ADR-0046 Schritt 1: Spiegel auf anode, Archiv je Stand, ansible_stand im Check-in

- Hintergrund-Thread spiegelt ansible-content (vorab), -stabil und
  provisioning-agent alle 60 s nach data/spiegel, Archiv (tar.gz + SHA-256)
  je Stand, die letzten drei je Stufe bleiben abrufbar
- Check-in liefert zusätzlich ansible_stand {commit, sha256}; ansible_repo
  bleibt als Rückfall
- GET /api/v1/agent/ansible/<commit>.tar.gz und
  /api/v1/agent/fassung/<commit>/<datei> nur mit Agent-Secret
  (Authorization + X-Tuxflotte-Geraet), nur zustehende Stände

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
adminandClaude Opus 5.5 committed 2026-10-20 19:28:15 +02:00
1 parent d90ce99180
commit 2fae14cfe0
1 file changed
+218 -5
+218 -5
View File
@@ -7,6 +7,7 @@ import mimetypes
import os
import re
import secrets
import shutil
import subprocess
import threading
import time
@@ -23,7 +24,7 @@ from cryptography.fernet import Fernet, InvalidToken
from fastapi import FastAPI, Header, HTTPException, Request
from pydantic import BaseModel
from fastapi.responses import FileResponse
from fastapi.responses import FileResponse, Response
LOG_PATH = Path("activation.log")
@@ -48,6 +49,23 @@ ANSIBLE_CONTENT_REPO_STABIL = os.environ.get("TUXFLOTTE_ANSIBLE_CONTENT_REPO_STA
re.sub(r"(\.git)?$", r"-stabil\1", ANSIBLE_CONTENT_REPO, count=1) if ANSIBLE_CONTENT_REPO else ""
)
AUSROLLSTUFEN = ("stabil", "vorab")
# Spiegel für Geräte (ADR-0046, 20.10.2026): anode hält Spiegel von
# ansible-content (je Ausrollstufe) und provisioning-agent, erzeugt je Stand
# ein Archiv und liefert beides nur gegen das Agent-Secret aus. Damit hängen
# die Geräte nicht mehr an einem öffentlich erreichbaren Gitea. Im Übergang
# schickt der Check-in weiter auch "ansible_repo" (Rückfall per Git).
AGENT_REPO = os.environ.get("TUXFLOTTE_AGENT_REPO", "") or (
re.sub(r"ansible-content(\.git)?$", r"provisioning-agent\1", ANSIBLE_CONTENT_REPO, count=1)
if ANSIBLE_CONTENT_REPO else ""
)
SPIEGEL_DIR = Path(os.environ.get("TUXFLOTTE_SPIEGEL_DIR", "data/spiegel")).resolve()
SPIEGEL_INTERVALL_SECONDS = 60
SPIEGEL_GIT_TIMEOUT_SECONDS = 120
# Je Stufe ausgelieferte Stände: der aktuelle und zwei vorige - ein Gerät,
# das seinen Stand kurz vor einer Freigabe erfahren hat, kann ihn noch holen.
SPIEGEL_STAENDE_JE_STUFE = 3
# Dateien, die Agent-Wechsel und Neuinstallation aus provisioning-agent holen.
AGENT_FASSUNG_DATEIEN = ("agent.py", "firstboot.sh", "tuxflotte-agent.service", "tuxflotte-firstboot.service")
ADMIN_TOKEN = os.environ.get("TUXFLOTTE_ADMIN_TOKEN", "")
KUNDENPLATTFORM_TOKEN = os.environ.get("TUXFLOTTE_KUNDENPLATTFORM_TOKEN", "")
AGENT_REPORTABLE_EVENT_TYPES = {"applied", "apply_failed", "removed", "remove_failed"}
@@ -134,6 +152,12 @@ def start_iso_build_expiry_sweep() -> None:
threading.Thread(target=run_iso_build_expiry_sweep, daemon=True).start()
@app.on_event("startup")
def start_spiegel() -> None:
spiegel_stand_laden()
threading.Thread(target=run_spiegel_schleife, daemon=True).start()
class ActivationRequest(BaseModel):
activation_code: str
device_fingerprint: str
@@ -3772,9 +3796,8 @@ def fetch_organizations():
) in cur.fetchall()
]
def ansible_repo_fuer_geraet(device_id: str) -> str:
"""Adresse des Rollen-Repos nach der Ausrollstufe der Organisation.
Unbekanntes Gerät: der freigegebene Stand."""
def ausrollstufe_fuer_geraet(device_id: str) -> str:
"""Ausrollstufe der Organisation des Geräts. Unbekanntes Gerät: stabil."""
with get_database_connection() as conn:
with conn.cursor() as cur:
@@ -3788,7 +3811,150 @@ def ansible_repo_fuer_geraet(device_id: str) -> str:
)
row = cur.fetchone()
return ANSIBLE_CONTENT_REPO if row and row[0] == "vorab" else ANSIBLE_CONTENT_REPO_STABIL
return "vorab" if row and row[0] == "vorab" else "stabil"
def ansible_repo_fuer_geraet(device_id: str) -> str:
"""Adresse des Rollen-Repos nach der Ausrollstufe der Organisation."""
return ANSIBLE_CONTENT_REPO if ausrollstufe_fuer_geraet(device_id) == "vorab" else ANSIBLE_CONTENT_REPO_STABIL
# --- Spiegel für Geräte (ADR-0046) -------------------------------------------
# Zustand je Ausrollstufe: {"commit", "sha256", "erlaubt": [neueste zuerst]}.
# Liegt zusätzlich in stand.json, damit nach einem Neustart sofort
# ausgeliefert wird, auch wenn Gitea gerade nicht erreichbar ist.
_spiegel_lock = threading.Lock()
_spiegel_staende: dict = {}
def _spiegel_repos() -> dict:
return {"vorab": ANSIBLE_CONTENT_REPO, "stabil": ANSIBLE_CONTENT_REPO_STABIL, "agent": AGENT_REPO}
def _spiegel_git(*args, binaer: bool = False):
ergebnis = subprocess.run(
["git", *args],
capture_output=True,
timeout=SPIEGEL_GIT_TIMEOUT_SECONDS,
env={**os.environ, "GIT_TERMINAL_PROMPT": "0"},
check=True,
)
return ergebnis.stdout if binaer else ergebnis.stdout.decode().strip()
def spiegel_stand_laden() -> None:
try:
daten = json.loads((SPIEGEL_DIR / "stand.json").read_text())
except (OSError, ValueError):
return
with _spiegel_lock:
_spiegel_staende.update({k: v for k, v in daten.items() if k in AUSROLLSTUFEN})
def _spiegel_holen(name: str, url: str) -> Path:
"""Bare-Spiegel anlegen oder aktualisieren, liefert seinen Pfad."""
pfad = SPIEGEL_DIR / f"{name}.git"
if (pfad / "HEAD").exists():
if _spiegel_git("-C", str(pfad), "remote", "get-url", "origin") != url:
_spiegel_git("-C", str(pfad), "remote", "set-url", "origin", url)
_spiegel_git("-C", str(pfad), "remote", "update", "--prune")
else:
neu = SPIEGEL_DIR / f".{name}.git.neu"
if neu.exists():
shutil.rmtree(neu)
_spiegel_git("clone", "--quiet", "--mirror", url, str(neu))
neu.rename(pfad)
return pfad
def _spiegel_archiv(stufe: str, pfad: Path) -> dict:
"""Archiv des aktuellen Stands einer Stufe (einmal je Commit) und die
Liste der auslieferbaren Stände; ältere Archive werden gelöscht."""
commit = _spiegel_git("-C", str(pfad), "rev-parse", "HEAD^{commit}")
archiv_dir = SPIEGEL_DIR / "archive" / stufe
archiv_dir.mkdir(parents=True, exist_ok=True)
archiv = archiv_dir / f"{commit}.tar.gz"
if not archiv.exists():
neu = archiv_dir / f".{commit}.tar.gz.neu"
_spiegel_git("-C", str(pfad), "archive", "--format=tar.gz", "-o", str(neu), commit)
neu.rename(archiv)
sha256 = hashlib.sha256(archiv.read_bytes()).hexdigest()
with _spiegel_lock:
bisher = _spiegel_staende.get(stufe, {}).get("erlaubt", [])
erlaubt = [commit] + [c for c in bisher if c != commit][: SPIEGEL_STAENDE_JE_STUFE - 1]
for datei in archiv_dir.glob("*.tar.gz"):
if datei.name.removesuffix(".tar.gz") not in erlaubt:
datei.unlink()
return {"commit": commit, "sha256": sha256, "erlaubt": erlaubt}
def spiegel_aktualisieren() -> None:
"""Ein Durchlauf: alle Spiegel holen, Archive erzeugen, Stand sichern.
Ein Fehler bei einem Repo lässt die anderen und den letzten guten Stand
dieser Stufe unberührt."""
SPIEGEL_DIR.mkdir(parents=True, exist_ok=True)
for name, url in _spiegel_repos().items():
if not url:
continue
try:
pfad = _spiegel_holen(name, url)
if name in AUSROLLSTUFEN:
stand = _spiegel_archiv(name, pfad)
with _spiegel_lock:
vorher = _spiegel_staende.get(name, {}).get("commit")
_spiegel_staende[name] = stand
if vorher != stand["commit"]:
print(f"[spiegel] {name}: {str(vorher)[:7]} -> {stand['commit'][:7]}")
except (subprocess.SubprocessError, OSError) as exc:
fehler = exc.stderr.decode(errors="replace").strip() if getattr(exc, "stderr", None) else exc
print(f"[spiegel] {name}: Fehler, letzter Stand bleibt: {fehler}")
with _spiegel_lock:
daten = json.dumps(_spiegel_staende, indent=2)
neu = SPIEGEL_DIR / ".stand.json.neu"
neu.write_text(daten)
neu.rename(SPIEGEL_DIR / "stand.json")
def run_spiegel_schleife() -> None:
"""Daemon-Thread wie run_iso_build_expiry_sweep(): ein Fehler darf die
Schleife nicht beenden."""
while True:
try:
spiegel_aktualisieren()
except Exception as exc:
print(f"[spiegel] Durchlauf-Fehler (nächster Versuch in {SPIEGEL_INTERVALL_SECONDS}s): {exc}")
time.sleep(SPIEGEL_INTERVALL_SECONDS)
def ansible_stand_fuer_geraet(device_id: str) -> dict | None:
"""Stand für den Check-in: Commit und Prüfsumme des Archivs der
Ausrollstufe. None, solange der Spiegel noch nichts hat - dann bleibt
dem Agent der Git-Weg über "ansible_repo"."""
stufe = ausrollstufe_fuer_geraet(device_id)
with _spiegel_lock:
stand = _spiegel_staende.get(stufe)
if not stand:
return None
return {"commit": stand["commit"], "sha256": stand["sha256"]}
def _geraet_aus_kopfzeilen(authorization: str | None, geraet: str | None) -> str:
"""Gerät für die GET-Endpunkte des Spiegels: Secret im Authorization-
Header, Geräte-ID in X-Tuxflotte-Geraet (nie in der URL)."""
if not authorization or not authorization.startswith("Bearer ") or not geraet or not _ist_uuid(geraet):
raise HTTPException(status_code=401, detail="Nicht angemeldet.")
if not verify_agent_secret(geraet, authorization.removeprefix("Bearer ")):
raise HTTPException(status_code=401, detail="Nicht angemeldet.")
return geraet
def set_ausrollstufe(organization_id: str, ausrollstufe: str) -> bool:
@@ -7222,6 +7388,9 @@ def agent_checkin(
# wenn er sich vom lokal tatsächlich gesetzten unterscheidet.
"hostname": hostname,
"ansible_repo": ansible_repo_fuer_geraet(payload.device_id),
# ADR-0046: Stand als Archiv von anode. Neuere Agenten nehmen ihn
# statt "ansible_repo", ältere ignorieren das Feld.
"ansible_stand": ansible_stand_fuer_geraet(payload.device_id),
# Persönliche Konten (ADR-0041, Schritt 3). Ältere Agenten ignorieren
# das Feld; die Rolle fasst nur hier genannte Konten an.
"persoenliche_konten": fetch_persoenliche_konten_fuer_agent(payload.device_id),
@@ -7229,6 +7398,50 @@ def agent_checkin(
}
@app.get("/api/v1/agent/ansible/{datei}")
def agent_ansible_archiv(
datei: str,
authorization: str | None = Header(default=None),
x_tuxflotte_geraet: str | None = Header(default=None),
):
"""Archiv eines ansible-content-Stands (ADR-0046) - nur ein Stand, der der
Ausrollstufe dieses Geräts zusteht."""
device_id = _geraet_aus_kopfzeilen(authorization, x_tuxflotte_geraet)
treffer = re.fullmatch(r"([0-9a-f]{40})\.tar\.gz", datei)
stufe = ausrollstufe_fuer_geraet(device_id)
with _spiegel_lock:
erlaubt = _spiegel_staende.get(stufe, {}).get("erlaubt", [])
if not treffer or treffer.group(1) not in erlaubt:
raise HTTPException(status_code=404, detail="Unbekannter Stand.")
archiv = SPIEGEL_DIR / "archive" / stufe / datei
if not archiv.is_file():
raise HTTPException(status_code=404, detail="Unbekannter Stand.")
return FileResponse(archiv, media_type="application/gzip")
@app.get("/api/v1/agent/fassung/{commit}/{datei}")
def agent_fassung(
commit: str,
datei: str,
authorization: str | None = Header(default=None),
x_tuxflotte_geraet: str | None = Header(default=None),
):
"""Eine Datei des Agenten in einer bestimmten Fassung (ADR-0046) - nur
Commits auf dem Hauptzweig von provisioning-agent, nur feste Dateien."""
_geraet_aus_kopfzeilen(authorization, x_tuxflotte_geraet)
pfad = SPIEGEL_DIR / "agent.git"
if not re.fullmatch(r"[0-9a-f]{40}", commit) or datei not in AGENT_FASSUNG_DATEIEN or not (pfad / "HEAD").exists():
raise HTTPException(status_code=404, detail="Unbekannte Fassung.")
try:
_spiegel_git("-C", str(pfad), "merge-base", "--is-ancestor", commit, "HEAD")
inhalt = _spiegel_git("-C", str(pfad), "show", f"{commit}:{datei}", binaer=True)
except (subprocess.SubprocessError, OSError):
raise HTTPException(status_code=404, detail="Unbekannte Fassung.")
return Response(content=inhalt, media_type="text/plain; charset=utf-8")
@app.post("/api/v1/agent/report")
def agent_report(
payload: AgentReportRequest,