Add read-only production runtime audit

This commit is contained in:
2026-08-17 11:17:40 +07:00
parent 057d4ed9dc
commit a1de4715a4
106 changed files with 6869 additions and 1782 deletions
+82
View File
@@ -0,0 +1,82 @@
"""Point the k3s practice cluster's ArgoCD Application at a freshly-built
image tag, then trigger an immediate sync.
Only touches `medical-chatbot-app` on the practice cluster
(argocd.realvuxbaro.me). Never touches production — the EC2 Compose
deployment isn't ArgoCD-managed at all.
Required env: ARGOCD_PRACTICE_URL, ARGOCD_PRACTICE_PASSWORD, IMAGE_TAG.
"""
from __future__ import annotations
import json
import os
import re
import sys
import urllib.error
import urllib.request
APP_NAME = "medical-chatbot-app"
IMAGES = ("vsf-duocthu-ai-service", "vsf-duocthu-web")
def call(base: str, method: str, path: str, token: str | None = None, body=None):
req = urllib.request.Request(
f"{base}{path}",
data=json.dumps(body).encode() if body is not None else None,
method=method,
headers={"Content-Type": "application/json"},
)
if token:
req.add_header("Authorization", f"Bearer {token}")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read()
return json.loads(raw) if raw else {}
except urllib.error.HTTPError as exc:
print(f"{method} {path} -> {exc.code}: {exc.read().decode(errors='replace')}", file=sys.stderr)
raise
def main() -> int:
base = os.environ["ARGOCD_PRACTICE_URL"].rstrip("/")
password = os.environ["ARGOCD_PRACTICE_PASSWORD"]
tag = os.environ["IMAGE_TAG"]
session = call(base, "POST", "/api/v1/session", body={"username": "admin", "password": password})
token = session["token"]
app = call(base, "GET", f"/api/v1/applications/{APP_NAME}", token=token)
values = app["spec"]["source"]["helm"]["values"]
for image in IMAGES:
pattern = re.compile(
rf"(repository:\s*ghcr\.io/baovu2k4/{re.escape(image)}\s*\n\s*tag:\s*)\S+"
)
values, count = pattern.subn(rf"\g<1>{tag}", values)
if count != 1:
print(f"Expected exactly one tag: line after {image}'s repository line, found {count}", file=sys.stderr)
return 1
app["spec"]["source"]["helm"]["values"] = values
call(base, "PUT", f"/api/v1/applications/{APP_NAME}", token=token, body=app)
# selfHeal (syncPolicy.automated) reacts to the PUT above on its own —
# often before this explicit call lands, which then 400s with "another
# operation is already in progress". That race means the sync we wanted
# is already happening; only a genuinely different failure is fatal.
try:
call(base, "POST", f"/api/v1/applications/{APP_NAME}/sync", token=token, body={})
except urllib.error.HTTPError as exc:
if exc.code == 400:
print(f"Explicit sync raced with autosync (expected under selfHeal) — continuing.")
else:
raise
print(f"{APP_NAME} pointed at tag {tag}; sync in progress.")
return 0
if __name__ == "__main__":
raise SystemExit(main())