tegwick
e84b68db01
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
200 lines
10 KiB
Python
200 lines
10 KiB
Python
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import unittest
|
|
import tempfile
|
|
import http.server
|
|
import threading
|
|
import shutil
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('requester_metadata', Path(__file__).resolve().parents[1] / 'scripts/informed_decision_requester_metadata.py')
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
|
|
|
|
def role(**updates):
|
|
return dict(bound_service_account_names=['informed-decision-eso'],
|
|
bound_service_account_namespaces=['informed-decision'], audience='openbao',
|
|
token_policies=[m.POLICY], token_ttl=900, token_max_ttl=900,
|
|
token_explicit_max_ttl=900, token_no_default_policy=True, **updates)
|
|
|
|
|
|
class MetadataTests(unittest.TestCase):
|
|
def test_role_rejects_widening_and_unbounded_token(self):
|
|
original = role()
|
|
for key, value in [('bound_service_account_names', ['*']),
|
|
('bound_service_account_namespaces', ['*']),
|
|
('audience', 'other'), ('token_policies', [m.POLICY, 'default']),
|
|
('token_ttl', 3600), ('token_max_ttl', 0),
|
|
('token_explicit_max_ttl', 0), ('token_no_default_policy', 1),
|
|
('token_period', 900), ('bound_service_account_namespace_selector', '*')]:
|
|
with self.subTest(key=key):
|
|
self.assertFalse(m.role_matches(dict(original, **{key: value})))
|
|
self.assertTrue(m.role_matches(original))
|
|
|
|
def test_differing_role_refuses_before_missing_policy_write(self):
|
|
calls = []
|
|
def command(args, payload=None, **kwargs):
|
|
calls.append(args)
|
|
return None if args[-1].startswith('sys/') else {'data': dict(role(), audience='wrong')}
|
|
with self.assertRaises(m.Refused) as caught:
|
|
m.apply('exact policy', {}, command=command)
|
|
self.assertEqual(caught.exception.code, 3)
|
|
self.assertTrue(all(args[0] == 'read' for args in calls))
|
|
|
|
def test_differing_policy_refuses_all_writes(self):
|
|
calls = []
|
|
def command(args, payload=None, **kwargs):
|
|
calls.append(args)
|
|
return {'data': {'policy': 'different'}} if args[-1].startswith('sys/') else None
|
|
with self.assertRaises(m.Refused) as caught:
|
|
m.apply('exact policy', {}, command=command)
|
|
self.assertEqual(caught.exception.code, 3)
|
|
self.assertTrue(all(args[0] == 'read' for args in calls))
|
|
|
|
def test_identical_objects_noop_without_write(self):
|
|
calls = []
|
|
def command(args, payload=None, **kwargs):
|
|
calls.append(args)
|
|
return {'data': {'policy': 'exact policy'}} if args[-1].startswith('sys/') else {'data': role()}
|
|
m.apply('exact policy', {}, command=command)
|
|
self.assertEqual(len(calls), 4)
|
|
self.assertTrue(all(args[0] == 'read' for args in calls))
|
|
|
|
def test_new_objects_write_only_two_metadata_paths_then_verify(self):
|
|
calls = []
|
|
saved = {}
|
|
def command(args, payload=None, **kwargs):
|
|
calls.append(args)
|
|
path = args[2]
|
|
if args[0] == 'write':
|
|
saved[path] = json.loads(payload)
|
|
return {}
|
|
if path not in saved:
|
|
return None
|
|
return {'data': saved[path]} if path.startswith('sys/') else {'data': role()}
|
|
m.apply('exact policy', {'audience': 'openbao'}, command=command)
|
|
self.assertEqual(set(saved), {'sys/policies/acl/' + m.POLICY, m.ROLE})
|
|
self.assertEqual([args[0] for args in calls], ['read', 'read', 'write', 'write', 'read', 'read'])
|
|
|
|
def test_exact_legacy_policy_upgrade_preserves_role(self):
|
|
saved, writes = [m.LEGACY_POLICY], []
|
|
def command(args, payload=None, **kwargs):
|
|
if args[0] == 'write':
|
|
writes.append(args[2]); saved[0] = json.loads(payload)['policy']; return {}
|
|
return {'data': {'policy': saved[0]}} if args[-1].startswith('sys/') else {'data': role()}
|
|
m.apply(m.EXPECTED_POLICY, {}, command=command)
|
|
self.assertEqual(writes, ['sys/policies/acl/' + m.POLICY])
|
|
self.assertEqual(saved[0], m.EXPECTED_POLICY)
|
|
|
|
def test_legacy_policy_upgrade_refuses_different_role_before_write(self):
|
|
calls = []
|
|
def command(args, **kwargs):
|
|
calls.append(args)
|
|
return {'data': {'policy': m.LEGACY_POLICY}} if args[-1].startswith('sys/') else {'data': dict(role(), audience='wrong')}
|
|
with self.assertRaises(m.Refused): m.apply(m.EXPECTED_POLICY, {}, command=command)
|
|
self.assertTrue(all(args[0] == 'read' for args in calls))
|
|
|
|
def test_postwrite_role_drift_fails_verification(self):
|
|
calls = []
|
|
def command(args, payload=None, **kwargs):
|
|
calls.append(args)
|
|
if kwargs.get('missing_ok'): return None
|
|
if args[0] == 'write': return {}
|
|
return {'data': {'policy': 'exact policy'}} if args[-1].startswith('sys/') else {'data': dict(role(), audience='wrong')}
|
|
with self.assertRaises(m.Refused) as caught:
|
|
m.apply('exact policy', {}, command=command)
|
|
self.assertEqual(caught.exception.code, 5)
|
|
|
|
def test_404_only_is_absence_and_failures_are_silent(self):
|
|
for code, stderr, missing in [(2, ('No value found at '+m.ROLE+'\n').encode(), True), (2, b'Code: 404', False), (2, b'Code: 403 SECRET-SENTINEL', False), (1, ('No value found at '+m.ROLE).encode(), False), (2, b'No value found at wrong-path', False)]:
|
|
with self.subTest(code=code, stderr=stderr), patch.object(m.subprocess, 'run', return_value=subprocess.CompletedProcess([], code, b'', stderr)):
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
|
|
if missing: self.assertIsNone(m.cli(['read', '-format=json', m.ROLE], missing_ok=True))
|
|
else:
|
|
with self.assertRaises(m.Refused): m.cli(['read', '-format=json', m.ROLE], missing_ok=True)
|
|
self.assertEqual(output.getvalue(), '')
|
|
|
|
def test_empty_204_write_is_success(self):
|
|
with patch.object(m.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, b'', b'')):
|
|
self.assertEqual(m.cli(['write', '-format=json', m.ROLE, '-'], b'{}'), {})
|
|
|
|
def test_proposed_source_refuses_before_network(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
ccr = Path(tmp) / 'proposed.yaml'
|
|
ccr.write_text(m.CCR.read_text().replace('status: approved', 'status: proposed'))
|
|
digest = m.hashlib.sha256(ccr.read_bytes()).hexdigest()
|
|
with patch.object(m, 'CCR', ccr), patch.object(m.subprocess, 'run', side_effect=AssertionError('must remain offline')):
|
|
with self.assertRaises(m.Refused) as caught: m.source(digest)
|
|
self.assertEqual(caught.exception.code, 2)
|
|
|
|
@unittest.skipUnless(shutil.which('bao'), 'installed bao CLI required')
|
|
def test_installed_cli_missing_create_readback_and_idempotent_retry(self):
|
|
objects, writes = {}, []
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
value = objects.get(self.path)
|
|
self.send_response(200 if value else 404)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps({'data': value} if value else {'errors': []}).encode())
|
|
def do_PUT(self):
|
|
value = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
|
|
writes.append(self.path)
|
|
objects[self.path] = value if '/sys/' in self.path else role()
|
|
self.send_response(204)
|
|
self.end_headers()
|
|
do_POST = do_PUT
|
|
def log_message(self, *args): pass
|
|
server = http.server.HTTPServer(('127.0.0.1', 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {
|
|
'PATH': os.environ['PATH'], 'HOME': tmp,
|
|
'BAO_ADDR': f'http://127.0.0.1:{server.server_port}',
|
|
'BAO_TOKEN': 'synthetic-test-only',
|
|
}, clear=True):
|
|
m.apply('exact policy', {'audience': 'openbao'})
|
|
m.apply('exact policy', {'audience': 'openbao'})
|
|
self.assertEqual(writes, ['/v1/sys/policies/acl/' + m.POLICY, '/v1/' + m.ROLE])
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join()
|
|
|
|
def test_failed_step_receipt_contains_only_fixed_metadata(self):
|
|
with tempfile.TemporaryDirectory() as tmp, patch.object(m, 'ROOT', Path(tmp)), \
|
|
patch.object(m.sys, 'argv', ['child', '--expected-ccr-sha256', 'fixture']), \
|
|
patch.object(m.Path, 'home', return_value=Path('/tmp/.warden-attended-login/fixture')), \
|
|
patch.dict(os.environ, {}, clear=True), \
|
|
patch.object(m, 'source', return_value=('policy', {})), \
|
|
patch.object(m, 'cli', side_effect=m.Refused(4)):
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
|
|
self.assertEqual(m.main(), 4)
|
|
self.assertEqual(output.getvalue(), '')
|
|
receipt = json.loads(next((Path(tmp)/'docs/evidence/requester-metadata').glob('*.json')).read_text())
|
|
self.assertEqual(receipt['phase'], 'identity')
|
|
self.assertEqual(receipt['exit_code'], 4)
|
|
self.assertEqual(set(receipt), {'schema', 'ccr', 'phase', 'exit_code', 'native_delivery_verified', 'at'})
|
|
|
|
def test_missing_arguments_silent(self):
|
|
with tempfile.TemporaryDirectory() as tmp, patch.object(m, 'ROOT', Path(tmp)), patch.object(m.sys, 'argv', ['metadata-child']):
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
|
|
self.assertEqual(m.main(), 2)
|
|
self.assertEqual(output.getvalue(), '')
|
|
receipt = json.loads(next((Path(tmp)/'docs/evidence/requester-metadata').glob('*.json')).read_text())
|
|
self.assertEqual(receipt['phase'], 'arguments')
|
|
self.assertEqual(receipt['exit_code'], 2)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|