Homelab Gateway

from pydantic import BaseModel, Field
from enum import Enum
from typing import Optional

class RecordType(str, Enum):
    A = "A"
    AAAA = "AAAA"
    CNAME = "CNAME"
    TXT = "TXT"
    MX = "MX"

class DNSRecord(BaseModel):
    """DNS record model with validation."""
    name: str = Field(..., description="Record name (e.g., 'api')")
    type: RecordType
    content: str = Field(..., description="Record value")
    ttl: int = Field(default=3600, ge=60, le=86400)
    disabled: bool = False

    class Config:
        use_enum_values = True

from fastapi import APIRouter, Depends, HTTPException
from typing import List

from ..models.dns import DNSRecord, Zone
from ..services.dns_service import DNSService

router = APIRouter(prefix="/api/dns", tags=["dns"])

@router.get("/zones", response_model=List[Zone])
async def list_zones(
    dns_service: DNSService = Depends()
) -> List[Zone]:
    """List all DNS zones."""
    return await dns_service.get_zones()

@router.post("/zones/{zone_id}/records", response_model=DNSRecord)
async def create_record(
    zone_id: str,
    record: DNSRecord,
    dns_service: DNSService = Depends()
) -> DNSRecord:
    """Create a new DNS record."""
    try:
        return await dns_service.create_record(zone_id, record)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

import httpx
from typing import Any, Dict, List
from functools import lru_cache

class PowerDNSClient:
    """Async client for PowerDNS API."""

    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip("/")
        self.headers = {"X-API-Key": api_key}

    async def _request(
        self,
        method: str,
        path: str,
        **kwargs
    ) -> Dict[str, Any]:
        async with httpx.AsyncClient() as client:
            response = await client.request(
                method,
                f"{self.base_url}{path}",
                headers=self.headers,
                **kwargs
            )
            response.raise_for_status()
            return response.json()

    async def get_zones(self, server_id: str = "localhost") -> List[Dict]:
        """Fetch all zones from PowerDNS."""
        return await self._request(
            "GET",
            f"/api/v1/servers/{server_id}/zones"
        )

    async def patch_zone(
        self,
        zone_id: str,
        rrsets: List[Dict],
        server_id: str = "localhost"
    ) -> None:
        """Update zone records."""
        await self._request(
            "PATCH",
            f"/api/v1/servers/{server_id}/zones/{zone_id}",
            json={"rrsets": rrsets}
        )

import httpx
from dataclasses import dataclass
from typing import Optional, Dict, Any

@dataclass
class CaddyConfig:
    """Parsed Caddy configuration."""
    apps: Dict[str, Any]
    admin: Dict[str, Any]

class CaddyClient:
    """Client for Caddy's admin API."""

    def __init__(self, admin_url: str = "http://localhost:2019"):
        self.admin_url = admin_url

    async def get_config(self) -> CaddyConfig:
        """Fetch current Caddy configuration."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"{self.admin_url}/config/")
            resp.raise_for_status()
            data = resp.json()
            return CaddyConfig(
                apps=data.get("apps", {}),
                admin=data.get("admin", {})
            )

    async def reload_config(self, config_path: str) -> bool:
        """Trigger Caddy to reload its configuration."""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.admin_url}/load",
                headers={"Content-Type": "text/caddyfile"},
                content=open(config_path).read()
            )
            return resp.status_code == 200

from dataclasses import dataclass
from typing import List, Optional

from ..clients.powerdns import PowerDNSClient
from ..clients.caddy import CaddyClient
from ..models.dns import DNSRecord, Zone

@dataclass
class GatewayService:
    """Coordinates DNS and proxy configuration."""

    dns_client: PowerDNSClient
    caddy_client: CaddyClient

    async def sync_dns_with_caddy(self, zone_id: str) -> int:
        """
        Ensure DNS records exist for all Caddy routes.
        Returns count of records created.
        """
        # Get current Caddy config
        caddy_config = await self.caddy_client.get_config()
        routes = self._extract_routes(caddy_config)

        # Get existing DNS records
        zone = await self.dns_client.get_zone(zone_id)
        existing = {r.name for r in zone.records}

        # Create missing records
        created = 0
        for route in routes:
            if route.hostname not in existing:
                await self.dns_client.create_record(
                    zone_id,
                    DNSRecord(
                        name=route.hostname,
                        type="A",
                        content=route.target_ip
                    )
                )
                created += 1

        return created