furtka-gateway/control_plane/app.py

import argparse
import os
from http.server import ThreadingHTTPServer

from control_plane import accounts
from control_plane.api import Context, Handler
from control_plane.db import Database
from control_plane.wgeasy import WgEasyClient
from control_plane.wgeasy_v15 import WgEasyV15Client

DEFAULT_BOX_LIMIT = 20
DEFAULT_ROUTE_LIMIT_PER_BOX = 20
DEFAULT_WG_ALLOWED_IPS = "10.8.0.0/24"


def build_context(db: Database | None = None) -> Context:
    mode = os.environ.get("GATEWAY_MODE", "single")
    db = db or Database()

    wg_easy_url = os.environ.get("WG_EASY_URL", "http://host.docker.internal:51821")
    wg_easy_password = os.environ.get("WG_EASY_ADMIN_PASSWORD", "")
    if os.environ.get("WG_EASY_API_VERSION", "14") == "15":
        wgeasy = WgEasyV15Client(
            base_url=wg_easy_url,
            username=os.environ.get("WG_EASY_ADMIN_USERNAME", ""),
            password=wg_easy_password,
        )
        # v15 defaults new clients to full-tunnel AllowedIPs and has no
        # equivalent to v14's WG_ALLOWED_IPS container env var for an
        # already-initialized instance (relevant for a BYO wg-easy that
        # existed before this control-plane ever connected to it) — see
        # wgeasy_v15.py's docstring. Enforced on every startup, not just
        # once, since a BYO instance's config isn't otherwise this
        # process's to assume.
        wgeasy.restrict_default_allowed_ips(
            [os.environ.get("GATEWAY_WG_ALLOWED_IPS", DEFAULT_WG_ALLOWED_IPS)]
        )
    else:
        wgeasy = WgEasyClient(base_url=wg_easy_url, admin_password=wg_easy_password)

    single_account_id = None
    if mode == "single":
        single_account_id = accounts.ensure_single_tenant_account(
            db, DEFAULT_BOX_LIMIT, DEFAULT_ROUTE_LIMIT_PER_BOX
        )

    # Independent of GATEWAY_MODE — mirrors whatever certResolver name (if
    # any) the operator's own Traefik already has configured. Unset means
    # no explicit resolver (`tls: {}` — see traefikconf.py), trusting that
    # Traefik's own default TLS applies (e.g. shared mode's wildcard
    # defaultGeneratedCert in traefik.shared.yml).
    cert_resolver = os.environ.get("GATEWAY_CERT_RESOLVER") or None

    return Context(
        db=db,
        wgeasy=wgeasy,
        mode=mode,
        cert_resolver=cert_resolver,
        single_account_id=single_account_id,
        box_limit=DEFAULT_BOX_LIMIT,
        route_limit_per_box=DEFAULT_ROUTE_LIMIT_PER_BOX,
    )


def serve(host: str = "0.0.0.0", port: int = 8090) -> None:
    server = ThreadingHTTPServer((host, port), Handler)
    server.ctx = build_context()
    server.serve_forever()


def main() -> None:
    parser = argparse.ArgumentParser(description="furtka-gateway control plane")
    parser.add_argument("--host", default="0.0.0.0")
    parser.add_argument("--port", type=int, default=8090)
    args = parser.parse_args()
    serve(args.host, args.port)


if __name__ == "__main__":
    main()