"""dms-api: HTTPS admin API for docker-mailserver (DMS), runs as a sidecar container.

Drop this on the same host as `mailserver`. It binds the docker unix socket
and exposes a small FastAPI that proxies each call to `docker exec mailserver
setup <subcommand>`.

Auth: single shared bearer token. All calls are TLS-protected when the API is
fronted by a reverse proxy with a real cert (Caddy / nginx + LE / Cloudflare).

Endpoints
---------
GET  /health                          liveness + container status
GET  /v1/accounts                     list email accounts
POST /v1/accounts                     add (body: email, password, quota?)
POST /v1/accounts/{email}/password   update password
DELETE /v1/accounts/{email}          remove account
GET  /v1/aliases                      list aliases
POST /v1/aliases                      add (body: source, target)
DELETE /v1/aliases                    remove (body: source, target)
GET  /v1/quotas                       list per-account quota
POST /v1/quotas                       set (body: email, quota)
DELETE /v1/quotas/{email}            delete quota
GET  /v1/queue                        postqueue -p
POST /v1/queue/flush                  postqueue -f
GET  /v1/logs/tail?n=200              tail /var/log/mail.log
POST /v1/raw                          raw command (whitelisted: setup *, postqueue *)
GET  /v1/dashboard                    container + queue + account count + disk

Optional
--------
* Caddy / nginx in front for TLS
* docker network: mailserver + dms-api in same compose file

Security notes
--------------
* No SSH key on the host; we only need docker.sock (group `docker`).
* The API is bound to 127.0.0.1 by default; expose via reverse proxy.
* `DMS_API_TOKEN` env var carries the bearer secret — rotate freely.
* Every command runs as the configured container user, in the container's
  filesystem, with no host shell access.
"""
from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import shlex
import subprocess
import time
from pathlib import Path
from typing import Any, Dict, List, Optional

from fastapi import Body, Depends, FastAPI, HTTPException, Query, Request
from fastapi.responses import JSONResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel, Field

# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
CONTAINER = os.environ.get("DMS_CONTAINER", "mailserver")
DMS_DATA_DIR = Path(os.environ.get("DMS_DATA_DIR", "/var/mail"))  # for disk usage
API_TOKEN = os.environ.get("DMS_API_TOKEN", "")  # required; checked at startup
API_BIND = os.environ.get("DMS_API_BIND", "127.0.0.1")
API_PORT = int(os.environ.get("DMS_API_PORT", "8791"))
LOG_LEVEL = os.environ.get("DMS_API_LOG_LEVEL", "INFO")

logging.basicConfig(level=LOG_LEVEL, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
log = logging.getLogger("dms-api")

if not API_TOKEN:
    raise SystemExit("DMS_API_TOKEN env var is required (any random string >= 16 chars)")

# ---------------------------------------------------------------------------
# Security
# ---------------------------------------------------------------------------
_bearer = HTTPBearer(auto_error=False)


def _check_token(creds: Optional[HTTPAuthorizationCredentials] = Depends(_bearer)) -> None:
    if creds is None or creds.scheme.lower() != "bearer":
        raise HTTPException(status_code=401, detail="missing bearer token")
    if not _ct_eq(creds.credentials, API_TOKEN):
        raise HTTPException(status_code=401, detail="invalid token")


def _ct_eq(a: str, b: str) -> bool:
    """Constant-time compare to avoid timing attacks."""
    if len(a) != len(b):
        return False
    r = 0
    for x, y in zip(a, b):
        r |= ord(x) ^ ord(y)
    return r == 0


# ---------------------------------------------------------------------------
# docker exec helper
# ---------------------------------------------------------------------------
async def _exec(cmd: List[str], timeout: int = 30) -> Dict[str, Any]:
    """Run `docker exec mailserver <cmd>` and capture stdout/stderr/exit_code."""
    argv = ["docker", "exec", CONTAINER] + cmd
    try:
        proc = await asyncio.create_subprocess_exec(
            *argv,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        try:
            stdout_b, stderr_b = await asyncio.wait_for(proc.communicate(), timeout=timeout)
        except asyncio.TimeoutError:
            proc.kill()
            return {"ok": False, "exit_code": 124, "stdout": "", "stderr": "[timeout]", "error": "timeout"}
        return {
            "ok": proc.returncode == 0,
            "exit_code": proc.returncode,
            "stdout": stdout_b.decode("utf-8", errors="replace"),
            "stderr": stderr_b.decode("utf-8", errors="replace"),
        }
    except Exception as exc:
        log.exception("exec failed: %s", argv)
        return {"ok": False, "exit_code": -1, "stdout": "", "stderr": str(exc), "error": str(exc)}


def _strip_ansi(s: str) -> str:
    return re.sub(r"\x1b\[[0-9;]*[A-Za-z]", "", s)


def _email_list(stdout: str) -> List[Dict[str, str]]:
    """Parse `setup email list` output.

    Output example:
        * admin@smail.icu ( 2.0K / ~ ) [0%]
            [ aliases -> postmaster@smail.icu ]
        * user@smail.icu ( 0 / ~ ) [0%]
    """
    out = []
    for line in stdout.splitlines():
        m = re.match(r"^\*\s+(\S+)\s+\(\s*(\S*)\s*/\s*(\S*)\s*\)\s*\[(\d+)%\]", line.strip())
        if m:
            out.append({
                "email": m.group(1),
                "used": m.group(2),
                "total": m.group(3),
                "quota_pct": int(m.group(4)),
            })
    return out


def _alias_list(stdout: str) -> List[Dict[str, str]]:
    out = []
    for line in stdout.splitlines():
        m = re.match(r"^\*\s+(\S+)\s+(\S+)", line.strip())
        if m:
            out.append({"alias": m.group(1), "target": m.group(2)})
    return out


# ---------------------------------------------------------------------------
# Request models
# ---------------------------------------------------------------------------
class CreateAccount(BaseModel):
    email: str
    password: str
    quota: Optional[str] = None  # e.g. "500M" — optional


class UpdatePassword(BaseModel):
    password: str


class AddAlias(BaseModel):
    source: str
    target: str


class SetQuota(BaseModel):
    email: str
    quota: str  # bytes or unit string


class RawCmd(BaseModel):
    subcommand: str = Field(..., description="setup, postqueue, dovecot-master, etc.")


# ---------------------------------------------------------------------------
# FastAPI app
# ---------------------------------------------------------------------------
app = FastAPI(title="dms-api", version="1.0.0", description="docker-mailserver admin API")


@app.middleware("http")
async def _rate_limit_simple(request: Request, call_next):
    """Tiny in-process token bucket: 30 req/sec per IP. Good enough for a small admin API."""
    ip = request.client.host if request.client else "?"
    bucket = _rate_limit_simple.buckets.setdefault(ip, [30.0, time.time()])
    bucket[0] = min(30.0, bucket[0] + (time.time() - bucket[1]) * 30.0)
    bucket[1] = time.time()
    if bucket[0] < 1.0:
        return JSONResponse({"detail": "rate-limited"}, status_code=429)
    bucket[0] -= 1.0
    return await call_next(request)
_rate_limit_simple.buckets = {}


@app.get("/health", include_in_schema=False)
async def health_no_auth():
    """Liveness probe (no auth) for the reverse proxy / docker."""
    return {"status": "ok", "container": CONTAINER, "version": app.version}


@app.get("/health/auth", include_in_schema=False)
async def health_auth(_: None = Depends(_check_token)):
    """Same as /health, but verifies the token is valid."""
    return {"status": "ok", "container": CONTAINER, "version": app.version}


# All routes below require the bearer token
AUTH = [_check_token]


@app.get("/v1/dashboard")
async def dashboard(_: None = Depends(_check_token)) -> Dict[str, Any]:
    """Health snapshot: container status, postqueue depth, account count, disk usage, log tail."""
    container = await _exec(["docker", "inspect", "-f", "{{.State.Running}}", CONTAINER], timeout=10)
    name_present = await _exec(["docker", "ps", "--format", "{{.Names}}"], timeout=10)
    queue = await _exec(["postqueue", "-p"], timeout=10)
    accounts = await _exec(["setup", "email", "list"], timeout=15)
    disk = await _exec(["df", "-h", "/var/mail"], timeout=10)
    log_tail = await _exec(["tail", "-n", "30", "/var/log/mail.log"], timeout=10)
    return {
        "ok": True,
        "data": {
            "container": {
                "running": "true" in container["stdout"].strip().lower(),
                "name_present": CONTAINER in name_present["stdout"],
                "raw": container["stdout"].strip(),
            },
            "queue_depth": _queue_depth(queue["stdout"]),
            "account_count": len(_email_list(accounts["stdout"])),
            "disk": disk["stdout"].strip().splitlines()[-1] if disk["ok"] else "",
            "log_tail": _strip_ansi(log_tail["stdout"])[-4000:],
        },
    }


def _queue_depth(stdout: str) -> int:
    if "Mail queue is empty" in stdout:
        return 0
    return sum(1 for ln in stdout.splitlines()
               if re.match(r"^[0-9A-F]{6,}\s+", ln))


@app.get("/v1/accounts")
async def accounts(_: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["setup", "email", "list"], timeout=15)
    return {
        "ok": r["ok"],
        "data": _email_list(r["stdout"]),
        "raw": _strip_ansi(r["stdout"]),
    }


@app.post("/v1/accounts")
async def accounts_create(body: CreateAccount, _: None = Depends(_check_token)) -> Dict[str, Any]:
    # Newer DMS rejects add without a password; pass it on the same line.
    add = await _exec(["setup", "email", "add", body.email, body.password], timeout=20)
    upd = await _exec(["setup", "email", "update", body.email, body.password], timeout=20)
    quota = None
    if body.quota:
        quota = await _exec(["setup", "quota", "set", body.email, body.quota], timeout=15)
    return {
        "ok": add["ok"] and upd["ok"],
        "data": {
            "add": add,
            "update": upd,
            "quota": quota,
        },
    }


@app.post("/v1/accounts/{email}/password")
async def accounts_update_password(email: str, body: UpdatePassword,
                                    _: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["setup", "email", "update", email, body.password], timeout=20)
    return {"ok": r["ok"], "data": r}


@app.delete("/v1/accounts/{email}")
async def accounts_delete(email: str, _: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["setup", "email", "del", email], timeout=15)
    return {"ok": r["ok"], "data": r}


@app.get("/v1/aliases")
async def aliases(_: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["setup", "alias", "list"], timeout=10)
    return {
        "ok": r["ok"],
        "data": _alias_list(r["stdout"]),
        "raw": _strip_ansi(r["stdout"]),
    }


@app.post("/v1/aliases")
async def aliases_add(body: AddAlias, _: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["setup", "alias", "add", body.source, body.target], timeout=10)
    return {"ok": r["ok"], "data": r}


@app.delete("/v1/aliases")
async def aliases_del(body: AddAlias, _: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["setup", "alias", "del", body.source, body.target], timeout=10)
    return {"ok": r["ok"], "data": r}


@app.get("/v1/quotas")
async def quotas(_: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["doveadm", "quota", "get", "-A"], timeout=15)
    return {"ok": r["ok"], "raw": _strip_ansi(r["stdout"]), "data": r}


@app.post("/v1/quotas")
async def quotas_set(body: SetQuota, _: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["setup", "quota", "set", body.email, body.quota], timeout=15)
    return {"ok": r["ok"], "data": r}


@app.delete("/v1/quotas/{email}")
async def quotas_del(email: str, _: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["setup", "quota", "del", email], timeout=10)
    return {"ok": r["ok"], "data": r}


@app.get("/v1/queue")
async def queue(_: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["postqueue", "-p"], timeout=10)
    return {"ok": r["ok"], "raw": _strip_ansi(r["stdout"]), "data": r}


@app.post("/v1/queue/flush")
async def queue_flush(_: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["postqueue", "-f"], timeout=10)
    return {"ok": r["ok"], "raw": r["stdout"], "data": r}


@app.get("/v1/logs/tail")
async def logs_tail(n: int = Query(default=200, ge=1, le=5000),
                   _: None = Depends(_check_token)) -> Dict[str, Any]:
    r = await _exec(["tail", "-n", str(n), "/var/log/mail.log"], timeout=15)
    return {"ok": r["ok"], "raw": _strip_ansi(r["stdout"]), "data": r}


_WHITELIST = re.compile(r"^(setup|postqueue|doveadm|dms-healthcheck)(\s.*)?$", re.DOTALL)


@app.post("/v1/raw")
async def raw(body: RawCmd, _: None = Depends(_check_token)) -> Dict[str, Any]:
    """Whitelisted escape hatch. Only commands starting with `setup `, `postqueue`, `doveadm`, or
    `dms-healthcheck` are accepted. Anything else returns 400."""
    if not _WHITELIST.match(body.subcommand.strip()):
        raise HTTPException(
            status_code=400,
            detail="subcommand must start with setup / postqueue / doveadm / dms-healthcheck",
        )
    argv = body.subcommand.strip().split()
    r = await _exec(argv, timeout=30)
    return {"ok": r["ok"], "raw": _strip_ansi(r["stdout"]), "data": r}


# ---------------------------------------------------------------------------
# Entrypoint
# ---------------------------------------------------------------------------
def main() -> None:
    import uvicorn
    uvicorn.run(
        "app:app",
        host=API_BIND,
        port=API_PORT,
        log_level=LOG_LEVEL.lower(),
        access_log=True,
    )


if __name__ == "__main__":
    main()