#!/usr/bin/env python3 """Offline, read-only allow-list planner. No network access or device writes. Python 3.10+. The CLI reads two JSON files and emits a plan to stdout. Read EXERCISE.md before opening this reference solution. """ from __future__ import annotations import argparse import hashlib import ipaddress import json import sys from dataclasses import asdict, dataclass from pathlib import Path from typing import Any class ValidationError(ValueError): """Input cannot safely be used to generate a plan.""" def identifier(value: Any, name: str) -> str: if not isinstance(value, str) or not value or value != value.strip(): raise ValidationError(f"{name} must be a nonempty, whitespace-trimmed string") if len(value) > 200 or any(ord(char) < 32 for char in value): raise ValidationError(f"{name} has invalid length or control characters") return value def prefix_key(prefix: str) -> tuple[int, int, int]: network = ipaddress.ip_network(prefix, strict=True) return network.version, int(network.network_address), network.prefixlen def canonical_prefixes(value: Any) -> tuple[str, ...]: if not isinstance(value, list) or len(value) > 10000: raise ValidationError("allowed_prefixes must be a list of at most 10000 entries") result: set[str] = set() for prefix in value: if not isinstance(prefix, str) or '/' not in prefix: raise ValidationError("each prefix must be an explicit CIDR string") try: canonical = str(ipaddress.ip_network(prefix, strict=True)) except ValueError as exc: raise ValidationError(f"invalid network prefix: {prefix!r}") from exc if canonical in result: raise ValidationError(f"duplicate normalized prefix: {canonical}") result.add(canonical) return tuple(sorted(result, key=prefix_key)) @dataclass(frozen=True) class Device: tenant: str device_id: str prefixes: tuple[str, ...] @property def key(self) -> tuple[str, str]: return self.tenant, self.device_id @dataclass(frozen=True) class Snapshot: revision: str devices: tuple[Device, ...] def parse_snapshot(raw: Any) -> Snapshot: if not isinstance(raw, dict) or set(raw) != {'complete', 'revision', 'devices'}: raise ValidationError("snapshot requires exactly complete, revision, and devices") # Do not accept 1 or the string 'true' as trustworthy completeness evidence. if raw['complete'] is not True: raise ValidationError("refusing an incomplete or unverified snapshot") revision = identifier(raw['revision'], 'revision') if not isinstance(raw['devices'], list): raise ValidationError("devices must be a list") records: list[Device] = [] keys: set[tuple[str, str]] = set() for item in raw['devices']: if not isinstance(item, dict) or set(item) != {'tenant', 'device_id', 'allowed_prefixes'}: raise ValidationError("each device requires exactly tenant, device_id, allowed_prefixes") record = Device(identifier(item['tenant'], 'tenant'), identifier(item['device_id'], 'device_id'), canonical_prefixes(item['allowed_prefixes'])) if record.key in keys: raise ValidationError(f"duplicate scoped device: {record.key}") keys.add(record.key) records.append(record) return Snapshot(revision, tuple(sorted(records, key=lambda d: d.key))) def fingerprint(snapshot: Snapshot) -> str: payload = json.dumps(asdict(snapshot), sort_keys=True, separators=(',', ':')) return hashlib.sha256(payload.encode('utf-8')).hexdigest() def plan_changes(desired_raw: Any, observed_raw: Any, *, expected_observed_revision: str, allow_removals: bool = False) -> dict[str, Any]: """Plan exact-prefix allow-list changes, not ordered firewall policies. A truthfully complete observed snapshot and matching managed device scope are required even when prefix removal has been explicitly authorized. The revision check is local evidence, not a distributed lock. A real executor must re-check current state before applying this plan. """ if type(allow_removals) is not bool: raise ValidationError('allow_removals must be a boolean') expected = identifier(expected_observed_revision, 'expected_observed_revision') desired, observed = parse_snapshot(desired_raw), parse_snapshot(observed_raw) if observed.revision != expected: raise ValidationError('observed revision does not match the expected revision') wanted = {d.key: d for d in desired.devices} actual = {d.key: d for d in observed.devices} if wanted.keys() != actual.keys(): raise ValidationError('managed device scope mismatch; device creation/deletion is out of scope') changes: list[dict[str, Any]] = [] for key in sorted(wanted): before, after = set(actual[key].prefixes), set(wanted[key].prefixes) add = sorted(after - before, key=prefix_key) remove = sorted(before - after, key=prefix_key) if remove and not allow_removals: raise ValidationError(f'prefix removal requires explicit approval for {key}') if add or remove: changes.append({'tenant': key[0], 'device_id': key[1], 'add': add, 'remove': remove}) return {'schema_version': 1, 'read_only': True, 'desired_revision': desired.revision, 'observed_revision': observed.revision, 'observed_fingerprint': fingerprint(observed), 'desired_fingerprint': fingerprint(desired), 'removals_authorized': allow_removals, 'changes': changes} def load_json(path: str) -> Any: data = Path(path).read_bytes() if len(data) > 2_000_000: raise ValidationError('fixture exceeds the 2 MB exercise limit') def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise ValidationError(f'duplicate JSON object key: {key}') result[key] = value return result return json.loads(data, object_pairs_hook=reject_duplicate_keys) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('desired') parser.add_argument('observed') parser.add_argument('--expected-observed-revision', required=True) parser.add_argument('--allow-removals', action='store_true') args = parser.parse_args() try: result = plan_changes(load_json(args.desired), load_json(args.observed), expected_observed_revision=args.expected_observed_revision, allow_removals=args.allow_removals) except (ValidationError, OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: print(json.dumps({'status': 'rejected', 'error': str(exc)}), file=sys.stderr) return 2 print(json.dumps(result, indent=2)) return 0 if __name__ == '__main__': raise SystemExit(main())