"""Small Basic Auth web UI for managing mailcat domain routes.""" from __future__ import annotations import argparse import base64 import html import json import os import secrets import shutil from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse from mail_routes import DOMAIN_ROUTES_FILE, domain_lookup, domain_routes, normalize_domain_route DEFAULT_CONFIG_FILE = Path(__file__).parent / "web_config.json" DEFAULT_HOST = "0.0.0.0" DEFAULT_PORT = 4321 def validate_routes(raw_routes: list[dict]) -> list[dict]: """Return normalized route dictionaries and reject duplicate domains.""" if not isinstance(raw_routes, list): raise ValueError("Routes must be a list") normalized: list[dict] = [] seen: dict[str, int] = {} for index, raw_rule in enumerate(raw_routes): domains, mailbox = normalize_domain_route(raw_rule) for domain in domains: if domain in seen: other = seen[domain] + 1 raise ValueError(f"Domain {domain!r} is already used by route {other}") seen[domain] = index normalized.append({"domains": domains, "mailbox": mailbox}) return normalized def load_routes(path: Path = DOMAIN_ROUTES_FILE) -> list[dict]: raw = json.loads(path.read_text(encoding="utf-8")) return validate_routes(raw) def save_routes(routes: list[dict], path: Path = DOMAIN_ROUTES_FILE) -> list[dict]: normalized = validate_routes(routes) tmp_path = path.with_name(f".{path.name}.tmp") backup_path = path.with_suffix(f"{path.suffix}.bak") tmp_path.write_text(json.dumps(normalized, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") if path.exists(): shutil.copy2(path, backup_path) os.replace(tmp_path, path) domain_routes.cache_clear() domain_lookup.cache_clear() return normalized def load_web_config(path: Path) -> tuple[str, str]: config = {} if path.exists(): config = json.loads(path.read_text(encoding="utf-8")) username = os.environ.get("MAILCAT_WEB_USER") or config.get("username") password = os.environ.get("MAILCAT_WEB_PASSWORD") or config.get("password") if not username or not password: raise SystemExit( f"Missing web credentials. Create {path} or set MAILCAT_WEB_USER and MAILCAT_WEB_PASSWORD." ) return str(username), str(password) def basic_auth_ok(header: str | None, username: str, password: str) -> bool: if not header or not header.startswith("Basic "): return False try: decoded = base64.b64decode(header[6:], validate=True).decode("utf-8") supplied_user, supplied_password = decoded.split(":", 1) except Exception: return False return secrets.compare_digest(supplied_user, username) and secrets.compare_digest( supplied_password, password ) def page_html() -> bytes: return f""" Mailcat routes

Mailcat routes

#	Domeinen	Map

Nieuwe regel

""".encode("utf-8") class RouteHandler(BaseHTTPRequestHandler): routes_file = DOMAIN_ROUTES_FILE username = "" password = "" def log_message(self, fmt: str, *args: object) -> None: print(f"{self.address_string()} - {fmt % args}") def _require_auth(self) -> bool: if basic_auth_ok(self.headers.get("Authorization"), self.username, self.password): return True self.send_response(HTTPStatus.UNAUTHORIZED) self.send_header("WWW-Authenticate", 'Basic realm="mailcat"') self.send_header("Content-Length", "0") self.end_headers() return False def _send(self, status: HTTPStatus, body: bytes, content_type: str) -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _json(self, status: HTTPStatus, payload: dict) -> None: self._send( status, json.dumps(payload, indent=2, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8", ) def _error(self, status: HTTPStatus, message: str) -> None: self._json(status, {"error": message}) def _read_route_body(self) -> dict: length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(length) payload = json.loads(body.decode("utf-8")) if not isinstance(payload, dict): raise ValueError("Request body must contain one route object") return payload def _route_index(self) -> int | None: parts = [part for part in urlparse(self.path).path.split("/") if part] if len(parts) != 3 or parts[:2] != ["api", "routes"]: return None try: return int(parts[2]) except ValueError: return None def do_GET(self) -> None: if not self._require_auth(): return path = urlparse(self.path).path if path == "/": self._send(HTTPStatus.OK, page_html(), "text/html; charset=utf-8") return if path == "/api/routes": try: self._json(HTTPStatus.OK, {"routes": load_routes(self.routes_file)}) except Exception as exc: self._error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) return self._error(HTTPStatus.NOT_FOUND, "Not found") def do_POST(self) -> None: if not self._require_auth(): return if urlparse(self.path).path != "/api/routes": self._error(HTTPStatus.NOT_FOUND, "Not found") return try: routes = load_routes(self.routes_file) routes.append(self._read_route_body()) self._json(HTTPStatus.CREATED, {"routes": save_routes(routes, self.routes_file)}) except json.JSONDecodeError: self._error(HTTPStatus.BAD_REQUEST, "Invalid JSON") except ValueError as exc: self._error(HTTPStatus.BAD_REQUEST, str(exc)) except Exception as exc: self._error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) def do_PUT(self) -> None: if not self._require_auth(): return index = self._route_index() if index is None: self._error(HTTPStatus.NOT_FOUND, "Not found") return try: routes = load_routes(self.routes_file) if index < 0 or index >= len(routes): raise ValueError("Route index out of range") routes[index] = self._read_route_body() self._json(HTTPStatus.OK, {"routes": save_routes(routes, self.routes_file)}) except json.JSONDecodeError: self._error(HTTPStatus.BAD_REQUEST, "Invalid JSON") except ValueError as exc: self._error(HTTPStatus.BAD_REQUEST, str(exc)) except Exception as exc: self._error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) def do_DELETE(self) -> None: if not self._require_auth(): return index = self._route_index() if index is None: self._error(HTTPStatus.NOT_FOUND, "Not found") return try: routes = load_routes(self.routes_file) if index < 0 or index >= len(routes): raise ValueError("Route index out of range") del routes[index] self._json(HTTPStatus.OK, {"routes": save_routes(routes, self.routes_file)}) except ValueError as exc: self._error(HTTPStatus.BAD_REQUEST, str(exc)) except Exception as exc: self._error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) def main() -> None: parser = argparse.ArgumentParser(description="Run the mailcat route management web UI.") parser.add_argument("--host", default=DEFAULT_HOST) parser.add_argument("--port", type=int, default=DEFAULT_PORT) parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG_FILE) parser.add_argument("--routes-file", type=Path, default=DOMAIN_ROUTES_FILE) args = parser.parse_args() username, password = load_web_config(args.config) RouteHandler.routes_file = args.routes_file RouteHandler.username = username RouteHandler.password = password server = ThreadingHTTPServer((args.host, args.port), RouteHandler) print(f"mailcat route web listening on http://{html.escape(args.host)}:{args.port}/") server.serve_forever() if __name__ == "__main__": main()