#!/usr/bin/env python3
"""Local HTTP/Unix-socket control service for the FEDEO host updater. No TCP listener."""
import argparse
import contextlib
import fcntl
from http.server import BaseHTTPRequestHandler
import importlib.util
import json
import os
from pathlib import Path
import re
import signal
import socketserver
import stat
import threading
import traceback
import urllib.parse

spec = importlib.util.spec_from_file_location("fedeo_updater", Path(__file__).with_name("selfhost-update.py"))
engine = importlib.util.module_from_spec(spec)
spec.loader.exec_module(engine)

REQUEST_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{7,79}\Z")
PHASES = {"preparing", "validating", "checked", "pulling", "maintenance", "backed_up", "applying",
          "retiring_matrix", "migrating", "starting", "verifying", "complete", "resolved"}
STATUSES = {"idle", "queued", "running", "checked", "up_to_date", "succeeded", "failed", "interrupted", "recovered"}
ERRORS = {
    "INVALID_REQUEST": "Ungültige Anfrage.",
    "NOT_FOUND": "Auftrag oder Endpunkt nicht gefunden.",
    "BUSY": "Eine Prüfung oder ein Update läuft bereits.",
    "SHUTTING_DOWN": "Der Dienst wird beendet.",
    "REQUEST_CONFLICT": "Diese Auftrags-ID wurde für eine andere Anfrage verwendet.",
    "CHECK_REQUIRED": "Zuerst genau diese Zielversion erfolgreich prüfen.",
    "COMPOSE_REVIEW_REQUIRED": "Die Compose-Datei muss zuerst auf dem Host geprüft und übernommen werden.",
    "RECOVERY_REQUIRED": "Ein vorheriges Update benötigt Wiederherstellung auf dem Host.",
    "PLAN_CHANGED": "Release oder Konfiguration haben sich geändert. Erneut prüfen.",
    "UPDATE_FAILED": "Auftrag fehlgeschlagen.",
    "LOGS_UNAVAILABLE": "Das Update-Protokoll kann gerade nicht gelesen werden.",
    "INTERRUPTED": "Der Auftrag wurde unterbrochen. Status auf dem Host prüfen.",
}


class ApiError(Exception):
    def __init__(self, code, status=400):
        self.code, self.status = code, status


def identifier(value):
    if not isinstance(value, str) or not REQUEST_ID.fullmatch(value):
        raise ApiError("INVALID_REQUEST")
    return value


def public_state(state):
    """An allowlist, never raw command logs, exceptions, host paths, config or diff contents."""
    result = {"status": state.get("status") if state.get("status") in STATUSES else "idle"}
    if state.get("phase") in PHASES:
        result["phase"] = state["phase"]
    for key in ("recoveryRequired", "maintenance", "mutationStarted"):
        if key in state:
            result[key] = bool(state[key])
    for key in ("version", "previousVersion"):
        if isinstance(state.get(key), str) and engine.VERSION.fullmatch(state[key]):
            result[key] = state[key]
    if re.fullmatch(r"[a-f0-9]{40}", str(state.get("revision", ""))):
        result["revision"] = state["revision"]
    for key in ("startedAt", "updatedAt"):
        if isinstance(state.get(key), str) and re.fullmatch(r"[0-9T:+.Z-]{10,40}", state[key]):
            result[key] = state[key]
    if state.get("requestId") and REQUEST_ID.fullmatch(str(state["requestId"])):
        result["requestId"] = state["requestId"]
    if state.get("command") in ("check", "apply"):
        result["command"] = state["command"]
    return result


def public_plan(plan):
    result = {k: v for k, v in public_state(plan).items() if k != "status"}
    for key in ("updateAvailable", "requiresComposeAdoption"):
        result[key] = bool(plan.get(key))
    if type(plan.get("estimatedBackupBytes")) is int:
        result["estimatedBackupBytes"] = plan["estimatedBackupBytes"]
    result["retiredServices"] = sorted(set(plan.get("retiredServices", [])) & engine.LEGACY_MATRIX_SERVICES)
    return result


LOG_BYTES = 6000
SECRET_KEY = r"[\w.-]*(?:password|passwd|secret|token|credential|authorization|(?:api|private|access|encryption|signing)[_-]?key|database[_-]?url|dsn|auth)[\w.-]*"
QUOTED_VALUE = r"\"(?:\\.|[^\"\\])*\"|'[^']*'"
SECRET_VALUE = re.compile(r"(?im)(?<![\w.-])([\"']?" + SECRET_KEY + r"[\"']?\s*[:=]\s*)(" + QUOTED_VALUE + r"|[^\r\n,}\]]+)")
CONFIG_SECRET = re.compile(r"(?im)^\s*(?:export\s+|-\s*)?[\"']?" + SECRET_KEY + r"[\"']?\s*[:=]\s*(" + QUOTED_VALUE + r"|[^\r\n]+)")


def read_private(path, *, tail=False):
    """Read only regular files, never follow a log/config symlink or block on a FIFO."""
    try:
        fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
    except FileNotFoundError:
        return "", False
    with os.fdopen(fd, "rb") as stream:
        size = os.fstat(stream.fileno()).st_size
        if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
            raise OSError("Not a regular file")
        limit = 32768 if tail else 1024 * 1024
        if size > limit and not tail:
            raise OSError("Configuration too large to redact safely")
        truncated = size > limit
        if truncated:
            stream.seek(-limit, os.SEEK_END)
            stream.readline(limit)  # Never expose a suffix of a secret on a cut line.
        return stream.read(limit).decode("utf-8", errors="replace"), truncated


def log_redactor(directory, job):
    # Include original values after an update changed the active configuration.
    paths = [directory / name for name in (".env", "docker-compose.yml", "docker-compose.override.yml", "docker-compose.override.yaml")]
    if job:
        paths += [job / "target.env", job / "target.compose.yml"]
        paths += [job / "backup/config" / name for name in (".env", "docker-compose.yml", "docker-compose.override.yml", "docker-compose.override.yaml")]
    values = {value for key, value in os.environ.items() if re.fullmatch(SECRET_KEY, key, re.I) and value}
    for path in paths:
        content, _ = read_private(path)
        for match in CONFIG_SECRET.finditer(content):
            value = match[1].strip()
            if value[:1] in ("'", '"'):
                if value.startswith('"'):
                    try:
                        values.add(json.loads(value))
                    except ValueError:
                        pass
                value = value[1:-1]
            else:
                value = value.split(" #", 1)[0].strip()
            if value:
                values.add(value)
        for match in re.finditer(r"[a-z][a-z0-9+.-]*://[^\s\"']+", content, re.I):
            password = urllib.parse.urlsplit(match[0]).password
            if password:
                values.update((password, urllib.parse.unquote(password)))
    values |= {urllib.parse.quote(value, safe="") for value in values}

    def redact(content):
        content = re.sub(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\))", "", content)
        for value in sorted(values, key=len, reverse=True):
            content = content.replace(value, "[REDACTED]")
        content = re.sub(r"(?is)-----BEGIN [^-]*PRIVATE KEY-----.*?(?:-----END [^-]*PRIVATE KEY-----|\Z)", "[REDACTED]", content)
        content = re.sub(r"(?i)([a-z][a-z0-9+.-]*://)[^\s/]*@", r"\1[REDACTED]@", content)
        content = re.sub(r"(?i)\b(Bearer|Basic|token)\s+[A-Za-z0-9._~+/=-]+", r"\1 [REDACTED]", content)
        content = SECRET_VALUE.sub(lambda m: m[1] + "[REDACTED]", content)
        content = re.sub(r"(?i)(--" + SECRET_KEY + r"\s+)(" + QUOTED_VALUE + r"|\S+)", r"\1[REDACTED]", content)
        return re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", content)
    return redact


class Controller:
    def __init__(self, directory, *, updater_factory=engine.Updater):
        self.directory = Path(directory).resolve()
        self.factory = updater_factory
        updater = self.factory(self.directory)
        self.home = updater.home / "service"
        self.home.mkdir(mode=0o700, exist_ok=True)
        if self.home.is_symlink():
            raise engine.UpdateError("Dienstverzeichnis darf kein symbolischer Link sein.")
        os.chmod(self.home, 0o700)
        self.gate = threading.Lock()
        self.worker = None
        self.closing = False
        self.active = None
        self.daemon_lock = (self.home / "daemon.lock").open("a")
        try:
            fcntl.flock(self.daemon_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            self.daemon_lock.close()
            raise engine.UpdateError("Für diese Instanz läuft bereits ein Update-Dienst.") from None
        # A persisted request is an audit record, not a queue that silently replays updates.
        try:
            for path in self.home.glob("*.json"):
                record = engine.read_json(path)
                if record.get("status") in ("queued", "running"):
                    record.update(status="interrupted", errorCode="INTERRUPTED", updatedAt=engine.now())
                    engine.write_json(path, record)
                if (REQUEST_ID.fullmatch(str(record.get("requestId", "")))
                        and (self.active is None or record.get("startedAt", "") > self.read(self.active).get("startedAt", ""))):
                    self.active = record["requestId"]
            updater.status()  # Detect an abandoned engine job without touching an active CLI job.
        except BaseException:
            self.daemon_lock.close()
            raise

    def path(self, request_id):
        return self.home / (identifier(request_id) + ".json")

    def read(self, request_id):
        record = engine.read_json(self.path(request_id))
        if not record:
            raise ApiError("NOT_FOUND", 404)
        return record

    def request_status(self, request_id):
        record = self.read(request_id)
        result = public_state(record)
        if record.get("plan"):
            result["plan"] = public_plan(record["plan"])
        code = record.get("errorCode")
        if code in ERRORS:
            result["error"] = {"code": code, "message": ERRORS[code]}
        if record["status"] == "running":
            state = self.factory(self.directory).status()
            if state.get("requestId") == request_id:
                result["progress"] = public_state(state)
        return result

    def request_logs(self, request_id):
        record = self.read(request_id)
        try:
            home = self.home.parent
            state = engine.read_json(home / "status.json", {})
            candidates = [Path(record["job"])] if record.get("job") else []
            if state.get("requestId") == request_id and state.get("job"):
                candidates.append(Path(state["job"]))
            # Older service versions did not persist the engine-job association.
            if not candidates:
                candidates = [path.parent for path in home.glob("*/status.json")]
            job = None
            for candidate in candidates:
                if candidate.parent != home or candidate.is_symlink() or not candidate.is_dir():
                    continue
                content, _ = read_private(candidate / "status.json")
                if content and json.loads(content).get("requestId") == request_id:
                    job = candidate
                    break
            redact = log_redactor(self.directory, job)
            files = [("service", self.home / (request_id + ".log"))]
            if job:
                files += [("events", job / "events.jsonl"), ("commands", job / "commands.log")]
            logs = []
            for source, path in files:
                content, truncated = read_private(path, tail=True)
                content = redact(content)
                encoded = content.encode("utf-8")
                if len(encoded) > LOG_BYTES:
                    content = encoded[-LOG_BYTES:].decode("utf-8", errors="ignore")
                    truncated = True
                if content or truncated:
                    logs.append({"source": source, "content": content, "truncated": truncated})
            return {"requestId": request_id, "logs": logs}
        except (OSError, ValueError, TypeError):
            raise ApiError("LOGS_UNAVAILABLE", 503) from None

    def status(self):
        updater = self.factory(self.directory)
        result = {"apiVersion": 1, "engine": public_state(updater.status()), "acceptingRequests": not self.closing,
                  "capabilities": {"requestLogs": True}}
        installed = engine.read_json(updater.home / "installed.json", {}).get("manifest", {})
        result["installed"] = {k: v for k, v in public_state({"version": installed.get("version"),
                              "revision": installed.get("commit")}).items() if k != "status"}
        if self.active:
            result["request"] = self.request_status(self.active)
        return result

    def submit(self, command, body):
        if not isinstance(body, dict) or command not in ("check", "apply"):
            raise ApiError("INVALID_REQUEST")
        allowed = {"requestId", "version"} | ({"checkRequestId"} if command == "apply" else set())
        if set(body) - allowed:
            raise ApiError("INVALID_REQUEST")
        request_id = identifier(body.get("requestId"))
        version = body.get("version")
        if version is not None and (not isinstance(version, str) or not engine.VERSION.fullmatch(version)):
            raise ApiError("INVALID_REQUEST")
        if command == "apply" and not version:
            raise ApiError("INVALID_REQUEST")
        check_id = identifier(body.get("checkRequestId")) if command == "apply" else None
        payload = {"command": command, "version": version, "checkRequestId": check_id}
        with self.gate:
            previous = engine.read_json(self.path(request_id))
            if previous:
                if previous["input"] != payload:
                    raise ApiError("REQUEST_CONFLICT", 409)
                return 200, self.request_status(request_id)
            if self.closing:
                raise ApiError("SHUTTING_DOWN", 503)
            if self.worker and self.worker.is_alive():
                raise ApiError("BUSY", 409)
            state = self.factory(self.directory).status()
            if state.get("recoveryRequired"):
                raise ApiError("RECOVERY_REQUIRED", 409)
            if state.get("status") == "running":
                raise ApiError("BUSY", 409)
            fingerprint = None
            if command == "apply":
                try:
                    check = self.read(check_id)
                except ApiError:
                    raise ApiError("CHECK_REQUIRED", 409) from None
                plan = check.get("plan", {})
                if (check.get("command") != "check" or check.get("status") != "checked"
                        or plan.get("version") != version or not plan.get("updateAvailable")
                        or not re.fullmatch(r"[a-f0-9]{64}", str(plan.get("approvalFingerprint", "")))):
                    raise ApiError("CHECK_REQUIRED", 409)
                if plan.get("requiresComposeAdoption"):
                    raise ApiError("COMPOSE_REVIEW_REQUIRED", 409)
                fingerprint = plan["approvalFingerprint"]
            record = {"requestId": request_id, "command": command, "input": payload,
                      "status": "queued", "startedAt": engine.now(), "updatedAt": engine.now()}
            engine.write_json(self.path(request_id), record)
            self.active = request_id
            self.worker = threading.Thread(target=self.run, args=(record, fingerprint), daemon=False)
            try:
                self.worker.start()
            except BaseException:
                record.update(status="failed", errorCode="UPDATE_FAILED", updatedAt=engine.now())
                engine.write_json(self.path(request_id), record)
                raise
            return 202, public_state(record)

    def run(self, record, fingerprint):
        request_id = record["requestId"]
        updater = None
        try:
            record.update(status="running", updatedAt=engine.now())
            engine.write_json(self.path(request_id), record)
            updater = self.factory(self.directory)
            result = updater.execute(record["command"], record["input"]["version"],
                                     expected_fingerprint=fingerprint, request_id=request_id)
            record.update(status=result["status"], plan=result["plan"])
            if record["command"] == "check" and not result["plan"]["updateAvailable"]:
                record["status"] = "up_to_date"
        except BaseException as error:
            code = "PLAN_CHANGED" if isinstance(error, engine.PlanChangedError) else "UPDATE_FAILED"
            record.update(status="failed", errorCode=code)
            with (self.home / (request_id + ".log")).open("a") as log:
                os.chmod(log.name, 0o600)
                traceback.print_exc(file=log)
        finally:
            if updater and updater.state and updater.state.get("requestId") == request_id:
                record["job"] = str(updater.job)
            record["updatedAt"] = engine.now()
            engine.write_json(self.path(request_id), record)

    def close(self):
        with self.gate:
            self.closing = True
            worker = self.worker
        if worker:
            worker.join()  # Normal service stop drains the active update instead of interrupting a migration.
        self.daemon_lock.close()


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.0"
    server_version = "FEDEO-Updater/1"
    sys_version = ""

    def setup(self):
        super().setup()
        self.connection.settimeout(5)

    def log_message(self, *_):
        pass

    def send_json(self, status, body):
        data = json.dumps(body, ensure_ascii=False).encode()
        self.close_connection = True
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(data)

    def dispatch(self):
        try:
            if self.headers.get("Origin") or self.headers.get("Transfer-Encoding"):
                raise ApiError("INVALID_REQUEST")
            controller = self.server.controller
            if self.command == "GET" and self.path == "/v1/status":
                self.send_json(200, controller.status())
            elif self.command == "GET" and self.path.startswith("/v1/requests/"):
                suffix = self.path.removeprefix("/v1/requests/")
                self.send_json(200, controller.request_logs(suffix.removesuffix("/logs"))
                               if suffix.endswith("/logs") else controller.request_status(suffix))
            elif self.command == "POST" and self.path in ("/v1/check", "/v1/apply"):
                lengths = self.headers.get_all("Content-Length", [])
                if (len(lengths) != 1 or not lengths[0].isdigit() or not 1 <= int(lengths[0]) <= 4096
                        or self.headers.get_content_type() != "application/json"):
                    raise ApiError("INVALID_REQUEST")
                body = json.loads(self.rfile.read(int(lengths[0])))
                status, result = controller.submit(self.path.rsplit("/", 1)[-1], body)
                self.send_json(status, result)
            else:
                raise ApiError("NOT_FOUND", 404)
        except ApiError as error:
            self.send_json(error.status, {"error": {"code": error.code, "message": ERRORS[error.code]}})
        except (ValueError, UnicodeError):
            self.send_json(400, {"error": {"code": "INVALID_REQUEST", "message": ERRORS["INVALID_REQUEST"]}})
        except (OSError, engine.UpdateError, KeyError, TypeError):
            with contextlib.suppress(OSError):
                self.send_json(500, {"error": {"code": "UPDATE_FAILED", "message": ERRORS["UPDATE_FAILED"]}})

    do_GET = dispatch
    do_POST = dispatch


class Server(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
    daemon_threads = True
    request_queue_size = 8

    def __init__(self, path, controller):
        self.controller = controller
        self.slots = threading.BoundedSemaphore(16)
        super().__init__(str(path), Handler)
        os.chmod(path, 0o660)

    def process_request(self, request, client_address):
        if not self.slots.acquire(blocking=False):
            self.shutdown_request(request)
            return
        try:
            super().process_request(request, client_address)
        except BaseException:
            self.slots.release()
            raise

    def process_request_thread(self, request, client_address):
        try:
            super().process_request_thread(request, client_address)
        finally:
            self.slots.release()

    def handle_error(self, *_):
        pass  # Do not send request data or tracebacks to the journal.


def serve(directory, socket_path):
    controller = None
    path = Path(socket_path)
    stop = threading.Event()
    server = None
    socket_lock = None
    try:
        path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
        parent = path.parent.stat()
        if path.parent.is_symlink() or parent.st_uid != 0 or parent.st_mode & 0o022:
            raise engine.UpdateError("Socket-Verzeichnis muss root gehören und darf nicht durch andere beschreibbar sein.")
        socket_lock = path.with_name(path.name + ".lock").open("a")
        try:
            fcntl.flock(socket_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            raise engine.UpdateError("An diesem Socket läuft bereits ein Update-Dienst.") from None
        controller = Controller(directory)
        if path.is_symlink() or (path.exists() and not stat.S_ISSOCK(path.stat().st_mode)):
            raise engine.UpdateError("Socket-Pfad ist bereits anderweitig belegt.")
        # Both the instance and endpoint are locked before replacing a stale socket.
        path.unlink(missing_ok=True)
        server = Server(path, controller)
        for sig in (signal.SIGTERM, signal.SIGINT):
            signal.signal(sig, lambda *_: stop.set())
        server.timeout = 0.5
        while not stop.is_set():
            server.handle_request()
    finally:
        if controller:
            controller.close()
        if server:
            server.server_close()
            path.unlink(missing_ok=True)
        if socket_lock:
            socket_lock.close()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", choices=("serve",))
    parser.add_argument("--directory", default="/opt/fedeo")
    parser.add_argument("--socket", default="/run/fedeo-updater/control.sock")
    args = parser.parse_args()
    if os.geteuid() != 0:
        parser.error("Der Host-Dienst muss als root laufen.")
    os.umask(0o077)
    try:
        serve(args.directory, args.socket)
    except (OSError, engine.UpdateError) as error:
        parser.exit(1, f"Update-Dienst konnte nicht gestartet werden: {error}\n")


if __name__ == "__main__":
    main()