NK-WP-0048 r5 adapters: OW-1.3-C1 attended change lane, RAPPS-1.3-C1 human actor, RAPPS-1.3-C2 handover observations

Lane key-cape-oidc-login -> keycape-attended-case-change in the register-client,
ensure-groups and bind-env descriptors; deliver-consumer-secret actor agent ->
human; verify_consumer reports a value-free, exactly-once probe record list bound
into the observation digest. Composition artifacts, type.yaml and admit/compose
tests are left for the orchestrator to regenerate.

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:59:14 +02:00
parent 239a528d62
commit cbadd359ba
13 changed files with 88 additions and 16 deletions

View file

@ -20,7 +20,7 @@ bound to anything else (including a literal), is refused.
`register-user-engine-portal.py` — closure checked by
`test_adapter_source_closure.py`), the
descriptor digest (`Pins.adapter`), case source and acceptance digests, and the
admitted invocation (step `bind-env`, actor `human`, lane `key-cape-oidc-login`,
admitted invocation (step `bind-env`, actor `human`, lane `keycape-attended-case-change`,
created object `keycape-env-binding`, not retry-safe, `parent-session` scope).
Every attempt is rebound to that invocation and re-checks source drift.
- `publish`, `authenticate` and `revoke_session` are mandatory injected

View file

@ -85,7 +85,7 @@ def source_digest():
def descriptor():
return Descriptor('warden.case.execution.v1', 'bind-env', 'net-kingdom', REFERENCE,
source_digest(), 'human', 'key-cape-oidc-login', 'client_id', 'change',
source_digest(), 'human', 'keycape-attended-case-change', 'client_id', 'change',
(CREATED,), False, ('parent-session',))

View file

@ -12,7 +12,7 @@ credential provider, OpenBao access or evidence trust root. Canonical 1.1 stays
the default: a 1.1 case, or a public client, is refused because the step does not
exist there. No `--allow-unimplemented` admission is used or implied.
Binding: descriptor (`agent` actor, lane `consumer-owner-deployment`, effect
Binding: descriptor (`human` actor, lane `consumer-owner-deployment`, effect
`change`, not retry-safe, `parent-session` revocation) pins this adapter source;
case, descriptor, target (`client_id`) and owner (`consumer_owner_repo`) must match
the admitted invocation; each attempt is rebound and the source rehashed.

View file

@ -39,7 +39,7 @@ def descriptor(owner):
if not isinstance(owner, str) or not IDENTIFIER.fullmatch(owner):
raise Refused('bounded consumer owner required')
return Descriptor('warden.case.execution.v1', 'deliver-consumer-secret', owner,
REFERENCE, source_digest(), 'agent', 'consumer-owner-deployment',
REFERENCE, source_digest(), 'human', 'consumer-owner-deployment',
'client_id', 'change', OBJECTS, False, ('parent-session',))

View file

@ -50,7 +50,7 @@ def source_digest():
def descriptor():
return Descriptor('warden.case.execution.v1', 'ensure-groups', 'net-kingdom', REFERENCE,
source_digest(), 'human', 'key-cape-oidc-login', 'client_id', 'change',
source_digest(), 'human', 'keycape-attended-case-change', 'client_id', 'change',
CREATED, False, ('parent-session',))

View file

@ -21,7 +21,12 @@ Apps-owned consumer verification and the hand-over receipt.
requires the consumer to enforce them (`consumer-enforces-groups`).
- **Injected capabilities (required, never strings or case parameters):**
`verify_consumer(client_id, groups)` must return exactly `client_id`,
`enforces_groups=True`, the case's `groups` and `login_verified=True`;
`enforces_groups=True`, the case's `groups`, `login_verified=True` and
`observations` (RAPPS-1.3-C2): a value-free list of `{probe, evidence}` records,
one each for `member-login`, `non-member-refused`, `membership-add-grants` and
`membership-remove-revokes`, evidence an owner evidence identifier (IDENTIFIER
pattern). Missing, extra, duplicate or unknown probes and any other key are
refused; the observation digest covers the records in canonical probe order;
`record_handover(client_id, observation_digest) -> (identifier, created)`;
`remove_handover(identifier) -> True`; `publish(bytes) -> identifier` and
`authenticate(identifier, bytes) -> True` for owner evidence;

View file

@ -26,6 +26,26 @@ SELECTABLE_TYPES = {'1.1': CANONICAL_TYPE, '1.2': 'oidc-browser-relying-party@1.
'1.3': 'oidc-browser-relying-party@1.3'}
OWNER_ROOTS = frozenset({'1.2', '1.3'})
RECEIPT = 'handover-receipt'
# Value-free probe records verify_consumer must report, each exactly once (RAPPS-1.3-C2).
PROBES = frozenset({'member-login', 'non-member-refused', 'membership-add-grants',
'membership-remove-revokes'})
SEEN_KEYS = frozenset({'client_id', 'enforces_groups', 'groups', 'login_verified', 'observations'})
def _observations(value):
"""Return the probe records in canonical order, or None if the contract differs."""
if not isinstance(value, list) or len(value) != len(PROBES):
return None
records = []
for rec in value:
if not (isinstance(rec, dict) and set(rec) == {'probe', 'evidence'}
and isinstance(rec['probe'], str) and rec['probe'] in PROBES
and isinstance(rec['evidence'], str) and IDENTIFIER.fullmatch(rec['evidence'])):
return None
records.append({'probe': rec['probe'], 'evidence': rec['evidence']})
if {r['probe'] for r in records} != PROBES:
return None
return sorted(records, key=lambda r: r['probe'])
def source_digest():
@ -66,8 +86,10 @@ def build(*, admitted: Invocation, case_source: bytes, case_acceptance: str,
"""Bind one admitted case; return a closed kernel Adapter.
verify_consumer(client_id, groups)->dict observes the consumer and must report
{'client_id', 'enforces_groups', 'groups', 'login_verified'} for exactly this
case. record_handover(client_id, observation_digest)->(identifier, created)
{'client_id', 'enforces_groups', 'groups', 'login_verified', 'observations'} for
exactly this case; 'observations' is a value-free list of {probe, evidence}
records, one per probe in PROBES, evidence an owner evidence IDENTIFIER. The
observation digest covers the whole report, observations included. record_handover(client_id, observation_digest)->(identifier, created)
durably records the hand-over receipt; remove_handover(identifier) removes only
that receipt. publish/authenticate are owner evidence capabilities.
revoke_session(invocation)->identifier revokes this attempt's parent session
@ -136,10 +158,15 @@ def build(*, admitted: Invocation, case_source: bytes, case_acceptance: str,
binding(inv, 'run')
with contained_output():
seen = verify_consumer(client_id, groups)
if not (isinstance(seen, dict) and set(seen) == {'client_id', 'enforces_groups', 'groups', 'login_verified'}
if not (isinstance(seen, dict) and set(seen) == SEEN_KEYS
and seen['client_id'] == client_id and seen['enforces_groups'] is True
and seen['login_verified'] is True and tuple(sorted(seen['groups'] or ())) == groups):
raise Refused('consumer verification contract differs')
probes = _observations(seen['observations'])
if probes is None:
raise Refused('consumer observation records differ')
seen = {'client_id': client_id, 'enforces_groups': True, 'groups': list(groups),
'login_verified': True, 'observations': probes}
observation = digest(json.dumps(seen, sort_keys=True, separators=(',', ':')).encode())
with contained_output():
receipt = record_handover(client_id, observation)

View file

@ -77,7 +77,7 @@ def source_digest(check_config_digest):
def descriptor(check_config_digest):
return Descriptor('warden.case.execution.v1', 'register-client', 'net-kingdom',
REFERENCE, source_digest(check_config_digest), 'human', 'key-cape-oidc-login',
REFERENCE, source_digest(check_config_digest), 'human', 'keycape-attended-case-change',
'client_id', 'change', ('keycape-client',), False,
('parent-session',))

View file

@ -51,7 +51,7 @@ class BindEnvTests(unittest.TestCase):
pins = Pins(digest(self.case), digest(b'type'), digest(desc_bytes),
digest(b'catalog'), digest(b'case-acceptance:test'), digest(b'run sheet'))
self.admitted = Invocation('bind-env', 'net-kingdom', 'forgejo', 'human',
'founder', 'key-cape-oidc-login', pins)
'founder', 'keycape-attended-case-change', pins)
self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run')
self.undo = replace(self.admitted, attempt_id='attempt-two', operation='rollback')
self.evidence, self.revocations = {}, []

View file

@ -36,7 +36,7 @@ class DeliveryTests(unittest.TestCase):
desc = json.dumps(asdict(a.descriptor(owner)), sort_keys=True, separators=(',', ':')).encode()
pins = Pins(digest(case), digest(b'type'), digest(desc), digest(b'catalog'),
digest(b'acceptance'), digest(b'run sheet'))
return Invocation('deliver-consumer-secret', owner, 'forgejo', 'agent', 'founder',
return Invocation('deliver-consumer-secret', owner, 'forgejo', 'human', 'founder',
'consumer-owner-deployment', pins)
def receipt(self, operation, names, want=None):

View file

@ -75,7 +75,7 @@ class EnsureGroupsTests(unittest.TestCase):
pins = Pins(digest(self.case), digest(b'type'), digest(desc_bytes),
digest(b'catalog'), digest(b'acceptance'), digest(b'run sheet'))
self.admitted = Invocation('ensure-groups', 'net-kingdom', 'forgejo', 'human',
'founder', 'key-cape-oidc-login', pins)
'founder', 'keycape-attended-case-change', pins)
self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run')
self.journal = self.make_journal()
self.prov = FakeProvisioner(self.journal)

View file

@ -33,7 +33,11 @@ class HandoverTests(unittest.TestCase):
self.rb = replace(self.admitted, attempt_id='attempt-two', operation='rollback')
self.evidence, self.receipts, self.revoked = {}, set(), []
self.observed = {'client_id': 'forgejo', 'enforces_groups': True,
'groups': ['forge-admins', 'forge-users'], 'login_verified': True}
'groups': ['forge-admins', 'forge-users'], 'login_verified': True,
'observations': [{'probe': p, 'evidence': 'apps-probe:' + p}
for p in ('member-login', 'non-member-refused',
'membership-add-grants', 'membership-remove-revokes')]}
self.recorded = []
self.kw = dict(admitted=self.admitted, case_source=self.case, case_acceptance='case-acceptance:test',
verify_consumer=lambda c, g: dict(self.observed), record_handover=self.record,
remove_handover=self.remove, publish=self.publish,
@ -50,6 +54,7 @@ class HandoverTests(unittest.TestCase):
return ref
def record(self, client_id, observation):
self.recorded.append(observation)
ref = 'handover:' + client_id
created = ref not in self.receipts
self.receipts.add(ref)
@ -129,6 +134,41 @@ class HandoverTests(unittest.TestCase):
self.setUp()
self.assertEqual(self.receipts, set())
def test_observation_record_refusals(self):
obs = self.observed['observations']
cases = {
'missing-field': None,
'not-a-list': {'member-login': 'apps-probe:x'},
'missing-probe': obs[:3],
'extra-probe': obs + [{'probe': 'admin-login', 'evidence': 'apps-probe:y'}],
'duplicate-probe': obs[:3] + [dict(obs[0])],
'unknown-probe': obs[:3] + [{'probe': 'admin-login', 'evidence': 'apps-probe:y'}],
'value-bearing-key': [{**obs[0], 'value': 'secret'}] + obs[1:],
'missing-evidence': [{'probe': 'member-login'}] + obs[1:],
'evidence-not-identifier': [{'probe': 'member-login', 'evidence': 'has space/x'}] + obs[1:],
'evidence-not-string': [{'probe': 'member-login', 'evidence': 7}] + obs[1:],
'record-not-dict': ['member-login'] + obs[1:],
}
for name, value in cases.items():
self.setUp()
if value is None:
del self.observed['observations']
else:
self.observed['observations'] = value
with self.subTest(name=name), self.assertRaises(Refused):
a.build(**self.kw).invoke(self.inv)
self.assertEqual(self.receipts, set())
def test_observations_bound_into_digest_order_independent(self):
def run(mutate):
self.setUp()
mutate(self.observed['observations'])
a.build(**self.kw).invoke(self.inv)
return self.recorded[-1]
base = run(lambda o: None)
self.assertEqual(base, run(lambda o: o.reverse()))
self.assertNotEqual(base, run(lambda o: o[0].update(evidence='apps-probe:changed')))
def test_canonical_default_and_explicit_selection_only(self):
with self.assertRaises(Refused):
a.build(**{**self.kw, 'version_root': '1.0'})

View file

@ -34,7 +34,7 @@ class RegistrationTests(unittest.TestCase):
pins = Pins(digest(self.case), digest(b'type'), digest(self.desc_bytes),
digest(b'catalog'), digest(b'case-acceptance:test'), digest(b'run sheet'))
self.admitted = Invocation('register-client', 'net-kingdom', 'forgejo', 'human',
'founder', 'key-cape-oidc-login', pins)
'founder', 'keycape-attended-case-change', pins)
self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run')
self.secret, self.raw = f.fixture()
self.effects = []
@ -262,7 +262,7 @@ class RegistrationTests(unittest.TestCase):
type_source = yaml.safe_dump(typ).encode()
catalog = b'fixture-only-catalog'
sheet = json.dumps(['register-client', 'net-kingdom', 'forgejo', 'human',
'founder', 'key-cape-oidc-login', 'fixture-acceptance', 'fixture-materialization',
'founder', 'keycape-attended-case-change', 'fixture-acceptance', 'fixture-materialization',
'fixture-descriptor', 'fixture-session'], separators=(',', ':')).encode()
pins = Pins(digest(case_source), digest(type_source), digest(self.desc_bytes),
digest(catalog), digest(b'fixture-acceptance'), digest(sheet))