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()