app.py

@@ -895,130 +895,6 @@ async def vault_reload(_=Depends(_verify)):
     return {"reloaded": True, "items": len(_vault_cache)}
 
 
-# --- VPS reverse-proxy and DNS management ---
-
-VPS_SSH = "root@46.225.230.72"
-VPS_IPV4 = "46.225.230.72"
-VPS_IPV6 = "2a01:4f8:1c19:9653::1"
-MANAGED_DNS_ZONES = {"guck.tv"}
-
-
-def _validate_proxy_route(domain: str, upstream: str) -> tuple[str, str, str, str]:
-    domain = domain.strip().lower().rstrip(".")
-    upstream = upstream.strip().lower()
-    if not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)+", domain):
-        raise ValueError("invalid domain")
-    zone = next((item for item in MANAGED_DNS_ZONES if domain.endswith(f".{item}")), None)
-    if not zone or domain == zone:
-        raise ValueError("domain is outside managed DNS zones or is a zone apex")
-    match = re.fullmatch(r"(127\.0\.0\.1|localhost):(\d{1,5})", upstream)
-    if not match or not 1 <= int(match.group(2)) <= 65535:
-        raise ValueError("upstream must be localhost with a valid TCP port")
-    return domain, upstream, zone, domain[: -(len(zone) + 1)]
-
-
-class ProxyRouteRequest(BaseModel):
-    domain: str
-    upstream: str
-
-
-def _remote_python(script: str, timeout: int = 30) -> tuple[int, str, str]:
-    encoded = base64.b64encode(script.encode()).decode()
-    return _ssh(VPS_SSH, f'python3 -c "import base64;exec(base64.b64decode(\'{encoded}\'))"', timeout=timeout)
-
-
-def _restore_caddy_backup(backup: str):
-    if not re.fullmatch(r"/app-config/caddy/Caddyfile\.bak-\d{8}T\d{6}Z", backup):
-        raise ValueError("invalid Caddy backup path")
-    script = f"""from pathlib import Path
-Path('/app-config/caddy/Caddyfile').write_bytes(Path({backup!r}).read_bytes())
-"""
-    rc, _out, err = _remote_python(script)
-    if rc != 0:
-        raise RuntimeError(f"Caddy rollback write failed: {err[-300:]}")
-    rc, _out, err = _ssh(VPS_SSH, "docker exec caddy caddy reload --config /etc/caddy/Caddyfile", timeout=30)
-    if rc != 0:
-        raise RuntimeError(f"Caddy rollback reload failed: {err[-300:]}")
-
-
-def _configure_caddy_route(domain: str, upstream: str) -> dict:
-    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
-    backup = f"/app-config/caddy/Caddyfile.bak-{timestamp}"
-    block = f"{domain} {{\n    reverse_proxy {upstream}\n}}\n\n"
-    script = f"""from pathlib import Path
-import re, shutil
-path = Path('/app-config/caddy/Caddyfile')
-backup = Path({backup!r})
-content = path.read_text()
-block = {block!r}
-pattern = re.compile(r'(?ms)^{re.escape(domain)}\\s*\\{{.*?^\\}}\\s*')
-shutil.copy2(path, backup)
-if pattern.search(content):
-    content = pattern.sub(block, content, count=1)
-else:
-    if content and not content.endswith('\\n'):
-        content += '\\n'
-    content += '\\n' + block
-path.write_text(content)
-print(backup)
-"""
-    rc, out, err = _remote_python(script)
-    if rc != 0:
-        raise RuntimeError(f"Caddyfile update failed: {err[-300:]}")
-    backup = out.strip() or backup
-    rc, _out, err = _ssh(VPS_SSH, "docker exec caddy caddy validate --config /etc/caddy/Caddyfile", timeout=30)
-    if rc != 0:
-        _restore_caddy_backup(backup)
-        raise RuntimeError(f"Caddy validation failed: {err[-300:]}")
-    rc, _out, err = _ssh(VPS_SSH, "docker exec caddy caddy reload --config /etc/caddy/Caddyfile", timeout=30)
-    if rc != 0:
-        _restore_caddy_backup(backup)
-        raise RuntimeError(f"Caddy reload failed: {err[-300:]}")
-    return {"status": "reloaded", "backup": backup}
-
-
-async def _upsert_dns_records(zone: str, name: str) -> dict:
-    token = _read("HETZNER_DNS_TOKEN")
-    if not token:
-        raise RuntimeError("HETZNER_DNS_TOKEN is unavailable")
-    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
-    api = "https://api.hetzner.cloud/v1"
-    async with httpx.AsyncClient(timeout=30) as client:
-        zones_response = await client.get(f"{api}/zones", headers=headers)
-        zones_response.raise_for_status()
-        zone_data = next((item for item in zones_response.json().get("zones", []) if item.get("name") == zone), None)
-        if not zone_data:
-            raise RuntimeError(f"DNS zone not found: {zone}")
-        zone_id = zone_data["id"]
-        rrsets_response = await client.get(f"{api}/zones/{zone_id}/rrsets", headers=headers)
-        rrsets_response.raise_for_status()
-        existing = {(item.get("name"), item.get("type")) for item in rrsets_response.json().get("rrsets", [])}
-        for record_type, value in (("A", VPS_IPV4), ("AAAA", VPS_IPV6)):
-            payload = {"name": name, "type": record_type, "ttl": 300, "records": [{"value": value, "comment": "Managed by Homelab Butler"}]}
-            if (name, record_type) in existing:
-                response = await client.put(f"{api}/zones/{zone_id}/rrsets/{name}/{record_type}", headers=headers, json=payload)
-            else:
-                response = await client.post(f"{api}/zones/{zone_id}/rrsets", headers=headers, json=payload)
-            response.raise_for_status()
-    return {"zone_id": zone_id, "records": ["A", "AAAA"]}
-
-
-@app.post("/vps/proxy-route")
-async def vps_proxy_route(req: ProxyRouteRequest, _=Depends(_verify)):
-    try:
-        domain, upstream, zone, name = _validate_proxy_route(req.domain, req.upstream)
-    except ValueError as exc:
-        raise HTTPException(400, str(exc)) from exc
-    caddy = await asyncio.to_thread(_configure_caddy_route, domain, upstream)
-    try:
-        dns = await _upsert_dns_records(zone, name)
-    except Exception:
-        await asyncio.to_thread(_restore_caddy_backup, caddy["backup"])
-        raise
-    _audit("/vps/proxy-route", "POST", 200, f"{domain} -> {upstream}")
-    return {"status": "configured", "domain": domain, "upstream": upstream, "caddy": caddy, "dns": dns}
-
-
 # --- VM Lifecycle Endpoints ---
 import subprocess as _sp
 

tests/test_app.py

@@ -261,51 +261,3 @@ def test_overview_is_compact_deterministic_and_light_model_friendly(monkeypatch)
     assert [item["code"] for item in result["findings"]] == [
         "host_unreachable", "service_auth_failed", "backup_warning", "disk_high"
     ]
-
-
-def test_proxy_route_validation_restricts_domain_and_upstream():
-    assert app._validate_proxy_route("speed.guck.tv", "127.0.0.1:8080") == (
-        "speed.guck.tv", "127.0.0.1:8080", "guck.tv", "speed"
-    )
-
-    for domain, upstream in [
-        ("guck.tv", "127.0.0.1:8080"),
-        ("speed.evil.example", "127.0.0.1:8080"),
-        ("speed.guck.tv", "10.0.0.1:8080"),
-        ("speed.guck.tv", "127.0.0.1:70000"),
-        ("speed.guck.tv;rm", "127.0.0.1:8080"),
-    ]:
-        try:
-            app._validate_proxy_route(domain, upstream)
-        except ValueError:
-            pass
-        else:
-            raise AssertionError(f"unsafe route accepted: {domain} -> {upstream}")
-
-
-def test_proxy_route_endpoint_configures_caddy_and_dns(monkeypatch):
-    calls = []
-
-    def fake_caddy(domain, upstream):
-        calls.append(("caddy", domain, upstream))
-        return {"status": "reloaded", "backup": "/app-config/caddy/Caddyfile.bak-test"}
-
-    async def fake_dns(zone, name):
-        calls.append(("dns", zone, name))
-        return {"zone_id": 123, "records": ["A", "AAAA"]}
-
-    monkeypatch.setattr(app, "_configure_caddy_route", fake_caddy)
-    monkeypatch.setattr(app, "_upsert_dns_records", fake_dns)
-    with TestClient(app.app) as client:
-        response = client.post(
-            "/vps/proxy-route",
-            headers={"Authorization": "Bearer test-token"},
-            json={"domain": "speed.guck.tv", "upstream": "127.0.0.1:8080"},
-        )
-
-    assert response.status_code == 200
-    assert response.json()["status"] == "configured"
-    assert calls == [
-        ("caddy", "speed.guck.tv", "127.0.0.1:8080"),
-        ("dns", "guck.tv", "speed"),
-    ]