diff --git a/src/openultrasast/cpg/queries/taint.sc b/src/openultrasast/cpg/queries/taint.sc index 9fdeffd..842736a 100644 --- a/src/openultrasast/cpg/queries/taint.sc +++ b/src/openultrasast/cpg/queries/taint.sc @@ -107,7 +107,8 @@ traceS: String = "", fixedOriginS: String = "", originAnchorsS: String = "", - quotedSanitizersS: String = "" + quotedSanitizersS: String = "", + guardsS: String = "" ): List[ujson.Obj] = { // Word-boundary matching, never substring. `resolveUrl` contains `resolve`, so a bare-substring sanitizer @@ -139,6 +140,7 @@ val strictNames = sanitizerNames.filterNot(quotedOnly.contains) val quotedNames = sanitizerNames.filter(quotedOnly.contains) val stringBuilders = Set(".concat", "encaps", ".addition", ".formatString") + val guardNames = split(guardsS).toSet val boundedNames = split(boundedS) // A relational comparison against an integer literal: `len > 250`, `n <= sizeof(buf)`. Equality and null @@ -688,8 +690,122 @@ } } + // A GUARD is a check, not a transformation, so it is judged by WHERE the flow is relative to it: + // * inside the branch its condition selects -- `if next_url and url_has_allowed_host_and_scheme(next_url)` + // around the redirect, `elseif (in_array($sortby, $allowed, true)) { ... }` around the query; + // * past it, when its failure exits or overwrites the value -- `if not ...(next_url): next_url = reverse(..)` + // -- and the check dominates the end of the flow; + // * through the arm of a ternary it selects -- `in_array(strtoupper($o), ['ASC','DESC'], true) ? $o : 'ASC'`. + // The guard must test a variable the path itself carries. Polarity is read off the condition: a positive + // guard sits under nothing but `&&`; a failure test is exactly one `!` under nothing but `||`. + val exitCalls = Set("exit", "die", "wp_die", "abort", ".throw") + + def carried(elements: List[io.shiftleft.codepropertygraph.generated.nodes.AstNode]): Set[String] = + elements.flatMap { + case i: io.shiftleft.codepropertygraph.generated.nodes.Identifier => List(i.name, i.code.trim) + case c: io.shiftleft.codepropertygraph.generated.nodes.Call if c.name == FIELD_ACCESS => List(c.code.trim) + case _ => Nil + }.toSet + + def guardsIn(root: io.shiftleft.codepropertygraph.generated.nodes.AstNode, names: Set[String]): List[io.shiftleft.codepropertygraph.generated.nodes.Call] = + if (guardNames.isEmpty) Nil + else + root.ast.isCall.l.filter(g => guardNames.contains(g.name)).filter { g => + g.argument.l.exists(_.ast.exists { + case i: io.shiftleft.codepropertygraph.generated.nodes.Identifier => names(i.name) || names(i.code.trim) + case c: io.shiftleft.codepropertygraph.generated.nodes.Call if c.name == FIELD_ACCESS => names(c.code.trim) + case _ => false + }) + } + + // Some(true): the guard holds wherever the condition does. Some(false): the condition says it FAILED. + def polarity(g: io.shiftleft.codepropertygraph.generated.nodes.AstNode, root: io.shiftleft.codepropertygraph.generated.nodes.AstNode): Option[Boolean] = { + var node = g + var nots, ands, ors = 0 + var other = false + while (node.id != root.id && !other) { + node._astIn.nextOption() match { + case Some(parent: io.shiftleft.codepropertygraph.generated.nodes.Call) => + parent.name match { + case ".logicalNot" => nots += 1 + case ".logicalAnd" => ands += 1 + case ".logicalOr" => ors += 1 + case _ => other = true + } + node = parent + case Some(parent: io.shiftleft.codepropertygraph.generated.nodes.AstNode) if parent.id == root.id => node = parent + case _ => other = true + } + } + if (other) None + else if (nots == 0 && ors == 0) Some(true) + else if (nots == 1 && ands == 0) Some(false) + else None + } + + def branches(cs: io.shiftleft.codepropertygraph.generated.nodes.ControlStructure) = { + val children = cs.astChildren.l.sortBy(_.order) + (children.find(_.order == 2), children.find(_.order == 3)) + } + + def contains(root: io.shiftleft.codepropertygraph.generated.nodes.AstNode, node: io.shiftleft.codepropertygraph.generated.nodes.AstNode) = + root.id == node.id || root.ast.exists(_.id == node.id) + + def guarded(elements: List[io.shiftleft.codepropertygraph.generated.nodes.AstNode]): Boolean = + guardNames.nonEmpty && elements.nonEmpty && { + val names = carried(elements) + val end = elements.last + val ternary = elements.indices.exists { k => + elements(k) match { + // The engine may enter through the CONDITION's own copy of the value (`in_array(strtoupper($o), ..)`), + // which never becomes the result. What the ternary can return is its arms; with the check holding on + // the carried value and the other arm unable to carry it, the result is the checked value or a constant. + case c: io.shiftleft.codepropertygraph.generated.nodes.Call if k > 0 && c.name == ".conditional" => + val args = c.argument.l + def carries(arm: io.shiftleft.codepropertygraph.generated.nodes.AstNode) = arm.ast.exists { + case i: io.shiftleft.codepropertygraph.generated.nodes.Identifier => names(i.name) || names(i.code.trim) + case f: io.shiftleft.codepropertygraph.generated.nodes.Call if f.name == FIELD_ACCESS => names(f.code.trim) + case _ => false + } + argumentHolding(c, elements(k - 1)).exists(a => a.argumentIndex == 1 || a.argumentIndex == 2) && + args.find(_.argumentIndex == 1).exists(cond => guardsIn(cond, names).exists(g => polarity(g, cond).contains(true))) && + !args.find(_.argumentIndex == 3).exists(carries) + case _ => false + } + } + ternary || (end match { + case cfg: io.shiftleft.codepropertygraph.generated.nodes.CfgNode => + val method = cfg.method + method.ast.isControlStructure.controlStructureType("IF").l.exists { cs => + cs.condition.l.headOption.exists { cond => + val checks = guardsIn(cond, names) + checks.nonEmpty && { + val (whenTrue, _) = branches(cs) + val inside = whenTrue.exists(b => contains(b, end)) && checks.exists(g => polarity(g, cond).contains(true)) + inside || (checks.exists(g => polarity(g, cond).contains(false)) && whenTrue.exists { b => + val bails = b.ast.exists { + case _: io.shiftleft.codepropertygraph.generated.nodes.Return => true + case c: io.shiftleft.codepropertygraph.generated.nodes.Call => exitCalls.contains(c.name) + case _ => false + } + val overwrites = b.ast.isCall.nameExact(ASSIGNMENT).l.exists(_.argument.l.find(_.argumentIndex == 1).exists { + case i: io.shiftleft.codepropertygraph.generated.nodes.Identifier => names(i.name) || names(i.code.trim) + case _ => false + }) + (bails || overwrites) && !contains(cs, end) && (cond match { + case c: io.shiftleft.codepropertygraph.generated.nodes.CfgNode => cfg.dominatedBy.exists(_.id == c.id) + case _ => false + }) + }) + } + } + } + case _ => false + }) + } + def cleansed(elements: List[io.shiftleft.codepropertygraph.generated.nodes.AstNode]): Boolean = - elements.exists(node => sanitizesHere(node, strictNames)) || sanitizedBetween(elements, strictNames) || + guarded(elements) || elements.exists(node => sanitizesHere(node, strictNames)) || sanitizedBetween(elements, strictNames) || (quotedNames.nonEmpty && elements.indices.exists { k => (sanitizesHere(elements(k), quotedNames) || (k + 1 < elements.size && wrappedBetween(elements(k), elements(k + 1), quotedNames))) && landsQuoted(elements, k + 1) @@ -1278,7 +1394,7 @@ .map { flow => val path = spliced(flow.elements.l) val elements = path.map(_.code) - val sanitized = sanitizerNames.nonEmpty && cleansed(path) + val sanitized = (sanitizerNames.nonEmpty || guardNames.nonEmpty) && cleansed(path) val sourceKind = sourceKindOf(path.headOption) ((sourceKind, elements.headOption.getOrElse("").take(200), sanitized), (elements.size, flow)) } @@ -1387,7 +1503,8 @@ field("trace"), field("fixedOrigin"), field("originAnchors"), - field("quotedSanitizers") + field("quotedSanitizers"), + field("guards") ): _* ) 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 52d7fdb..841b3df 100644 --- a/src/openultrasast/model/specs.py +++ b/src/openultrasast/model/specs.py @@ -124,6 +124,8 @@ class TaintSpec: origin_anchors: tuple[str, ...] = () # The sanitizers that protect only a value inside a quoted literal (`esc_sql`); a subset of `sanitizers`. quoted_sanitizers: tuple[str, ...] = () + # Checks that make a value safe where they held (see `SanitizerFact.guard`); never in `sanitizers`. + guards: tuple[str, ...] = () @dataclass(frozen=True) @@ -257,7 +259,8 @@ def taint_specs( {f"{call}:{fact.value_arg if call in fact.returns else 0}" for fact in scoped.dispatches if fact.returns for call in fact.apply} ) ) - sanitizers = tuple(sorted({call for fact in scoped.sanitizers if not _is_shape(fact) for call in fact.calls})) + sanitizers = tuple(sorted({call for fact in scoped.sanitizers if not _is_shape(fact) and not fact.guard for call in fact.calls})) + guards = tuple(sorted({call for fact in scoped.sanitizers if fact.guard for call in fact.calls})) safe_shapes = tuple(sorted({call for fact in scoped.sanitizers if _is_shape(fact) for call in fact.calls})) quoted = tuple(sorted({call for fact in scoped.sanitizers if not _is_shape(fact) and fact.quoted_only for call in fact.calls})) by_family: dict[str, set[str]] = {} @@ -290,6 +293,7 @@ def taint_specs( fixed_origin=family_id in fixed_origin, origin_anchors=tuple(sorted(anchors.get(family_id, ()))), quoted_sanitizers=quoted, + guards=guards, ) for family_id, calls in sorted(by_family.items()) } diff --git a/src/openultrasast/model/taint.py b/src/openultrasast/model/taint.py index 20fa0b9..f4214e4 100644 --- a/src/openultrasast/model/taint.py +++ b/src/openultrasast/model/taint.py @@ -63,6 +63,7 @@ def request_params( "fixedOrigin": "true" if spec.fixed_origin else "false", "originAnchors": spec.origin_anchors, "quotedSanitizers": spec.quoted_sanitizers, + "guards": spec.guards, } diff --git a/src/openultrasast/ruleset/semantic/php.toml b/src/openultrasast/ruleset/semantic/php.toml index 66c98c4..f9988e7 100644 --- a/src/openultrasast/ruleset/semantic/php.toml +++ b/src/openultrasast/ruleset/semantic/php.toml @@ -182,6 +182,15 @@ calls = ["pg_escape_literal"] # into a query. Adding them would make that CVE, and its whole class, invisible. The sanitizer list is # currently flat across families, which cannot express "escapes for HTML but not for SQL"; until it can, only # SQL-specific cleansers belong here. +# An allowlist CHECK. Ultimate Member's fix for CVE-2024-1071 is exactly this, twice over: +# `$order = in_array( strtoupper( $order ), array( 'ASC', 'DESC' ), true ) ? $order : 'ASC'` and +# `elseif ( in_array( $sortby, $this->core_search_fields, true ) ) { ... }`. A guard, not a cleanser -- see +# `guard` in semantic/facts.py. +[[sanitizer]] +id = "allowlist" +calls = ["in_array"] +guard = true + [[sanitizer]] id = "wpdb_prepare" calls = ["$wpdb->prepare"] diff --git a/src/openultrasast/ruleset/semantic/python.toml b/src/openultrasast/ruleset/semantic/python.toml index ee30059..4a07f87 100644 --- a/src/openultrasast/ruleset/semantic/python.toml +++ b/src/openultrasast/ruleset/semantic/python.toml @@ -138,3 +138,10 @@ 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"] + +# Django's documented redirect-target check (`is_safe_url` is its pre-3.0 name). wger's fix for its open +# redirect is `if not url_has_allowed_host_and_scheme(next_url, ...): next_url = reverse(...)`. +[[sanitizer]] +id = "url_allowlist" +calls = ["url_has_allowed_host_and_scheme", "is_safe_url"] +guard = true diff --git a/src/openultrasast/semantic/facts.py b/src/openultrasast/semantic/facts.py index 436bb9b..f116a82 100644 --- a/src/openultrasast/semantic/facts.py +++ b/src/openultrasast/semantic/facts.py @@ -60,6 +60,10 @@ class SanitizerFact: # An ESCAPE, not a cleansing: `esc_sql` backslashes quotes, which protects a value inside a quoted SQL # literal and nothing else. `ORDER BY u.{$sortby}` is Ultimate Member's CVE-2024-1071, escaped and injectable. quoted_only: bool = False + # A CHECK, not a transformation: `in_array($v, $allowed, true)`, `url_has_allowed_host_and_scheme(v)`. It + # makes a value safe only where it held -- inside the branch it guards, past an exit on its failure, or in + # the arm of a ternary it selects. Never a cleansing call on a path. + guard: bool = False @dataclass(frozen=True) @@ -331,6 +335,7 @@ def _sanitizers(value: object, language: str, path: Path) -> list[SanitizerFact] parameterized=bool(item.get("parameterized", False)), literal_format_arg=literal if isinstance(literal, int) else None, quoted_only=_flag(item.get("quoted_only"), "quoted_only", path, ident), + guard=_flag(item.get("guard"), "guard", path, ident), ) ) return rows diff --git a/tests/test_cpg_queries_compile.py b/tests/test_cpg_queries_compile.py index 4a33859..ba9052a 100644 --- a/tests/test_cpg_queries_compile.py +++ b/tests/test_cpg_queries_compile.py @@ -1281,3 +1281,125 @@ def test_a_source_pattern_is_a_token_not_a_prefix(tmp_path: Path) -> None: flows = {name: [r for r in payload[name] if isinstance(r, dict) and r.get("sourceKind")] for name in ("host", "parameter")} assert not flows["host"], f"request.get_host() was read as request.get: {flows['host']}" assert flows["parameter"], "the request parameter is no longer a source" + + +def _taint_rows(tmp_path: Path, frontend_name: str, files: dict[str, str], requests: dict[str, dict]) -> dict: + from openultrasast.cpg.backend import extract_payload + + frontend = __import__("shutil").which(frontend_name) + if frontend is None: + pytest.skip(f"{frontend_name} is not installed") + src = tmp_path / "src" + src.mkdir() + for name, text in files.items(): + (src / name).write_text(text) + 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:]}") + path = tmp_path / "requests.json" + path.write_text(json.dumps(requests)) + done = subprocess.run( + [ + _joern() or "joern", + "--script", + str((QUERIES / "taint.sc").resolve()), + "--param", + f"cpgFile={cpg}", + "--param", + f"requestsFile={path}", + ], + 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) + return payload + + +def _unsanitized(payload: dict, name: str) -> list: + return [r for r in payload[name] if isinstance(r, dict) and r.get("sourceKind") and not r.get("sanitized")] + + +@pytest.mark.skipif(_joern() is None, reason="joern is not installed on this machine") +def test_a_redirect_validator_guards_only_where_it_held(tmp_path: Path) -> None: + """wger's fix: `if not url_has_allowed_host_and_scheme(next_url, ...): next_url = reverse(...)`, and + `if next_url and url_has_allowed_host_and_scheme(next_url, ...): return HttpResponseRedirect(next_url)`. + The check guards where it held -- not a different variable, and not a failure that is only logged.""" + from openultrasast.model.specs import taint_specs + + spec = taint_specs(language="python")["untrusted_destination"] + assert "url_has_allowed_host_and_scheme" in spec.guards + views = ( + "from django.http import HttpResponseRedirect\n" + "from django.utils.http import url_has_allowed_host_and_scheme\n\n" + "def overwritten(request):\n" + " n = request.GET.get('next', '/')\n" + " if not url_has_allowed_host_and_scheme(n, allowed_hosts=None):\n" + " n = '/home'\n" + " return HttpResponseRedirect(n)\n\n" + "def branch(request):\n" + " n = request.GET.get('next')\n" + " if n and url_has_allowed_host_and_scheme(n, allowed_hosts=None):\n" + " return HttpResponseRedirect(n)\n" + " return HttpResponseRedirect('/')\n\n" + "def unguarded(request):\n" + " return HttpResponseRedirect(request.GET.get('next'))\n\n" + "def other_variable(request):\n" + " n = request.GET.get('next')\n" + " m = request.GET.get('back')\n" + " if url_has_allowed_host_and_scheme(m, allowed_hosts=None):\n" + " return HttpResponseRedirect(n)\n" + " return HttpResponseRedirect('/')\n\n" + "def only_logged(request):\n" + " n = request.GET.get('next')\n" + " if not url_has_allowed_host_and_scheme(n, allowed_hosts=None):\n" + " print('suspicious')\n" + " return HttpResponseRedirect(n)\n" + ) + names = ("overwritten", "branch", "unguarded", "other_variable", "only_logged") + common = {"sources": ",".join(spec.sources), "sinks": ",".join(spec.sinks), "guards": ",".join(spec.guards), "file": "views.py"} + payload = _taint_rows(tmp_path, "pysrc2cpg", {"views.py": views}, {n: {**common, "function": n} for n in names}) + reported = {n for n in names if _unsanitized(payload, n)} + assert reported == {"unguarded", "other_variable", "only_logged"}, {n: payload[n] for n in names} + + +@pytest.mark.skipif(_joern() is None, reason="joern is not installed on this machine") +def test_an_allowlist_check_guards_its_branch_and_its_ternary_arm(tmp_path: Path) -> None: + """Ultimate Member's CVE-2024-1071 fix: `in_array( strtoupper( $order ), array( 'ASC', 'DESC' ), true ) ? + $order : 'ASC'` and `elseif ( in_array( $sortby, $core, true ) ) { ... }`. The unguarded `else`, and the + arm of a ternary the check does NOT select, stay reported.""" + from openultrasast.model.specs import taint_specs + + spec = taint_specs(language="php")["injection"] + assert "in_array" in spec.guards and "in_array" not in spec.sanitizers + app = ( + "query( "SELECT 1 FROM t ORDER BY a {$o}" ); }\n' + "function wrong_arm() { global $wpdb; $o = $_POST['o'];\n" + " $o = in_array( $o, array( 'ASC', 'DESC' ), true ) ? 'ASC' : $o;\n" + ' return $wpdb->query( "SELECT 1 FROM t ORDER BY a {$o}" ); }\n' + "function branch() { global $wpdb; $s = $_POST['s']; $core = array( 'login' );\n" + ' if ( in_array( $s, $core, true ) ) { return $wpdb->query( "SELECT 1 FROM t ORDER BY u.{$s}" ); }\n' + " return null; }\n" + "function else_branch() { global $wpdb; $s = $_POST['s']; $core = array( 'login' );\n" + " if ( in_array( $s, $core, true ) ) { return null; }\n" + ' else { return $wpdb->query( "SELECT 1 FROM t ORDER BY u.{$s}" ); } }\n' + ) + names = ("ternary", "wrong_arm", "branch", "else_branch") + common = { + "sources": ",".join(spec.sources), + "sinks": ",".join(spec.sinks), + "sanitizers": ",".join(spec.sanitizers), + "guards": ",".join(spec.guards), + "file": "app.php", + } + payload = _taint_rows(tmp_path, "php2cpg", {"app.php": app}, {n: {**common, "function": n} for n in names}) + reported = {n for n in names if _unsanitized(payload, n)} + assert reported == {"wrong_arm", "else_branch"}, {n: payload[n] for n in names}