feat(detection): add insecure-deserialization rule for Python (pickle, yaml.load, marshal, shelve)

signetry_core/pipeline/findings/deterministic.py

@@ -255,6 +255,34 @@ def visit_Call(self, node: ast.Call) -> None:  # noqa: C901 - dispatch table by
                       "Use bcrypt/scrypt/Argon2 for passwords; SHA-256+ for integrity.",
                       0.8, "CWE-327")
 
+        # --- Insecure deserialization (pickle, yaml.load, marshal, shelve) ---
+        if target in ("pickle.loads", "pickle.load", "_pickle.loads", "_pickle.load",
+                      "marshal.loads", "marshal.load", "shelve.open") and args:
+            if not isinstance(args[0], ast.Constant):
+                self._add(node, "py.insecure_deserialization", "deserialization", Severity.HIGH,
+                          f"Insecure deserialization via {target}()",
+                          f"{target}() on untrusted input allows arbitrary code execution (RCE) via object instantiation.",
+                          "Avoid deserializing untrusted data with pickle/marshal/shelve; use safe data formats like JSON.",
+                          0.85, "CWE-502")
+
+        if target == "yaml.load" and args:
+            is_safe_loader = False
+            for kw in node.keywords:
+                if kw.arg == "Loader":
+                    loader_name = _attr_chain(kw.value)
+                    if loader_name in ("yaml.SafeLoader", "SafeLoader", "yaml.CSafeLoader", "CSafeLoader"):
+                        is_safe_loader = True
+            if len(args) >= 2:
+                loader_name = _attr_chain(args[1])
+                if loader_name in ("yaml.SafeLoader", "SafeLoader", "yaml.CSafeLoader", "CSafeLoader"):
+                    is_safe_loader = True
+            if not is_safe_loader:
+                self._add(node, "py.insecure_deserialization", "deserialization", Severity.HIGH,
+                          "Unsafe YAML deserialization via yaml.load()",
+                          "yaml.load() without SafeLoader allows arbitrary Python object instantiation and code execution.",
+                          "Use yaml.safe_load() or pass Loader=yaml.SafeLoader.",
+                          0.85, "CWE-502")
+
         # --- Path traversal: open(<tainted path>) ---
         if target in ("open",) and args:
             a0 = args[0]

tests/test_findings_engine.py

@@ -263,6 +263,62 @@ def test_detects_ssrf():
     assert "ssrf" in _cats(src)
 
 
+def test_detects_insecure_deserialization_pickle_and_marshal():
+    src_pickle = (
+        "import pickle, flask\n"
+        "def handler():\n"
+        "    raw = flask.request.data\n"
+        "    return pickle.loads(raw)\n"
+    )
+    assert "deserialization" in _cats(src_pickle)
+
+    src_marshal = (
+        "import marshal\n"
+        "def run_code(user_bytes):\n"
+        "    return marshal.loads(user_bytes)\n"
+    )
+    assert "deserialization" in _cats(src_marshal)
+
+    src_shelve = (
+        "import shelve\n"
+        "def open_db(user_file):\n"
+        "    return shelve.open(user_file)\n"
+    )
+    assert "deserialization" in _cats(src_shelve)
+
+
+def test_detects_unsafe_yaml_load():
+    src_no_loader = (
+        "import yaml\n"
+        "def parse(raw_yaml):\n"
+        "    return yaml.load(raw_yaml)\n"
+    )
+    assert "deserialization" in _cats(src_no_loader)
+
+    src_unsafe_loader = (
+        "import yaml\n"
+        "def parse(raw_yaml):\n"
+        "    return yaml.load(raw_yaml, Loader=yaml.Loader)\n"
+    )
+    assert "deserialization" in _cats(src_unsafe_loader)
+
+
+def test_no_false_positive_on_safe_deserialization():
+    src_safe = (
+        "import json, yaml, pickle\n"
+        "from yaml import SafeLoader\n"
+        "def safe_parsers(user_input):\n"
+        "    d1 = json.loads(user_input)\n"
+        "    d2 = yaml.safe_load(user_input)\n"
+        "    d3 = yaml.load(user_input, Loader=yaml.SafeLoader)\n"
+        "    d4 = yaml.load(user_input, Loader=SafeLoader)\n"
+        "    d5 = yaml.load(user_input, yaml.SafeLoader)\n"
+        "    d6 = pickle.loads(b'hardcoded_constant_bytes')\n"
+        "    return d1, d2, d3, d4, d5, d6\n"
+    )
+    assert "deserialization" not in _cats(src_safe)
+
+
 def test_detects_ssti():
     src = (
         "import flask\n"