9ce42da97e
The reader login succeeded; the child died at preflight. Use bao for the KV read, accept parent/sibling deny shapes, and record a failure class. Assistant: grok Assistant-Session: 01a0a23b-3bf0-7341-b4e5-9dc05f72573a
173 lines
7.1 KiB
Python
173 lines
7.1 KiB
Python
"""Silent create-only token-exchange proof; no sitting POST."""
|
|
import base64
|
|
import json
|
|
import os
|
|
import stat
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from urllib.error import HTTPError
|
|
from urllib.parse import urlencode
|
|
from urllib.request import ProxyHandler, Request, build_opener, HTTPRedirectHandler
|
|
import jwt
|
|
|
|
ROOT = Path('/home/worsch/railiance-platform')
|
|
RECEIPT = ROOT / 'docs/evidence/2026-09-15-sitting-requester-exchange.json'
|
|
POLICY = 'workload-kv-read-informed-decision-sitting-requester-client'
|
|
KV = 'platform/data/workloads/informed-decision/sitting-requester'
|
|
SIBLING = 'platform/data/workloads/secrets-engine/approval-requester'
|
|
PARENT = 'platform/metadata/workloads/informed-decision'
|
|
ISSUER = 'https://kc.coulomb.social'
|
|
CLIENT_ID = 'informed-decision-sitting-requester'
|
|
|
|
|
|
class NoRedirect(HTTPRedirectHandler):
|
|
def redirect_request(self, *args, **kwargs):
|
|
return None
|
|
|
|
|
|
class Refused(Exception):
|
|
pass
|
|
|
|
|
|
def http(url, *, body=None, headers=None):
|
|
req = Request(url, data=body, headers=headers or {})
|
|
try:
|
|
with build_opener(ProxyHandler({}), NoRedirect()).open(req, timeout=20) as response:
|
|
content = response.read(1048577)
|
|
status = response.status
|
|
except HTTPError as error:
|
|
status = error.code
|
|
content = error.read(1048577)
|
|
error.close()
|
|
if len(content) > 1048576:
|
|
raise Refused('response_too_large')
|
|
if not content:
|
|
return status, {}
|
|
try:
|
|
return status, json.loads(content)
|
|
except json.JSONDecodeError:
|
|
raise Refused('non_json_http_response')
|
|
|
|
|
|
def bao(*args):
|
|
import subprocess
|
|
p = subprocess.run(['bao', *args], capture_output=True, text=True, timeout=20)
|
|
if p.returncode:
|
|
raise Refused('metadata_failed')
|
|
return json.loads(p.stdout)
|
|
|
|
|
|
def capabilities(result):
|
|
if isinstance(result, dict):
|
|
result = result.get('data', result).get('capabilities', result)
|
|
return result if isinstance(result, list) else []
|
|
|
|
|
|
def is_deny(result):
|
|
caps = set(capabilities(result))
|
|
return caps <= {'deny'} or not caps.intersection({'read', 'create', 'update', 'delete', 'list', 'patch', 'sudo'})
|
|
|
|
|
|
def check_identity(data):
|
|
policies = set(data.get('policies', [])) | set(data.get('identity_policies', []))
|
|
if POLICY not in policies or policies - {POLICY, 'default'}:
|
|
raise Refused('reader_identity_failed')
|
|
if not data.get('entity_id') or not 0 < int(data.get('ttl') or 0) <= 900:
|
|
raise Refused('reader_identity_failed')
|
|
|
|
|
|
def secret_from_read(payload):
|
|
body = payload.get('data', payload)
|
|
inner = body.get('data', body)
|
|
version = (body.get('metadata') or {}).get('version')
|
|
secret = inner.get('CLIENT_SECRET') if isinstance(inner, dict) else None
|
|
if version != 1 or not secret:
|
|
raise Refused('requester_delivery_failed')
|
|
return secret
|
|
|
|
|
|
def main(receipt):
|
|
if Path.home().parent.name != '.warden-attended-login' or os.getenv('BAO_TOKEN') or os.getenv('VAULT_TOKEN'):
|
|
raise Refused('attended_reader_required')
|
|
receipt['phase'] = 'envelope_ok'
|
|
check_identity(bao('token', 'lookup', '-format=json')['data'])
|
|
receipt['phase'] = 'identity_ok'
|
|
if capabilities(bao('token', 'capabilities', '-format=json', KV)) != ['read']:
|
|
raise Refused('reader_scope_failed:kv')
|
|
if not is_deny(bao('token', 'capabilities', '-format=json', SIBLING)):
|
|
raise Refused('reader_scope_failed:sibling')
|
|
if not is_deny(bao('token', 'capabilities', '-format=json', PARENT)):
|
|
raise Refused('reader_scope_failed:parent')
|
|
receipt['phase'] = 'scope_ok'
|
|
helper = Path.home() / '.vault-token'
|
|
info = helper.lstat()
|
|
if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o600 or info.st_uid != os.getuid():
|
|
raise Refused('private_helper_required')
|
|
secret = secret_from_read(bao('read', '-format=json', KV))
|
|
receipt['phase'] = 'secret_present'
|
|
receipt['secret_present'] = True
|
|
|
|
def exchange(scope, credential=secret):
|
|
auth = base64.b64encode((CLIENT_ID + ':' + credential).encode()).decode()
|
|
return http(ISSUER + '/token', body=urlencode({'grant_type': 'client_credentials', 'scope': scope}).encode(),
|
|
headers={'Authorization': 'Basic ' + auth, 'Content-Type': 'application/x-www-form-urlencoded'})
|
|
|
|
status, tokens = exchange('approval:create')
|
|
if status != 200 or 'access_token' not in tokens:
|
|
raise Refused('requester_exchange_failed')
|
|
receipt['phase'] = 'exchange_ok'
|
|
token = tokens['access_token']
|
|
status, jwks = http(ISSUER + '/jwks')
|
|
if status != 200:
|
|
raise Refused('jwks_failed')
|
|
header = jwt.get_unverified_header(token)
|
|
keys = [key for key in jwks['keys'] if key['kid'] == header.get('kid')]
|
|
if header.get('alg') != 'RS256' or len(keys) != 1:
|
|
raise Refused('signing_key_failed')
|
|
claims = jwt.decode(token, jwt.PyJWK.from_dict(keys[0]).key, algorithms=['RS256'], issuer=ISSUER,
|
|
audience='approval-engine', options={'strict_aud': True, 'require': ['sub', 'iat', 'exp', 'iss', 'aud']})
|
|
roles = claims.get('roles')
|
|
if isinstance(roles, str):
|
|
roles = [roles]
|
|
if claims.get('sub') != 'informed-decision':
|
|
raise Refused('requester_claims_failed:sub')
|
|
if claims.get('tenant') != 'tenant:platform':
|
|
raise Refused('requester_claims_failed:tenant')
|
|
if claims.get('principal_type') != 'service':
|
|
raise Refused('requester_claims_failed:principal_type')
|
|
if claims.get('scope') != 'approval:create':
|
|
raise Refused('requester_claims_failed:scope')
|
|
if roles != ['informed-decision-sitting-requester']:
|
|
raise Refused('requester_claims_failed:roles')
|
|
if claims.get('groups') not in (None, []):
|
|
raise Refused('requester_claims_failed:groups')
|
|
if int(claims['exp']) - int(claims['iat']) != 900:
|
|
raise Refused('requester_claims_failed:ttl')
|
|
receipt['phase'] = 'claims_ok'
|
|
for scope in ('approval:approve', 'approval:consume', 'approval:read'):
|
|
if exchange(scope)[0] != 400:
|
|
raise Refused('excess_scope_not_refused')
|
|
if exchange('approval:create', 'invalid-synthetic-credential')[0] != 401:
|
|
raise Refused('wrong_secret_not_refused')
|
|
receipt.update(status='verified', phase='requester_verified', signature_verified=True,
|
|
excess_scopes_refused=True, wrong_secret_refused=True, reader_scope_verified=True,
|
|
sitting_post=False)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if RECEIPT.exists():
|
|
try:
|
|
existing = json.loads(RECEIPT.read_text())
|
|
except Exception:
|
|
raise SystemExit(1)
|
|
if existing.get('status') == 'verified':
|
|
raise SystemExit(1)
|
|
receipt = {'observed_at': datetime.now(timezone.utc).isoformat(),
|
|
'status': 'failed', 'phase': 'preflight', 'credential_values_emitted': False, 'sitting_post': False}
|
|
try:
|
|
main(receipt)
|
|
except Exception as error:
|
|
receipt['failure'] = str(error) if isinstance(error, Refused) else 'contained_operation_failed'
|
|
raise SystemExit(1) from None
|
|
finally:
|
|
RECEIPT.write_text(json.dumps(receipt, indent=2) + '\n')
|