"""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""" {assigned} služeb
""" if not rows: rows = '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í.

{rows}
Název IP/CIDR Výchozí metody Poznámka Stav Přiřazeno Akce

Přidat IP adresu

""" def _render_assignment_card( mode: str, addresses: list[dict], apps: list[dict], assignment_map: dict[tuple[int, str], dict], selected_address: dict | None, selected_app: dict | None, ) -> str: address_tab = "btn" if mode == MODE_ADDRESS else "btn btn-secondary" service_tab = "btn" if mode == MODE_SERVICE else "btn btn-secondary" selected_address_id = int(selected_address["id"]) if selected_address else 0 selected_app_id = selected_app.get("id", "") if selected_app else "" switcher = f"""
Podle IP adresy Podle služby
""" if mode == MODE_ADDRESS: if not addresses: return f"""

Přiřazení

{switcher}

Nejdřív přidejte alespoň jednu IP adresu do katalogu.

""" options = "".join( ''.format( value=int(item["id"]), sel=" selected" if int(item["id"]) == selected_address_id else "", text=html.escape( "{label} ({ip}){off}".format( label=item.get("label", "") or "", ip=item.get("ip_cidr", "") or "", off="" if item.get("is_enabled", 1) else " – vypnuto", ) ), ) for item in addresses ) selector = f"""
""" disabled_note = "" if not selected_address.get("is_enabled", 1): disabled_note = ( '

Adresa je vypnutá — přiřazení se ukládají, ale všechna pravidla ' "zůstanou neaktivní, dokud adresu nezapnete v katalogu výše.

" ) row_html = "" for item in apps: app_id = item.get("id", "") or "" key = html.escape(app_id, quote=True) rule = assignment_map.get((selected_address_id, app_id)) methods = rule.get("methods") if rule else selected_address.get("default_methods", "WRITE") row_html += f""" {html.escape(item.get("name", "") or app_id)}
{html.escape(app_id)}
""" if not row_html: row_html = 'Zatím nejsou evidované žádné služby.' action = f"{PAGE_URL}/addresses/{selected_address_id}/assignments" heading = "Do kterých služeb pustit adresu {label} ({ip})".format( label=html.escape(selected_address.get("label", "") or ""), ip=html.escape(selected_address.get("ip_cidr", "") or ""), ) first_column = "Přiřadit" second_column = "Služba" else: if not apps: return f"""

Přiřazení

{switcher}

Zatím nejsou evidované žádné služby.

""" options = "".join( ''.format( value=html.escape(item.get("id", "") or "", quote=True), sel=" selected" if (item.get("id", "") or "") == selected_app_id else "", text=html.escape("{name} ({id})".format(name=item.get("name", "") or item.get("id", ""), id=item.get("id", ""))), ) for item in apps ) selector = f"""
""" disabled_note = "" row_html = "" for item in addresses: address_id = int(item["id"]) key = str(address_id) rule = assignment_map.get((address_id, selected_app_id)) methods = rule.get("methods") if rule else item.get("default_methods", "WRITE") off_pill = "" if item.get("is_enabled", 1) else ' vypnuto' row_html += f""" {html.escape(item.get("label", "") or "")}{off_pill}
{html.escape(item.get("ip_cidr", "") or "")}
""" if not row_html: row_html = 'Katalog je zatím prázdný — přidejte první IP adresu výše.' action = f"{PAGE_URL}/apps/{html.escape(selected_app_id, quote=True)}/assignments" heading = "Které IP adresy pustit do služby {name}".format( name=html.escape(selected_app.get("name", "") or selected_app_id), ) first_column = "Přiřadit" second_column = "IP adresa" return f"""

Přiřazení

{switcher} {selector}

{heading}

{disabled_note}
{row_html}
{first_column} {second_column} Metody

Odškrtnuté řádky se při uložení smažou. Metody: WRITE = POST, PUT, PATCH, DELETE; ALL = všechny běžné metody včetně GET.

""" def _render_overview_card(addresses: list[dict], assignments: list[dict]) -> str: by_address: dict[int, list[dict]] = {} for item in assignments: by_address.setdefault(int(item["address_id"]), []).append(item) rows = "" for address in addresses: address_id = int(address["id"]) items = by_address.get(address_id, []) if items: pills = "".join( '{name} '.format( muted="" if item.get("is_enabled", 1) else " pill-muted", app=html.escape(item.get("app_id", "") or "", quote=True), methods=html.escape(item.get("methods", "") or "", quote=True), name=html.escape(item.get("app_name", "") or item.get("app_id", "") or ""), ) for item in items ) else: pills = 'bez přiřazení' state = "" if address.get("is_enabled", 1) else ' vypnuto' rows += f""" {html.escape(address.get("label", "") or "")}{state}
{html.escape(address.get("ip_cidr", "") or "")}
{pills} """ if not rows: rows = 'Katalog je zatím prázdný.' return f"""

Přehled

{rows}
IP adresa Přiřazené služby
""" def _render_manual_card(manual_rules: list[dict]) -> str: if not manual_rules: return "" rows = "" for rule in manual_rules: rid = html.escape(str(rule.get("id", ""))) app_id = html.escape(rule.get("app_id", "") or "", quote=True) rows += f""" {html.escape(rule.get("app_name", "") or rule.get("app_id", "") or "")} {html.escape(rule.get("ip_cidr", "") or "")} {html.escape(rule.get("methods", "") or "")} {html.escape(rule.get("description", "") or "")} {"Ano" if rule.get("is_enabled", 1) else "Ne"} """ return f"""

Pravidla mimo katalog

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.

{rows}
Import Služba IP/CIDR Metody Popis Aktivní
""" 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.

{render_action_buttons(["regenerate-caddy"], PAGE_URL)}
""" 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." ), )