Common: Add PasswordRestorationEvent

monkey/common/agent_events/__init__.py

@@ -4,3 +4,4 @@ from .ping_scan_event import PingScanEvent
 from .tcp_scan_event import TCPScanEvent
 from .exploitation_event import ExploitationEvent
 from .propagation_event import PropagationEvent
+from .password_restoration_event import PasswordRestorationEvent

monkey/common/agent_events/password_restoration_event.py

@@ -0,0 +1,18 @@
+from ipaddress import IPv4Address
+
+from . import AbstractAgentEvent
+
+
+class PasswordRestorationEvent(AbstractAgentEvent):
+    """
+    An event that occurs when a password has been restored on the target
+    system
+
+    Attributes:
+        :param target: The IP of the target system on which the
+            restoration was performed
+        :param success: If the password restoration was successful
+    """
+
+    target: IPv4Address
+    success: bool

monkey/tests/unit_tests/common/agent_events/test_password_restoration_event.py

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

vulture_allowlist.py

@@ -7,7 +7,13 @@ from common.agent_configuration.agent_sub_configurations import (
     CustomPBAConfiguration,
     ScanTargetConfiguration,
 )
-from common.agent_events import ExploitationEvent, PingScanEvent, PropagationEvent, TCPScanEvent
+from common.agent_events import (
+    ExploitationEvent,
+    PasswordRestorationEvent,
+    PingScanEvent,
+    PropagationEvent,
+    TCPScanEvent,
+)
 from common.credentials import Credentials, LMHash, NTHash
 from common.tags import (
     T1021_ATTACK_TECHNIQUE_TAG,
@@ -336,6 +342,7 @@ T1222_ATTACK_TECHNIQUE_TAG
 T1570_ATTACK_TECHNIQUE_TAG
 HostExploiter._publish_propagation_event
 HostExploiter._publish_exploitation_event
+PasswordRestorationEvent
 
 # pydantic base models
 underscore_attrs_are_private