policy-nexus/tools/guardrails.py
tegwick f784f017b0
Some checks failed
Build and publish policy-nexus image / build-and-push (push) Failing after 1m21s
Publish canon-derived build-mode guardrail source and evaluator
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
2026-10-11 21:30:47 +02:00

98 lines
4.5 KiB
Python

#!/usr/bin/env python3
"""Publish and evaluate the restricted machine-checkable canon rule, fail closed.
This deliberately accepts only scalar mappings and block lists used by the canon
contract. Unsupported YAML, unknown categories and malformed records never grant
execution. Lane classification is not a credential or runtime approval token.
"""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import re
from render import split_frontmatter
def extract_rule(source: str) -> dict:
_meta, body = split_frontmatter(source)
blocks = re.findall(r'^```yaml\n(.*?)^```\s*$', body, re.M | re.S)
candidates = [b for b in blocks if re.search(r'^rule: build-mode-change-authorization\s*$', b, re.M)]
if len(candidates) != 1:
raise ValueError('expected exactly one build-mode authorization YAML block')
rule = {}
current = None
for line in candidates[0].splitlines():
text = line.split('#', 1)[0].rstrip()
if not text.strip():
continue
item = re.fullmatch(r' - ([a-z][a-z0-9_-]*)', text)
if item and current and isinstance(rule[current], list):
rule[current].append(item[1])
continue
field = re.fullmatch(r'([a-z][a-z0-9_]*):(?: ([a-z0-9_.-]+))?', text)
if not field or field[1] in rule:
raise ValueError('unsupported or duplicate machine rule YAML field')
current = field[1]
value = field[2]
rule[current] = [] if value is None else float(value) if re.fullmatch(r'\d+\.\d+', value) else value
validate_rule(rule)
return rule
def validate_rule(rule: dict) -> None:
fields = {'rule', 'version', 'applies_while', 'blue_lane_requires', 'red_lane_categories', 'orange_lane_categories'}
if not isinstance(rule, dict) or set(rule) != fields:
raise ValueError('machine rule fields differ from supported contract')
if (rule['rule'] != 'build-mode-change-authorization' or rule['version'] != 0.1
or rule['applies_while'] != 'build-mode'):
raise ValueError('unsupported machine rule identity, version or lifetime')
for name in ('blue_lane_requires', 'red_lane_categories', 'orange_lane_categories'):
values = rule[name]
if (not isinstance(values, list) or not values
or any(not isinstance(v, str) or not re.fullmatch(r'[a-z][a-z0-9_-]*', v) for v in values)
or len(values) != len(set(values))):
raise ValueError(f'invalid {name}')
if set(rule['red_lane_categories']) & set(rule['orange_lane_categories']):
raise ValueError('lane categories overlap')
def evaluate(rule: dict, record: dict, *, build_mode: bool) -> dict:
validate_rule(rule)
if not isinstance(record, dict):
raise ValueError('change record must be an object')
if not build_mode:
return {'lane': 'incomplete', 'reason': 'standing grant expired or build mode unconfirmed'}
categories = record.get('categories')
if not isinstance(categories, list) or any(not isinstance(c, str) for c in categories):
return {'lane': 'incomplete', 'reason': 'categories must be an explicit list'}
known = set(rule['red_lane_categories']) | set(rule['orange_lane_categories'])
unknown = sorted(set(categories) - known)
if unknown:
return {'lane': 'incomplete', 'unknown_categories': unknown}
for lane in ('red', 'orange'):
declared = sorted(set(categories) & set(rule[f'{lane}_lane_categories']))
if declared:
return {'lane': lane, 'categories': declared}
missing = [key for key in rule['blue_lane_requires']
if not isinstance(record.get(key), str) or not record[key].strip()]
return {'lane': 'incomplete', 'missing': missing} if missing else {'lane': 'blue'}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('rule', type=Path, help='published JSON rule artifact')
parser.add_argument('record', type=Path, help='JSON change record')
parser.add_argument('--build-mode', action='store_true', help='confirm build mode has not closed down')
args = parser.parse_args()
try:
result = evaluate(json.loads(args.rule.read_text()), json.loads(args.record.read_text()), build_mode=args.build_mode)
except (ValueError, OSError, TypeError) as exc:
parser.error(str(exc))
print(json.dumps(result, sort_keys=True))
return 0 if result['lane'] == 'blue' else 1
if __name__ == '__main__':
raise SystemExit(main())