bin/promote-online-production.py
#!/usr/bin/env python3
"""Promote an already verified local production release to the explicit online target."""
from __future__ import annotations
import argparse
import json
import re
import shlex
import shutil
import subprocess
import sys
import tarfile
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from urllib.request import urlopen
ENVIRONMENT_MARKER = re.compile(
r'<p class="environment-marker" role="status">(?:Testumgebung|Lokale Produktion)</p>'
)
def fail(message: str) -> None:
raise RuntimeError(message)
def run(command: list[str], *, capture: bool = False) -> str:
result = subprocess.run(command, text=True, stdout=subprocess.PIPE if capture else None,
stderr=subprocess.STDOUT if capture else None)
if result.returncode:
fail("command failed: " + " ".join(command[:2]))
return result.stdout or ""
def read_config(path: Path) -> dict:
if not path.is_file():
fail("Online-Konfiguration fehlt: " + str(path))
if path.stat().st_mode & 0o077:
fail("Online-Konfiguration muss Modus 0600 haben: " + str(path))
value = json.loads(path.read_text(encoding="utf-8"))
required = (("ssh", "host"), ("ssh", "user"), ("remote", "root"),
("remote", "serviceName"), ("remote", "healthUrl"), ("public", "baseUrl"))
for section, key in required:
if not isinstance(value.get(section, {}).get(key), str) or "SET_" in value[section][key]:
fail("Online-Konfiguration enthält keinen gültigen Wert für " + section + "." + key)
if value["public"]["baseUrl"] != "https://wissen.guggat.de":
fail("Online-Ziel muss https://wissen.guggat.de sein")
if not re.fullmatch(r"/[A-Za-z0-9_./-]+", value["remote"]["root"]):
fail("Unsicherer Remote-Installationspfad")
return value
def latest_local_proof(runtime: Path) -> Path:
candidates = sorted((runtime / "production" / "reports" / "promotion").glob("*.json"), reverse=True)
for candidate in candidates:
try:
if json.loads(candidate.read_text(encoding="utf-8")).get("status") == "PASSED":
return candidate
except json.JSONDecodeError:
continue
fail("Kein erfolgreicher lokaler Produktionsnachweis gefunden; zuerst promote-local-production.py ausführen")
def public_landing_url(config: dict) -> str:
landing_path = config["public"].get("landingPath", "/reports/")
if not isinstance(landing_path, str) or not re.fullmatch(r"/[A-Za-z0-9_./-]*", landing_path):
fail("Online-Konfiguration enthält keinen sicheren public.landingPath")
return config["public"]["baseUrl"].rstrip("/") + landing_path
def stage_online_web(source: Path, destination: Path) -> None:
"""Copy the verified local web release and remove local-only environment labels."""
shutil.copytree(source, destination)
for page in destination.rglob("*.html"):
content = page.read_text(encoding="utf-8")
online_content = ENVIRONMENT_MARKER.sub("", content)
if online_content != content:
page.write_text(online_content, encoding="utf-8")
def ssh_base(config: dict) -> list[str]:
ssh = config["ssh"]
command = ["ssh", "-p", str(ssh.get("port", 22)), "-o", "BatchMode=yes"]
if ssh.get("knownHostsFile"):
command += ["-o", "UserKnownHostsFile=" + str(Path(ssh["knownHostsFile"]).expanduser())]
if ssh.get("identityFile"):
command += ["-i", str(Path(ssh["identityFile"]).expanduser())]
command.append(f'{ssh["user"]}@{ssh["host"]}')
return command
def remote(config: dict, script: str, *, capture: bool = False) -> str:
return run(ssh_base(config) + ["sh -lc " + shlex.quote(script)], capture=capture)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--engineering-home", type=Path, default=Path.home() / "engineering")
parser.add_argument("--config", type=Path, default=Path.home() / ".config/netzwerksolution/promotion/production-online.json")
parser.add_argument("--roundtrip-id", default="manual")
args = parser.parse_args()
home = args.engineering_home.expanduser().resolve()
runtime = home / "runtime"
config = read_config(args.config.expanduser())
proof = latest_local_proof(runtime)
install = runtime / "production" / "installations" / "engineering-platform"
landing_source = runtime / "public" / "guggat-landing"
if not (install / "bin" / "engineering-platform").is_file() or not (install / "web" / "reports" / "index.html").is_file():
fail("Lokale Produktionsinstallation oder Reports fehlen")
marker = json.loads((install / "web" / "deployment-marker.json").read_text(encoding="utf-8"))
deployment_id = marker.get("deploymentId")
if not isinstance(deployment_id, str) or not deployment_id:
fail("Lokaler Versionsnachweis deployment-marker.json fehlt")
if not (landing_source / "index.html").is_file():
fail("Versionierte Server-Landing-Page fehlt: " + str(landing_source / "index.html"))
evidence_dir = runtime / "production" / "reports" / "promotion-online"
evidence_dir.mkdir(parents=True, exist_ok=True)
release_id = datetime.now(timezone.utc).strftime("ONLINE-%Y%m%d-%H%M%S")
landing_url = public_landing_url(config)
evidence = {"promotionId": release_id, "target": "production-online", "status": "FAILED",
"roundtripId": args.roundtrip_id, "localProof": str(proof), "publicUrl": config["public"]["baseUrl"],
"publicLandingUrl": landing_url}
root = config["remote"]["root"].rstrip("/")
previous = ""
try:
remote(config, f"mkdir -p '{root}/releases/{release_id}'")
with tempfile.TemporaryDirectory(prefix="online-release-") as tmp:
archive = Path(tmp) / "release.tar.gz"
staged_web = Path(tmp) / "web"
stage_online_web(install / "web", staged_web)
with tarfile.open(archive, "w:gz") as tar:
tar.add(install / "bin", arcname="bin")
tar.add(staged_web, arcname="web")
tar.add(landing_source, arcname="landing")
destination = f'{config["ssh"]["user"]}@{config["ssh"]["host"]}:{root}/releases/{release_id}/release.tar.gz'
copy = ["scp", "-P", str(config["ssh"].get("port", 22)), "-o", "BatchMode=yes"]
if config["ssh"].get("knownHostsFile"):
copy += ["-o", "UserKnownHostsFile=" + str(Path(config["ssh"]["knownHostsFile"]).expanduser())]
if config["ssh"].get("identityFile"):
copy += ["-i", str(Path(config["ssh"]["identityFile"]).expanduser())]
run(copy + [str(archive), destination])
previous = remote(config, f"readlink '{root}/current' || true", capture=True).strip()
remote(config, f"cd '{root}/releases/{release_id}' && tar -xzf release.tar.gz && rm release.tar.gz && ln -sfn 'releases/{release_id}' '{root}/current.new' && mv -Tf '{root}/current.new' '{root}/current' && systemctl --user restart '{config['remote']['serviceName']}'")
remote(config, f"curl --fail --silent --show-error '{config['remote']['healthUrl']}' > /dev/null")
with urlopen(landing_url, timeout=15) as response:
page = response.read().decode("utf-8", errors="replace")
if config["public"].get("landingNeedle", "Elevator Pitch") not in page:
fail("Öffentliche Landing Page enthält den erwarteten Nachweis nicht")
with urlopen(config["public"]["baseUrl"].rstrip("/") + "/deployment-marker.json", timeout=15) as response:
public_marker = json.loads(response.read().decode("utf-8"))
if public_marker.get("deploymentId") != deployment_id:
fail("Öffentlicher Versionsnachweis stimmt nicht mit der lokalen Produktion überein")
server_landing_url = config["public"].get("serverLandingUrl", "https://guggat.de/")
with urlopen(server_landing_url, timeout=15) as response:
server_landing = response.read().decode("utf-8", errors="replace")
if "Betriebszugänge" not in server_landing or "Impressum von miradlo" not in server_landing:
fail("Öffentliche Server-Landing-Page enthält die erwarteten Nachweise nicht")
evidence.update({"status": "PASSED", "remotePreviousRelease": previous,
"remoteRelease": f"releases/{release_id}", "deploymentId": deployment_id})
except Exception as exc:
evidence["error"] = str(exc)
if previous:
try:
remote(config, f"ln -sfn '{previous}' '{root}/current.new' && mv -Tf '{root}/current.new' '{root}/current' && systemctl --user restart '{config['remote']['serviceName']}'")
evidence["rollback"] = "PASSED"
except Exception as rollback_error:
evidence["rollback"] = "MANUAL_RECOVERY_REQUIRED"
evidence["rollbackError"] = str(rollback_error)
raise
finally:
(evidence_dir / f"{release_id}.json").write_text(json.dumps(evidence, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(evidence_dir / f"{release_id}.json")
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except RuntimeError as error:
print("ERROR: " + str(error), file=sys.stderr)
raise SystemExit(1)