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()