From 562b670a4366868062c2b8c0c8e777a845e56459 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Fri, 25 Sep 2026 12:37:22 +0200 Subject: [PATCH] feat(rules): Django request sources; a destination with a fixed origin is not untrusted python.toml's sources were Flask-only, so no flow could start anywhere on a Django project. OpenCVE's SSRF reads its URL from `form.cleaned_data`, and the scan answered the sink file and found nothing. Added: request.GET/POST/body/COOKIES/FILES/META, DRF's query_params, and cleaned_data. With those sources known, OpenCVE reported three redirects that cannot leave the site: `reverse("cves") + "?" + request.GET.urlencode()` (twice) and `reverse(route, kwargs=...)`. Destination sink facts can now set `prefix_fixes_origin` and `origin_anchors`. A path is cut when it enters a concatenation, f-string or `+=` through anything other than its leftmost operand, and that operand fixes the origin. Origin-fixing operands: - a literal `/x`, `?` or `#`; - a literal of the form scheme://host/; - an anchor call; - a variable whose every assignment fixes the origin. A path is also cut when it passes through an anchor. `"/" + x` (//evil) and `"https://" + host` stay reported. A Joern-gated test covers all seven shapes. Exploratory (retune after protocol-v1): OpenCVE vulnerable and fixed pins are both 0 findings, down from 2 and 3. The SSRF itself is still missed, because the flow crosses a registry dispatch (`REGISTRY.get(t)(extras).run()`) that the engine does not follow. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo --- src/openultrasast/cpg/queries/taint.sc | 86 ++++++++++++++++++- src/openultrasast/model/specs.py | 11 +++ src/openultrasast/model/taint.py | 2 + .../ruleset/semantic/javascript.toml | 2 + .../ruleset/semantic/python.toml | 21 +++++ src/openultrasast/semantic/facts.py | 16 ++++ tests/test_cpg_queries_compile.py | 74 ++++++++++++++++ 7 files changed, 209 insertions(+), 3 deletions(-) diff --git a/src/openultrasast/cpg/queries/taint.sc b/src/openultrasast/cpg/queries/taint.sc index 9ea3bee..f6e1b3c 100644 --- a/src/openultrasast/cpg/queries/taint.sc +++ b/src/openultrasast/cpg/queries/taint.sc @@ -104,7 +104,9 @@ contextS: String = "", contextFilterS: String = "", contextPathsS: String = "", - traceS: String = "" + traceS: String = "", + fixedOriginS: String = "", + originAnchorsS: String = "" ): List[ujson.Obj] = { // Word-boundary matching, never substring. `resolveUrl` contains `resolve`, so a bare-substring sanitizer @@ -400,6 +402,82 @@ } } + // A destination is untrusted only where the value can choose WHERE it goes. OpenCVE redirects to + // `reverse("cves") + "?" + request.GET.urlencode()` and to `reverse(route, kwargs={...})`: the request + // decides the query string or a path segment, and the redirect stays on the site. So, for a sink fact that + // says so, a path is cut when it enters a concatenation through anything but its LEFTMOST operand and that + // operand already fixes the origin, or when it passes through an anchor call such as `reverse`. What does + // NOT fix an origin is kept deliberately: `"/" + x` becomes `//evil.example`, and `"https://" + host` is the + // attack itself. + val originFixing = fixedOriginS == "true" + val originAnchors = split(originAnchorsS).toSet + val schemeHostThenPath = java.util.regex.Pattern.compile("^[A-Za-z][A-Za-z0-9+.-]*://[^/?#\\\\]+[/?#]") + val concatenations = Set(".addition", ".assignmentPlus", ".formatString") + val fixesOriginMemo = scala.collection.mutable.Map.empty[Long, Boolean] + + def literalFixesOrigin(code: String): Boolean = { + // The literal's TEXT, without its quotes. With the closing quote left on, `"/"` read as `/"` and passed + // the path test -- and `"/" + x` is the protocol-relative `//evil.example`. + val quoted = "(?s)^(?:[rRbBuUfF]{1,2})?(['\"`])(.*)\\1$".r + val text = code.trim match { + case quoted(_, inner) => inner + case other => other + } + text.startsWith("?") || text.startsWith("#") || text.matches("(?s)^/[^/\\\\].*") || schemeHostThenPath.matcher(text).find() + } + + def fixesOrigin(node: io.shiftleft.codepropertygraph.generated.nodes.AstNode, depth: Int): Boolean = + depth <= 3 && fixesOriginMemo.getOrElseUpdate(node.id, node match { + case l: io.shiftleft.codepropertygraph.generated.nodes.Literal => literalFixesOrigin(l.code) + case c: io.shiftleft.codepropertygraph.generated.nodes.Call if originAnchors.contains(c.name) => true + case c: io.shiftleft.codepropertygraph.generated.nodes.Call if concatenations.contains(c.name) => + c.argument.l.sortBy(_.argumentIndex).headOption.exists(a => fixesOrigin(a, depth + 1)) + case i: io.shiftleft.codepropertygraph.generated.nodes.Identifier => + // A variable fixes the origin when EVERY assignment to it in the method does. + val writes = i.method.ast.isCall.nameExact(ASSIGNMENT).l.filter(_.argument.l.sortBy(_.argumentIndex).headOption.exists { + case t: io.shiftleft.codepropertygraph.generated.nodes.Identifier => t.name == i.name + case _ => false + }) + writes.nonEmpty && writes.forall(w => w.argument.l.find(_.argumentIndex == 2).exists(v => fixesOrigin(v, depth + 1))) + case _ => false + }) + + // The argument of `call` that holds `node` -- the node itself or anywhere inside it. The engine does not + // always step through the argument: `reverse("cves") + ("?" + request.GET.urlencode())` goes from + // `request.GET` straight to the outer `+`. + def argumentHolding( + call: io.shiftleft.codepropertygraph.generated.nodes.Call, + node: io.shiftleft.codepropertygraph.generated.nodes.AstNode + ): Option[io.shiftleft.codepropertygraph.generated.nodes.Expression] = + call.argument.l.find(a => a.id == node.id || a.ast.exists(_.id == node.id)) + + def originFixed(elements: List[io.shiftleft.codepropertygraph.generated.nodes.AstNode]): Boolean = + originFixing && elements.indices.exists { k => + elements(k) match { + // Through an anchor: the value only fills the route's arguments. The anchor may be the path's last + // element, which is the sink's own argument. + case call: io.shiftleft.codepropertygraph.generated.nodes.Call if k > 0 && originAnchors.contains(call.name) => true + // `url += "&" + q`: the engine goes from the right-hand side to the TARGET `url`, never visiting the + // `+=` itself. The value is appended to what `url` already held. + case target: io.shiftleft.codepropertygraph.generated.nodes.Identifier if k > 0 && target.argumentIndex == 1 => + target.astParent match { + case augmented: io.shiftleft.codepropertygraph.generated.nodes.Call if augmented.name == ".assignmentPlus" => + argumentHolding(augmented, elements(k - 1)).exists(_.argumentIndex == 2) && fixesOrigin(target, 0) + case _ => false + } + case call: io.shiftleft.codepropertygraph.generated.nodes.Call if k > 0 && concatenations.contains(call.name) => + // The argument the path entered by, as in `offValueDispatch`: the first of the run of elements + // right before the call that lie inside its arguments. + var j = k - 1 + while (j >= 0 && argumentHolding(call, elements(j)).isDefined) j -= 1 + j < k - 1 && argumentHolding(call, elements(j + 1)).exists { entered => + val leftmost = call.argument.l.sortBy(_.argumentIndex).headOption + entered.argumentIndex > 1 && leftmost.exists(l => l.argumentIndex < entered.argumentIndex && fixesOrigin(l, 0)) + } + case _ => false + } + } + def plausible(flow: io.joern.dataflowengineoss.language.Path): Boolean = { val elements = flow.elements.l val throughSink = elements.dropRight(1).exists { @@ -407,7 +485,7 @@ familySinkMemo.getOrElseUpdate(c.id, familySinkCall(c)) case _ => false } - !throughSink && !offValueDispatch(elements) && !(elements.exists { + !throughSink && !offValueDispatch(elements) && !originFixed(elements) && !(elements.exists { case c: io.shiftleft.codepropertygraph.generated.nodes.Call => c.name == ".fieldAccess" case _ => false } && absorbedByObject(elements)) @@ -1180,7 +1258,9 @@ field("contextEvidence"), if (field("contextFilter").nonEmpty) field("contextFilter") else contextFilter, if (field("contextPaths").nonEmpty) field("contextPaths") else contextPaths, - field("trace") + field("trace"), + field("fixedOrigin"), + field("originAnchors") ): _* ) println(ujson.write(ujson.Obj("id" -> id, "rows" -> rows, "ms" -> ((System.nanoTime() - started) / 1000000L).toDouble))) diff --git a/src/openultrasast/model/specs.py b/src/openultrasast/model/specs.py index 87bcbfa..c5060f4 100644 --- a/src/openultrasast/model/specs.py +++ b/src/openultrasast/model/specs.py @@ -118,6 +118,10 @@ class TaintSpec: # Which argument of each apply call its RESULT comes from, as `call:position`; position 0 means the call # returns nothing a flow can carry. Only calls a fact describes appear here. dispatch_value: tuple[str, ...] = () + # Whether a prefix that fixes the origin makes a destination safe, and the calls that build such a + # prefix (`reverse`). From the sink facts; only destination families set them. + fixed_origin: bool = False + origin_anchors: tuple[str, ...] = () @dataclass(frozen=True) @@ -255,6 +259,8 @@ def taint_specs( safe_shapes = tuple(sorted({call for fact in scoped.sanitizers if _is_shape(fact) for call in fact.calls})) by_family: dict[str, set[str]] = {} bounded: dict[str, set[str]] = {} + fixed_origin: set[str] = set() + anchors: dict[str, set[str]] = {} for sink in scoped.sinks: family = families.family_of_cwe(sink.cwe) if family is None: @@ -264,6 +270,9 @@ def taint_specs( # is on the fact already, so which sinks these are is read from the data rather than declared here. if sink.cwe in OVERFLOW_CWES: bounded.setdefault(family.id, set()).update(sink.calls) + if sink.prefix_fixes_origin: + fixed_origin.add(family.id) + anchors.setdefault(family.id, set()).update(sink.origin_anchors) return { family_id: TaintSpec( family=family_id, @@ -275,6 +284,8 @@ def taint_specs( safe_shape_sinks=safe_shapes, dispatch_apply=dispatch_apply, dispatch_value=dispatch_value, + fixed_origin=family_id in fixed_origin, + origin_anchors=tuple(sorted(anchors.get(family_id, ()))), ) for family_id, calls in sorted(by_family.items()) } diff --git a/src/openultrasast/model/taint.py b/src/openultrasast/model/taint.py index 1818798..a45d1e7 100644 --- a/src/openultrasast/model/taint.py +++ b/src/openultrasast/model/taint.py @@ -60,6 +60,8 @@ def request_params( # `apply_filters` is, which makes a general mechanism into a WordPress feature. "dispatchApply": spec.dispatch_apply, "dispatchValue": spec.dispatch_value, + "fixedOrigin": "true" if spec.fixed_origin else "false", + "originAnchors": spec.origin_anchors, } diff --git a/src/openultrasast/ruleset/semantic/javascript.toml b/src/openultrasast/ruleset/semantic/javascript.toml index baec1e7..5b6d7e2 100644 --- a/src/openultrasast/ruleset/semantic/javascript.toml +++ b/src/openultrasast/ruleset/semantic/javascript.toml @@ -123,12 +123,14 @@ calls = [ "https.request", ] rule_ids = [] +prefix_fixes_origin = true [[sink]] id = "redirect" cwe = "CWE-601" calls = ["redirect", "location.assign", "location.replace"] rule_ids = [] +prefix_fixes_origin = true # --- model-grounded-detection: sanitizers ----------------------------------------------------------------- # This file shipped with no [[sanitizer]] entries at all, so no JavaScript fix of any family could be diff --git a/src/openultrasast/ruleset/semantic/python.toml b/src/openultrasast/ruleset/semantic/python.toml index 34038f6..ee30059 100644 --- a/src/openultrasast/ruleset/semantic/python.toml +++ b/src/openultrasast/ruleset/semantic/python.toml @@ -16,6 +16,23 @@ patterns = [ "request.get", ] +# Django and Django REST framework. The source vocabulary above is Flask's alone, so on a Django project no +# flow could begin anywhere: OpenCVE's notification test POSTs to a URL read from `form.cleaned_data`, and the +# scan answered the file and found nothing. `cleaned_data` is what a form bound to the request hands back -- +# validated for type, not for trust. +[[source]] +id = "django-request" +patterns = [ + "request.GET", + "request.POST", + "request.body", + "request.COOKIES", + "request.FILES", + "request.META", + "request.query_params", + "cleaned_data", +] + [[source]] id = "argv" patterns = ["sys.argv"] @@ -111,9 +128,13 @@ id = "outbound.request" cwe = "CWE-918" calls = ["requests.get", "requests.post", "requests.request", "urlopen", "urlretrieve", "httpx.get", "httpx.post"] rule_ids = [] +prefix_fixes_origin = true [[sink]] id = "redirect" cwe = "CWE-601" calls = ["redirect", "HttpResponseRedirect", "url_for"] rule_ids = [] +# Django's `reverse` builds an in-site path from a route NAME; request data can only fill its path segments. +prefix_fixes_origin = true +origin_anchors = ["reverse", "reverse_lazy"] diff --git a/src/openultrasast/semantic/facts.py b/src/openultrasast/semantic/facts.py index 241cb9b..c55a7eb 100644 --- a/src/openultrasast/semantic/facts.py +++ b/src/openultrasast/semantic/facts.py @@ -42,6 +42,12 @@ class SinkFact: language: str weak_literals: tuple[str, ...] = () format_arg: int | None = None + # A DESTINATION sink (a redirect, an outbound request) is dangerous only where the value can choose where + # it goes. With `prefix_fixes_origin`, a value appended after a prefix that already fixes the origin -- a + # path `/x`, a `?query`, `scheme://host/`, or an `origin_anchors` call such as Django's `reverse()` -- is + # not an untrusted destination, and neither is one that only fills an anchor call's arguments. + prefix_fixes_origin: bool = False + origin_anchors: tuple[str, ...] = () @dataclass(frozen=True) @@ -263,6 +269,14 @@ def _sources(value: object, language: str, path: Path) -> list[SourceFact]: return rows +def _flag(value: object, field: str, path: Path, ident: str) -> bool: + if value is None: + return False + if not isinstance(value, bool): + raise FactLoadError(f"{path} sink {ident} has a non-boolean {field}") + return value + + def _sinks(value: object, language: str, path: Path) -> list[SinkFact]: rows: list[SinkFact] = [] for item in _items(value): @@ -287,6 +301,8 @@ def _sinks(value: object, language: str, path: Path) -> list[SinkFact]: language=language, weak_literals=_strings(item.get("weak_literals"), "weak_literals", path), format_arg=format_arg if isinstance(format_arg, int) else None, + prefix_fixes_origin=_flag(item.get("prefix_fixes_origin"), "prefix_fixes_origin", path, ident), + origin_anchors=_strings(item.get("origin_anchors"), "origin_anchors", path), ) ) return rows diff --git a/tests/test_cpg_queries_compile.py b/tests/test_cpg_queries_compile.py index 0a4b421..c8f5d85 100644 --- a/tests/test_cpg_queries_compile.py +++ b/tests/test_cpg_queries_compile.py @@ -1018,3 +1018,77 @@ def test_node_vm_is_an_eval_and_a_package_name_is_not_a_sink(tmp_path: Path) -> assert any("runInNewContext" in str(r["sink"]) for r in flows["vulnerable"]), flows["vulnerable"] assert not flows["parsed"], f"a package name was read as a sink: {flows['parsed']}" assert any("query" in str(r["sink"]) for r in flows["raw"]), f"a real query call was lost: {flows['raw']}" + + +@pytest.mark.skipif(_joern() is None, reason="joern is not installed on this machine") +def test_a_destination_whose_origin_is_fixed_is_not_untrusted(tmp_path: Path) -> None: + """OpenCVE redirects to `reverse("cves") + "?" + request.GET.urlencode()` and to `reverse(route, kwargs=...)`: + the request picks a query string or a path segment, never the site. Both were reported as open redirects + once Django's sources were known. A prefix that does NOT fix the origin -- a scheme with the host still to + come, or no prefix at all -- must still be reported. The shipped spec supplies sinks, anchors and flag.""" + from openultrasast.cpg.backend import extract_payload + from openultrasast.model.specs import taint_specs + + frontend = __import__("shutil").which("pysrc2cpg") + if frontend is None: + pytest.skip("pysrc2cpg is not installed") + spec = taint_specs(language="python")["untrusted_destination"] + assert spec.fixed_origin and "reverse" in spec.origin_anchors, "the shipped redirect fact no longer fixes origins" + src = tmp_path / "src" + src.mkdir() + (src / "views.py").write_text( + "from django.shortcuts import redirect\nfrom django.urls import reverse\n\n" + "def appended(request):\n" + ' url = reverse("cves") + ("?" + request.GET.urlencode() if request.GET else "")\n' + " return redirect(url)\n\n" + "def augmented(request):\n" + ' url = reverse("x")\n' + " if request.GET:\n" + ' url += "&" + request.GET.urlencode()\n' + " return redirect(url)\n\n" + "def formatted(request):\n" + " return redirect(f\"/items?{request.GET.get('q')}\")\n\n" + "def anchored(request):\n" + ' return redirect(reverse("m", kwargs={"n": request.POST.get("n")}))\n\n' + "def host(request):\n" + ' return redirect("https://" + request.GET.get("host") + "/x")\n\n' + "def bare(request):\n" + ' return redirect(request.GET.get("next"))\n\n' + "def slash(request):\n" + ' return redirect("/" + request.GET.get("path"))\n' + ) + cpg = tmp_path / "cpg.bin" + built = subprocess.run([frontend, str(src), "-o", str(cpg)], capture_output=True, text=True, timeout=600, check=False) + if not cpg.is_file(): + pytest.skip(f"could not build a sample cpg: {(built.stderr or '')[-200:]}") + common = { + "sources": ",".join(spec.sources), + "sinks": ",".join(spec.sinks), + "file": "views.py", + "fixedOrigin": "true", + "originAnchors": ",".join(spec.origin_anchors), + } + names = ("appended", "augmented", "formatted", "anchored", "host", "bare", "slash") + requests = tmp_path / "requests.json" + requests.write_text(json.dumps({name: {**common, "function": name} for name in names})) + done = subprocess.run( + [ + _joern() or "joern", + "--script", + str((QUERIES / "taint.sc").resolve()), + "--param", + f"cpgFile={cpg}", + "--param", + f"requestsFile={requests}", + ], + capture_output=True, + text=True, + timeout=900, + check=False, + cwd=str(tmp_path), + ) + assert done.returncode == 0, (done.stderr or done.stdout or "")[-500:] + payload = extract_payload(done.stdout or "") + assert isinstance(payload, dict) + reported = {name for name in names if any(isinstance(r, dict) and r.get("sink") and r.get("sourceKind") for r in payload[name])} + assert reported == {"host", "bare", "slash"}, {name: payload[name] for name in names}