tegwick
f3dbff6a77
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
106 lines
5.8 KiB
Python
106 lines
5.8 KiB
Python
"""Temporary native proof. Emits only fixed phases and booleans; never raw failures."""
|
|
import base64
|
|
import json
|
|
import stat
|
|
import sys
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode, quote_plus
|
|
from urllib.request import Request, build_opener, ProxyHandler, HTTPRedirectHandler
|
|
from urllib.error import HTTPError
|
|
from informed_decision.requester import RequesterTokenProvider, CLIENT_ID
|
|
from informed_decision.http_transport import JSONTransport
|
|
|
|
BAO = 'http://openbao.openbao.svc:8200/v1/'
|
|
ISSUER = 'https://kc.coulomb.social'
|
|
ROLE = 'informed-decision-sitting-requester-eso'
|
|
POLICY = 'workload-kv-read-informed-decision-sitting-requester-eso'
|
|
SECRET = Path('/var/run/secrets/informed-decision/requester/client-secret')
|
|
|
|
class NoRedirect(HTTPRedirectHandler):
|
|
def redirect_request(self, *args, **kwargs): return None
|
|
|
|
def http(method, url, body=None, token=None, form=False, client_auth=None):
|
|
headers = {'Content-Type': 'application/x-www-form-urlencoded' if form else 'application/json'}
|
|
if token: headers['X-Vault-Token'] = token
|
|
if client_auth: headers['Authorization'] = 'Basic ' + client_auth
|
|
raw = body if form else (json.dumps(body).encode() if body is not None else None)
|
|
req = Request(url, data=raw, headers=headers, method=method)
|
|
try: response = build_opener(ProxyHandler({}), NoRedirect()).open(req, timeout=5)
|
|
except HTTPError as error: response = error
|
|
with response:
|
|
status = response.code
|
|
content = response.read(262145)
|
|
if len(content) > 262144: raise ValueError()
|
|
return status, json.loads(content) if content else {}
|
|
|
|
def run(negative=False):
|
|
receipt = {'complete': False, 'phase': 'start', 'negative_identity': negative}
|
|
token = None
|
|
def check(name, condition):
|
|
receipt['phase'] = name
|
|
receipt[name] = bool(condition)
|
|
if not condition: raise ValueError()
|
|
try:
|
|
receipt['phase'] = 'login'
|
|
status, result = http('POST', BAO+'auth/kubernetes/login',
|
|
{'role': ROLE, 'jwt': Path('/identity/openbao').read_text()})
|
|
if status == 200: token = result.get('auth', {}).get('client_token')
|
|
if negative:
|
|
check('wrong_identity_refused', status == 403 and token is None)
|
|
else:
|
|
auth = result.get('auth', {})
|
|
check('exact_identity', status == 200 and bool(token)
|
|
and auth.get('metadata', {}).get('service_account_name') == 'informed-decision-eso'
|
|
and auth.get('metadata', {}).get('service_account_namespace') == 'informed-decision')
|
|
status, result = http('GET', BAO+'auth/token/lookup-self', token=token)
|
|
data = result.get('data', {})
|
|
check('exact_policy_and_ttl', status == 200 and data.get('policies') == [POLICY]
|
|
and not data.get('identity_policies') and 0 < data.get('ttl', 0) <= 900
|
|
and data.get('creation_ttl') == 900 and data.get('explicit_max_ttl') == 900)
|
|
status, _ = http('GET', BAO+'platform/data/workloads/secrets-engine/approval-requester', token=token)
|
|
check('sibling_get_refused', status == 403)
|
|
status, _ = http('LIST', BAO+'platform/metadata/workloads/informed-decision', token=token)
|
|
check('parent_list_refused', status == 403)
|
|
status, _ = http('POST', BAO+'auth/kubernetes/login',
|
|
{'role': ROLE, 'jwt': Path('/identity/wrong-audience').read_text()})
|
|
check('wrong_audience_refused', status == 403)
|
|
check('projection_mode', stat.S_IMODE(SECRET.stat().st_mode) == 0o440)
|
|
secret = SECRET.read_text()
|
|
status, result = http('GET', BAO+'platform/data/workloads/informed-decision/sitting-requester', token=token)
|
|
check('exact_path_projection_matches', status == 200 and bool(secret)
|
|
and result.get('data', {}).get('data', {}).get('CLIENT_SECRET') == secret)
|
|
receipt['phase'] = 'requester_exchange'
|
|
class ObservedTransport:
|
|
def request(self, method, url, **kwargs):
|
|
step = 'issuer_token_request' if method == 'POST' else 'issuer_jwks_request'
|
|
receipt['phase'] = step
|
|
status, result = JSONTransport().request(method, url, **kwargs)
|
|
receipt[step+'_ok'] = status == 200
|
|
return status, result
|
|
provider = RequesterTokenProvider(ISSUER, str(SECRET), transport=ObservedTransport())
|
|
check('signed_create_only_exchange', bool(provider()))
|
|
credentials = base64.b64encode((quote_plus(CLIENT_ID)+':'+quote_plus(secret)).encode()).decode()
|
|
for scope in ('approval:approve', 'approval:read', 'approval:consume'):
|
|
status, result = http('POST', ISSUER+'/token', form=True, client_auth=credentials, body=urlencode({
|
|
'grant_type': 'client_credentials', 'scope': scope}).encode())
|
|
check(scope.replace(':', '_')+'_refused', status in (400, 403) and 'access_token' not in result)
|
|
receipt['complete'] = True
|
|
except BaseException:
|
|
receipt['complete'] = False
|
|
finally:
|
|
if token:
|
|
try:
|
|
status, _ = http('POST', BAO+'auth/token/revoke-self', body={}, token=token)
|
|
receipt['reader_revoked'] = status in (200, 204)
|
|
status, _ = http('GET', BAO+'auth/token/lookup-self', token=token)
|
|
receipt['revoked_lookup_refused'] = status == 403
|
|
if not receipt['reader_revoked'] or not receipt['revoked_lookup_refused']: receipt['complete'] = False
|
|
except BaseException:
|
|
receipt['reader_revoked'] = False
|
|
receipt['complete'] = False
|
|
return receipt
|
|
|
|
if __name__ == '__main__':
|
|
result = run('--negative' in sys.argv)
|
|
print(json.dumps(result, sort_keys=True))
|
|
sys.exit(0 if result['complete'] else 1)
|