ops-mason/scripts/verify-keycape-client-custody.py
tegwick 969fe218df Prepare independently reviewed native custody capability verification
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
2026-10-11 14:16:58 +02:00

325 lines
16 KiB
Python

#!/usr/bin/env python3
"""MASON-WP-0009-T05 attended, value-free lane verifier; no KV GETs.
The existing operator/ESO policies do NOT grant capabilities-self. An existing
read-only ops-mason-survey grant queries their tokens via sys/capabilities.
This file prepares a founder-run procedure, not a new authorization surface.
"""
import argparse
import hashlib
import json
import os
from pathlib import Path
import subprocess
import selectors
import signal
import sys
import time
import urllib.request
import urllib.error
DATA = 'platform/data/workloads/keycape/secrets-engine-openbao'
OP_ROLE = 'keycape-secrets-engine-openbao-operator-workload-kv-read'
ESO_ROLE = 'external-secrets-keycape-secrets-engine-openbao'
POLICY = 'workload-kv-read-keycape-secrets-engine-openbao-'
PATHS = [DATA, DATA.replace('/data/', '/metadata/'),
'platform/data/workloads/keycape', 'platform/data/workloads/keycape/',
DATA + '-other', 'platform/data/workloads/keycape/other',
'platform/data/workloads/keycape/*', 'sys/policies/acl/ops-mason-survey',
'auth/token/create', 'auth/token/revoke-self']
BASE = 'http://127.0.0.1:18200'
class Refused(Exception):
pass
class RecoveredLoginRefused(Refused):
"""Retain only an already recovered session for finally cleanup, never proof."""
def __init__(self, code, auth):
super().__init__(code)
token = auth.get('client_token') if isinstance(auth, dict) else None
self.token = token if isinstance(token, str) and 0 < len(token) <= 4096 and not any(c.isspace() for c in token) else ''
class Transport(Exception):
pass
class Api:
"""Hard-coded endpoint allowlist prevents secret GETs, role/policy writes."""
def __init__(self):
self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}),
NoRedirect())
def call(self, endpoint, token='', body=None):
allowed = {'auth/token/lookup-self', 'auth/token/revoke-self',
'sys/capabilities', 'sys/capabilities-self',
'auth/kubernetes/login'}
if endpoint not in allowed:
raise Refused('endpoint_not_allowlisted')
req = urllib.request.Request(BASE + '/v1/' + endpoint,
data=None if body is None else json.dumps(body).encode(),
headers={'Content-Type': 'application/json', 'X-Vault-Token': token},
method='GET' if endpoint == 'auth/token/lookup-self' else 'POST')
try:
with self.opener.open(req, timeout=20) as response:
if endpoint == 'auth/token/revoke-self':
# OpenBao's native self-revoke success is 204 with no body.
if response.status != 204 or response.read(1):
raise Refused('unconfirmed_revoke_response')
return {}
if response.status != 200 or response.headers.get_content_type() != 'application/json':
raise Refused('invalid_status_or_content_type')
length = response.headers.get('Content-Length')
if length is not None and (not length.isdecimal() or int(length) > 65536):
raise Refused('oversized_or_invalid_content_length')
raw = response.read(65537)
if len(raw) > 65536:
raise Refused('oversized_response')
result = json.loads(raw)
if not isinstance(result, dict):
raise Refused('invalid_response_envelope')
if 'errors' in result:
if endpoint == 'auth/kubernetes/login':
raise RecoveredLoginRefused('native_error_envelope', result.get('auth'))
raise Refused('native_error_envelope')
key = 'auth' if endpoint == 'auth/kubernetes/login' else 'data'
if not isinstance(result.get(key), dict):
raise Refused('invalid_endpoint_response_shape')
return result
except urllib.error.HTTPError as error:
error.close() # never read/print server errors
raise Refused('http_' + str(error.code)) from None
except (OSError, ValueError, urllib.error.URLError):
raise Transport('transport_or_invalid_response') from None
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *args, **kwargs):
raise Refused('redirect_refused')
def digest(value):
return hashlib.sha256(value.encode()).hexdigest()
def accessor_digest(value):
if not isinstance(value, str) or not 1 <= len(value) <= 256 or any(c.isspace() for c in value):
raise Refused('missing_or_invalid_native_accessor')
return digest(value)
def checked_caps(response, expected):
data = response.get('data', response)
# Only batch-shaped named results accepted, never substitute a default.
if not isinstance(data, dict) or not all(path in data for path in expected):
raise Refused('incomplete_capability_matrix')
result = {}
for path, caps in expected.items():
actual = data[path]
if not isinstance(actual, list) or sorted(actual) != sorted(caps):
raise Refused('capability_mismatch')
result[path] = actual
return result
def survey(api, token):
info = api.call('auth/token/lookup-self', token)['data']
if info.get('identity_policies'):
raise Refused('query_identity_policy_additions')
policies = set(info.get('policies', []))
if policies not in ({'ops-mason-survey'}, {'ops-mason-survey', 'default'}):
raise Refused('query_policy_mismatch')
if info.get('display_name') != 'token-claude-MASON-WP-0009-T05' or (type(info.get('ttl')) is not int or not 0 < info['ttl'] <= 2700):
raise Refused('query_identity_or_ttl_mismatch')
expected = {DATA: ['deny'], 'sys/policies/acl/ops-mason-survey': ['deny'],
'auth/token/create': ['deny'], 'sys/capabilities': ['update']}
checked_caps(api.call('sys/capabilities-self', token, {'paths': list(expected)}), expected)
return info
def safe_env(home):
# Never inherit credential variables, token helpers, proxies, CLI plugins.
return {'PATH': '/home/worsch/.local/bin:/usr/local/bin:/usr/bin:/bin', 'HOME': home,
'BAO_ADDR': BASE, 'VAULT_ADDR': BASE,
'DISPLAY': os.environ.get('DISPLAY', ''),
'WSL_INTEROP': os.environ.get('WSL_INTEROP', ''),
'WSL_DISTRO_NAME': os.environ.get('WSL_DISTRO_NAME', '')}
def run(argv, env):
# Capture each stream with a real memory bound while the child runs; never
# retain an unbounded PIPE response and only check its length afterwards.
process = None
try:
process = subprocess.Popen(argv, env=env, stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True)
buffers = {process.stdout: bytearray(), process.stderr: bytearray()}
deadline = time.monotonic() + 300
with selectors.DefaultSelector() as selector:
for stream in buffers:
selector.register(stream, selectors.EVENT_READ)
while selector.get_map():
remaining = deadline - time.monotonic()
if remaining <= 0:
raise Transport('native_transport_or_timeout')
for key, _ in selector.select(min(remaining, 1)):
chunk = os.read(key.fileobj.fileno(), 8192)
if not chunk:
selector.unregister(key.fileobj)
continue
target = buffers[key.fileobj]
if len(target) + len(chunk) > 65536:
raise Refused('oversized_native_response')
target.extend(chunk)
remaining = deadline - time.monotonic()
if remaining <= 0:
raise Transport('native_transport_or_timeout')
code = process.wait(timeout=remaining)
return code, bytes(buffers[process.stdout]), bytes(buffers[process.stderr])
except (OSError, subprocess.TimeoutExpired):
raise Transport('native_transport_or_timeout') from None
finally:
if process is not None:
if process.poll() is None:
try:
os.killpg(process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
process.wait()
process.stdout.close()
process.stderr.close()
def oidc(env, runner=run):
# -no-store is mandatory; output/errors stay inside the founder's process.
code, raw, err = runner(['/home/worsch/.local/bin/bao', 'login', '-no-store', '-format=json',
'-method=oidc', '-path=netkingdom', 'role=' + OP_ROLE], env)
if code:
# A candidate refusal still needs independent issuer + correlated
# OpenBao owner evidence. HTTP 403, cancellation or timeout never pass.
claim_refusal = b'claim "groups" does not match' in err or b'claim "groups" not found' in err
return None, 'group_refusal_candidate' if claim_refusal else 'unproved_login_failure'
try:
result = json.loads(raw)
if not isinstance(result, dict) or not isinstance(result.get('auth'), dict):
raise Refused('malformed_native_login')
if 'errors' in result:
raise RecoveredLoginRefused('native_error_envelope', result['auth'])
return result['auth'], 'login_succeeded'
except (ValueError, KeyError, TypeError):
raise Refused('malformed_native_login') from None
def eso(api, runner=run):
# Existing owner SSH certificate, no kubeconfig/Secret reads. Separate
# TokenRequest (10m, openbao audience); no ESO controller session revoked.
env = {'PATH': '/usr/local/bin:/usr/bin:/bin', 'HOME': os.path.expanduser('~')}
rc, out, _ = runner(['ssh', '-oBatchMode=yes', 'railiance01',
'kubectl get nodes -o json'], env)
if rc:
raise Transport('cluster_identity_unproved')
try:
nodes = json.loads(out)['items']
addresses = {a['address'] for n in nodes for a in n['status']['addresses']}
if len(nodes) != 1 or '92.205.62.239' not in addresses:
raise Refused('wrong_cluster')
except (KeyError, TypeError, ValueError):
raise Refused('invalid_cluster_identity') from None
rc, jwt, _ = runner(['ssh', '-oBatchMode=yes', 'railiance01',
'kubectl -n external-secrets create token external-secrets --audience=openbao --duration=10m'], env)
if rc or not jwt.strip():
raise Transport('tokenrequest_failed')
return api.call('auth/kubernetes/login', body={'role': ESO_ROLE,
'jwt': jwt.decode().strip()})['auth']
def lane_matrix(api, query, auth, mode):
token = auth.get('client_token')
if not isinstance(token, str) or not token or any(c.isspace() for c in token):
raise Refused('malformed_lane_token')
metadata = auth.get('metadata', {})
expected_role = OP_ROLE if mode == 'operator' else ESO_ROLE
if metadata.get('role') != expected_role:
raise Refused('lane_role_mismatch')
if mode == 'eso' and (metadata.get('service_account_name') != 'external-secrets' or
metadata.get('service_account_namespace') != 'external-secrets'):
raise Refused('eso_identity_mismatch')
if auth.get('identity_policies'):
raise Refused('lane_identity_policy_additions')
expected_policy = POLICY + ('operator' if mode == 'operator' else 'eso')
if set(auth.get('policies', [])) != {expected_policy}:
raise Refused('lane_policy_mismatch')
if (type(auth.get('lease_duration')) is not int or not 0 < auth['lease_duration'] <= (300 if mode == 'operator' else 900)):
raise Refused('lane_ttl_mismatch')
expected = {p: ['deny'] for p in PATHS}
expected[DATA] = ['read']
expected['auth/token/revoke-self'] = ['update']
return checked_caps(api.call('sys/capabilities', query,
{'token': token, 'paths': PATHS}), expected)
def main():
parser = argparse.ArgumentParser()
parser.add_argument('mode', choices=['operator', 'eso', 'nonmember'])
parser.add_argument('--survey-token-fd', type=int, required=True)
parser.add_argument('--approval-reference', required=True)
parser.add_argument('--approved-source-sha256', required=True)
parser.add_argument('--receipt', type=Path, required=True)
parser.add_argument('--attended', action='store_true')
args = parser.parse_args()
if not args.attended or not sys.stdin.isatty() or args.survey_token_fd < 3:
parser.error('founder terminal attendance and separate survey token FD required')
if any(os.environ.get(k) for k in ('BAO_TOKEN', 'VAULT_TOKEN', 'BAO_NAMESPACE', 'VAULT_NAMESPACE')):
parser.error('ambient authority refused')
source_sha256 = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
if args.approved_source_sha256 != source_sha256:
parser.error('approved source digest differs; no session requested')
# Reservation prevents overwrites/symlink substitution before credentials.
fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
receipt = {'schema': 'mason-client-custody-native/v1', 'task': 'MASON-WP-0009-T05',
'mode': args.mode, 'source_sha256': source_sha256,
'approval_reference': args.approval_reference, 'at': int(time.time()),
'secret_reads': 0, 'result': 'unproved', 'owner_acceptance': 'pending',
'query_revoke': False, 'lane_revoke': False}
api = Api()
query = ''
query_valid = False
lane = ''
try:
with os.fdopen(args.survey_token_fd, 'r') as handle:
query = handle.read(4097).strip()
if not query or len(query) > 4096 or any(c.isspace() for c in query):
raise Refused('malformed_query_token')
query_info = survey(api, query)
query_valid = True
receipt['query_accessor_sha256'] = accessor_digest(query_info.get('accessor'))
if args.mode == 'eso':
auth = eso(api)
else:
import tempfile
with tempfile.TemporaryDirectory(prefix='mason-oidc-') as home:
auth, result = oidc(safe_env(home))
receipt['native_login_result'] = result
if auth is None:
# Never pronounce the nonmember check passed from CLI text.
raise Refused(result)
lane = auth.get('client_token', '')
receipt['lane_accessor_sha256'] = accessor_digest(auth.get('accessor'))
if args.mode == 'nonmember':
raise Refused('nonmember_obtained_token')
receipt['capabilities'] = lane_matrix(api, query, auth, args.mode)
receipt['result'] = 'capability_matrix_observed'
except RecoveredLoginRefused as error:
lane = error.token
receipt['failure_class'] = str(error)
except (Refused, Transport) as error:
receipt['failure_class'] = str(error)
except Exception:
receipt['failure_class'] = 'invalid_or_unexpected_response'
finally:
if lane:
try:
api.call('auth/token/revoke-self', lane, {})
receipt['lane_revoke'] = True
except Exception:
receipt['result'] = 'unproved'
if query_valid:
try:
api.call('auth/token/revoke-self', query, {})
receipt['query_revoke'] = True
except Exception:
receipt['result'] = 'unproved'
query = lane = ''
with os.fdopen(fd, 'w') as handle:
json.dump(receipt, handle, indent=2, sort_keys=True)
handle.write('\n')
return 0 if receipt['result'] == 'capability_matrix_observed' else 1
if __name__ == '__main__':
sys.exit(main())