f5b7ff5118
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a10391-6355-7b41-b0fa-3e14e874f091
203 lines
13 KiB
Python
203 lines
13 KiB
Python
"""Draft-only offline exact-path lane scaffolding; no execution surface."""
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from datetime import date
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
|
|
import jsonschema
|
|
import yaml
|
|
|
|
REPO_DIR = Path(__file__).resolve().parents[1]
|
|
MAX_INPUT_BYTES = 128 * 1024
|
|
SECRET_MARKERS = ('AGE-SECRET-KEY-1', '-----BEGIN', 'hvs.', 'hvb.', 'hvc.', 'ghp_', 'github_pat_', 'npm_', 'BAO_TOKEN=', 'VAULT_TOKEN=', 'OPENBAO_ROOT_TOKEN=')
|
|
|
|
|
|
class UniqueLoader(yaml.SafeLoader):
|
|
pass
|
|
|
|
|
|
def unique_mapping(loader, node):
|
|
result = {}
|
|
for key_node, value_node in node.value:
|
|
key = loader.construct_object(key_node)
|
|
if not isinstance(key, str) or key in result:
|
|
raise ValueError('Invalid or duplicate metadata key')
|
|
result[key] = loader.construct_object(value_node)
|
|
return result
|
|
|
|
|
|
UniqueLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, unique_mapping)
|
|
|
|
|
|
def read_metadata(path):
|
|
with Path(path).open('rb') as stream:
|
|
raw = stream.read(MAX_INPUT_BYTES + 1)
|
|
if len(raw) > MAX_INPUT_BYTES:
|
|
raise ValueError('Metadata input exceeds size limit')
|
|
text = raw.decode('utf-8')
|
|
if any(marker in text for marker in SECRET_MARKERS) or re.search(r'(?<![A-Za-z])sk-[A-Za-z0-9]', text):
|
|
raise ValueError('Metadata input contains rejected secret material')
|
|
try:
|
|
data = yaml.load(text, Loader=UniqueLoader)
|
|
canonical = json.dumps(data, allow_nan=False)
|
|
if any(marker in canonical for marker in SECRET_MARKERS) or re.search(r"(?<![A-Za-z])sk-[A-Za-z0-9]", canonical):
|
|
raise ValueError("Rejected parsed secret material")
|
|
except (yaml.YAMLError, TypeError, RecursionError, ValueError) as exc:
|
|
raise ValueError('Invalid metadata document') from None
|
|
return data, hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def validate_request(data):
|
|
schema = json.loads((REPO_DIR / 'schemas/catalog-lane-scaffold.schema.json').read_text())
|
|
if not jsonschema.Draft202012Validator(schema).is_valid(data):
|
|
raise ValueError('Request does not match closed platform scaffold schema v1')
|
|
def text_values(value):
|
|
if isinstance(value, dict):
|
|
return [item for nested in value.values() for item in text_values(nested)]
|
|
if isinstance(value, list):
|
|
return [item for nested in value for item in text_values(nested)]
|
|
return [value] if isinstance(value, str) else []
|
|
if any(any(ord(char) < 32 for char in value) for value in text_values(data)):
|
|
raise ValueError('Control characters are not supported in scaffold metadata')
|
|
date.fromisoformat(data['date'])
|
|
bao = data['openbao']
|
|
if not bao['kv_path'].startswith(bao['mount'] + '/'):
|
|
raise ValueError('KV path must use the explicitly declared mount')
|
|
auth = bao['auth']
|
|
units = {'s': 1, 'm': 60, 'h': 3600}
|
|
ttl = auth['ttl']
|
|
if int(ttl[:-1]) * units[ttl[-1]] > 900:
|
|
raise ValueError('Prospective Kubernetes token TTL exceeds fifteen minutes')
|
|
eso = data['eso']
|
|
if len(auth['service_account_names']) != 1 or len(auth['service_account_namespaces']) != 1:
|
|
raise ValueError('One explicit ESO auth identity is supported')
|
|
keys = [item['secret_key'] for item in eso['data']]
|
|
properties = [item['property'] for item in eso['data']]
|
|
if len(set(keys)) != len(keys) or len(set(properties)) != len(properties) or set(properties) != set(bao['fields']):
|
|
raise ValueError('ESO mappings must cover the explicit field set exactly once')
|
|
|
|
|
|
def verify_catalog_join(request, catalog):
|
|
# Warden validates its own full input; platform checks only the shared seam.
|
|
if not isinstance(catalog, dict):
|
|
raise ValueError('Catalog pointer manifest must be an object')
|
|
for field in ('schema_version', 'lane_id', 'title', 'consumer_repo', 'owner_repo'):
|
|
if catalog.get(field) != request[field]:
|
|
raise ValueError('Cross-owner fixture identity mismatch')
|
|
if catalog.get('risk') != request['risk']['classification']:
|
|
raise ValueError('Cross-owner risk mismatch')
|
|
if catalog.get('canon_ref') != 'railiance-platform:credential-change-requests/' + request['ccr_id'] + '.yaml':
|
|
raise ValueError('Catalog CCR destination pointer mismatch')
|
|
ref = catalog.get('platform_request_ref')
|
|
if not isinstance(ref, str) or not re.fullmatch(r'railiance-platform:[a-zA-Z0-9_./-]+', ref) or '..' in ref:
|
|
raise ValueError('Catalog needs a non-secret owner request pointer')
|
|
|
|
|
|
def render_artifacts(data, policy_renderer, auth_renderer, request_digest, catalog_digest=None):
|
|
validate_request(data)
|
|
data = deepcopy(data)
|
|
bao = data['openbao']
|
|
auth = bao['auth']
|
|
lane = data['lane_id']
|
|
policy_file = 'openbao/policies/' + bao['policy_name'] + '.hcl'
|
|
ccr = {'id': data['ccr_id'], 'kind': 'credential-change-request', 'schema_version': 1,
|
|
'request_type': 'workload-kv-read', 'title': data['title'], 'status': 'proposed',
|
|
'created': data['date'], 'updated': data['date'], 'requester': data['requester'],
|
|
'review': {'required': True, 'required_approvers': data['review_approvers'], 'comments': []},
|
|
'target': data['target'],
|
|
'openbao': {'mount': bao['mount'], 'kv_path': bao['kv_path'], 'fields': bao['fields'],
|
|
'policy_name': bao['policy_name'], 'policy_file': policy_file, 'metadata_read': False,
|
|
'auth': {'method': 'kubernetes', 'mount': auth['mount'], 'role': auth['role'],
|
|
'audience': auth['audience'], 'ttl': auth['ttl'],
|
|
'token_max_ttl': auth['ttl'], 'token_explicit_max_ttl': auth['ttl'],
|
|
'token_no_default_policy': True, 'policies': [bao['policy_name']],
|
|
'bound_claims': {'service_account_names': auth['service_account_names'],
|
|
'service_account_namespaces': auth['service_account_namespaces']},
|
|
'bound_claims_confirmed': False}},
|
|
'access_frontdoor': {'type': 'external-secrets', 'catalog_id': lane,
|
|
'readiness': 'pending-review', 'resolvable': False},
|
|
'risk': data['risk'],
|
|
'verification': {'positive': ['Owner-reviewed native exact identity and exact path proof required'],
|
|
'negative': ['Owner-reviewed wrong identity, sibling GET and parent LIST denials required'],
|
|
'activation_conditions': ['Exact CCR approval, confirmed identity, attended custody and native receipt acceptance required']},
|
|
'lifecycle': {'deactivate': 'Owner-reviewed delivery withdrawal and token revocation; exact rollback receipt required',
|
|
'rotate': 'Separate version-guarded owner approval and contained value delivery required',
|
|
'compromised': 'Owner incident process; no automatic read, reseed or cleanup'},
|
|
'scaffold': {'draft_only': True, 'consumer_repo': data['consumer_repo'],
|
|
'owner_repo': data['owner_repo'], 'request_sha256': request_digest,
|
|
'catalog_request_sha256': catalog_digest}}
|
|
annotation = {'railiance.io/credential-change': data['ccr_id'], 'railiance.io/admission': 'pending-review'}
|
|
eso = data['eso']
|
|
store = {'apiVersion': 'external-secrets.io/v1', 'kind': 'ClusterSecretStore',
|
|
'metadata': {'name': eso['store_name'], 'annotations': deepcopy(annotation)},
|
|
'spec': {'provider': {'vault': {'server': eso['server'], 'path': bao['mount'], 'version': 'v2',
|
|
'auth': {'kubernetes': {'mountPath': auth['mount'], 'role': auth['role'],
|
|
'audiences': [auth['audience']],
|
|
'serviceAccountRef': {'name': auth['service_account_names'][0],
|
|
'namespace': auth['service_account_namespaces'][0]}}}}},
|
|
'conditions': [{'namespaces': [eso['namespace']]}]}}
|
|
secret = {'apiVersion': 'external-secrets.io/v1', 'kind': 'ExternalSecret',
|
|
'metadata': {'name': eso['external_secret_name'], 'namespace': eso['namespace'], 'annotations': deepcopy(annotation)},
|
|
'spec': {'refreshInterval': eso['refresh_interval'],
|
|
'secretStoreRef': {'kind': 'ClusterSecretStore', 'name': eso['store_name']},
|
|
'target': {'name': eso['secret_name'], 'creationPolicy': 'Owner', 'deletionPolicy': 'Retain',
|
|
'template': {'metadata': {'annotations': deepcopy(annotation)}}},
|
|
'data': [{'secretKey': item['secret_key'],
|
|
'remoteRef': {'key': bao['kv_path'][len(bao['mount']) + 1:], 'property': item['property']}}
|
|
for item in eso['data']]}}
|
|
evidence = {'schema_version': 1, 'lane_id': lane, 'ccr_id': data['ccr_id'], 'status': 'not-executed',
|
|
'draft_only': True, 'verified': False, 'request_sha256': request_digest,
|
|
'required_owner_checks': ['Exact revision/digest and required owner reviews',
|
|
'Identity/auth claims confirmed separately; no inferred workload identity',
|
|
'Metadata/capability readback through reviewed output-suppressing owner tool',
|
|
'No Secret bodies, values or raw subprocess errors in evidence',
|
|
'Native scoped positive/negative, cleanup/revocation and rollback receipts after approval']}
|
|
artifacts = {'credential-change-requests/' + data['ccr_id'] + '.yaml': yaml.safe_dump(ccr, sort_keys=False),
|
|
policy_file: '# DRAFT ONLY: exact CCR review and confirmed identity required.\n' + policy_renderer(ccr),
|
|
'openbao/auth/' + auth['role'] + '.draft.json': json.dumps({'draft_only': True, 'status': 'proposed',
|
|
'bound_claims_confirmed': False,
|
|
'auth_payload': auth_renderer(ccr)}, indent=2) + '\n',
|
|
'manifests/' + lane + '.draft.yaml': '# DRAFT REVIEW ARTIFACT. Do not apply. No custody/admission supplied.\n' + yaml.safe_dump_all([store, secret], sort_keys=False),
|
|
'docs/evidence/' + lane + '.checklist.json': json.dumps(evidence, indent=2) + '\n',
|
|
'README.md': 'Draft-only review bundle. No admission, approval, confirmed identity, credential value or executed receipt.\n'
|
|
'Review the exact request digest and owner destinations before promoting any file.\n'
|
|
'ESO target metadata is explicit to contain source last-applied annotations.\n'}
|
|
return artifacts
|
|
|
|
|
|
def scaffold(input_path, output_dir, policy_renderer, auth_renderer, catalog_manifest=None):
|
|
data, digest = read_metadata(input_path)
|
|
catalog_digest = None
|
|
if catalog_manifest is not None:
|
|
catalog, catalog_digest = read_metadata(catalog_manifest)
|
|
validate_request(data)
|
|
verify_catalog_join(data, catalog)
|
|
artifacts = render_artifacts(data, policy_renderer, auth_renderer, digest, catalog_digest)
|
|
output = Path(output_dir).absolute()
|
|
if output.is_symlink() or output.exists() or output.resolve().is_relative_to(REPO_DIR.resolve()):
|
|
raise ValueError('Output must be a new staging directory outside the owner repository')
|
|
if any((p / '.git').exists() for p in output.resolve().parents if p != Path(tempfile.gettempdir()).resolve()):
|
|
raise ValueError('Output staging must not be inside a repository')
|
|
output.mkdir(mode=0o700, parents=False)
|
|
for relative, content in artifacts.items():
|
|
path = output / relative
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open('x', encoding='utf-8') as stream:
|
|
stream.write(content)
|
|
return artifacts
|
|
|
|
|
|
def command_scaffold(args, policy_renderer, auth_renderer):
|
|
try:
|
|
artifacts = scaffold(args.input, args.output_dir, policy_renderer, auth_renderer, args.catalog_manifest)
|
|
except (ValueError, OSError, UnicodeError, RecursionError):
|
|
print('Scaffold refused: invalid metadata, mismatched owner pointers, or unsafe staging output.', file=sys.stderr)
|
|
return 2
|
|
print('Prepared ' + str(len(artifacts)) + ' draft review artifacts; nothing admitted or executed.')
|
|
return 0
|