access-engine/tools/verify_runtime_consumer_migration.py

#!/usr/bin/env python3
"""Deterministic T06 source classification. No credentials, network or mutation.

This reports source preparation, NEVER native readiness or workplan closure.
Owner-authenticated live allow/deny, predecessor and no-traffic/soak evidence
must be independently collected and reviewed by the migration coordinator.
"""
from dataclasses import dataclass
from pathlib import Path
import argparse
import ast
import tomllib
import yaml
import hashlib
import json
import re
from urllib.parse import urlparse

@dataclass(frozen=True)
class Consumer:
    owner: str
    repos: tuple
    pins: tuple
    paths: tuple
    network_policy: bool

CONSUMERS = (
 Consumer("ops-warden", ("ops-warden", "ops-bridge"), ("ops-warden", "ops-warden-fetch"), ("ops-warden/examples/warden.production.example.yaml",), False),
 Consumer("secrets-engine", ("secrets-engine",), ("secrets-engine",), ("secrets-engine/tools/exercise_approval_identity.py",), False),
 Consumer("user-engine", ("user-engine", "net-kingdom"), ("user-engine", "application-readback"), ("net-kingdom/sso-mfa/k8s/user-engine/runtime.yaml",), True),
 Consumer("tenant-engine", ("tenant-engine",), ("tenant-engine",), ("tenant-engine/deploy/base/tenant-engine.yaml", "tenant-engine/railiance/app.toml"), True),
 Consumer("informed-decision", ("informed-decision",), ("informed-decision-lifecycle", "informed-decision-sitting", "informed-decision-t03"), ("informed-decision/deploy/sitting-admission.json", "informed-decision/deploy/t03-admission.json", "informed-decision/deploy/lifecycle/admission.candidate.json", "informed-decision/deploy/lifecycle/service.candidate.json", "informed-decision/tools/render_deployment.py"), True),
)

class MigrationReadiness:
    def __init__(self, estate_root):
        self.root = Path(estate_root).resolve()

    def classify(self, consumer):
        sources = []; texts = []; documents = []
        for relative in consumer.paths:
            path = self.root / relative
            # Never follow a redirected public path into a secret/private tree.
            if path.is_symlink() or (path.exists() and not path.resolve().is_relative_to(self.root)):
                raise ValueError("public source path escapes bounded estate root")
            if not path.is_file():
                sources.append({"path": relative, "state": "missing"}); continue
            raw = path.read_bytes(); text = raw.decode()
            source = {"path": relative, "state": "present", "sha256": hashlib.sha256(raw).hexdigest()}
            try:
                if path.suffix == ".py":
                    # Literal configuration strings only, never execute source.
                    values = [n.value for n in ast.walk(ast.parse(text)) if isinstance(n, ast.Constant) and isinstance(n.value, str)]
                    texts.extend(values)
                    for value in values:
                        try:
                            documents.extend(doc for doc in yaml.safe_load_all(value) if isinstance(doc, (dict, list)))
                        except yaml.YAMLError:
                            pass
                elif path.suffix == ".toml":
                    doc = tomllib.loads(text); documents.append(doc)
                    texts.append(json.dumps(doc, sort_keys=True))
                else:
                    # JSON is a YAML subset; parse documents to exclude comments.
                    docs = [doc for doc in yaml.safe_load_all(text) if doc is not None]
                    documents.extend(docs)
                    texts.extend(json.dumps(doc, sort_keys=True) for doc in docs)
            except (ValueError, SyntaxError, yaml.YAMLError):
                source["state"] = "invalid"
            sources.append(source)
        text = "\n".join(texts)
        old_urls = []; new_urls = []
        endpoints = []
        for pin in consumer.pins:
            old = f"flex-auth-{pin}.flex-auth.svc"
            new = f"access-engine-{pin}.access-engine.svc"
            observed = []
            for value in re.findall(r"https?://[^\s\"'<>]+", text):
                try:
                    observed.append(urlparse(value.rstrip(".,;")))
                except ValueError:
                    continue  # templates/malformed URLs are not endpoint evidence
            def has_endpoint(host):
                for value in observed:
                    try:
                        if value.hostname in (host, host + ".cluster.local") and value.port == 8080:
                            return True
                    except ValueError:
                        continue
                return False
            if has_endpoint(old): old_urls.append(pin)
            if has_endpoint(new): new_urls.append(pin)
            endpoints.append({"pin": pin, "old_origin": f"http://{old}.cluster.local:8080", "new_origin": f"http://{new}.cluster.local:8080", "audience": "access-engine"})
        # Only runtime audience and namespace-selector literals count. Env names,
        # caller mount paths, prose and historical package names do not.
        audiences = set(); namespaces = set()
        def collect(value):
            if isinstance(value, dict):
                audience = value.get("audience")
                if isinstance(audience, str): audiences.add(audience)
                if value.get("kind") == "NetworkPolicy":
                    spec = value.get("spec", {})
                    for rule in spec.get("egress", []) if isinstance(spec, dict) else []:
                        if not isinstance(rule, dict): continue
                        for peer in rule.get("to", []):
                            if not isinstance(peer, dict): continue
                            selector = peer.get("namespaceSelector", {})
                            labels = selector.get("matchLabels", {}) if isinstance(selector, dict) else {}
                            if isinstance(labels, dict) and isinstance(labels.get("kubernetes.io/metadata.name"), str):
                                namespaces.add(labels["kubernetes.io/metadata.name"])
                for child in value.values(): collect(child)
            elif isinstance(value, list):
                for child in value: collect(child)
        for doc in documents: collect(doc)
        audience_old = "flex-auth" in audiences
        audience_new = "access-engine" in audiences
        egress_old = "flex-auth" in namespaces
        egress_new = "access-engine" in namespaces
        reasons = []
        if any(s["state"] != "present" for s in sources): reasons.append("inventory source files missing/invalid; owner must reconcile")
        if len(new_urls) != len(consumer.pins): reasons.append("not every new pin endpoint appears in bounded public source")
        if not audience_new: reasons.append("new projected-token audience not observed in public source")
        if consumer.network_policy and not egress_new: reasons.append("new namespace-selector egress not observed; add before switching")
        if old_urls and consumer.network_policy and egress_new and not egress_old: reasons.append("old endpoint remains but old namespace-selector removed; unsafe migration order")
        if reasons: phase = "NEEDS_OWNER_SOURCE_WORK"
        elif old_urls or audience_old: phase = "SOURCE_DUAL_PREPARATION_REVIEW_REQUIRED"
        else: phase = "SOURCE_SWITCHED_NATIVE_PROOF_REQUIRED"
        return {"owner": consumer.owner, "source_owners": list(consumer.repos), "phase": phase, "native_ready": False, "sources": sources, "old_endpoint_pins": old_urls, "new_endpoint_pins": new_urls, "audience": {"old": audience_old, "new": audience_new}, "egress_namespace": {"old": egress_old, "new": egress_new}, "reasons": reasons, "handoff": {"endpoints": endpoints, "order": ["verify T02/T05 and ordered predecessor live proof", "owner adds new egress while retaining old", "owner switches URL and projected audience for one consumer", "prove live allow and deny against every new pin", "observe old-twin logs with bounded no-traffic/soak evidence", "owner removes old egress only after evidence review"], "required_live_evidence": ["owner identity and exact deployed revision", "old/new image digest and package correspondence", "authenticated allow and deny exact decision bindings", "old-twin log observation interval and no-traffic results; silence/source strings alone are insufficient"], "special": "ops-warden operator config and ops-bridge ClusterIP tunnels require owner evidence; no new self resource.system grant may be inferred from policy routing lane" if consumer.owner == "ops-warden" else "CLI/operator runtime config requires owner evidence" if consumer.owner == "secrets-engine" else "public source does not establish live deployment"}}

    def report(self):
        return {"schema": "runtime-consumer-migration-source-v1", "workplan": "FLEX-WP-0036", "task": "FLEX-WP-0036-T06", "scope": "deterministic public-source readiness classification, no native admission", "dependency_state": "T02/T05 and ordered live proof require coordinator verification; this tool cannot assert completion", "consumers": [self.classify(c) for c in CONSUMERS]}

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--estate-root", type=Path, default=Path(__file__).resolve().parents[2])
    args = parser.parse_args()
    print(json.dumps(MigrationReadiness(args.estate_root).report(), indent=2, sort_keys=True))

if __name__ == "__main__": main()