diff --git a/signetry_core/pipeline/findings/deterministic.py b/signetry_core/pipeline/findings/deterministic.py index 13f9ee3..d83711f 100644 --- a/signetry_core/pipeline/findings/deterministic.py +++ b/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() --- if target in ("open",) and args: a0 = args[0] diff --git a/tests/test_findings_engine.py b/tests/test_findings_engine.py index 9a23994..74072bb 100644 --- a/tests/test_findings_engine.py +++ b/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"