diff --git a/apps/betterangels-backend/common/tests/test_org_mutation_gating.py b/apps/betterangels-backend/common/tests/test_org_mutation_gating.py new file mode 100644 index 0000000000..1c1b49009b --- /dev/null +++ b/apps/betterangels-backend/common/tests/test_org_mutation_gating.py @@ -0,0 +1,141 @@ +"""Meta-test — every mutation in a grant-cutover module must gate on grants. + +ADR 0001 makes the grant predicates the authority for cut-over domains. A new +or edited mutation that forgets its gate fails silently in one of two ways: it +serves an unauthorized caller, or it refuses a legitimate one. This tripwire +walks the cut-over schema modules and fails unless each mutation resolver: + +1. gates in its own body — ``require_can(`` / ``can_anywhere(`` / ``can_obj(`` / + ``visible(`` or a scoped load (``permission=`` — ADR §2.6: the scoped + ``*_get``/``*_queryset`` load *is* the write check), or +2. delegates to a service/selector function in its own app that does, or +3. is listed in ``GATE_EXEMPT`` — deliberately, with the reason. + +It is a heuristic tripwire, not a proof: the gate test is a string match over +resolver and delegated sources, one delegation hop deep. When it fails, either +add the gate or add an exemption entry explaining why the mutation needs none. +""" + +import ast +from importlib import import_module +from pathlib import Path + +import pytest + +# Apps whose org-scoped mutations are grant-gated (ADR 0001). A domain joins +# this list the moment it cuts over — see the readiness matrix in the ADR §4.1. +GRANT_GATED_MODULES = ("accounts.schema", "reports.schema", "shelters.schema", "teams.schema") + +# ``permission=`` counts because the scoped selectors (``*_get``/``*_queryset``) +# take the permission and scope by the caller's grants — the ADR §2.6 write +# check for update/delete. +GATE_MARKERS = ("require_can(", "can_anywhere(", "can_obj(", "visible(", "permission=") + +# Mutations allowed to skip a grant gate, each with the reason it needs none. +GATE_EXEMPT = { + ("accounts.schema", "login"): "authentication — no org in scope", + ("accounts.schema", "logout"): "authentication — no org in scope", + ("accounts.schema", "update_current_user"): "self-service — the caller's own account, never org-scoped", + ("accounts.schema", "update_user_profile"): "self-service — the caller's own profile, never org-scoped", + ("accounts.schema", "delete_current_user"): "self-service — the caller's own account, never org-scoped", + ("accounts.schema", "create_organization"): ( + "org creation itself — no org authority exists yet; eligibility lives in create_organization_service" + ), +} + + +def _app_dir(app: str) -> Path: + return Path(import_module(app).__path__[0]) + + +def _mutation_resolvers(app: str) -> list[tuple[str, str]]: + """(name, source) for every ``@mutation``-decorated def in ``class Mutation``.""" + src = (_app_dir(app) / "schema.py").read_text() + tree = ast.parse(src) + resolvers: list[tuple[str, str]] = [] + for node in ast.walk(tree): + if not isinstance(node, ast.ClassDef) or node.name != "Mutation": + continue + for item in node.body: + if not isinstance(item, ast.FunctionDef): + continue + if not any("mutation" in ast.unparse(decorator) for decorator in item.decorator_list): + continue + resolvers.append((item.name, ast.get_source_segment(src, item) or "")) + return resolvers + + +def _gated(source: str) -> bool: + return any(marker in source for marker in GATE_MARKERS) + + +def _delegated_sources(app: str) -> dict[str, list[str]]: + """Function name → sources, across the app's ``services``/``selectors`` modules.""" + root = _app_dir(app) + paths = [root / "services.py", root / "selectors.py"] + for package in ("services", "selectors"): + directory = root / package + if directory.is_dir(): + paths.extend(directory.glob("*.py")) + + sources: dict[str, list[str]] = {} + for path in paths: + if not path.exists(): + continue + module_src = path.read_text() + try: + tree = ast.parse(module_src) + except SyntaxError: # pragma: no cover — a broken module fails elsewhere first + continue + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef): + sources.setdefault(node.name, []).append(ast.get_source_segment(module_src, node) or "") + return sources + + +def _called_names(source: str) -> set[str]: + names: set[str] = set() + for node in ast.walk(ast.parse(source)): + if not isinstance(node, ast.Call): + continue + func = node.func + if isinstance(func, ast.Name): + names.add(func.id) + elif isinstance(func, ast.Attribute): + names.add(func.attr) + return names + + +@pytest.mark.parametrize("module_name", GRANT_GATED_MODULES) +def test_every_mutation_in_a_grant_cutover_module_gates(module_name: str) -> None: + app = module_name.split(".", 1)[0] + delegates = _delegated_sources(app) + + ungated: list[str] = [] + for name, source in _mutation_resolvers(app): + if (module_name, name) in GATE_EXEMPT or _gated(source): + continue + delegated_gated = any( + _gated(delegate_source) for called in _called_names(source) for delegate_source in delegates.get(called, []) + ) + if not delegated_gated: + ungated.append(name) + + assert not ungated, ( + f"{module_name}: mutation(s) without a grant gate: {sorted(ungated)}. " + "Gate with require_can()/can_obj() — or a scoped *_get/*_queryset load in the resolver " + "or a service it delegates to — or add a GATE_EXEMPT entry with the reason." + ) + + +def test_exemptions_name_real_mutations_and_carry_a_reason() -> None: + for (module_name, name), reason in GATE_EXEMPT.items(): + assert reason, f"{module_name}.{name}: exemption needs a reason" + resolvers = {resolver_name for resolver_name, _ in _mutation_resolvers(module_name.split(".", 1)[0])} + assert name in resolvers, f"{module_name}.{name} is exempted but no such mutation exists any more" + + +@pytest.mark.parametrize("module_name", GRANT_GATED_MODULES) +def test_configured_modules_still_exist(module_name: str) -> None: + """A renamed/moved module must fail loudly, not silently leave the net open.""" + assert (_app_dir(module_name.split(".", 1)[0]) / "schema.py").exists() diff --git a/apps/betterangels-backend/common/tests/utils.py b/apps/betterangels-backend/common/tests/utils.py index f18cfebe51..f3807d6830 100644 --- a/apps/betterangels-backend/common/tests/utils.py +++ b/apps/betterangels-backend/common/tests/utils.py @@ -216,9 +216,7 @@ def make_legacy_only_holder(*, organization: Organization, template_name: str, u convert when they apply. """ organization.add_user(user) - add_legacy_membership( - user, group=make_permission_group(organization=organization, template_name=template_name) - ) + add_legacy_membership(user, group=make_permission_group(organization=organization, template_name=template_name)) return user