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

Publish reviewed KEY39 lifecycle source and preserve failed review history
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
This commit is contained in:
tegwick 2026-10-11 15:33:14 +02:00
parent 98f09f4d4b
commit dfc357eb8a
6 changed files with 1540 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

53
docs/reviews/2026-10-11-key39-lifecycle-v3-netkingdom-review.json Normal file
View file

@ -0,0 +1,53 @@
{
"reviewer": "net-kingdom-owner (agent net-kingdom, codex)",
"decision": "changes_requested",
"review_id": "NK-REVIEW-20261011-KEY39-LIFECYCLE-V3",
"work_item_id": "NK-WP-ADHOC-2026-10-11-T07",
"reviewed_proposal_id": "KEY-WP-0039-T02/attended-lifecycle-v3",
"reviewed_manifest_sha256": "fbc31fda2b6194747547b9fbe876095914c3d53231c79abd43a010dbb3273efa",
"reviewed_executor_sha256": "1161b3666ebf1f3eb322c7d5112584bce7e074da7e7650c5840e3c5de9311194",
"reviewed_tests_sha256": "75e26a9be4422be539f7e0a4d46c3dd6d8e2eab21d3cb6e4ba03cfe8c7c1b9cd",
"reviewed_runner_sha256": "bb9610babf44a959342bac700bb5712e97f400b44c4d0659d21df4426e616e19",
"previous_review_sha256": "f7293bc3cc896e23840d767bf86bd379f202a665f6a334485081831ad067d262",
"comment": "NetKingdom INTENT Taxonomy declaration and Principles 6/7 require bounded exposure with owner-held execution. Same completed work item NK-WP-ADHOC-2026-10-11-T07 independently reviews frozen v3 manifest fbc31fda2b6194747547b9fbe876095914c3d53231c79abd43a010dbb3273efa. V2 findings are resolved, but the new aggregate cleanup bound bypass below needs correction. This is source judgement only, not founder approval.",
"v2_findings_status": {
"NK-KEY39-V2-F01": "resolved: strict finite nonbool expiry and observation validation, JSON constants refused, expiry limited to maxsession",
"NK-KEY39-V2-F02": "resolved: total SIGALRM request guard clips wall and monotonic deadline; postcompletion recheck; known late accepted session revoked privately; unknown partial response retained"
},
"findings": [
{
"id": "NK-KEY39-V3-F01",
"severity": "medium",
"location": "openbao-negative-session.py:delete_temporary",
"counterexample": "With CLEANING=True and CLEANUP_MONOTONIC_DEADLINE=time.monotonic()-1, mocked subprocess.run records one Secret DELETE; only subsequent guarded k absence GET raises cleanup_budget_expired.",
"impact": "The final native Secret deletion bypasses the promised nonrenewable aggregate cleanup deadline and its remaining-time cap.",
"requested_change": "Route the exact UID/resourceVersion-precondition raw DeleteOptions request through guarded k or equivalent precheck/remaining-budget/total-timeout/postcheck. Add expired-budget no-call and near-expiry total-duration regressions."
}
],
"validation": {
"supplied_tests": 39,
"independent_tests": 15,
"all_passed": true,
"counterexamples_reproduced": true
},
"coverage": [
"Prior eleven independent checks adapted to v3; v2 counterexamples now rejected",
"Real SIGALRM slow-body interruption and signal/timer restoration",
"Partial accepted login timeout records only unknown-outcome boolean",
"Aggregate cleanup is nonrenewable and guarded k refuses after expiry",
"Direct final delete bypass reproduced independently",
"CAS/source/runtime pins, temporary registration/environment isolation, output suppression, normal cleanup and hard-interruption recovery inspected unchanged"
],
"authority_boundary": {
"source_preparation_only": true,
"native_actions": false,
"owner_checkout_writes": false,
"credential_access": false,
"secret_reads": false,
"foreign_reviews": false,
"founder_approval": false,
"native_acceptance": false,
"KEY39_task_closure": false
},
"work_record_anchor_sha256": "aa56b9cd64301f031cedd3cbd703e1698605fe97b0fee7b9ba45343333ff54f5"
}

64
docs/reviews/2026-10-11-key39-lifecycle-v4-netkingdom-review.json Normal file
View file

@ -0,0 +1,64 @@
{
"reviewer": "net-kingdom-owner (agent net-kingdom, codex)",
"decision": "approved",
"review_id": "NK-REVIEW-20261011-KEY39-LIFECYCLE-V4",
"work_item_id": "NK-WP-ADHOC-2026-10-11-T07",
"reviewed_proposal_id": "KEY-WP-0039-T02/attended-lifecycle-v4",
"reviewed_manifest_sha256": "d079f85c47d5c2635e1ee633ea4d1346b1737ce2be9ca273f22a38e620f418d8",
"reviewed_source_hashes": {
"sso-mfa/k8s/keycape/openbao-negative-session.py": "b4687c2d07819c43b45d0cddcad0488b62eb49175b2a06a43f7806a0838d3a4d",
"sso-mfa/k8s/keycape/test_openbao_negative_session.py": "ea2b5f7466c1da016f528093daf168c274238d4803eb5424d7353c66adbecf43"
},
"reviewed_runner_sha256": "bb9610babf44a959342bac700bb5712e97f400b44c4d0659d21df4426e616e19",
"previous_review_sha256": "06ac6616e75eb7690836fbce94b30ebecd3a674a3b256118d180ed13c90db3b1",
"original_v2_review_sha256": "f7293bc3cc896e23840d767bf86bd379f202a665f6a334485081831ad067d262",
"comment": "NetKingdom INTENT Taxonomy declaration, Principles 6/7 and SECURITY-COMPANION sections 3/7 support this bounded owner-held source preparation. Exact frozen v4 manifest d079f85c47d5c2635e1ee633ea4d1346b1737ce2be9ca273f22a38e620f418d8 reviewed under same completed own work item NK-WP-ADHOC-2026-10-11-T07. Technical source agreement: all independently reproduced v2/v3 gaps resolve. This owner source judgement supplies neither human approval nor native execution authority/acceptance and does not close KEY39 T02.",
"findings": [],
"prior_findings_status": {
"NK-KEY39-V2-F01": "resolved: finite nonbool approval and role observation values; integer bounded session; nonfinite JSON constants refused",
"NK-KEY39-V2-F02": "resolved: total Linux main-thread SIGALRM bound across connect/headers/body; finite per-request caps clipped to wall and monotonic action deadlines; postcompletion check and bounded late-session revoke",
"NK-KEY39-V3-F01": "resolved: final UID/resourceVersion raw DeleteOptions now uses guarded k; expired aggregate issues no delete and slow delete is clipped to remaining time"
},
"independent_coverage": [
"Exact manifest/executor/test/runner pins independently verified; originals unchanged",
"CAS patch tests require UID and resourceVersion; both dry runs and baseline UID/spec/config pins inspected",
"Six temporary registrations/env refs and separate Secret with no production credential replacement inspected; single replica Recreate causes two intended restarts",
"Captured subprocess stdout/stderr, fixed refusal messages, private stdin, value-free receipts and exclusive receipt creation inspected; synthetic raw error suppression verified",
"Normal cleanup refuses concurrent config/spec drift, restores baseline/readback, proves precise removed-client denial then UID/resourceVersion-precondition delete",
"Hard-interruption recovery requires separate recovery-approved envelope, reconstructs exact registry/spec baseline, refuses partial registrations/env and baseline drift; uncertain creation UID needs independent owner acknowledgement; issued OpenBao sessions require separate reconciliation",
"Fixed token/discovery/login/revoke allowlist, no ambient proxy/redirect, response close and byte bound inspected; unreviewed paths/methods refuse",
"Future/stale role observations now refuse; monotonic probe cleanup reserve inspected; two residual deadline counterexamples reproduced",
"Real elapsed slow HTTP interruption and SIGALRM timer/handler restoration",
"Late full accepted OpenBao session revoked privately then result refused; partial accepted response timeout retains value-free unknown-outcome evidence",
"Cleanup aggregate deadline starts once and cannot renew; guarded k refuses after expiry; exact final delete now shares it",
"Real elapsed slow deletion clipped to remaining cleanup budget; approved runtime/source pins and precise Secret deletion identity preserved"
],
"validation": {
"supplied_lifecycle_tests": 41,
"unchanged_reviewed_runner_tests": 17,
"independent_tests": 17,
"all_passed": true,
"network_used": false,
"kubernetes_used": false
},
"limitations": [
"Linux main-thread setitimer required; preexisting timer or worker thread refuses",
"Fixture checks establish source behavior only; no native role/policy, headroom, runtime readiness or successful cleanup observed",
"Owner-held token producer remains unresolved; no credential route authorized",
"Unknown/failed session revocation requires separate owner reconciliation; cleanup alone is not revocation proof",
"Founder exact source/runtime/window approval and each independent owner acknowledgment remain necessary before native execution"
],
"authority_boundary": {
"source_preparation_only": true,
"native_actions": false,
"owner_checkout_writes": false,
"credential_access": false,
"secret_reads": false,
"db_access": false,
"foreign_reviews": false,
"founder_approval": false,
"native_acceptance": false,
"KEY39_task_closure": false
},
"work_record_anchor_sha256": "3650c1eeadd74e150e528766523a284b03d02004749c04e9b6518d42f3e32531"
}

71
docs/reviews/2026-10-11-key39-lifecycle-v5-netkingdom-review.json Normal file
View file

@ -0,0 +1,71 @@
{
"reviewer": "net-kingdom-owner (agent net-kingdom, codex)",
"decision": "approved",
"review_id": "NK-REVIEW-20261011-KEY39-LIFECYCLE-V5",
"work_item_id": "NK-WP-ADHOC-2026-10-11-T07",
"reviewed_proposal_id": "KEY-WP-0039-T02/attended-lifecycle-v5",
"reviewed_manifest_sha256": "d1f997624db46693a7126a6eb6435f6b3a353d5bd67dcc8a539683e4b7147285",
"reviewed_source_hashes": {
"sso-mfa/k8s/keycape/openbao-negative-session.py": "3bcceebc845b17492040c76223d77088257c547a2b890198e7b5c2b04a0cd360",
"sso-mfa/k8s/keycape/test_openbao_negative_session.py": "a428a4a5ea60bd6c4ad10481e133227ad696a4a08653bf39b554b47431fb9f8c"
},
"reviewed_runner_sha256": "bb9610babf44a959342bac700bb5712e97f400b44c4d0659d21df4426e616e19",
"previous_review_sha256": "4f4498203a31bd46dbc3715c0df1cbb9eee63bed5b2c1baf430ff3c4a37c13cb",
"original_v2_review_sha256": "f7293bc3cc896e23840d767bf86bd379f202a665f6a334485081831ad067d262",
"comment": "NetKingdom INTENT Taxonomy declaration, Principles 6/7 and SECURITY-COMPANION sections 3/7 support bounded owner-held source preparation. Exact frozen v5 manifest d1f997624db46693a7126a6eb6435f6b3a353d5bd67dcc8a539683e4b7147285 reviewed under same completed own work item NK-WP-ADHOC-2026-10-11-T07. Technical source agreement includes bounded exceptional HTTP response cleanup and preserves all earlier failed/revised owner evidence. This owner judgement supplies neither human approval nor native execution authority/acceptance and does not close KEY39 T02.",
"findings": [],
"prior_findings_status": {
"NK-KEY39-V2-F01": "resolved: finite nonbool approval and role observation values; integer bounded session; nonfinite JSON constants refused",
"NK-KEY39-V2-F02": "resolved: total Linux main-thread SIGALRM bound across connect/headers/body; finite per-request caps clipped to wall and monotonic action deadlines; postcompletion check and bounded late-session revoke",
"NK-KEY39-V3-F01": "resolved: final UID/resourceVersion raw DeleteOptions now uses guarded k; expired aggregate issues no delete and slow delete is clipped to remaining time",
"Platform-v4-exceptional-close": "resolved in exact v5: normal close remains within request timer; one exceptional close uses at most 20 ms clipped to positive original remainder and finite nonrenewable cleanup remainder; expired cleanup performs no close; abort failure remains unconfirmed, never successful proof"
},
"independent_coverage": [
"Exact manifest/executor/test/runner pins independently verified; originals unchanged",
"CAS patch tests require UID and resourceVersion; both dry runs and baseline UID/spec/config pins inspected",
"Six temporary registrations/env refs and separate Secret with no production credential replacement inspected; single replica Recreate causes two intended restarts",
"Captured subprocess stdout/stderr, fixed refusal messages, private stdin, value-free receipts and exclusive receipt creation inspected; synthetic raw error suppression verified",
"Normal cleanup refuses concurrent config/spec drift, restores baseline/readback, proves precise removed-client denial then UID/resourceVersion-precondition delete",
"Hard-interruption recovery requires separate recovery-approved envelope, reconstructs exact registry/spec baseline, refuses partial registrations/env and baseline drift; uncertain creation UID needs independent owner acknowledgement; issued OpenBao sessions require separate reconciliation",
"Fixed token/discovery/login/revoke allowlist, no ambient proxy/redirect, response close and byte bound inspected; unreviewed paths/methods refuse",
"Future/stale role observations now refuse; monotonic probe cleanup reserve inspected; two residual deadline counterexamples reproduced",
"Real elapsed slow HTTP interruption and SIGALRM timer/handler restoration",
"Late full accepted OpenBao session revoked privately then result refused; partial accepted response timeout retains value-free unknown-outcome evidence",
"Cleanup aggregate deadline starts once and cannot renew; guarded k refuses after expiry; exact final delete now shares it",
"Real elapsed slow deletion clipped to remaining cleanup budget; approved runtime/source pins and precise Secret deletion identity preserved",
"Independent early RuntimeError read plus slow exceptional close; fixed error and no protected text in receipt",
"Independent timed-out body plus slow abort cap; real elapsed timer restores handler and timer",
"Independent normal-close error on known accepted session; private bounded self-revoke then abort failure",
"Independent expired aggregate exceptional-close zero call/no deadline renewal; prior guard suite adapted only for newly added value-free abort receipt booleans"
],
"validation": {
"supplied_lifecycle_tests": 46,
"unchanged_reviewed_runner_tests": 17,
"independent_tests": 21,
"all_passed": true,
"network_used": false,
"kubernetes_used": false,
"review_test_adaptation": "Earlier receipt-equality assertion expected only unknown-outcome flag; v5 legitimately adds abort booleans. Updated assertion verifies those booleans and value-free shape; no source defect inferred."
},
"limitations": [
"Linux main-thread setitimer required; preexisting timer or worker thread refuses",
"Fixture checks establish source behavior only; no native role/policy, headroom, runtime readiness or successful cleanup observed",
"Owner-held token producer remains unresolved; no credential route authorized",
"Unknown/failed session revocation requires separate owner reconciliation; cleanup alone is not revocation proof",
"Founder exact source/runtime/window approval and each independent owner acknowledgment remain necessary before native execution"
],
"authority_boundary": {
"source_preparation_only": true,
"native_actions": false,
"owner_checkout_writes": false,
"credential_access": false,
"secret_reads": false,
"db_access": false,
"foreign_reviews": false,
"founder_approval": false,
"native_acceptance": false,
"KEY39_task_closure": false
},
"work_record_anchor_sha256": "17c1e1a798516509d6ddb4d5c93090910cc6d3dd7b516827abf4fd3031ef80b3",
"platform_previous_review_sha256": "c0f9da4850d7b6ce8a41401bfb7cddc39a60cd7bc8767de89d76b5f93198a2d9"
}

712
sso-mfa/k8s/keycape/openbao-negative-session.py Normal file
View file

@ -0,0 +1,712 @@
#!/usr/bin/env python3
"""Founder-attended KEY39 lifecycle. Silent; private values never leave memory/K8s.
inspect is metadata-only (no Secret read). execute requires an exact approved
envelope, source/runtime pins and founder-held private tokens on stdin. No login,
custody fetch, production credential, role or policy mutation is implemented.
"""
import argparse
import base64
import copy
import hashlib
import importlib.util
import json
import math
import os
from pathlib import Path
import secrets
import signal
import subprocess
import threading
import time
from contextlib import contextmanager
import urllib.parse
import urllib.request
import urllib.error
import yaml
CLUSTER = 'a553c742-0115-43d4-99a4-a5ca56fe0786'
TARGET = 'sso/keycape'
PROPOSAL = 'KEY-WP-0039-T02/attended-lifecycle-v5'
IDS = ['openbao-probe-' + c for c in ['aud', 'sub', 'tenant', 'roles', 'scope', 'expiry']]
ENVS = ['KEYCAPE_' + c.upper().replace('-', '_') + '_CLIENT_SECRET' for c in IDS]
TEMP = 'keycape-openbao-negative-key39'
BAO_ORIGINS = {'http://127.0.0.1:18200', 'http://openbao.openbao.svc:8200'}
REQUEST_DEADLINE = None
BUSINESS_MONOTONIC_DEADLINE = None
CLEANUP_MONOTONIC_DEADLINE = None
CLEANING = False
MAX_RESPONSE = 256 * 1024
HTTP_TOTAL_SECONDS = 15
CLEANUP_TOTAL_SECONDS = 300
HTTP_ABORT_SECONDS = 0.02
class Refused(Exception):
pass
def require(ok, reason):
if not ok:
raise Refused(reason)
def digest(data):
return hashlib.sha256(data).hexdigest()
def canonical(value):
return digest(json.dumps(value, sort_keys=True, separators=(',', ':')).encode())
def finite_number(value):
try:
return type(value) in (int, float) and math.isfinite(value)
except OverflowError:
return False
def read_json(raw):
def reject_constant(_):
raise Refused('nonfinite_json_constant')
return json.loads(raw, parse_constant=reject_constant)
def begin_cleanup():
global CLEANUP_MONOTONIC_DEADLINE
if CLEANUP_MONOTONIC_DEADLINE is None:
CLEANUP_MONOTONIC_DEADLINE = time.monotonic() + CLEANUP_TOTAL_SECONDS
def check_deadline():
if CLEANING:
require(CLEANUP_MONOTONIC_DEADLINE is not None
and time.monotonic() < CLEANUP_MONOTONIC_DEADLINE, 'cleanup_budget_expired')
return
if REQUEST_DEADLINE is not None and not CLEANING:
require(finite_number(REQUEST_DEADLINE), 'approved_window_invalid')
require(time.time() < REQUEST_DEADLINE, 'approved_window_expired')
if BUSINESS_MONOTONIC_DEADLINE is not None:
require(time.monotonic() < BUSINESS_MONOTONIC_DEADLINE, 'session_business_budget_expired')
def request_budget(cap, cleanup=False):
require(finite_number(cap) and 0 < cap <= 30, 'request_timeout_invalid')
budget = float(cap)
if cleanup or CLEANING:
if CLEANUP_MONOTONIC_DEADLINE is not None:
budget = min(budget, CLEANUP_MONOTONIC_DEADLINE - time.monotonic())
else:
check_deadline()
if REQUEST_DEADLINE is not None:
budget = min(budget, REQUEST_DEADLINE - time.time())
if BUSINESS_MONOTONIC_DEADLINE is not None:
budget = min(budget, BUSINESS_MONOTONIC_DEADLINE - time.monotonic())
require(budget > 0 and math.isfinite(budget), 'request_window_exhausted')
return budget
@contextmanager
def total_timeout(seconds):
"""Linux main-thread hard elapsed bound, including blocking headers/body reads."""
require(finite_number(seconds) and 0 < seconds <= 30, 'request_timeout_invalid')
require(threading.current_thread() is threading.main_thread()
and hasattr(signal, 'setitimer'), 'main_thread_linux_timer_required')
previous_timer = signal.getitimer(signal.ITIMER_REAL)
require(previous_timer == (0.0, 0.0), 'existing_alarm_timer_refused')
previous_handler = signal.getsignal(signal.SIGALRM)
def expired(*_):
raise Refused('request_total_timeout')
started = time.monotonic()
signal.signal(signal.SIGALRM, expired)
signal.setitimer(signal.ITIMER_REAL, seconds)
try:
yield
require(time.monotonic() - started <= seconds, 'request_total_timeout')
finally:
signal.setitimer(signal.ITIMER_REAL, 0)
signal.signal(signal.SIGALRM, previous_handler)
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *args, **kwargs):
return None
class SessionTransport:
"""Fixed origins, no ambient proxy, bounded reads, same guards in cleanup."""
def __init__(self, bao_origin, receipt=None):
require(bao_origin in BAO_ORIGINS, 'bao_origin_invalid')
self.bao_origin = bao_origin
self.receipt = receipt if receipt is not None else {}
self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())
def request(self, url, *, data=None, headers=None, method='POST', timeout=15):
allowed = {'https://kc.coulomb.social/token',
'https://kc.coulomb.social/.well-known/openid-configuration',
self.bao_origin + '/v1/auth/keycape-services/login',
self.bao_origin + '/v1/auth/token/revoke-self'}
require(url in allowed, 'http_target_not_reviewed')
require(method == ('GET' if url.endswith('/.well-known/openid-configuration') else 'POST'),
'http_method_not_reviewed')
cleanup = url.endswith('/v1/auth/token/revoke-self')
budget = request_budget(min(timeout, HTTP_TOTAL_SECONDS), cleanup=cleanup)
total_deadline = time.monotonic() + budget
req = urllib.request.Request(url, data=data, headers=headers or {}, method=method)
response, status, raw = None, None, None
try:
with total_timeout(budget):
try:
response = self.opener.open(req, timeout=budget)
except urllib.error.HTTPError as error:
response = error
except (urllib.error.URLError, OSError):
if url == self.bao_origin + '/v1/auth/keycape-services/login':
self.receipt['openbao_login_outcome_unknown'] = True
return None, b''
status = response.getcode()
raw = response.read(MAX_RESPONSE + 1)
require(len(raw) <= MAX_RESPONSE, 'http_response_oversized')
response.close()
response = None
if not cleanup:
check_deadline()
return status, raw
except Refused:
if url == self.bao_origin + '/v1/auth/keycape-services/login' and status in (None, 200):
# A late response cannot pass; preserve acquired-token cleanup when available.
if raw is not None and status == 200:
self.revoke_late_session(raw)
else:
self.receipt['openbao_login_outcome_unknown'] = True
raise
except (urllib.error.URLError, OSError):
if url == self.bao_origin + '/v1/auth/keycape-services/login':
if raw is not None and status == 200:
self.revoke_late_session(raw)
else:
self.receipt['openbao_login_outcome_unknown'] = True
return None, b''
except Exception:
if url == self.bao_origin + '/v1/auth/keycape-services/login':
if raw is not None and status == 200:
self.revoke_late_session(raw)
else:
self.receipt['openbao_login_outcome_unknown'] = True
raise Refused('http_response_failed') from None
finally:
if response is not None:
self.abort_response(response, total_deadline)
def abort_response(self, response, original_deadline):
"""One bounded exceptional close; never retry or reset the cleanup clock."""
self.receipt['http_abort_close_attempted'] = False
self.receipt['http_abort_close_succeeded'] = False
begin_cleanup()
try:
cap = request_budget(HTTP_ABORT_SECONDS, cleanup=True)
remaining = original_deadline - time.monotonic()
if remaining > 0:
cap = min(cap, remaining)
self.receipt['http_abort_close_attempted'] = True
with total_timeout(cap):
response.close()
self.receipt['http_abort_close_succeeded'] = True
except Exception:
self.receipt['http_response_abort_unconfirmed'] = True
raise Refused('http_response_abort_unconfirmed') from None
def revoke_late_session(self, raw):
self.receipt['late_acceptance_revoke_attempted'] = False
self.receipt['late_acceptance_revoke_succeeded'] = False
try:
token = read_json(raw)['auth']['client_token']
require(isinstance(token, str) and token, 'late_session_response_invalid')
except (KeyError, TypeError, ValueError, Refused):
self.receipt['openbao_login_outcome_unknown'] = True
return
begin_cleanup()
self.receipt['late_acceptance_revoke_attempted'] = True
try:
status, _ = self.post(self.bao_origin + '/v1/auth/token/revoke-self',
body={}, headers={'X-Vault-Token': token})
self.receipt['late_acceptance_revoke_succeeded'] = status == 204
except Exception:
self.receipt['late_acceptance_revoke_succeeded'] = False
finally:
del token
def post(self, url, form=None, body=None, headers=None, timeout=15):
headers = dict(headers or {})
if form is not None:
data = urllib.parse.urlencode(form).encode()
headers['Content-Type'] = 'application/x-www-form-urlencoded'
else:
data = json.dumps(body or {}).encode()
headers['Content-Type'] = 'application/json'
return self.request(url, data=data, headers=headers, timeout=timeout)
def discovery(self):
status, raw = self.request('https://kc.coulomb.social/.well-known/openid-configuration', method='GET')
require(status == 200, 'discovery_failed')
return json.loads(raw)
def k(arguments, payload=None):
# Execute on railiance01 in the founder's session, never workstation defaults.
check_deadline()
budget = request_budget(30)
with total_timeout(budget):
result = subprocess.run(['kubectl', '--kubeconfig=/etc/rancher/k3s/k3s.yaml',
'--request-timeout=20s', *arguments],
input=None if payload is None else json.dumps(payload),
text=True, capture_output=True, timeout=budget)
check_deadline()
require(result.returncode == 0, 'kubernetes_operation_failed')
return json.loads(result.stdout) if result.stdout.strip() else None
def get(kind, name):
return k(['-n', 'sso', 'get', kind, name, '-o', 'json'])
def meta(obj):
return {key: obj['metadata'][key] for key in ['uid', 'resourceVersion']}
def patch(kind, obj, path, value, dry=False):
ops = [{'op': 'test', 'path': '/metadata/' + key, 'value': val}
for key, val in meta(obj).items()]
ops.append({'op': 'replace', 'path': path, 'value': value})
args = ['-n', 'sso', 'patch', kind, obj['metadata']['name'], '--type=json',
'--patch-file=/dev/stdin', '-o', 'json']
if dry:
args += ['--dry-run=server']
return k(args, ops)
def registrations():
reference = {'displayName': 'Temporary KEY39 negative probe',
'allowedScopes': ['openbao:login'], 'grantTypes': ['client_credentials'],
'clientType': 'confidential', 'serviceSubject': 'service:secrets-engine',
'tenant': 'tenant:platform', 'roles': ['secrets-engine'], 'tokenLifetime': '1m'}
result = []
for index, client_id in enumerate(IDS):
c = dict(reference, clientId=client_id, secretRef='env:' + ENVS[index])
if index:
c['audience'] = 'secrets-engine-openbao'
if index == 1:
c['serviceSubject'] = 'service:openbao-probe'
if index == 2:
c['tenant'] = 'tenant:openbao-probe'
if index == 3:
c['roles'] = ['openbao-probe']
if index == 4:
c['allowedScopes'] = ['openbao:probe']
result.append(c)
return result
class UniqueLoader(yaml.SafeLoader):
pass
def unique(loader, node):
result = {}
for key_node, value_node in node.value:
key = loader.construct_object(key_node)
require(key not in result, 'duplicate_yaml_key')
result[key] = loader.construct_object(value_node)
return result
UniqueLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, unique)
def candidate_config(raw):
config = yaml.load(raw, Loader=UniqueLoader)
clients = config['clients']
require(isinstance(clients, list) and clients, 'clients_required')
ids = [c['clientId'] for c in clients]
require(len(ids) == len(set(ids)) and not set(ids).intersection(IDS), 'existing_probe_or_duplicate')
require(not {c.get('secretRef') for c in clients}.intersection('env:' + e for e in ENVS),
'probe_secretref_collision')
production = [c for c in clients if c['clientId'] == 'secrets-engine-openbao']
require(len(production) == 1, 'production_client_missing')
expected = registrations()[-1]
for field in ['allowedScopes', 'grantTypes', 'clientType', 'serviceSubject', 'tenant', 'roles']:
require(production[0].get(field) == expected[field], 'production_claim_contract_drift')
require(production[0].get('audience', production[0]['clientId']) == 'secrets-engine-openbao',
'production_audience_drift')
nodes = yaml.compose(raw)
node = next(v for key, v in nodes.value if key.value == 'clients')
require(isinstance(node, yaml.nodes.SequenceNode) and not node.flow_style,
'block_client_sequence_required')
index = node.end_mark.index
start = raw.rfind('\n', 0, index) + 1
if not raw[start:index].strip():
index = start
fragment = yaml.safe_dump(registrations(), sort_keys=False)
fragment = ''.join(' ' * node.start_mark.column + line if line.strip() else line
for line in fragment.splitlines(True))
if index and raw[index - 1] != '\n':
fragment = '\n' + fragment
updated = raw[:index] + fragment + raw[index:]
intended = copy.deepcopy(config)
intended['clients'].extend(registrations())
require(yaml.load(updated, Loader=UniqueLoader) == intended, 'unrelated_config_changed')
return updated
def candidate_spec(spec):
result = copy.deepcopy(spec)
require(result.get('replicas', 1) == 1 and result['strategy']['type'] == 'Recreate',
'single_replica_recreate_required')
container = next(c for c in result['template']['spec']['containers'] if c['name'] == 'keycape')
env = container['env']
require(not {e['name'] for e in env}.intersection(ENVS), 'probe_env_collision')
env.extend({'name': e, 'valueFrom': {'secretKeyRef': {'name': TEMP, 'key': c}}}
for c, e in zip(IDS, ENVS))
return result
def ready(spec, timeout=150):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
dep = get('deployment', 'keycape')
require(dep['spec'] == spec, 'deployment_drift_during_readiness')
status = dep.get('status', {})
if (status.get('observedGeneration') == dep['metadata']['generation']
and all(status.get(key) == 1 for key in
['updatedReplicas', 'readyReplicas', 'availableReplicas', 'replicas'])):
return True
time.sleep(3)
raise Refused('readiness_timeout')
def verify_sources(envelope, recovery=False):
require(envelope.get('proposal_id') == PROPOSAL and envelope.get('target') == TARGET,
'proposal_target_mismatch')
require(envelope.get('disposition') == ('recovery-approved' if recovery else 'approved')
and envelope.get('founder_decision_id'),
'founder_exact_go_required')
require(envelope.get('source_hashes') and
str(Path(__file__).resolve()) in envelope['source_hashes'], 'executor_pin_required')
for filename, expected in envelope['source_hashes'].items():
require(digest(Path(filename).read_bytes()) == expected, 'reviewed_source_changed')
required = ['net-kingdom', 'key-cape', 'railiance-platform']
require(all(envelope.get('owner_acknowledgments', {}).get(owner) for owner in required),
'owner_acknowledgments_required')
expiry = envelope['valid_until_epoch']
maximum = envelope['maximum_session_seconds']
require(finite_number(expiry) and type(maximum) is int and
420 <= maximum <= 900 and 0 < expiry - time.time() <= maximum,
'attended_window_invalid')
if recovery:
return None
role = envelope['role_observation']
require(role['mount'] == 'keycape-services' and role['role'] == 'secrets-engine-openbao'
and role['issuer'] == 'https://kc.coulomb.social' and role['algorithm'] == 'RS256'
and role['bindings_exact'] is True and role['policy_exact'] is True,
'role_owner_observation_required')
leeway = role['raw_expiration_leeway']
require(type(leeway) is int and -1 <= leeway <= 150, 'leeway_outside_session_bound')
effective = 0 if leeway == -1 else leeway or 150
require(role['effective_expiration_leeway'] == effective, 'leeway_normalization_mismatch')
observed = role['observed_at_epoch']
require(finite_number(observed) and 0 <= time.time() - observed <= 300,
'role_observation_stale_or_future')
require(envelope['bao_addr'] in BAO_ORIGINS, 'bao_origin_invalid')
return effective
def load_probe(path):
spec = importlib.util.spec_from_file_location('key39_reviewed_probe', path)
result = importlib.util.module_from_spec(spec)
spec.loader.exec_module(result)
return result
def removed_client_denied(status, raw):
try:
body = json.loads(raw)
return status == 401 and body.get('error') == 'invalid_profile_usage' and body.get('feature') == 'client_id'
except (ValueError, AttributeError, TypeError):
return False
def recovered_config(raw):
config = yaml.load(raw, Loader=UniqueLoader)
clients = config['clients']
ids = [c['clientId'] for c in clients]
present = set(ids).intersection(IDS)
require(len(ids) == len(set(ids)) and present in [set(), set(IDS)], 'partial_probe_registry')
if not present:
return raw
require(clients[-6:] == registrations(), 'recovery_registration_drift')
node = next(v for key, v in yaml.compose(raw).value if key.value == 'clients')
first = node.value[-6].start_mark.index
start = raw.rfind('\n', 0, first) + 1
end = node.end_mark.index
line_start = raw.rfind('\n', 0, end) + 1
if not raw[line_start:end].strip():
end = line_start
restored = raw[:start] + raw[end:]
require(candidate_config(restored) == raw, 'recovery_byte_roundtrip_failed')
return restored
def recovered_spec(spec):
restored = copy.deepcopy(spec)
container = next(c for c in restored['template']['spec']['containers'] if c['name'] == 'keycape')
env = container['env']
expected = [{'name': e, 'valueFrom': {'secretKeyRef': {'name': TEMP, 'key': c}}}
for c, e in zip(IDS, ENVS)]
matches = [e for e in env if e['name'] in ENVS]
require(matches in [[], expected], 'partial_or_drifted_probe_env')
container['env'] = [e for e in env if e['name'] not in ENVS]
return restored
def delete_temporary(created):
# The exact raw API delete needs the same aggregate/per-request guards as patches.
k(['delete', '--raw=/api/v1/namespaces/sso/secrets/' + TEMP, '-f', '/dev/stdin'],
{'apiVersion': 'v1', 'kind': 'DeleteOptions', 'preconditions': meta(created)})
require(k(['-n', 'sso', 'get', 'secret', TEMP, '--ignore-not-found', '-o', 'json']) is None,
'temporary_secret_still_present')
def recover(envelope, receipt):
global REQUEST_DEADLINE, BUSINESS_MONOTONIC_DEADLINE, CLEANUP_MONOTONIC_DEADLINE, CLEANING
verify_sources(envelope, recovery=True)
# Separately approved recovery obligations are cleanup, never business writes.
REQUEST_DEADLINE = None
BUSINESS_MONOTONIC_DEADLINE = None
CLEANUP_MONOTONIC_DEADLINE = None
begin_cleanup()
CLEANING = True
transport = SessionTransport(envelope['bao_addr'], receipt=receipt)
require(k(['get', 'namespace', 'kube-system', '-o', 'json'])['metadata']['uid'] == CLUSTER,
'cluster_identity_changed')
config = get('secret', 'keycape-config')
dep = get('deployment', 'keycape')
raw = base64.b64decode(config['data']['config.yaml'], validate=True).decode()
restored = recovered_config(raw)
spec = recovered_spec(dep['spec'])
require(config['metadata']['uid'] == envelope['config_metadata']['uid']
and dep['metadata']['uid'] == envelope['deployment_metadata']['uid']
and digest(restored.encode()) == envelope['registry_sha256']
and canonical(spec) == envelope['deployment_spec_sha256'], 'recovery_baseline_drift')
temp = get('secret', TEMP)
require(set(temp.get('data', {})) == set(IDS)
and temp['metadata']['uid'] == envelope['temporary_secret_uid'], 'recovery_custody_identity_required')
values = {c: base64.b64decode(temp['data'][c], validate=True).decode() for c in IDS}
if raw != restored:
patch('secret', config, '/data/config.yaml', base64.b64encode(restored.encode()).decode())
if dep['spec'] != spec:
patch('deployment', dep, '/spec', spec)
ready(spec)
after = get('secret', 'keycape-config')
require(after['data'] == dict(config['data'], **{'config.yaml': base64.b64encode(restored.encode()).decode()}),
'recovery_readback_failed')
require(envelope['runner_path'] in envelope['source_hashes'], 'runner_pin_required')
probe = load_probe(envelope['runner_path'])
for client, value in values.items():
basic = base64.b64encode((client + ':' + value).encode()).decode()
status, body = transport.post('https://kc.coulomb.social/token',
form={'grant_type': 'client_credentials', 'scope': 'openbao:login'},
headers={'Authorization': 'Basic ' + basic})
require(removed_client_denied(status, body), 'removed_client_denial_unconfirmed')
observed = get('secret', TEMP)
require(observed['data'] == temp['data'] and observed['metadata']['uid'] == temp['metadata']['uid'],
'recovery_temporary_secret_drift')
delete_temporary(observed)
receipt.update(cleanup_verified=True, recovery_only=True)
def execute(envelope, private, receipt, checkpoint=lambda: None):
global REQUEST_DEADLINE, BUSINESS_MONOTONIC_DEADLINE, CLEANUP_MONOTONIC_DEADLINE, CLEANING
leeway = verify_sources(envelope)
REQUEST_DEADLINE = envelope['valid_until_epoch']
BUSINESS_MONOTONIC_DEADLINE = time.monotonic() + min(
REQUEST_DEADLINE - time.time(), envelope['maximum_session_seconds'] - 180)
CLEANUP_MONOTONIC_DEADLINE = None
CLEANING = False
runner_path = envelope['runner_path']
require(runner_path in envelope['source_hashes'], 'runner_pin_required')
probe = load_probe(runner_path)
transport = SessionTransport(envelope['bao_addr'], receipt=receipt)
discovery = transport.discovery()
require(discovery.get('issuer') == 'https://kc.coulomb.social' and
discovery.get('token_endpoint') == 'https://kc.coulomb.social/token',
'issuer_discovery_changed')
require(isinstance(private.get('human_token'), str) and private['human_token']
and isinstance(private.get('coding_agent_token'), str) and private['coding_agent_token'],
'founder_private_tokens_required')
require(k(['get', 'namespace', 'kube-system', '-o', 'json'])['metadata']['uid'] == CLUSTER,
'cluster_identity_changed')
before = get('secret', 'keycape-config')
deployment = get('deployment', 'keycape')
require(not before['metadata'].get('ownerReferences'), 'controller_owned_config')
require(meta(before) == envelope['config_metadata']
and meta(deployment) == envelope['deployment_metadata']
and canonical(deployment['spec']) == envelope['deployment_spec_sha256'],
'approved_runtime_revision_changed')
raw = base64.b64decode(before['data']['config.yaml'], validate=True).decode()
require(digest(raw.encode()) == envelope['registry_sha256'], 'approved_registry_changed')
updated = base64.b64encode(candidate_config(raw).encode()).decode()
intended_data = dict(before['data'], **{'config.yaml': updated})
candidate = candidate_spec(deployment['spec'])
ready(deployment['spec'])
# Exact GET existence is tested via create (never adopt/delete an existing Secret).
values = {c: secrets.token_urlsafe(48) for c in IDS}
temporary = {'apiVersion': 'v1', 'kind': 'Secret', 'metadata': {'namespace': 'sso', 'name': TEMP},
'type': 'Opaque', 'data': {c: base64.b64encode(v.encode()).decode() for c, v in values.items()}}
patch('secret', before, '/data/config.yaml', updated, dry=True)
patch('deployment', deployment, '/spec', candidate, dry=True)
k(['create', '-f', '/dev/stdin', '--dry-run=server', '-o', 'json'], temporary)
created = None
mutations_attempted = False
began = time.monotonic()
try:
# Set before API call: uncertain response must still trigger reconciliation.
check_deadline()
mutations_attempted = True
created = k(['create', '-f', '/dev/stdin', '-o', 'json'], temporary)
receipt['temporary_secret_uid'] = created['metadata']['uid']
checkpoint()
patch('secret', before, '/data/config.yaml', updated)
patch('deployment', deployment, '/spec', candidate)
ready(candidate)
require(get('secret', 'keycape-config')['data'] == intended_data, 'registry_drift')
require(time.monotonic() - began < 300, 'probe_start_too_late')
# Current admitted agent token stays founder-owned; no existing secret fetched.
original_exchange = probe.exchange
def exchange(post, endpoint, client, secret, scope):
if client == 'codex-railiance-platform':
return private['coding_agent_token']
return original_exchange(post, endpoint, client, secret, scope)
probe.exchange = exchange
supplied = dict(values, **{'codex-railiance-platform': 'founder-held-token'})
def bounded_post(*args, **kwargs):
if not args[0].endswith('/v1/auth/token/revoke-self'):
require(time.monotonic() - began < envelope['maximum_session_seconds'] - 180,
'session_cleanup_reserve_reached')
return transport.post(*args, **kwargs)
def bounded_sleep(seconds):
while seconds > 0:
check_deadline()
require(time.monotonic() - began < envelope['maximum_session_seconds'] - 180,
'session_cleanup_reserve_reached')
part = min(seconds, 3)
time.sleep(part)
seconds -= part
result = probe.run(bounded_post, supplied, 'https://kc.coulomb.social/token',
envelope['bao_addr'], 'keycape-services', 'secrets-engine-openbao',
human_token=private['human_token'], leeway=leeway, sleep=bounded_sleep)
receipt['negative_probe'] = result
checkpoint()
check_deadline()
require(time.monotonic() - began <= envelope['maximum_session_seconds'], 'session_window_exceeded')
require(result['all_refused'], 'negative_proof_incomplete')
finally:
if mutations_attempted:
begin_cleanup()
CLEANING = True
# Never restore over an unrelated edit; keep custody for owner recovery on drift.
now = get('secret', 'keycape-config')
require(now['metadata']['uid'] == before['metadata']['uid'] and
now['data'] in [before['data'], intended_data], 'cleanup_registry_drift')
if now['data'] == intended_data:
patch('secret', now, '/data/config.yaml', before['data']['config.yaml'])
now = get('deployment', 'keycape')
require(now['metadata']['uid'] == deployment['metadata']['uid'] and
now['spec'] in [deployment['spec'], candidate], 'cleanup_deployment_drift')
if now['spec'] == candidate:
patch('deployment', now, '/spec', deployment['spec'])
ready(deployment['spec'])
require(get('secret', 'keycape-config')['data'] == before['data'], 'cleanup_readback_failed')
# No fresh JWT is minted: deleted clients must fail at the issuer.
for client in IDS:
basic = base64.b64encode((client + ':' + values[client]).encode()).decode()
status, body = transport.post('https://kc.coulomb.social/token',
form={'grant_type': 'client_credentials', 'scope': 'openbao:login'},
headers={'Authorization': 'Basic ' + basic})
require(removed_client_denied(status, body),
'removed_client_denial_unconfirmed')
if created is None:
# A timed-out create may have succeeded; compare generated bytes before delete.
observed = get('secret', TEMP)
require(observed.get('data') == temporary['data'], 'temporary_secret_identity_unconfirmed')
created = observed
observed = get('secret', TEMP)
require(observed['metadata']['uid'] == created['metadata']['uid']
and observed.get('data') == temporary['data'], 'temporary_secret_drift')
delete_temporary(observed)
receipt['cleanup_verified'] = True
def main():
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument('mode', choices=['inspect', 'founder-preflight', 'execute', 'recover'])
ap.add_argument('--envelope')
ap.add_argument('--confirm')
ap.add_argument('--receipt', required=True)
args = ap.parse_args()
receipt = {'proposal_id': PROPOSAL, 'target': TARGET, 'native_accepted': False,
'values_emitted': False, 'cleanup_verified': False}
# Do not truncate or follow a link; protected receipts cannot overwrite credentials.
fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
def checkpoint():
os.lseek(fd, 0, os.SEEK_SET)
os.ftruncate(fd, 0)
os.write(fd, json.dumps(receipt, sort_keys=True, indent=2).encode())
os.fsync(fd)
rc = 1
try:
if args.mode in ['inspect', 'founder-preflight']:
require(k(['get', 'namespace', 'kube-system', '-o', 'json'])['metadata']['uid'] == CLUSTER,
'cluster_identity_changed')
dep = get('deployment', 'keycape')
receipt.update(deployment_metadata=meta(dep), deployment_spec_sha256=canonical(dep['spec']),
image=next(c['image'] for c in dep['spec']['template']['spec']['containers']
if c['name'] == 'keycape'), mode='metadata_only')
if args.mode == 'founder-preflight':
require(args.confirm == 'FOUNDER ATTENDED KEY39 PREFLIGHT', 'founder_attendance_required')
config = get('secret', 'keycape-config')
require(not config['metadata'].get('ownerReferences'), 'controller_owned_config')
raw = base64.b64decode(config['data']['config.yaml'], validate=True).decode()
candidate_config(raw)
candidate_spec(dep['spec'])
require(k(['-n', 'sso', 'get', 'secret', TEMP, '--ignore-not-found', '-o', 'json']) is None,
'temporary_secret_already_exists')
ready(dep['spec'])
receipt.update(config_metadata=meta(config), registry_sha256=digest(raw.encode()),
baseline_ready=True, mode='founder_protected_preflight',
probe_ids_absent=True, probe_env_absent=True, temporary_secret_absent=True)
else:
require(args.envelope is not None, 'approval_envelope_required')
envelope = read_json(Path(args.envelope).read_text())
require(Path(args.envelope).stat().st_mode & 0o022 == 0, 'approval_envelope_writable')
signal.signal(signal.SIGTERM, lambda *_: (_ for _ in ()).throw(Refused('session_interrupted')))
if args.mode == 'recover':
recover(envelope, receipt)
else:
private = read_json(__import__('sys').stdin.read())
execute(envelope, private, receipt, checkpoint=checkpoint)
receipt['negative_and_cleanup_verified'] = True
rc = 0
except Refused as error:
receipt.update(status='incomplete_requires_owner_review', reason=str(error))
except BaseException:
# Fixed outcome only: exception text may contain protected Kubernetes/HTTP data.
receipt['status'] = 'incomplete_requires_owner_review'
finally:
checkpoint()
os.close(fd)
return rc
if __name__ == '__main__':
raise SystemExit(main())

603
sso-mfa/k8s/keycape/test_openbao_negative_session.py Normal file
View file

@ -0,0 +1,603 @@
import base64
import copy
import importlib.util
import json
from pathlib import Path
import time
import io
import os
import urllib.error
import signal
import threading
import unittest
from unittest.mock import patch
spec = importlib.util.spec_from_file_location('session', Path(__file__).with_name('openbao-negative-session.py'))
s = importlib.util.module_from_spec(spec)
spec.loader.exec_module(s)
def baseline():
reference = dict(s.registrations()[-1], clientId='secrets-engine-openbao',
secretRef='env:PRODUCTION', tokenLifetime='15m')
return '# preserve this comment\nclients:\n' + s.yaml.safe_dump([reference], sort_keys=False) + 'other: unchanged\n'
def envelope():
return {'proposal_id': s.PROPOSAL, 'target': s.TARGET, 'disposition': 'approved',
'founder_decision_id': 'fixture-decision',
'source_hashes': {str(Path(s.__file__).resolve()): s.digest(Path(s.__file__).read_bytes())},
'owner_acknowledgments': {o: 'fixture-owner-review' for o in
['net-kingdom', 'key-cape', 'railiance-platform']},
'valid_until_epoch': time.time() + 900, 'maximum_session_seconds': 900,
'bao_addr': 'http://openbao.openbao.svc:8200',
'role_observation': {'mount': 'keycape-services', 'role': 'secrets-engine-openbao',
'issuer': 'https://kc.coulomb.social', 'algorithm': 'RS256',
'bindings_exact': True, 'policy_exact': True, 'raw_expiration_leeway': 0,
'effective_expiration_leeway': 150, 'observed_at_epoch': time.time()}}
class Tests(unittest.TestCase):
def setUp(self):
s.REQUEST_DEADLINE = None
s.BUSINESS_MONOTONIC_DEADLINE = None
s.CLEANUP_MONOTONIC_DEADLINE = None
s.CLEANING = False
def test_preserve_original_bytes(self):
raw = baseline()
result = s.candidate_config(raw)
start = result.index('clientId: openbao-probe-aud')
self.assertIn('# preserve this comment', result[:start])
parsed = s.yaml.safe_load(result)
self.assertEqual(parsed['clients'][:-6], s.yaml.safe_load(raw)['clients'])
self.assertEqual(parsed['other'], 'unchanged')
def test_isolation_and_sixth_matching_expiry(self):
clients = s.registrations()
self.assertNotIn('audience', clients[0])
for c in clients[1:]:
self.assertEqual(c['audience'], 'secrets-engine-openbao')
for key in ['roles', 'tenant', 'serviceSubject', 'allowedScopes']:
self.assertEqual(clients[-1][key], s.yaml.safe_load(baseline())['clients'][0][key])
def test_preexisting_probe_refused(self):
raw = baseline().replace('other: unchanged', '- clientId: openbao-probe-aud\nother: unchanged')
with self.assertRaises(s.Refused):
s.candidate_config(raw)
def test_duplicate_yaml_refused(self):
with self.assertRaises(s.Refused):
s.candidate_config(baseline() + 'other: shadow\n')
def test_production_drift_refused(self):
with self.assertRaises(s.Refused):
s.candidate_config(baseline().replace('tenant:platform', 'tenant:other'))
def test_env_collision_refused(self):
spec = {'replicas': 1, 'strategy': {'type': 'Recreate'}, 'template': {'spec': {'containers': [
{'name': 'keycape', 'env': [{'name': s.ENVS[0], 'value': 'fixture'}]}]}}}
with self.assertRaises(s.Refused):
s.candidate_spec(spec)
def test_recovery_roundtrip_preserves_every_baseline_byte(self):
raw = baseline()
self.assertEqual(s.recovered_config(s.candidate_config(raw)), raw)
self.assertEqual(s.recovered_config(raw), raw)
def test_recovery_claim_drift_and_partial_registration_refused(self):
raw = s.candidate_config(baseline())
with self.assertRaises(s.Refused):
s.recovered_config(raw.replace('service:openbao-probe', 'service:changed'))
with self.assertRaises(s.Refused):
s.recovered_config(raw.replace('clientId: openbao-probe-aud', 'clientId: different-probe'))
def test_recovery_env_roundtrip(self):
spec = {'replicas': 1, 'strategy': {'type': 'Recreate'}, 'template': {'spec': {'containers': [
{'name': 'keycape', 'env': [{'name': 'ORIGINAL', 'value': 'fixture'}]}]}}}
self.assertEqual(s.recovered_spec(s.candidate_spec(spec)), spec)
def test_approval_is_not_optional(self):
e = envelope()
del e['founder_decision_id']
with self.assertRaises(s.Refused):
s.verify_sources(e)
def test_source_change_refused(self):
e = envelope()
e['source_hashes'][str(Path(s.__file__).resolve())] = '0' * 64
with self.assertRaises(s.Refused):
s.verify_sources(e)
def test_owner_missing_refused(self):
e = envelope()
del e['owner_acknowledgments']['net-kingdom']
with self.assertRaises(s.Refused):
s.verify_sources(e)
def test_leeway_default_and_disabled(self):
e = envelope()
self.assertEqual(s.verify_sources(e), 150)
e['role_observation'].update(raw_expiration_leeway=-1, effective_expiration_leeway=0)
self.assertEqual(s.verify_sources(e), 0)
def test_stale_role_refused(self):
e = envelope()
e['role_observation']['observed_at_epoch'] -= 301
with self.assertRaises(s.Refused):
s.verify_sources(e)
def test_wrong_effective_leeway_refused(self):
e = envelope()
e['role_observation']['effective_expiration_leeway'] = 0
with self.assertRaises(s.Refused):
s.verify_sources(e)
def test_bao_credentials_in_origin_refused(self):
e = envelope()
e['bao_addr'] = 'https://user:password@example.invalid'
with self.assertRaises(s.Refused):
s.verify_sources(e)
def test_future_role_observation_refused(self):
e = envelope()
e['role_observation']['observed_at_epoch'] = time.time() + 60
with self.assertRaises(s.Refused):
s.verify_sources(e)
def test_arbitrary_https_origin_refused(self):
e = envelope()
e['bao_addr'] = 'https://attacker.invalid'
with self.assertRaises(s.Refused):
s.verify_sources(e)
def test_hostile_proxy_is_disabled(self):
build = s.urllib.request.build_opener
with patch.dict(os.environ, {'HTTPS_PROXY':'http://hostile-proxy.invalid:8000',
'HTTP_PROXY':'http://hostile-proxy.invalid:8000'}), \
patch.object(s.urllib.request, 'build_opener', wraps=build) as observed:
transport = s.SessionTransport('http://openbao.openbao.svc:8200')
handlers = observed.call_args.args
self.assertIsInstance(handlers[0], s.urllib.request.ProxyHandler)
self.assertEqual(handlers[0].proxies, {})
self.assertIsInstance(handlers[1], s.NoRedirect)
self.assertIsNone(handlers[1].redirect_request(None, None, 302, '', {}, 'https://hostile.invalid'))
def test_redirect_never_follows_or_counts_as_refusal(self):
transport = s.SessionTransport('http://openbao.openbao.svc:8200')
response = urllib.error.HTTPError('https://kc.coulomb.social/token', 302,
'redirect', {}, io.BytesIO(b'{}'))
with patch.object(transport.opener, 'open', side_effect=response) as observed:
status, raw = transport.post('https://kc.coulomb.social/token', form={})
self.assertEqual(status, 302)
self.assertEqual(observed.call_count, 1)
def test_oversized_http_body_is_closed_and_refused(self):
transport = s.SessionTransport('http://openbao.openbao.svc:8200')
response = type('Response', (), {'getcode':lambda self: 200,
'read':lambda self, size: b'x' * size, 'close':lambda self: setattr(self, 'closed', True)})()
with patch.object(transport.opener, 'open', return_value=response):
with self.assertRaisesRegex(s.Refused, 'http_response_oversized'):
transport.post('https://kc.coulomb.social/token', form={})
self.assertTrue(response.closed)
def test_expired_window_prevents_protected_kubernetes_and_http(self):
s.REQUEST_DEADLINE = time.time() - 1
with patch.object(s.subprocess, 'run') as backend:
with self.assertRaisesRegex(s.Refused, 'approved_window_expired'):
s.k(['create', '-f', '/dev/stdin'], {})
backend.assert_not_called()
transport = s.SessionTransport('http://openbao.openbao.svc:8200')
with patch.object(transport.opener, 'open') as backend:
with self.assertRaisesRegex(s.Refused, 'approved_window_expired'):
transport.post('https://kc.coulomb.social/token', form={})
backend.assert_not_called()
def test_cleanup_revoke_allowed_after_window_expiry(self):
s.REQUEST_DEADLINE = time.time() - 1
transport = s.SessionTransport('http://openbao.openbao.svc:8200')
response = type('Response', (), {'getcode':lambda self: 204,
'read':lambda self, size: b'', 'close':lambda self: None})()
with patch.object(transport.opener, 'open', return_value=response):
self.assertEqual(transport.post('http://openbao.openbao.svc:8200/v1/auth/token/revoke-self'), (204, b''))
def test_http_read_has_finite_size(self):
transport = s.SessionTransport('http://openbao.openbao.svc:8200')
sizes = []
response = type('Response', (), {'getcode':lambda self: 200,
'read':lambda self, size: sizes.append(size) or b'{}', 'close':lambda self: None})()
with patch.object(transport.opener, 'open', return_value=response):
transport.post('https://kc.coulomb.social/token')
self.assertEqual(sizes, [s.MAX_RESPONSE + 1])
def test_nonfinite_and_boolean_timestamps_refused(self):
for value in [float('inf'), float('-inf'), float('nan'), True, False, 'Infinity', 10 ** 1000]:
for field in ['valid_until_epoch', 'observed_at_epoch']:
with self.subTest(field=field, value_type=type(value).__name__):
e = envelope()
if field == 'valid_until_epoch':
e[field] = value
else:
e['role_observation'][field] = value
with self.assertRaises(s.Refused):
s.verify_sources(e)
def test_finite_far_future_and_noninteger_window_refused(self):
e = envelope()
e['valid_until_epoch'] = time.time() + 86400
with self.assertRaises(s.Refused):
s.verify_sources(e)
for value in [True, 900.0, float('inf'), float('nan'), '900']:
e = envelope()
e['maximum_session_seconds'] = value
with self.assertRaises(s.Refused):
s.verify_sources(e)
def test_recovery_also_requires_finite_typed_window(self):
e = envelope()
e['disposition'] = 'recovery-approved'
e['valid_until_epoch'] = float('inf')
with self.assertRaises(s.Refused):
s.verify_sources(e, recovery=True)
def test_nonfinite_json_constants_refused(self):
for constant in ['Infinity', '-Infinity', 'NaN']:
with self.assertRaises(s.Refused):
s.read_json('{"valid_until_epoch":' + constant + '}')
self.assertFalse(s.finite_number(s.read_json('{"valid_until_epoch":1e999}')['valid_until_epoch']))
def test_hard_timer_covers_slow_trickle_body_and_closes_response(self):
transport = s.SessionTransport('http://openbao.openbao.svc:8200')
class Trickle:
def getcode(self):
return 200
def read(self, size):
for _ in range(100):
time.sleep(0.01)
return b'{}'
def close(self):
self.closed = True
response = Trickle()
previous = signal.getsignal(signal.SIGALRM)
started = time.monotonic()
with patch.object(transport.opener, 'open', return_value=response):
with self.assertRaisesRegex(s.Refused, 'request_total_timeout'):
transport.post('https://kc.coulomb.social/token', timeout=0.04)
self.assertLess(time.monotonic() - started, 0.25)
self.assertTrue(response.closed)
self.assertIs(signal.getsignal(signal.SIGALRM), previous)
self.assertEqual(signal.getitimer(signal.ITIMER_REAL), (0.0, 0.0))
def test_hard_timer_spans_headers_and_clips_to_business_window(self):
transport = s.SessionTransport('http://openbao.openbao.svc:8200')
s.REQUEST_DEADLINE = time.time() + 60
s.BUSINESS_MONOTONIC_DEADLINE = time.monotonic() + 0.04
def slow_headers(*args, **kwargs):
time.sleep(1)
raise AssertionError('inactivity timeout would have permitted this')
started = time.monotonic()
with patch.object(transport.opener, 'open', side_effect=slow_headers):
with self.assertRaisesRegex(s.Refused, 'request_total_timeout'):
transport.post('https://kc.coulomb.social/token')
self.assertLess(time.monotonic() - started, 0.25)
def test_late_full_acceptance_is_revoked_then_refused(self):
receipt = {}
transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
token = 'not-a-real-late-token'
class Late:
def getcode(self):
return 200
def read(self, size):
s.REQUEST_DEADLINE = time.time() - 1
return json.dumps({'auth': {'client_token': token}}).encode()
def close(self):
pass
revoke = type('Response', (), {'getcode':lambda self: 204,
'read':lambda self, size: b'', 'close':lambda self: None})()
with patch.object(transport.opener, 'open', side_effect=[Late(), revoke]) as calls:
with self.assertRaisesRegex(s.Refused, 'approved_window_expired'):
transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login')
self.assertEqual(calls.call_count, 2)
self.assertEqual(calls.call_args_list[1].args[0].full_url,
'http://openbao.openbao.svc:8200/v1/auth/token/revoke-self')
self.assertTrue(receipt['late_acceptance_revoke_attempted'])
self.assertTrue(receipt['late_acceptance_revoke_succeeded'])
self.assertNotIn(token, json.dumps(receipt))
def test_late_acceptance_failed_revoke_remains_unconfirmed(self):
receipt = {}
transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
class Late:
def getcode(self):
return 200
def read(self, size):
s.REQUEST_DEADLINE = time.time() - 1
return b'{"auth":{"client_token":"fake-token"}}'
def close(self):
pass
failed = type('Response', (), {'getcode':lambda self: 403,
'read':lambda self, size: b'{}', 'close':lambda self: None})()
with patch.object(transport.opener, 'open', side_effect=[Late(), failed]):
with self.assertRaises(s.Refused):
transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login')
self.assertTrue(receipt['late_acceptance_revoke_attempted'])
self.assertFalse(receipt['late_acceptance_revoke_succeeded'])
def test_partial_accepted_response_timeout_retains_unknown_outcome(self):
receipt = {}
transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
response = type('Response', (), {'getcode':lambda self: 200,
'read':lambda self, size: time.sleep(1), 'close':lambda self: None})()
with patch.object(transport.opener, 'open', return_value=response):
with self.assertRaisesRegex(s.Refused, 'request_total_timeout'):
transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login', timeout=0.04)
self.assertTrue(receipt['openbao_login_outcome_unknown'])
def test_cleanup_has_finite_aggregate_deadline_and_restoration(self):
s.REQUEST_DEADLINE = time.time() - 1
s.begin_cleanup()
first = s.CLEANUP_MONOTONIC_DEADLINE
s.begin_cleanup()
self.assertEqual(s.CLEANUP_MONOTONIC_DEADLINE, first)
s.CLEANUP_MONOTONIC_DEADLINE = time.monotonic() - 1
s.CLEANING = True
with self.assertRaisesRegex(s.Refused, 'cleanup_budget_expired'):
s.check_deadline()
transport = s.SessionTransport('http://openbao.openbao.svc:8200')
with patch.object(transport.opener, 'open') as backend:
with self.assertRaises(s.Refused):
transport.post('http://openbao.openbao.svc:8200/v1/auth/token/revoke-self')
backend.assert_not_called()
def test_expired_cleanup_cannot_issue_native_secret_delete(self):
s.CLEANING = True
s.CLEANUP_MONOTONIC_DEADLINE = time.monotonic() - 1
obj = {'metadata': {'uid':'temporary', 'resourceVersion':'7'}}
with patch.object(s.subprocess, 'run') as native:
with self.assertRaisesRegex(s.Refused, 'cleanup_budget_expired'):
s.delete_temporary(obj)
native.assert_not_called()
def test_secret_delete_total_timeout_clipped_to_cleanup_budget(self):
s.CLEANING = True
s.CLEANUP_MONOTONIC_DEADLINE = time.monotonic() + 0.04
obj = {'metadata': {'uid':'temporary', 'resourceVersion':'7'}}
captured = {}
def slow_delete(args, **kwargs):
captured.update(kwargs)
time.sleep(1)
return type('Result', (), {'returncode':0, 'stdout':'{}'})()
started = time.monotonic()
with patch.object(s.subprocess, 'run', side_effect=slow_delete):
with self.assertRaisesRegex(s.Refused, 'request_total_timeout'):
s.delete_temporary(obj)
self.assertLess(time.monotonic() - started, 0.25)
self.assertLessEqual(captured['timeout'], 0.04)
self.assertEqual(json.loads(captured['input'])['preconditions'], {'uid':'temporary', 'resourceVersion':'7'})
def test_early_read_failure_slow_close_is_bounded_once(self):
receipt = {}
transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
class Response:
closes = 0
def getcode(self):
return 200
def read(self, size):
raise s.Refused('synthetic_read_failure')
def close(self):
self.closes += 1
time.sleep(0.07)
response = Response()
previous = signal.getsignal(signal.SIGALRM)
started = time.monotonic()
with patch.object(transport.opener, 'open', return_value=response):
with self.assertRaisesRegex(s.Refused, 'http_response_abort_unconfirmed'):
transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login', timeout=0.01)
self.assertLess(time.monotonic() - started, 0.05)
self.assertEqual(response.closes, 1)
self.assertTrue(receipt['openbao_login_outcome_unknown'])
self.assertTrue(receipt['http_abort_close_attempted'])
self.assertFalse(receipt['http_abort_close_succeeded'])
self.assertTrue(receipt['http_response_abort_unconfirmed'])
self.assertIs(signal.getsignal(signal.SIGALRM), previous)
self.assertEqual(signal.getitimer(signal.ITIMER_REAL), (0.0, 0.0))
def test_total_read_timeout_then_slow_close_uses_only_abort_cap(self):
receipt = {}
transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
class Response:
closes = 0
def getcode(self):
return 200
def read(self, size):
time.sleep(0.1)
def close(self):
self.closes += 1
time.sleep(0.1)
response = Response()
started = time.monotonic()
with patch.object(transport.opener, 'open', return_value=response):
with self.assertRaisesRegex(s.Refused, 'http_response_abort_unconfirmed'):
transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login', timeout=0.01)
self.assertLess(time.monotonic() - started, 0.06)
self.assertEqual(response.closes, 1)
self.assertTrue(receipt['openbao_login_outcome_unknown'])
self.assertFalse(receipt['http_abort_close_succeeded'])
def test_oversized_body_slow_abort_remains_unproved_and_bounded(self):
receipt = {}
transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
response = type('Response', (), {'getcode':lambda self: 200,
'read':lambda self, size: b'x' * size,
'close':lambda self: time.sleep(0.1)})()
started = time.monotonic()
with patch.object(transport.opener, 'open', return_value=response):
with self.assertRaisesRegex(s.Refused, 'http_response_abort_unconfirmed'):
transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login', timeout=0.01)
self.assertLess(time.monotonic() - started, 0.05)
self.assertTrue(receipt['openbao_login_outcome_unknown'])
self.assertFalse(receipt['http_abort_close_succeeded'])
def test_raising_normal_close_then_slow_abort_is_private_bounded_unknown(self):
receipt = {}
transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
class Response:
closes = 0
def getcode(self):
return 200
def read(self, size):
return b'{}'
def close(self):
self.closes += 1
if self.closes == 1:
raise RuntimeError('private protected error must not escape')
time.sleep(0.1)
response = Response()
previous = signal.getsignal(signal.SIGALRM)
started = time.monotonic()
with patch.object(transport.opener, 'open', return_value=response):
with self.assertRaisesRegex(s.Refused, '^http_response_abort_unconfirmed$'):
transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login', timeout=0.01)
self.assertLess(time.monotonic() - started, 0.05)
self.assertEqual(response.closes, 2)
self.assertTrue(receipt['openbao_login_outcome_unknown'])
self.assertFalse(receipt['http_abort_close_succeeded'])
self.assertIs(signal.getsignal(signal.SIGALRM), previous)
self.assertEqual(signal.getitimer(signal.ITIMER_REAL), (0.0, 0.0))
def test_expired_cleanup_never_attempts_exceptional_close_or_resets(self):
receipt = {}
transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
deadline = time.monotonic() - 1
s.CLEANUP_MONOTONIC_DEADLINE = deadline
response = type('Response', (), {'close':lambda self: self.fail('must not close')})()
with patch.object(response, 'close') as closes:
with self.assertRaisesRegex(s.Refused, 'http_response_abort_unconfirmed'):
transport.abort_response(response, time.monotonic() - 1)
closes.assert_not_called()
self.assertEqual(s.CLEANUP_MONOTONIC_DEADLINE, deadline)
self.assertFalse(receipt['http_abort_close_attempted'])
self.assertTrue(receipt['http_response_abort_unconfirmed'])
def test_thread_or_existing_timer_refused_without_overwrite(self):
results = []
def in_thread():
try:
with s.total_timeout(0.1):
results.append('unreachable')
except s.Refused:
results.append('refused')
thread = threading.Thread(target=in_thread)
thread.start()
thread.join()
self.assertEqual(results, ['refused'])
signal.setitimer(signal.ITIMER_REAL, 1)
try:
with self.assertRaisesRegex(s.Refused, 'existing_alarm_timer_refused'):
with s.total_timeout(0.1):
pass
self.assertGreater(signal.getitimer(signal.ITIMER_REAL)[0], 0.5)
finally:
signal.setitimer(signal.ITIMER_REAL, 0)
def test_cleanup_refuses_concurrent_registry_change(self):
self.lifecycle(drift=True)
def test_deleted_client_contract_is_specific(self):
self.assertTrue(s.removed_client_denied(401, b'{"error":"invalid_profile_usage","feature":"client_id"}'))
for status, body in [(503, b'{}'), (401, b'{"error":"invalid_profile_usage","feature":"Authorization"}'),
(401, b'{"error":"invalid_client"}'), (401, b'not-json')]:
self.assertFalse(s.removed_client_denied(status, body))
def test_cleanup_even_when_probe_fails(self):
self.lifecycle(probe_pass=False)
def test_complete_lifecycle(self):
self.lifecycle()
def lifecycle(self, drift=False, probe_pass=True):
raw = baseline()
dep_spec = {'replicas': 1, 'strategy': {'type': 'Recreate'}, 'template': {'spec': {'containers': [
{'name': 'keycape', 'image': 'fixture@sha256:abc', 'env': []}]}}}
objects = {'keycape-config': {'metadata': {'name': 'keycape-config', 'uid': 'cfg', 'resourceVersion': '1'},
'data': {'config.yaml': base64.b64encode(raw.encode()).decode(),
'key.pem': 'FAKE_KEY_NOT_READ'}},
'keycape': {'metadata': {'name': 'keycape', 'uid': 'dep', 'resourceVersion': '1'},
'spec': dep_spec}}
original = copy.deepcopy(objects)
e = envelope()
e.update(runner_path='fake-runner', config_metadata=s.meta(objects['keycape-config']),
deployment_metadata=s.meta(objects['keycape']), deployment_spec_sha256=s.canonical(dep_spec),
registry_sha256=s.digest(raw.encode()))
e['source_hashes']['fake-runner'] = 'fixture'
receipt = {'cleanup_verified': False}
class Probe:
def exchange(self, *args):
return 'fixture-token'
def http_post(self, *args, **kwargs):
return 401, b'{"error":"invalid_profile_usage","feature":"client_id"}'
def run(self, *args, **kwargs):
if drift:
objects['keycape-config']['data']['unrelated'] = 'concurrent-owner-change'
return {'all_refused': probe_pass}
def kube(args, payload=None):
if args[0] == 'get':
return {'metadata': {'uid': s.CLUSTER}}
if args[0] == 'create':
if '--dry-run=server' in args:
return copy.deepcopy(payload)
created = copy.deepcopy(payload)
created['metadata'].update(uid='temp', resourceVersion='1')
objects[s.TEMP] = created
return created
if args[0] == 'delete':
self.assertEqual(payload['preconditions'], {'uid':'temp', 'resourceVersion':'1'})
del objects[s.TEMP]
return {'kind':'Status', 'status':'Success'}
if '--ignore-not-found' in args:
return objects.get(s.TEMP)
raise AssertionError('unexpected kube call')
def patched(kind, obj, path, value, dry=False):
result = copy.deepcopy(obj)
if path == '/spec':
result['spec'] = copy.deepcopy(value)
else:
result['data']['config.yaml'] = value
if not dry:
objects[obj['metadata']['name']] = result
return result
def delete(*args, **kwargs):
del objects[s.TEMP]
return type('Result', (), {'returncode': 0})()
class Discovery:
def __enter__(self):
import io
return io.StringIO('{"issuer":"https://kc.coulomb.social","token_endpoint":"https://kc.coulomb.social/token"}')
def __exit__(self, *args):
pass
with patch.object(s, 'verify_sources', return_value=150), patch.object(s, 'load_probe', return_value=Probe()), \
patch.object(s, 'SessionTransport') as transport_factory, \
patch.object(s, 'k', side_effect=kube), patch.object(s, 'get', side_effect=lambda kind, name: copy.deepcopy(objects[name])), \
patch.object(s, 'patch', side_effect=patched), patch.object(s, 'ready', return_value=True), \
patch.object(s.subprocess, 'run', side_effect=delete):
transport_factory.return_value.discovery.return_value = {
'issuer':'https://kc.coulomb.social', 'token_endpoint':'https://kc.coulomb.social/token'}
transport_factory.return_value.post.side_effect = Probe().http_post
if drift or not probe_pass:
with self.assertRaises(s.Refused):
s.execute(e, {'human_token': 'fake', 'coding_agent_token': 'fake'}, receipt)
else:
s.execute(e, {'human_token': 'fake', 'coding_agent_token': 'fake'}, receipt)
if drift:
self.assertFalse(receipt['cleanup_verified'])
self.assertIn(s.TEMP, objects)
self.assertEqual(objects['keycape-config']['data']['unrelated'], 'concurrent-owner-change')
else:
self.assertTrue(receipt['cleanup_verified'])
self.assertNotIn(s.TEMP, objects)
self.assertEqual(objects['keycape-config']['data'], original['keycape-config']['data'])
self.assertEqual(objects['keycape']['spec'], original['keycape']['spec'])
if __name__ == '__main__':
unittest.main()

37
workplans/ADHOC-2026-10-11.md
View file

@ -196,3 +196,40 @@ formal owner review under the temporary source-review stage. This completed
local review task is distinct from KEY39 T02 and all native owner obligations.
No founder decision, deployment, credential access, Secret read, grant, live
acceptance or task closure in other repositories is claimed.
T07 same-trial v3 refinement review: independently reviewed exact manifest
fbc31fda2b6194747547b9fbe876095914c3d53231c79abd43a010dbb3273efa.
Both v2 deadline counterexamples now refuse; a real elapsed timer interrupts
slow HTTP body reads and restores signal state. Thirty-nine supplied and fifteen
independent checks pass, including a retained new failure reproduction: final
temporary Secret deletion bypasses the aggregate cleanup deadline through an
unguarded subprocess call. Changes requested for that exact source path only;
v2 changes-requested evidence and frozen stages stay unchanged. No native
authority, owner acceptance or KEY39 task closure is inferred.
T07 same-trial final v4 source refinement review: exact frozen manifest
d079f85c47d5c2635e1ee633ea4d1346b1737ce2be9ca273f22a38e620f418d8
routes the exact temporary Secret deletion through guarded k. The v3 expired
cleanup reproduction now makes zero native calls; a real slow-delete timer is
clipped to remaining cleanup time. Both earlier v2 counterexamples now refuse,
late full OpenBao acceptance is self-revoked privately before failure, partial
acceptance stays unknown, and aggregate cleanup does not renew. Forty-one
supplied lifecycle tests, seventeen unchanged-runner tests and seventeen
independent checks pass. NetKingdom source agreement only; earlier unsuccessful
reviews and immutable sources remain history in this same trial. Native token
production, runtime authority, quiet-window/headroom, execution, revocation/
cleanup and other owners' task closure remain separate obligations.
T07 same-trial v5 exceptional HTTP-close refinement review: frozen manifest
d1f997624db46693a7126a6eb6435f6b3a353d5bd67dcc8a539683e4b7147285
retains NetKingdom's prior v4 agreement and Platform's distinct v4
changes-requested finding as history. Exceptional response close now has one
20 ms abort cap clipped to positive original request time and the same
nonrenewable aggregate cleanup deadline; expired aggregate performs no close.
Independent early-read and timed-read failure/slow-close checks, known accepted
session revocation despite normal-close error, signal restoration, value-free
unknown outcomes and prior deadline/delete/recovery guards pass. Forty-six
supplied lifecycle tests, seventeen unchanged-runner tests and twenty-one
independent checks pass. NetKingdom exact source agreement only; same trial,
no new task, native authority, credential access, runtime acceptance or closure.