diff --git a/src/openultrasast/ruleset/obligations/javascript.toml b/src/openultrasast/ruleset/obligations/javascript.toml index ae5df3a..2145340 100644 --- a/src/openultrasast/ruleset/obligations/javascript.toml +++ b/src/openultrasast/ruleset/obligations/javascript.toml @@ -12,7 +12,7 @@ sensitivity = "high" [[operation]] id = "orm-write" kind = "protected_write" -calls = ["update", "updateOne", "updateMany", "deleteOne", "deleteMany", "delete", "create", "insert", "upsert", "save", "remove", "findByIdAndUpdate", "findByIdAndDelete"] +calls = ["updateOne", "updateMany", "deleteOne", "deleteMany", "findByIdAndUpdate", "findByIdAndDelete", "upsert", "insertOne", "insertMany", "prisma.update", "prisma.delete", "prisma.create"] requires = ["identity_constraint", "ownership_check"] sensitivity = "high" @@ -33,7 +33,7 @@ sensitivity = "high" [[operation]] id = "security-setting" kind = "security_setting" -calls = ["cors", "cookie", "session", "helmet", "listen", "use", "setHeader", "createClient", "from"] +calls = ["cors", "cookie", "session", "helmet", "app.listen", "setHeader", "createClient"] resource_arg = 0 requires = ["non_permissive_value"] sensitivity = "medium" diff --git a/src/openultrasast/ruleset/obligations/python.toml b/src/openultrasast/ruleset/obligations/python.toml index b9f2233..84a54fc 100644 --- a/src/openultrasast/ruleset/obligations/python.toml +++ b/src/openultrasast/ruleset/obligations/python.toml @@ -12,7 +12,7 @@ sensitivity = "high" [[operation]] id = "orm-write" kind = "protected_write" -calls = ["add", "delete", "update", "save", "commit", "insert_one", "update_one", "delete_one", "bulk_update"] +calls = ["session.add", "session.delete", "session.merge", "query.update", "query.delete", "objects.update", "objects.create", "objects.filter", "insert_one", "update_one", "delete_one", "bulk_update"] requires = ["identity_constraint", "ownership_check"] sensitivity = "high" @@ -33,7 +33,7 @@ sensitivity = "high" [[operation]] id = "security-setting" kind = "security_setting" -calls = ["CORS", "cross_origin", "set_cookie", "Fernet", "run", "config.update", "TrustedHostMiddleware", "add_middleware", "set_debug"] +calls = ["CORS", "cross_origin", "set_cookie", "app.run", "config.update", "TrustedHostMiddleware", "add_middleware", "set_debug"] resource_arg = 0 requires = ["non_permissive_value"] sensitivity = "medium" @@ -42,7 +42,7 @@ sensitivity = "medium" id = "route-guard" kind = "path_guard" decorators = ["login_required", "jwt_required", "auth_required", "roles_required", "permission_required", "admin_required", "requires_auth", "authenticated", "token_required"] -calls = ["token_validator", "verify_jwt_in_request", "get_jwt_identity", "check_auth", "require_login", "authenticate", "abort", "login_required", "current_user_required"] +calls = ["token_validator", "verify_jwt_in_request", "get_jwt_identity", "check_auth", "require_login", "authenticate", "login_required", "current_user_required"] [[discharger]] id = "identity-from-context" @@ -59,7 +59,7 @@ calls = ["check_owner", "is_owner", "assert_owner", "ensure_owner", "owns", "can [[discharger]] id = "safe-setting" kind = "non_permissive_value" -calls = ["CORS", "cross_origin", "set_cookie", "run", "config.update", "add_middleware"] +calls = ["CORS", "cross_origin", "set_cookie", "app.run", "config.update", "add_middleware"] permissive_values = ["'*'", "\"*\"", "True", "0.0.0.0", "None", "False", "allow_all", "disable"] [[discharger]] diff --git a/src/openultrasast/ruleset/obligations/typescript.toml b/src/openultrasast/ruleset/obligations/typescript.toml index 880a3ba..8e3c118 100644 --- a/src/openultrasast/ruleset/obligations/typescript.toml +++ b/src/openultrasast/ruleset/obligations/typescript.toml @@ -12,7 +12,7 @@ sensitivity = "high" [[operation]] id = "orm-write" kind = "protected_write" -calls = ["update", "updateOne", "updateMany", "deleteOne", "deleteMany", "delete", "create", "insert", "upsert", "save", "remove", "findByIdAndUpdate", "findByIdAndDelete"] +calls = ["updateOne", "updateMany", "deleteOne", "deleteMany", "findByIdAndUpdate", "findByIdAndDelete", "upsert", "insertOne", "insertMany", "prisma.update", "prisma.delete", "prisma.create"] requires = ["identity_constraint", "ownership_check"] sensitivity = "high" @@ -33,7 +33,7 @@ sensitivity = "high" [[operation]] id = "security-setting" kind = "security_setting" -calls = ["cors", "cookie", "session", "helmet", "listen", "use", "setHeader", "createClient", "from"] +calls = ["cors", "cookie", "session", "helmet", "app.listen", "setHeader", "createClient"] resource_arg = 0 requires = ["non_permissive_value"] sensitivity = "medium" diff --git a/src/openultrasast/semantic/obligations/__init__.py b/src/openultrasast/semantic/obligations/__init__.py index bc2ea9a..c08480a 100644 --- a/src/openultrasast/semantic/obligations/__init__.py +++ b/src/openultrasast/semantic/obligations/__init__.py @@ -6,6 +6,7 @@ ownership check, a non-permissive value, validated input). Facts are closed data findings never claim a proof rung. """ +from .check import ObligationFinding, ObligationResult, check_obligations, findings_to_static from .facts import ( DISCHARGER_KINDS, OPERATION_KINDS, @@ -25,10 +26,14 @@ __all__ = [ "DischargerFact", "ObligationFacts", "ObligationFactsError", + "ObligationFinding", + "ObligationResult", "ObligationShape", "OperationFact", + "check_obligations", "derive_obligation", "explain_skip", + "findings_to_static", "load_obligation_facts", "obligation_mechanisms", ] diff --git a/src/openultrasast/semantic/obligations/operations.py b/src/openultrasast/semantic/obligations/operations.py index cd88ea7..58770cb 100644 --- a/src/openultrasast/semantic/obligations/operations.py +++ b/src/openultrasast/semantic/obligations/operations.py @@ -103,6 +103,11 @@ def find_discharges( return tuple(_dedupe(witnesses)) +def valid_witness(witness: Discharge) -> bool: + """An identity constraint witnesses a discharge only when its value comes from the authenticated context.""" + return witness.kind != "identity_constraint" or witness.provenance == "authenticated_context" + + def covers(witness: Discharge, operation: Operation) -> bool: """A witness discharges an operation when it sits in the same function, has a kind the operation requires, and, for an identity constraint, binds its value from the authenticated context. Ordering along the path is the Dominance protocol's @@ -111,7 +116,7 @@ def covers(witness: Discharge, operation: Operation) -> bool: return False if witness.kind not in operation.requires: return False - if witness.kind == "identity_constraint" and witness.provenance != "authenticated_context": + if not valid_witness(witness): return False return not (witness.kind == "non_permissive_value" and witness.line != operation.line) @@ -130,7 +135,7 @@ def _decorator_witnesses(path: str, function: FunctionIR, fact: DischargerFact, out: list[Discharge] = [] for decorator in decorators: name = _decorator_name(decorator) - if any(candidate == name or candidate in decorator for candidate in fact.decorators): + if any(candidate == name for candidate in fact.decorators): # the parsed decorator name only, never free text out.append( Discharge( path=path, @@ -245,4 +250,4 @@ def _dedupe(witnesses: Sequence[Discharge]) -> list[Discharge]: return out -__all__ = ["SCOPES", "Discharge", "Operation", "covers", "find_discharges", "find_operations"] +__all__ = ["SCOPES", "Discharge", "Operation", "covers", "find_discharges", "find_operations", "valid_witness"] diff --git a/src/openultrasast/semantic/obligations/siblings.py b/src/openultrasast/semantic/obligations/siblings.py new file mode 100644 index 0000000..879a74a --- /dev/null +++ b/src/openultrasast/semantic/obligations/siblings.py @@ -0,0 +1,125 @@ +"""Sibling handler sets and consistency anomalies (authorization-obligations, Req 3). + +Handlers that operate on the same resource under the same router or module are siblings. When at least the configured +minimum number of siblings discharge an obligation and one does not (Req 3.2), that one is the anomaly; its evidence is +the siblings that discharge. A set with fewer members than the minimum is counted as under-populated and never produces +a finding (Req 3.3). Nothing here touches other findings. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass, field +from pathlib import Path + +from .facts import DISCHARGER_KINDS +from .operations import Discharge, Operation, valid_witness + + +@dataclass(frozen=True) +class SiblingSet: + key: tuple[str, str] # (module path without suffix, resource token) + handlers: tuple[str, ...] # "path::function" + discharging: dict[str, tuple[str, ...]] # discharger kind -> handlers that carry a valid witness of that kind + operation_kinds: tuple[str, ...] + access_levels: dict[str, str] = field(default_factory=dict) # handler -> entry-point access classification + + +@dataclass(frozen=True) +class Anomaly: + handler: str + missing: str # discharger kind the siblings practice and this handler lacks + operation_kind: str + siblings: tuple[str, ...] # handlers that discharge + key: tuple[str, str] + access_level: str = "unknown" + evidence: tuple[str, ...] = () + + +def sibling_sets(entries: Sequence[object], operations: Sequence[Operation], discharges: Sequence[Discharge]) -> tuple[SiblingSet, ...]: + """Group entry-point handlers by (module, resource) over the operations they perform; record who discharges what.""" + by_handler_ops: dict[str, list[Operation]] = {} + for operation in operations: + by_handler_ops.setdefault(f"{operation.path}::{operation.function}", []).append(operation) + access: dict[str, str] = {} + for entry in entries: + name = getattr(entry, "function_name", None) + path = getattr(entry, "path", None) + if name and path: + access[f"{path}::{name}"] = str(getattr(entry, "access_level", "unknown")) + valid: dict[str, set[str]] = {} + for witness in discharges: + if valid_witness(witness): + valid.setdefault(f"{witness.path}::{witness.function}", set()).add(witness.kind) + groups: dict[tuple[str, str], dict[str, list[Operation]]] = {} + for handler, ops in by_handler_ops.items(): + if handler not in access: + continue # only entry-point handlers are siblings; helpers are reached through them + module = _module_of(handler.split("::", 1)[0]) + for operation in ops: + if operation.resource is None: + continue + groups.setdefault((module, operation.resource), {}).setdefault(handler, []).append(operation) + sets: list[SiblingSet] = [] + for key, members in sorted(groups.items()): + handlers = tuple(sorted(members)) + discharging = { + kind: tuple(sorted(handler for handler in handlers if kind in valid.get(handler, set()))) for kind in DISCHARGER_KINDS + } + kinds = tuple(sorted({operation.kind for ops in members.values() for operation in ops})) + sets.append( + SiblingSet( + key=key, + handlers=handlers, + discharging={kind: who for kind, who in discharging.items() if who}, + operation_kinds=kinds, + access_levels={handler: access.get(handler, "unknown") for handler in handlers}, + ) + ) + return tuple(sets) + + +def consistency_anomalies(sets: Sequence[SiblingSet], *, min_siblings: int) -> tuple[tuple[Anomaly, ...], tuple[tuple[str, str], ...]]: + """Anomalies where at least ``min_siblings`` siblings discharge a kind another sibling lacks; sets with fewer members than + ``min_siblings`` are returned as under-populated instead. One threshold, read as Req 3.2 states it.""" + anomalies: list[Anomaly] = [] + under: list[tuple[str, str]] = [] + for group in sets: + if len(group.handlers) < min_siblings: + under.append(group.key) + continue + for kind, dischargers in sorted(group.discharging.items()): + if len(dischargers) < min_siblings: + continue # one sibling does not set a norm + for handler in group.handlers: + if handler in dischargers: + continue + access = group.access_levels.get(handler, "unknown") + evidence: list[str] = [f"{sibling} discharges {kind}" for sibling in dischargers] + if kind == "path_guard": + evidence.append( + f"route access: {access}; siblings: " + + ", ".join(sorted({group.access_levels.get(s, "unknown") for s in dischargers})) + ) + for operation_kind in group.operation_kinds: + anomalies.append( + Anomaly( + handler=handler, + missing=kind, + operation_kind=operation_kind, + siblings=dischargers, + key=group.key, + access_level=access, + evidence=tuple(evidence), + ) + ) + return tuple(anomalies), tuple(under) + + +def _module_of(path: str) -> str: + """The module path without its suffix: `api/v1/books.py` -> `api/v1/books`. Two same-named files in different + directories are different modules, so their handlers are never siblings.""" + return Path(path).with_suffix("").as_posix() + + +__all__ = ["Anomaly", "SiblingSet", "consistency_anomalies", "sibling_sets"] diff --git a/tests/test_obligation_facts.py b/tests/test_obligation_facts.py index 15219a2..c13b054 100644 --- a/tests/test_obligation_facts.py +++ b/tests/test_obligation_facts.py @@ -60,9 +60,6 @@ def test_loader_rejects_kinds_and_fields_outside_the_closed_sets(tmp_path: Path) (tmp_path / "python.toml").write_text(good + '\n[[discharger]]\nid = "g"\nkind = "maybe_guard"\ndecorators = ["login_required"]\n') with pytest.raises(ObligationFactsError, match="maybe_guard"): load_obligation_facts(tmp_path) - (tmp_path / "python.toml").write_text(good + '\n[[discharger]]\nid = "g"\nkind = "maybe_guard"\ndecorators = ["login_required"]\n') - with pytest.raises(ObligationFactsError, match="maybe_guard"): - load_obligation_facts(tmp_path) def test_flow_facts_loader_ignores_the_obligation_facts_directory() -> None: diff --git a/tests/test_obligation_operations.py b/tests/test_obligation_operations.py index 3680a54..f4c9d65 100644 --- a/tests/test_obligation_operations.py +++ b/tests/test_obligation_operations.py @@ -10,9 +10,12 @@ APP = ( "from flask import request\n\n\n" "@app.route('/books/')\n@login_required\ndef guarded(title):\n" " return Book.query.filter_by(book_title=title).first()\n\n\n" - "@app.route('/books/mine/<title>')\ndef constrained(title):\n" + "@app.route('/books/mine/<title>')\n@login_required\ndef constrained(title):\n" " user_id = request.user.id\n" " return Book.query.filter_by(user_id=user_id, book_title=title).first()\n\n\n" + "@app.route('/books/shared/<title>')\ndef constrained_too(title):\n" + " owner = current_user.id\n" + " return Book.query.filter_by(owner=owner, book_title=title).first()\n\n\n" "@app.route('/books/any/<title>')\ndef leaky(title):\n" " owner = request.args.get('owner')\n" " return Book.query.filter_by(user_id=owner, book_title=title).first()\n\n\n" @@ -39,9 +42,10 @@ def _entry(function: str, line: int, end: int, access: str, evidence: list[str]) ENTRIES = [ _entry("guarded", 6, 7, "authenticated", ["@login_required"]), - _entry("constrained", 11, 13, "public", []), - _entry("leaky", 17, 19, "public", []), - _entry("health", 23, 24, "public", []), + _entry("constrained", 12, 14, "authenticated", ["@login_required"]), + _entry("constrained_too", 18, 20, "public", []), + _entry("leaky", 24, 26, "public", []), + _entry("health", 30, 31, "public", []), ] @@ -54,6 +58,7 @@ def test_find_operations_names_kind_fact_function_and_resource() -> None: assert [(op.function, op.kind, op.resource) for op in ops] == [ ("guarded", "protected_read", "book"), ("constrained", "protected_read", "book"), + ("constrained_too", "protected_read", "book"), ("leaky", "protected_read", "book"), ] assert all(op.path == "app.py" and op.fact_id == "orm-read" and op.line > 0 for op in ops) @@ -73,6 +78,7 @@ def test_find_discharges_reports_witness_kind_provenance_and_scope() -> None: by_function.setdefault(w.function, []).append((w.kind, w.provenance, w.scope)) assert ("path_guard", "authenticated_context", "decorator") in by_function["guarded"] assert ("identity_constraint", "authenticated_context", "statement") in by_function["constrained"] + assert ("identity_constraint", "authenticated_context", "statement") in by_function["constrained_too"] assert ("identity_constraint", "request_input", "statement") in by_function["leaky"] # present, but discharges nothing assert "health" not in by_function assert all(w.path == "app.py" and w.fact_id for w in witnesses) @@ -87,6 +93,6 @@ def test_discharge_witness_covers_operation_only_when_kind_and_provenance_fit() ops = {op.function: op for op in find_operations(ir, facts, text=APP)} witnesses = find_discharges(ir, facts, load_facts().for_language("python"), text=APP, entries=ENTRIES) covering = {op.function: [w.kind for w in witnesses if covers(w, op)] for op in ops.values()} - assert covering["constrained"] == ["identity_constraint"] + assert covering["constrained"] == ["identity_constraint"] and covering["constrained_too"] == ["identity_constraint"] assert covering["leaky"] == [] # request-bound identity does not cover assert covering["guarded"] == [] # a path guard does not cover a protected read (only privileged actions) diff --git a/tests/test_obligation_policy.py b/tests/test_obligation_policy.py index bc3abb0..dc2b109 100644 --- a/tests/test_obligation_policy.py +++ b/tests/test_obligation_policy.py @@ -42,6 +42,7 @@ def test_unknown_fields_and_values_are_rejected_by_name(tmp_path: Path) -> None: (GOOD.replace('sensitivity = "high"', 'sensitivity = "critical"'), "critical"), (GOOD + "\nallow_everything = true\n", "allow_everything"), (GOOD.replace('identity_source = "request.user"', 'identity_source = "whatever.works"'), "identity_source"), + (GOOD + '\n[[route]]\npath = "/x"\naccess = "public"\ncolor = "blue"\n', "color"), ): path.write_text(bad) with pytest.raises(PolicyError, match=needle): diff --git a/tests/test_obligation_siblings.py b/tests/test_obligation_siblings.py new file mode 100644 index 0000000..5961737 --- /dev/null +++ b/tests/test_obligation_siblings.py @@ -0,0 +1,111 @@ +"""authorization-obligations task 2.2: sibling handler sets and consistency anomalies (offline).""" + +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) +from test_obligation_operations import APP, ENTRIES # noqa: E402 + +from openultrasast.semantic.facts import load_facts # noqa: E402 +from openultrasast.semantic.ir import parse_file # noqa: E402 + + +def _inputs(): + from openultrasast.semantic.obligations import load_obligation_facts + from openultrasast.semantic.obligations.operations import find_discharges, find_operations + + ir = parse_file("app.py", APP, "python") + facts = load_obligation_facts().for_language("python") + ops = find_operations(ir, facts, text=APP) + wit = find_discharges(ir, facts, load_facts().for_language("python"), text=APP, entries=ENTRIES) + return ops, wit + + +def test_sibling_sets_group_by_router_and_resource_and_record_who_discharges() -> None: + from openultrasast.semantic.obligations.siblings import sibling_sets + + ops, wit = _inputs() + sets = sibling_sets(ENTRIES, ops, wit) + assert len(sets) == 1 + (books,) = sets + assert books.key == ("app", "book") + assert set(books.handlers) == {"app.py::guarded", "app.py::constrained", "app.py::constrained_too", "app.py::leaky"} + assert books.discharging["identity_constraint"] == ( + "app.py::constrained", + "app.py::constrained_too", + ) # leaky's request-bound one does not count + assert books.discharging["path_guard"] == ("app.py::constrained", "app.py::guarded") + assert books.operation_kinds == ("protected_read",) + + +def test_consistency_anomalies_respect_the_minimum_and_name_the_siblings() -> None: + from openultrasast.semantic.obligations.siblings import consistency_anomalies, sibling_sets + + ops, wit = _inputs() + sets = sibling_sets(ENTRIES, ops, wit) + anomalies, under = consistency_anomalies(sets, min_siblings=2) + by_handler = {(a.handler, a.missing): a for a in anomalies} + # tasks.md observable: two siblings constrain by owner -> one anomaly per unconstrained handler naming both siblings + assert by_handler[("app.py::leaky", "identity_constraint")].siblings == ("app.py::constrained", "app.py::constrained_too") + assert by_handler[("app.py::guarded", "identity_constraint")].siblings == ("app.py::constrained", "app.py::constrained_too") + assert by_handler[("app.py::leaky", "identity_constraint")].operation_kind == "protected_read" + assert ("app.py::constrained", "identity_constraint") not in by_handler + assert under == () + # Req 3.2: the configured minimum applies to how many siblings discharge, not to one sibling setting the norm + anomalies_three, _ = consistency_anomalies(sets, min_siblings=3) + assert not [a for a in anomalies_three if a.missing == "identity_constraint"] # only two discharge + anomalies_strict, under_strict = consistency_anomalies(sets, min_siblings=5) + assert anomalies_strict == () and under_strict == (("app", "book"),) # Req 3.3: under-populated, never reported + + +def test_public_route_among_authenticated_siblings_is_a_missing_path_guard_with_access_evidence() -> None: + from openultrasast.semantic.obligations.siblings import consistency_anomalies, sibling_sets + + ops, wit = _inputs() + sets = sibling_sets(ENTRIES, ops, wit) + anomalies, _ = consistency_anomalies(sets, min_siblings=2) + guard_anomalies = [a for a in anomalies if a.missing == "path_guard"] + assert {a.handler for a in guard_anomalies} == {"app.py::constrained_too", "app.py::leaky"} + assert all(a.access_level == "public" and "authenticated" in " ".join(a.evidence) for a in guard_anomalies) + assert all(a.siblings == ("app.py::constrained", "app.py::guarded") for a in guard_anomalies) + + +def test_same_named_files_in_different_modules_are_not_siblings() -> None: + """Round-2 finding: the key must be the module path, not the file's basename.""" + from openultrasast.mapping import EntryPointRecord + from openultrasast.semantic.obligations.operations import Operation + from openultrasast.semantic.obligations.siblings import sibling_sets + + def entry(path: str, function: str) -> EntryPointRecord: + return EntryPointRecord( + path=path, + line=1, + end_line=2, + function_name=function, + name=function, + kind="route", + access_level="public", + trust_boundary="http_request", + access_evidence=[], + conditions=[], + provenance="t", + rationale="", + ) # type: ignore[arg-type] + + def op(path: str, function: str) -> Operation: + return Operation( + path=path, + line=2, + function=function, + kind="protected_read", + fact_id="orm-read", + resource="book", + requires=("identity_constraint",), + ) + + entries = [entry("api/v1/books.py", "list_a"), entry("admin/legacy/books.py", "list_b")] + sets = sibling_sets(entries, [op("api/v1/books.py", "list_a"), op("admin/legacy/books.py", "list_b")], []) + assert {group.key for group in sets} == {("api/v1/books", "book"), ("admin/legacy/books", "book")} + assert all(len(group.handlers) == 1 for group in sets)