Merge pull request #2385 from guardicore/2269-propagation-event

Define Propagation Event
This commit is contained in:
Mike Salvatore 2022-10-03 10:34:33 -04:00 committed by GitHub
commit 07839a46ae
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 109 additions and 4 deletions

View File

@ -2,3 +2,4 @@ from .abstract_agent_event import AbstractAgentEvent
from .credentials_stolen_events import CredentialsStolenEvent
from .ping_scan_event import PingScanEvent
from .tcp_scan_event import TCPScanEvent
from .propagation_event import PropagationEvent

View File

@ -0,0 +1,22 @@
from ipaddress import IPv4Address
from pydantic import Field
from . import AbstractAgentEvent
class PropagationEvent(AbstractAgentEvent):
"""
An event that occurs when the Agent propagates on a host
Attributes:
:param target: IP address of the propagated system
:param success: Status of the propagation
:param exploiter_name: Name of the exploiter that propagated
:param error_message: Message if an error occurs during propagation
"""
target: IPv4Address
success: bool
exploiter_name: str
error_message: str = Field(default="")

View File

@ -0,0 +1,82 @@
from ipaddress import IPv4Address
from uuid import UUID
import pytest
from common.agent_events import PropagationEvent
TARGET_IP_STR = "192.168.1.10"
AGENT_ID = UUID("012e7238-7b81-4108-8c7f-0787bc3f3c10")
TIMESTAMP = 1664371327.4067292
PROPAGATION_EVENT = PropagationEvent(
source=AGENT_ID,
timestamp=TIMESTAMP,
target=IPv4Address(TARGET_IP_STR),
success=True,
exploiter_name="SSHExploiter",
)
PROPAGATION_OBJECT_DICT = {
"source": AGENT_ID,
"timestamp": TIMESTAMP,
"target": IPv4Address(TARGET_IP_STR),
"success": True,
"exploiter_name": "SSHExploiter",
"error_message": "",
}
PROPAGATION_SIMPLE_DICT = {
"source": str(AGENT_ID),
"timestamp": TIMESTAMP,
"target": TARGET_IP_STR,
"success": "true",
"exploiter_name": "SSHExploiter",
"error_message": "",
}
@pytest.mark.parametrize(
"propagation_event_dict", [PROPAGATION_OBJECT_DICT, PROPAGATION_SIMPLE_DICT]
)
def test_constructor(propagation_event_dict):
assert PropagationEvent(**propagation_event_dict) == PROPAGATION_EVENT
@pytest.mark.parametrize(
"key, value",
[
("target", None),
("success", "not-a-bool"),
("exploiter_name", None),
("error_message", None),
],
)
def test_construct_invalid_field__type_error(key, value):
invalid_type_dict = PROPAGATION_SIMPLE_DICT.copy()
invalid_type_dict[key] = value
with pytest.raises(TypeError):
PropagationEvent(**invalid_type_dict)
@pytest.mark.parametrize(
"key, value",
[
("target", "not-an-ip"),
],
)
def test_construct_invalid_field__value_error(key, value):
invalid_type_dict = PROPAGATION_SIMPLE_DICT.copy()
invalid_type_dict[key] = value
with pytest.raises(ValueError):
PropagationEvent(**invalid_type_dict)
def test_construct__extra_fields_forbidden():
extra_field_dict = PROPAGATION_SIMPLE_DICT.copy()
extra_field_dict["extra_field"] = 99 # red balloons
with pytest.raises(ValueError):
PropagationEvent(**extra_field_dict)

View File

@ -7,7 +7,7 @@ from common.agent_configuration.agent_sub_configurations import (
CustomPBAConfiguration,
ScanTargetConfiguration,
)
from common.agent_events import PingScanEvent, TCPScanEvent
from common.agent_events import PingScanEvent, PropagationEvent, TCPScanEvent
from common.credentials import Credentials, LMHash, NTHash
from common.types import NetworkPort
from infection_monkey.exploit.log4shell_utils.ldap_server import LDAPServerFactory
@ -311,13 +311,13 @@ IAgentLogRepository
IAgentLogRepository.upsert_agent_log
IAgentLogRepository.get_agent_log
# TODO: Remove once #2268 is closed
PingScanEvent
# TODO: Remove once #2267 is closed
TCPScanEvent
TCPScanEvent.port_status
# TODO: Remove once #2269 is close
PropagationEvent
# pydantic base models
underscore_attrs_are_private
extra