Merge pull request #1929 from blacklanternsecurity/magic

Add native filetype + compression detection

bbot/core/event/base.py

@@ -503,12 +503,13 @@ def scope_distance(self, scope_distance):
             for t in list(self.tags):
                 if t.startswith("distance-"):
                     self.remove_tag(t)
-            if scope_distance == 0:
-                self.add_tag("in-scope")
-                self.remove_tag("affiliate")
-            else:
-                self.remove_tag("in-scope")
-                self.add_tag(f"distance-{new_scope_distance}")
+            if self.host:
+                if scope_distance == 0:
+                    self.add_tag("in-scope")
+                    self.remove_tag("affiliate")
+                else:
+                    self.remove_tag("in-scope")
+                    self.add_tag(f"distance-{new_scope_distance}")
             self._scope_distance = new_scope_distance
             # apply recursively to parent events
             parent_scope_distance = getattr(self.parent, "scope_distance", None)
@@ -1018,20 +1019,21 @@ def __init__(self, *args, **kwargs):
 class DictPathEvent(DictEvent):
     def sanitize_data(self, data):
         new_data = dict(data)
+        new_data["path"] = str(new_data["path"])
         file_blobs = getattr(self.scan, "_file_blobs", False)
         folder_blobs = getattr(self.scan, "_folder_blobs", False)
         blob = None
         try:
-            data_path = Path(data["path"])
-            if data_path.is_file():
+            self._data_path = Path(data["path"])
+            if self._data_path.is_file():
                 self.add_tag("file")
                 if file_blobs:
-                    with open(data_path, "rb") as file:
+                    with open(self._data_path, "rb") as file:
                         blob = file.read()
-            elif data_path.is_dir():
+            elif self._data_path.is_dir():
                 self.add_tag("folder")
                 if folder_blobs:
-                    blob = self._tar_directory(data_path)
+                    blob = self._tar_directory(self._data_path)
         except KeyError:
             pass
         if blob:
@@ -1540,7 +1542,23 @@ def _pretty_string(self):
 
 
 class FILESYSTEM(DictPathEvent):
-    pass
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        if self._data_path.is_file():
+            # detect type of file content using magic
+            from bbot.core.helpers.libmagic import get_magic_info, get_compression
+
+            extension, mime_type, description, confidence = get_magic_info(self.data["path"])
+            self.data["magic_extension"] = extension
+            self.data["magic_mime_type"] = mime_type
+            self.data["magic_description"] = description
+            self.data["magic_confidence"] = confidence
+            # detection compression
+            compression = get_compression(mime_type)
+            if compression:
+                self.add_tag("compressed")
+                self.add_tag(f"{compression}-archive")
+                self.data["compression"] = compression
 
 
 class RAW_DNS_RECORD(DictHostEvent, DnsEvent):

bbot/core/helpers/libmagic.py

@@ -0,0 +1,68 @@
+import puremagic
+
+
+def get_magic_info(file):
+
+    magic_detections = puremagic.magic_file(file)
+    if magic_detections:
+        magic_detections.sort(key=lambda x: x.confidence, reverse=True)
+        detection = magic_detections[0]
+        return detection.extension, detection.mime_type, detection.name, detection.confidence
+    return "", "", "", 0
+
+
+def get_compression(mime_type):
+    mime_type = mime_type.lower()
+    # from https://github.com/cdgriffith/puremagic/blob/master/puremagic/magic_data.json
+    compression_map = {
+        "application/gzip": "gzip",  # Gzip compressed file
+        "application/zip": "zip",  # Zip archive
+        "application/x-bzip2": "bzip2",  # Bzip2 compressed file
+        "application/x-xz": "xz",  # XZ compressed file
+        "application/x-7z-compressed": "7z",  # 7-Zip archive
+        "application/vnd.rar": "rar",  # RAR archive
+        "application/x-lzma": "lzma",  # LZMA compressed file
+        "application/x-compress": "compress",  # Unix compress file
+        "application/zstd": "zstd",  # Zstandard compressed file
+        "application/x-lz4": "lz4",  # LZ4 compressed file
+        "application/x-tar": "tar",  # Tar archive
+        "application/x-zip-compressed-fb2": "zip",  # Zip archive (FB2)
+        "application/epub+zip": "zip",  # EPUB book (Zip archive)
+        "application/pak": "pak",  # PAK archive
+        "application/x-lha": "lha",  # LHA archive
+        "application/arj": "arj",  # ARJ archive
+        "application/vnd.ms-cab-compressed": "cab",  # Microsoft Cabinet archive
+        "application/x-sit": "sit",  # StuffIt archive
+        "application/binhex": "binhex",  # BinHex encoded file
+        "application/x-lrzip": "lrzip",  # Long Range ZIP
+        "application/x-alz": "alz",  # ALZip archive
+        "application/x-tgz": "tgz",  # Gzip compressed Tar archive
+        "application/x-gzip": "gzip",  # Gzip compressed file
+        "application/x-lzip": "lzip",  # Lzip compressed file
+        "application/x-zstd-compressed-tar": "zstd",  # Zstandard compressed Tar archive
+        "application/x-lz4-compressed-tar": "lz4",  # LZ4 compressed Tar archive
+        "application/vnd.comicbook+zip": "zip",  # Comic book archive (Zip)
+        "application/vnd.palm": "palm",  # Palm OS data
+        "application/fictionbook2+zip": "zip",  # FictionBook 2.0 (Zip)
+        "application/fictionbook3+zip": "zip",  # FictionBook 3.0 (Zip)
+        "application/x-cpio": "cpio",  # CPIO archive
+        "application/x-java-pack200": "pack200",  # Java Pack200 archive
+        "application/x-par2": "par2",  # PAR2 recovery file
+        "application/x-rar-compressed": "rar",  # RAR archive
+        "application/java-archive": "zip",  # Java Archive (JAR)
+        "application/x-webarchive": "zip",  # Web archive (Zip)
+        "application/vnd.android.package-archive": "zip",  # Android package (APK)
+        "application/x-itunes-ipa": "zip",  # iOS application archive (IPA)
+        "application/x-stuffit": "sit",  # StuffIt archive
+        "application/x-archive": "ar",  # Unix archive
+        "application/x-qpress": "qpress",  # Qpress archive
+        "application/x-xar": "xar",  # XAR archive
+        "application/x-ace": "ace",  # ACE archive
+        "application/x-zoo": "zoo",  # Zoo archive
+        "application/x-arc": "arc",  # ARC archive
+        "application/x-zstd-compressed-tar": "zstd",  # Zstandard compressed Tar archive
+        "application/x-lz4-compressed-tar": "lz4",  # LZ4 compressed Tar archive
+        "application/vnd.comicbook-rar": "rar",  # Comic book archive (RAR)
+    }
+
+    return compression_map.get(mime_type, "")

bbot/test/test_step_1/test_events.py

@@ -925,3 +925,44 @@ def test_event_closest_host():
         vuln = scan.make_event(
             {"path": "/tmp/asdf.txt", "description": "test", "severity": "HIGH"}, "VULNERABILITY", parent=event3
         )
+
+
+def test_event_magic():
+    from bbot.core.helpers.libmagic import get_magic_info, get_compression
+
+    import base64
+
+    zip_base64 = "UEsDBAoDAAAAAOMmZ1lR4FaHBQAAAAUAAAAIAAAAYXNkZi50eHRhc2RmClBLAQI/AwoDAAAAAOMmZ1lR4FaHBQAAAAUAAAAIACQAAAAAAAAAIICkgQAAAABhc2RmLnR4dAoAIAAAAAAAAQAYAICi2B77MNsBgKLYHvsw2wGAotge+zDbAVBLBQYAAAAAAQABAFoAAAArAAAAAAA="
+    zip_bytes = base64.b64decode(zip_base64)
+    zip_file = Path("/tmp/.bbottestzipasdkfjalsdf.zip")
+    with open(zip_file, "wb") as f:
+        f.write(zip_bytes)
+
+    # test magic helpers
+    extension, mime_type, description, confidence = get_magic_info(zip_file)
+    assert extension == ".zip"
+    assert mime_type == "application/zip"
+    assert description == "PKZIP Archive file"
+    assert confidence > 0
+    assert get_compression(mime_type) == "zip"
+
+    # test filesystem event - file
+    scan = Scanner()
+    event = scan.make_event({"path": zip_file}, "FILESYSTEM", parent=scan.root_event)
+    assert event.data == {
+        "path": "/tmp/.bbottestzipasdkfjalsdf.zip",
+        "magic_extension": ".zip",
+        "magic_mime_type": "application/zip",
+        "magic_description": "PKZIP Archive file",
+        "magic_confidence": 0.9,
+        "compression": "zip",
+    }
+    assert event.tags == {"file", "zip-archive", "compressed"}
+
+    # test filesystem event - folder
+    scan = Scanner()
+    event = scan.make_event({"path": "/tmp"}, "FILESYSTEM", parent=scan.root_event)
+    assert event.data == {"path": "/tmp"}
+    assert event.tags == {"folder"}
+
+    zip_file.unlink()

pyproject.toml

@@ -55,6 +55,7 @@ setproctitle = "^1.3.3"
 yara-python = "^4.5.1"
 pyzmq = "^26.0.3"
 httpx = "^0.27.0"
+puremagic = "^1.28"
 
 [tool.poetry.group.dev.dependencies]
 flake8 = ">=6,<8"