scripts/deploy_rotation_qa.py
Rama main · réplica de lectura del repositorio original.
"""Real shared-Secret rotation through installed Deploy, KVM and Nginx."""
import concurrent.futures
import contextlib
import http.client
import json
import subprocess
import threading
import time
import uuid
def run_rotation(api, wait_for, traffic, network, spec_a, port, evidence):
services = []
secret = None
held_ip = None
stop = threading.Event()
lock = threading.Lock()
threads = []
executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
started = time.monotonic()
metrics = {"requests": {}, "httpFailures": [], "monitorErrors": [], "maxLiveVMs": 0,
"maxLiveVCPUs": 0, "stages": []}
evidence["secretRotation"] = metrics
def path(service, suffix=""):
return "/api/micro-services/" + service["id"] + suffix
def current(service):
return api.call("GET", path(service, "/deployment"))
def checks():
with lock:
assert not metrics["httpFailures"], metrics["httpFailures"]
assert not metrics["monitorErrors"], metrics["monitorErrors"]
def ready(service):
record = current(service)
assert record["phase"] not in ["failed", "stopped"], record
return record if record["phase"] == "running" else None
def edit(value=None):
summary = next(item for item in api.call("GET", "/api/secrets") if item["id"] == secret)
body = {"name": summary["name"], "description": "edited by disposable rotation QA", "version": summary["version"]}
if value is not None:
body["value"] = value
return api.call("PUT", "/api/secrets/" + secret, body)
def load(service):
domain = service["spec"]["domain"]
allowed = [(200, "unrelated")] if service["unrelated"] else [(200, label) for label in ["old", "next", "latest"]]
while not stop.is_set():
failure = None
try:
response = traffic("127.0.0.1", port, domain, "/cgi-bin/rotation")
if response not in allowed:
failure = repr(response)
except Exception as error:
failure = str(error)
with lock:
metrics["requests"][domain] += 1
if failure and len(metrics["httpFailures"]) < 10:
metrics["httpFailures"].append({"domain": domain, "error": failure})
stop.wait(.03)
def observe():
while not stop.is_set():
try:
live = [vm for vm in api.call("GET", "/api/vms") if vm["state"] in ["running", "starting", "paused"]]
cpus = sum(vm["cpu"] for vm in live)
assert len(live) <= 4 and cpus <= 4, "Secret rotation exceeded one shared surge replica"
with lock:
metrics["maxLiveVMs"] = max(metrics["maxLiveVMs"], len(live))
metrics["maxLiveVCPUs"] = max(metrics["maxLiveVCPUs"], cpus)
except (OSError, http.client.HTTPException):
pass # API restart must not affect the separate HTTP measurements.
except Exception as error:
with lock:
if len(metrics["monitorErrors"]) < 10:
metrics["monitorErrors"].append(str(error))
stop.wait(.1)
def latest():
checks()
records = [current(service) for service in services]
if any(record["phase"] != "running" for record in records):
return None
for service, record in zip(services, records):
expected = "unrelated" if service["unrelated"] else "latest"
if traffic("127.0.0.1", port, service["spec"]["domain"], "/cgi-bin/rotation") != (200, expected):
return None
assert record["spec"] == service["spec"] and record["digest"] == spec_a["image"].split("@", 1)[1], record
public = json.dumps(record) + json.dumps(api.call("GET", path(service, "/releases")))
assert not any(value in public for value in ["qa-secret-v0", "qa-secret-v1", "qa-secret-v2", "qa-secret-bad"])
return records
try:
secret = api.call("POST", "/api/secrets", {"name": "rotation-qa-" + uuid.uuid4().hex[:8], "value": "qa-secret-v0"})["id"]
for index in range(3):
name = f"rotate-{index}-" + uuid.uuid4().hex[:8]
unrelated = index == 2
spec = {**spec_a, "name": name, "domain": name + ".example.test", "healthPath": "/cgi-bin/rotation-health",
"startupTimeout": 4, "drainTimeout": 0 if unrelated else 45,
"env": {"UNRELATED": "1"} if unrelated else {},
"secrets": {} if unrelated else {"PASSWORD": secret}}
service = {"id": api.call("POST", "/api/micro-services", spec)["id"], "spec": spec, "unrelated": unrelated}
services.append(service)
metrics["requests"][spec["domain"]] = 0
api.call("POST", path(service, "/deployment"), {"microNetworkId": network})
service["initial"] = wait_for(lambda: ready(service), timeout=180)
service["spec"] = service["initial"]["spec"].copy()
for target, arguments in [(observe, ())] + [(load, (service,)) for service in services]:
thread = threading.Thread(target=target, args=arguments, daemon=True)
threads.append(thread)
thread.start()
initial_ids = [service["initial"]["id"] for service in services]
initial_vms = {vm["id"] for vm in api.call("GET", "/api/vms")}
edit() # A metadata edit increments Secret version but must not roll any VM.
time.sleep(12)
checks()
assert [current(service)["id"] for service in services] == initial_ids
assert {vm["id"] for vm in api.call("GET", "/api/vms")} == initial_vms
metrics["stages"].append("metadata edit retained all replicas")
first = services[0]
held_ip = first["initial"]["instances"][0]["ipv4"]
held = executor.submit(traffic, "127.0.0.1", port, first["spec"]["domain"], "/cgi-bin/hold")
wait_for(lambda: traffic(held_ip, 8080, "fixture", "/cgi-bin/holding") == (200, "holding"), timeout=10)
edit("qa-secret-v1")
wait_for(lambda: current(first)["phase"] == "draining", timeout=180)
assert not held.done()
assert traffic("127.0.0.1", port, first["spec"]["domain"], "/cgi-bin/rotation") == (200, "next")
edit("qa-secret-v2")
edit("qa-secret-v2")
pid = api.call("GET", "/api/pipe")["units"][0]["mainPid"]
subprocess.run(["systemctl", "restart", "firecrab-api.service"], check=True)
api.ready()
assert api.call("GET", "/api/pipe")["units"][0]["mainPid"] == pid
traffic(held_ip, 8080, "fixture", "/cgi-bin/release")
assert held.result(timeout=10) == (200, "release-A")
held_ip = None
rotated = wait_for(latest, timeout=300, message="shared Secret values did not converge after restart")
assert all(rotated[i]["id"] != initial_ids[i] for i in [0, 1])
assert rotated[2]["id"] == initial_ids[2]
assert rotated[2]["instances"] == services[2]["initial"]["instances"]
metrics["stages"].append("shared rotation, rapid edits, pinned image, restart, retained HTTP, unrelated VM preservation")
print("PASS shared Secret rotation and rapid edits recover after API restart; image/config and unrelated VM retained", flush=True)
stable_vms = {vm["id"] for vm in api.call("GET", "/api/vms")}
edit("qa-secret-bad")
def bad_attempt():
checks()
record = current(first)
probing = record["phase"] == "starting" and any(
instance["ipv4"] and not instance["healthy"] for instance in record["instances"])
return record if record["id"] != rotated[0]["id"] and probing else None
attempted = wait_for(bad_attempt, timeout=180, message="unhealthy rotation did not reach a booted candidate")
# Restore desired values while the captured bad candidate is still probing.
# That attempt must keep its private snapshot and fail without another retry race.
edit("qa-secret-v2")
def failed():
record = api.call("GET", path(first, "/deployments/" + attempted["id"]))
return record if record["phase"] == "failed" else None
rejected = wait_for(failed, timeout=180)
assert rejected["update"]["completed"], rejected
wait_for(latest, timeout=180)
time.sleep(12)
records = latest()
assert records and [record["id"] for record in records] == [record["id"] for record in rotated]
assert {vm["id"] for vm in api.call("GET", "/api/vms")} == stable_vms
api.call("DELETE", "/api/secrets/" + secret, allowed=(409,))
metrics["stages"].append("unhealthy Secret candidate rejected, previous values/VMs retained, in-use deletion rejected")
checks()
assert metrics["maxLiveVMs"] == 4 and metrics["maxLiveVCPUs"] == 4, metrics
assert min(metrics["requests"].values()) > 20, metrics
except BaseException as error:
metrics["scenarioError"] = str(error)
raise
finally:
stop.set()
for thread in threads:
thread.join(timeout=35)
if held_ip:
with contextlib.suppress(Exception):
traffic(held_ip, 8080, "fixture", "/cgi-bin/release")
executor.shutdown(wait=True, cancel_futures=True)
errors = []
for service in services:
try:
for _ in range(3):
api.call("POST", path(service, "/stop"))
result = wait_for(lambda: (record if (record := current(service))["phase"] in ["stopped", "failed"] else None), timeout=180)
if result["phase"] == "stopped":
break
time.sleep(1)
assert result["phase"] == "stopped", result
api.call("DELETE", path(service))
except Exception as error:
errors.append(str(error))
if secret:
try:
api.call("DELETE", "/api/secrets/" + secret)
except Exception as error:
errors.append(str(error))
metrics["elapsedSeconds"] = round(time.monotonic() - started, 2)
metrics["cleanupErrors"] = errors
assert not errors, errors
checks()
print(f"PASS Secret rotation: {sum(metrics['requests'].values())} continuous HTTP requests, zero errors; dedicated services/Secret removed", flush=True)