Agent: Add T1145 attack telemetry

monkey/infection_monkey/credential_collectors/ssh_collector/ssh_handler.py

@@ -6,6 +6,7 @@ from typing import Dict, Iterable
 
 from common.utils.attack_utils import ScanStatus
 from infection_monkey.telemetry.attack.t1005_telem import T1005Telem
+from infection_monkey.telemetry.attack.t1145_telem import T1145Telem
 
 logger = logging.getLogger(__name__)
 
@@ -81,6 +82,9 @@ def _get_ssh_files(usr_info: Iterable[Dict]) -> Iterable[Dict]:
                                             T1005Telem(
                                                 ScanStatus.USED, "SSH key", "Path: %s" % private
                                             ).send()
+                                            T1145Telem(
+                                                ScanStatus.USED, info["name"], info["home_dir"]
+                                            ).send()
                                         else:
                                             continue
                                 except (IOError, OSError):

monkey/infection_monkey/telemetry/attack/t1145_telem.py

@@ -0,0 +1,19 @@
+from infection_monkey.telemetry.attack.attack_telem import AttackTelem
+
+
+class T1145Telem(AttackTelem):
+    def __init__(self, status, name, home_dir):
+        """
+        T1145 telemetry.
+        :param status: ScanStatus of technique
+        :param name: Username from which ssh keypair is taken
+        :param home_dir: Home directory where we found the ssh keypair
+        """
+        super(T1145Telem, self).__init__("T1145", status)
+        self.name = name
+        self.home_dir = home_dir
+
+    def get_data(self):
+        data = super(T1145Telem, self).get_data()
+        data.update({"name": self.name, "home_dir": self.home_dir})
+        return data

monkey/tests/unit_tests/infection_monkey/telemetry/attack/test_t1145_telem.py

@@ -0,0 +1,28 @@
+import json
+
+import pytest
+
+from common.utils.attack_utils import ScanStatus
+from infection_monkey.telemetry.attack.t1145_telem import T1145Telem
+
+NAME = "ubuntu"
+HOME_DIR = "/home/ubuntu"
+STATUS = ScanStatus.USED
+
+
+@pytest.fixture
+def T1145_telem_test_instance():
+    return T1145Telem(STATUS, NAME, HOME_DIR)
+
+
+def test_T1145_send(T1145_telem_test_instance, spy_send_telemetry):
+    T1145_telem_test_instance.send()
+    expected_data = {
+        "status": STATUS.value,
+        "technique": "T1145",
+        "name": NAME,
+        "home_dir": HOME_DIR,
+    }
+    expected_data = json.dumps(expected_data, cls=T1145_telem_test_instance.json_encoder)
+    assert spy_send_telemetry.data == expected_data
+    assert spy_send_telemetry.telem_category == "attack"