File size: 2,506 Bytes
daaa6ed 36ce73b daaa6ed 36ce73b daaa6ed 36ce73b daaa6ed 36ce73b daaa6ed 36ce73b daaa6ed 36ce73b daaa6ed 36ce73b daaa6ed 36ce73b daaa6ed 36ce73b daaa6ed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | """
Port management API.
Single Responsibility: only handles port CRUD for zones.
Reverse proxy logic is in proxy.py — separate reason to change.
"""
from fastapi import APIRouter, Depends, Form, HTTPException
from auth import AuthUser, get_current_user
from config import MIN_PORT, MAX_PORT
from storage import load_meta, save_meta, check_zone_owner
router = APIRouter(prefix="/api/zones/{zone_name}/ports", tags=["ports"])
def _validate_port(port: int):
if not (MIN_PORT <= port <= MAX_PORT):
raise ValueError(f"Port must be between {MIN_PORT} and {MAX_PORT}")
def _validate_zone(meta: dict, zone_name: str):
if zone_name not in meta:
raise ValueError(f"Zone '{zone_name}' does not exist")
@router.get("")
def list_ports(zone_name: str, user: AuthUser = Depends(get_current_user)):
try:
check_zone_owner(zone_name, user.sub, user.role)
meta = load_meta()
_validate_zone(meta, zone_name)
return meta[zone_name].get("ports", [])
except ValueError as e:
raise HTTPException(400, str(e))
@router.post("")
def add_port(zone_name: str, port: int = Form(...), label: str = Form(""), user: AuthUser = Depends(get_current_user)):
try:
_validate_port(port)
check_zone_owner(zone_name, user.sub, user.role)
meta = load_meta()
_validate_zone(meta, zone_name)
ports = meta[zone_name].setdefault("ports", [])
for p in ports:
if p["port"] == port:
raise ValueError(f"Port {port} already mapped in zone '{zone_name}'")
entry = {"port": port, "label": label or f"Port {port}"}
ports.append(entry)
save_meta(meta)
return entry
except ValueError as e:
raise HTTPException(400, str(e))
@router.delete("/{port}")
def remove_port(zone_name: str, port: int, user: AuthUser = Depends(get_current_user)):
try:
check_zone_owner(zone_name, user.sub, user.role)
meta = load_meta()
_validate_zone(meta, zone_name)
ports = meta[zone_name].get("ports", [])
before = len(ports)
meta[zone_name]["ports"] = [p for p in ports if p["port"] != port]
if len(meta[zone_name]["ports"]) == before:
raise ValueError(f"Port {port} not found in zone '{zone_name}'")
save_meta(meta)
return {"ok": True}
except ValueError as e:
raise HTTPException(400, str(e))
|