coulomb/net-kingdom
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

NK-WP-0048-T24: ensure-groups adapter source candidate

Browse source 
Binds group creation to the T14/T21 PostgreSQL journal seam (exact case,
source and principal binding; refuses untrusted journals), canonical 1.1
case only, created-only rollback via the journal receipt. Source candidate;
owner review and native action remain separate gates.

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 07:56:25 +02:00
parent 2019e823f0
commit 5a2e7d67a2
3 changed files with 416 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

47
sso-mfa/k8s/keycape/ensure-groups-adapter.md Normal file
View file

@ -0,0 +1,47 @@
# Ensure-groups adapter source candidate (NK-WP-0048-T24)
Source candidate only, modelled on the T13 registration adapter. It is
unexercised against any live system and is not pinned, admitted or reviewed.
## Binding
- `build()` takes an admitted `ensure-groups` invocation, the exact case bytes,
an injected provisioner, and owner `publish`/`authenticate` evidence
capabilities.
- Group creation binds to the T14/T21 journal seam. The provisioner's
`group_provenance` must be a `PostgresGroupProvenance`. Its `GroupJournalBinding`
must name this case pin, the reviewed adapter source digest and the admitted
principal. A missing journal, a local SQLite journal or a differing binding is
refused before any call.
- The case must be canonical `oidc-browser-relying-party@1.1`. `access_groups`
and `initial_members` are parsed exactly, and members outside `access_groups`
are refused.
- The idempotency key and correlation id are derived from the case pin, so
every attempt resumes the same journal receipt. A replay returns `no_change`
and never recreates removed state.
- The source digest pins the adapter plus `provisioner.py`,
`group_provenance.py` and `group_provenance_postgres.py`.
## Rollback
Rollback calls `revoke_service_groups` against the original journal receipt.
It accepts only `service-group`/`service-group-membership` created objects.
The journal, not the caller, decides what was created, and preexisting or
shared state is preserved. `delete_created_if_empty` defaults to `false`.
Enabling it requires an owner fact that no other case uses the groups. This
adapter does not supply that fact.
`verify` requires the authenticated owner evidence and a journal readback
state (`ensured` / `revoked`) matching the outcome. It fails closed on any
journal error.
## Not supplied here
- NetKingdom (directory) owner source review before pinning.
- Native CNPG journal admission, trusted continuity/checkpoint callbacks and
protected `GroupJournalBinding` construction (T14).
- Writer isolation or CAS against external LDAP writers.
- Any native action.
Tests use synthetic fakes only (no SQL, no LLDAP):
`PYTHONPATH=~/ops-warden/src ~/ops-warden/.venv/bin/python -m pytest sso-mfa/k8s/keycape/test_ensure_groups_adapter.py`

183
sso-mfa/k8s/keycape/ensure-groups-adapter.py Normal file
View file

@ -0,0 +1,183 @@
"""Unexercised, injection-only ensure-groups seam for Warden's v1 kernel (T24).
Source candidate only. No CLI, credentials, LLDAP session, SQL connection or
evidence trust root is installed here. Group creation is bound to T14's
journal/provenance seam: the injected provisioner must carry a
PostgresGroupProvenance whose GroupJournalBinding names this exact case,
principal and reviewed adapter source, with independent continuity callbacks.
SQLite/local journals, a missing journal or a differing binding are refused.
Owner source review and every native action remain separate gates.
"""
from __future__ import annotations
from dataclasses import asdict, replace
import json
import sys
from pathlib import Path
import yaml
from warden.case_run import Adapter, Descriptor, Invocation, OwnerResult, contained_output, digest, IDENTIFIER
from warden.case_receipt_state import Outcome, Refused
ROOT = Path(__file__).resolve().parent
PROVISIONER = ROOT.parents[2] / 'identity-provisioner'
SOURCE_FILES = (('sso-mfa/k8s/keycape', 'ensure-groups-adapter.py'),
('identity-provisioner', 'provisioner.py'),
('identity-provisioner', 'group_provenance.py'),
('identity-provisioner', 'group_provenance_postgres.py'))
REFERENCE = 'net-kingdom/ensure-groups@v1'
CASE_TYPE = 'oidc-browser-relying-party@1.1' # canonical 1.1 stays the default
CREATED = ('service-group', 'service-group-membership')
if str(PROVISIONER) not in sys.path:
sys.path.insert(0, str(PROVISIONER))
from group_provenance_postgres import PostgresGroupProvenance # noqa: E402 (no I/O on import)
def source_digest():
"""Pin adapter plus the provisioner/journal import closure, in path order."""
base = ROOT.parents[2]
files = {f'{d}/{n}': digest((base / d / n).read_bytes()) for d, n in SOURCE_FILES}
return digest(json.dumps({'files': files}, sort_keys=True, separators=(',', ':')).encode())
def descriptor():
return Descriptor('warden.case.execution.v1', 'ensure-groups', 'net-kingdom', REFERENCE,
source_digest(), 'human', 'key-cape-oidc-login', 'client_id', 'change',
CREATED, False, ())
def _request(case):
spec = case.get('spec') if isinstance(case, dict) else None
if not isinstance(spec, dict) or spec.get('type') != CASE_TYPE:
raise Refused('canonical oidc-browser-relying-party@1.1 case required')
params = spec.get('parameters')
if not isinstance(params, dict) or not isinstance(params.get('access_groups'), str):
raise Refused('access_groups required')
groups = params['access_groups'].split(',')
if len(set(groups)) != len(groups):
raise Refused('duplicate access group')
members = {}
raw = params.get('initial_members', '')
if not isinstance(raw, str):
raise Refused('initial_members must be a string')
for entry in filter(None, raw.split(';')):
group, _, users = entry.partition('=')
if group not in groups or group in members or not users:
raise Refused('initial_members names a group outside access_groups')
members[group] = users.split('+')
return params.get('client_id'), groups, members
def build(*, admitted: Invocation, case_source: bytes, provisioner, publish, authenticate,
delete_created_if_empty: bool = False):
"""Bind one admitted case to the owner journal; return a closed kernel Adapter.
provisioner exposes ensure_service_groups/revoke_service_groups and carries
the trusted owner journal as group_provenance. publish(bytes)->identifier and
authenticate(identifier, bytes)->True are owner evidence capabilities.
The idempotency key and correlation derive from the case pin, so every
attempt resumes the one journal receipt and never recreates removed state.
Rollback reverses only journal-confirmed creations; preexisting/shared
groups and memberships are preserved. Ambiguity propagates as uncertain.
"""
reviewed = source_digest()
desc = descriptor()
if type(admitted) is not Invocation or not all(
isinstance(v, str) and IDENTIFIER.fullmatch(v)
for v in (admitted.step, admitted.owner, admitted.target,
admitted.actor, admitted.principal, admitted.lane)):
raise Refused('bounded invocation required')
if (admitted.step, admitted.owner, admitted.actor, admitted.lane) != (
desc.step, desc.owner, desc.actor, desc.lane):
raise Refused('ensure-groups invocation boundary differs')
if admitted.operation != 'admission' or admitted.attempt_id:
raise Refused('expected admission invocation')
if type(case_source) is not bytes or len(case_source) > 262144 or digest(case_source) != admitted.pins.case:
raise Refused('case source pin differs')
descriptor_bytes = json.dumps(asdict(desc), sort_keys=True, separators=(',', ':')).encode()
if admitted.pins.adapter != digest(descriptor_bytes):
raise Refused('descriptor pin differs')
journal = getattr(provisioner, 'group_provenance', None)
if type(journal) is not PostgresGroupProvenance:
raise Refused('trusted owner journal callbacks required')
binding = journal.binding
if (binding.case_digest, binding.source_digest, binding.principal) != (
admitted.pins.case, reviewed, admitted.principal):
raise Refused('owner journal binding differs from this case')
if not all(callable(getattr(provisioner, m, None)) for m in ('ensure_service_groups', 'revoke_service_groups')):
raise Refused('owner group operations required')
if not all(callable(x) for x in (publish, authenticate)) or type(delete_created_if_empty) is not bool:
raise Refused('owner evidence capabilities required')
with contained_output():
client_id, groups, members = _request(yaml.safe_load(case_source))
if client_id != admitted.target:
raise Refused('case target differs')
key = 'warden-ensure:' + admitted.pins.case
request = {'groups': groups, 'members': members, 'correlation_id': admitted.pins.case}
journal_binding = provisioner._group_binding(groups, members, admitted.pins.case)
records = {}
def check_attempt(inv, operation):
if (replace(inv, attempt_id='', operation='admission') != admitted
or inv.operation != operation or not IDENTIFIER.fullmatch(inv.attempt_id)):
raise Refused('attempt binding differs')
if source_digest() != reviewed:
raise Refused('owner source changed')
def execute(inv, rollback, created=()):
with contained_output():
check_attempt(inv, 'rollback' if rollback else 'run')
if rollback:
if not created or set(created) - set(CREATED):
raise Refused('only journal-confirmed creations may be rolled back')
result = provisioner.revoke_service_groups(dict(
request, idempotency_key=key + ':revoke', ensure_idempotency_key=key,
ensure_correlation_id=admitted.pins.case, delete_if_empty=delete_created_if_empty))
expected = {'groups_revoked', 'groups_revoked_and_deleted'}
else:
result = provisioner.ensure_service_groups(dict(request, idempotency_key=key))
expected = {'groups_ensured'}
if (result.status not in expected or result.ensure_idempotency_key != key
or result.ensure_correlation_id != admitted.pins.case
or result.external_subject != ','.join(sorted(groups))):
raise Refused('owner journal result contract differs')
objects = tuple(name for name, items in zip(CREATED, (result.created_groups, result.created_memberships)) if items)
if rollback:
outcome = Outcome.ROLLED_BACK
elif result.resumed or not objects:
outcome, objects = Outcome.NO_CHANGE, ()
else:
outcome = Outcome.SUCCESS
# Names and ids only: the journal holds no secret values.
payload = json.dumps({'schema': 'netkingdom.ensure-groups.attempt.v1',
'invocation': asdict(inv), 'source': reviewed, 'outcome': outcome.value,
'created': objects, 'journal_key': key, 'status': result.status,
'created_groups': list(result.created_groups),
'created_memberships': list(result.created_memberships)},
sort_keys=True, separators=(',', ':')).encode()
with contained_output():
evidence = publish(payload)
if not isinstance(evidence, str) or not IDENTIFIER.fullmatch(evidence) or evidence in records:
raise Refused('owner evidence identifier invalid or reused')
owner = OwnerResult(outcome, evidence, objects, rollback,
attempt_id=inv.attempt_id, operation=inv.operation)
records[evidence] = (inv, payload, owner)
return owner
def verify(inv, result):
record = records.get(result.evidence) if type(result) is OwnerResult else None
if record is None or record[0] != inv or record[2] != result:
return False
with contained_output():
try:
state = journal.get(key, journal_binding)['state']
except Exception:
return False
want = 'revoked' if result.outcome is Outcome.ROLLED_BACK else 'ensured'
return state == want and authenticate(result.evidence, record[1]) is True
return Adapter(REFERENCE, reviewed, lambda inv: execute(inv, False), verify,
lambda inv, created: execute(inv, True, created))

186
sso-mfa/k8s/keycape/test_ensure_groups_adapter.py Normal file
View file

@ -0,0 +1,186 @@
"""Ensure-groups adapter against synthetic journal and provisioner fakes only.
No SQL connection, LLDAP session or native action happens here; the fake
connector raises if anything tries to open it.
"""
from dataclasses import asdict, replace
import importlib.util
import json
from pathlib import Path
import unittest
from unittest.mock import patch
import uuid
from warden.case_run import Invocation, digest
from warden.case_receipt_state import Outcome, Pins, Refused
import yaml
ROOT = Path(__file__).parent
spec = importlib.util.spec_from_file_location('ensure_groups_adapter', ROOT / 'ensure-groups-adapter.py')
a = importlib.util.module_from_spec(spec)
spec.loader.exec_module(a)
from group_provenance_postgres import GroupJournalBinding, PostgresGroupProvenance # noqa: E402
from group_provenance import GroupProvenance # noqa: E402
from provisioner import GroupResult, LLDAPProvisioner # noqa: E402
class Continuity:
def check(self, *args):
return True
def advance(self, *args):
return True
def no_connect():
raise AssertionError('no SQL connection in source tests')
class FakeProvisioner:
_group_binding = staticmethod(LLDAPProvisioner._group_binding)
def __init__(self, journal):
self.group_provenance = journal
self.calls = []
self.state = None
self.created_groups = ({'name': 'forgejo-users', 'id': 7},)
self.created_memberships = ({'group': 'forgejo-users', 'user': 'alice'},)
def _result(self, payload, status, resumed):
return GroupResult('netkingdom-lldap', ','.join(sorted(payload['groups'])), status, resumed,
payload.get('ensure_idempotency_key', payload['idempotency_key']),
payload.get('ensure_correlation_id', payload['correlation_id']),
self.created_groups, self.created_memberships)
def ensure_service_groups(self, payload):
self.calls.append(('ensure', payload))
resumed = self.state == 'ensured'
self.state = 'ensured'
return self._result(payload, 'groups_ensured', resumed)
def revoke_service_groups(self, payload):
self.calls.append(('revoke', payload))
if self.state != 'ensured':
raise RuntimeError('group provenance unresolved')
self.state = 'revoked'
return self._result(payload, 'groups_revoked', False)
class EnsureGroupsTests(unittest.TestCase):
def setUp(self):
self.case = yaml.safe_dump({'spec': {'type': 'oidc-browser-relying-party@1.1', 'parameters': {
'client_id': 'forgejo', 'access_groups': 'forgejo-users,forgejo-admins',
'initial_members': 'forgejo-users=alice'}}}).encode()
desc_bytes = json.dumps(asdict(a.descriptor()), sort_keys=True, separators=(',', ':')).encode()
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)
self.inv = replace(self.admitted, attempt_id='attempt-one', operation='run')
self.journal = self.make_journal()
self.prov = FakeProvisioner(self.journal)
self.evidence = {}
self.kw = dict(admitted=self.admitted, case_source=self.case, provisioner=self.prov,
publish=self.publish, authenticate=lambda r, p: self.evidence.get(r) == p)
get = patch.object(PostgresGroupProvenance, 'get', lambda s, k, b: {'state': self.prov.state})
get.start()
self.addCleanup(get.stop)
def make_journal(self, **override):
values = dict(tenant_scope='platform', directory='netkingdom-lldap',
case_digest=digest(self.case), source_digest=a.source_digest(),
principal='founder', epoch=str(uuid.uuid4()))
values.update(override)
return PostgresGroupProvenance(connect=no_connect, binding=GroupJournalBinding(**values),
continuity=Continuity())
def publish(self, payload):
ref = 'owner-evidence:' + str(len(self.evidence))
self.evidence[ref] = payload
return ref
def test_run_creates_verifies_and_rolls_back_only_creations(self):
adapter = a.build(**self.kw)
result = adapter.invoke(self.inv)
self.assertEqual(result.outcome, Outcome.SUCCESS)
self.assertEqual(result.created, ('service-group', 'service-group-membership'))
self.assertTrue(adapter.verify(self.inv, result))
ensure = self.prov.calls[0][1]
self.assertEqual(ensure['idempotency_key'], 'warden-ensure:' + digest(self.case))
self.assertEqual(ensure['members'], {'forgejo-users': ['alice']})
rb_inv = replace(self.inv, attempt_id='attempt-two', operation='rollback')
rolled = adapter.rollback(rb_inv, result.created)
self.assertEqual(rolled.outcome, Outcome.ROLLED_BACK)
revoke = self.prov.calls[1][1]
self.assertEqual(revoke['ensure_idempotency_key'], ensure['idempotency_key'])
self.assertIs(revoke['delete_if_empty'], False)
self.assertTrue(adapter.verify(rb_inv, rolled))
self.assertFalse(adapter.verify(self.inv, result)) # journal now says revoked
def test_replay_resumes_without_creation(self):
adapter = a.build(**self.kw)
adapter.invoke(self.inv)
again = adapter.invoke(replace(self.inv, attempt_id='attempt-two'))
self.assertEqual((again.outcome, again.created), (Outcome.NO_CHANGE, ()))
def test_refuses_untrusted_or_differing_journal(self):
for journal in (None, GroupProvenance.__new__(GroupProvenance),
self.make_journal(case_digest=digest(b'other')),
self.make_journal(principal='someone-else'),
self.make_journal(source_digest=digest(b'other source'))):
self.prov.group_provenance = journal
with self.assertRaises(Refused):
a.build(**self.kw)
def test_refuses_case_and_invocation_drift(self):
bad = [dict(case_source=self.case + b'\n'),
dict(admitted=replace(self.admitted, target='gitea')),
dict(admitted=replace(self.admitted, step='mint-secret')),
dict(publish=None),
dict(delete_created_if_empty='yes')]
for override in bad:
with self.assertRaises(Refused):
a.build(**{**self.kw, **override})
def test_refuses_non_canonical_or_malformed_case(self):
for params, kind in (({'client_id': 'forgejo', 'access_groups': 'g'}, 'oidc-browser-relying-party@1.2'),
({'client_id': 'forgejo', 'access_groups': 'g', 'initial_members': 'h=alice'}, None),
({'client_id': 'forgejo'}, None)):
case = yaml.safe_dump({'spec': {'type': kind or a.CASE_TYPE, 'parameters': params}}).encode()
pins = replace(self.admitted.pins, case=digest(case))
self.prov.group_provenance = self.make_journal(case_digest=digest(case))
with self.assertRaises(Refused):
a.build(**{**self.kw, 'case_source': case, 'admitted': replace(self.admitted, pins=pins)})
def test_rollback_refuses_unconfirmed_or_foreign_objects(self):
adapter = a.build(**self.kw)
rb_inv = replace(self.inv, operation='rollback')
for created in ((), ('keycape-client',)):
with self.assertRaises(Refused):
adapter.rollback(rb_inv, created)
with self.assertRaises(Refused): # journal has no ensured receipt (contained owner failure)
adapter.rollback(rb_inv, ('service-group',))
self.assertEqual([c[0] for c in self.prov.calls], ['revoke'])
def test_attempt_binding_and_contract_drift(self):
adapter = a.build(**self.kw)
with self.assertRaises(Refused):
adapter.invoke(replace(self.inv, principal='other'))
with self.assertRaises(Refused):
adapter.invoke(replace(self.inv, operation='rollback'))
self.prov.ensure_service_groups = lambda p: GroupResult(
'netkingdom-lldap', 'x', 'groups_ensured', False, 'k', 'c', (), ())
with self.assertRaises(Refused):
adapter.invoke(self.inv)
def test_verify_fails_closed_on_journal_or_evidence_failure(self):
adapter = a.build(**self.kw)
result = adapter.invoke(self.inv)
self.evidence.clear()
self.assertFalse(adapter.verify(self.inv, result))
with patch.object(PostgresGroupProvenance, 'get', side_effect=RuntimeError):
self.assertFalse(adapter.verify(self.inv, result))
if __name__ == '__main__':
unittest.main()