mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
feat(authorization-obligations): the obligation checker, path-aware and function-local
Task 2.5. An obligated operation that is reached and has no dominating discharger is reported at suspicion, labeled by why the obligation exists: a declared policy clause, the practice of sibling handlers, or, in the degraded no-path mode, the operation fact alone. With path records and neither a clause nor a sibling norm nothing justifies the obligation and nothing is reported. A declared public route waives the path guard wherever it is required, not only where siblings practice it (Req 4.4). A malformed store row is recorded as a degradation naming the record instead of being skipped silently. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Opus 5
Claude Opus 5
parent
8e26dfd1e2
commit
2249de8d2e
@@ -0,0 +1,308 @@
|
||||
"""The obligation checker (authorization-obligations, Req 3–6).
|
||||
|
||||
An obligated operation is reported when it is reached (a path record ends at it, or, without path records, its function is
|
||||
an entry point) and none of the dischargers it accepts dominates it. The label says why the obligation exists: a declared
|
||||
policy clause (``declared_policy_violation``), the practice of sibling handlers (``consistency_violation``), or, in the
|
||||
degraded mode without path records, the operation fact alone (``function_local``). With path records and neither a clause
|
||||
nor a sibling norm, nothing justifies the obligation and nothing is reported (design gating 3). Findings are ``suspicion``;
|
||||
the obligation, the missing discharger, the evidence and the known fix travel as data.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from ...findings import StaticFinding
|
||||
from ..facts import SemanticFacts
|
||||
from ..ir import FileIR
|
||||
from .dominance import Dominance
|
||||
from .facts import ObligationFacts
|
||||
from .operations import Discharge, Operation, find_discharges, find_operations, valid_witness
|
||||
from .policy import DeclaredPolicy
|
||||
from .siblings import Anomaly, SiblingSet, consistency_anomalies, sibling_sets
|
||||
|
||||
LABELS = ("declared_policy_violation", "consistency_violation", "function_local")
|
||||
_ROUTE_LITERAL = re.compile(r"@\w[\w.]*\.(?:route|get|post|put|delete|patch|api_route|websocket)\(\s*['\"]([^'\"]+)['\"]")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ObligationFinding:
|
||||
operation: Operation
|
||||
missing: str # discharger kind
|
||||
provenance: str | None # when a discharger of that kind exists but binds the wrong provenance
|
||||
label: str # LABELS
|
||||
evidence: tuple[str, ...] # sibling handler ids or the policy clause
|
||||
known_fix: str | None # mechanism record id whose obligation shape matches
|
||||
sensitivity: str
|
||||
route: str | None = None
|
||||
intent: str | None = None # hunter adjudication (task 3.3): public | protected | unknown
|
||||
intent_rationale: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ObligationResult:
|
||||
findings: tuple[ObligationFinding, ...]
|
||||
discharges: tuple[Discharge, ...]
|
||||
sibling_sets: tuple[SiblingSet, ...]
|
||||
under_populated: tuple[tuple[str, str], ...] = ()
|
||||
operations: int = 0
|
||||
degradations: tuple[dict[str, object], ...] = field(default=())
|
||||
policy_version: str | None = None
|
||||
|
||||
|
||||
def check_obligations(
|
||||
*,
|
||||
irs: Mapping[str, tuple[FileIR, str]],
|
||||
entries: Sequence[object],
|
||||
facts: ObligationFacts,
|
||||
flow_facts: SemanticFacts,
|
||||
policy: DeclaredPolicy | None,
|
||||
paths: Sequence[object],
|
||||
dominance: Dominance,
|
||||
store_shapes: Sequence[object],
|
||||
min_siblings: int,
|
||||
extra_witnesses: Sequence[Discharge] = (),
|
||||
) -> ObligationResult:
|
||||
operations: list[Operation] = []
|
||||
witnesses: list[Discharge] = list(extra_witnesses)
|
||||
texts: dict[str, str] = {}
|
||||
degradations: list[dict[str, object]] = []
|
||||
unsupported: set[str] = set()
|
||||
for path, (ir, text) in irs.items():
|
||||
texts[path] = text
|
||||
if not ir.parse_ok:
|
||||
if ir.reason == "language_unsupported" and ir.language not in unsupported:
|
||||
unsupported.add(ir.language)
|
||||
degradations.append({"stage": "obligations", "reason": "obligations_language_unsupported", "language": ir.language})
|
||||
continue
|
||||
scoped = facts.for_language(ir.language)
|
||||
flow = flow_facts.for_language(ir.language)
|
||||
operations.extend(find_operations(ir, scoped, text=text))
|
||||
witnesses.extend(find_discharges(ir, scoped, flow, text=text, entries=entries))
|
||||
sets = sibling_sets(entries, operations, witnesses)
|
||||
anomalies, under = consistency_anomalies(sets, min_siblings=min_siblings)
|
||||
anomaly_index: dict[tuple[str, str], Anomaly] = {(a.handler, a.missing): a for a in anomalies}
|
||||
entry_handlers = {f"{getattr(e, 'path', '')}::{getattr(e, 'function_name', '')}" for e in entries}
|
||||
reached = _reached(operations, paths, entry_handlers)
|
||||
if not paths and operations:
|
||||
for language in sorted({irs[op.path][0].language for op in operations if op.path in irs}):
|
||||
degradations.append(
|
||||
{
|
||||
"stage": "obligations",
|
||||
"reason": "obligations_function_local",
|
||||
"language": language,
|
||||
"files": sorted({op.path for op in operations if op.path in irs and irs[op.path][0].language == language}),
|
||||
"detail": "no path records; obligations evaluated within entry-point handlers only",
|
||||
}
|
||||
)
|
||||
findings: list[ObligationFinding] = []
|
||||
invalid_rows = tuple(sorted(str(getattr(row, "id", "")) for row in store_shapes if not _usable_shape(row)))
|
||||
if invalid_rows:
|
||||
degradations.append({"stage": "obligations", "reason": "obligations_store_row_invalid", "records": list(invalid_rows)})
|
||||
for operation, path_record in reached:
|
||||
handler = f"{operation.path}::{operation.function}"
|
||||
route = _route_path(operation, texts.get(operation.path, ""))
|
||||
declared_access = policy.route_access(route) if policy is not None and route else None
|
||||
same = [w for w in witnesses if _relevant(w, operation)]
|
||||
dominated_kinds = {w.kind for w in same if valid_witness(w) and dominance.dominates(w, operation, path_record)}
|
||||
# (a) the operation's own obligation: discharged when any accepted discharger dominates. A declared public route
|
||||
# waives the path guard here too (Req 4.4), so the operation falls through to the next kind it accepts.
|
||||
required = [kind for kind in operation.requires if not (kind == "path_guard" and declared_access == "public")]
|
||||
if required and not (set(required) & dominated_kinds):
|
||||
missing = next(
|
||||
(kind for kind in required if (handler, kind) in anomaly_index),
|
||||
required[0],
|
||||
)
|
||||
finding = _finding(
|
||||
operation, missing, same, anomaly_index.get((handler, missing)), policy, route, declared_access, store_shapes, paths, facts
|
||||
)
|
||||
if finding is not None:
|
||||
findings.append(finding)
|
||||
# (b) consistency anomalies on kinds the siblings practice beyond the operation's own requirement (a path guard, say)
|
||||
for kind in sorted({a.missing for (h, _), a in anomaly_index.items() if h == handler} - set(operation.requires)):
|
||||
if kind in dominated_kinds:
|
||||
continue
|
||||
if kind == "path_guard" and declared_access == "public":
|
||||
continue # declared public: the guard obligation is waived, the identity one is not
|
||||
finding = _finding(
|
||||
operation, kind, same, anomaly_index[(handler, kind)], policy, route, declared_access, store_shapes, paths, facts
|
||||
)
|
||||
if finding is not None:
|
||||
findings.append(finding)
|
||||
return ObligationResult(
|
||||
findings=tuple(_dedupe(findings)),
|
||||
discharges=tuple(witnesses),
|
||||
sibling_sets=sets,
|
||||
under_populated=under,
|
||||
operations=len(operations),
|
||||
degradations=tuple(degradations),
|
||||
policy_version=policy.version_hash if policy is not None else None,
|
||||
)
|
||||
|
||||
|
||||
def findings_to_static(result: ObligationResult) -> list[StaticFinding]:
|
||||
statics: list[StaticFinding] = []
|
||||
for finding in result.findings:
|
||||
op = finding.operation
|
||||
tags = [f"obligation:{op.kind}", f"discharger:{finding.missing}", f"obligation_evidence:{finding.label}"]
|
||||
if op.resource:
|
||||
tags.append(f"resource:{op.resource}")
|
||||
if finding.known_fix:
|
||||
tags.append(f"mechanism:{finding.known_fix}")
|
||||
why = "; ".join(finding.evidence) if finding.evidence else "the operation fact alone (function-local mode)"
|
||||
resource = op.resource or "a resource"
|
||||
provenance = f" (a {finding.missing} is present but bound from {finding.provenance})" if finding.provenance else ""
|
||||
statics.append(
|
||||
StaticFinding(
|
||||
finding_id=f"obligation:{op.kind}:{op.path}:{op.line}:{finding.missing}",
|
||||
path=op.path,
|
||||
title=f"Obligation not discharged: {op.kind} on {resource} without {finding.missing}",
|
||||
severity={"high": "high", "medium": "medium", "low": "low"}.get(finding.sensitivity, "medium"),
|
||||
confidence="low",
|
||||
evidence_level="suspicion",
|
||||
rationale=(
|
||||
f"{op.kind} on {resource} in {op.function} is reached without a dominating {finding.missing}{provenance}. "
|
||||
f"Why the obligation exists: {why}. Label: {finding.label}."
|
||||
+ (f" Known fix learned from the corpus: {finding.known_fix}." if finding.known_fix else "")
|
||||
),
|
||||
line=op.line,
|
||||
function_name=op.function,
|
||||
reachability_status="unknown",
|
||||
reachability_evidence=[],
|
||||
reachability_conditions=[],
|
||||
tags=tags,
|
||||
ranking_priority=0.0,
|
||||
)
|
||||
)
|
||||
return statics
|
||||
|
||||
|
||||
def _finding(
|
||||
operation: Operation,
|
||||
missing: str,
|
||||
same: Sequence[Discharge],
|
||||
anomaly: Anomaly | None,
|
||||
policy: DeclaredPolicy | None,
|
||||
route: str | None,
|
||||
declared_access: str | None,
|
||||
store_shapes: Sequence[object],
|
||||
paths: Sequence[object],
|
||||
facts: ObligationFacts,
|
||||
) -> ObligationFinding | None:
|
||||
wrong = next((w for w in same if w.kind == missing and not valid_witness(w)), None)
|
||||
declared = policy.resources.get(operation.resource or "") if policy is not None else None
|
||||
# A declared clause covers the obligation it speaks about: a protected resource covers identity and ownership
|
||||
# dischargers; a route declared authenticated or role-restricted covers the path guard.
|
||||
resource_clause = declared is not None and missing in {"identity_constraint", "ownership_check"}
|
||||
route_clause = missing == "path_guard" and declared_access in {"authenticated", "role"}
|
||||
if resource_clause and declared is not None:
|
||||
label = "declared_policy_violation"
|
||||
clause = f"resource {declared.name} declared sensitivity {declared.sensitivity}"
|
||||
if declared.identity_field:
|
||||
clause += f", identity field {declared.identity_field}"
|
||||
evidence: tuple[str, ...] = (clause,)
|
||||
if anomaly is not None:
|
||||
evidence = evidence + anomaly.siblings
|
||||
elif route_clause:
|
||||
label = "declared_policy_violation"
|
||||
evidence = (f"route {route} declared {declared_access}",) + (anomaly.siblings if anomaly is not None else ())
|
||||
elif anomaly is not None:
|
||||
label = "consistency_violation"
|
||||
evidence = anomaly.siblings + anomaly.evidence[len(anomaly.siblings) :]
|
||||
elif paths:
|
||||
return None # path-aware mode: without a clause or a sibling norm, nothing justifies the obligation (gating 3)
|
||||
else:
|
||||
label = "function_local" # degraded mode: the operation fact alone, recorded as a degradation
|
||||
evidence = ()
|
||||
sensitivity = (declared.sensitivity if declared is not None else None) or _fact_sensitivity(operation, facts)
|
||||
return ObligationFinding(
|
||||
operation=operation,
|
||||
missing=missing,
|
||||
provenance=wrong.provenance if wrong is not None else None,
|
||||
label=label,
|
||||
evidence=evidence,
|
||||
known_fix=_known_fix(operation, missing, store_shapes),
|
||||
sensitivity=sensitivity,
|
||||
route=route,
|
||||
)
|
||||
|
||||
|
||||
def _reached(operations: Sequence[Operation], paths: Sequence[object], entry_handlers: set[str]) -> list[tuple[Operation, object | None]]:
|
||||
if not paths:
|
||||
return [(op, None) for op in operations if f"{op.path}::{op.function}" in entry_handlers]
|
||||
reached: list[tuple[Operation, object | None]] = []
|
||||
for op in operations:
|
||||
for record in paths:
|
||||
hops = getattr(record, "hops", None) or ()
|
||||
if not hops:
|
||||
continue
|
||||
last = hops[-1][0] if isinstance(hops[-1], (tuple, list)) else str(hops[-1])
|
||||
if str(last) == f"{op.path}::{op.function}":
|
||||
reached.append((op, record))
|
||||
break
|
||||
return reached
|
||||
|
||||
|
||||
def _relevant(witness: Discharge, operation: Operation) -> bool:
|
||||
return witness.scope == "hop" or (witness.path, witness.function) == (operation.path, operation.function)
|
||||
|
||||
|
||||
def _usable_shape(record: object) -> bool:
|
||||
"""A store row is usable when its shape dict names both an operation kind and a discharger kind."""
|
||||
shape = getattr(record, "shape", None)
|
||||
return isinstance(shape, dict) and bool(shape.get("operation_kind")) and bool(shape.get("discharger_kind"))
|
||||
|
||||
|
||||
def _known_fix(operation: Operation, missing: str, store_shapes: Sequence[object]) -> str | None:
|
||||
for record in store_shapes:
|
||||
if not _usable_shape(record):
|
||||
continue
|
||||
shape = getattr(record, "shape", None) or {}
|
||||
if isinstance(shape, dict) and shape.get("operation_kind") == operation.kind and shape.get("discharger_kind") == missing:
|
||||
return str(getattr(record, "id", "")) or None
|
||||
return None
|
||||
|
||||
|
||||
def _fact_sensitivity(operation: Operation, facts: ObligationFacts) -> str:
|
||||
for fact in facts.operations:
|
||||
if fact.id == operation.fact_id:
|
||||
return fact.sensitivity
|
||||
return "medium"
|
||||
|
||||
|
||||
def _route_path(operation: Operation, text: str) -> str | None:
|
||||
lines = text.splitlines()
|
||||
index = _def_index(lines, operation.function)
|
||||
if index is None:
|
||||
return None
|
||||
cursor = index - 1
|
||||
while cursor >= 0 and lines[cursor].strip().startswith("@"):
|
||||
match = _ROUTE_LITERAL.search(lines[cursor].strip())
|
||||
if match:
|
||||
return match.group(1)
|
||||
cursor -= 1
|
||||
return None
|
||||
|
||||
|
||||
def _def_index(lines: Sequence[str], function: str) -> int | None:
|
||||
pattern = re.compile(rf"^\s*(?:async\s+)?(?:def|function)\s+{re.escape(function)}\b")
|
||||
for index, line in enumerate(lines):
|
||||
if pattern.match(line):
|
||||
return index
|
||||
return None
|
||||
|
||||
|
||||
def _dedupe(findings: Sequence[ObligationFinding]) -> list[ObligationFinding]:
|
||||
seen: set[tuple[str, int, str, str]] = set()
|
||||
out: list[ObligationFinding] = []
|
||||
for finding in findings:
|
||||
key = (finding.operation.path, finding.operation.line, finding.operation.kind, finding.missing)
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
out.append(finding)
|
||||
return out
|
||||
|
||||
|
||||
__all__ = ["LABELS", "ObligationFinding", "ObligationResult", "check_obligations", "findings_to_static"]
|
||||
@@ -0,0 +1,221 @@
|
||||
"""authorization-obligations task 2.5: the checker, function-local and path-aware (offline)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
from test_obligation_operations import APP, ENTRIES, _entry # noqa: E402
|
||||
|
||||
from openultrasast.semantic.facts import load_facts # noqa: E402
|
||||
from openultrasast.semantic.ir import parse_file # noqa: E402
|
||||
|
||||
POLICY = (
|
||||
'version = 1\nidentity_source = "request.user"\n\n[[resource]]\nname = "book"\nsensitivity = "high"\nidentity_field = "user_id"\n\n'
|
||||
'[[route]]\npath = "/books/any/*"\naccess = "public"\n'
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _Path:
|
||||
hops: tuple[tuple[str, int], ...]
|
||||
|
||||
|
||||
def _check(
|
||||
*, policy_text: str | None = None, paths=(), dominance=None, min_siblings: int = 2, store_shapes=(), tmp_path: Path | None = None
|
||||
):
|
||||
from openultrasast.semantic.obligations import load_obligation_facts
|
||||
from openultrasast.semantic.obligations.check import check_obligations
|
||||
from openultrasast.semantic.obligations.dominance import OrderDominance
|
||||
from openultrasast.semantic.obligations.policy import load_declared_policy
|
||||
|
||||
policy = None
|
||||
if policy_text is not None and tmp_path is not None:
|
||||
(tmp_path / "obligations.toml").write_text(policy_text)
|
||||
policy = load_declared_policy(tmp_path / "obligations.toml")
|
||||
ir = parse_file("app.py", APP, "python")
|
||||
return check_obligations(
|
||||
irs={"app.py": (ir, APP)},
|
||||
entries=ENTRIES,
|
||||
facts=load_obligation_facts(),
|
||||
flow_facts=load_facts(),
|
||||
policy=policy,
|
||||
paths=paths,
|
||||
dominance=dominance or OrderDominance(texts={"app.py": APP}),
|
||||
store_shapes=store_shapes,
|
||||
min_siblings=min_siblings,
|
||||
)
|
||||
|
||||
|
||||
def test_function_local_run_reports_the_consistency_violation_with_siblings_and_a_degradation() -> None:
|
||||
result = _check()
|
||||
identity = [f for f in result.findings if f.missing == "identity_constraint"]
|
||||
assert {f.operation.function for f in identity} == {"leaky", "guarded"}
|
||||
leaky = next(f for f in identity if f.operation.function == "leaky")
|
||||
assert leaky.label == "consistency_violation" and leaky.provenance == "request_input"
|
||||
assert "app.py::constrained" in leaky.evidence and "app.py::constrained_too" in leaky.evidence # names both siblings
|
||||
assert leaky.operation.resource == "book" and leaky.operation.kind == "protected_read"
|
||||
assert not [f for f in result.findings if f.operation.function == "constrained" and f.missing == "identity_constraint"]
|
||||
(degradation,) = [d for d in result.degradations if d["reason"] == "obligations_function_local"]
|
||||
assert degradation["language"] == "python" and degradation["files"] == ["app.py"] # Req 5.3: names the file
|
||||
assert result.sibling_sets and result.discharges
|
||||
|
||||
|
||||
def test_declared_policy_outranks_consistency_and_public_routes_drop_only_the_guard(tmp_path: Path) -> None:
|
||||
result = _check(policy_text=POLICY, tmp_path=tmp_path)
|
||||
leaky = [f for f in result.findings if f.operation.function == "leaky"]
|
||||
identity = next(f for f in leaky if f.missing == "identity_constraint")
|
||||
assert identity.label == "declared_policy_violation" and any("resource book" in e for e in identity.evidence)
|
||||
assert not [f for f in leaky if f.missing == "path_guard"] # /books/any/* is declared public
|
||||
constrained_guard = [f for f in result.findings if f.operation.function == "constrained_too" and f.missing == "path_guard"]
|
||||
assert constrained_guard and constrained_guard[0].label == "consistency_violation" # not declared public: the guard anomaly stands
|
||||
|
||||
|
||||
def test_path_records_gate_reachability_and_a_dominating_hop_discharges() -> None:
|
||||
from openultrasast.semantic.obligations.operations import Discharge
|
||||
|
||||
class _HopAware:
|
||||
def __init__(self) -> None:
|
||||
self.calls = 0
|
||||
|
||||
def dominates(self, witness: Discharge, operation, path) -> bool: # type: ignore[no-untyped-def]
|
||||
self.calls += 1
|
||||
return witness.scope == "hop"
|
||||
|
||||
reached = (_Path(hops=(("api.py::gateway", 3), ("app.py::leaky", 25))),)
|
||||
result = _check(paths=reached)
|
||||
assert {f.operation.function for f in result.findings} == {"leaky"} # only the operation a path reaches
|
||||
assert not any(d["reason"] == "obligations_function_local" for d in result.degradations)
|
||||
dominance = _HopAware()
|
||||
hop_guard = Discharge(
|
||||
path="api.py", line=1, function="gateway", kind="identity_constraint", fact_id="x", provenance="authenticated_context", scope="hop"
|
||||
)
|
||||
covered = _check(paths=reached, dominance=dominance, store_shapes=(), min_siblings=2)
|
||||
assert dominance.calls == 0 # leaky's only witness is request-bound, so there is nothing valid to ask dominance about
|
||||
from openultrasast.semantic.obligations import load_obligation_facts
|
||||
from openultrasast.semantic.obligations.check import check_obligations
|
||||
|
||||
ir = parse_file("app.py", APP, "python")
|
||||
with_hop = check_obligations(
|
||||
irs={"app.py": (ir, APP)},
|
||||
entries=ENTRIES,
|
||||
facts=load_obligation_facts(),
|
||||
flow_facts=load_facts(),
|
||||
policy=None,
|
||||
paths=reached,
|
||||
dominance=dominance,
|
||||
store_shapes=(),
|
||||
min_siblings=2,
|
||||
extra_witnesses=(hop_guard,),
|
||||
)
|
||||
assert dominance.calls > 0 # the hop witness is valid, so the checker consulted the injected Dominance implementation
|
||||
assert not [f for f in with_hop.findings if f.operation.function == "leaky" and f.missing == "identity_constraint"]
|
||||
assert covered.findings # without the hop witness the finding stands
|
||||
|
||||
|
||||
def test_findings_become_suspicion_static_findings_with_known_fix_and_labels(tmp_path: Path) -> None:
|
||||
from openultrasast.semantic.mechanisms import MechanismStore, append_from_pair
|
||||
from openultrasast.semantic.obligations.check import findings_to_static
|
||||
from openultrasast.semantic.obligations.shapes import ObligationShape, obligation_mechanisms
|
||||
|
||||
store = MechanismStore(tmp_path / "m.jsonl")
|
||||
shape = ObligationShape(
|
||||
language="python",
|
||||
operation_kind="protected_read",
|
||||
discharger_kind="identity_constraint",
|
||||
provenance="authenticated_context",
|
||||
resource_class="owned",
|
||||
mechanism="missing_auth_guard",
|
||||
)
|
||||
record = append_from_pair(store, shape, summary="s", cwe="CWE-639", pair="vampi", provenance="human", tier="seeded")
|
||||
result = _check(store_shapes=obligation_mechanisms(store.load()))
|
||||
leaky = next(f for f in result.findings if f.operation.function == "leaky" and f.missing == "identity_constraint")
|
||||
assert leaky.known_fix == record.id
|
||||
statics = findings_to_static(result)
|
||||
ids = {s.finding_id for s in statics}
|
||||
guarded = next(f for f in result.findings if f.operation.function == "guarded" and f.missing == "identity_constraint")
|
||||
assert f"obligation:protected_read:app.py:{leaky.operation.line}:identity_constraint" in ids # exact id contract
|
||||
assert f"obligation:protected_read:app.py:{guarded.operation.line}:identity_constraint" in ids
|
||||
static = next(s for s in statics if s.line == leaky.operation.line and "discharger:identity_constraint" in s.tags)
|
||||
assert static.evidence_level == "suspicion"
|
||||
assert {
|
||||
"obligation:protected_read",
|
||||
"discharger:identity_constraint",
|
||||
"obligation_evidence:consistency_violation",
|
||||
f"mechanism:{record.id}",
|
||||
} <= set(static.tags)
|
||||
assert "app.py::constrained" in static.rationale
|
||||
|
||||
|
||||
def test_with_path_records_an_unjustified_obligation_yields_no_finding() -> None:
|
||||
"""Round-2 finding: `function_local` is the degraded mode's label only; with paths and no sibling or clause, nothing."""
|
||||
reached = (_Path(hops=(("api.py::gateway", 3), ("app.py::leaky", 25))),)
|
||||
result = _check(paths=reached, min_siblings=5) # every set under-populated: no consistency evidence, no policy
|
||||
assert result.findings == () and result.under_populated == (("app", "book"),)
|
||||
assert not any(d["reason"] == "obligations_function_local" for d in result.degradations)
|
||||
assert all(f.label != "consistency_violation" or f.evidence for f in _check(paths=reached, min_siblings=2).findings)
|
||||
|
||||
|
||||
ADMIN = (
|
||||
"from flask import request\n\n\n"
|
||||
"@app.route('/admin/reset/<name>')\ndef reset(name):\n"
|
||||
" set_password(name, 'temporary')\n"
|
||||
" return Book.query.filter_by(book_title=name).first()\n"
|
||||
)
|
||||
ADMIN_ENTRIES = [_entry("reset", 4, 6, "public", [])]
|
||||
ADMIN_POLICY = (
|
||||
'version = 1\nidentity_source = "request.user"\n\n[[resource]]\nname = "book"\nsensitivity = "high"\n\n'
|
||||
'[[route]]\npath = "/admin/*"\naccess = "public"\n'
|
||||
)
|
||||
|
||||
|
||||
def _check_admin(tmp_path: Path, policy_text: str | None):
|
||||
from openultrasast.semantic.obligations import load_obligation_facts
|
||||
from openultrasast.semantic.obligations.check import check_obligations
|
||||
from openultrasast.semantic.obligations.dominance import OrderDominance
|
||||
from openultrasast.semantic.obligations.policy import load_declared_policy
|
||||
|
||||
policy = None
|
||||
if policy_text is not None:
|
||||
(tmp_path / "obligations.toml").write_text(policy_text)
|
||||
policy = load_declared_policy(tmp_path / "obligations.toml")
|
||||
ir = parse_file("app.py", ADMIN, "python")
|
||||
return check_obligations(
|
||||
irs={"app.py": (ir, ADMIN)},
|
||||
entries=ADMIN_ENTRIES,
|
||||
facts=load_obligation_facts(),
|
||||
flow_facts=load_facts(),
|
||||
policy=policy,
|
||||
paths=(),
|
||||
dominance=OrderDominance(texts={"app.py": ADMIN}),
|
||||
store_shapes=(),
|
||||
min_siblings=2,
|
||||
)
|
||||
|
||||
|
||||
def test_a_declared_public_route_never_reports_its_missing_path_guard_but_keeps_the_resource_obligation(tmp_path: Path) -> None:
|
||||
"""Round-3 finding (Req 4.4): the waiver must also cover the operation's own requirement, not only sibling anomalies."""
|
||||
without = _check_admin(tmp_path, None)
|
||||
assert [(f.operation.kind, f.missing) for f in without.findings if f.operation.kind == "privileged_action"] == [
|
||||
("privileged_action", "path_guard")
|
||||
]
|
||||
declared = _check_admin(tmp_path, ADMIN_POLICY)
|
||||
assert not [f for f in declared.findings if f.missing == "path_guard"] # /admin/* is declared public: no guard obligation
|
||||
identity = [f for f in declared.findings if f.operation.kind == "protected_read" and f.missing == "identity_constraint"]
|
||||
assert identity and identity[0].label == "declared_policy_violation" # the declared resource still owes its identity constraint
|
||||
|
||||
|
||||
def test_a_malformed_store_row_is_a_degradation_not_a_crash() -> None:
|
||||
"""Design Error Handling: store row with malformed obligation shape -> `obligations_store_row_invalid`."""
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _Row:
|
||||
id: str
|
||||
shape: object
|
||||
|
||||
result = _check(store_shapes=(_Row(id="corpus:bad", shape="garbage"), _Row(id="corpus:worse", shape={"family": "obligation"})))
|
||||
(degradation,) = [d for d in result.degradations if d["reason"] == "obligations_store_row_invalid"]
|
||||
assert degradation["records"] == ["corpus:bad", "corpus:worse"]
|
||||
assert all(f.known_fix is None for f in result.findings)
|
||||
Reference in New Issue
Block a user