"""Probe generators — the SAP-style negative tests, generated from the composition instead of written by hand,
so a use case added tomorrow is probed tomorrow.
permission_matrix every persona × every use case, through BOTH doors: no identity → 401; a permission the
role lacks → exactly "403: missing permission: <perm>"; a role that holds it is not refused
(and does not crash on valid input). With `rows=`, another person of the same role gets 404 —
never 403 — on every use case that takes the owner's record keys.
idempotency every command declared idempotent=True, replayed with the same input: the replay — and a
replay through the OTHER door — answers what the first call answered, and an observation
made after each call (the record read back) does not move.
mass_assignment every command with a body, sent extra fields (every field its answer shows and the usual
suspects: owner, status, version, roles…) set to sentinels: refused, or ignored — never applied.
Each takes `exceptions`: {use_case or (persona, use_case): "the reason"} — known exceptions, allow-listed by
name with a reason, reported as notes on the probe. A command the idempotency/mass-assignment probes cannot
build input for needs a RECIPE — `recipes={"place_order": lambda s: ("clerk", {...})}` — or an exception: a
missing recipe fails the probe, so a new command cannot slip past untested.
Live: the matrix never calls a command a persona MAY run (placeholder inputs would change the deployment);
those cells are reported as not called. Refusals, 404s and every query are called.
"""
import enum
import inspect
import typing
import uuid
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from typing import Any, Callable, Optional
from pydantic import BaseModel
from erp_platform.scenarios.core import Response, Run, Scenario, ScenarioFailed
from erp_platform.scenarios.personas import ANONYMOUS, split
_EMPTY = inspect.Parameter.empty
SUSPECTS = ("owner", "status", "version", "created_by", "created_at", "approved", "approved_by", "roles",
"permissions", "is_admin", "tenant", "deleted")
def placeholder(annotation: Any) -> Any:
"""A value that passes validation for this annotation, so a call reaches the permission check."""
origin, args = typing.get_origin(annotation), typing.get_args(annotation)
if origin is typing.Annotated:
return placeholder(args[0])
if isinstance(annotation, type) and issubclass(annotation, BaseModel):
return {n: placeholder(f.annotation) for n, f in annotation.model_fields.items() if f.is_required()}
if origin is typing.Union:
return placeholder(next(a for a in args if a is not type(None)))
if origin is typing.Literal:
return args[0]
if origin in (list, tuple, set, typing.List):
return []
if origin is dict or annotation is dict:
return {}
if isinstance(annotation, type) and issubclass(annotation, enum.Enum):
return next(iter(annotation)).value
return {int: 1, float: 1.0, bool: False, Decimal: "1", date: "2026-01-01",
datetime: "2026-01-01T00:00:00Z"}.get(annotation, "x")
def _exempt(uc) -> Optional[str]:
if getattr(uc.fn, "__capability__", None):
return "a capability: its replay is the platform's claim (tests/test_capabilities.py)"
return getattr(uc.fn, "__erp_probe_exempt__", None)
def _use_cases(comp) -> list:
from erp_platform import compose
_, hidden = compose.capability_wiring(comp, [])
return [(m.name, uc) for m in comp.members for uc in m.brick.use_cases if (m.name, uc.name) not in hidden]
def _args(uc, given: dict) -> dict:
out = {}
for n, ann, default in uc.inputs:
if n in given:
out[n] = given[n]
elif default is _EMPTY or n in uc.path_params:
out[n] = placeholder(ann)
return out
def _exception(exceptions: dict, persona: str, name: str) -> Optional[str]:
role, _ = split(persona)
return exceptions.get((persona, name)) or exceptions.get((role, name)) or exceptions.get(name)
def _fail_all(problems: list, first: Optional[Response]) -> None:
if problems:
raise ScenarioFailed(f"{len(problems)} problem(s):\n " + "\n ".join(problems), first)
# ── the permission matrix ──────────────────────────────────────────────────────────────────
@dataclass
class Rows:
"""Row visibility for the matrix: `make(s)` creates records AS `owner` and returns their keys
({"order_id": ...}); `other` — another person, same role — must get 404 on every use case taking them."""
make: Callable[[Run], dict]
owner: str
other: str
def permission_matrix(*, rows: Optional[Rows] = None, exceptions: Optional[dict] = None,
personas: Optional[list] = None, expect: Optional[dict] = None,
wildcards: bool = True) -> Callable:
"""`expect` overrides a cell's answer by (persona or role, use case) → status: hr's manager gets 404, not
403, on show_sensitive (SC-HR-20) — a 403 would confirm the record exists."""
exceptions = dict(exceptions or {})
expect = dict(expect or {})
def generate(cat, comp) -> list:
everyone = list(personas or cat.roles.roles) + [ANONYMOUS]
out = []
for persona in everyone:
out.append(Scenario(
f"PROBE-PERM-{persona.upper().replace('#', '-')}",
f"{persona}: every use case through both doors" + (
" — 401 with no identity" if persona == ANONYMOUS else
" — refused without the permission, served with it"),
_matrix_fn(persona, exceptions, expect), tags=("permissions", "probe"), personas=(persona,), generated=True))
if rows is not None:
out.append(Scenario(
f"PROBE-ROWS-{split(rows.other)[0].upper()}",
f"another {split(rows.other)[0]} gets 404 — never 403 — on every record of {rows.owner}, both doors",
_rows_fn(rows, exceptions), tags=("permissions", "probe"), personas=(rows.owner, rows.other),
generated=True))
if wildcards:
out += _wildcard_scenarios(cat, comp, exceptions)
return out
return generate
# ── wildcard grants (0.9.2): `*:*:*`, `<module>:*:*` and malformed grants, against every use case ───────────
MALFORMED_GRANTS = ["*", "*:*", "*:*:*:*", "x:*", "X:*:*", "x:i*:*", " *:*:*", "", "a-b:c:d", 7]
def _wildcard_scenarios(cat, comp, exceptions: dict) -> list:
from erp_platform import auth
gated = [(b, uc) for b, uc in _use_cases(comp) if uc.permission and auth.is_concrete(uc.permission)]
spaces = sorted({uc.permission.split(":")[0] for _, uc in gated})
probes = [("probe_star", ["*:*:*"], lambda p: not auth.is_machine(p),
"`*:*:*` passes every person use case's gate and no machine one")]
probes += [(f"probe_{ns}_star", [f"{ns}:*:*"], (lambda ns: lambda p: p.split(":")[0] == ns and not auth.is_machine(p))(ns),
f"`{ns}:*:*` is confined to {ns}") for ns in spaces]
probes.append(("probe_malformed", list(MALFORMED_GRANTS), lambda p: False, "malformed grants pass nothing"))
out = []
for persona, grants, passes, title in probes:
if persona in cat.roles.roles:
raise ValueError(f"{cat.roles.source}: the role name {persona!r} is reserved for a generated probe")
cat.roles.synthetic[persona] = grants
out.append(Scenario(f"PROBE-WILD-{persona.removeprefix('probe_').removesuffix('_star').upper().replace('_', '-') or 'STAR'}", title,
_wild_fn(persona, passes, exceptions), tags=("permissions", "probe"),
personas=(persona,), modes=("inprocess",), generated=True))
return out
def _wild_fn(persona: str, passes: Callable, exceptions: dict) -> Callable:
def probe(s: Run) -> None:
from erp_platform import auth
actor = s.as_(persona)
problems, first = [], None
for brick, uc in _use_cases(s.runner.comp):
if not uc.permission or not auth.is_concrete(uc.permission) or _exception(exceptions, persona, uc.name):
continue
gate = f"missing permission: {uc.permission}"
should = passes(uc.permission)
for door in (("http", "mcp") if uc.tool else ("http",)):
r = actor.call(uc.name, _args(uc, {}), door=door)
if should and (r.status_code == 401 or (r.status_code == 403 and r.reason == gate)):
problems.append(f"{persona} {uc.name} via {door}: refused at the gate ({r.status_code} {r.reason!r})")
elif not should and (r.status_code != 403 or r.reason != gate):
problems.append(f"{persona} {uc.name} via {door}: answered {r.status_code} {r.reason!r}, not 403 {gate!r}")
else:
continue
first = first or r
_fail_all(problems, first)
return probe
def _matrix_fn(persona: str, exceptions: dict, expect: dict) -> Callable:
def probe(s: Run) -> None:
roles = s.catalogue.roles
actor = s.anonymous if persona == ANONYMOUS else s.as_(*_split_actor(persona))
problems, first, not_called = [], None, 0
if persona == ANONYMOUS:
r = actor.tool("(any)")
if r.status_code != 401 or "bearer" not in r.headers.get("www-authenticate", "").lower():
problems.append(f"MCP with no identity answered {r.status_code}, not 401 with a Bearer challenge")
first = first or r
for brick, uc in _use_cases(s.runner.comp):
why = _exception(exceptions, persona, uc.name)
if why:
s.note(f"{uc.name}: not probed ({why})")
continue
if persona == ANONYMOUS:
expected = 401
elif uc.permission and not roles.holds(persona, uc.permission):
expected = 403
else:
expected = None # allowed
override = expect.get((persona, uc.name), expect.get((split(persona)[0], uc.name)))
if override is not None and persona != ANONYMOUS:
expected = override
if expected is None and s.mode == "live" and uc.kind == "command":
not_called += 1
continue
args = _args(uc, {})
for door in (("http",) if persona == ANONYMOUS or not uc.tool else ("http", "mcp")):
r = actor.call(uc.name, dict(args), door=door)
bad = None
if override is not None and persona != ANONYMOUS:
if r.status_code != expected:
bad = f"answered {r.status_code} {r.reason!r}, not {expected} (the matrix's expect=)"
elif expected == 401 and r.status_code != 401:
bad = f"answered {r.status_code}, not 401"
elif expected == 403 and (r.status_code != 403 or r.reason != f"missing permission: {uc.permission}"):
bad = f"answered {r.status_code} {r.reason!r}, not 403 'missing permission: {uc.permission}'"
elif expected is None and (r.status_code in (401, 403) or r.status_code >= 500):
bad = f"holds the permission but answered {r.status_code} {r.reason!r}"
if bad:
problems.append(f"{persona} {uc.name} via {door}: {bad}")
first = first or r
if not_called:
s.note(f"{not_called} command(s) {persona} may run were not called live (they would change the deployment)")
_fail_all(problems, first)
return probe
def _split_actor(persona: str) -> tuple:
role, n = split(persona)
return role, n
def _rows_fn(rows: Rows, exceptions: dict) -> Callable:
def probe(s: Run) -> None:
s.given(f"records made by {rows.owner}")
made = rows.make(s)
other = s.as_(*_split_actor(rows.other))
roles = s.catalogue.roles
s.when(f"{rows.other} asks for each of them, through both doors")
problems, first = [], None
for brick, uc in _use_cases(s.runner.comp):
if not uc.path_params or not uc.path_params <= set(made):
continue
why = _exception(exceptions, rows.other, uc.name)
if why:
s.note(f"{uc.name}: not probed ({why})")
continue
if uc.permission and not roles.holds(rows.other, uc.permission):
continue # the matrix covers the 403
for door in (("http", "mcp") if uc.tool else ("http",)):
r = other.call(uc.name, _args(uc, made), door=door)
if r.status_code != 404:
problems.append(f"{rows.other} {uc.name} via {door}: answered {r.status_code} {r.reason!r}, not 404")
first = first or r
s.then("every one is not found")
_fail_all(problems, first)
return probe
# ── idempotency ────────────────────────────────────────────────────────────────────────────
def _recipe(recipes: dict, s: Run, name: str) -> tuple:
out = recipes[name](s)
persona, args = out[0], dict(out[1])
observe = out[2] if len(out) > 2 else None
return persona, args, observe
def idempotency(*, recipes: dict, exceptions: Optional[dict] = None, ignore: tuple = ()) -> Callable:
exceptions = dict(exceptions or {})
def generate(cat, comp) -> list:
out = []
for brick, uc in _use_cases(comp):
if uc.kind != "command" or not uc.idempotent or _exempt(uc):
continue
out.append(Scenario(f"PROBE-IDEM-{uc.name.upper()}",
f"{uc.name} replayed with the same input changes nothing, through either door",
_idem_fn(uc.name, recipes, exceptions, ignore), tags=("idempotency", "probe"),
generated=True))
return out
return generate
def _idem_fn(name: str, recipes: dict, exceptions: dict, ignore: tuple) -> Callable:
def probe(s: Run) -> None:
why = exceptions.get(name)
if why:
s.note(f"not probed ({why})")
return
if name not in recipes:
raise ScenarioFailed(f"{name} is declared idempotent and has no recipe: add one to idempotency(recipes=) "
"or list it in exceptions with the reason")
persona, args, observe = _recipe(recipes, s, name)
actor = s.as_(*_split_actor(persona))
s.when(f"{persona} runs {name}")
first = actor.call(name, dict(args)).ok()
seen = observe(s, first) if observe else None
s.then("the same input again — through the API, then through MCP — answers the same and moves nothing")
for door in (("http", "mcp") if s.use_case(name)[1].tool else ("http",)):
again = actor.call(name, dict(args), door=door)
again.same_as(first, ignore=ignore)
if observe:
observe(s, again).same_as(seen, ignore=ignore)
return probe
# ── mass assignment ────────────────────────────────────────────────────────────────────────
def mass_assignment(*, recipes: dict, exceptions: Optional[dict] = None, extra: tuple = ()) -> Callable:
exceptions = dict(exceptions or {})
def generate(cat, comp) -> list:
out = []
for brick, uc in _use_cases(comp):
if uc.kind != "command" or uc.method == "GET" or _exempt(uc):
continue
body = [n for n, _, _ in uc.inputs if n not in uc.path_params]
if not body:
continue
out.append(Scenario(f"PROBE-MASS-{uc.name.upper()}",
f"{uc.name} given fields it did not ask for refuses or ignores them — never applies them",
_mass_fn(uc.name, recipes, exceptions, tuple(extra)), tags=("mass-assignment", "probe"),
generated=True))
return out
return generate
def _mass_fn(name: str, recipes: dict, exceptions: dict, extra: tuple) -> Callable:
def probe(s: Run) -> None:
why = exceptions.get(name)
if why:
s.note(f"not probed ({why})")
return
if name not in recipes:
raise ScenarioFailed(f"{name} takes a body and has no recipe: add one to mass_assignment(recipes=) "
"or list it in exceptions with the reason")
problems, first = [], None
for door in (("http", "mcp") if s.use_case(name)[1].tool else ("http",)):
persona, args, observe = _recipe(recipes, s, name)
actor = s.as_(*_split_actor(persona))
sentinel = f"mass-assign-{uuid.uuid4().hex[:8]}"
# what the recipe answers shows the fields worth trying; the usual suspects besides
brick, uc = s.use_case(name)
fields = [f for f in dict.fromkeys(list(SUSPECTS) + list(extra)) if f not in args and f not in uc.path_params]
s.when(f"{persona} runs {name} via {door} with {len(fields)} extra fields set to {sentinel}")
r = actor.call(name, {**args, **{f: sentinel for f in fields}}, door=door)
if r.status_code == 422:
continue # refused: fine
if not 200 <= r.status_code < 300:
problems.append(f"via {door}: answered {r.status_code} {r.reason!r} (neither refused 422 nor served)")
first = first or r
continue
seen = [r] + ([observe(s, r)] if observe else [])
applied = sorted({f for resp in seen if isinstance(resp.body, dict) for f in fields
if _contains(resp.body, sentinel, f)})
if applied:
problems.append(f"via {door}: applied {', '.join(applied)} = {sentinel} from the request")
first = first or r
s.then("nothing extra was applied")
_fail_all(problems, first)
return probe
def _contains(body: Any, sentinel: str, field: str) -> bool:
if isinstance(body, dict):
return any((k == field and v == sentinel) or _contains(v, sentinel, field) for k, v in body.items())
if isinstance(body, list):
return any(_contains(v, sentinel, field) for v in body)
return False