diff --git a/src/openultrasast/model/pipeline.py b/src/openultrasast/model/pipeline.py index 993c29e..7dcb292 100644 --- a/src/openultrasast/model/pipeline.py +++ b/src/openultrasast/model/pipeline.py @@ -28,8 +28,10 @@ from typing import Any from ..cpg.backend import CpgResult from ..redaction import redact_secrets -from .ladder import Rung -from .specs import TaintSpec +from .config_value import verdict as config_verdict +from .dominance import verdict as dominance_verdict +from .ladder import Rung, Verdict +from .specs import ConfigSpec, DominanceSpec, TaintSpec from .taint import verdict as taint_verdict logger = logging.getLogger(__name__) @@ -62,9 +64,26 @@ class ModelFinding: contradiction: str = "" +def _arbitrate( + cpg: CpgResult, spec: TaintSpec | DominanceSpec | ConfigSpec, *, function: str, parameter_sources: bool +) -> Verdict | None: + """Route a family to the arbiter that can actually decide it. + + Dispatching on the spec type rather than on a family name keeps this honest: a family without an arbiter + cannot be silently routed to the wrong one. The pipeline first shipped taint-only, so access_control and + config_secrets -- whose arbiters had worked since group 4 -- produced nothing, and every one of their + pairs read as `both_silent` on the first end-to-end run. + """ + if isinstance(spec, DominanceSpec): + return dominance_verdict(cpg, spec, function=function) + if isinstance(spec, ConfigSpec): + return config_verdict(cpg, spec, function=function) + return taint_verdict(cpg, spec, function=function, parameter_sources=parameter_sources) + + def scan_region( cpg: CpgResult, - spec: TaintSpec, + spec: TaintSpec | DominanceSpec | ConfigSpec, *, function: str = "", client: Any | None = None, @@ -73,7 +92,7 @@ def scan_region( parameter_sources: bool = False, ) -> list[ModelFinding]: """Findings for one region and family, ordered strongest first.""" - answer = taint_verdict(cpg, spec, function=function, parameter_sources=parameter_sources) + answer = _arbitrate(cpg, spec, function=function, parameter_sources=parameter_sources) if answer is not None and answer.rung is Rung.ENTAILED: # The graph decided. No model call, and no candidate needs asking about: this region has its finding. @@ -115,7 +134,7 @@ def scan_region( return findings -def _sanitizer_question(spec: TaintSpec, witness: str, function: str) -> str: +def _sanitizer_question(spec: TaintSpec | DominanceSpec | ConfigSpec, witness: str, function: str) -> str: source, _, sink = witness.partition(" -> ") return ( f"You are judging ONE candidate a static model already found. Do not look for other issues.\n\n" diff --git a/tests/test_model_pipeline.py b/tests/test_model_pipeline.py index 00d4e7b..f32f8a2 100644 --- a/tests/test_model_pipeline.py +++ b/tests/test_model_pipeline.py @@ -119,3 +119,47 @@ def test_the_candidate_budget_is_bounded() -> None: many = tuple({"id": f"a:{i}:x", "text": "db.raw(q)", "line": i} for i in range(50)) scan_region(_cpg([]), _spec(), function="run", client=client, model="m", candidates=many) assert client.calls == MAX_JUDGED_CANDIDATES + + +# --- family dispatch ------------------------------------------------------------------------------------ +# +# The pipeline first shipped taking a TaintSpec and calling taint_verdict, so access_control and +# config_secrets -- which have working arbiters since group 4 -- produced nothing at all. On the first +# vibe-py run every one of their pairs scored `both_silent`, which read as the model failing when in fact +# there was no code path to the arbiter. Same class as the ceiling harness routing config through taint. + + +def _dominance_spec(): # type: ignore[no-untyped-def] + from openultrasast.model.specs import DominanceSpec + + return DominanceSpec(family="access_control", language="python", + operations=("filter_by",), dischargers=("current_user",)) + + +def _config_spec(): # type: ignore[no-untyped-def] + from openultrasast.model.specs import ConfigSpec + + return ConfigSpec(family="config_secrets", language="python", settings=("CORS",), permissive=('"*"',)) + + +def test_an_absence_family_is_arbitrated_by_dominance_not_taint() -> None: + from openultrasast.model.ladder import Rung + from openultrasast.model.pipeline import scan_region + + rows = [{"operation": "Note.query.filter_by(id=n)", "opLine": "9", "opMethod": "leaky", "dominatingGuards": []}, + {"operation": "Note.query.filter_by(id=n)", "opLine": "3", "opMethod": "safe", + "dominatingGuards": ["current_user.id"]}] + findings = scan_region(_cpg(rows), _dominance_spec(), function="leaky", client=None, model="", candidates=()) + assert len(findings) == 1 and findings[0].rung is Rung.ENTAILED + assert findings[0].family == "access_control" + + +def test_a_configuration_family_is_arbitrated_by_constant_abstraction() -> None: + from openultrasast.model.ladder import Rung + from openultrasast.model.pipeline import scan_region + + rows = [{"setting": 'CORS(app, origins="*")', "line": "3", "method": "create_app", + "literalArgs": ['"*"'], "args": ["app"]}] + findings = scan_region(_cpg(rows), _config_spec(), function="create_app", client=None, model="", candidates=()) + assert len(findings) == 1 and findings[0].rung is Rung.ENTAILED + assert findings[0].family == "config_secrets"