"""Centrální správa IP access — katalog pojmenovaných IP adres + hromadné přiřazování ke službám.
Doplněk k per-service kartě v detailu služby: adresu ("Klient X – kancelář") založím jednou
a pak ji jedním formulářem přiřadím do libovolného počtu služeb. Přiřazení se ukládá do
existující tabulky ``app_ip_access_rules``, ze které generuje pravidla Caddyfile — ta zůstává
beze změny zdrojem pravdy, přibyla jen vazba ``address_id`` na katalog.
"""
import html
from urllib.parse import urlencode
from fastapi import APIRouter, Depends, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from ..auth import is_admin, require_developer
from ..db.apps import get_apps
from ..db.audit import log_audit_event
from ..db.ip_access import (
create_ip_address,
delete_ip_address,
get_assignments,
get_ip_address,
get_ip_addresses,
get_manual_rules,
import_manual_rules,
set_address_assignments,
set_app_assignments,
update_ip_address,
)
from ..routes.apps import IP_ACCESS_METHOD_OPTIONS
from ..routes.runtime import render_action_buttons
from ..templates.layout import page
router = APIRouter()
PAGE_URL = "/portal/admin/ip-access"
MODE_ADDRESS = "address"
MODE_SERVICE = "service"
def _clean(value: str | None) -> str:
return (value or "").strip()
def _checked(value) -> str:
return " checked" if value else ""
def _method_options(selected: str) -> str:
selected_upper = _clean(selected).upper()
options = list(IP_ACCESS_METHOD_OPTIONS)
if selected_upper and selected_upper not in options:
options.append(selected_upper)
out = ""
for opt in options:
is_sel = " selected" if opt == selected_upper else ""
out += f''
return out
def _redirect(mode: str, address_id=None, app_id=None, message: str = "", error: str = ""):
params: dict[str, str] = {"mode": mode}
if address_id:
params["address_id"] = str(address_id)
if app_id:
params["app_id"] = str(app_id)
if message:
params["message"] = message
if error:
params["error"] = error
return RedirectResponse(url=f"{PAGE_URL}?{urlencode(params)}", status_code=303)
async def form_lists(request: Request) -> dict[str, list[str]]:
"""Opakovaná pole formuláře (matice přiřazení) načtená přímo z requestu.
``list[str] = Form(...)`` se napříč verzemi FastAPI chová u opakovaných polí nejednotně,
``form.getlist()`` je spolehlivé. Async dependency + sync endpoint znamená, že samotný
handler (a jeho blokující sqlite dotazy) dál běží v threadpoolu jako zbytek portálu.
"""
form = await request.form()
return {name: form.getlist(name) for name in ("row_key", "row_methods", "assigned_keys", "rule_ids")}
def _parse_rows(rows: dict[str, list[str]]) -> dict[str, str]:
"""Z paralelních polí formuláře vytáhne {klíč řádku: metody} jen pro zaškrtnuté řádky.
Skryté ``row_key`` + ``row_methods`` se posílají za každý řádek (proto jsou zarovnané),
zaškrtnutá políčka chodí zvlášť v ``assigned_keys`` — nezaškrtnutý checkbox prohlížeč
neodešle, takže by pole rozházel.
"""
checked = set(rows.get("assigned_keys") or [])
methods_by_key: dict[str, str] = {}
for key, methods in zip(rows.get("row_key") or [], rows.get("row_methods") or []):
methods_by_key[key] = _clean(methods).upper() or "WRITE"
return {key: methods_by_key.get(key, "WRITE") for key in checked if key in methods_by_key}
# --- render -----------------------------------------------------------------
def _render_catalog_card(addresses: list[dict]) -> str:
rows = ""
for address in addresses:
aid = html.escape(str(address.get("id", "")))
form_id = f"ipaddr-{aid}"
label = html.escape(address.get("label", "") or "", quote=True)
ip_cidr = html.escape(address.get("ip_cidr", "") or "", quote=True)
note = html.escape(address.get("note", "") or "", quote=True)
enabled = bool(address.get("is_enabled", 1))
assigned = int(address.get("assigned_count") or 0)
row_class = "" if enabled else " rule-disabled"
rows += f"""
Katalog je zatím prázdný — přidejte první IP adresu níže.
'
return f"""
IP adresy
Pojmenovaný katalog adres. Změna IP/CIDR nebo názvu se automaticky propíše do všech
služeb, kde je adresa přiřazená. Vypnutím adresy odeberete přístup všude naráz —
přiřazení zůstanou zachovaná pro pozdější zapnutí.
Pravidla zadaná přímo v detailu služby. Fungují dál beze změny; importem se jen
navážou na katalog (adresa se najde podle IP/CIDR, jinak se založí), aby šly spravovat
hromadně odtud.
"""
def _render_apply_card(user: dict) -> str:
if not is_admin(user):
return """
Aplikovat na gateway
Změny jsou uložené v databázi. Pregenerování Caddy může spustit jen administrátor
(Admin → Environment → Regenerate Caddy).
"""
return f"""
Aplikovat na gateway
Uložením se změní jen databáze. Aby se pravidla projevila na gateway, spusťte
pregenerování Caddyfile — akce nevolá docker přímo, vytvoří úlohu do fronty
(zpracuje worker). Historii najdete v Úlohách.
"""
def _render_page(user: dict, mode: str, address_id: str, app_id: str, message: str = "", error: str = "") -> str:
addresses = get_ip_addresses()
apps = get_apps()
assignments = get_assignments()
manual_rules = get_manual_rules()
assignment_map = {(int(item["address_id"]), item["app_id"]): item for item in assignments}
mode = MODE_SERVICE if mode == MODE_SERVICE else MODE_ADDRESS
selected_address = None
if addresses:
selected_address = next((item for item in addresses if str(item["id"]) == str(address_id)), addresses[0])
selected_app = None
if apps:
selected_app = next((item for item in apps if str(item.get("id")) == str(app_id)), apps[0])
notice = ""
if message:
notice = f'
{html.escape(message)}
'
if error:
notice = f'
{html.escape(error)}
'
total_assignments = len(assignments)
active_addresses = sum(1 for item in addresses if item.get("is_enabled", 1))
body = f"""
IP Access
Centrální správa přístupů podle IP. Adresu založíte jednou v katalogu a pak ji
přepínačem níže přiřadíte do libovolného počtu služeb — typicky dvě IP klienta do
deseti služeb. Přiřazení se ukládají do stejných pravidel, která používá detail
služby i generátor Caddyfile.
{notice}
IP adres v katalogu{len(addresses)}
Aktivních adres{active_addresses}
Přiřazení celkem{total_assignments}
Pravidel mimo katalog{len(manual_rules)}
{_render_assignment_card(mode, addresses, apps, assignment_map, selected_address, selected_app)}
{_render_catalog_card(addresses)}
{_render_overview_card(addresses, assignments)}
{_render_manual_card(manual_rules)}
{_render_apply_card(user)}
"""
return page("IP Access", body, user=user)
# --- endpoints --------------------------------------------------------------
@router.get("/admin/ip-access", response_class=HTMLResponse)
def ip_access_page(
request: Request,
mode: str = MODE_ADDRESS,
address_id: str = "",
app_id: str = "",
message: str = "",
error: str = "",
user=Depends(require_developer),
):
return _render_page(user, mode, address_id, app_id, message=message, error=error)
@router.post("/admin/ip-access/addresses")
def add_ip_address(
label: str = Form(...),
ip_cidr: str = Form(...),
default_methods: str = Form("WRITE"),
note: str = Form(""),
is_enabled: str | None = Form(None),
user=Depends(require_developer),
):
clean_label = _clean(label)
clean_ip = _clean(ip_cidr)
clean_methods = _clean(default_methods).upper() or "WRITE"
if not clean_label:
return _redirect(MODE_ADDRESS, error="Název adresy nesmí být prázdný.")
if not clean_ip:
return _redirect(MODE_ADDRESS, error="IP/CIDR nesmí být prázdné.")
address_id = create_ip_address(clean_label, clean_ip, _clean(note), clean_methods, bool(is_enabled))
log_audit_event(
user,
action="ip_access.address.created",
target_type="ip_address",
target_id=str(address_id),
metadata={"label": clean_label, "ip_cidr": clean_ip, "default_methods": clean_methods, "is_enabled": bool(is_enabled)},
)
return _redirect(MODE_ADDRESS, address_id=address_id, message=f"IP adresa {clean_label} přidána do katalogu.")
@router.post("/admin/ip-access/addresses/{address_id}/update")
def save_ip_address(
address_id: int,
label: str = Form(...),
ip_cidr: str = Form(...),
default_methods: str = Form("WRITE"),
note: str = Form(""),
is_enabled: str | None = Form(None),
user=Depends(require_developer),
):
existing = get_ip_address(address_id)
if not existing:
return _redirect(MODE_ADDRESS, error="IP adresa nebyla nalezena.")
clean_label = _clean(label)
clean_ip = _clean(ip_cidr)
clean_methods = _clean(default_methods).upper() or "WRITE"
if not clean_label:
return _redirect(MODE_ADDRESS, address_id=address_id, error="Název adresy nesmí být prázdný.")
if not clean_ip:
return _redirect(MODE_ADDRESS, address_id=address_id, error="IP/CIDR nesmí být prázdné.")
affected = update_ip_address(address_id, clean_label, clean_ip, _clean(note), clean_methods, bool(is_enabled))
log_audit_event(
user,
action="ip_access.address.updated",
target_type="ip_address",
target_id=str(address_id),
metadata={
"label": clean_label,
"ip_cidr": clean_ip,
"default_methods": clean_methods,
"is_enabled": bool(is_enabled),
"propagated_rules": affected,
},
)
note_text = f" Propsáno do {affected} přiřazených pravidel." if affected else ""
return _redirect(MODE_ADDRESS, address_id=address_id, message=f"IP adresa {clean_label} uložena.{note_text}")
@router.post("/admin/ip-access/addresses/{address_id}/delete")
def remove_ip_address(address_id: int, user=Depends(require_developer)):
existing = get_ip_address(address_id)
if not existing:
return _redirect(MODE_ADDRESS, error="IP adresa nebyla nalezena.")
affected = delete_ip_address(address_id)
log_audit_event(
user,
action="ip_access.address.deleted",
target_type="ip_address",
target_id=str(address_id),
metadata={"label": existing.get("label"), "ip_cidr": existing.get("ip_cidr"), "removed_rules": affected},
)
return _redirect(
MODE_ADDRESS,
message=f"IP adresa {existing.get('label')} smazána včetně {affected} přiřazení.",
)
@router.post("/admin/ip-access/addresses/{address_id}/assignments")
def save_address_assignments(
address_id: int,
rows: dict[str, list[str]] = Depends(form_lists),
user=Depends(require_developer),
):
address = get_ip_address(address_id)
if not address:
return _redirect(MODE_ADDRESS, error="IP adresa nebyla nalezena.")
selected = _parse_rows(rows)
known_apps = {item.get("id", "") for item in get_apps()}
unknown = sorted(set(selected) - known_apps)
if unknown:
return _redirect(
MODE_ADDRESS,
address_id=address_id,
error="Neznámé služby v požadavku: " + ", ".join(unknown),
)
counts = set_address_assignments(address_id, {app: (methods, True) for app, methods in selected.items()})
log_audit_event(
user,
action="ip_access.assignments.updated",
target_type="ip_address",
target_id=str(address_id),
metadata={"label": address.get("label"), "app_ids": sorted(selected), **counts},
)
return _redirect(
MODE_ADDRESS,
address_id=address_id,
message=(
f"Přiřazení adresy {address.get('label')} uloženo — {len(selected)} služeb "
f"(+{counts['created']} / ~{counts['updated']} / -{counts['deleted']}). "
"Nezapomeňte pregenerovat Caddy."
),
)
@router.post("/admin/ip-access/apps/{app_id}/assignments")
def save_app_ip_assignments(
app_id: str,
rows: dict[str, list[str]] = Depends(form_lists),
user=Depends(require_developer),
):
apps = get_apps()
app = next((item for item in apps if item.get("id") == app_id), None)
if not app:
return _redirect(MODE_SERVICE, error="Služba nebyla nalezena.")
selected = _parse_rows(rows)
known_addresses = {str(item["id"]) for item in get_ip_addresses()}
unknown = sorted(set(selected) - known_addresses)
if unknown:
return _redirect(
MODE_SERVICE,
app_id=app_id,
error="Neznámé IP adresy v požadavku: " + ", ".join(unknown),
)
counts = set_app_assignments(app_id, {int(key): (methods, True) for key, methods in selected.items()})
log_audit_event(
user,
action="ip_access.assignments.updated",
target_type="app",
target_id=app_id,
metadata={"address_ids": sorted(int(key) for key in selected), **counts},
)
return _redirect(
MODE_SERVICE,
app_id=app_id,
message=(
f"Přiřazení služby {app.get('name') or app_id} uloženo — {len(selected)} IP adres "
f"(+{counts['created']} / ~{counts['updated']} / -{counts['deleted']}). "
"Nezapomeňte pregenerovat Caddy."
),
)
@router.post("/admin/ip-access/manual-rules/import")
def import_manual_ip_rules(rows: dict[str, list[str]] = Depends(form_lists), user=Depends(require_developer)):
rule_ids = [int(value) for value in rows.get("rule_ids") or [] if value.isdigit()]
if not rule_ids:
return _redirect(MODE_ADDRESS, error="Nebyla vybraná žádná pravidla k importu.")
counts = import_manual_rules(rule_ids)
log_audit_event(
user,
action="ip_access.manual_rules.imported",
target_type="ip_address",
target_id=None,
metadata={"rule_ids": rule_ids, **counts},
)
return _redirect(
MODE_ADDRESS,
message=(
f"Importováno {counts['rules_linked']} pravidel, "
f"nově založeno {counts['addresses_created']} adres v katalogu."
),
)