tegwick
6f33abddcf
All checks were successful
Build and Publish Container Image / build-and-push (push) Successful in 41s
Assistant: codex Assistant-Model: gpt-5.6-luna Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
58 lines
2.5 KiB
Python
58 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Read bounded probe logs on stdin and emit only a validated metadata receipt."""
|
|
import json
|
|
import re
|
|
import sys
|
|
|
|
ISSUERS = {'https://auth.coulomb.social', 'http://auth.coulomb.social',
|
|
'http://authelia.sso.svc.cluster.local:9091'}
|
|
FAILURES = {'authorization_callback_refused', 'token_exchange_error', 'token_exchange_refused',
|
|
'token_response_invalid', 'id_token_verification_error', 'issuer_outside_reviewed_set',
|
|
'id_token_issuer_mismatch', 'id_token_audience_mismatch', 'id_token_expired',
|
|
'id_token_validity_window', 'id_token_signature', 'provider_metadata_unavailable',
|
|
'provider_keys_unavailable', 'id_token_nonce_mismatch', 'probe_deadline', 'probe_listener_stopped'}
|
|
|
|
|
|
def sanitize(raw):
|
|
if len(raw) > 8192:
|
|
raise ValueError()
|
|
lines = raw.splitlines()
|
|
if not lines:
|
|
raise ValueError()
|
|
data = json.loads(lines[0])
|
|
if not isinstance(data, dict) or data.get('schema') != 'keycape.upstream-issuer-proof.v1':
|
|
raise ValueError()
|
|
if data.get('tokens_retained') is not False or data.get('downstream_credential_issued') is not False:
|
|
raise ValueError()
|
|
keys = {'schema', 'status', 'tokens_retained', 'downstream_credential_issued'}
|
|
if 'observed_at' in data:
|
|
if not isinstance(data['observed_at'], str) or not re.fullmatch(r'\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ', data['observed_at']):
|
|
raise ValueError()
|
|
keys.add('observed_at')
|
|
if data.get('status') == 'verified':
|
|
if len(lines) != 1 or data.get('issuer') not in ISSUERS or 'observed_at' not in data:
|
|
raise ValueError()
|
|
keys.add('issuer')
|
|
for field in ['signature_verified', 'audience_verified', 'validity_window_verified', 'nonce_verified']:
|
|
if data.get(field) is not True:
|
|
raise ValueError()
|
|
keys.add(field)
|
|
elif data.get('status') == 'failed' and data.get('failure') in FAILURES:
|
|
if len(lines) > 2 or (len(lines) == 2 and lines[1] != 'issuer proof failed'):
|
|
raise ValueError()
|
|
keys.add('failure')
|
|
else:
|
|
raise ValueError()
|
|
if set(data) != keys:
|
|
raise ValueError()
|
|
return data
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
result = sanitize(sys.stdin.read(8193))
|
|
except Exception:
|
|
print('No valid metadata-only issuer proof receipt.', file=sys.stderr)
|
|
raise SystemExit(1) from None
|
|
print(json.dumps(result, indent=2))
|
|
raise SystemExit(0 if result['status'] == 'verified' else 1)
|