3759850cc4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
109 lines
5.4 KiB
Python
109 lines
5.4 KiB
Python
from __future__ import annotations
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
def load(name, filename):
|
|
spec = importlib.util.spec_from_file_location(name, ROOT / 'scripts' / filename)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
lane = load('preflight_lane', 'state_hub_preflight_lane.py')
|
|
cc = load('credential_change_preflight', 'credential-change.py')
|
|
|
|
class SigningLaneTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.ccr = cc.load_yaml(lane.CCR)
|
|
|
|
def test_data_only_policy_and_bounded_auth_survive_plan_generation(self):
|
|
self.assertNotIn('/metadata/', cc.generated_policy_hcl(self.ccr))
|
|
self.assertIn('path "auth/token/lookup-self"', cc.generated_policy_hcl(self.ccr))
|
|
self.assertIn('path "auth/token/revoke-self"', cc.generated_policy_hcl(self.ccr))
|
|
auth = cc.auth_payload(self.ccr)
|
|
self.assertEqual(auth['audience'], 'openbao')
|
|
self.assertEqual(auth['token_explicit_max_ttl'], '15m')
|
|
self.assertEqual(auth['token_max_ttl'], '15m')
|
|
self.assertIs(auth['token_no_default_policy'], True)
|
|
legacy = copy.deepcopy(self.ccr)
|
|
del legacy['openbao']['metadata_read']
|
|
self.assertIn('/metadata/', cc.generated_policy_hcl(legacy))
|
|
|
|
def test_malformed_security_options_rejected(self):
|
|
for section, key, value in [('openbao', 'metadata_read', 'false'),
|
|
('openbao', 'token_self_lifecycle', 'true'),
|
|
('auth', 'audience', ''),
|
|
('auth', 'token_max_ttl', 900),
|
|
('auth', 'token_no_default_policy', 'true')]:
|
|
ccr = copy.deepcopy(self.ccr)
|
|
target = ccr['openbao'] if section == 'openbao' else ccr['openbao']['auth']
|
|
target[key] = value
|
|
errors = []
|
|
cc.validate_workload_kv_read(ccr, errors, [])
|
|
self.assertTrue(any(key in error for error in errors), errors)
|
|
|
|
def test_proposed_ccr_cannot_reach_writer(self):
|
|
proposed = copy.deepcopy(self.ccr)
|
|
proposed['status'] = 'proposed'
|
|
import tempfile, yaml
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
file = Path(tmp) / 'ccr.yaml'
|
|
file.write_text(yaml.safe_dump(proposed))
|
|
with patch.object(lane, 'CCR', file):
|
|
with self.assertRaisesRegex(lane.LaneError, 'approved_ccr_required'):
|
|
lane.approved_contract()
|
|
|
|
def test_fence_rejects_running_terminating_or_autoscaled_api(self):
|
|
for replicas, pods, hpas in [(1, [], []), (0, [{'metadata': {'deletionTimestamp': 'now'}}], []),
|
|
(0, [], [{}])]:
|
|
responses = [SimpleNamespace(stdout=json.dumps(x).encode()) for x in
|
|
[{'spec': {'replicas': replicas}}, {'items': pods}, {'items': hpas}]]
|
|
with patch.object(lane, 'command', side_effect=responses):
|
|
with self.assertRaises(lane.LaneError):
|
|
lane.assert_fenced(['kubectl'])
|
|
|
|
def test_rotation_fence_failure_precedes_key_generation(self):
|
|
args = SimpleNamespace(action='rotate', expected_version=1, kubeconfig='/fixture')
|
|
result = SimpleNamespace(stdout=json.dumps({'data': {'policies': ['platform-admin']}}).encode())
|
|
with patch.object(lane, 'approved_contract', return_value=({}, '')), \
|
|
patch.object(lane, 'assert_cluster'), \
|
|
patch.object(lane, 'bao', return_value=result), \
|
|
patch.object(lane, 'assert_fenced', side_effect=lane.LaneError('fence')), \
|
|
patch.object(lane.secrets, 'token_hex') as generate:
|
|
with self.assertRaises(lane.LaneError):
|
|
lane.run(args, {})
|
|
generate.assert_not_called()
|
|
|
|
def test_wrong_cluster_precedes_any_openbao_access_or_generation(self):
|
|
args = SimpleNamespace(action='provision', expected_version=0, kubeconfig='/fixture')
|
|
response = SimpleNamespace(stdout=json.dumps({'metadata': {'uid': 'other-cluster'}}).encode())
|
|
with patch.object(lane, 'approved_contract', return_value=({}, '')), \
|
|
patch.object(lane, 'command', return_value=response), \
|
|
patch.object(lane, 'bao') as access, \
|
|
patch.object(lane.secrets, 'token_hex') as generate:
|
|
with self.assertRaisesRegex(lane.LaneError, 'primary_cluster_identity_mismatch'):
|
|
lane.run(args, {})
|
|
access.assert_not_called()
|
|
generate.assert_not_called()
|
|
|
|
def test_unapproved_key_format_or_policy_does_not_pass_contract(self):
|
|
import tempfile, yaml
|
|
ccr = copy.deepcopy(self.ccr)
|
|
ccr['status'] = 'approved'
|
|
ccr['access_frontdoor']['resolvable'] = False
|
|
ccr['openbao']['auth']['bound_claims_confirmed'] = True
|
|
ccr['openbao']['auth']['audience'] = 'kubernetes'
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
file = Path(tmp) / 'ccr.yaml'; file.write_text(yaml.safe_dump(ccr))
|
|
with patch.object(lane, 'CCR', file):
|
|
with self.assertRaisesRegex(lane.LaneError, 'exact_contract_required'):
|
|
lane.approved_contract()
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|