Merge 1.3 successor source support in adapters and composer

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 903097@bnt-lap001
Assistant-Session: aa320c8f-c3fa-46e7-8541-0b58e1e45d23
This commit is contained in:
tegwick 2026-10-17 11:21:53 +02:00
commit acda26c7a9
12 changed files with 151 additions and 52 deletions

View file

@ -9,6 +9,13 @@ map, a duplicate or unknown step, any mismatch, and any change to canonical 1.1.
Nothing here writes a type file, pins a descriptor, admits anything or touches Nothing here writes a type file, pins a descriptor, admits anything or touches
a live system. A composed map is a successor candidate that needs a reviewed a live system. A composed map is a successor candidate that needs a reviewed
revision under canon section 6 and owner review (T16) before any pinning. revision under canon section 6 and owner review (T16) before any pinning.
Generalised (NK-WP-0048) to a selectable successor version via --version
('1.2' default, or '1.3' = 1.2 content plus execution_descriptors). The type is
loaded from security-case-type-revisions/<version>/ through the explicit version
root, and each version-aware adapter must expose SELECTABLE_TYPES[<version>].
The file name is kept so review targets do not move; with the default the
output is byte-identical to the 1.2-only composer.
""" """
from __future__ import annotations from __future__ import annotations
@ -25,7 +32,10 @@ from warden.case_receipt_state import Refused
ROOT = Path(__file__).resolve().parent ROOT = Path(__file__).resolve().parent
REPO = ROOT.parents[2] REPO = ROOT.parents[2]
DEFAULT_ROOT = REPO / 'security-case-types' DEFAULT_ROOT = REPO / 'security-case-types'
VERSION_ROOTS = {'1.2': REPO / 'security-case-type-revisions' / '1.2'} REVISIONS = REPO / 'security-case-type-revisions'
SUCCESSOR_VERSIONS = ('1.2', '1.3')
DEFAULT_VERSION = '1.2'
VERSION_ROOTS = {'1.2': REVISIONS / '1.2'}
TYPE_ID = 'oidc-browser-relying-party' TYPE_ID = 'oidc-browser-relying-party'
TYPE_REF = f'{TYPE_ID}@1.2' TYPE_REF = f'{TYPE_ID}@1.2'
CANONICAL_REF = f'{TYPE_ID}@1.1' CANONICAL_REF = f'{TYPE_ID}@1.1'
@ -76,18 +86,34 @@ def encode(desc: Descriptor) -> bytes:
return json.dumps(asdict(desc), sort_keys=True, separators=(',', ':')).encode() return json.dumps(asdict(desc), sort_keys=True, separators=(',', ':')).encode()
def load_types(default_root=DEFAULT_ROOT, version_roots=None): def _version(version):
"""Return (1.2 spec, canonical 1.1 file digest) via explicit selection only.""" if version not in SUCCESSOR_VERSIONS:
raise Refused(f'unsupported successor version {version!r}; allowed {list(SUCCESSOR_VERSIONS)}')
return version
def _type_ref(version):
return f'{TYPE_ID}@{_version(version)}'
def load_types(default_root=DEFAULT_ROOT, version_roots=None, version=DEFAULT_VERSION):
"""Return (<version> spec, canonical 1.1 file digest) via explicit selection only."""
version, type_ref = _version(version), _type_ref(version)
if version_roots is None:
root = REVISIONS / version
if not root.is_dir():
raise Refused(f'revision directory security-case-type-revisions/{version} absent')
version_roots = {version: root}
v = _validator() v = _validator()
roots = dict(VERSION_ROOTS if version_roots is None else version_roots) roots = dict(version_roots)
if set(roots) != {'1.2'}: if set(roots) != {version}:
raise Refused('only an explicit 1.2 version root may be selected') raise Refused(f'only an explicit {version} version root may be selected')
issues, data, _ = v.load_type(Path(default_root), TYPE_REF, version_roots=roots) issues, data, _ = v.load_type(Path(default_root), type_ref, version_roots=roots)
if issues or data['spec'].get('version') != '1.2': if issues or str(data['spec'].get('version')) != version:
raise Refused('1.2 type not loadable through explicit version-root selection') raise Refused(f'{version} type not loadable through explicit version-root selection')
issues, default_12, _ = v.load_type(Path(default_root), TYPE_REF) issues, default_12, _ = v.load_type(Path(default_root), type_ref)
if not issues: if not issues:
raise Refused('1.2 is reachable by default; canonical 1.1 default changed') raise Refused(f'{version} is reachable by default; canonical 1.1 default changed')
issues, canonical, type_dir = v.load_type(Path(default_root), CANONICAL_REF) issues, canonical, type_dir = v.load_type(Path(default_root), CANONICAL_REF)
if issues or str(canonical['spec'].get('version')) != '1.1': if issues or str(canonical['spec'].get('version')) != '1.1':
raise Refused('canonical 1.1 default not loadable') raise Refused('canonical 1.1 default not loadable')
@ -101,13 +127,14 @@ def _resolve(value, params):
raise Refused('type owner template has no bound parameter') from None raise Refused('type owner template has no bound parameter') from None
def adapter_descriptors(check_config_digest, consumer_owner): def adapter_descriptors(check_config_digest, consumer_owner, version=DEFAULT_VERSION):
"""Ask every step adapter for its descriptor; T13 needs the checker digest.""" """Ask every step adapter for its descriptor; T13 needs the checker digest."""
type_ref = _type_ref(version)
out = [] out = []
for step, (filename, needs_12) in ADAPTERS.items(): for step, (filename, needs_12) in ADAPTERS.items():
m = _load(f'compose_{step.replace("-", "_")}', ROOT / filename) m = _load(f'compose_{step.replace("-", "_")}', ROOT / filename)
if needs_12 and getattr(m, 'SELECTABLE_TYPES', {}).get('1.2') != TYPE_REF: if needs_12 and getattr(m, 'SELECTABLE_TYPES', {}).get(version) != type_ref:
raise Refused(f'{step}: adapter has no explicit 1.2 version-root path') raise Refused(f'{step}: adapter has no explicit {version} version-root path')
if step == 'register-client': if step == 'register-client':
desc, source = m.descriptor(check_config_digest), m.source_digest(check_config_digest) desc, source = m.descriptor(check_config_digest), m.source_digest(check_config_digest)
elif step in ('deliver-consumer-secret', 'hand-over'): elif step in ('deliver-consumer-secret', 'hand-over'):
@ -120,11 +147,12 @@ def adapter_descriptors(check_config_digest, consumer_owner):
return out return out
def compose(type_spec, candidates, params, canonical_digest_before, *, default_root=DEFAULT_ROOT): def compose(type_spec, candidates, params, canonical_digest_before, *, default_root=DEFAULT_ROOT,
"""Return {step: descriptor digest} for the complete 1.2 set, or raise Refused.""" version=DEFAULT_VERSION, version_roots=None):
"""Return {step: descriptor digest} for the complete <version> set, or raise Refused."""
steps = [s['id'] for s in type_spec.get('follow_through', ())] steps = [s['id'] for s in type_spec.get('follow_through', ())]
if len(steps) != 7 or len(set(steps)) != 7: if len(steps) != 7 or len(set(steps)) != 7:
raise Refused('1.2 type does not declare exactly seven distinct steps') raise Refused(f'{version} type does not declare exactly seven distinct steps')
by_step, sources = {}, {} by_step, sources = {}, {}
for step, desc in candidates: for step, desc in candidates:
if step in by_step: if step in by_step:
@ -145,16 +173,16 @@ def compose(type_spec, candidates, params, canonical_digest_before, *, default_r
for entry in type_spec['follow_through']: for entry in type_spec['follow_through']:
desc = by_step[entry['id']] desc = by_step[entry['id']]
if desc.owner != _resolve(entry['owner'], params): if desc.owner != _resolve(entry['owner'], params):
raise Refused(f"{entry['id']}: owner {desc.owner!r} does not match the 1.2 type") raise Refused(f"{entry['id']}: owner {desc.owner!r} does not match the {version} type")
if 'lane' in entry and desc.lane != entry['lane']: if 'lane' in entry and desc.lane != entry['lane']:
raise Refused(f"{entry['id']}: lane {desc.lane!r} does not match the 1.2 type") raise Refused(f"{entry['id']}: lane {desc.lane!r} does not match the {version} type")
_, after = load_types(default_root) _, after = load_types(default_root, version_roots, version)
if after != canonical_digest_before: if after != canonical_digest_before:
raise Refused('canonical 1.1 type changed during composition') raise Refused('canonical 1.1 type changed during composition')
return {s: digest(sources[s]) for s in steps} return {s: digest(sources[s]) for s in steps}
def review(type_spec, candidates, params): def review(type_spec, candidates, params, version=DEFAULT_VERSION):
"""Per-step findings without stopping at the first refusal (for the artifact).""" """Per-step findings without stopping at the first refusal (for the artifact)."""
entries = {e['id']: e for e in type_spec.get('follow_through', ())} entries = {e['id']: e for e in type_spec.get('follow_through', ())}
rows = [] rows = []
@ -166,7 +194,7 @@ def review(type_spec, candidates, params):
except Refused: except Refused:
findings.append('refused by Warden v1 Descriptor.decode') findings.append('refused by Warden v1 Descriptor.decode')
if entry is None: if entry is None:
findings.append('step not in 1.2 type') findings.append(f'step not in {version} type')
else: else:
try: try:
owner = _resolve(entry['owner'], params) owner = _resolve(entry['owner'], params)
@ -189,18 +217,19 @@ def review(type_spec, candidates, params):
return rows return rows
def artifact(check_config_digest, params): def artifact(check_config_digest, params, version=DEFAULT_VERSION):
type_spec, canonical = load_types() type_spec, canonical = load_types(version=version)
candidates = adapter_descriptors(check_config_digest, params['consumer_owner_repo']) candidates = adapter_descriptors(check_config_digest, params['consumer_owner_repo'], version)
try: try:
composed, refusal = compose(type_spec, candidates, params, canonical), None composed, refusal = compose(type_spec, candidates, params, canonical, version=version), None
except Refused as exc: except Refused as exc:
composed, refusal = None, str(exc) composed, refusal = None, str(exc)
return { return {
'schema': 'netkingdom.descriptor-composition-candidate.v1', 'schema': 'netkingdom.descriptor-composition-candidate.v1',
'task': 'NK-WP-0048-T30', 'task': 'NK-WP-0048-T30',
'type': TYPE_REF, 'type': _type_ref(version),
'selection': 'explicit version root 1.2=security-case-type-revisions/1.2; default root unchanged', 'selection': (f'explicit version root {version}=security-case-type-revisions/{version}; '
'default root unchanged'),
'canonical_default': {'type': CANONICAL_REF, 'type_yaml_digest': canonical, 'canonical_default': {'type': CANONICAL_REF, 'type_yaml_digest': canonical,
'unchanged': True}, 'unchanged': True},
'status': 'not_an_admission', 'status': 'not_an_admission',
@ -210,7 +239,7 @@ def artifact(check_config_digest, params):
'complete_executable_descriptor_set': composed is not None, 'complete_executable_descriptor_set': composed is not None,
'execution_descriptors': composed, 'execution_descriptors': composed,
'refusal': refusal, 'refusal': refusal,
'steps': review(type_spec, candidates, params), 'steps': review(type_spec, candidates, params, version),
'owner_review': False, 'admitted': False, 'dispatch_permitted': False, 'owner_review': False, 'admitted': False, 'dispatch_permitted': False,
} }
@ -220,8 +249,13 @@ def main(argv=None):
p.add_argument('--check-config-digest', required=True) p.add_argument('--check-config-digest', required=True)
p.add_argument('--consumer-owner-repo', required=True) p.add_argument('--consumer-owner-repo', required=True)
p.add_argument('--out', type=Path) p.add_argument('--out', type=Path)
p.add_argument('--version', choices=SUCCESSOR_VERSIONS, default=DEFAULT_VERSION)
a = p.parse_args(argv) a = p.parse_args(argv)
doc = artifact(a.check_config_digest, {'consumer_owner_repo': a.consumer_owner_repo}) try:
doc = artifact(a.check_config_digest, {'consumer_owner_repo': a.consumer_owner_repo}, a.version)
except Refused as exc:
sys.stderr.write(f'refused: {exc}\n')
return 3
text = json.dumps(doc, indent=2, sort_keys=True) + '\n' text = json.dumps(doc, indent=2, sort_keys=True) + '\n'
if a.out: if a.out:
a.out.write_text(text) a.out.write_text(text)

View file

@ -4,7 +4,8 @@ Exercise status: unexercised
Source-only, injection-only evidence seam for the type 1.2 step Source-only, injection-only evidence seam for the type 1.2 step
`deliver-consumer-secret` (NK-WP-0048-T28), modelled on the T13 registration `deliver-consumer-secret` (NK-WP-0048-T28), modelled on the T13 registration
adapter. Warden does not execute the consumer owner's apply adapter. It accepts exactly type 1.2 and its successor 1.3 (1.2 plus
`execution_descriptors`) cases and refuses canonical 1.1 and any other type. Warden does not execute the consumer owner's apply
(`warden_executes: false`); the adapter binds one admitted case and accepts a (`warden_executes: false`); the adapter binds one admitted case and accepts a
closed, value-free owner status receipt. It installs no command, kubectl call, closed, value-free owner status receipt. It installs no command, kubectl call,
credential provider, OpenBao access or evidence trust root. Canonical 1.1 stays credential provider, OpenBao access or evidence trust root. Canonical 1.1 stays

View file

@ -21,6 +21,8 @@ ROOT = Path(__file__).resolve().parent
SOURCE_FILES = ('deliver-consumer-secret-adapter.py',) SOURCE_FILES = ('deliver-consumer-secret-adapter.py',)
REFERENCE = 'net-kingdom/deliver-consumer-secret@v1' REFERENCE = 'net-kingdom/deliver-consumer-secret@v1'
TYPE = 'oidc-browser-relying-party@1.2' TYPE = 'oidc-browser-relying-party@1.2'
# Exactly 1.2 and its successor 1.3 (1.2 + execution_descriptors); 1.1 and others refused.
ACCEPTED_TYPES = frozenset({TYPE, 'oidc-browser-relying-party@1.3'})
OBJECTS = ('consumer-eso-sa', 'consumer-secretstore', 'consumer-externalsecret') OBJECTS = ('consumer-eso-sa', 'consumer-secretstore', 'consumer-externalsecret')
# Rollback order: ExternalSecret/owned target, then SecretStore, then SA. # Rollback order: ExternalSecret/owned target, then SecretStore, then SA.
ROLLBACK_ORDER = tuple(reversed(OBJECTS)) ROLLBACK_ORDER = tuple(reversed(OBJECTS))
@ -101,8 +103,8 @@ def build(*, admitted: Invocation, case_source: bytes, owner_apply, owner_remove
case = yaml.safe_load(case_source) case = yaml.safe_load(case_source)
spec = case.get('spec') if isinstance(case, dict) else None spec = case.get('spec') if isinstance(case, dict) else None
params = spec.get('parameters') if isinstance(spec, dict) else None params = spec.get('parameters') if isinstance(spec, dict) else None
if not isinstance(params, dict) or spec.get('type') != TYPE: if not isinstance(params, dict) or spec.get('type') not in ACCEPTED_TYPES:
raise Refused('type 1.2 case required; canonical 1.1 has no delivery step') raise Refused('type 1.2 or 1.3 case required; canonical 1.1 has no delivery step')
if params.get('client_type') != 'confidential': if params.get('client_type') != 'confidential':
raise Refused('delivery applies to confidential clients only') raise Refused('delivery applies to confidential clients only')
for key in ('client_id', 'consuming_service', 'consumer_namespace', 'consumer_owner_repo'): for key in ('client_id', 'consuming_service', 'consumer_namespace', 'consumer_owner_repo'):

View file

@ -10,10 +10,10 @@
"dispatch_permitted": false, "dispatch_permitted": false,
"execution_descriptors": { "execution_descriptors": {
"bind-env": "sha256:d2511dce3d2ff1d380f845920aa168e6fe09c0f0c382482a796a5bf13619905d", "bind-env": "sha256:d2511dce3d2ff1d380f845920aa168e6fe09c0f0c382482a796a5bf13619905d",
"deliver-consumer-secret": "sha256:a74220eddab2709e90e6349b45940f68233c169009cccd0cdeea25a5541774a6", "deliver-consumer-secret": "sha256:9c289f646516923a4c448a2d3cdd45b33115516e6a171b7955924d8b5d852d2b",
"deliver-secret": "sha256:75aaa7346aebdea2a0eb0ba993995515b5248ffefc34caf8e713525e9b58da85", "deliver-secret": "sha256:75aaa7346aebdea2a0eb0ba993995515b5248ffefc34caf8e713525e9b58da85",
"ensure-groups": "sha256:775dd71d7da43d16437dcdd9cba81ea6eef5803ab74e93c24d4994a764180135", "ensure-groups": "sha256:4411b5cb8944ee76ca1ade62c0c246f061981fb18f0b9afd05934a8935598dc3",
"hand-over": "sha256:24d1ab891bfccaac1d691245d495ced2571d940d055c5e387076bb57dc68444e", "hand-over": "sha256:84f0fac401ea2f1e7c305fb1753a34838ac062655199fcfbbd998f39268c3000",
"mint-secret": "sha256:c92957227235db04953809d92b260d32c8b947472a8e8fd17a258ade3ae8da70", "mint-secret": "sha256:c92957227235db04953809d92b260d32c8b947472a8e8fd17a258ade3ae8da70",
"register-client": "sha256:3ccc8eed545cf31f8a48feb8afa0c1355e126d9549eef9737d92d0c1b87708a8" "register-client": "sha256:3ccc8eed545cf31f8a48feb8afa0c1355e126d9549eef9737d92d0c1b87708a8"
}, },
@ -28,11 +28,11 @@
"status": "not_an_admission", "status": "not_an_admission",
"steps": [ "steps": [
{ {
"candidate_descriptor_digest": "sha256:775dd71d7da43d16437dcdd9cba81ea6eef5803ab74e93c24d4994a764180135", "candidate_descriptor_digest": "sha256:4411b5cb8944ee76ca1ade62c0c246f061981fb18f0b9afd05934a8935598dc3",
"descriptor": { "descriptor": {
"actor": "human", "actor": "human",
"adapter": "net-kingdom/ensure-groups@v1", "adapter": "net-kingdom/ensure-groups@v1",
"adapter_source": "sha256:691ace82aec8a3d2ba09fa8c884fcfae247397bc77644e16936b3da1e7942739", "adapter_source": "sha256:87cc12ada32c4d1296070dd1e8586f35151fcb181ecd89d5d1fbb20c78b50a6f",
"allowed_created": [ "allowed_created": [
"service-group", "service-group",
"service-group-membership" "service-group-membership"
@ -181,11 +181,11 @@
"type_owner": "net-kingdom" "type_owner": "net-kingdom"
}, },
{ {
"candidate_descriptor_digest": "sha256:a74220eddab2709e90e6349b45940f68233c169009cccd0cdeea25a5541774a6", "candidate_descriptor_digest": "sha256:9c289f646516923a4c448a2d3cdd45b33115516e6a171b7955924d8b5d852d2b",
"descriptor": { "descriptor": {
"actor": "agent", "actor": "agent",
"adapter": "net-kingdom/deliver-consumer-secret@v1", "adapter": "net-kingdom/deliver-consumer-secret@v1",
"adapter_source": "sha256:7dd5509d7dee83c2a8ff1bcd4704c336f8dbc4586223830cad34b92e6fa0bfff", "adapter_source": "sha256:8ec63233218f2f9571e1f0605ae29b4378f3126d0ff7f5192e35efa4291f72cd",
"allowed_created": [ "allowed_created": [
"consumer-eso-sa", "consumer-eso-sa",
"consumer-secretstore", "consumer-secretstore",
@ -216,11 +216,11 @@
"type_owner": "{consumer_owner_repo}" "type_owner": "{consumer_owner_repo}"
}, },
{ {
"candidate_descriptor_digest": "sha256:24d1ab891bfccaac1d691245d495ced2571d940d055c5e387076bb57dc68444e", "candidate_descriptor_digest": "sha256:84f0fac401ea2f1e7c305fb1753a34838ac062655199fcfbbd998f39268c3000",
"descriptor": { "descriptor": {
"actor": "human", "actor": "human",
"adapter": "net-kingdom/apps-handover@v1", "adapter": "net-kingdom/apps-handover@v1",
"adapter_source": "sha256:997bde2b03e546d6a86e73f9634f272736ab76b4ff6d2c4ad592f4efd890d352", "adapter_source": "sha256:ceaefd44168fa49d4ab1bb0f4870fe43ffd3b6a963b59d53970c97b1ed871155",
"allowed_created": [ "allowed_created": [
"handover-receipt" "handover-receipt"
], ],

View file

@ -21,7 +21,8 @@ unexercised against any live system and is not pinned, admitted or reviewed.
refused before any call. refused before any call.
- By default the case must be canonical `oidc-browser-relying-party@1.1`. - By default the case must be canonical `oidc-browser-relying-party@1.1`.
`oidc-browser-relying-party@1.2` is accepted only when `build` is called with `oidc-browser-relying-party@1.2` is accepted only when `build` is called with
an explicit `version_root='1.2'` (as in `handover-adapter.py`); an unknown root an explicit `version_root='1.2'` (as in `handover-adapter.py`), and successor
`@1.3` only with `version_root='1.3'`; an unknown root
or a case type that differs from the selected root is refused. The 1.2 or a case type that differs from the selected root is refused. The 1.2
revision leaves the ensure-groups step and its `access_groups` / revision leaves the ensure-groups step and its `access_groups` /
`initial_members` parameters unchanged, so both roots share one parser. `initial_members` parameters unchanged, so both roots share one parser.

View file

@ -29,10 +29,11 @@ SOURCE_FILES = (('sso-mfa/k8s/keycape', 'ensure-groups-adapter.py'),
('identity-provisioner', 'group_provenance_postgres.py')) ('identity-provisioner', 'group_provenance_postgres.py'))
REFERENCE = 'net-kingdom/ensure-groups@v1' REFERENCE = 'net-kingdom/ensure-groups@v1'
CASE_TYPE = 'oidc-browser-relying-party@1.1' # canonical 1.1 stays the default CASE_TYPE = 'oidc-browser-relying-party@1.1' # canonical 1.1 stays the default
# 1.2 is reachable only by explicit version-root selection, never by default. # 1.2 and 1.3 are reachable only by explicit version-root selection, never by default.
# Its ensure-groups step and access_groups/initial_members parameters are # Its ensure-groups step and access_groups/initial_members parameters are
# unchanged from 1.1, so the same request parsing applies to both. # unchanged from 1.1, so the same request parsing applies to both.
SELECTABLE_TYPES = {'1.1': CASE_TYPE, '1.2': 'oidc-browser-relying-party@1.2'} SELECTABLE_TYPES = {'1.1': CASE_TYPE, '1.2': 'oidc-browser-relying-party@1.2',
'1.3': 'oidc-browser-relying-party@1.3'}
CREATED = ('service-group', 'service-group-membership') CREATED = ('service-group', 'service-group-membership')
if str(PROVISIONER) not in sys.path: if str(PROVISIONER) not in sys.path:
@ -100,7 +101,7 @@ def build(*, admitted: Invocation, case_source: bytes, provisioner, publish, aut
The idempotency key and correlation derive from the case pin, so every The idempotency key and correlation derive from the case pin, so every
attempt resumes the one journal receipt and never recreates removed state. attempt resumes the one journal receipt and never recreates removed state.
version_root selects the accepted case type: '1.1' (default, canonical) or version_root selects the accepted case type: '1.1' (default, canonical) or
'1.2' only when passed explicitly. Unknown roots are refused. '1.2' / '1.3' only when passed explicitly. Unknown roots are refused.
Rollback reverses only journal-confirmed creations; preexisting/shared Rollback reverses only journal-confirmed creations; preexisting/shared
groups and memberships are preserved. Ambiguity propagates as uncertain. groups and memberships are preserved. Ambiguity propagates as uncertain.

View file

@ -13,7 +13,9 @@ Apps-owned consumer verification and the hand-over receipt.
case must carry a bounded `consumer_owner_repo` equal to it. The lane name is case must carry a bounded `consumer_owner_repo` equal to it. The lane name is
still a candidate value for Apps owner review. still a candidate value for Apps owner review.
- **Case binding:** canonical `oidc-browser-relying-party@1.1` is the default; - **Case binding:** canonical `oidc-browser-relying-party@1.1` is the default;
`@1.2` is reached only through an explicit `version_root='1.2'` selection. `@1.2` is reached only through an explicit `version_root='1.2'` selection, and
successor `@1.3` (1.2 plus `execution_descriptors`) only through
`version_root='1.3'`, with the same `consumer_owner_repo` owner handling.
No `--allow-unimplemented` admission exists. Case, descriptor and acceptance No `--allow-unimplemented` admission exists. Case, descriptor and acceptance
pins are checked; `access_groups` must be non-empty, because 1.1 hand-over pins are checked; `access_groups` must be non-empty, because 1.1 hand-over
requires the consumer to enforce them (`consumer-enforces-groups`). requires the consumer to enforce them (`consumer-enforces-groups`).

View file

@ -20,8 +20,11 @@ ROOT = Path(__file__).resolve().parent
SOURCE_FILES = ('handover-adapter.py',) SOURCE_FILES = ('handover-adapter.py',)
REFERENCE = 'net-kingdom/apps-handover@v1' REFERENCE = 'net-kingdom/apps-handover@v1'
CANONICAL_TYPE = 'oidc-browser-relying-party@1.1' CANONICAL_TYPE = 'oidc-browser-relying-party@1.1'
# 1.2 is reachable only by explicit version-root selection, never by default. # 1.2 and 1.3 are reachable only by explicit version-root selection, never by default.
SELECTABLE_TYPES = {'1.1': CANONICAL_TYPE, '1.2': 'oidc-browser-relying-party@1.2'} # 1.3 is 1.2 plus execution_descriptors, so 1.2 owner handling applies to it.
SELECTABLE_TYPES = {'1.1': CANONICAL_TYPE, '1.2': 'oidc-browser-relying-party@1.2',
'1.3': 'oidc-browser-relying-party@1.3'}
OWNER_ROOTS = frozenset({'1.2', '1.3'})
RECEIPT = 'handover-receipt' RECEIPT = 'handover-receipt'
@ -52,9 +55,9 @@ def _case(case_source, version_root):
if not groups or not all(IDENTIFIER.fullmatch(g) for g in groups): if not groups or not all(IDENTIFIER.fullmatch(g) for g in groups):
raise Refused('hand-over requires access_groups the consumer enforces') raise Refused('hand-over requires access_groups the consumer enforces')
owner = params.get('consumer_owner_repo') owner = params.get('consumer_owner_repo')
if version_root == '1.2' and (not isinstance(owner, str) or not IDENTIFIER.fullmatch(owner)): if version_root in OWNER_ROOTS and (not isinstance(owner, str) or not IDENTIFIER.fullmatch(owner)):
raise Refused('1.2 hand-over requires a bounded consumer_owner_repo') raise Refused('1.2/1.3 hand-over requires a bounded consumer_owner_repo')
return client_id, groups, owner if version_root == '1.2' else None return client_id, groups, owner if version_root in OWNER_ROOTS else None
def build(*, admitted: Invocation, case_source: bytes, case_acceptance: str, def build(*, admitted: Invocation, case_source: bytes, case_acceptance: str,

View file

@ -89,6 +89,21 @@ class ComposeTests(unittest.TestCase):
with self.assertRaisesRegex(Refused, 'explicit 1.2'): with self.assertRaisesRegex(Refused, 'explicit 1.2'):
c.load_types(version_roots={}) c.load_types(version_roots={})
def test_version_selection(self):
absent = not (c.REVISIONS / '1.3').is_dir()
for bad in ('1.1', '1.4', '', None):
with self.assertRaises(Refused):
c.load_types(version=bad)
with self.assertRaises(Refused): # a 1.2 root cannot be selected as 1.3
c.load_types(version='1.3', version_roots={'1.2': c.VERSION_ROOTS['1.2']})
if absent:
with self.assertRaises(Refused):
c.load_types(version='1.3')
self.assertEqual(c.main(['--check-config-digest', 'sha256:' + '0' * 64,
'--consumer-owner-repo', 'railiance-apps', '--version', '1.3']), 3)
with self.assertRaises(SystemExit):
c.main(['--check-config-digest', 'x', '--consumer-owner-repo', 'y', '--version', '1.1'])
def test_canonical_default_bytes_untouched(self): def test_canonical_default_bytes_untouched(self):
path = c.DEFAULT_ROOT / c.TYPE_ID / 'type.yaml' path = c.DEFAULT_ROOT / c.TYPE_ID / 'type.yaml'
before = path.read_bytes() before = path.read_bytes()

View file

@ -85,6 +85,18 @@ class DeliveryTests(unittest.TestCase):
ad.rollback(self.rinv, ('consumer-externalsecret',)) ad.rollback(self.rinv, ('consumer-externalsecret',))
self.assertNotIn('remove', [e[0] for e in self.effects if isinstance(e, tuple)]) self.assertNotIn('remove', [e[0] for e in self.effects if isinstance(e, tuple)])
def test_accepts_exactly_1_2_and_1_3(self):
self.assertEqual(a.ACCEPTED_TYPES, {'oidc-browser-relying-party@1.2', 'oidc-browser-relying-party@1.3'})
case = self.case_bytes('oidc-browser-relying-party@1.3')
admitted = self.invocation(case)
result = self.adapter(case=case, admitted=admitted).invoke(
replace(admitted, attempt_id=self.inv.attempt_id, operation=self.inv.operation))
self.assertEqual(result.outcome, Outcome.SUCCESS)
for t in ('oidc-browser-relying-party@1.1', 'oidc-browser-relying-party@1.4', 'other@1.3'):
case = self.case_bytes(t)
with self.assertRaises(Refused):
self.adapter(case=case, admitted=self.invocation(case))
def test_refuses_canonical_11_public_and_bad_binding(self): def test_refuses_canonical_11_public_and_bad_binding(self):
for case in (self.case_bytes('oidc-browser-relying-party@1.1'), for case in (self.case_bytes('oidc-browser-relying-party@1.1'),
self.case_bytes(client_type='public')): self.case_bytes(client_type='public')):

View file

@ -206,6 +206,15 @@ class EnsureGroupsTests(unittest.TestCase):
self.assertEqual(result.outcome, Outcome.SUCCESS) self.assertEqual(result.outcome, Outcome.SUCCESS)
self.assertEqual(self.prov.calls[0][1]['members'], {'forgejo-users': ['alice']}) self.assertEqual(self.prov.calls[0][1]['members'], {'forgejo-users': ['alice']})
def test_explicit_version_root_selects_1_3(self):
kw = self._rebind('oidc-browser-relying-party@1.3')
for root in (None, '1.2'): # default 1.1 and a different successor root refuse
with self.assertRaises(Refused):
a.build(**kw) if root is None else a.build(**kw, version_root=root)
adapter = a.build(**kw, version_root='1.3')
result = adapter.invoke(replace(kw['admitted'], attempt_id='attempt-one', operation='run'))
self.assertEqual(result.outcome, Outcome.SUCCESS)
def test_version_root_must_match_case_type_and_be_known(self): def test_version_root_must_match_case_type_and_be_known(self):
with self.assertRaises(Refused): with self.assertRaises(Refused):
a.build(**self.kw, version_root='1.2') # 1.1 case under 1.2 root a.build(**self.kw, version_root='1.2') # 1.1 case under 1.2 root

View file

@ -149,6 +149,25 @@ class HandoverTests(unittest.TestCase):
with self.assertRaises(Refused): # admitted owner differs from the case's consumer_owner_repo with self.assertRaises(Refused): # admitted owner differs from the case's consumer_owner_repo
a.build(**{**kw12, 'admitted': replace(admitted12, owner='apps')}) a.build(**{**kw12, 'admitted': replace(admitted12, owner='apps')})
def test_1_3_explicit_selection_with_1_2_owner_handling(self):
case13 = case_bytes(a.SELECTABLE_TYPES['1.3'])
admitted13 = replace(self.admitted, pins=replace(self.admitted.pins, case=digest(case13)))
with self.assertRaises(Refused): # 1.3 never reached by default
a.build(**{**self.kw, 'admitted': admitted13, 'case_source': case13})
with self.assertRaises(Refused): # 1.3 case must name consumer_owner_repo
a.build(**{**self.kw, 'admitted': admitted13, 'case_source': case13, 'version_root': '1.3'})
case13 = case_bytes(a.SELECTABLE_TYPES['1.3'], {**PARAMS, 'consumer_owner_repo': 'railiance-apps'})
desc = json.dumps(asdict(a.descriptor('railiance-apps')), sort_keys=True, separators=(',', ':')).encode()
admitted13 = replace(self.admitted, owner='railiance-apps',
pins=replace(self.admitted.pins, case=digest(case13), adapter=digest(desc)))
kw13 = {**self.kw, 'admitted': admitted13, 'case_source': case13, 'version_root': '1.3'}
with self.assertRaises(Refused): # a 1.3 case under the 1.2 root
a.build(**{**kw13, 'version_root': '1.2'})
inv = replace(admitted13, attempt_id='attempt-one', operation='run')
self.assertEqual(a.build(**kw13).invoke(inv).outcome, Outcome.SUCCESS)
with self.assertRaises(Refused): # admitted owner differs from the case's consumer_owner_repo
a.build(**{**kw13, 'admitted': replace(admitted13, owner='apps')})
def test_binding_refusals(self): def test_binding_refusals(self):
bad = [{'case_acceptance': 'other'}, {'case_source': self.case + b'\n#'}, bad = [{'case_acceptance': 'other'}, {'case_source': self.case + b'\n#'},
{'admitted': replace(self.admitted, owner='net-kingdom')}, {'admitted': replace(self.admitted, owner='net-kingdom')},