add processing for the token src_data

thinkst · Nov 20, 2024 · c08e8a9 · c08e8a9
1 parent 17da502
commit c08e8a9
Show file tree

Hide file tree

Showing 3 changed files with 75 additions and 0 deletions.
diff --git a/canarytokens/channel.py b/canarytokens/channel.py
@@ -206,6 +206,12 @@ def gather_alert_details(
         additional_data = hit.get_additional_data_for_notification()
         if canarydrop.cmd_process:
             additional_data["cmd_process"] = canarydrop.cmd_process
+        if canarydrop.windows_fake_fs_root:
+            additional_data["windows_fake_fs_root"] = canarydrop.windows_fake_fs_root
+        if canarydrop.windows_fake_fs_file_structure:
+            additional_data[
+                "windows_fake_fs_file_structure"
+            ] = canarydrop.windows_fake_fs_file_structure
 
         return TokenAlertDetails(
             channel=cls.CHANNEL,

diff --git a/canarytokens/tokens.py b/canarytokens/tokens.py
@@ -56,6 +56,9 @@
     ),
     re.IGNORECASE,
 )
+windows_fake_fs_pattern = re.compile(
+    r"u([A-Za-z0-9]*)\.f([A-Za-z0-9]*)\.i([A-Za-z0-9]*)", re.IGNORECASE
+)
 
 # to validate decoded sql username, not a data extractor:
 sql_decoded_username = re.compile(r"[A-Za-z0-9\!\#\'\-\.\\\^\_\~]+")
@@ -69,6 +72,7 @@
     "desktop_ini_browsing": desktop_ini_browsing_pattern,
     "log4_shell": log4_shell_pattern,
     "cmd_process": cmd_process_pattern,
+    "windows_fake_fs": windows_fake_fs_pattern,
     "sql_server_username": sql_server_username,
 }
 
@@ -245,6 +249,25 @@ def _cmd_process(matches: Match[AnyStr]) -> dict[str, dict[str, AnyStr]]:
 
         return {"src_data": data}
 
+    @staticmethod
+    def _windows_fake_fs(matches: Match[AnyStr]) -> dict[str, dict[str, AnyStr]]:
+        """"""
+        invocation_id = matches.group(1)
+        file_name = matches.group(2)
+        process_name = matches.group(3)
+        data = {
+            "windows_fake_fs_file_name": "(not obtained)",
+            "windows_fake_fs_process_name": "(not obtained)",
+        }
+        if invocation_id:
+            data["windows_fake_fs_invocation_id"] = invocation_id[0:].lower()
+        if file_name and file_name != "u":
+            data["windows_fake_fs_file_name"] = file_name[0:].lower()
+        if process_name and process_name != "c":
+            data["windows_fake_fs_process_name"] = process_name[0:].lower()
+
+        return {"src_data": data}
+
     @staticmethod
     def _desktop_ini_browsing(matches: Match[AnyStr]) -> dict[str, dict[str, AnyStr]]:
         username = matches.group(1)

diff --git a/tests/units/test_tokens.py b/tests/units/test_tokens.py
@@ -88,6 +88,52 @@ def test_cmd_process_pattern(
     assert data["src_data"].get("cmd_invocation_id") == cmd_invocation_id
 
 
+@pytest.mark.parametrize(
+    "query, invocation_id, file_name, process_name, ",
+    [
+        # (
+        #     "cbrokenpc.UN.ubrokenuser.CMD.someid78.sometoken.com",
+        #     "brokenpc",
+        #     "brokenuser",
+        #     "someid78",
+        # ),
+        # ("cbrokenpc.UN.ubrokenuser.CMD.sometoken.com", "brokenpc", "brokenuser", None),
+        # ("c.UN.ubrokenuser.CMD.sometoken.com", "(not obtained)", "brokenuser", None),
+        # ("cbrokenpc.UN.u.CMD.sometoken.com", "brokenpc", "(not obtained)", None),
+        (
+            "u7595.fMRXWGIDCFZSG6Y3Y.iMV4HA3DPOJSXELTFPBSQ.someid.sometoken.com",
+            "7595",
+            "MRXWGIDCFZSG6Y3Y",
+            "MV4HA3DPOJSXELTFPBSQ",
+        ),
+        (
+            "u7595.f.iMV4HA3DPOJSXELTFPBSQ.someid.sometoken.com",
+            "7595",
+            "(not obtained)",
+            "MV4HA3DPOJSXELTFPBSQ",
+        ),
+        (
+            "u7595.fMRXWGIDCFZSG6Y3Y.i.someid.sometoken.com",
+            "7595",
+            "MRXWGIDCFZSG6Y3Y",
+            "(not obtained)",
+        ),
+        (
+            "u7595.f.i.someid.sometoken.com",
+            "7595",
+            "(not obtained)",
+            "(not obtained)",
+        ),
+    ],
+)
+def test_windows_fake_fs_pattern(query, invocation_id, file_name, process_name):
+    m = t.windows_fake_fs_pattern.match(query)
+    data = t.Canarytoken._windows_fake_fs(m)
+    assert data["src_data"]["windows_fake_fs_invocation_id"] == invocation_id.lower()
+    assert data["src_data"]["windows_fake_fs_file_name"] == file_name.lower()
+    assert data["src_data"]["windows_fake_fs_process_name"] == process_name.lower()
+
+
 def test_canarytoken_create_and_fetch():
     ct = t.Canarytoken()
     ct_new = t.Canarytoken(value=ct.value())