f5b7ff5118
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a10391-6355-7b41-b0fa-3e14e874f091
150 lines
8.4 KiB
Python
150 lines
8.4 KiB
Python
import argparse
|
|
from copy import deepcopy
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import urllib.request
|
|
|
|
import yaml
|
|
|
|
REPO_DIR = Path(__file__).resolve().parents[1]
|
|
SCRIPTS = REPO_DIR / 'scripts'
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
import catalog_lane_scaffold as scaffold
|
|
SPEC = importlib.util.spec_from_file_location('ccr_scaffold_helpers', SCRIPTS / 'credential-change.py')
|
|
ccr = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(ccr)
|
|
|
|
|
|
class ScaffoldTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.request = REPO_DIR / 'examples/catalog-lane-scaffold/platform-request.yaml'
|
|
self.catalog = REPO_DIR / 'examples/catalog-lane-scaffold/warden-request.yaml'
|
|
self.data, self.digest = scaffold.read_metadata(self.request)
|
|
|
|
def render(self, data=None):
|
|
return scaffold.render_artifacts(data or self.data, ccr.generated_policy_hcl, ccr.auth_payload, self.digest)
|
|
|
|
def write_request(self, data):
|
|
path = self.root / 'request.yaml'
|
|
path.write_text(yaml.safe_dump(data))
|
|
return path
|
|
|
|
def test_existing_ccr_validator_accepts_only_proposed_unconfirmed_draft(self):
|
|
output = self.root / 'bundle'
|
|
scaffold.scaffold(self.request, output, ccr.generated_policy_hcl, ccr.auth_payload, self.catalog)
|
|
path = output / 'credential-change-requests/CCR-2099-9001.yaml'
|
|
with patch.object(ccr, 'REPO_DIR', output):
|
|
data, errors, warnings = ccr.validate_ccr(path)
|
|
self.assertEqual(errors, [])
|
|
self.assertTrue(any('not confirmed' in w for w in warnings))
|
|
self.assertEqual(data['status'], 'proposed')
|
|
self.assertFalse(data['access_frontdoor']['resolvable'])
|
|
self.assertFalse(data['openbao']['auth']['bound_claims_confirmed'])
|
|
self.assertEqual(data['review']['comments'], [])
|
|
self.assertTrue(ccr.applier_readiness_blockers(data))
|
|
self.assertEqual(set(ccr.auth_payload(data)['bound_service_account_names']), {'example-consumer-eso'})
|
|
|
|
def test_policy_and_eso_exact_scope_annotation_containment_and_no_values(self):
|
|
artifacts = self.render()
|
|
policy = artifacts['openbao/policies/workload-kv-read-example-consumer.hcl']
|
|
self.assertIn('path "platform/data/workloads/example-consumer/config"', policy)
|
|
self.assertEqual(policy.count('path "'), 1)
|
|
self.assertNotIn('metadata/', policy)
|
|
store, secret = list(yaml.safe_load_all(artifacts['manifests/example-workload-kv.draft.yaml']))
|
|
self.assertEqual(store['spec']['conditions'], [{'namespaces': ['example-consumer']}])
|
|
self.assertEqual(store['spec']['provider']['vault']['auth']['kubernetes']['audiences'], ['openbao'])
|
|
target = secret['spec']['target']
|
|
self.assertEqual(target['template']['metadata']['annotations']['railiance.io/admission'], 'pending-review')
|
|
self.assertNotIn('kubectl.kubernetes.io/last-applied-configuration', json.dumps(target))
|
|
self.assertEqual(secret['spec']['data'], [{'secretKey': 'client-secret', 'remoteRef': {'key': 'workloads/example-consumer/config', 'property': 'CLIENT_SECRET'}}])
|
|
self.assertNotIn('dataFrom', secret['spec'])
|
|
self.assertNotIn('Secret', [store['kind'], secret['kind']])
|
|
evidence = json.loads(artifacts['docs/evidence/example-workload-kv.checklist.json'])
|
|
self.assertFalse(evidence['verified'])
|
|
self.assertEqual(evidence['status'], 'not-executed')
|
|
|
|
def test_closed_schema_and_authority_injection_refused(self):
|
|
for location, key, value in [('top','status','approved'),('auth','bound_claims_confirmed',True),('top','secret_values',{'CLIENT_SECRET':'not-a-value'})]:
|
|
data = deepcopy(self.data)
|
|
target = data if location == 'top' else data['openbao']['auth']
|
|
target[key] = value
|
|
with self.assertRaises(ValueError): self.render(data)
|
|
|
|
def test_missing_identity_and_risk_are_not_inferred(self):
|
|
for location, key in [('auth','service_account_names'),('top','risk')]:
|
|
data = deepcopy(self.data)
|
|
target = data if location == 'top' else data['openbao']['auth']
|
|
del target[key]
|
|
with self.assertRaises(ValueError): self.render(data)
|
|
|
|
def test_wildcards_traversal_and_hcl_injection_refused(self):
|
|
for path in ['platform/workloads/*','platform/../other','platform/workloads/config\n','platform/a/" { capabilities = ["sudo"] }']:
|
|
data = deepcopy(self.data);data['openbao']['kv_path'] = path
|
|
with self.assertRaises(ValueError): self.render(data)
|
|
|
|
def test_wrong_mount_and_overlong_ttl_refused(self):
|
|
data = deepcopy(self.data);data['openbao']['mount'] = 'other'
|
|
with self.assertRaises(ValueError): self.render(data)
|
|
data = deepcopy(self.data);data['openbao']['auth']['ttl'] = '1h'
|
|
with self.assertRaises(ValueError): self.render(data)
|
|
|
|
def test_field_mapping_omissions_duplicates_and_extra_fields_refused(self):
|
|
for mapping in [[], [{'secret_key':'one','property':'OTHER'}], [{'secret_key':'a','property':'CLIENT_SECRET'},{'secret_key':'b','property':'CLIENT_SECRET'}]]:
|
|
data = deepcopy(self.data);data['eso']['data'] = mapping
|
|
with self.assertRaises(ValueError): self.render(data)
|
|
|
|
def test_cross_owner_pointer_fixture_and_unknown_identity_preserved(self):
|
|
catalog, _ = scaffold.read_metadata(self.catalog)
|
|
scaffold.verify_catalog_join(self.data, catalog)
|
|
self.assertIn('unknown_reason', catalog['workload_ref'])
|
|
for key,value in [('lane_id','other'),('risk','high'),('canon_ref','railiance-platform:credential-change-requests/OTHER.yaml'),('platform_request_ref','other-owner:request.yaml')]:
|
|
bad=deepcopy(catalog);bad[key]=value
|
|
with self.assertRaises(ValueError): scaffold.verify_catalog_join(self.data,bad)
|
|
|
|
def test_canonical_outputs_deterministic_and_input_immutable(self):
|
|
before = self.request.read_bytes();data_before=deepcopy(self.data)
|
|
self.assertEqual(self.render(), self.render())
|
|
self.assertEqual(self.data,data_before)
|
|
self.assertEqual(self.request.read_bytes(),before)
|
|
|
|
def test_collision_and_repo_output_refused_without_overwrite(self):
|
|
output=self.root/'occupied';output.mkdir();sentinel=output/'sentinel';sentinel.write_text('keep')
|
|
with self.assertRaises(ValueError): scaffold.scaffold(self.request,output,ccr.generated_policy_hcl,ccr.auth_payload)
|
|
self.assertEqual(sentinel.read_text(),'keep')
|
|
with self.assertRaises(ValueError): scaffold.scaffold(self.request,REPO_DIR/'live-output',ccr.generated_policy_hcl,ccr.auth_payload)
|
|
|
|
def test_no_network_or_subprocess_in_generation(self):
|
|
with patch.object(urllib.request,'urlopen',side_effect=AssertionError('network forbidden')),patch.object(subprocess,'run',side_effect=AssertionError('execution forbidden')):
|
|
artifacts=scaffold.scaffold(self.request,self.root/'isolated',ccr.generated_policy_hcl,ccr.auth_payload,self.catalog)
|
|
self.assertEqual(len(artifacts),6)
|
|
|
|
def test_duplicate_yaml_and_secret_material_refused_without_echo(self):
|
|
for text in ['lane_id: a\nlane_id: b\n', 'title: hvs.CANARY_NEVER_ECHO\n', 'title: "\\x68vs.CANARY_NEVER_ECHO"\n']:
|
|
path=self.root/'bad.yaml';path.write_text(text)
|
|
args=argparse.Namespace(input=path,output_dir=self.root/'bad-output',catalog_manifest=None)
|
|
stream=io.StringIO()
|
|
with patch('sys.stderr',stream): self.assertEqual(scaffold.command_scaffold(args,ccr.generated_policy_hcl,ccr.auth_payload),2)
|
|
self.assertNotIn('CANARY',stream.getvalue())
|
|
self.assertFalse((self.root/'bad-output').exists())
|
|
|
|
def test_oversized_input_refused(self):
|
|
path=self.root/'big.yaml';path.write_text('a'*(scaffold.MAX_INPUT_BYTES+1))
|
|
with self.assertRaises(ValueError): scaffold.read_metadata(path)
|
|
|
|
def test_schema_rejects_boolean_version_and_nonexistent_calendar_day(self):
|
|
for version,day in [(True,'2026-10-04'),(1,'2026-02-31')]:
|
|
data=deepcopy(self.data);data['schema_version']=version;data['date']=day
|
|
with self.assertRaises(ValueError): self.render(data)
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|