diff --git a/src/openultrasast/rank.py b/src/openultrasast/rank.py index 7157518..9452d1a 100644 --- a/src/openultrasast/rank.py +++ b/src/openultrasast/rank.py @@ -1,12 +1,16 @@ from __future__ import annotations import json -from dataclasses import asdict, dataclass +from collections.abc import Collection +from dataclasses import asdict, dataclass, replace from pathlib import Path -from typing import Protocol +from typing import TYPE_CHECKING, Protocol from .preprocess import FileTarget +if TYPE_CHECKING: + from .findings import StaticFinding + class RankingError(ValueError): """Raised when ranker output does not match the expected schema.""" @@ -234,3 +238,28 @@ def _reachability_from_hints(hints: list[dict[str, object]]) -> int: elif access_level == "review-required" or access_level == "local-only": score = max(score, 2) return score + + +# authorization-obligations Req 6.4: obligation findings rank by resource sensitivity and evidence label, and never above +# a sandbox-proven finding in the same run. Every other finding keeps its priority. +_OBLIGATION_LABEL_WEIGHT = {"declared_policy_violation": 0.9, "consistency_violation": 0.6, "function_local": 0.3} +_OBLIGATION_SENSITIVITY_WEIGHT = {"high": 1.0, "medium": 0.7, "low": 0.4} +_PROVEN_LEVELS = frozenset({"crash_reproduced", "exploit_demonstrated", "patch_validated", "root_cause_explained"}) + + +def rank_obligations(findings: list[StaticFinding], *, proven_ids: Collection[str] = ()) -> list[StaticFinding]: + """Rank obligation findings strictly below every proven finding in the run. Proof rungs live on verdicts, not on + ``StaticFinding``, so the sandbox's proven finding ids are passed in; ``evidence_level`` is honoured as well.""" + proven = [f.ranking_priority for f in findings if f.evidence_level in _PROVEN_LEVELS or f.finding_id in proven_ids] + ceiling = min(proven) if proven else None + ranked: list[StaticFinding] = [] + for finding in findings: + if not finding.finding_id.startswith("obligation:"): + ranked.append(finding) + continue + label = next((tag.split(":", 1)[1] for tag in finding.tags if tag.startswith("obligation_evidence:")), "function_local") + weight = _OBLIGATION_SENSITIVITY_WEIGHT.get(finding.severity, 0.7) * _OBLIGATION_LABEL_WEIGHT.get(label, 0.3) + if ceiling is not None and weight >= ceiling: + weight = max(ceiling * 0.99, 0.0) * _OBLIGATION_LABEL_WEIGHT.get(label, 0.3) / 0.9 # strictly below the lowest proven finding + ranked.append(replace(finding, ranking_priority=round(weight, 4))) + return ranked diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index 7af33d8..5f2637c 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -20,11 +20,14 @@ def write_markdown_report( verdicts: Sequence[object] | None = None, overlay: Sequence[object] | None = None, mechanisms: Mapping[str, Mapping[str, object]] | None = None, + obligations: Mapping[str, Mapping[str, object]] | None = None, + obligations_summary: Mapping[str, object] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications or []) overlay_by_id = _overlay_by_id(overlay or []) cited: dict[str, Mapping[str, object]] = {} + cited_obligations: dict[str, Mapping[str, object]] = {} lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", "", "## Inventory", ""] if not findings: lines.append("No quick-mode findings were emitted.") @@ -60,9 +63,15 @@ def write_markdown_report( lines.append(f"- Mechanism: `{mechanism_id}` — {info.get('summary', '')}") lines.append(f"- Known fix guard: `{info.get('guard', 'none')}`") lines.append(f"- Learned from pairs: `{', '.join(_pairs_of(info))}`") + info_o = (obligations or {}).get(finding.finding_id) + if info_o is not None: + cited_obligations[finding.finding_id] = info_o + _append_obligation_lines(lines, info_o, (mechanisms or {}).get(str(info_o.get("known_fix") or ""))) lines.extend(["", finding.rationale, ""]) if cited: _append_mechanisms(lines, cited) + if cited_obligations: + _append_obligations(lines, cited_obligations, obligations_summary) if overlay: _append_overlay(lines, overlay) if complexity_map is not None: @@ -145,6 +154,7 @@ def write_sarif_report( path: Path, overlay: Sequence[object] | None = None, mechanisms: Mapping[str, Mapping[str, object]] | None = None, + obligations: Mapping[str, Mapping[str, object]] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -164,7 +174,11 @@ def write_sarif_report( }, "results": [ _sarif_result( - finding, verification_by_id.get(finding.finding_id), overlay_by_id.get(finding.finding_id), mechanisms=mechanisms + finding, + verification_by_id.get(finding.finding_id), + overlay_by_id.get(finding.finding_id), + mechanisms=mechanisms, + obligation=(obligations or {}).get(finding.finding_id), ) for finding in findings ], @@ -189,6 +203,7 @@ def write_manifest( worth_fixing: dict[str, object] | None = None, provenance: dict[str, object] | None = None, variants: dict[str, object] | None = None, + obligations: dict[str, object] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -227,6 +242,8 @@ def write_manifest( payload["provenance"] = provenance if variants is not None: payload["variants"] = variants + if obligations is not None: + payload["obligations"] = obligations path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") @@ -269,6 +286,7 @@ def _sarif_result( overlay: object | None = None, *, mechanisms: Mapping[str, Mapping[str, object]] | None = None, + obligation: Mapping[str, object] | None = None, ) -> dict[str, object]: properties: dict[str, object] = { "finding_id": finding.finding_id, @@ -291,6 +309,15 @@ def _sarif_result( properties["mechanism_summary"] = str(info.get("summary", "")) properties["mechanism_guard"] = str(info.get("guard", "none")) properties["mechanism_pairs"] = _pairs_of(info) + if obligation is not None: + properties["obligation_kind"] = str(obligation.get("obligation", "")) + properties["obligation_resource"] = obligation.get("resource") + properties["obligation_missing"] = str(obligation.get("missing", "")) + properties["obligation_provenance"] = obligation.get("provenance") + properties["obligation_label"] = str(obligation.get("label", "")) + properties["obligation_evidence"] = _strings_of(obligation, "evidence") + properties["obligation_known_fix"] = obligation.get("known_fix") + properties["obligation_intent"] = obligation.get("intent") return { "ruleId": _rule_id(finding), "level": _sarif_level(finding.severity), @@ -408,3 +435,44 @@ def _append_mechanisms(lines: list[str], cited: Mapping[str, Mapping[str, object def _pairs_of(info: Mapping[str, object]) -> list[str]: pairs = info.get("pairs") return [str(p) for p in pairs] if isinstance(pairs, list | tuple) else [] + + +def _append_obligation_lines(lines: list[str], info: Mapping[str, object], mechanism: Mapping[str, object] | None) -> None: + resource = info.get("resource") or "a resource" + provenance = f" (`{info.get('provenance')}`)" if info.get("provenance") else "" + evidence = ", ".join(_strings_of(info, "evidence")) or "the operation fact alone" + lines.append(f"- Obligation: `{info.get('obligation')}` on `{resource}`") + lines.append(f"- Missing discharger: `{info.get('missing')}`{provenance}") + lines.append(f"- Evidence: `{info.get('label')}` — {evidence}") + if info.get("known_fix"): + guard = mechanism.get("guard") if mechanism else info.get("missing") + pairs = ", ".join(_pairs_of(mechanism)) if mechanism else "the mechanism store" + lines.append(f"- Known fix: `{guard}` learned from `{pairs}` ({info.get('known_fix')})") + if info.get("intent"): + lines.append(f"- Intent: `{info.get('intent')}`") + + +def _append_obligations(lines: list[str], cited: Mapping[str, Mapping[str, object]], summary: Mapping[str, object] | None = None) -> None: + lines.extend( + ["## Obligations", "", "Operations reached without a dominating discharger; suspicions labeled by why the obligation exists.", ""] + ) + if summary is not None: + version = summary.get("policy_version") + lines.append(f"Sibling sets evaluated: {summary.get('sibling_sets', 0)} (under-populated: {summary.get('under_populated', 0)})") + lines.append(f"Policy version: {f'`{version}`' if version else 'none'}") + lines.append("") + sets = summary.get("sets") + if isinstance(sets, list | tuple): + for group in sets: + if isinstance(group, Mapping): + lines.append(f"- `{group.get('module')}` / `{group.get('resource')}`: {group.get('handlers')} handlers") + lines.append("") + for finding_id, info in sorted(cited.items()): + resource = info.get("resource") or "a resource" + lines.append(f"- `{finding_id}`: `{info.get('obligation')}` on `{resource}` without `{info.get('missing')}` ({info.get('label')})") + lines.append("") + + +def _strings_of(info: Mapping[str, object], key: str) -> list[str]: + value = info.get(key) + return [str(item) for item in value] if isinstance(value, list | tuple) else [] diff --git a/tests/test_obligation_reports.py b/tests/test_obligation_reports.py new file mode 100644 index 0000000..ca20512 --- /dev/null +++ b/tests/test_obligation_reports.py @@ -0,0 +1,150 @@ +"""authorization-obligations task 3.2: reports and ranking name the obligation (offline).""" + +from __future__ import annotations + +import json +from pathlib import Path + +from openultrasast.findings import StaticFinding +from openultrasast.reports import write_markdown_report, write_sarif_report + + +def _obligation() -> StaticFinding: + return StaticFinding( + finding_id="obligation:protected_read:app.py:18:identity_constraint", + path="app.py", + title="Obligation not discharged: protected_read on book without identity_constraint", + severity="high", + confidence="low", + evidence_level="suspicion", + rationale=( + "protected_read on book in leaky is reached without a dominating identity_constraint (a identity_constraint is present " + "but bound from request_input). Why the obligation exists: app.py::constrained discharges identity_constraint. " + "Label: consistency_violation. Known fix learned from the corpus: corpus:abc." + ), + line=18, + function_name="leaky", + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=[ + "obligation:protected_read", + "discharger:identity_constraint", + "obligation_evidence:consistency_violation", + "resource:book", + "mechanism:corpus:abc", + ], + ranking_priority=0.0, + ) + + +def _proven() -> StaticFinding: + return StaticFinding( + **{ + **_obligation().__dict__, + "finding_id": "python-os-command:app.py:3", + "evidence_level": "crash_reproduced", + "tags": ["injection"], + "line": 3, + "ranking_priority": 1.0, + } + ) + + +OBLIGATIONS = { + "obligation:protected_read:app.py:18:identity_constraint": { + "obligation": "protected_read", + "resource": "book", + "missing": "identity_constraint", + "provenance": "request_input", + "label": "consistency_violation", + "evidence": ["app.py::constrained"], + "known_fix": "corpus:abc", + "intent": None, + } +} +MECHANISMS = { + "corpus:abc": { + "summary": "missing_auth_guard: protected_read without identity_constraint; fix binds authenticated_context", + "guard": "identity_constraint", + "pairs": ["vampi"], + "cwe": "CWE-639", + } +} + + +def test_markdown_names_obligation_missing_discharger_evidence_and_known_fix(tmp_path: Path) -> None: + path = tmp_path / "report.md" + write_markdown_report([_obligation()], path, mechanisms=MECHANISMS, obligations=OBLIGATIONS) + text = path.read_text() + assert "- Obligation: `protected_read` on `book`" in text + assert "- Missing discharger: `identity_constraint` (`request_input`)" in text + assert "- Evidence: `consistency_violation`" in text and "app.py::constrained" in text + assert "- Known fix: `identity_constraint`" in text and "vampi" in text + assert "- Evidence level: `suspicion`" in text + assert "## Obligations" in text + summary = { + "sibling_sets": 2, + "under_populated": 1, + "sets": [{"module": "api/books", "resource": "book", "handlers": 3}], + "policy_version": "abc123", + } + write_markdown_report([_obligation()], path, mechanisms=MECHANISMS, obligations=OBLIGATIONS, obligations_summary=summary) + text = path.read_text() + assert "Sibling sets evaluated: 2 (under-populated: 1)" in text and "- `api/books` / `book`: 3 handlers" in text + assert "Policy version: `abc123`" in text + + +def test_sarif_carries_obligation_properties(tmp_path: Path) -> None: + path = tmp_path / "report.sarif" + write_sarif_report([_obligation()], [], path, mechanisms=MECHANISMS, obligations=OBLIGATIONS) + (result,) = json.loads(path.read_text())["runs"][0]["results"] + props = result["properties"] + assert props["obligation_kind"] == "protected_read" and props["obligation_missing"] == "identity_constraint" + assert props["obligation_label"] == "consistency_violation" and props["obligation_evidence"] == ["app.py::constrained"] + assert props["obligation_known_fix"] == "corpus:abc" and props["evidence_level"] == "suspicion" + + +def test_obligation_ranking_weighs_sensitivity_and_label_but_stays_below_proven_findings() -> None: + from openultrasast.rank import rank_obligations + + declared = StaticFinding( + **{ + **_obligation().__dict__, + "finding_id": "obligation:protected_read:app.py:20:identity_constraint", + "tags": ["obligation:protected_read", "discharger:identity_constraint", "obligation_evidence:declared_policy_violation"], + "line": 20, + } + ) + local = StaticFinding( + **{ + **_obligation().__dict__, + "finding_id": "obligation:protected_read:app.py:22:identity_constraint", + "tags": ["obligation:protected_read", "discharger:identity_constraint", "obligation_evidence:function_local"], + "severity": "medium", + "line": 22, + } + ) + ranked = rank_obligations([_proven(), _obligation(), declared, local]) + by_id = {f.finding_id: f.ranking_priority for f in ranked} + assert by_id[declared.finding_id] > by_id[_obligation().finding_id] > by_id[local.finding_id] > 0 + assert max(by_id[f] for f in by_id if f.startswith("obligation:")) < by_id["python-os-command:app.py:3"] + assert by_id["python-os-command:app.py:3"] == 1.0 # non-obligation findings untouched + + +def test_obligation_cap_binds_on_verdict_ids_because_the_pipeline_never_sets_proof_levels_on_static_findings() -> None: + """Round-3 finding: proof rungs live on verdicts, so the cap is keyed on the proven finding ids the sandbox returned.""" + from openultrasast.rank import rank_obligations + + proven_by_sandbox = StaticFinding(**{**_proven().__dict__, "evidence_level": "static_corroboration", "ranking_priority": 0.35}) + ranked = rank_obligations([proven_by_sandbox, _obligation()], proven_ids={proven_by_sandbox.finding_id}) + by_id = {f.finding_id: f.ranking_priority for f in ranked} + assert 0 < by_id[_obligation().finding_id] < 0.35 and by_id[proven_by_sandbox.finding_id] == 0.35 + unproven = rank_obligations([proven_by_sandbox, _obligation()]) + assert {f.finding_id: f.ranking_priority for f in unproven}[_obligation().finding_id] > 0.35 # no verdict, no cap + + +def test_reports_without_obligations_are_unchanged(tmp_path: Path) -> None: + path = tmp_path / "report.md" + write_markdown_report([_proven()], path) + assert "Obligation:" not in path.read_text() and "## Obligations" not in path.read_text()