tegwick
969fe218df
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
298 lines
17 KiB
Python
298 lines
17 KiB
Python
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()
|