security: prepare scoped Playwright consumer registrations and owner delivery

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2

docs/playwright-registration.md

@@ -0,0 +1,117 @@
+# Playwright registration owner contract
+
+Source preparation under NK-WP-0050; live custody, rollout and consumer acceptance
+remain open. Implements the registration part of RPF-WP-0056 and
+CUST-WP-0080-T05. The existing browser SecurityCaseType remains unchanged: its
+HTTPS/standard-scope admission does not cover this loopback/custom-scope use.
+No case acceptance record is fabricated or borrowed from Forgejo/OpenBao.
+
+## Prepare the non-secret records
+
+Run `python3 sso-mfa/k8s/keycape/playwright-registration.py` for the public
+client, or add `--agents <identity-references.json>` for the complete proposed
+Playwright agent inventory. The latter is an array with one object per existing
+registered identity:
+
+```json
+[
+  {
+    "agent_id": "test-agent",
+    "service_subject": "service:test-agent:platform",
+    "tenant": "tenant:test",
+    "identity_record": "synthetic/agent-record.json"
+  }
+]
+```
+
+This example is synthetic; it is not an admitted consumer. The identity owner
+must confirm each real record, subject, tenant and agent assignment. A reference
+string is not evidence of registration. Initial eligibility already covers all
+identified platform humans and coding agents; no new membership decision is
+needed. The compiler does not provision identities or retrieve/mint credentials.
+
+| Field | Public human client | Each coding agent |
+| --- | --- | --- |
+| client ID | `playwright-cli` | `playwright-agent-<agent_id>` |
+| grant | authorization code, public PKCE | client credentials, confidential |
+| audience | `playwright-kubernetes` | `playwright-kubernetes` |
+| scope | `openid profile groups playwright:connect` | `playwright:connect` |
+| redirect | exact `http://127.0.0.1:18764/callback` | none |
+| subject/tenant | existing directory user and tenant | exact registered subject/tenant supplied by owner |
+| role | existing human roles | only `coding-agent` |
+| lifetime | existing issuer setting, explicit and at most one hour | 15 minutes |
+
+The public client has no per-client lifetime override; KeyCape supports that only
+for client-credentials clients. Existing issuer lifetime outside the accepted
+bound refuses preparation and needs owner reconciliation; this tool does not
+change every client's lifetime. No public tenant override or MFA opt-out is added.
+The callback, dedicated audience and custom scope match Platform's candidate.
+The resource uses access tokens; ID tokens retain the client audience.
+
+Each agent gets a distinct proposed env binding from
+`keycape-<client-id>-client`, key `client-secret`, to
+`KEYCAPE_<CLIENT_ID_WITH_UNDERSCORES>_CLIENT_SECRET`. These names describe a
+proposed delivery contract, **not existing Secrets or admitted custody**. Platform
+must bind an exact OpenBao/ESO and consumer delivery route with its own source
+task and acceptance. No existing OpenBao client or secret is reused. Secret
+values must never enter this identity-reference file, Git, command arguments,
+coordination messages or receipts.
+
+## Registry and activation boundary
+
+The pure `prepare_registry(secret, identities, operation)` adapter reuses
+`keycape-client.py`'s byte-preserving insertion/removal. It validates the entire
+inventory first, refuses differing existing records, ambiguous YAML, duplicate
+IDs, shared secret references and unreviewed registrations for this dedicated
+audience. An identical registration is a no-op. A separate existing OpenBao
+registration for the same agent subject is preserved, with its distinct audience
+and secret reference. Unrelated configuration, including signing material, is
+unchanged. The encoded result contains private registry data and must stay inside
+the protected activation process; the CLI never accepts or emits that registry.
+
+This is not a live apply command. NK-WP-0050-T02 must finish the exact owner
+activation path using existing registry/Deployment operations:
+
+1. Bind verified identity inventory and source digests to the proposal. Obtain
+   the governed per-agent custody and delivery references from Platform.
+2. Pin current cluster, issuer image and object identities. Read the registry
+   only within the existing protected owner execution path. Run the actual
+   pinned KeyCape `check-config` on live and proposed configurations, with the
+   exact Deployment env names and policy-store mode; suppress private diagnostics.
+3. Prepare UID/resource-version/data-bound registry mutation and exact env
+   bindings. Preserve all unrelated data and authentication methods. Retain
+   created-only provenance, restart prerequisites, readback and guarded rollback
+   before proposing any live action. A compare-and-swap conflict means reconcile,
+   never replay a stale whole registry.
+4. Obtain the scoped live action disposition under the existing supervision
+   rules, execute its exact revision, verify config/env/image/readiness and
+   existing human/service login. No new runtime authority comes from this file.
+
+`operation=remove` is an exact inventory removal proposal, not automatic
+rollback: the activation adapter must remove only registrations it created,
+preserve pre-existing identical entries and refuse intervening drift. Keep
+custody material until recovery/retention requirements are met. Disabling/removing
+registration stops new issuance after the issuer loads the change; issued JWTs
+remain valid until expiry, and established connections need separate termination
+through the transport owner. No immediate global revocation is claimed.
+
+## Verification and completion
+
+Python tests exercise preservation, add/no-op/remove, lifetime bounds, identity
+conflicts, distinct credentials and contained invalid-input output. The native
+Go fixture `tests/playwright_registration_native_test.go` is copied into a
+disposable committed KeyCape checkout at `src/internal/authclient/` and consumes
+the merged synthetic configuration through `PLAYWRIGHT_SYNTHETIC_CONFIG`.
+It loads the config and domain registrations, performs native token/JWKS
+exchanges for both client kinds, and checks dedicated claims, wrong secret,
+scope, audience, signature, PKCE and callback refusals. It uses synthetic users
+and seeded authorization sessions; it does not prove upstream login, production
+image compatibility, native Kubernetes CEL validation or consumer eligibility.
+
+T03 needs actual issuer issuance and native API allow/deny receipts, one real
+human and one real registered coding agent independently running Platform's
+consumer command and provider responsive/keyboard trial, renewal, revocation and
+active connection termination. Cluster authentication and ops-bridge transport
+remain their owners' implementations. Platform/provider must accept the receipts
+before managed discovery is published. Same-image abrupt provider recovery is
+already evidenced; cross-version rollback is still explicitly unproven.

sso-mfa/k8s/keycape/playwright-registration.py

@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+"""Prepare Playwright client records; no credential access or live mutation.
+
+The registry adapter below is for an owner-controlled activation ceremony. Its
+encoded result contains the original private registry and must stay in memory.
+The CLI emits only newly constructed non-secret client/binding proposals.
+"""
+from __future__ import annotations
+
+import argparse
+import copy
+from decimal import Decimal
+import importlib.util
+import json
+from pathlib import Path
+import re
+import sys
+
+spec = importlib.util.spec_from_file_location('playwright_registry', Path(__file__).with_name('keycape-client.py'))
+registry = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(registry)
+require = registry.require
+Refused = registry.Refused
+AUDIENCE = 'playwright-kubernetes'
+SCOPE = 'playwright:connect'
+CALLBACK = 'http://127.0.0.1:18764/callback'
+ISSUER = 'https://kc.coulomb.social'
+SLUG = re.compile(r'[a-z0-9][a-z0-9-]{0,39}\Z')
+IDENTIFIER = re.compile(r'[a-z0-9][a-z0-9:._-]{0,159}\Z')
+
+
+def public_client():
+    return {'clientId': 'playwright-cli', 'audience': AUDIENCE,
+            'displayName': 'Playwright platform consumer CLI',
+            'redirectUris': [CALLBACK],
+            'allowedScopes': ['openid', 'profile', 'groups', SCOPE],
+            'grantTypes': ['authorization_code'], 'clientType': 'public'}
+
+
+def agent_client(identity):
+    """Reference an existing registered identity; never manufacture its subject/tenant."""
+    require(isinstance(identity, dict) and set(identity) ==
+            {'agent_id', 'service_subject', 'tenant', 'identity_record'}, 'agent_identity_fields_required')
+    require(all(isinstance(v, str) for v in identity.values()), 'agent_identity_strings_required')
+    require(SLUG.fullmatch(identity['agent_id']) is not None, 'agent_id_invalid')
+    require(identity['service_subject'].startswith('service:') and
+            IDENTIFIER.fullmatch(identity['service_subject']) is not None and
+            len(identity['service_subject']) > len('service:'), 'registered_service_subject_required')
+    require(identity['tenant'].startswith('tenant:') and
+            IDENTIFIER.fullmatch(identity['tenant']) is not None and
+            len(identity['tenant']) > len('tenant:'), 'registered_tenant_required')
+    # This is a reference for owner review, not proof that registration exists.
+    reference = identity['identity_record']
+    require(1 <= len(reference) <= 512 and all(32 < ord(c) < 127 for c in reference),
+            'identity_record_reference_required')
+    cid = 'playwright-agent-' + identity['agent_id']
+    return {'clientId': cid, 'audience': AUDIENCE,
+            'displayName': 'Playwright agent ' + identity['agent_id'],
+            'allowedScopes': [SCOPE], 'grantTypes': ['client_credentials'],
+            'clientType': 'confidential', 'secretRef': 'env:' + registry.env_name(cid),
+            'serviceSubject': identity['service_subject'], 'tenant': identity['tenant'],
+            'roles': ['coding-agent'], 'tokenLifetime': '15m'}
+
+
+def clients(identities):
+    require(isinstance(identities, list) and len(identities) <= 100, 'agent_inventory_invalid')
+    result = [public_client()] + [agent_client(identity) for identity in identities]
+    for field in ['clientId', 'serviceSubject', 'secretRef']:
+        values = [c[field] for c in result if field in c]
+        require(len(values) == len(set(values)), 'duplicate_agent_identity_or_binding')
+    return result
+
+
+def human_lifetime(config):
+    """Bound the existing issuer setting; do not alter all humans' token lifetime."""
+    raw = config.get('tokenLifetime')
+    require(isinstance(raw, str) and 0 < len(raw) <= 128, 'explicit_human_token_lifetime_required')
+    units = {'h': Decimal(3600), 'm': Decimal(60), 's': Decimal(1),
+             'ms': Decimal('.001'), 'us': Decimal('.000001'),
+             'µs': Decimal('.000001'), 'μs': Decimal('.000001'), 'ns': Decimal('.000000001')}
+    parts = re.findall(r'([0-9]+(?:\.[0-9]*)?|\.[0-9]+)(ns|us|µs|μs|ms|s|m|h)', raw)
+    require(parts and ''.join(value + unit for value, unit in parts) == raw,
+            'human_token_lifetime_invalid')
+    seconds = sum((Decimal(value) * units[unit] for value, unit in parts), Decimal(0))
+    require(1 <= seconds <= 3600, 'human_token_lifetime_outside_cluster_bound')
+    return seconds
+
+
+def prepare_registry(secret, identities, operation='add'):
+    """Byte-preserving proposal only. Caller owns CAS, check-config and readback.
+
+    Refuse drift before constructing any result. A missing client during removal
+    is an idempotent no-op. Removal does not revoke issued tokens/connections.
+    """
+    require(operation in ('add', 'remove'), 'operation_invalid')
+    wanted = clients(identities)
+    _, config, _ = registry.pin.issuer_document(secret)
+    require(config.get('issuer') == ISSUER, 'keycape_issuer_mismatch')
+    human_lifetime(config)
+    require(isinstance(config.get('clients'), list), 'client_sequence_required')
+    for existing in config['clients']:
+        require(isinstance(existing, dict), 'client_shape_unexpected')
+        if existing.get('audience') == AUDIENCE:
+            require(existing in wanted, 'unreviewed_playwright_audience_registration')
+    # Preflight every record against the same original, including collisions.
+    for client in wanted:
+        _, current, _ = registry._inspect(secret, client)
+        if 'serviceSubject' in client:
+            for other in current['clients']:
+                if other['clientId'] != client['clientId'] and other.get('audience') == AUDIENCE:
+                    require(other.get('serviceSubject') != client['serviceSubject'],
+                            'playwright_subject_already_registered')
+        matching = [c for c in current['clients'] if c['clientId'] == client['clientId']]
+        require(not matching or matching == [client], 'existing_registration_differs_requires_reconciliation')
+    candidate = copy.deepcopy(secret)
+    changed_ids = []
+    merge = registry.replacement if operation == 'add' else registry.removal
+    for client in wanted:
+        encoded, changed = merge(candidate, client)
+        candidate['data']['config.yaml'] = encoded
+        if changed:
+            changed_ids.append(client['clientId'])
+    return candidate['data']['config.yaml'], changed_ids
+
+
+def unique_object(pairs):
+    result = {}
+    for key, value in pairs:
+        require(key not in result, 'duplicate_input_field')
+        result[key] = value
+    return result
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__.split('\n', 1)[0])
+    parser.add_argument('--agents', help='Non-secret JSON array of existing agent identity references')
+    args = parser.parse_args(argv)
+    try:
+        identities = []
+        if args.agents:
+            with Path(args.agents).open('rb') as stream:
+                raw = stream.read(65537)
+            require(len(raw) <= 65536, 'agent_inventory_too_large')
+            identities = json.loads(raw, object_pairs_hook=unique_object)
+        records = clients(identities)
+        print(json.dumps({'schema': 'netkingdom.playwright-registration.v1',
+                          'status': 'source_proposal', 'clients': records,
+                          'identity_references': {c['clientId']: identity['identity_record']
+                                                  for c, identity in zip(records[1:], identities)},
+                          'deployment_env': [registry.env_binding(c) for c in records if 'secretRef' in c],
+                          'identity_evidence_verified': False,
+                          'custody_admitted': False, 'live_activated': False}, indent=2))
+        return 0
+    except Refused as error:
+        print(json.dumps({'status': 'refused', 'reason': str(error)}))
+    except Exception:
+        print(json.dumps({'status': 'refused', 'reason': 'invalid_registration_input'}))
+    return 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())

sso-mfa/k8s/keycape/test_playwright_registration.py

@@ -0,0 +1,150 @@
+import base64
+import contextlib
+import copy
+import importlib.util
+import io
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+import yaml
+
+spec = importlib.util.spec_from_file_location('playwright_registration', Path(__file__).with_name('playwright-registration.py'))
+p = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(p)
+IDENTITY = {'agent_id': 'test-agent', 'service_subject': 'service:test-agent:platform',
+            'tenant': 'tenant:test', 'identity_record': 'synthetic/agent-record.json'}
+
+
+def fixture():
+    raw = ('# preserve the issuer and existing clients\nissuer: https://kc.coulomb.social\n'
+           'port: 8080\nprivateKeyPem: /synthetic/key.pem\ntokenLifetime: 15m\n'
+           'authelia:\n  issuer: https://auth.coulomb.social\n  clientSecret: synthetic-private-marker\n'
+           'clients:\n  - clientId: existing\n    clientType: public\n'
+           '    redirectUris: [https://example.test/callback]\n'
+           '    allowedScopes: [openid]\n    grantTypes: [authorization_code]\n'
+           '# trailing configuration comment\nlogLevel: info\n')
+    return {'metadata': {'uid': 'synthetic', 'resourceVersion': '7'},
+            'data': {'config.yaml': base64.b64encode(raw.encode()).decode(), 'other': 'preserve'}}, raw
+
+
+def decode(value):
+    return base64.b64decode(value).decode()
+
+
+def altered(secret, change):
+    result = copy.deepcopy(secret)
+    config = yaml.safe_load(decode(result['data']['config.yaml']))
+    change(config)
+    result['data']['config.yaml'] = base64.b64encode(yaml.safe_dump(config, sort_keys=False).encode()).decode()
+    return result
+
+
+class RegistrationTests(unittest.TestCase):
+    def test_public_matches_platform_and_agent_has_distinct_delivery(self):
+        client = p.public_client()
+        self.assertEqual(client['redirectUris'], ['http://127.0.0.1:18764/callback'])
+        self.assertNotIn('tokenLifetime', client)
+        self.assertNotIn('secretRef', client)
+        agent = p.agent_client(IDENTITY)
+        self.assertEqual(agent['allowedScopes'], ['playwright:connect'])
+        self.assertEqual(agent['roles'], ['coding-agent'])
+        self.assertEqual(agent['tenant'], IDENTITY['tenant'])
+        self.assertEqual(agent['serviceSubject'], IDENTITY['service_subject'])
+        self.assertEqual(agent['tokenLifetime'], '15m')
+        self.assertEqual(p.registry.env_binding(agent)['valueFrom']['secretKeyRef'],
+                         {'name': 'keycape-playwright-agent-test-agent-client', 'key': 'client-secret'})
+
+    def test_unknown_or_missing_identity_fields_refused(self):
+        for identity in [{}, [], dict(IDENTITY, secret='not-emitted'), dict(IDENTITY, tenant=''),
+                         dict(IDENTITY, service_subject='human:someone'),
+                         dict(IDENTITY, identity_record='bad\nreference'),
+                         dict(IDENTITY, agent_id='bad_$agent'), dict(IDENTITY, tenant=True)]:
+            with self.subTest(identity=identity), self.assertRaises(p.Refused):
+                p.agent_client(identity)
+
+    def test_duplicate_agent_or_subject_refused(self):
+        for second in [IDENTITY, dict(IDENTITY, agent_id='second'),
+                       dict(IDENTITY, service_subject='service:second')]:
+            with self.subTest(second=second), self.assertRaises(p.Refused):
+                p.clients([IDENTITY, second])
+
+    def test_go_lifetimes_are_bounded_without_global_change(self):
+        for value in ['15m', '1h', '1h0m0s', '3599.5s', '1000ms', '1.5m', '.5m']:
+            with self.subTest(value=value): p.human_lifetime({'tokenLifetime': value})
+        for value in [None, 900, '', '0s', '-1m', '1h1ns', '0.5s', 'NaN', '15mgarbage', '1d']:
+            with self.subTest(value=value), self.assertRaises(p.Refused):
+                p.human_lifetime({'tokenLifetime': value})
+
+    def test_add_preserves_bytes_and_is_repeatable_then_exact_remove(self):
+        secret, raw = fixture()
+        original = copy.deepcopy(secret)
+        encoded, changed = p.prepare_registry(secret, [IDENTITY])
+        self.assertEqual(changed, ['playwright-cli', 'playwright-agent-test-agent'])
+        updated = decode(encoded)
+        prefix, suffix = raw.split('logLevel:')
+        self.assertTrue(updated.startswith(prefix))
+        self.assertTrue(updated.endswith('logLevel:' + suffix))
+        self.assertEqual(secret, original)
+        current = copy.deepcopy(secret); current['data']['config.yaml'] = encoded
+        self.assertEqual(p.prepare_registry(current, [IDENTITY]), (encoded, []))
+        restored, removed = p.prepare_registry(current, [IDENTITY], 'remove')
+        self.assertEqual(decode(restored), raw)
+        self.assertEqual(removed, changed)
+        self.assertEqual(p.prepare_registry(secret, [IDENTITY], 'remove'),
+                         (secret['data']['config.yaml'], []))
+
+    def test_refuses_drift_conflicts_and_unreviewed_audience(self):
+        base, _ = fixture()
+        public = p.public_client(); agent = p.agent_client(IDENTITY)
+        cases = [lambda c: c.update(issuer='https://other.test'),
+                 lambda c: c.update(tokenLifetime='2h'),
+                 lambda c: c['clients'].append(dict(public, allowedScopes=['openid'])),
+                 lambda c: c['clients'].append(dict(public, clientId='other-consumer')),
+                 lambda c: c['clients'].append(dict(agent, clientId='duplicate-subject')),
+                 lambda c: c['clients'].append({'clientId': 'other', 'secretRef': agent['secretRef']}),
+                 lambda c: c['clients'].append(copy.deepcopy(c['clients'][0]))]
+        for change in cases:
+            for operation in ['add', 'remove']:
+                with self.subTest(change=change, operation=operation), self.assertRaises(p.Refused):
+                    p.prepare_registry(altered(base, change), [IDENTITY], operation)
+
+    def test_same_agent_can_keep_separate_existing_openbao_client(self):
+        base, _ = fixture()
+        existing = dict(p.agent_client(IDENTITY), clientId='existing-openbao-agent',
+                        audience='openbao', allowedScopes=['openbao:login'],
+                        secretRef='env:EXISTING_OPENBAO_SECRET')
+        secret = altered(base, lambda c: c['clients'].append(existing))
+        encoded, _ = p.prepare_registry(secret, [IDENTITY])
+        self.assertIn(existing, yaml.safe_load(decode(encoded))['clients'])
+
+    def test_cli_outputs_only_proposal_and_no_false_admission(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'agents.json'; path.write_text(json.dumps([IDENTITY]))
+            out = io.StringIO()
+            with contextlib.redirect_stdout(out): code = p.main(['--agents', str(path)])
+            self.assertEqual(code, 0)
+            result = json.loads(out.getvalue())
+            self.assertEqual(result['status'], 'source_proposal')
+            for flag in ['identity_evidence_verified', 'custody_admitted', 'live_activated']:
+                self.assertFalse(result[flag])
+            self.assertEqual(len(result['clients']), 2)
+            self.assertEqual(result['identity_references']['playwright-agent-test-agent'],
+                             IDENTITY['identity_record'])
+
+    def test_cli_contains_invalid_private_or_duplicate_input(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'input.json'
+            for raw in ['synthetic-private-marker', '[{"agent_id":"a","agent_id":"b"}]',
+                        json.dumps([dict(IDENTITY, secret='synthetic-private-marker')]), 'x' * 65537]:
+                path.write_text(raw)
+                out = io.StringIO()
+                with contextlib.redirect_stdout(out): code = p.main(['--agents', str(path)])
+                self.assertEqual(code, 1)
+                self.assertNotIn('synthetic-private-marker', out.getvalue())
+                self.assertEqual(json.loads(out.getvalue())['status'], 'refused')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/playwright_registration_native_test.go

@@ -0,0 +1,120 @@
+// Copy into a disposable KeyCape src/internal/authclient checkout. This uses
+// its native token endpoint/JWKS verifier, with a synthetic directory and seeded
+// authorization sessions. It does not prove upstream login or Kubernetes access.
+package authclient
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/base64"
+	"net/url"
+	"os"
+	"testing"
+	"time"
+
+	"keycape/internal/config"
+	"keycape/internal/domain"
+	"keycape/internal/server/oidc"
+)
+
+type playwrightUsers struct{ users }
+
+func (playwrightUsers) LookupUser(context.Context, string) (*domain.User, error) {
+	return &domain.User{ID: "user:playwright-test", Username: "test", Enabled: true, Tenant: "tenant:test"}, nil
+}
+
+func TestNetKingdomPlaywrightGeneratedRegistration(t *testing.T) {
+	path := os.Getenv("PLAYWRIGHT_SYNTHETIC_CONFIG")
+	if path == "" {
+		t.Fatal("synthetic generated config required")
+	}
+	cfg, err := config.Load(path)
+	if err != nil {
+		t.Fatal("generated config did not load")
+	}
+	if len(config.ValidateConfig(cfg)) != 0 {
+		t.Fatal("native config validation failed")
+	}
+	records, err := cfg.Registrations()
+	if err != nil {
+		t.Fatal("native registration conversion failed")
+	}
+	c, d, h := provider(t)
+	h.Users = playwrightUsers{}
+	h.ClientConfig = map[string]*domain.Client{}
+	for i := range records {
+		record := &records[i]
+		if record.ClientType == "confidential" {
+			record.ClientSecret = "synthetic-playwright-fixture"
+		}
+		h.ClientConfig[record.ClientID] = record
+	}
+	ctx := context.Background()
+	agentID := "playwright-agent-test-agent"
+	form := url.Values{"grant_type": {"client_credentials"}, "scope": {"playwright:connect"}}
+	token, err := c.Exchange(ctx, d, form, agentID, "synthetic-playwright-fixture", "playwright-kubernetes", "")
+	if err != nil {
+		t.Fatal("generated service exchange failed")
+	}
+	claims, err := c.Verify(ctx, d, token.AccessToken, "playwright-kubernetes", "")
+	if err != nil || claims["principal_type"] != "service" || claims["sub"] != "service:test-agent:platform" || claims["tenant"] != "tenant:test" {
+		t.Fatal("service claim contract mismatch")
+	}
+	roles, ok := claims["roles"].([]any)
+	if !ok || len(roles) != 1 || roles[0] != "coding-agent" {
+		t.Fatal("agent role mismatch")
+	}
+	if token.ExpiresIn != 900 {
+		t.Fatal("agent lifetime mismatch")
+	}
+	if _, err = c.Verify(ctx, d, token.AccessToken, "openbao", ""); err == nil {
+		t.Fatal("wrong audience accepted")
+	}
+	if _, err = c.Verify(ctx, d, token.AccessToken+"tampered", "playwright-kubernetes", ""); err == nil {
+		t.Fatal("tampered token accepted")
+	}
+	if _, err = c.Exchange(ctx, d, form, agentID, "wrong", "playwright-kubernetes", ""); err == nil {
+		t.Fatal("wrong secret accepted")
+	}
+	form.Set("scope", "openbao:login")
+	if _, err = c.Exchange(ctx, d, form, agentID, "synthetic-playwright-fixture", "playwright-kubernetes", ""); err == nil {
+		t.Fatal("excess scope accepted")
+	}
+
+	verifier := "synthetic-playwright-verifier-with-enough-length-12345"
+	digest := sha256.Sum256([]byte(verifier))
+	callback := "http://127.0.0.1:18764/callback"
+	exchange := func(sentVerifier, sentCallback string) (Tokens, error) {
+		code := h.Sessions.Create(&oidc.PKCESession{ClientID: "playwright-cli", Username: "test", Nonce: "synthetic-nonce",
+			Scopes:        []string{"openid", "profile", "groups", "playwright:connect"},
+			PKCEChallenge: base64.RawURLEncoding.EncodeToString(digest[:]), RedirectURI: callback, ExpiresAt: time.Now().Add(time.Minute)})
+		fields := url.Values{"grant_type": {"authorization_code"}, "client_id": {"playwright-cli"}, "code": {code},
+			"redirect_uri": {sentCallback}, "code_verifier": {sentVerifier}, "scope": {"openid playwright:connect"}}
+		return c.Exchange(ctx, d, fields, "playwright-cli", "", "playwright-kubernetes", "synthetic-nonce")
+	}
+	human, err := exchange(verifier, callback)
+	if err != nil {
+		t.Fatal("public PKCE token exchange failed")
+	}
+	claims, err = c.Verify(ctx, d, human.AccessToken, "playwright-kubernetes", "")
+	if err != nil || claims["principal_type"] != "human" || claims["sub"] != "user:playwright-test" || claims["tenant"] != "tenant:test" {
+		t.Fatal("human claim contract mismatch")
+	}
+	if _, err = c.Verify(ctx, d, human.IDToken, "playwright-kubernetes", ""); err == nil {
+		t.Fatal("ID token accepted as resource token")
+	}
+	if _, err = exchange("wrong-verifier", callback); err == nil {
+		t.Fatal("wrong PKCE verifier accepted")
+	}
+	if _, err = exchange(verifier, "http://127.0.0.1:18765/callback"); err == nil {
+		t.Fatal("wrong callback accepted")
+	}
+	delete(h.ClientConfig, agentID)
+	form.Set("scope", "playwright:connect")
+	if _, err = c.Exchange(ctx, d, form, agentID, "synthetic-playwright-fixture", "playwright-kubernetes", ""); err == nil {
+		t.Fatal("removed registration still issues")
+	}
+	if _, err = c.Verify(ctx, d, token.AccessToken, "playwright-kubernetes", ""); err != nil {
+		t.Fatal("issued JWT lifetime limitation changed")
+	}
+}

workplans/NK-WP-0050-playwright-consumer-registration.md

@@ -0,0 +1,84 @@
+---
+id: NK-WP-0050
+type: workplan
+title: "Register Playwright human and agent consumers through governed identity delivery"
+domain: infotech
+repo: net-kingdom
+status: active
+flavor: implementation
+owner: codex
+topic_slug: netkingdom
+created: "2026-10-13"
+updated: "2026-10-13"
+depends_on: []
+related: [CUST-WP-0080, RPF-WP-0056, RAPP-PLAYWRIGHT-WP-0004]
+---
+
+Accept NetKingdom's concrete implementation request
+`c6b4dfb0-44d5-4f12-8854-2c38ab6dc454` (original
+`94a0096c-f34c-463c-b336-b13cf629b968`). This is the missing registration-owner
+implementation, not another cross-owner coordination plan. Coordination remains
+CUST-WP-0080-T05; Platform and provider retain consumer/service acceptance.
+Initial eligibility of all identified platform humans and coding agents is
+already decided. Do not request another membership approval or reuse a shared
+admin identity, kubeconfig, SSH identity or existing OpenBao credential.
+
+## Prepare exact registrations and preservation checks
+
+```task
+id: NK-WP-0050-T01
+status: done
+priority: high
+```
+
+`docs/playwright-registration.md` owns the bounded registration contract.
+`sso-mfa/k8s/keycape/playwright-registration.py` constructs the exact public
+PKCE record and distinct per-agent confidential registrations from existing
+identity references, with dedicated audience/scope and 15-minute agent lifetime.
+It reuses the existing byte-preserving registry adapter, refuses drift/shared
+references/unreviewed audience registrations and bounds the existing human
+issuer lifetime without modifying it. CLI emits only source proposals; no
+credential access, live mutation or fabricated SecurityCaseType acceptance.
+
+Verified: nine new compiler/merge tests and 64 existing registration tests pass.
+The broader 159-test directory run passed 157 and failed two unchanged short-window
+OpenBao input tests; all seven tests in that module passed unchanged on rerun.
+Native KeyCape config/startup suites and the generated-registration token/JWKS
+fixture pass against committed issuer source 1344fc80e56315c5c99f58993f9ee02ce79e8a76.
+Public output exactly matches Platform's candidate. Synthetic identity references
+and local fixtures are not issuer-owner or real consumer acceptance.
+
+## Deliver custody integration and guarded registration activation
+
+```task
+id: NK-WP-0050-T02
+status: todo
+priority: high
+depends_on: [NK-WP-0050-T01]
+```
+
+Resolve the exact real agent records and Platform custody/delivery source task;
+add protected registration/config/Deployment integration using existing owner
+operations, exact source/object/revision bindings, pinned native check-config,
+created-only provenance, readback and guarded rollback. Prepare the concrete
+action before any needed privileged approval. Source work can proceed while
+owner acceptance is collected; custody and live grants must not be invented.
+Neither compiler output nor another client's acceptance authorizes live apply.
+
+## Verify issuer and real consumer acceptance
+
+```task
+id: NK-WP-0050-T03
+status: wait
+priority: high
+depends_on: [NK-WP-0050-T02, RPF-WP-0056-T03]
+blocking_reason: "Await owner-scoped registry/custody activation, native cluster authentication and individual transport, then real human and agent acceptance under CUST-WP-0080-T05."
+```
+
+Retain real human/agent issuance and identity/tenant/audience/scope/expiry proof,
+native negative cases, preserved existing login, renewal, removal and active
+connection termination. Hand the exact receipts to RPF-WP-0056-T03 and provider
+RAPP-PLAYWRIGHT-WP-0004-T01/T03. Do not introduce a reciprocal prerequisite on
+Platform activation: T02 owns issuer preparation/activation, T03 consumes final
+Platform end-to-end acceptance. Close only after the registration owner's
+requirements are proven; hand any residuals to live work records first.