furtka-gateway/control_plane/routes.py

"""Publish/unpublish routes: a DB row plus a Traefik dynamic-config file.

The gateway never sees a Furtka app's manifest — a publish call only ever
carries {app_name, subdomain, port} from the box. So this module is the
place that independently enforces what the manifest-side `internet.viable`
check can't: subdomain shape/reserved labels, port bounds, and per-box
route caps, regardless of what the calling box claims.
"""

from __future__ import annotations

import re
import sqlite3
import uuid
from datetime import UTC, datetime

from control_plane import traefikconf
from control_plane.db import Database

_SUBDOMAIN_RE = re.compile(
    r"^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?(\.[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?)*$"
)
_APP_NAME_RE = re.compile(r"^[a-z][a-z0-9_-]*$")
_RESERVED_LABELS = {"www", "api", "admin", "traefik", "healthz", "gateway", "wg-easy"}
_MAX_SUBDOMAIN_LEN = 253


class InvalidRoute(Exception):
    pass


class RouteLimitExceeded(Exception):
    pass


class SubdomainTaken(Exception):
    pass


def _validate(app_name: str, subdomain: str, target_port: int) -> None:
    if not isinstance(app_name, str) or not _APP_NAME_RE.match(app_name):
        raise InvalidRoute(f"invalid app_name {app_name!r}")
    if (
        not isinstance(subdomain, str)
        or len(subdomain) > _MAX_SUBDOMAIN_LEN
        or not _SUBDOMAIN_RE.match(subdomain)
    ):
        raise InvalidRoute(f"invalid subdomain {subdomain!r}")
    first_label = subdomain.split(".", 1)[0]
    if first_label in _RESERVED_LABELS:
        raise InvalidRoute(f"subdomain label {first_label!r} is reserved")
    if not isinstance(target_port, int) or isinstance(target_port, bool):
        raise InvalidRoute(f"port must be an integer, got {target_port!r}")
    if not (1 <= target_port <= 65535):
        raise InvalidRoute(f"port {target_port} out of range 1-65535")


def _route_count(db: Database, box_id: str) -> int:
    row = db.query_one(
        "SELECT COUNT(*) AS n FROM routes WHERE box_id = ? AND enabled = 1", (box_id,)
    )
    return row["n"]


def publish_route(
    db: Database,
    box_id: str,
    account_id: str,
    app_name: str,
    subdomain: str,
    target_port: int,
    route_limit_per_box: int,
    cert_resolver: str | None,
) -> dict:
    _validate(app_name, subdomain, target_port)

    if _route_count(db, box_id) >= route_limit_per_box:
        raise RouteLimitExceeded(box_id)

    if db.query_one("SELECT id FROM routes WHERE subdomain = ?", (subdomain,)) is not None:
        raise SubdomainTaken(subdomain)

    box = db.query_one("SELECT wg_allowed_ip FROM boxes WHERE id = ?", (box_id,))
    if box is None:
        raise InvalidRoute(f"unknown box {box_id!r}")

    route_id = uuid.uuid4().hex
    now = datetime.now(UTC).isoformat()
    db.execute(
        """
        INSERT INTO routes
            (id, box_id, account_id, app_name, subdomain, target_port, enabled, created_at)
        VALUES (?, ?, ?, ?, ?, ?, 1, ?)
        """,
        (route_id, box_id, account_id, app_name, subdomain, target_port, now),
    )
    target_ip = box["wg_allowed_ip"].split("/")[0]
    traefikconf.write_route(route_id, subdomain, target_ip, target_port, cert_resolver)
    return {"route_id": route_id, "public_url": f"https://{subdomain}/"}


def unpublish_route(db: Database, box_id: str, route_id: str) -> bool:
    row = db.query_one("SELECT id FROM routes WHERE id = ? AND box_id = ?", (route_id, box_id))
    if row is None:
        return False
    db.execute("DELETE FROM routes WHERE id = ?", (route_id,))
    traefikconf.remove_route(route_id)
    return True


def unpublish_all_for_box(db: Database, box_id: str) -> None:
    for row in db.query_all("SELECT id FROM routes WHERE box_id = ?", (box_id,)):
        traefikconf.remove_route(row["id"])
    db.execute("DELETE FROM routes WHERE box_id = ?", (box_id,))


def list_routes_for_box(db: Database, box_id: str) -> list[sqlite3.Row]:
    return db.query_all(
        "SELECT id, app_name, subdomain, target_port, enabled, created_at "
        "FROM routes WHERE box_id = ?",
        (box_id,),
    )