'
notice = ""
if message:
@@ -908,11 +924,19 @@ def app_detail(app_id: str, request: Request, message: str = "", error: str = ""
WRITE = POST, PUT, PATCH, DELETE; ALL = všechny běžné metody včetně GET.
Pravidla se zatím pouze ukládají (Caddy je začne vynucovat později).
+
+
+ Hromadné přiřazení IP adres
+
+ Pojmenované adresy z katalogu (sloupec Zdroj) se spravují centrálně —
+ tam jednou adresou obsloužíte i deset služeb.
+
IP/CIDR
Methods
Description
+
Zdroj
Enabled
Actions
@@ -1281,12 +1305,19 @@ def save_app_ip_access_rule(
rule_ip = clean_optional(ip_cidr)
rule_methods = clean_optional(methods).upper()
- if not rule_ip:
- return redirect_app_detail(app_id, anchor="ip-access", error="IP/CIDR nesmí být prázdné.")
+ rule_description = description.strip()
if not rule_methods:
return redirect_app_detail(app_id, anchor="ip-access", error="Methods nesmí být prázdné.")
- update_app_ip_access_rule(rule_id, app_id, rule_ip, rule_methods, description.strip(), bool(is_enabled))
+ # Pravidla z katalogu IP adres si IP/CIDR ani popis nedrží sama – ty patří katalogu, jinak by
+ # se obě místa rozešla. Editovatelné odtud zůstávají jen metody a zapnutí/vypnutí.
+ if existing.get("address_id"):
+ rule_ip = existing.get("ip_cidr") or ""
+ rule_description = existing.get("description") or ""
+ elif not rule_ip:
+ return redirect_app_detail(app_id, anchor="ip-access", error="IP/CIDR nesmí být prázdné.")
+
+ update_app_ip_access_rule(rule_id, app_id, rule_ip, rule_methods, rule_description, bool(is_enabled))
log_audit_event(
user,
action="app.ip_access_rule.updated",
diff --git a/app/routes/ip_access.py b/app/routes/ip_access.py
new file mode 100644
index 0000000..19341d5
--- /dev/null
+++ b/app/routes/ip_access.py
@@ -0,0 +1,759 @@
+"""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."
+ ),
+ )
diff --git a/app/static/styles.css b/app/static/styles.css
index 9569e25..f623649 100644
--- a/app/static/styles.css
+++ b/app/static/styles.css
@@ -483,6 +483,29 @@ button:hover,
background: #f3f5f7;
}
+/* Pravidla z katalogu IP adres: IP/CIDR a popis se spravují centrálně, tady jen ke čtení. */
+.ip-rule-row input[readonly] {
+ background: #f3f5f7;
+ color: var(--muted);
+ cursor: not-allowed;
+}
+
+/* Hromadné akce nad maticí přiřazení (označit vše / nastavit metody všem). */
+.assignment-tools {
+ margin: 12px 0;
+ padding: 10px 12px;
+ background: #eef6f9;
+ border: 1px solid var(--border);
+ border-radius: 10px;
+}
+
+.assignment-tools label {
+ font-size: 13px;
+ font-weight: 700;
+ color: var(--secondary);
+ margin-left: 8px;
+}
+
.info-dot {
position: relative;
display: inline-flex;
diff --git a/app/templates/layout.py b/app/templates/layout.py
index 02b4f60..c193e71 100644
--- a/app/templates/layout.py
+++ b/app/templates/layout.py
@@ -14,6 +14,7 @@ ADMIN_MENU_ITEMS = [
("Audit", "/portal/audit", "fa-clipboard-list", False),
("Environment", "/portal/admin/environment", "fa-sliders", True),
("Incidenty", "/portal/incidents", "fa-triangle-exclamation", False),
+ ("IP Access", "/portal/admin/ip-access", "fa-shield-halved", False),
("Nasazení", "/portal/deployments", "fa-rocket", False),
("Plánované skripty", "/portal/scheduled-scripts", "fa-calendar-days", False),
("Služby", "/portal/apps", "fa-server", False),
@@ -58,6 +59,10 @@ def page(title: str, body: str, user=None) -> str:
' Alerting'
' Pro vývojáře'
)
+ # IP Access spravují i developeři, ale admin menu se jim nezobrazuje – proto jen jim
+ # přidáváme odkaz do hlavní navigace, aby stránku nemuseli hledat přes detail služby.
+ if can_operate and role != "admin":
+ nav_links += ' IP Access'
admin_menu = ""
if role == "admin":