Prepare independently reviewed native custody capability verification
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
This commit is contained in:
parent
c52be782d0
commit
969fe218df
7 changed files with 1000 additions and 1 deletions
325
scripts/verify-keycape-client-custody.py
Normal file
325
scripts/verify-keycape-client-custody.py
Normal file
|
|
@ -0,0 +1,325 @@
|
|||
#!/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())
|
||||
Loading…
Add table
Add a link
Reference in a new issue