0e98465c6e
The reuse-surface recovery lane policy carries the exact extra data path approved under CCR-2026-0046; the test predated it and required exactly three paths. It now allows one exact data path per approved CCR naming the same policy file and keeps the wildcard, metadata and list bans. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 509845@bnt-lap001 Assistant-Session: ed7a7d95-d8eb-455c-b594-963fb1830f76
75 lines
3.7 KiB
Python
75 lines
3.7 KiB
Python
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / 'scripts'))
|
|
import repair_eso_kubernetes_auth as repair
|
|
|
|
|
|
APPROVED_CCR_STATUSES = {'approved', 'applied', 'verified', 'active'}
|
|
|
|
|
|
def approved_extension_paths(policy_file, lane_kv_path):
|
|
"""Exact data paths that approved CCRs add to a recovery lane's policy file."""
|
|
paths = []
|
|
for ccr_path in sorted((ROOT / 'credential-change-requests').glob('CCR-*.yaml')):
|
|
ccr = yaml.safe_load(ccr_path.read_text())
|
|
openbao = ccr.get('openbao') or {}
|
|
if ccr.get('status') not in APPROVED_CCR_STATUSES or openbao.get('policy_file') != policy_file:
|
|
continue
|
|
mount = openbao['mount']
|
|
data_path = mount + '/data/' + openbao['kv_path'][len(mount) + 1:]
|
|
if data_path != lane_kv_path:
|
|
paths.append(data_path)
|
|
return paths
|
|
|
|
|
|
class EsoRecoveryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.lane = json.loads((ROOT / 'openbao/eso-auth-recovery/lanes.json').read_text())[0]
|
|
|
|
def test_role_rejects_wildcard_namespace_added_policy_and_unbounded_ttl(self):
|
|
native = {k:900 if v == '15m' else v for k,v in repair.role_payload(self.lane).items()}
|
|
repair.check_role(native, self.lane)
|
|
for key,value in [('bound_service_account_namespaces',['*']),('token_policies',['root']),('token_explicit_max_ttl',0),('audience','other')]:
|
|
with self.assertRaises(repair.LaneError):
|
|
repair.check_role(dict(native, **{key:value}), self.lane)
|
|
|
|
def test_custody_disagreement_stops_before_serviceaccount_or_auth_write(self):
|
|
identity = SimpleNamespace(stdout=json.dumps({'data':{'policies':['platform-admin']}}).encode())
|
|
with patch.object(repair,'assert_cluster'), patch.object(repair,'bao',return_value=identity) as bao, patch.object(repair,'command') as kube, patch.object(repair,'compare_custody',side_effect=repair.LaneError('custody_disagrees')):
|
|
with self.assertRaises(repair.LaneError):
|
|
repair.run(SimpleNamespace(action='apply',kubeconfig='/fixture'), {'lanes':[]})
|
|
kube.assert_not_called()
|
|
self.assertEqual(bao.call_count,1)
|
|
|
|
def test_no_credential_values_or_parent_access_in_generated_policies(self):
|
|
for lane in json.loads((ROOT / 'openbao/eso-auth-recovery/lanes.json').read_text()):
|
|
policy_file='openbao/policies/'+lane['policy']+'.hcl'
|
|
policy=(ROOT / policy_file).read_text()
|
|
self.assertNotIn('*',policy)
|
|
self.assertNotIn('/metadata/',policy)
|
|
self.assertNotIn('"list"',policy)
|
|
# Lane path plus token lookup-self/revoke-self, plus one exact data path
|
|
# per approved CCR that extends this policy file (e.g. CCR-2026-0046).
|
|
extra=approved_extension_paths(policy_file, lane['kv_path'])
|
|
self.assertEqual(policy.count('path "'),3+len(extra))
|
|
for path in [lane['kv_path'], 'auth/token/lookup-self', 'auth/token/revoke-self', *extra]:
|
|
self.assertIn('path "'+path+'"',policy)
|
|
|
|
def test_expired_token_response_is_distinct_from_authority_failure(self):
|
|
for code in [400,403]:
|
|
self.assertTrue(repair.invalid_token_lookup(SimpleNamespace(returncode=2,stderr=f'Code: {code}. Errors: bad token'.encode())))
|
|
for error in [b'Code: 403. permission denied',b'Code: 500. bad token',b'connection refused']:
|
|
self.assertFalse(repair.invalid_token_lookup(SimpleNamespace(returncode=2,stderr=error)))
|
|
|
|
|
|
if __name__=='__main__':
|
|
unittest.main()
|