Add sitting-requester create-only token-exchange proof.
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Reader lane login, sibling-path denial, and KeyCape exchange checks.
No sitting POST. Attended wrapper selects the exact OIDC reader role.

Assistant: grok
Assistant-Session: 01a0a23b-3bf0-7341-b4e5-9dc05f72573a
This commit is contained in:
codex 2026-09-15 20:38:47 +02:00
parent 50f031091c
commit 73178f4ae2
6 changed files with 172 additions and 2 deletions

View file

@ -6,6 +6,12 @@ import shutil
import sys
LANES = (
'openbao-platform-admin-login',
'informed-decision-sitting-requester-login',
)
def reviewed_command(args):
if args and args[0] == '--':
args = args[1:]
@ -23,6 +29,20 @@ def reviewed_command(args):
return [str(command), *args[1:]]
def parse_argv(argv):
args = list(argv)
lane = 'openbao-platform-admin-login'
if args[:1] == ['--lane']:
if len(args) < 2:
raise SystemExit('lane required')
lane, args = args[1], args[2:]
elif args and args[0].startswith('--lane='):
lane, args = args[0].split('=', 1)[1], args[1:]
if lane not in LANES:
raise SystemExit('unsupported attended login lane')
return lane, reviewed_command(args)
TUNNEL = 'http://127.0.0.1:18200'
@ -32,6 +52,7 @@ def contained_env():
env['VAULT_ADDR'] = TUNNEL
env.pop('BAO_TOKEN', None)
env.pop('VAULT_TOKEN', None)
env['WARDEN_ROUTING_CATALOG'] = '/home/worsch/ops-warden/registry/routing/catalog.yaml'
if not any(shutil.which(x) for x in ('xdg-open', 'x-www-browser', 'www-browser')):
if not Path('/mnt/c/Windows/System32/WindowsPowerShell/v1.0/powershell.exe').is_file():
raise SystemExit('No supported attended browser launcher is available')
@ -40,8 +61,8 @@ def contained_env():
def main():
args = reviewed_command(sys.argv[1:])
os.execvpe('warden', ['warden', 'access', 'openbao-platform-admin-login', '--exec', '--', *args], contained_env())
lane, args = parse_argv(sys.argv[1:])
os.execvpe('warden', ['warden', 'access', lane, '--exec', '--', *args], contained_env())
if __name__ == '__main__':

View file

@ -0,0 +1,119 @@
"""Silent create-only token-exchange proof; no sitting POST."""
import base64
import json
import os
import stat
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
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 ValueError('response_too_large')
return status, json.loads(content) if content else {}
def bao(*args):
import subprocess
p = subprocess.run(['bao', *args], capture_output=True, text=True, timeout=20)
if p.returncode:
raise ValueError('metadata_failed')
return json.loads(p.stdout)
def check_identity(data):
policies = set(data.get('policies', [])) | set(data.get('identity_policies', []))
if POLICY not in policies or policies - {POLICY, 'default'} or not data.get('entity_id') or not 0 < data.get('ttl', 0) <= 900:
raise ValueError('reader_identity_failed')
def main(receipt):
if Path.home().parent.name != '.warden-attended-login' or os.getenv('BAO_TOKEN') or os.getenv('VAULT_TOKEN'):
raise ValueError('attended_reader_required')
check_identity(bao('token', 'lookup', '-format=json')['data'])
for path, expected in ((KV, ['read']), (SIBLING, ['deny']), (PARENT, ['deny'])):
result = bao('token', 'capabilities', '-format=json', path)
if isinstance(result, dict):
result = result.get('data', result).get('capabilities', result)
if result != expected:
raise ValueError('reader_scope_failed')
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 ValueError('private_helper_required')
addr = os.environ.get('BAO_ADDR', 'http://127.0.0.1:18200').rstrip('/')
status, data = http(addr + '/v1/' + KV + '?version=1', headers={'X-Vault-Token': helper.read_text().strip()})
if status != 200 or data['data']['metadata']['version'] != 1:
raise ValueError('requester_delivery_failed')
secret = data['data']['data']['CLIENT_SECRET']
del data
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:
raise ValueError('requester_exchange_failed')
token = tokens['access_token']
status, jwks = http(ISSUER + '/jwks')
if status != 200:
raise ValueError('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 ValueError('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']})
expected = {'sub': 'informed-decision', 'tenant': 'tenant:platform', 'principal_type': 'service',
'scope': 'approval:create', 'roles': ['informed-decision-sitting-requester'], 'groups': []}
if any(claims.get(key) != value for key, value in expected.items()) or claims['exp'] - claims['iat'] != 900:
raise ValueError('requester_claims_failed')
for scope in ('approval:approve', 'approval:consume', 'approval:read'):
if exchange(scope)[0] != 400:
raise ValueError('excess_scope_not_refused')
if exchange('approval:create', 'invalid-synthetic-credential')[0] != 401:
raise ValueError('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():
raise SystemExit(1)
receipt = {'observed_at': __import__('datetime').datetime.now(__import__('datetime').timezone.utc).isoformat(),
'status': 'failed', 'phase': 'preflight', 'credential_values_emitted': False, 'sitting_post': False}
try:
main(receipt)
except Exception:
raise SystemExit(1) from None
finally:
RECEIPT.write_text(json.dumps(receipt, indent=2) + '\n')

View file

@ -0,0 +1,3 @@
#!/usr/bin/env bash
set -euo pipefail
exec python3 "$(dirname "$0")/prove-sitting-requester-exchange.py" "$@" >/dev/null 2>&1

View file

@ -87,6 +87,17 @@ def test_receipt_records_apply_and_failure(tmp_path, monkeypatch):
assert json.loads(failed.read_text())['status'] == 'bao_write_failed'
def test_attended_wrapper_selects_sitting_requester_lane():
spec = importlib.util.spec_from_file_location(
'attended', Path(__file__).resolve().parents[1] / 'scripts/openbao-attended-exec.py')
wrapper = importlib.util.module_from_spec(spec)
spec.loader.exec_module(wrapper)
command = Path(__file__).resolve().parents[1] / 'scripts/prove-sitting-requester-exchange.sh'
lane, args = wrapper.parse_argv(['--lane', 'informed-decision-sitting-requester-login', str(command)])
assert lane == 'informed-decision-sitting-requester-login'
assert args[0] == str(command.resolve())
def test_attended_wrapper_pins_operator_tunnel_address(monkeypatch):
spec = importlib.util.spec_from_file_location(
'attended', Path(__file__).resolve().parents[1] / 'scripts/openbao-attended-exec.py')

View file

@ -34,5 +34,18 @@ class SittingRequesterProvisioningTests(unittest.TestCase):
self.assertEqual(named['serviceSubject'], 'informed-decision')
class SittingRequesterExchangeTests(unittest.TestCase):
def test_exchange_helper_does_not_post_sittings(self):
spec = importlib.util.spec_from_file_location(
'exchange', Path(__file__).resolve().parents[1] / 'scripts/prove-sitting-requester-exchange.py')
exchange = importlib.util.module_from_spec(spec)
spec.loader.exec_module(exchange)
source = Path(exchange.__file__).read_text()
self.assertNotIn('/v1/approvals', source)
self.assertEqual(exchange.CLIENT_ID, 'informed-decision-sitting-requester')
self.assertEqual(exchange.POLICY, 'workload-kv-read-informed-decision-sitting-requester-client')
self.assertIn('approval-requester', exchange.SIBLING)
if __name__ == '__main__':
unittest.main()

View file

@ -58,3 +58,6 @@ KV version 1, ESO Ready/SecretSynced, KeyCape single Ready replica, CCRs
applied. No sitting POST. Remaining: create-only token-exchange proof
(positive create scope, refuse approve/consume, sibling path deny). Evidence:
`docs/evidence/2026-09-15-sitting-requester-provision.json`.
Exchange proof uses reader lane `informed-decision-sitting-requester-login`
and `scripts/prove-sitting-requester-exchange.sh`. It does not POST sittings.