feat(pre-push-safety-net): reproduce evaluated Joern runtime

2026-10-17 12:09:27 +00:00 · 2026-09-12 18:32:18 +02:00
parent 7e3b3bbc8b
commit 56cfc9da89
8 changed files with 252 additions and 11 deletions
@@ -0,0 +1,117 @@
+"""Exercise the installed engine, not a mock. Run inside the shipped image as its non-root user."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+from openultrasast.cpg.backend import DATAFLOW_OVERLAY, JoernBackend, joern_version
+
+
+def require(condition: object, message: str) -> None:
+    if not condition:
+        raise RuntimeError(message)
+
+
+def main() -> None:
+    require(os.getuid() != 0, "run as the image's non-root user so unreadable-file checks are meaningful")
+    version = joern_version()
+    require(version, "installed Joern version is unknown")
+    php = Path(shutil.which("php") or "/missing-php").resolve()
+    require(php.read_bytes()[:4] == b"\x7fELF", "PHP must be a native executable, not a stdio wrapper")
+    parser = Path(shutil.which("php2cpg") or "/missing-php2cpg").resolve().parent
+    parser /= "frontends/php2cpg/bin/php-parser/php-parser.php"
+    probe = "$s = file_get_contents($argv[1]); if ($s === false || strlen($s) === 0) exit(3); echo strlen($s);"
+
+    def php_bytes(path: Path) -> int:
+        result = subprocess.run([str(php), "-r", probe, str(path)], check=True, capture_output=True, text=True, timeout=15)
+        return int(result.stdout)
+
+    report: dict[str, object] = {
+        "joern_version": ".".join(map(str, version)),
+        "php_version": subprocess.run([str(php), "-v"], check=True, capture_output=True, text=True, timeout=15).stdout.splitlines()[0],
+        "php_parser_bytes": php_bytes(parser),
+        "purpose": "runtime and query smoke; does not qualify vulnerability detection or hook latency",
+        "languages": {},
+    }
+    languages = report["languages"]
+    assert isinstance(languages, dict)
+    with tempfile.TemporaryDirectory(prefix="ousast-engine-smoke-") as temporary:
+        root = Path(temporary)
+        queries = root / "queries"
+        queries.mkdir()
+        (queries / "source_witness.sc").write_text("""@main def exec(cpgFile: String) = {
+  importCpg(cpgFile)
+  println("---OUSAST-CPG-BEGIN---")
+  println(ujson.write(ujson.Obj(
+    "files" -> ujson.Arr.from(cpg.file.name.l),
+    "handlers" -> cpg.method.nameExact("handler").size,
+    "evalCalls" -> cpg.call.nameExact("eval").size
+  )))
+  println("---OUSAST-CPG-END---")
+}
+""")
+        for language, frontend, filename, content in (
+            ("php", "php2cpg", "probe.php", "<?php\nfunction handler($req) { eval($req); }\n"),
+            ("javascript", "jssrc2cpg", "probe.js", "function handler(req) { eval(req.query.code); }\n"),
+        ):
+            require(shutil.which(frontend), f"missing installed {frontend}")
+            source_root = root / language
+            source_root.mkdir()
+            source = source_root / filename
+            source.write_text(content)
+            data = source.read_bytes()
+            require(data, f"unreadable/empty {source}")
+            if language == "php":
+                require(php_bytes(source) == len(data), "PHP and host source byte counts differ")
+            print(f"{language}: opened {filename}, {len(data)} bytes; building graph", file=sys.stderr, flush=True)
+            backend = JoernBackend(build_timeout=180, query_timeout=180, heap_mb=1024)
+            started = time.monotonic()
+            graph = backend.build(source_root, language=language)
+            require(graph, f"{language} graph build failed: {backend.last_failure}")
+            assert graph is not None
+            try:
+                require(not graph.unparsed, f"{language} unparsed input: {graph.unparsed}")
+                census = backend.query(graph.cpg_path, "census", {})
+                require(isinstance(census, dict), f"{language} census returned no valid payload")
+                assert isinstance(census, dict)
+                require(int(census["files"]) > 0 and int(census["methods"]) > 0, f"empty {language} census")
+                require(DATAFLOW_OVERLAY in str(census["overlays"]).split(","), f"{language} missing dataflow overlay: {census}")
+                witness = JoernBackend(queries_dir=queries, query_timeout=180, heap_mb=1024).query(graph.cpg_path, "source_witness", {})
+                require(isinstance(witness, dict), f"{language} source witness unanswered")
+                assert isinstance(witness, dict)
+                require(any(str(path).endswith(filename) for path in witness["files"]), f"{language} source absent from graph")
+                require(witness["handlers"] > 0 and witness["evalCalls"] > 0, f"{language} expected source nodes missing: {witness}")
+                languages[language] = {
+                    "frontend": frontend,
+                    "source_bytes": len(data),
+                    "source_sha256": hashlib.sha256(data).hexdigest(),
+                    "graph_bytes": graph.cpg_path.stat().st_size,
+                    "census": census,
+                    "source_witness": witness,
+                    "elapsed_seconds": round(time.monotonic() - started, 3),
+                }
+                print(f"{language}: census and source witness verified", file=sys.stderr, flush=True)
+            finally:
+                if graph.cleanup:
+                    graph.cleanup()
+            if language == "php":
+                source.chmod(0)
+                try:
+                    refused = backend.build(source_root, language=language)
+                    require(refused is None and "cannot read" in backend.last_failure, "unreadable source did not fail explicitly")
+                    report["unreadable_source"] = {"status": "explicit_failure", "reason": backend.last_failure}
+                finally:
+                    source.chmod(0o600)
+    print(json.dumps(report, indent=2))
+
+
+if __name__ == "__main__":
+    main()