diff --git a/src/openultrasast/semantic/mechanisms.py b/src/openultrasast/semantic/mechanisms.py index 3e9db86..a2ff748 100644 --- a/src/openultrasast/semantic/mechanisms.py +++ b/src/openultrasast/semantic/mechanisms.py @@ -14,6 +14,7 @@ from pathlib import Path from ..complexity.map import Hotspot from ..findings import StaticFinding from ..index import CodeChunk, EmbeddingClient, VectorIndex, VectorRecord, write_vector_index +from .obligations.shapes import ObligationShape from .overlay import OverlayRecord from .variants import Shape @@ -96,14 +97,14 @@ def append_mechanism( _TIER_RANK = {"seeded": 1, "reviewed": 2} -def corpus_mechanism_id(shape: Shape) -> str: +def corpus_mechanism_id(shape: Shape | ObligationShape) -> str: """Deterministic id from the shape key, so the same shape gets the same record on every machine (lever-addressable).""" return "corpus:" + hashlib.sha1(shape.key().encode()).hexdigest()[:16] def append_from_pair( store: MechanismStore, - shape: Shape, + shape: Shape | ObligationShape, *, summary: str, cwe: str, @@ -120,7 +121,15 @@ def append_from_pair( if existing is not None and _TIER_RANK.get(existing.review_tier, 0) > _TIER_RANK.get(tier, 0): tier = existing.review_tier # a shape keeps the strongest review it ever had inherited = tuple(tag for tag in (existing.tags if existing is not None else ()) if tag.startswith(("provenance:", "tier:"))) - positions = ", ".join(f"arg{p} ({k})" for p, k in zip(shape.source_positions, shape.source_kinds, strict=True)) + if isinstance(shape, Shape): + positions = ", ".join(f"arg{p} ({k})" for p, k in zip(shape.source_positions, shape.source_kinds, strict=True)) + exploitable = f"{positions} reaches {shape.sink_name}/{shape.arity}; the fix added {shape.guard}" + keywords: tuple[str, ...] = (shape.sink_name, shape.guard, *shape.source_kinds) + guard = shape.guard + else: # obligation shape: the discharger the fix carries is the 'guard' the reports show + exploitable = f"{shape.operation_kind} reached without {shape.discharger_kind}; the fix binds {shape.provenance}" + keywords = (shape.operation_kind, shape.discharger_kind, shape.provenance, shape.resource_class) + guard = shape.discharger_kind record = Mechanism( id=mechanism_id, summary=summary, @@ -129,18 +138,18 @@ def append_from_pair( tags=( shape.mechanism, f"mechanism:{shape.mechanism}", - f"guard:{shape.guard}", + f"guard:{guard}", *sorted({*inherited, f"provenance:{provenance}", f"tier:{tier}"}), ), - keywords=(shape.sink_name, shape.guard, *shape.source_kinds), - what_made_it_exploitable=f"{positions} reaches {shape.sink_name}/{shape.arity}; the fix added {shape.guard}", + keywords=keywords, + what_made_it_exploitable=exploitable, source_finding_id="", source_repo="", origin="corpus", review_tier=tier, pairs=pairs, shape=shape.to_dict(), - guard=shape.guard, + guard=guard, ) store.append(record) return record diff --git a/src/openultrasast/semantic/seed.py b/src/openultrasast/semantic/seed.py index 0e671fe..f483fa5 100644 --- a/src/openultrasast/semantic/seed.py +++ b/src/openultrasast/semantic/seed.py @@ -16,6 +16,8 @@ from ..preprocess import LANGUAGE_BY_EXTENSION from .facts import FactLoadError, SemanticFacts, load_facts from .ir import FileIR, parse_file from .mechanisms import MechanismStore, append_from_pair +from .obligations.facts import ObligationFacts, load_obligation_facts +from .obligations.shapes import ABSENCE_MECHANISMS, ObligationShape, derive_obligation, explain_skip from .variants import Shape, derive_shape LANGUAGE_UNSUPPORTED = "variants_language_unsupported" @@ -85,7 +87,7 @@ def export_mechanisms(cases: Sequence[PairCase], store: MechanismStore, *, facts if lessons: seeded.append(case.name) else: - skipped.append((case.name, "no_labeled_sink_call_site")) + skipped.append((case.name, _skip_detail(case, vuln_ir, fixed_ir, vuln_text, fixed_text, loaded))) return ExportReport( seeded=len(seeded), records=len(record_ids), @@ -97,9 +99,9 @@ def export_mechanisms(cases: Sequence[PairCase], store: MechanismStore, *, facts @dataclass(frozen=True) class Lesson: - """One shape a trusted pair teaches, with the CWE and a text-free summary.""" + """One shape a trusted pair teaches, with the CWE and a text-free summary (a sink shape or an obligation shape).""" - shape: Shape + shape: Shape | ObligationShape cwe: str summary: str @@ -125,6 +127,27 @@ def pair_lessons( for row in case.expected: if not row.function or not row.mechanism: continue + obligation_kind = getattr(row, "obligation", None) + if obligation_kind or row.mechanism in ABSENCE_MECHANISMS: + obligation = derive_obligation( + vuln_ir, + fixed_ir, + function=row.function, + obligation=obligation_kind, + mechanism=row.mechanism, + facts=_obligation_facts(), + flow_facts=facts, + vuln_text=vuln_text, + fixed_text=fixed_text, + ) + if obligation is not None: + summary = ( + f"{obligation.mechanism}: {obligation.operation_kind} without {obligation.discharger_kind} ({row.cwe}); " + f"fix binds {obligation.provenance}" + ) + lessons.append(Lesson(shape=obligation, cwe=row.cwe, summary=summary)) + continue + # an absence row that teaches no obligation may still teach a sink shape (a weak literal at a sink, say) shape = derive_shape( vuln_ir, fixed_ir, @@ -143,6 +166,32 @@ def pair_lessons( return lessons +def _skip_detail(case: PairCase, vuln_ir: FileIR, fixed_ir: FileIR, vuln_text: str, fixed_text: str, flow_facts: SemanticFacts) -> str: + """Why a parsing, labeled pair taught nothing: the obligation derivation's reason for absence rows, else the sink reason.""" + for row in case.expected: + if row.function and row.mechanism and (getattr(row, "obligation", None) or row.mechanism in ABSENCE_MECHANISMS): + return explain_skip( + vuln_ir, + fixed_ir, + function=row.function, + vuln_text=vuln_text, + fixed_text=fixed_text, + obligation=getattr(row, "obligation", None), + facts=_obligation_facts(), + flow_facts=flow_facts, + ) + return "no_labeled_sink_call_site" + + +_OBLIGATION_FACTS: list[ObligationFacts] = [] + + +def _obligation_facts() -> ObligationFacts: + if not _OBLIGATION_FACTS: + _OBLIGATION_FACTS.append(load_obligation_facts()) + return _OBLIGATION_FACTS[0] + + def _skip_reason(case: PairCase) -> str | None: if case.review_tier not in GATING_TIERS: return f"tier:{case.review_tier}" diff --git a/src/openultrasast/semantic/variant_search.py b/src/openultrasast/semantic/variant_search.py index 8444d7e..9c058fe 100644 --- a/src/openultrasast/semantic/variant_search.py +++ b/src/openultrasast/semantic/variant_search.py @@ -38,7 +38,8 @@ def search_tree( max_mechanisms: int, ) -> SearchResult: """Variant search over the parsed files of a tree with the store's corpus shapes (bounded by ``max_mechanisms``).""" - records = corpus_mechanisms(store.load())[: max(max_mechanisms, 0)] + records = [record for record in corpus_mechanisms(store.load()) if (record.shape or {}).get("family", "sink") == "sink"] + records = records[: max(max_mechanisms, 0)] shapes: list[Shape] = [] ids: dict[str, str] = {} degradations: list[dict[str, object]] = [] diff --git a/tests/test_obligation_export.py b/tests/test_obligation_export.py new file mode 100644 index 0000000..70ac011 --- /dev/null +++ b/tests/test_obligation_export.py @@ -0,0 +1,158 @@ +"""authorization-obligations task 1.3: obligation shapes travel through the existing exporter and store.""" + +from __future__ import annotations + +from pathlib import Path + +from openultrasast.benchmark import ExpectedFinding +from openultrasast.pairs import PairCase +from openultrasast.semantic.mechanisms import MechanismStore + +VULN = ( + "from flask import request\n\n\ndef get_book(book_title):\n" + " resp = token_validator(request.headers.get('Authorization'))\n if 'error' in resp:\n return None\n" + " book = Book.query.filter_by(book_title=book_title).first()\n return book\n" +) +FIXED = ( + "from flask import request\n\n\ndef get_book(book_title):\n" + " resp = token_validator(request.headers.get('Authorization'))\n if 'error' in resp:\n return None\n" + " user = User.query.filter_by(username=resp['sub']).first()\n" + " book = Book.query.filter_by(user=user, book_title=book_title).first()\n return book\n" +) +FLOW_VULN = "from flask import request\nimport os\n\n\ndef run():\n cmd = request.args.get('cmd')\n os.system(cmd)\n" +FLOW_FIXED = ( + "from flask import request\nimport os\n\n\ndef run():\n cmd = request.args.get('cmd')\n" + + " if cmd not in ALLOWED:\n return None\n os.system(cmd)\n" +) + + +def _case( + tmp_path: Path, name: str, vuln: str, fixed: str, *, function: str, mechanism: str, tier: str = "seeded", sink: str | None = None +) -> PairCase: + (tmp_path / f"{name}-v.py").write_text(vuln) + (tmp_path / f"{name}-f.py").write_text(fixed) + return PairCase( + name=name, + slice="vibe-py", + language="python", + origin="test", + vuln_file=tmp_path / f"{name}-v.py", + fixed_file=tmp_path / f"{name}-f.py", + relpath="app.py", + expected=( + ExpectedFinding( + cwe="CWE-639", vulnerability_class="x", path="app.py", evidence="", function=function, mechanism=mechanism, sink=sink + ), + ), + min_recall=1.0, + fix_policy="silent", + provenance="human", + review_tier=tier, + reviewer="t" if tier == "reviewed" else "", + ) + + +def test_exporter_writes_obligation_rows_deduped_with_provenance_and_skips_untrusted(tmp_path: Path) -> None: + from openultrasast.semantic.obligations.shapes import obligation_mechanisms + from openultrasast.semantic.seed import export_mechanisms + + store = MechanismStore(tmp_path / "mechanisms.jsonl") + cases = [ + _case(tmp_path, "a", VULN, FIXED, function="get_book", mechanism="missing_auth_guard"), + _case( + tmp_path, + "b", + VULN.replace("book", "item"), + FIXED.replace("book", "item"), + function="get_item", + mechanism="missing_auth_guard", + tier="reviewed", + ), + _case(tmp_path, "advisory", VULN, FIXED, function="get_book", mechanism="missing_auth_guard", tier="advisory"), + _case(tmp_path, "flow", FLOW_VULN, FLOW_FIXED, function="run", mechanism="source_reaches_sink", sink="os.system"), + ] + report = export_mechanisms(cases, store) + rows = store.load() + obligations = obligation_mechanisms(rows) + assert len(obligations) == 1 and obligations[0].pairs == ("a", "b") and obligations[0].origin == "corpus" + assert ( + obligations[0].shape is not None + and obligations[0].shape["family"] == "obligation" + and obligations[0].guard == "identity_constraint" + ) + sinks = [r for r in rows if r.shape and r.shape.get("family", "sink") != "obligation"] + assert len(sinks) == 1 and sinks[0].shape is not None and sinks[0].shape["sink_name"] == "system" # flow rows unaffected + assert report.seeded == 3 and dict(report.skipped)["advisory"] == "tier:advisory" + text = (tmp_path / "mechanisms.jsonl").read_text() + assert "book_title" not in text and "resp" not in text and "app.py" not in text + + +def test_absence_pair_without_a_lesson_is_skipped_with_the_derivation_reason(tmp_path: Path) -> None: + from openultrasast.semantic.seed import export_mechanisms + + store = MechanismStore(tmp_path / "m.jsonl") + report = export_mechanisms([_case(tmp_path, "same", VULN, VULN, function="get_book", mechanism="missing_auth_guard")], store) + assert store.load() == () and dict(report.skipped)["same"].startswith("no_discharger_added:identity_constraint") + + +def test_sink_search_ignores_obligation_rows(tmp_path: Path) -> None: + from openultrasast.preprocess import preprocess_repository + from openultrasast.semantic.facts import load_facts + from openultrasast.semantic.seed import export_mechanisms + from openultrasast.semantic.variant_search import search_tree + + store = MechanismStore(tmp_path / "m.jsonl") + export_mechanisms([_case(tmp_path, "a", VULN, FIXED, function="get_book", mechanism="missing_auth_guard")], store) + repo = tmp_path / "repo" + repo.mkdir() + (repo / "views.py").write_text(VULN) + _, targets = preprocess_repository(repo) + result = search_tree(repo, targets, store, load_facts(), max_mechanisms=500) + assert result.mechanisms_searched == 0 and result.hits == () and result.degradations == () + + +def test_real_vibe_py_absence_pairs_seed_at_least_one_obligation_shape(tmp_path: Path) -> None: + from openultrasast.pairs import load_pair_catalog, select_slice, select_vendored + from openultrasast.semantic.obligations.shapes import obligation_mechanisms + from openultrasast.semantic.seed import export_mechanisms + + store = MechanismStore(tmp_path / "m.jsonl") + report = export_mechanisms(select_vendored(select_slice(load_pair_catalog(), "vibe-py")), store) + rows = obligation_mechanisms(store.load()) + assert rows and any("vampi-books-get-by-title" in r.pairs and "vampi-users-update-password" in r.pairs for r in rows) + vampi = next(r for r in rows if "vampi-books-get-by-title" in r.pairs) + assert vampi.shape is not None + assert (vampi.shape["operation_kind"], vampi.shape["discharger_kind"], vampi.shape["provenance"], vampi.shape["resource_class"]) == ( + "protected_read", + "identity_constraint", + "authenticated_context", + "owned", + ) + assert all( + r.shape is not None and r.shape["operation_kind"] in {"protected_read", "protected_write", "privileged_action", "security_setting"} + for r in rows + ) + flow_rows = [r for r in store.load() if r.shape is not None and r.shape.get("family", "sink") == "sink"] + assert report.records == len(flow_rows) + len(rows) and len(flow_rows) >= 1 # both families, nothing lost to the other + + +def test_absence_row_without_an_obligation_lesson_still_teaches_its_sink_shape(tmp_path: Path) -> None: + """The fallback that keeps the flow store unchanged: a `missing_auth_guard` row whose twin adds no discharger but whose + labeled sink call carries a source still seeds a sink shape (the threatbyte pairs do exactly this).""" + from openultrasast.semantic.obligations.shapes import obligation_mechanisms + from openultrasast.semantic.seed import export_mechanisms + + vuln = ( + "import sqlite3\n\n\ndef get(user_id):\n cur = sqlite3.connect('x').cursor()\n" + " cur.execute('SELECT * FROM users WHERE id = ' + user_id)\n return cur.fetchone()\n" + ) + fixed = ( + "import sqlite3\n\n\ndef login(username):\n cur = sqlite3.connect('x').cursor()\n" + " cur.execute('SELECT * FROM users WHERE name = ?', (username,))\n return cur.fetchone()\n" + ) + store = MechanismStore(tmp_path / "m.jsonl") + report = export_mechanisms([_case(tmp_path, "tb", vuln, fixed, function="get", mechanism="missing_auth_guard", sink="execute")], store) + rows = store.load() + assert report.seeded == 1 and obligation_mechanisms(rows) == () + assert len(rows) == 1 and rows[0].shape is not None and rows[0].shape.get("family", "sink") == "sink" + assert rows[0].shape["sink_name"] == "execute"