140 lines
9.2 KiB
Python
140 lines
9.2 KiB
Python
|
#!/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()
|