"""model-grounded-detection task 2.2: the taint verdict, and what separates entailed from corroborated. An `ENTAILED` verdict is the static analogue of a reproduced crash: the CPG establishes the flow on its own and the LLM's assertion is no longer load-bearing. `CORROBORATED` means a path exists but a sanitizer may break it, so the judgement of sufficiency is the LLM's residual — checked against the graph, never averaged over runs. """ from __future__ import annotations from pathlib import Path def _cpg(rows: list[dict[str, object]]): # type: ignore[no-untyped-def] from openultrasast.cpg.backend import CpgResult return CpgResult(cpg_path=Path("cpg.bin"), run=lambda query, params: rows) def _spec(): # type: ignore[no-untyped-def] from openultrasast.model.specs import TaintSpec return TaintSpec( family="injection", language="python", sources=("request.args",), sinks=("os.system", "execute"), sanitizers=("escape", "quote"), ) def test_an_unsanitized_source_to_sink_path_is_entailed() -> None: from openultrasast.model.ladder import Rung from openultrasast.model.taint import verdict rows = [ {"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['x']", "sanitized": False, "length": 3} ] answer = verdict(_cpg(rows), _spec(), function="run") assert answer is not None and answer.rung is Rung.ENTAILED assert answer.family == "injection" assert "request.args" in answer.witness and "os.system" in answer.witness, "the witness names the flow it found" def test_a_path_through_a_sanitizer_is_corroborated_not_entailed() -> None: """The model saw the flow but cannot judge whether the sanitizer is sufficient. That residual is the LLM's.""" from openultrasast.model.ladder import Rung from openultrasast.model.taint import verdict rows = [{"sink": "execute(q)", "sinkLine": "9", "sinkMethod": "run", "source": "request.args['q']", "sanitized": True, "length": 4}] answer = verdict(_cpg(rows), _spec(), function="run") assert answer is not None and answer.rung is Rung.CORROBORATED def test_no_path_yields_no_verdict_so_the_caller_reports_suspicion() -> None: from openultrasast.model.taint import verdict assert verdict(_cpg([]), _spec(), function="run") is None def test_a_query_that_failed_is_not_read_as_an_absence_of_flow() -> None: """`None` from the engine means "could not decide", which is not the same as "no path" — Req 4.3.""" from openultrasast.cpg.backend import CpgResult from openultrasast.model.taint import verdict broken = CpgResult(cpg_path=Path("cpg.bin"), run=lambda query, params: None) assert verdict(broken, _spec(), function="run") is None def test_the_verdict_is_restricted_to_the_labeled_function() -> None: from openultrasast.model.taint import verdict rows = [ { "sink": "os.system(x)", "sinkLine": "40", "sinkMethod": "elsewhere", "source": "request.args['x']", "sanitized": False, "length": 2, } ] assert verdict(_cpg(rows), _spec(), function="run") is None, "a flow in another function does not judge this one" def test_the_shortest_unsanitized_flow_wins_and_the_sequence_is_deterministic() -> None: """Req 6.4: two runs over one CPG give an identical verdict sequence, so there is nothing to average.""" from openultrasast.model.ladder import Rung from openultrasast.model.taint import verdict rows = [ {"sink": "os.system(a)", "sinkLine": "7", "sinkMethod": "run", "source": "request.args['b']", "sanitized": True, "length": 9}, {"sink": "os.system(a)", "sinkLine": "7", "sinkMethod": "run", "source": "request.args['a']", "sanitized": False, "length": 2}, ] first = verdict(_cpg(rows), _spec(), function="run") second = verdict(_cpg(rows), _spec(), function="run") assert first == second assert first is not None and first.rung is Rung.ENTAILED and "request.args['a']" in first.witness def test_the_spec_drives_the_query_parameters() -> None: """The security modelling is ours; the traversal is the engine's. The spec is what crosses the seam.""" from openultrasast.cpg.backend import CpgResult from openultrasast.model.taint import verdict seen: dict[str, object] = {} def run(query: str, params: dict[str, object]) -> object: seen["query"] = query seen.update(params) return [] verdict(CpgResult(cpg_path=Path("cpg.bin"), run=run), _spec(), function="run") assert seen["query"] == "taint" assert seen["sources"] == ("request.args",) assert seen["sinks"] == ("os.system", "execute") assert seen["sanitizers"] == ("escape", "quote") assert seen["function"] == "run" # --- the safe-shape test (task 2.3 finding) ------------------------------------------------------------ # # Taint reachability alone does not distinguish a parameterised fix from an interpolated bug: the flow # `request.args["name"] -> name -> execute` exists on BOTH sides of the canonical SQL pair. # # vuln db.execute("select ... '" + name + "'") one argument, interpolated # fixed db.execute("select ... = %s", (name,)) two arguments, bound # # What separates them is the SHAPE of the sink call, which is exactly what `TaintSpec.safe_shape_sinks` # preserves. Without this test the model entails both sides and arbitrates nothing. def _shape_spec(): # type: ignore[no-untyped-def] from openultrasast.model.specs import TaintSpec return TaintSpec( family="injection", language="python", sources=("request.args",), sinks=("execute",), sanitizers=(), safe_shape_sinks=("execute",), ) def test_a_parameterised_sink_call_is_not_a_vulnerability_even_though_the_flow_reaches_it() -> None: from openultrasast.model.taint import verdict rows = [ { "sink": 'db.execute("select ... = %s", (name,))', "sinkLine": "5", "sinkMethod": "run", "source": "request.args['name']", "sanitized": False, "length": 4, "sinkArity": 2, "sinkArg0Literal": True, } ] assert verdict(_cpg(rows), _shape_spec(), function="run") is None, "a bound query is the fix, not the bug" def test_an_interpolated_sink_call_of_the_same_name_is_entailed() -> None: from openultrasast.model.ladder import Rung from openultrasast.model.taint import verdict rows = [ { "sink": 'db.execute("select ... \'" + name + "\'")', "sinkLine": "5", "sinkMethod": "run", "source": "request.args['name']", "sanitized": False, "length": 4, "sinkArity": 1, "sinkArg0Literal": False, } ] answer = verdict(_cpg(rows), _shape_spec(), function="run") assert answer is not None and answer.rung is Rung.ENTAILED def test_the_shape_test_only_applies_to_sinks_declared_safe_in_that_shape() -> None: """`os.system(x)` with one argument is not "safe because it has one argument" -- the rule is per sink.""" from openultrasast.model.ladder import Rung from openultrasast.model.taint import verdict rows = [ { "sink": "os.system(full)", "sinkLine": "7", "sinkMethod": "run", "source": "request.args['cmd']", "sanitized": False, "length": 5, "sinkArity": 1, "sinkArg0Literal": False, } ] answer = verdict(_cpg(rows), _spec(), function="run") # os.system is not in safe_shape_sinks assert answer is not None and answer.rung is Rung.ENTAILED def test_a_row_without_shape_fields_is_judged_on_the_flow_alone() -> None: """Older rows, or an engine that could not report arity, must not be silently treated as safe.""" from openultrasast.model.ladder import Rung from openultrasast.model.taint import verdict rows = [{"sink": "db.execute(q)", "sinkLine": "5", "sinkMethod": "run", "source": "request.args['n']", "sanitized": False, "length": 3}] answer = verdict(_cpg(rows), _shape_spec(), function="run") assert answer is not None and answer.rung is Rung.ENTAILED def test_parameter_sources_are_opt_in_and_reach_the_query() -> None: """A function-level pair's parameters are its trust boundary; a whole repository's are not. 39 of the 50 injection pairs carry no framework source token at all — their untrusted input arrives as a function parameter — and the flat-IR baseline this feature is measured against counted those as sources. So the flag must exist and must default to off. """ from openultrasast.cpg.backend import CpgResult from openultrasast.model.taint import verdict seen: dict[str, object] = {} def run(query: str, params: dict[str, object]) -> object: seen.update(params) return [] verdict(CpgResult(cpg_path=Path("c.bin"), run=run), _spec(), function="run") assert seen["parameterSources"] == "false", "treating every parameter as untrusted is not the default" verdict(CpgResult(cpg_path=Path("c.bin"), run=run), _spec(), function="run", parameter_sources=True) assert seen["parameterSources"] == "true" # --- closure scoping (group 4 finding) ----------------------------------------------------------------- # # JavaScript is callback-heavy, so the method containing a sink routinely is not the labeled function: # # function requestListener(req, res) { // labeled # fs.stat(possibleFilename, function(err, stats) { # fs.readFileSync(possibleFilename) // sink, enclosing method is 0 # }) # } # # Filtering on `sinkMethod == function` dropped every such sink -- the whole `path` family measured 0/8. # The query now decides scope by line-range containment and reports it; the Python side trusts that flag and # keeps the name comparison only as a fallback for rows that carry no flag. def test_a_sink_in_a_nested_closure_is_in_scope_when_the_query_says_so() -> None: from openultrasast.model.ladder import Rung from openultrasast.model.taint import verdict rows = [ { "sink": "fs.readFileSync(possibleFilename)", "sinkLine": "6", "sinkMethod": "0", "source": "req", "sanitized": False, "length": 4, "inLabeledScope": True, } ] answer = verdict(_cpg(rows), _spec(), function="requestListener") assert answer is not None and answer.rung is Rung.ENTAILED assert "0" not in answer.witness, "the witness names the flow, not the synthetic closure name" def test_a_sink_in_an_unrelated_function_stays_out_of_scope() -> None: from openultrasast.model.taint import verdict rows = [ { "sink": "fs.readFileSync('/etc/' + name)", "sinkLine": "11", "sinkMethod": "unrelated", "source": "name", "sanitized": False, "length": 2, "inLabeledScope": False, } ] assert verdict(_cpg(rows), _spec(), function="requestListener") is None def test_rows_without_the_scope_flag_fall_back_to_the_method_name() -> None: """A row from an engine that cannot report scope must not silently become in-scope.""" from openultrasast.model.taint import verdict rows = [ {"sink": "os.system(x)", "sinkLine": "9", "sinkMethod": "0", "source": "request.args['x']", "sanitized": False, "length": 2} ] assert verdict(_cpg(rows), _spec(), function="run") is None def test_a_flow_from_a_real_source_outranks_a_shorter_one_from_a_bare_parameter() -> None: """Shortest-wins alone picks the wrong witness once closures are in scope. In the angular-http-server shape the closure yields three unsanitized flows to one sink: `req.url.split('?')[0]` (11 steps, the actual vulnerability), `res` (3 steps) and `this` (7). Shortest would report `res -> fs.readFileSync(...)`, which names no attacker input and is simply untrue. A flow whose source matches the spec's source model outranks one from a bare captured name; shortest still breaks ties, so the order stays total and the verdict deterministic. """ from openultrasast.model.ladder import Rung from openultrasast.model.taint import verdict rows = [ { "sink": "fs.readFileSync(p)", "sinkLine": "6", "sinkMethod": "0", "source": "res", "sanitized": False, "length": 3, "inLabeledScope": True, }, { "sink": "fs.readFileSync(p)", "sinkLine": "6", "sinkMethod": "0", "source": "this", "sanitized": False, "length": 7, "inLabeledScope": True, }, { "sink": "fs.readFileSync(p)", "sinkLine": "6", "sinkMethod": "0", "source": "request.args['f']", "sanitized": False, "length": 11, "inLabeledScope": True, }, ] answer = verdict(_cpg(rows), _spec(), function="requestListener") assert answer is not None and answer.rung is Rung.ENTAILED assert "request.args['f']" in answer.witness, f"witness named no real source: {answer.witness}" assert verdict(_cpg(rows), _spec(), function="requestListener") == answer # still deterministic def test_a_modelled_source_flow_decides_even_when_a_parameter_flow_is_unsanitized() -> None: """Parameter sources are a fallback, never an override. On the fixed angular-http-server side the real flow `req.url -> ... -> readFileSync` IS sanitized by path.normalize, but the captured `res` parameter also reaches the sink unsanitized. Judging on the union reports the fix as still vulnerable, on the strength of a flow that names no attacker input. When any modelled source reaches the sink, those flows are the evidence and the bare-parameter ones are ignored. """ from openultrasast.model.ladder import Rung from openultrasast.model.taint import verdict rows = [ { "sink": "fs.readFileSync(safe)", "sinkLine": "41", "sinkMethod": "0", "source": "res", "sanitized": False, "length": 3, "inLabeledScope": True, }, { "sink": "fs.readFileSync(safe)", "sinkLine": "41", "sinkMethod": "0", "source": "request.args['f']", "sanitized": True, "length": 22, "inLabeledScope": True, }, ] answer = verdict(_cpg(rows), _spec(), function="requestListener") assert answer is not None and answer.rung is Rung.CORROBORATED, "a sanitized real flow is not an entailment" assert "request.args['f']" in answer.witness def test_parameter_flows_still_decide_when_no_modelled_source_reaches_the_sink() -> None: from openultrasast.model.ladder import Rung from openultrasast.model.taint import verdict rows = [ { "sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "username", "sanitized": False, "length": 2, "inLabeledScope": True, } ] answer = verdict(_cpg(rows), _spec(), function="run") assert answer is not None and answer.rung is Rung.ENTAILED def test_a_region_scopes_its_question_to_its_own_file() -> None: """A region with no enclosing function must not be answered with the whole repository. Sending `function=""` unscoped matched every hit in the tree and the driver attributed all of it to whichever region asked: one permissive literal in app.py became eight identical entailed findings in eight files that do not contain it. """ from openultrasast.model.config_value import request_params as config_params from openultrasast.model.specs import config_specs, taint_specs from openultrasast.model.taint import request_params as taint_params taint = taint_params(taint_specs(language="python")["injection"], function="", file="models/user_model.py") config = config_params(config_specs(language="python")["config_secrets"], function="", file="config.py") assert taint["file"] == "models/user_model.py" assert config["file"] == "config.py" # The pair path stays unscoped, so the committed pair measurements remain reproducible. assert taint_params(taint_specs(language="python")["injection"], function="run")["file"] == ""