import importlib.util from pathlib import Path import unittest SCRIPT = Path(__file__).resolve().parents[1] / 'scripts/verify-keycape-client-custody.py' spec = importlib.util.spec_from_file_location('custody', SCRIPT) m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) NATIVE_API = m.Api class Fake: def __init__(self, info=None, changes=None): self.calls = [] self.info = info or {'policies': ['ops-mason-survey', 'default'], 'display_name': 'token-claude-MASON-WP-0009-T05', 'ttl': 2700, 'accessor': 'test-query-accessor'} self.changes = changes or {} def call(self, endpoint, token='', body=None): self.calls.append((endpoint, token, body)) if endpoint == 'auth/token/lookup-self': return {'data': self.info} if endpoint == 'sys/capabilities-self': result = {m.DATA: ['deny'], 'sys/policies/acl/ops-mason-survey': ['deny'], 'auth/token/create': ['deny'], 'sys/capabilities': ['update']} else: result = {p: ['deny'] for p in m.PATHS} result[m.DATA] = ['read'] result['auth/token/revoke-self'] = ['update'] result.update(self.changes) return {'data': result} class Tests(unittest.TestCase): def auth(self, mode='operator'): return {'client_token': 'test-only-not-a-credential', 'accessor': 'test-accessor', 'policies': [m.POLICY + mode], 'lease_duration': 300, 'metadata': {'role': m.OP_ROLE if mode == 'operator' else m.ESO_ROLE, 'service_account_name': 'external-secrets', 'service_account_namespace': 'external-secrets'}} def test_query_identity_and_data_deny(self): self.assertEqual(m.survey(Fake(), 'test-query')['accessor'], 'test-query-accessor') def test_broad_ambient_query_refused(self): for policies in [['root'], ['ops-mason-survey', 'platform-admin']]: with self.subTest(policies=policies), self.assertRaises(m.Refused): m.survey(Fake(info={'policies': policies}), 'test-query') def test_wrong_task_and_unbounded_grant_refused(self): for field, value in [('display_name', 'claude-another-task'), ('ttl', 2701), ('ttl', 0)]: fake = Fake() fake.info[field] = value with self.subTest(field=field, value=value), self.assertRaises(m.Refused): m.survey(fake, 'test-query') def test_query_cannot_read_value_or_create_token(self): for path in [m.DATA, 'auth/token/create', 'sys/policies/acl/ops-mason-survey']: with self.subTest(path=path), self.assertRaises(m.Refused): m.survey(Fake(changes={path: ['read']}), 'test-query') def test_operator_and_eso_native_matrix(self): for mode in ['operator', 'eso']: fake = Fake() result = m.lane_matrix(fake, 'test-query', self.auth(mode), mode) self.assertEqual(result[m.DATA], ['read']) self.assertEqual(fake.calls[0][0], 'sys/capabilities') self.assertEqual(fake.calls[0][1], 'test-query') def test_data_write_and_missing_read_refused(self): for caps in [['read', 'update'], ['deny'], ['sudo', 'read']]: with self.subTest(caps=caps), self.assertRaises(m.Refused): m.lane_matrix(Fake(changes={m.DATA: caps}), 'test-query', self.auth(), 'operator') def test_parent_sibling_metadata_and_policy_escalation_refused(self): for path in m.PATHS[1:-1]: with self.subTest(path=path), self.assertRaises(m.Refused): m.lane_matrix(Fake(changes={path: ['read']}), 'test-query', self.auth(), 'operator') def test_incomplete_matrix_refused(self): with self.assertRaises(m.Refused): m.checked_caps({'data': {m.DATA: ['read']}}, {m.DATA: ['read'], 'other': ['deny']}) def test_wrong_policy_default_identity_role_ttl_refused(self): for field, value in [('policies', ['root']), ('policies', [m.POLICY+'operator', 'default']), ('metadata', {'role': 'platform-admin'}), ('lease_duration', 301), ('client_token', 'invalid token')]: auth = self.auth() auth[field] = value with self.subTest(field=field, value=value), self.assertRaises(m.Refused): m.lane_matrix(Fake(), 'test-query', auth, 'operator') def test_wrong_eso_namespace_refused(self): auth = self.auth('eso') auth['metadata']['service_account_namespace'] = 'sso' with self.assertRaises(m.Refused): m.lane_matrix(Fake(), 'test-query', auth, 'eso') def test_login_failure_is_never_negative_acceptance(self): for err in [b'connection refused', b'HTTP 403 permission denied', b'user cancelled', b'claim "groups" does not match any associated bound claim values']: result, category = m.oidc({}, lambda *args: (1, b'', err)) self.assertIsNone(result) self.assertIn(category, ['unproved_login_failure', 'group_refusal_candidate']) self.assertNotIn('accepted', category) def test_no_store_and_fixed_native_role(self): calls = [] def runner(argv, env): calls.append(argv) return 1, b'', b'cancelled' m.oidc({}, runner) self.assertIn('-no-store', calls[0]) self.assertIn('role=' + m.OP_ROLE, calls[0]) def test_api_rejects_secret_gets_role_writes_and_generic_endpoints(self): api = m.Api() for endpoint in [m.DATA, m.DATA.replace('/data/', '/metadata/'), 'sys/policies/acl/new', 'auth/netkingdom/role/new', 'auth/token/create']: with self.subTest(endpoint=endpoint), self.assertRaises(m.Refused): api.call(endpoint, 'test-only') def test_credentials_not_in_child_env(self): import os from unittest.mock import patch with patch.dict(os.environ, {'BAO_TOKEN': 'test', 'VAULT_TOKEN': 'test', 'HTTPS_PROXY': 'https://wrong', 'BAO_NAMESPACE': 'other'}): env = m.safe_env('/tmp/test-only') for name in ['BAO_TOKEN', 'VAULT_TOKEN', 'HTTPS_PROXY', 'BAO_NAMESPACE']: self.assertNotIn(name, env) self.assertEqual(env['HOME'], '/tmp/test-only') def test_eso_cluster_mismatch_stops_before_tokenrequest(self): calls = [] def runner(argv, env): calls.append(argv) return 0, b'{"items": [{"status": {"addresses": [{"address": "10.0.0.1"}]}}]}', b'' with self.assertRaises(m.Refused): m.eso(Fake(), runner) self.assertEqual(len(calls), 1) def exercise_main(self, auth, failure=None, mode='operator', query_info=None, bad_revoke_response=None, login_error=None): import tempfile import os from unittest.mock import patch with tempfile.TemporaryDirectory() as directory: credential = Path(directory) / 'test-fixture' credential.write_text('test-only-query') token_fd = os.open(credential, os.O_RDONLY) receipt = Path(directory) / 'receipt.json' fake = Fake() if query_info is not None: fake.info.update(query_info) old_call = fake.call def call(endpoint, token='', body=None): if endpoint == 'auth/token/revoke-self' and bad_revoke_response is not None: native = NATIVE_API() native.opener = self.opener(bad_revoke_response) return native.call(endpoint, token, body) if endpoint == 'auth/token/revoke-self' and token == failure: raise m.Transport('test_cleanup_transport') return old_call(endpoint, token, body) fake.call = call argv = ['verifier', mode, '--survey-token-fd', str(token_fd), '--approval-reference', 'test-not-an-approval', '--approved-source-sha256', m.hashlib.sha256(SCRIPT.read_bytes()).hexdigest(), '--attended', '--receipt', str(receipt)] with patch.object(m.sys, 'argv', argv), patch.object(m.sys.stdin, 'isatty', return_value=True), \ patch.dict(os.environ, {}, clear=True), patch.object(m, 'Api', return_value=fake), \ patch.object(m, 'oidc', return_value=(auth, 'login_succeeded'), side_effect=login_error): code = m.main() return code, m.json.loads(receipt.read_text()), fake.calls def test_cleanup_attempted_even_when_policy_check_fails(self): auth = self.auth() auth['policies'] = ['root'] code, receipt, calls = self.exercise_main(auth) self.assertEqual(code, 1) self.assertTrue(receipt['lane_revoke']) self.assertTrue(receipt['query_revoke']) self.assertNotIn('test-only-not-a-credential', m.json.dumps(receipt)) def test_cleanup_transport_invalidates_positive_matrix(self): for token in ['test-only-query', 'test-only-not-a-credential']: code, receipt, calls = self.exercise_main(self.auth(), failure=token) self.assertEqual(code, 1) self.assertEqual(receipt['result'], 'unproved') self.assertEqual(len([c for c in calls if c[0] == 'auth/token/revoke-self']), 1) def test_unexpected_nonmember_success_is_revoked_and_fails(self): code, receipt, calls = self.exercise_main(self.auth(), mode='nonmember') self.assertEqual(code, 1) self.assertEqual(receipt['failure_class'], 'nonmember_obtained_token') self.assertTrue(receipt['lane_revoke']) self.assertEqual(receipt['owner_acceptance'], 'pending') def response(self, status=200, content='application/json', raw=b'{"data":{}}', length=None): from email.message import Message headers = Message() if content: headers['Content-Type'] = content if length is not None: headers['Content-Length'] = length class Response: def __enter__(self): return self def __exit__(self, *args): pass def read(self, count): return raw[:count] response = Response() response.status = status response.headers = headers return response def opener(self, response): from types import SimpleNamespace return SimpleNamespace(open=lambda *args, **kwargs: response) def test_mixed_login_error_preserves_recovered_token_cleanup(self): raw=m.json.dumps({'errors':['private-synthetic'], 'auth':self.auth()}).encode() with self.assertRaises(m.RecoveredLoginRefused) as error: m.oidc({},lambda *args:(0,raw,b'')) code, receipt, _ = self.exercise_main(self.auth(), login_error=error.exception) self.assertEqual(code,1) self.assertTrue(receipt['lane_revoke']) self.assertTrue(receipt['query_revoke']) self.assertNotIn('private-synthetic',m.json.dumps(receipt)) native=m.Api();native.opener=self.opener(self.response(raw=raw)) with self.assertRaises(m.RecoveredLoginRefused) as error: native.call('auth/kubernetes/login',body={}) self.assertEqual(error.exception.token,'test-only-not-a-credential') def test_endpoint_http_success_contracts(self): for endpoint, raw in [('auth/token/lookup-self', b'{"data":{}}'), ('sys/capabilities', b'{"data":{}}'), ('auth/kubernetes/login', b'{"auth":{}}')]: api = m.Api();api.opener = self.opener(self.response(raw=raw)) self.assertIsInstance(api.call(endpoint, 'test', {}), dict) api = m.Api();api.opener = self.opener(self.response(status=204, content='', raw=b'')) self.assertEqual(api.call('auth/token/revoke-self', 'test', {}), {}) def test_bad_metadata_status_content_size_shape_errors_refused(self): responses = [self.response(status=status) for status in [201,202,206,301,302,307,308]] responses += [self.response(content='text/plain'), self.response(length='65537'), self.response(length='-1'), self.response(raw=b'x'*65537), self.response(raw=b'private-synthetic-invalid-json'), self.response(raw=b'{"data":[]}'), self.response(raw=b'{}'), self.response(raw=b'{"errors":[],"data":{}}'), self.response(raw=b'{"errors":["private-synthetic"],"data":{}}')] for response in responses: api=m.Api();api.opener=self.opener(response) with self.subTest(response=response), self.assertRaises((m.Refused,m.Transport)): api.call('auth/token/lookup-self', 'synthetic') def test_revoke202_wrong_content_error_body_never_confirmed_in_receipt(self): responses = [self.response(status=status) for status in [200,201,202,206]] responses += [self.response(status=204, raw=b'error')] for response in responses: code, receipt, _ = self.exercise_main(self.auth(), bad_revoke_response=response) self.assertEqual(code, 1) self.assertEqual(receipt['result'], 'unproved') self.assertFalse(receipt['lane_revoke']) self.assertFalse(receipt['query_revoke']) def test_valid204_main_cleanup_confirmation(self): code, receipt, _ = self.exercise_main(self.auth(), bad_revoke_response=self.response(status=204, content='', raw=b'')) self.assertEqual(code, 0) self.assertTrue(receipt['lane_revoke']) self.assertTrue(receipt['query_revoke']) def test_missing_invalid_accessors_fail_with_recovered_token_cleanup(self): for accessor in [None, '', 123, ' ', 'x'*257]: auth=self.auth();auth['accessor']=accessor code, receipt, _ = self.exercise_main(auth) self.assertEqual(code, 1) self.assertTrue(receipt['lane_revoke']) self.assertTrue(receipt['query_revoke']) self.assertNotIn('lane_accessor_sha256',receipt) code, receipt, _ = self.exercise_main(self.auth(), query_info={'accessor':accessor}) self.assertEqual(code, 1) self.assertTrue(receipt['query_revoke']) self.assertNotIn('query_accessor_sha256',receipt) def test_ttl_boolean_zero_negative_overbudget_refused(self): for ttl in [True,False,0,-1,301,'300']: auth=self.auth();auth['lease_duration']=ttl with self.subTest(ttl=ttl), self.assertRaises(m.Refused): m.lane_matrix(Fake(), 'test-query', auth, 'operator') def test_approved_revision_mismatch_precedes_receipt_token_or_native(self): from unittest.mock import patch import os argv=['verifier','operator','--survey-token-fd','99','--approval-reference','test', '--approved-source-sha256','wrong','--attended','--receipt','/tmp/must-not-exist-mason-test'] with patch.object(m.sys,'argv',argv), patch.object(m.sys.stdin,'isatty',return_value=True), \ patch.dict(os.environ,{},clear=True), patch.object(m.os,'open',side_effect=AssertionError('no IO')): with self.assertRaises(SystemExit) as error: m.main() self.assertEqual(error.exception.code,2) def test_subprocess_capture_is_bounded_while_running(self): import sys with self.assertRaises(m.Refused): m.run([sys.executable,'-c', 'import sys; sys.stdout.write("x"*1000000)'], {'PATH':'/usr/bin:/bin'}) result=m.run([sys.executable,'-c','import sys; print("synthetic"); sys.stderr.write("synthetic error")'],{'PATH':'/usr/bin:/bin'}) self.assertEqual(result,(0,b'synthetic\n',b'synthetic error')) def test_real_relative_redirect_and_hostile_proxy_refused(self): import threading import os from unittest.mock import patch from http.server import BaseHTTPRequestHandler, HTTPServer calls=[] class Handler(BaseHTTPRequestHandler): def do_GET(self): calls.append(self.path) self.send_response(302);self.send_header('Location','/private');self.end_headers() def log_message(self,*args):pass server=HTTPServer(('127.0.0.1',0),Handler) thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() try: with patch.object(m,'BASE','http://127.0.0.1:'+str(server.server_port)), \ patch.dict(os.environ, {'HTTP_PROXY':'http://127.0.0.1:1','http_proxy':'http://127.0.0.1:1','NO_PROXY':'','no_proxy':''}): with self.assertRaises(m.Refused): m.Api().call('auth/token/lookup-self','synthetic-not-a-credential') finally: server.shutdown();server.server_close();thread.join() self.assertEqual(calls,['/v1/auth/token/lookup-self']) def test_redirect_refused(self): with self.assertRaises(m.Refused): m.NoRedirect().redirect_request(None, None, None, None, None, None) if __name__ == '__main__': unittest.main()