scripts/deploy_auto_qa.py
Rama main · réplica de lectura del repositorio original.
"""Authenticated registry metadata and automatic Deploy on the disposable Linux QA host."""
import base64
import contextlib
import http.server
import http.client
import json
import os
from pathlib import Path
import shutil
import ssl
import subprocess
import tempfile
import threading
import time
import uuid
from urllib.parse import urlsplit
class MetadataRegistry:
def __init__(self, fixtures, address=("127.0.0.1", 443), tls=None):
self.tags = {"stable": fixtures[0][0], "v1.0.0": fixtures[0][0], "v1.1.0": fixtures[1][0], "v2.0.0-rc.1": fixtures[3][0]}
self.manifests = {digest: manifest for digest, manifest, _ in fixtures}
self.blobs = {key: value for _, _, blobs in fixtures for key, value in blobs.items()}
self.authenticated = 0
self.owner = None
self.proxies = {}
self.proxy_calls = []
self.ca_file = None
registry = self
expected = "Basic " + base64.b64encode(b"team:local-deploy-qa-only").decode()
class Handler(http.server.BaseHTTPRequestHandler):
def proxy(self):
host = self.headers.get("Host", "").split(":", 1)[0]
port = registry.proxies.get(host)
if port is None:
return False
length = int(self.headers.get("Content-Length", "0"))
assert 0 <= length <= 8192
headers = {"Host": host, "Content-Type": "application/json"}
if self.headers.get("Authorization"):
headers["Authorization"] = self.headers["Authorization"]
with contextlib.closing(http.client.HTTPConnection("127.0.0.1", port, timeout=15)) as connection:
connection.request(self.command, self.path, self.rfile.read(length) if length else None, headers)
response = connection.getresponse()
body = response.read(65537)
assert len(body) <= 65536
self.send_response(response.status)
self.send_header("Content-Length", str(len(body)))
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(body)
registry.proxy_calls.append((self.command, self.path))
return True
def do_POST(self):
if not self.proxy():
self.send_error(405)
def do_GET(self):
if self.proxy():
return
if self.headers.get("Authorization") != expected:
self.send_response(401)
self.send_header("WWW-Authenticate", 'Basic realm="Deploy QA"')
self.send_header("Content-Length", "0")
self.end_headers()
return
registry.authenticated += 1
path = urlsplit(self.path).path
key = path.rsplit("/", 1)[-1]
manifest = registry.manifests.get(registry.tags.get(key, key))
if path == "/v2/team/deploy-qa/tags/list":
body, mime = json.dumps({"tags": sorted(registry.tags)}).encode(), "application/json"
elif "/manifests/" in path:
body, mime = manifest, "application/vnd.oci.image.manifest.v1+json"
elif "/blobs/" in path:
body, mime = registry.blobs.get(key), "application/octet-stream"
else:
body, mime = b"{}" if path in ["/v2", "/v2/"] else None, "application/json"
self.send_response(200 if body is not None else 404)
body = body if body is not None else b"missing"
self.send_header("Content-Length", str(len(body)))
self.send_header("Content-Type", mime)
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
self.server = http.server.ThreadingHTTPServer(address, Handler)
if tls:
self.server.socket = tls.wrap_socket(self.server.socket, server_side=True)
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
def close(self):
self.server.shutdown()
self.server.server_close()
self.thread.join()
@contextlib.contextmanager
def trusted_registry(fixtures, proxy_hosts=()):
"""Use normal TLS/Auth with a private hosts entry; never contact or publish to GHCR."""
assert os.geteuid() == 0, "registry trust fixture requires the disposable root QA runner"
assert all(host == "ci.example.test" for host in proxy_hosts), "only the private CI QA hostname is allowed"
names = ["ghcr.io", *proxy_hosts]
hosts = Path("/etc/hosts")
assert not any(set(names).intersection(line.split("#", 1)[0].split()) for line in hosts.read_text().splitlines()), "refusing an existing GHCR hosts mapping"
environment = subprocess.check_output(["systemctl", "show", "firecrab-api.service", "--property=Environment", "--value"], text=True)
assert "SSL_CERT_FILE=" not in environment, "refusing an existing custom registry trust file"
dropin = Path("/run/systemd/system/firecrab-api.service.d/90-deploy-registry-qa.conf")
assert not dropin.exists(), "refusing an existing registry QA drop-in"
directory = Path(tempfile.mkdtemp(prefix="qa-registry-", dir="/var/lib/firecrab"))
directory.chmod(0o755)
marker = "127.0.0.1 " + " ".join(names) + " # firecrab-deploy-qa-" + uuid.uuid4().hex
registry = None
appended = installed = False
try:
def openssl(*args):
subprocess.run(["openssl", *args], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
openssl("req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "1", "-subj", "/CN=Deploy QA root", "-keyout", str(directory / "ca.key"), "-out", str(directory / "ca.pem"), "-addext", "basicConstraints=critical,CA:TRUE")
openssl("req", "-newkey", "rsa:2048", "-nodes", "-subj", "/CN=ghcr.io", "-keyout", str(directory / "leaf.key"), "-out", str(directory / "leaf.csr"))
(directory / "extensions").write_text("subjectAltName=" + ",".join("DNS:" + host for host in names) + "\nbasicConstraints=critical,CA:FALSE\nkeyUsage=digitalSignature,keyEncipherment\nextendedKeyUsage=serverAuth\n")
openssl("x509", "-req", "-days", "1", "-in", str(directory / "leaf.csr"), "-CA", str(directory / "ca.pem"), "-CAkey", str(directory / "ca.key"), "-CAcreateserial", "-extfile", str(directory / "extensions"), "-out", str(directory / "leaf.pem"))
(directory / "ca.pem").chmod(0o644)
tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
tls.load_cert_chain(directory / "leaf.pem", directory / "leaf.key")
registry = MetadataRegistry(fixtures, tls=tls) # Binding refuses a pre-existing listener.
registry.ca_file = directory / "ca.pem"
with hosts.open("a") as output:
output.write("\n" + marker + "\n")
appended = True
dropin.parent.mkdir(parents=True, exist_ok=True)
dropin.write_text(f'[Service]\nEnvironment="SSL_CERT_FILE={directory / "ca.pem"}"\n')
installed = True
subprocess.run(["systemctl", "daemon-reload"], check=True)
subprocess.run(["systemctl", "restart", "firecrab-api.service"], check=True)
yield registry
finally:
errors = []
actions = []
if installed:
actions.append(dropin.unlink)
if appended:
actions.append(lambda: hosts.write_text("\n".join(line for line in hosts.read_text().splitlines() if line != marker) + "\n"))
if installed:
actions.extend([lambda: subprocess.run(["systemctl", "daemon-reload"], check=True),
lambda: subprocess.run(["systemctl", "restart", "firecrab-api.service"], check=True)])
if registry:
actions.append(registry.close)
actions.append(lambda: shutil.rmtree(directory))
for action in actions:
try:
action()
except Exception as error:
errors.append(error)
if errors:
raise ExceptionGroup("registry fixture cleanup failed", errors)
def run_auto(api, wait_for, traffic, network, spec_a, fixtures, port, evidence):
metrics = {"requests": 0, "httpFailures": [], "monitorErrors": [], "maxLiveVMs": 0, "maxLiveVCPUs": 0,
"httpTimeoutSeconds": 5, "stages": []}
evidence["automaticDeploy"] = metrics
services = []
stop = threading.Event()
loader = None
started = time.monotonic()
def path(service, suffix=""):
return "/api/micro-services/" + service["id"] + suffix
def current(service):
return api.call("GET", path(service, "/deployment"))
def status(service):
return api.call("GET", path(service, "/auto-deploy"))
def ready(service, expected):
record = current(service)
assert record["phase"] not in ["failed", "cancelled", "stopped"], record
return record if record["phase"] == "running" and record["digest"] == expected else None
def monitor():
while not stop.is_set():
for service in services.copy():
try:
value = traffic("127.0.0.1", port, service["spec"]["domain"])
metrics["requests"] += 1
if value not in [(200, "release-A"), (200, "release-B"), (200, "release-C")]:
metrics["httpFailures"].append(value)
except Exception as error:
metrics["httpFailures"].append(str(error))
try:
live = [vm for vm in api.call("GET", "/api/vms") if vm["state"] not in ["created", "stopped", "failed"]]
metrics["maxLiveVMs"] = max(metrics["maxLiveVMs"], len(live))
metrics["maxLiveVCPUs"] = max(metrics["maxLiveVCPUs"], sum(vm["cpu"] for vm in live))
except (OSError, http.client.HTTPException):
pass # API restart must not interrupt the independently served HTTP traffic.
except Exception as error:
metrics["monitorErrors"].append(str(error))
stop.wait(.05)
def create():
name = "auto-" + uuid.uuid4().hex[:8]
spec = {**spec_a, "name": name, "domain": name + ".example.test", "replicas": 1}
service = {"id": api.call("POST", "/api/micro-services", spec)["id"], "spec": spec}
services.append(service)
api.call("POST", path(service, "/deployment"), {"microNetworkId": network})
return service, wait_for(lambda: ready(service, fixtures[0][0]))
def rollback(service, active, original):
api.call("POST", path(service, "/rollback"), {"operationId": str(uuid.uuid4()), "expectedReleaseId": active["id"], "releaseId": original["id"]})
return wait_for(lambda: ready(service, fixtures[0][0]))
with trusted_registry(fixtures) as registry:
try:
api.ready()
service, original = create()
unrelated, unrelated_original = create()
loader = threading.Thread(target=monitor, daemon=True)
loader.start()
policy = {"reference": "ghcr.io/team/deploy-qa:stable", "intervalSeconds": 60}
configured = api.call("PUT", path(service, "/auto-deploy"), policy)
assert configured["policy"] == policy
wait_for(lambda: status(service)["observedDigest"] == fixtures[0][0])
registry.tags["stable"] = fixtures[1][0]
active = wait_for(lambda: ready(service, fixtures[1][0]), timeout=180)
assert active["spec"]["image"] == "ghcr.io/team/deploy-qa@" + fixtures[1][0]
assert status(service)["operationId"] == active["id"]
metrics["stages"].append("authenticated tag change submitted a pinned automatic Deploy")
restored = rollback(service, active, original)
assert status(service)["pause"] == "rollback"
subprocess.run(["systemctl", "restart", "firecrab-api.service"], check=True)
api.ready()
deadline = time.monotonic() + 65
while time.monotonic() < deadline:
assert current(service)["id"] == restored["id"]
assert status(service)["pause"] == "rollback"
time.sleep(1)
metrics["stages"].append("rollback suppressed the same digest across API restart and a full poll interval")
registry.tags["stable"] = fixtures[3][0]
active = wait_for(lambda: ready(service, fixtures[3][0]), timeout=180)
assert status(service)["pause"] is None
metrics["stages"].append("a newly observed digest resumed monitoring")
restored = rollback(service, active, original)
api.call("POST", path(service, "/auto-deploy/resume"))
active = wait_for(lambda: ready(service, fixtures[3][0]), timeout=180)
assert active["id"] != restored["id"]
api.call("POST", path(service, "/auto-deploy/pause"))
assert status(service)["pause"] == "manual"
api.call("PUT", path(service, "/auto-deploy"), None)
assert status(service)["policy"] is None
assert current(unrelated)["id"] == unrelated_original["id"]
metrics["stages"].append("explicit Resume deployed the same digest; pause/disable left the unrelated service unchanged")
metrics["authenticatedMetadataRequests"] = registry.authenticated
assert registry.authenticated > 0
finally:
stop.set()
if loader:
loader.join(timeout=10)
for service in services:
api.ready()
api.call("POST", path(service, "/stop"))
wait_for(lambda: current(service)["phase"] == "stopped")
api.call("DELETE", path(service))
metrics["durationSeconds"] = round(time.monotonic() - started, 3)
assert metrics["requests"] > 20 and not metrics["httpFailures"] and not metrics["monitorErrors"], metrics
assert metrics["maxLiveVMs"] <= 4 and metrics["maxLiveVCPUs"] <= 4, metrics
api.ready()
assert not api.call("GET", "/api/vms")
metrics["cleanup"] = "service VMs, policies, private TLS trust, hosts mapping and registry listener removed"
print(f"PASS authenticated automatic Deploy, rollback/restart pause and resume; {metrics['requests']} HTTP requests, zero errors", flush=True)