"""docker-mailserver web management UI (FastAPI backend). Endpoints --------- GET / SPA shell (static/index.html) GET /api/hosts list configured hosts GET /api/dashboard health snapshot for the selected host GET /api/accounts list of email accounts POST /api/accounts add an account POST /api/accounts/update change password POST /api/accounts/delete remove account POST /api/accounts/restrict toggle login for an account GET /api/aliases list aliases POST /api/aliases add alias POST /api/aliases/delete remove alias GET /api/quotas list per-account quota POST /api/quotas set quota POST /api/quotas/delete delete quota GET /api/queue postqueue -p POST /api/queue/flush postqueue -f GET /api/logs/tail SSE stream of `tail -F /var/log/mail.log` GET /api/raw?cmd=... escape hatch: run any whitelisted command POST /api/login password -> session cookie POST /api/logout clear session Auth ---- A single admin password is read from the env var ``DMS_UI_ADMIN_PASSWORD`` (default ``admin``). A successful login sets a signed-ish cookie; subsequent JSON calls must include it. """ from __future__ import annotations import asyncio import json import logging import os import re import secrets import shlex import time from pathlib import Path from typing import Any, Dict, List, Optional from fastapi import Cookie, Depends, FastAPI, HTTPException, Query, Request from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse from fastapi.staticfiles import StaticFiles import ssh_client from ssh_client import DMSSSH, CommandResult, run_command logging.basicConfig(level=os.environ.get("DMS_UI_LOG_LEVEL", "INFO")) # --------------------------------------------------------------------------- # Container command helper # --------------------------------------------------------------------------- DEFAULT_CONTAINER = os.environ.get("DMS_UI_CONTAINER", "mailserver") def _container_cmd(cmd: str, container: str = "") -> str: """Wrap a host shell command so it runs inside the DMS docker container. Everything sent over SSH by default runs on the host. DMS-specific tools (setup, postqueue, doveadm, etc.) live *inside* the mailserver container. We wrap such commands with `docker exec sh -c "..."` so they resolve correctly. Anything else (df, tail, docker ps, docker inspect) is left untouched and runs on the host. """ c = container or DEFAULT_CONTAINER bare = cmd.lstrip() if not bare.startswith(("setup ", "setup ", "postqueue", "doveadm", "dovecot-master", "dms-healthcheck")): return cmd # Use double-quotes for sh -c and escape embedded ", \, $, `. # Single-quote wrapping used here for sh -c so internal ' are safe; # we still escape single quotes inside the cmd (the form '''). quoted = cmd.replace("'", "'\''") return f"docker exec {c} sh -c '{quoted}'" log = logging.getLogger("dms-ui.app") ROOT = Path(__file__).resolve().parent CONFIG_PATH = Path(os.environ.get("DMS_UI_CONFIG", str(ROOT / "config.json"))) CONFIG_EXAMPLE = ROOT / "config.example.json" ADMIN_PASSWORD = os.environ.get("DMS_UI_ADMIN_PASSWORD", "admin") SESSION_COOKIE = "dms_ui_session" SESSION_MAX_AGE = int(os.environ.get("DMS_UI_SESSION_MAX_AGE", str(8 * 3600))) ACTIVE_HOST = os.environ.get("DMS_UI_DEFAULT_HOST", "") # optional default # Map of session_token -> issued_at. We deliberately keep this in-process. # Restarting the server forces re-login — acceptable for a single-tenant UI. _SESSIONS: Dict[str, float] = {} app = FastAPI(title="dms-ui", version="0.1.0") app.mount("/static", StaticFiles(directory=str(ROOT / "static")), name="static") # --------------------------------------------------------------------------- # Config loading # --------------------------------------------------------------------------- def _load_hosts() -> List[Dict[str, Any]]: path = CONFIG_PATH if CONFIG_PATH.exists() else CONFIG_EXAMPLE if not path.exists(): log.warning("no config at %s and no example; using empty host list", CONFIG_PATH) return [] try: data = json.loads(path.read_text(encoding="utf-8")) except Exception as exc: log.error("failed to parse %s: %s", path, exc) return [] hosts = data.get("hosts", []) if isinstance(data, dict) else [] # Normalize dict-of-hosts to list-of-hosts. Each entry may also carry its # short id as a separate key (e.g. {"smail": {...}}) — preserve that. if isinstance(hosts, dict): host_list: List[Dict[str, Any]] = [] for h_id, h in hosts.items(): if not isinstance(h, dict): continue h = dict(h) h.setdefault("id", h_id) host_list.append(h) hosts = host_list cleaned: List[Dict[str, Any]] = [] for h in hosts: if not isinstance(h, dict) or not h.get("ssh_alias"): continue cleaned.append( { "id": h.get("id") or h["ssh_alias"], "name": h.get("name") or h["ssh_alias"], "ssh_alias": h["ssh_alias"], "container": h.get("container", "mailserver"), "description": h.get("description", ""), # Never echo the password to the browser. "has_password": bool(h.get("password")), } ) return cleaned def _find_host(name: Optional[str]) -> Dict[str, Any]: hosts = _load_hosts() if not hosts: raise HTTPException(status_code=503, detail="no hosts configured") target = name or ACTIVE_HOST or hosts[0].get("id") or hosts[0]["name"] for h in hosts: if h.get("id") == target or h.get("name") == target or h.get("ssh_alias") == target: return h raise HTTPException(status_code=404, detail=f"host {target!r} not found") def _password_for(host: Dict[str, Any]) -> Optional[str]: if CONFIG_PATH.exists(): try: data = json.loads(CONFIG_PATH.read_text(encoding="utf-8")) for h in data.get("hosts", []): if h.get("name") == host["name"] or h.get("ssh_alias") == host["ssh_alias"]: return h.get("password") except Exception: # pragma: no cover pass return None # --------------------------------------------------------------------------- # Auth # --------------------------------------------------------------------------- def _check_session(session: Optional[str] = Cookie(default=None, alias=SESSION_COOKIE)) -> str: if not session or session not in _SESSIONS: raise HTTPException(status_code=401, detail="not authenticated") issued = _SESSIONS[session] if time.time() - issued > SESSION_MAX_AGE: _SESSIONS.pop(session, None) raise HTTPException(status_code=401, detail="session expired") # sliding expiry _SESSIONS[session] = time.time() return session def _response(result: CommandResult, data: Any = None) -> Dict[str, Any]: return { "ok": result.ok, "stdout": result.stdout, "stderr": result.stderr, "exit_code": result.exit_code, "data": data, "error": "" if result.ok else (result.stderr.strip() or f"rc={result.exit_code}"), } # --------------------------------------------------------------------------- # Static + login routes # --------------------------------------------------------------------------- @app.get("/", response_class=HTMLResponse) def index() -> HTMLResponse: return HTMLResponse((ROOT / "static" / "index.html").read_text(encoding="utf-8")) @app.post("/api/login") def login(request: Request) -> JSONResponse: body: Dict[str, Any] = {} try: body = asyncio.run(_read_json(request)) # tiny body, fine except Exception: body = {} password = str(body.get("password", "")) if not secrets.compare_digest(password, ADMIN_PASSWORD): raise HTTPException(status_code=401, detail="invalid password") token = secrets.token_urlsafe(32) _SESSIONS[token] = time.time() resp = JSONResponse({"ok": True}) resp.set_cookie( SESSION_COOKIE, token, max_age=SESSION_MAX_AGE, httponly=True, samesite="lax", ) return resp @app.post("/api/logout") def logout(session: Optional[str] = Cookie(default=None, alias=SESSION_COOKIE)) -> JSONResponse: if session: _SESSIONS.pop(session, None) resp = JSONResponse({"ok": True}) resp.delete_cookie(SESSION_COOKIE) return resp async def _read_json(request: Request) -> Dict[str, Any]: raw = await request.body() if not raw: return {} return json.loads(raw.decode("utf-8")) # --------------------------------------------------------------------------- # Host + dashboard # --------------------------------------------------------------------------- @app.get("/api/hosts") def list_hosts(_: str = Depends(_check_session)) -> Dict[str, Any]: return {"ok": True, "data": _load_hosts()} @app.get("/api/dashboard") def dashboard(host: Optional[str] = Query(default=None), _: str = Depends(_check_session)) -> Dict[str, Any]: h = _find_host(host) ssh = DMSSSH(h["ssh_alias"], password=_password_for(h), timeout=20) container = ssh.run("docker inspect -f '{{.State.Running}}' mailserver 2>/dev/null || echo no") queue = ssh.run(_container_cmd("postqueue -p | tail -n +2 | grep -c '^[0-9A-F]' || true")) accounts = ssh.run(_container_cmd("setup email list 2>&1 | tail -n +3 | grep -E '^\\| ' | wc -l")) disk = ssh.run("df -h /var/mail | tail -n 1") log_tail = ssh.run("tail -n 30 /var/log/mail.log 2>/dev/null") container_name = ssh.run("docker ps --format '{{.Names}}' | grep -E '^mailserver$' || true") return _response( CommandResult(stdout="", stderr="", exit_code=0), data={ "host": h["name"], "container": { "running": "true" in container.stdout.strip().lower(), "name_present": "mailserver" in container_name.stdout, "raw": container.stdout.strip(), }, "queue_depth": int(queue.stdout.strip() or 0), "account_count": int(accounts.stdout.strip() or 0), "disk": disk.stdout.strip(), "log_tail": log_tail.stdout, }, ) # --------------------------------------------------------------------------- # Accounts # --------------------------------------------------------------------------- _EMAIL_LIST_RE = re.compile( r"^\|\s*(\S+@\S+)\s*\|\s*(\S*)\s*\|", re.MULTILINE ) @app.get("/api/accounts") def accounts(host: Optional[str] = Query(default=None), _: str = Depends(_check_session)) -> Dict[str, Any]: h = _find_host(host) res = run_command(h["ssh_alias"], _container_cmd("setup email list 2>&1"), password=_password_for(h)) items = [] for line in res.stdout.splitlines(): m = _EMAIL_LIST_RE.match(line.strip()) if m: items.append({"email": m.group(1), "extra": m.group(2)}) return _response(res, data=items) @app.post("/api/accounts") def add_account(request: Request, _: str = Depends(_check_session)) -> Dict[str, Any]: body = asyncio.run(_read_json(request)) host = body.get("host") email = (body.get("email") or "").strip() password = body.get("password") or "" quota = (body.get("quota") or "").strip() if not email or "@" not in email or not password: raise HTTPException(status_code=400, detail="email + password required") h = _find_host(host) pwd = _password_for(h) # Newer docker-mailserver rejects add without a password; pass it on the # same line to keep the call atomic. add = run_command( h["ssh_alias"], f"docker exec -i mailserver setup email add {shlex.quote(email)} {shlex.quote(password)}", password=pwd, ) # Run an update as a safety net in case the container's add path stored an # empty password anyway (older DMS versions). upd = run_command( h["ssh_alias"], f"docker exec -i mailserver setup email update {shlex.quote(email)} {shlex.quote(password)}", password=pwd, ) quota_res = None if quota: quota_res = run_command( h["ssh_alias"], f"docker exec -i mailserver setup quota set {shlex.quote(email)} {shlex.quote(quota)}", password=pwd, ) return _response( add if not add.ok else (quota_res or upd), data={"add": add.to_dict(), "update": upd.to_dict(), "quota": quota_res.to_dict() if quota_res else None}, ) @app.post("/api/accounts/update") def update_account(request: Request, _: str = Depends(_check_session)) -> Dict[str, Any]: body = asyncio.run(_read_json(request)) h = _find_host(body.get("host")) email = (body.get("email") or "").strip() password = body.get("password") or "" if not email or not password: raise HTTPException(status_code=400, detail="email + password required") res = run_command( h["ssh_alias"], f"docker exec -i mailserver setup email update {shlex.quote(email)} {shlex.quote(password)}", password=_password_for(h), ) return _response(res) @app.post("/api/accounts/delete") def delete_account(request: Request, _: str = Depends(_check_session)) -> Dict[str, Any]: body = asyncio.run(_read_json(request)) h = _find_host(body.get("host")) email = (body.get("email") or "").strip() if not email: raise HTTPException(status_code=400, detail="email required") res = run_command( h["ssh_alias"], f"docker exec -i mailserver setup email del -y {shlex.quote(email)}", password=_password_for(h), ) return _response(res) @app.post("/api/accounts/restrict") def restrict_account(request: Request, _: str = Depends(_check_session)) -> Dict[str, Any]: body = asyncio.run(_read_json(request)) h = _find_host(body.get("host")) email = (body.get("email") or "").strip() restrict = bool(body.get("restrict", True)) if not email: raise HTTPException(status_code=400, detail="email required") args = "restrict" if restrict else "unrestrict" res = run_command( h["ssh_alias"], f"docker exec -i mailserver setup email {args} {shlex.quote(email)}", password=_password_for(h), ) return _response(res) # --------------------------------------------------------------------------- # Aliases # --------------------------------------------------------------------------- _ALIAS_RE = re.compile(r"^\|\s*(\S+@\S+)\s*\|\s*(\S+)\s*\|\s*(\S*)\s*\|", re.MULTILINE) @app.get("/api/aliases") def aliases(host: Optional[str] = Query(default=None), _: str = Depends(_check_session)) -> Dict[str, Any]: h = _find_host(host) res = run_command(h["ssh_alias"], _container_cmd("setup alias list 2>&1"), password=_password_for(h)) items = [] for line in res.stdout.splitlines(): m = _ALIAS_RE.match(line.strip()) if m: items.append({"source": m.group(1), "target": m.group(2), "extra": m.group(3)}) return _response(res, data=items) @app.post("/api/aliases") def add_alias(request: Request, _: str = Depends(_check_session)) -> Dict[str, Any]: body = asyncio.run(_read_json(request)) h = _find_host(body.get("host")) # Accept either (alias, recipient) or (source, target) — different frontends. source = (body.get("source") or body.get("alias") or "").strip() target = (body.get("target") or body.get("recipient") or "").strip() if not source or not target: raise HTTPException(status_code=400, detail="source + target required") res = run_command( h["ssh_alias"], f"docker exec -i mailserver setup alias add {shlex.quote(source)} {shlex.quote(target)}", password=_password_for(h), ) return _response(res) @app.post("/api/aliases/delete") def delete_alias(request: Request, _: str = Depends(_check_session)) -> Dict[str, Any]: body = asyncio.run(_read_json(request)) h = _find_host(body.get("host")) # Accept either (alias, recipient) or (source, target) — different frontends. source = (body.get("source") or body.get("alias") or "").strip() target = (body.get("target") or body.get("recipient") or "").strip() if not source or not target: raise HTTPException(status_code=400, detail="source + target required") res = run_command( h["ssh_alias"], f"docker exec -i mailserver setup alias del {shlex.quote(source)} {shlex.quote(target)}", password=_password_for(h), ) return _response(res) # --------------------------------------------------------------------------- # Quotas # --------------------------------------------------------------------------- @app.get("/api/quotas") def quotas(host: Optional[str] = Query(default=None), _: str = Depends(_check_session)) -> Dict[str, Any]: h = _find_host(host) # `doveadm quota get -u ` per account is slow; use `setup email list` + `setup quota list` if present. list_res = run_command(h["ssh_alias"], _container_cmd("setup email list 2>&1"), password=_password_for(h)) accounts: List[str] = [] for line in list_res.stdout.splitlines(): m = _EMAIL_LIST_RE.match(line.strip()) if m: accounts.append(m.group(1)) quotas: Dict[str, str] = {} for acc in accounts: q = run_command( h["ssh_alias"], f"docker exec -i mailserver setup quota get {shlex.quote(acc)} 2>/dev/null", password=_password_for(h), ) quotas[acc] = q.stdout.strip() or "(none)" return _response(list_res, data={"accounts": accounts, "quotas": quotas}) @app.post("/api/quotas") def set_quota(request: Request, _: str = Depends(_check_session)) -> Dict[str, Any]: body = asyncio.run(_read_json(request)) h = _find_host(body.get("host")) email = (body.get("email") or "").strip() quota = (body.get("quota") or "").strip() if not email or not quota: raise HTTPException(status_code=400, detail="email + quota required") res = run_command( h["ssh_alias"], f"docker exec -i mailserver setup quota set {shlex.quote(email)} {shlex.quote(quota)}", password=_password_for(h), ) return _response(res) @app.post("/api/quotas/delete") def delete_quota(request: Request, _: str = Depends(_check_session)) -> Dict[str, Any]: body = asyncio.run(_read_json(request)) h = _find_host(body.get("host")) email = (body.get("email") or "").strip() if not email: raise HTTPException(status_code=400, detail="email required") res = run_command( h["ssh_alias"], f"docker exec -i mailserver setup quota del {shlex.quote(email)}", password=_password_for(h), ) return _response(res) # --------------------------------------------------------------------------- # Queue # --------------------------------------------------------------------------- @app.get("/api/queue") def queue(host: Optional[str] = Query(default=None), _: str = Depends(_check_session)) -> Dict[str, Any]: h = _find_host(host) res = run_command(h["ssh_alias"], _container_cmd("postqueue -p"), password=_password_for(h)) return _response(res) @app.post("/api/queue/flush") def queue_flush(host: Optional[str] = Query(default=None), _: str = Depends(_check_session)) -> Dict[str, Any]: h = _find_host(host) res = run_command(h["ssh_alias"], _container_cmd("postqueue -f"), password=_password_for(h)) return _response(res) # --------------------------------------------------------------------------- # Logs (SSE) # --------------------------------------------------------------------------- @app.get("/api/logs/tail") def logs_tail( host: Optional[str] = Query(default=None), lines: int = Query(default=200, ge=10, le=5000), _: str = Depends(_check_session), ) -> StreamingResponse: h = _find_host(host) # First send the last N lines synchronously, then stream new lines. ssh = DMSSSH(h["ssh_alias"], password=_password_for(h), timeout=10) initial = ssh.run(f"tail -n {int(lines)} /var/log/mail.log 2>/dev/null") return StreamingResponse( _log_stream(h, initial.stdout, password=_password_for(h)), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) async def _log_stream(host: Dict[str, Any], initial: str, password: Optional[str]): yield _sse("snapshot", initial or "(log is empty)\n") # Use a queue + a thread to bridge paramiko's blocking iter into async. q: asyncio.Queue = asyncio.Queue() loop = asyncio.get_running_loop() sentinel = object() def pump() -> None: try: stream = ssh_client.StreamingSSH( host["ssh_alias"], "tail -F /var/log/mail.log", password=password, ) for chunk in iter(stream): loop.call_soon_threadsafe(q.put_nowait, chunk) except Exception as exc: # pragma: no cover - defensive loop.call_soon_threadsafe(q.put_nowait, f"[stream error] {exc}\n") finally: loop.call_soon_threadsafe(q.put_nowait, sentinel) import threading t = threading.Thread(target=pump, daemon=True) t.start() while True: item = await q.get() if item is sentinel: break yield _sse("line", item if isinstance(item, str) else str(item)) def _sse(event: str, data: str) -> bytes: safe = data.replace("\r\n", "\n") payload = "".join(f"data: {line}\n" for line in safe.split("\n")) return f"event: {event}\n{payload}\n\n".encode("utf-8") # --------------------------------------------------------------------------- # Raw escape hatch (whitelisted commands only) # --------------------------------------------------------------------------- _WHITELIST = re.compile( r"^(?:" r"setup\s+(?:email|alias|quota|dovecot-master|relay)(?:\s+\S+){0,6}" r"|postqueue(?:\s+-[a-z]+)?" r"|postsuper(?:\s+-[a-z]+)?" r"|doveadm\s+\S.*" r"|docker\s+(?:ps|inspect|logs|exec|stats)\s.*" r"|df\s+-h\s+/var/mail" r"|ls\s+/var/mail(?:\s+-la)?" r"|tail\s+(?:-F|-f|\d+)\s+/var/log/mail\.log" r"|cat\s+/var/log/mail\.log" r")$" ) @app.get("/api/raw") def raw( cmd: str = Query(..., max_length=400), host: Optional[str] = Query(default=None), _: str = Depends(_check_session), ) -> Dict[str, Any]: if not _WHITELIST.match(cmd): raise HTTPException(status_code=400, detail="command not whitelisted") h = _find_host(host) res = run_command(h["ssh_alias"], cmd, password=_password_for(h)) return _response(res) # --------------------------------------------------------------------------- # Entrypoint # --------------------------------------------------------------------------- def main() -> None: import argparse import uvicorn parser = argparse.ArgumentParser(description="docker-mailserver web UI") parser.add_argument("--host", default=os.environ.get("DMS_UI_BIND", "127.0.0.1")) parser.add_argument("--port", type=int, default=int(os.environ.get("DMS_UI_PORT", "8792"))) args = parser.parse_args() log.info("starting dms-ui on %s:%d (config=%s)", args.host, args.port, CONFIG_PATH) uvicorn.run(app, host=args.host, port=args.port, log_level="info") if __name__ == "__main__": main()