tegwick
c0c3221855
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
453 lines
24 KiB
Python
453 lines
24 KiB
Python
"""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"])
|