diff --git a/benchmarks/pairs/README.md b/benchmarks/pairs/README.md index 5956036..daef66d 100644 --- a/benchmarks/pairs/README.md +++ b/benchmarks/pairs/README.md @@ -97,8 +97,9 @@ when a hunter model is configured. The improve loop rejects a change that regresses any profile on the holdout split. Harvest for every slice goes through `benchmarks/pairs/harvest.py --slice ` -(modes `name`, `line_range`, `hunk`, `enclosing`; comment and string mentions -never anchor) and `benchmarks/pairs/catalog_gen.py --slice ` regenerates +(modes `name`, `line_range`, `hunk`, `enclosing`, `handler_context`; comment +and string mentions never anchor; `handler_context` keeps the labeled handler +with its decorators and the statements that register it) and `benchmarks/pairs/catalog_gen.py --slice ` regenerates the catalog from `recipes.toml`. Maintainer only; CI never fetches. ## Commands diff --git a/benchmarks/pairs/harvest.py b/benchmarks/pairs/harvest.py index 701b899..a556814 100644 --- a/benchmarks/pairs/harvest.py +++ b/benchmarks/pairs/harvest.py @@ -37,7 +37,7 @@ ALLOWED_HOSTS = { "hg-edge.mozilla.org", } BLOCKED_TOKENS = ("fixfox", "zenodo.org/record") -MODES = ("name", "line_range", "hunk", "enclosing") +MODES = ("name", "line_range", "hunk", "enclosing", "handler_context") _SUFFIX_LANGUAGE = { ".c": "c", ".h": "c", @@ -194,6 +194,28 @@ def extract_python_block(source: str, function_name: str) -> str: raise RecipeError(f"function not found: {function_name}") +def extract_handler_context(source: str, function_name: str, language: str = "c") -> str: + """The named function (with its decorators) followed by every statement outside it that names the function as an + identifier: `app.add_url_rule(..., view_func=fn)`, `router.get('/x', fn)`, `path('x', fn)`, `app.use(fn)`. Comment and + string mentions never count (the masked text is searched). One excerpt per side, so guards living in decorators, + routers and middleware are visible to the teacher and the scorer (authorization-obligations, Req 7.1).""" + body = extract_function(source, function_name, language) + lines = source.splitlines() + start = _start_line_of(source, body) - 1 + end = start + len(body.splitlines()) + masked_lines = mask_comments_and_strings(source, language).splitlines() + mention = re.compile(rf"(? str: """Inclusive 1-based line range.""" lines = source.splitlines(keepends=True) @@ -384,7 +406,7 @@ def validate_recipe(recipe: dict[str, Any], *, require_license: bool = True) -> raise RecipeError(f"{name}: unknown mode {mode!r}; expected one of {MODES}") if not recipe.get("path"): raise RecipeError(f"{name}: path required") - if mode == "name" and not recipe.get("function"): + if mode in {"name", "handler_context"} and not recipe.get("function"): raise RecipeError(f"{name}: path and function required") if mode == "line_range" and not (isinstance(recipe.get("line_start"), int) and isinstance(recipe.get("line_end"), int)): raise RecipeError(f"{name}: line_range mode needs integer line_start and line_end") @@ -456,8 +478,9 @@ def extract_pair_with_lines(recipe: dict[str, Any], parent_src: str, fix_src: st raise RecipeError(f"{recipe.get('name')}: line {line}/{fix_line} is not inside a function") return (extract_line_range(parent_src, *parent_bounds), parent_bounds[0]), (extract_line_range(fix_src, *fix_bounds), fix_bounds[0]) function = str(recipe["function"]) - parent_text = extract_function(parent_src, function, language) - fix_text = extract_function(fix_src, function, language) + extract = extract_handler_context if mode == "handler_context" else extract_function + parent_text = extract(parent_src, function, language) + fix_text = extract(fix_src, function, language) return (parent_text, _start_line_of(parent_src, parent_text)), (fix_text, _start_line_of(fix_src, fix_text)) @@ -511,7 +534,8 @@ def extract_pair(recipe: dict[str, Any], parent_src: str, fix_src: str) -> tuple raise RecipeError(f"{recipe.get('name')}: line {line}/{fix_line} is not inside a function") return extract_line_range(parent_src, *parent_bounds), extract_line_range(fix_src, *fix_bounds) function = str(recipe["function"]) - return extract_function(parent_src, function, language), extract_function(fix_src, function, language) + extract = extract_handler_context if mode == "handler_context" else extract_function + return extract(parent_src, function, language), extract(fix_src, function, language) def harvest_recipe(recipe: dict[str, Any], root: Path, *, require_license: bool = True) -> tuple[Path, Path]: diff --git a/benchmarks/pairs/mechanisms.toml b/benchmarks/pairs/mechanisms.toml index 3e3f967..5ffdefa 100644 --- a/benchmarks/pairs/mechanisms.toml +++ b/benchmarks/pairs/mechanisms.toml @@ -1,6 +1,6 @@ # Closed mechanism vocabulary for pair labels (pair-corpus-honesty, Req 2.1). # Every [[pair.expected]] row names one of these ids. Bump `version` when the set changes. -version = "1" +version = "2" [[mechanism]] id = "source_reaches_sink" @@ -78,6 +78,18 @@ definition = "A validator exists but accepts inputs it should reject; the fix st id = "cross_artifact" definition = "The bug is only decidable with an artifact outside the vendored files (a properties file, a schema, a build flag)." +[[mechanism]] +id = "unconstrained_protected_read" +definition = "A read of a protected resource reaches the store without an identity constraint or ownership check bound from the authenticated context (authorization-obligations, protected_read)." + +[[mechanism]] +id = "unconstrained_protected_write" +definition = "A write to a protected resource reaches the store without an identity constraint or ownership check bound from the authenticated context (authorization-obligations, protected_write)." + +[[mechanism]] +id = "unguarded_privileged_action" +definition = "A privileged action is reached by a handler without a dominating path guard or ownership check (authorization-obligations, privileged_action)." + [[mechanism]] id = "other" definition = "Not yet classified. Present so bulk-mined pairs load; review should replace it." diff --git a/src/openultrasast/benchmark.py b/src/openultrasast/benchmark.py index 88eca9c..b3714ce 100644 --- a/src/openultrasast/benchmark.py +++ b/src/openultrasast/benchmark.py @@ -27,6 +27,7 @@ class ExpectedFinding: function: str | None = None sink: str | None = None mechanism: str | None = None + obligation: str | None = None # additive (authorization-obligations, Req 7.2): an OPERATION_KIND the labeled function owes @dataclass(frozen=True) diff --git a/src/openultrasast/pairs.py b/src/openultrasast/pairs.py index a216187..1046738 100644 --- a/src/openultrasast/pairs.py +++ b/src/openultrasast/pairs.py @@ -39,6 +39,7 @@ from .ruleset import PatternRule from .semantic import OverlayRecord, adjudicate from .semantic.facts import FactLoadError, load_facts from .semantic.functions import RESERVED_FUNCTION_LABELS, function_at, named_function_ranges, spans_named +from .semantic.obligations.facts import OPERATION_KINDS DEFAULT_CATALOG = Path("benchmarks/pairs/catalog.toml") DEFAULT_SAST_CATALOG = Path("benchmarks/pairs/sast/catalog.toml") @@ -557,6 +558,9 @@ def _parse_expected(item: dict[str, object], name: str, mechanisms: frozenset[st raise CatalogError(f"pair {name}: expected row is missing mechanism") if mechanism not in mechanisms: raise CatalogError(f"pair {name}: unknown mechanism {mechanism!r}; see benchmarks/pairs/mechanisms.toml") + obligation = item.get("obligation") + if obligation is not None and obligation not in OPERATION_KINDS: + raise CatalogError(f"pair {name}: unknown obligation {obligation!r}; expected one of {sorted(OPERATION_KINDS)}") return ExpectedFinding( cwe=str(item["cwe"]), vulnerability_class=str(item.get("class", item.get("vulnerability_class", "unknown"))), @@ -567,6 +571,7 @@ def _parse_expected(item: dict[str, object], name: str, mechanisms: frozenset[st function=str(item["function"]) if "function" in item else None, sink=str(item["sink"]) if "sink" in item else None, mechanism=mechanism, + obligation=str(obligation) if obligation is not None else None, ) @@ -626,9 +631,10 @@ class _Side: records: list[OverlayRecord] parse_failed: int ranges: dict[str, tuple[tuple[str, int, int], ...]] + obligations: list[StaticFinding] = field(default_factory=list) # obligation statics, computed for obligation-labeled rows only -def _overlay_scan(target: Path, ruleset: tuple[PatternRule, ...] | None = None) -> _Side: +def _overlay_scan(target: Path, ruleset: tuple[PatternRule, ...] | None = None, *, obligations: bool = False) -> _Side: targets = _targets(target) findings = [f for f in quick_scan_findings(target, targets, rank_targets(targets), ruleset) if f.status != "shadow"] records = adjudicate(root=target, targets=targets, findings=findings) @@ -645,7 +651,43 @@ def _overlay_scan(target: Path, ruleset: tuple[PatternRule, ...] | None = None) parse_failed += 1 else: ranges[item.path] = named - return _Side(findings=findings, records=records, parse_failed=parse_failed, ranges=ranges) + statics = _obligation_scan(target, targets) if obligations else [] + return _Side(findings=findings, records=records, parse_failed=parse_failed, ranges=ranges, obligations=statics) + + +def _obligation_scan(root: Path, targets: Sequence[FileTarget]) -> list[StaticFinding]: + """Obligation findings for one side in function-local mode (no path records, no declared policy, no store): the + scorer's additive rule for obligation-labeled rows (Req 7.2). Never runs for rows without the label.""" + from .config import ObligationsConfig + from .semantic.ir import parse_file + from .semantic.obligations import check_obligations, findings_to_static, load_obligation_facts + from .semantic.obligations.dominance import OrderDominance + + try: + flow_facts = load_facts() + except FactLoadError: + return [] + irs: dict[str, tuple[object, str]] = {} + texts: dict[str, str] = {} + for target in targets: + try: + text = (root / target.path).read_text(errors="ignore") + except OSError: + continue + texts[target.path] = text + irs[target.path] = (parse_file(target.path, text, target.language), text) + result = check_obligations( + irs=irs, # type: ignore[arg-type] + entries=analyze_entry_points(root, list(targets)), + facts=load_obligation_facts(), + flow_facts=flow_facts, + policy=None, + paths=(), + dominance=OrderDominance(texts=texts), + store_shapes=(), + min_siblings=ObligationsConfig().min_siblings, + ) + return findings_to_static(result) def _inventory_only(case: PairCase, ruleset: tuple[PatternRule, ...] | None = None) -> PairOutcome: @@ -777,15 +819,16 @@ def _inventory_outcome( def _evaluate_overlay_pair(case: PairCase, vuln_root: Path, fix_root: Path, ruleset: tuple[PatternRule, ...] | None = None) -> PairOutcome: - vuln = _overlay_scan(vuln_root, ruleset) - fixed = _overlay_scan(fix_root, ruleset) + obligations = any(row.obligation for row in case.expected) + vuln = _overlay_scan(vuln_root, ruleset, obligations=obligations) + fixed = _overlay_scan(fix_root, ruleset, obligations=obligations) loss: dict[str, int] = { "parse_failed_vuln": vuln.parse_failed, "parse_failed_fixed": fixed.parse_failed, "unadjudicated_vuln": sum(1 for record in vuln.records if record.disposition == "unadjudicated"), "unadjudicated_fixed": sum(1 for record in fixed.records if record.disposition == "unadjudicated"), } - if not _adjudicated(vuln.records) and not _adjudicated(fixed.records): + if not _adjudicated(vuln.records) and not _adjudicated(fixed.records) and not (vuln.obligations or fixed.obligations): # Parser/facts could not adjudicate this language; score inventory so # labeled calibration does not collapse when tree-sitter grammars are absent. loss["unresolved_labels"] = sum(1 for row in case.expected if _label_unresolved(row, vuln.ranges)) @@ -793,16 +836,27 @@ def _evaluate_overlay_pair(case: PairCase, vuln_root: Path, fix_root: Path, rule detected_vuln = _detections(vuln.records) detected_fix = _detections(fixed.records) expected_total = len(case.expected) - matched = sum(1 for row in case.expected if _overlay_matches_expected(row, detected_vuln, case.language, vuln.ranges)) + matched = sum( + 1 + for row in case.expected + if _overlay_matches_expected(row, detected_vuln, case.language, vuln.ranges) + or _obligation_matches_expected(row, vuln.obligations, vuln.ranges) + ) recall = matched / expected_total if expected_total else 1.0 detected = recall + 1e-12 >= case.min_recall - overlay_leaks = _overlay_leaks(case, detected_fix, fixed.ranges) + overlay_leaks = _overlay_leaks(case, detected_fix, fixed.ranges) + _obligation_leaks(case, fixed.obligations, fixed.ranges) silent = not overlay_leaks unresolved = sum(1 for row in case.expected if _label_unresolved(row, vuln.ranges)) misses = tuple( f"{row.rule_id or '-'}:{row.path}:{row.cwe}" + ("!unresolved" if _label_unresolved(row, vuln.ranges) else "") for row in case.expected - if not _overlay_matches_expected(row, detected_vuln, case.language, vuln.ranges) + if not ( + _overlay_matches_expected(row, detected_vuln, case.language, vuln.ranges) + or _obligation_matches_expected(row, vuln.obligations, vuln.ranges) + ) + ) + obligation_kinds = ( + ("obligation",) if any(_obligation_matches_expected(row, vuln.obligations, vuln.ranges) for row in case.expected) else () ) return PairOutcome( name=case.name, @@ -828,7 +882,7 @@ def _evaluate_overlay_pair(case: PairCase, vuln_root: Path, fix_root: Path, rule known_limit=case.known_limit, split=case.split, weak_labels=sum(1 for row in case.expected if is_weak_label(row)), - detection_kinds=tuple(sorted({record.disposition for record in detected_vuln})), + detection_kinds=tuple(sorted({record.disposition for record in detected_vuln} | set(obligation_kinds))), parse_failed_vuln=loss["parse_failed_vuln"], parse_failed_fixed=loss["parse_failed_fixed"], unadjudicated_vuln=loss["unadjudicated_vuln"], @@ -976,6 +1030,31 @@ def _hunter_matches_expected( return bool(expected.rule_id and expected.rule_id in finding.finding_id) +def _obligation_matches_expected(expected: ExpectedFinding, statics: Sequence[StaticFinding], ranges: FunctionRanges | None) -> bool: + """The one additive rule (Req 7.2): a row with `obligation` is detected by a finding tagged `obligation:` whose line + lies inside the labeled function. Nothing else about the scorer changes; rows without the label never reach here.""" + if not expected.obligation or not expected.function: + return False + tag = f"obligation:{expected.obligation}" + for finding in statics: + if tag not in finding.tags or finding.line is None: + continue + spans = spans_named((ranges or {}).get(finding.path, ()), expected.function) + if any(start <= finding.line <= end for start, end in spans): + return True + return False + + +def _obligation_leaks(case: PairCase, statics: Sequence[StaticFinding], ranges: FunctionRanges | None) -> tuple[str, ...]: + if not statics: + return () + if case.fix_policy == "silent": + return tuple(finding.finding_id for finding in statics if any(row.obligation for row in case.expected)) + return tuple( + finding.finding_id for finding in statics if any(_obligation_matches_expected(row, [finding], ranges) for row in case.expected) + ) + + def _overlay_leaks(case: PairCase, detected_fix: Sequence[OverlayRecord], ranges: FunctionRanges | None = None) -> tuple[str, ...]: if case.fix_policy == "silent": return tuple(record.proposal_id for record in detected_fix) diff --git a/tests/test_pair_corpus.py b/tests/test_pair_corpus.py index 9752594..5b6c965 100644 --- a/tests/test_pair_corpus.py +++ b/tests/test_pair_corpus.py @@ -985,3 +985,57 @@ def test_vibe_py_and_agent_vfc_carry_pointer_rows_and_the_default_run_skips_them payload = json.loads(buf.getvalue()) skipped = [item for item in payload["degradations"] if item["reason"] == "pointer_pair_skipped"] assert len(skipped) == len(pointers) and payload["overall"]["pairs"] == 35 + + +_OBLIGATION_HEAD = ( + '[[pair]]\nname = "leaky-books"\nslice = "vibe-py"\nlanguage = "python"\nvuln = "v.py"\nfixed = "f.py"\nrelpath = "app.py"\n' + 'provenance = "human"\nreview_tier = "seeded"\n\n' + '[[pair.expected]]\ncwe = "CWE-639"\nclass = "broken_access_control"\npath = "app.py"\nfunction = "leaky"\n' + 'mechanism = "unconstrained_protected_read"\n' +) +_LEAKY = ( + "from flask import request\n\n\n" + "@app.route('/books/')\ndef leaky(title):\n" + " return Book.query.filter_by(book_title=title).first()\n" +) +_CONSTRAINED = ( + "from flask import request\n\n\n" + "@app.route('/books/<title>')\ndef leaky(title):\n" + " user_id = request.user.id\n" + " return Book.query.filter_by(user_id=user_id, book_title=title).first()\n" +) + + +def test_obligation_labeled_row_loads_and_is_detected_by_an_obligation_finding(tmp_path: Path) -> None: + """authorization-obligations 4.1 (Req 7.2, 7.3): the additive `obligation` label and the scorer's one additive rule.""" + from openultrasast.pairs import _load_catalog_file, load_mechanisms + + _, ids = load_mechanisms() + assert {"unconstrained_protected_read", "unconstrained_protected_write", "unguarded_privileged_action"} <= ids + (tmp_path / "v.py").write_text(_LEAKY) + (tmp_path / "f.py").write_text(_CONSTRAINED) + (tmp_path / "catalog.toml").write_text(_OBLIGATION_HEAD + 'obligation = "protected_read"\n') + (case,) = _load_catalog_file(tmp_path / "catalog.toml") + assert case.expected[0].obligation == "protected_read" + outcome = evaluate_pair(case) + assert outcome.detected_vuln and outcome.vuln_matched == 1 and "obligation" in outcome.detection_kinds + assert outcome.silent_fix and outcome.pair_correct # the constrained twin discharges the obligation + # the same finding outside the labeled function is not a detection + outside = _LEAKY.replace("def leaky(", "def someone_else(") + (tmp_path / "v2.py").write_text(outside) + (tmp_path / "catalog2.toml").write_text(_OBLIGATION_HEAD.replace('vuln = "v.py"', 'vuln = "v2.py"') + 'obligation = "protected_read"\n') + (elsewhere,) = _load_catalog_file(tmp_path / "catalog2.toml") + assert not evaluate_pair(elsewhere).detected_vuln + + +def test_catalog_rejects_an_obligation_kind_outside_the_closed_set(tmp_path: Path) -> None: + from openultrasast.pairs import CatalogError, _load_catalog_file + + (tmp_path / "v.py").write_text(_LEAKY) + (tmp_path / "f.py").write_text(_CONSTRAINED) + (tmp_path / "catalog.toml").write_text(_OBLIGATION_HEAD + 'obligation = "guess"\n') + with pytest.raises(CatalogError, match="unknown obligation 'guess'"): + _load_catalog_file(tmp_path / "catalog.toml") + (tmp_path / "plain.toml").write_text(_OBLIGATION_HEAD) # rows without the field load as before + (case,) = _load_catalog_file(tmp_path / "plain.toml") + assert case.expected[0].obligation is None diff --git a/tests/test_pair_harvest.py b/tests/test_pair_harvest.py index 0f9fec8..805d111 100644 --- a/tests/test_pair_harvest.py +++ b/tests/test_pair_harvest.py @@ -522,3 +522,42 @@ def test_agent_vfc_builder_writes_toml_booleans() -> None: b = runpy.run_path(str(Path("benchmarks/pairs/agent-vfc/build_recipes.py")), run_name="agent_builder") (row,) = tomllib.loads(b["to_toml"]([{"name": "p", "vendored": False, "line": 3, "reviewer": "pending"}]))["recipe"] assert row["vendored"] is False and row["line"] == 3 + + +def test_handler_context_mode_keeps_decorators_and_registration() -> None: + """authorization-obligations 4.1 (Req 7.1): the handler travels with the statements that register it.""" + lib = _lib() + source = ( + "from flask import request\n\n" + "@app.route('/books/<title>')\n" + "@login_required\n" + "def guarded(title):\n" + " return Book.query.filter_by(book_title=title).first()\n\n\n" + "def other():\n" + " return 1\n\n\n" + "app.add_url_rule('/legacy/books/<title>', view_func=guarded)\n" + "# guarded is mentioned in a comment and 'guarded' in a string: neither is a registration\n" + "label = 'guarded'\n" + ) + excerpt = lib["extract_handler_context"](source, "guarded", "python") + assert excerpt.startswith("@app.route('/books/<title>')\n@login_required\ndef guarded(title):") + assert "app.add_url_rule('/legacy/books/<title>', view_func=guarded)" in excerpt + assert "def other" not in excerpt and "mentioned in a comment" not in excerpt and "label = 'guarded'" not in excerpt + js = "function list(req, res) {\n return Book.find({});\n}\n\nfunction other() {}\n\nrouter.get('/books', requireAuth, list);\n" + js_excerpt = lib["extract_handler_context"](js, "list", "javascript") + assert "function list(req, res)" in js_excerpt and "router.get('/books', requireAuth, list);" in js_excerpt + assert "function other" not in js_excerpt + recipe = { + "name": "r", + "parent": "p", + "commit": "c", + "path": "app.py", + "mode": "handler_context", + "function": "guarded", + "license": "MIT", + } + lib["validate_recipe"](recipe) + with pytest.raises(lib["RecipeError"], match="function"): + lib["validate_recipe"]({**recipe, "function": ""}) + (vuln, start), (fixed, _) = lib["extract_pair_with_lines"]({**recipe}, source, source.replace("@login_required\n", "")) + assert start == 3 and "add_url_rule" in vuln and "@login_required" not in fixed and "add_url_rule" in fixed