"""Node-side adapter for sender_transition; read/dry-run unless explicitly armed. The attended capsule/entry point must bind these sources and the proposal. This module deliberately has no CLI. Secret reads use PartialObjectMetadata only. """ from __future__ import annotations import fcntl import http.client import importlib.util import json import os import socket import subprocess import tempfile import time from pathlib import Path from sender_transition import IMAGE, desired, digest, require NS = "email-connect" CLUSTER = "a553c742-0115-43d4-99a4-a5ca56fe0786" NAMESPACE = "5678c5d7-e4f5-4a62-8f24-1c6e1d906319" DEPLOYMENT = "2e98e52a-a2d1-4c52-9ac8-6efbc0a18e72" CONFIG = "558d3338-c4c2-490c-a5e5-af206c39eb51" FOUNDATION = "sha256:f8da6c831160e1c6354ce54f936809d2d4fb48eb45eb23484dc0bcdfe0bb52f0" FOUNDATION_SPEC = "cf5e1a5782243ebb72790ac7e9b868f44df027c1608ad37325264d290e70f537" PVC = "email-connect-data" CAPTURE_LOCK = Path("/var/lib/email-connect-backup/capture.lock") ROOT = Path(__file__).resolve().parent def module(name): spec = importlib.util.spec_from_file_location(name.replace("-", "_"), ROOT / (name + ".py")) value = importlib.util.module_from_spec(spec) spec.loader.exec_module(value) return value def stable(obj): require(obj is not None and not obj["metadata"].get("deletionTimestamp"), "object_missing_or_deleting") return {"uid": obj["metadata"]["uid"], "spec": obj.get("spec"), "data": obj.get("data")} class Native: def __init__(self, *, execute=False, journal=None, lock_path=CAPTURE_LOCK): self.kube = ["k3s", "kubectl", "--kubeconfig=/etc/rancher/k3s/k3s.yaml"] self.execute = execute self.journal = journal self.lock_path = lock_path self.lock_fd = None self.foundation_pins = None self.capture = module("capture-ledger") self.archive = module("ledger_archive") self.observer = module("inspect-sender-ledger") self.rehearsal = module("rehearse-ledger") self.capture.k = self.k self.receiver_uid = None self.node_name = None def k(self, *args, body=None, timeout=180): result = subprocess.run(self.kube + list(args), input=body, capture_output=True, timeout=timeout, check=False) require(result.returncode == 0, "native_command_failed") return result.stdout def get(self, kind, name=None, *, namespace=NS): require(kind.lower() in {"namespace", "deployment", "configmap", "pvc", "pv", "nodes", "externalsecret", "service", "networkpolicies", "serviceaccount", "secretstore", "pods", "pod", "replicaset"}, "resource_query_outside_reviewed_scope") args = ["get", kind] if name: args += [name, "--ignore-not-found"] if namespace: args += ["-n", namespace] raw = self.k(*args, "-o", "json", timeout=30) return json.loads(raw) if raw.strip() else None def metadata(self, name): require(name in {"email-connect-runtime", "email-connect-caller-user-engine", "email-connect-platform-smtp"}, "secret_metadata_scope_drift") path = f"/api/v1/namespaces/{NS}/secrets/{name}" with tempfile.TemporaryDirectory(prefix="sender-metadata-") as directory: sockpath = str(Path(directory) / "api.sock") proc = subprocess.Popen(self.kube + [ "proxy", "--unix-socket=" + sockpath, "--accept-paths=^" + path + "$", "--reject-methods=^(POST|PUT|PATCH|DELETE)$", ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) conn = None try: for _ in range(50): if Path(sockpath).exists(): break require(proc.poll() is None, "metadata_proxy_failed") time.sleep(0.1) sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(10) try: sock.connect(sockpath) except BaseException: sock.close() raise conn = http.client.HTTPConnection("localhost", timeout=10) conn.sock = sock conn.request("GET", path, headers={ "Accept": "application/json;as=PartialObjectMetadata;g=meta.k8s.io;v=v1"}) response = conn.getresponse() if response.status == 404: return None require(response.status == 200, "metadata_query_failed") value = json.load(response) require(value.get("kind") == "PartialObjectMetadata" and value.get("apiVersion") == "meta.k8s.io/v1" and "data" not in value and "stringData" not in value, "metadata_only_response_required") meta = value["metadata"] require(not meta.get("deletionTimestamp"), "secret_deleting") return {"uid": meta["uid"], "owners": meta.get("ownerReferences", [])} finally: if conn: conn.close() proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=5) def observe(self): require(stable(self.get("namespace", "kube-system", namespace=None))["uid"] == CLUSTER, "cluster_identity_changed") require(stable(self.get("namespace", NS, namespace=None))["uid"] == NAMESPACE, "namespace_identity_changed") deployment = self.get("deployment", "email-connect") config = self.get("configmap", "email-connect-config") require(stable(deployment)["uid"] == DEPLOYMENT and stable(config)["uid"] == CONFIG, "application_identity_changed") pvc = self.get("pvc", PVC) require(pvc.get("status", {}).get("phase") == "Bound", "ledger_pvc_not_bound") pv = self.get("pv", pvc["spec"]["volumeName"], namespace=None) require(pv["spec"]["claimRef"]["uid"] == pvc["metadata"]["uid"], "ledger_volume_binding_changed") terms = pv["spec"]["nodeAffinity"]["required"]["nodeSelectorTerms"] require(len(terms) == 1 and not terms[0].get("matchFields"), "volume_node_scope_changed") expressions = terms[0]["matchExpressions"] require(len(expressions) == 1 and expressions[0]["key"] == "kubernetes.io/hostname" and expressions[0]["operator"] == "In" and len(expressions[0]["values"]) == 1, "volume_node_scope_changed") hostname = expressions[0]["values"][0] nodes = [n for n in self.get("nodes", namespace=None)["items"] if n["metadata"].get("labels", {}).get("kubernetes.io/hostname") == hostname] require(len(nodes) == 1, "volume_node_identity_ambiguous") node = nodes[0] self.node_name = node["metadata"]["name"] volumes = deployment["spec"]["template"]["spec"]["volumes"] policy = next(v["configMap"]["name"] for v in volumes if v["name"] == "caller-policy") require(policy == "email-connect-callers-user-engine-276257849bf2", "caller_policy_changed") projection = self.get("externalsecret", "email-connect-runtime") old_secret = self.metadata("email-connect-runtime") caller_secret = self.metadata("email-connect-caller-user-engine") require(old_secret and caller_secret, "existing_custody_missing") policies = self.get("networkpolicies")["items"] if self.foundation_pins is not None: require(self.projection(self.journal.entries[0]["proposal"]) == self.foundation_pins, "new_projection_identity_changed") return { "identity": {"cluster": CLUSTER, "namespace": NAMESPACE, "deployment": DEPLOYMENT, "config": CONFIG}, "deployment_spec": deployment["spec"], "config_data": config["data"], "protected": { "pvc": stable(pvc), "pv": stable(pv), "node": {"name": self.node_name, "uid": node["metadata"]["uid"], "hostname": hostname}, "service": stable(self.get("service", "email-connect")), "caller_policy": stable(self.get("configmap", policy)), "old_projection": stable(projection), "old_secret": old_secret, "caller_secret": caller_secret, "network_policies": sorted((stable(p) for p in policies), key=lambda x: x["uid"]), }, } def prepare(self): require(not self.execute, "preparation_must_be_read_only") before = self.observe() dep = self.get("deployment", "email-connect") require(dep["status"].get("readyReplicas") == 1 and dep["status"].get("observedGeneration") == dep["metadata"]["generation"], "receiver_not_ready") after = desired(before) for spec in (dict(before["deployment_spec"], replicas=0), dict(after["deployment_spec"], replicas=0), after["deployment_spec"]): self.patch_deployment(before["deployment_spec"], spec) self.patch_config(before["config_data"], after["config_data"]) body = self.helper_manifest("sender-preflight-helper") self.helper_egress(body) projected = json.loads(self.k("create", "--dry-run=server", "-f", "-", "-o", "json", body=json.dumps(body).encode(), timeout=30)) require(projected["spec"]["automountServiceAccountToken"] is False, "helper_token_mount_changed") require(self.observe() == before, "preparation_changed") return {"before": before, "after": after, "native_server_dry_runs": 5} def armed(self): require(self.execute and self.journal is not None and self.lock_fd is not None, "attended_locked_execution_required") def acquire(self): require(self.execute and self.journal is not None, "attended_execution_required") require(self.lock_path == CAPTURE_LOCK, "capture_lock_path_changed") self.archive.private_directory(self.lock_path.parent) self.lock_fd = os.open(self.lock_path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) try: self.archive.private_file(self.lock_path) fcntl.flock(self.lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except BaseException: os.close(self.lock_fd) self.lock_fd = None raise def close(self): if self.lock_fd is not None: os.close(self.lock_fd) self.lock_fd = None def patch(self, kind, name, field, expected, value, uid): if self.execute: self.armed() obj = self.get(kind, name) require(stable(obj)["uid"] == uid and obj[field] == expected, "compare_and_swap_drift") patch = [{"op": "test", "path": "/metadata/uid", "value": uid}, {"op": "test", "path": "/metadata/resourceVersion", "value": obj["metadata"]["resourceVersion"]}, {"op": "test", "path": "/" + field, "value": expected}, {"op": "replace", "path": "/" + field, "value": value}] args = ["patch", kind, name, "-n", NS, "--type=json", "--patch-file=/dev/stdin", "-o", "json"] if not self.execute: args.append("--dry-run=server") result = json.loads(self.k(*args, body=json.dumps(patch).encode(), timeout=30)) require(result["metadata"]["uid"] == uid and result[field] == value, "patch_result_not_verified") def patch_deployment(self, expected, value): self.patch("deployment", "email-connect", "spec", expected, value, DEPLOYMENT) def patch_config(self, expected, value): require(self.get("deployment", "email-connect")["spec"]["replicas"] == 0 or not self.execute, "receiver_not_paused_for_config") self.patch("configmap", "email-connect-config", "data", expected, value, CONFIG) def projection(self, proposal): foundation = proposal["foundation"] require(digest(foundation) == FOUNDATION_SPEC, "foundation_contract_changed") pins = {} for wanted in foundation["resources"]: kind, name = wanted["kind"], wanted["metadata"]["name"] obj = self.get(kind, name) require(obj is not None and obj.get("apiVersion") == wanted["apiVersion"], "foundation_resource_missing") pins[kind] = stable(obj)["uid"] if kind == "ServiceAccount": require(obj.get("automountServiceAccountToken") is False and not obj.get("secrets") and not obj.get("imagePullSecrets"), "service_account_changed") else: require(obj["spec"] == wanted["spec"], "foundation_spec_changed") require(any(c.get("type") == "Ready" and c.get("status") == "True" for c in obj.get("status", {}).get("conditions", [])), "foundation_not_ready") if kind == "ExternalSecret": require(obj.get("status", {}).get("refreshTime"), "projection_refresh_missing") meta = self.metadata("email-connect-platform-smtp") require(meta is not None and any(o.get("uid") == pins["ExternalSecret"] and o.get("controller") is True for o in meta["owners"]), "projection_owner_changed") pins["Secret"] = meta["uid"] return pins def admission(self, proposal): self.armed() receipt = proposal["foundation_receipt"] require(receipt.get("schema") == "platform.email-connect-smtp-foundation.v1" and receipt.get("proposal_digest") == FOUNDATION and receipt.get("status") == "foundation_admitted_projection_ready" and receipt.get("application_sender_changed") is False, "verified_owner_foundation_receipt_required") for key in ("native_binding_verified", "native_consumer_auth_verified", "exact_read_verified", "mutation_sibling_metadata_denied", "bounded_ttl", "other_service_account_denied", "wrong_audience_denied", "workload_token_revocation_verified", "projection_ready"): require(receipt.get(key) is True, "foundation_proof_incomplete") pins = self.projection(proposal) require(pins["ServiceAccount"] == receipt["service_account_uid"], "admitted_identity_changed") self.journal.record("native_foundation_verified", pins=pins) self.foundation_pins = pins def helper_manifest(self, name): body = self.capture.helper(name) # Read-only PVC, no credentials, explicit scheduling on the existing node. require(self.node_name is not None, "volume_node_not_observed") body["spec"]["nodeName"] = self.node_name return body def helper_egress(self, body): self.capture.require_no_helper_egress(self.get("networkpolicies")["items"], body["metadata"]["labels"]) def helpers(self): require(self.journal is not None, "journal_required") return {e["name"]: e for e in self.journal.entries if e["phase"] == "native_helper_created"} def exact_helper(self, obj, record): require(stable(obj)["uid"] == record["uid"] and obj["spec"] == record["spec"], "helper_identity_or_spec_changed") self.helper_egress(obj) def fence(self, journal): self.armed() self.k("wait", "--for=delete", "pod", "-n", NS, "-l", "app.kubernetes.io/name=email-connect", "--timeout=120s") require(self.get("deployment", "email-connect")["spec"]["replicas"] == 0, "receiver_not_quiesced") self.check_consumers(allow_receiver=False) journal.record("native_writers_fenced", deployment_uid=DEPLOYMENT) def check_consumers(self, *, allow_receiver): known_helpers = self.helpers() for pod in self.get("pods")["items"]: if any(v.get("persistentVolumeClaim", {}).get("claimName") == PVC for v in pod["spec"].get("volumes", [])): record = known_helpers.get(pod["metadata"]["name"]) if allow_receiver and pod["metadata"]["uid"] == self.receiver_uid: require(not pod["metadata"].get("deletionTimestamp"), "receiver_is_terminating") continue require(record is not None, "unowned_ledger_pvc_consumer") self.exact_helper(pod, record) def ensure_helper(self, journal): self.armed() for name, record in reversed(list(self.helpers().items())): obj = self.get("pod", name) if obj is not None: self.exact_helper(obj, record) self.k("wait", "--for=condition=Ready", "pod/" + name, "-n", NS, "--timeout=120s") return name # A create with no durable returned UID cannot authorize adoption/deletion. for entry in journal.entries: if entry["phase"] == "native_helper_create_intent" and entry["name"] not in self.helpers(): require(self.get("pod", entry["name"]) is None, "unconfirmed_helper_requires_inspection") name = "sender-ledger-" + journal.entries[0]["proposal_digest"][:12] + "-" + str(len(journal.entries)) body = self.helper_manifest(name) self.helper_egress(body) require(self.get("pod", name) is None, "helper_name_occupied") projected = json.loads(self.k("create", "--dry-run=server", "-f", "-", "-o", "json", body=json.dumps(body).encode(), timeout=30)) journal.record("native_helper_create_intent", name=name, spec=projected["spec"]) obj = json.loads(self.k("create", "-f", "-", "-o", "json", body=json.dumps(body).encode())) require(obj["spec"] == projected["spec"], "helper_create_spec_changed") journal.record("native_helper_created", name=name, uid=obj["metadata"]["uid"], spec=obj["spec"]) self.k("wait", "--for=condition=Ready", "pod/" + name, "-n", NS, "--timeout=120s") self.exact_helper(self.get("pod", name), self.helpers()[name]) return name def snapshot(self, journal, *, quiesced): self.armed() if quiesced: self.fence(journal) name = self.ensure_helper(journal) self.check_consumers(allow_receiver=not quiesced) destination = journal.path.parent / ("snapshot-" + str(len(journal.entries))) journal.record("native_snapshot_intent", name=name, directory=destination.name, writers_quiesced=quiesced) destination.mkdir(mode=0o700) captured = self.capture.capture_database(name, destination) captured["writers_quiesced"] = quiesced path = destination / "database.sqlite3" outcomes = self.observer.inspect(path) self.check_consumers(allow_receiver=not quiesced) require(captured["ledger_fingerprint"] == outcomes["ledger_fingerprint"], "snapshot_drift") if quiesced: # Run the existing source-bound no-provider rehearsal in helper memory. archive_source = (ROOT / "ledger_archive.py").read_text() prefix = ("import types,sys\nif not __debug__:raise RuntimeError('assertions_required')\n" "m=types.ModuleType('ledger_archive')\n" "sys.modules['ledger_archive']=m\nexec(" + repr(archive_source) + ",m.__dict__)\n") code = prefix + self.rehearsal.CODE proof = json.loads(self.k("exec", "-i", "-n", NS, name, "--", "python", "-c", code, body=path.read_bytes())) require(proof.get("status") == "isolated_ledger_verified" and proof.get("mail_sent") is False and proof["ledger_fingerprint"] == captured["ledger_fingerprint"], "isolated_rehearsal_failed") journal.record("native_snapshot_rehearsed", proof=proof) journal.record("native_snapshot_captured", name=name, directory=destination.name, proof=captured) return path, outcomes def verify(self, target, journal): self.armed() self.k("rollout", "status", "deployment/email-connect", "-n", NS, "--timeout=180s", timeout=240) require(self.observe() == target, "receiver_spec_not_verified") pods = [p for p in self.get("pods")["items"] if p["metadata"].get("labels", {}).get("app.kubernetes.io/name") == NS] require(len(pods) == 1, "receiver_pod_count_changed") pod = pods[0] require(not pod["metadata"].get("deletionTimestamp") and pod["status"]["phase"] == "Running", "receiver_not_running") statuses = pod["status"].get("containerStatuses", []) require(len(statuses) == 1 and statuses[0].get("ready") is True and statuses[0]["imageID"].endswith(IMAGE.split("@", 1)[1]), "receiver_image_not_verified") wanted = target["deployment_spec"]["template"]["spec"]["containers"][0] actual = pod["spec"]["containers"] require(len(actual) == 1 and all(actual[0].get(k) == wanted.get(k) for k in ("image", "env", "envFrom")), "receiver_environment_references_changed") owner = next(o for o in pod["metadata"]["ownerReferences"] if o.get("controller")) require(owner["kind"] == "ReplicaSet", "receiver_owner_changed") rs = self.get("replicaset", owner["name"]) require(rs["metadata"]["uid"] == owner["uid"] and any( o.get("uid") == DEPLOYMENT and o.get("controller") for o in rs["metadata"]["ownerReferences"]), "receiver_deployment_owner_changed") sender = target["config_data"]["EMAIL_CONNECT_SENDER"] code = '''import json,os,urllib.request,urllib.error if not __debug__:raise RuntimeError('assertions_required') assert os.environ['EMAIL_CONNECT_SENDER']==__EXPECTED_SENDER__ assert os.environ['EMAIL_CONNECT_SMTP_USERNAME']==__EXPECTED_SENDER__ assert os.environ['EMAIL_CONNECT_CALLERS_FILE']=='/etc/email-connect-callers/callers.json' assert os.environ.get('EMAIL_CONNECT_SMTP_PASSWORD') for path in ['/healthz','/readyz']: with urllib.request.urlopen('http://127.0.0.1:8080'+path,timeout=5) as r:assert r.status==200 request=urllib.request.Request('http://127.0.0.1:8080/v1/send',data=b'{}',method='POST', headers={'Content-Type':'application/json','Authorization':'Bearer synthetic-invalid'}) try: urllib.request.urlopen(request,timeout=5);raise RuntimeError('unauthenticated_request_accepted') except urllib.error.HTTPError as e: assert e.code==401 and json.load(e).get('error')=='unauthorized' print(json.dumps({'health_ready':True,'public_sender_verified':True,'scoped_caller_mode':True, 'smtp_password_present':True,'unauthenticated_denial':401,'mail_sent':False,'private_values_printed':False})) '''.replace("__EXPECTED_SENDER__", repr(sender)) proof = json.loads(self.k("exec", "-n", NS, pod["metadata"]["name"], "-c", NS, "--", "python", "-c", code)) self.receiver_uid = pod["metadata"]["uid"] journal.record("native_receiver_verified", pod_uid=self.receiver_uid, proof=proof) return proof def cleanup(self, journal): self.armed() for name, record in self.helpers().items(): obj = self.get("pod", name) if obj is None: continue self.exact_helper(obj, record) journal.record("native_helper_delete_intent", name=name, uid=record["uid"]) options = {"apiVersion": "v1", "kind": "DeleteOptions", "preconditions": {"uid": record["uid"]}} self.k("delete", "--raw=/api/v1/namespaces/" + NS + "/pods/" + name, "-f", "/dev/stdin", body=json.dumps(options).encode()) self.k("wait", "--for=delete", "pod/" + name, "-n", NS, "--timeout=120s") require(self.get("pod", name) is None, "helper_cleanup_unverified") journal.record("native_helper_deleted", name=name, uid=record["uid"])