45 lines
2.5 KiB
Python
45 lines
2.5 KiB
Python
|
import copy
|
||
|
|
import importlib.util
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from unittest.mock import patch
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
ROOT=Path(__file__).resolve().parents[1]
|
||
|
|
spec=importlib.util.spec_from_file_location('clock_custody',ROOT/'scripts/railiance_clock_custody.py')
|
||
|
|
m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
|
||
|
|
|
||
|
|
class CustodyTests(unittest.TestCase):
|
||
|
|
def setUp(self):self.contract=yaml.safe_load(m.CCR.read_text())
|
||
|
|
def test_valid(self):self.assertEqual(m.validate_contract(self.contract),[])
|
||
|
|
def test_scope_and_approval_drift(self):
|
||
|
|
for change in [lambda c:c['delivery'].update(host='other'),lambda c:c['openbao'].update(runtime_access=True),lambda c:c['review'].update(comments=[]),lambda c:c['openbao'].update(initial_cas=1)]:
|
||
|
|
c=copy.deepcopy(self.contract);change(c);self.assertTrue(m.validate_contract(c))
|
||
|
|
def run_failure(self,reply,expected_stage):
|
||
|
|
with tempfile.TemporaryDirectory() as d:
|
||
|
|
out=Path(d)/'receipt.json'
|
||
|
|
with patch.object(sys,'argv',['helper','--receipt',str(out)]),patch.object(m,'bao',side_effect=reply),patch.object(m,'run_cmd') as cmd:
|
||
|
|
self.assertEqual(m.main(),1);cmd.assert_not_called()
|
||
|
|
receipt=json.loads(out.read_text());self.assertEqual(receipt['stage'],expected_stage)
|
||
|
|
self.assertNotIn('PRIVATE KEY',out.read_text())
|
||
|
|
def test_root_denied(self):
|
||
|
|
self.run_failure([{'data':{'policies':['root','platform-admin'],'path':'auth/netkingdom/login'}}],'attended_identity')
|
||
|
|
def test_wrong_auth_mount_denied(self):
|
||
|
|
self.run_failure([{'data':{'policies':['platform-admin'],'path':'auth/token/create'}}],'attended_identity')
|
||
|
|
def test_audit_required(self):
|
||
|
|
self.run_failure([{'data':{'policies':['platform-admin'],'path':'auth/netkingdom/login'}},{}],'attended_identity')
|
||
|
|
def test_cas_failure_prevents_delivery(self):
|
||
|
|
self.run_failure([{'data':{'policies':['platform-admin'],'path':'auth/netkingdom/login'}},{'file/':{}},m.CustodyError('contained_command_failed')],'custody')
|
||
|
|
def test_receipt_never_overwrites(self):
|
||
|
|
with tempfile.TemporaryDirectory() as d:
|
||
|
|
out=Path(d)/'receipt.json';out.write_text('previous evidence')
|
||
|
|
with patch.object(sys,'argv',['helper','--receipt',str(out)]),patch.object(m,'bao') as b:
|
||
|
|
with self.assertRaises(FileExistsError):m.main()
|
||
|
|
b.assert_not_called()
|
||
|
|
self.assertEqual(out.read_text(),'previous evidence')
|
||
|
|
|
||
|
|
if __name__=='__main__':unittest.main()
|