126 lines
3.3 KiB
Python
126 lines
3.3 KiB
Python
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
import json
|
|
|
|
from fastapi import FastAPI, Request
|
|
from pydantic import BaseModel
|
|
|
|
|
|
CONFIG_PATH = Path("config.json")
|
|
LOG_PATH = Path("activation.log")
|
|
|
|
app = FastAPI(title="Provisioning Activation Server", version="0.1.0")
|
|
|
|
|
|
class ActivationRequest(BaseModel):
|
|
activation_code: str
|
|
hostname: str | None = None
|
|
machine_id: str | None = None
|
|
client_version: str | None = None
|
|
|
|
|
|
def load_config() -> dict:
|
|
with CONFIG_PATH.open("r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def write_log(entry: dict) -> None:
|
|
with LOG_PATH.open("a", encoding="utf-8") as f:
|
|
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {
|
|
"status": "ok",
|
|
"service": "provisioning-server",
|
|
"version": "0.1.0"
|
|
}
|
|
|
|
|
|
@app.post("/api/v1/activate")
|
|
def activate(payload: ActivationRequest, request: Request):
|
|
config = load_config()
|
|
activation_entry = config.get("activation_codes", {}).get(payload.activation_code)
|
|
|
|
success = activation_entry is not None
|
|
|
|
write_log({
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"event": "activate",
|
|
"success": success,
|
|
"remote_ip": request.client.host if request.client else None,
|
|
"activation_code": payload.activation_code,
|
|
"hostname": payload.hostname,
|
|
"machine_id": payload.machine_id,
|
|
"client_version": payload.client_version
|
|
})
|
|
|
|
if not success:
|
|
return {
|
|
"success": False,
|
|
"error": "invalid_activation_code",
|
|
"message": "Der Aktivierungscode ist ungültig."
|
|
}
|
|
|
|
profiles = config.get("profiles", {})
|
|
available_profiles = [
|
|
profiles[profile_id]
|
|
for profile_id in activation_entry.get("profile_ids", [])
|
|
if profile_id in profiles
|
|
]
|
|
|
|
return {
|
|
"success": True,
|
|
"customer": activation_entry["customer"],
|
|
"profiles": available_profiles
|
|
}
|
|
|
|
|
|
@app.get("/api/v1/profiles/{profile_id}")
|
|
def get_profile(profile_id: str):
|
|
config = load_config()
|
|
profile = config.get("profiles", {}).get(profile_id)
|
|
|
|
if not profile:
|
|
return {
|
|
"success": False,
|
|
"error": "profile_not_found",
|
|
"message": "Das angeforderte Profil wurde nicht gefunden."
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"profile": profile
|
|
}
|
|
|
|
@app.get("/api/v1/profiles/{profile_id}/bootstrap")
|
|
def get_profile_bootstrap(profile_id: str):
|
|
config = load_config()
|
|
profile = config.get("profiles", {}).get(profile_id)
|
|
|
|
if not profile:
|
|
return {
|
|
"success": False,
|
|
"error": "profile_not_found",
|
|
"message": "Das angeforderte Profil wurde nicht gefunden."
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"profile": {
|
|
"id": profile["id"],
|
|
"label": profile["label"],
|
|
"distribution": profile["distribution"],
|
|
"version": profile["version"]
|
|
},
|
|
"installer": profile.get("installer"),
|
|
"ansible": {
|
|
"repo": profile.get("ansible_repo"),
|
|
"mode": "pull"
|
|
},
|
|
"ssh": {
|
|
"authorized_keys_url": "https://anode.stallinux.de/keys/default.pub"
|
|
}
|
|
}
|