3759850cc4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
50 lines
2.6 KiB
Python
50 lines
2.6 KiB
Python
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / 'scripts'))
|
|
import repair_eso_kubernetes_auth as repair
|
|
|
|
|
|
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=(ROOT / 'openbao/policies' / (lane['policy']+'.hcl')).read_text()
|
|
self.assertNotIn('*',policy)
|
|
self.assertNotIn('/metadata/',policy)
|
|
self.assertNotIn('"list"',policy)
|
|
self.assertEqual(policy.count('path "'),3)
|
|
self.assertIn('path "'+lane['kv_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()
|