diff --git a/app/db/apps.py b/app/db/apps.py index 5dbb137..83e6f07 100644 --- a/app/db/apps.py +++ b/app/db/apps.py @@ -414,11 +414,13 @@ def get_app_ip_access_rules(app_id: str): rows = con.execute( """ - SELECT id, app_id, ip_cidr, methods, description, - COALESCE(is_enabled, 1) AS is_enabled, created_at, updated_at - FROM app_ip_access_rules - WHERE app_id = ? - ORDER BY id + SELECT r.id, r.app_id, r.ip_cidr, r.methods, r.description, + COALESCE(r.is_enabled, 1) AS is_enabled, r.created_at, r.updated_at, + r.address_id, a.label AS address_label + FROM app_ip_access_rules r + LEFT JOIN ip_addresses a ON a.id = r.address_id + WHERE r.app_id = ? + ORDER BY r.id """, (app_id,), ).fetchall() @@ -433,10 +435,12 @@ def get_app_ip_access_rule(rule_id: int, app_id: str): row = con.execute( """ - SELECT id, app_id, ip_cidr, methods, description, - COALESCE(is_enabled, 1) AS is_enabled, created_at, updated_at - FROM app_ip_access_rules - WHERE id = ? AND app_id = ? + SELECT r.id, r.app_id, r.ip_cidr, r.methods, r.description, + COALESCE(r.is_enabled, 1) AS is_enabled, r.created_at, r.updated_at, + r.address_id, a.label AS address_label + FROM app_ip_access_rules r + LEFT JOIN ip_addresses a ON a.id = r.address_id + WHERE r.id = ? AND r.app_id = ? """, (rule_id, app_id), ).fetchone() diff --git a/app/db/ip_access.py b/app/db/ip_access.py new file mode 100644 index 0000000..2fee89f --- /dev/null +++ b/app/db/ip_access.py @@ -0,0 +1,368 @@ +"""Katalog pojmenovaných IP adres a jejich přiřazování ke službám. + +Zdrojem pravdy pro vynucování zůstává tabulka ``app_ip_access_rules`` (jeden řádek = jedna +IP/CIDR + metody pro jednu službu) — tu čte generátor Caddyfile. Tabulka ``ip_addresses`` je +nad ní jen pojmenovaný katalog: adresu založím jednou a pak ji přiřazuji do N služeb. +Přiřazený řádek má ``address_id`` na katalog a ``ip_cidr`` zkopírovanou z katalogu, ručně +zadaná pravidla mají ``address_id = NULL`` a fungují dál bez změny. +""" + +from app.db.database import get_connection +from app.db.migrations import run_migrations + + +def get_ip_addresses() -> list[dict]: + """Katalog adres včetně počtu služeb, do kterých je adresa přiřazená.""" + run_migrations() + con = get_connection() + + rows = con.execute( + """ + SELECT + a.id, + a.label, + a.ip_cidr, + a.note, + COALESCE(a.default_methods, 'WRITE') AS default_methods, + COALESCE(a.is_enabled, 1) AS is_enabled, + a.created_at, + a.updated_at, + ( + SELECT COUNT(*) + FROM app_ip_access_rules r + WHERE r.address_id = a.id + ) AS assigned_count, + ( + SELECT COUNT(*) + FROM app_ip_access_rules r + WHERE r.address_id = a.id AND COALESCE(r.is_enabled, 1) = 1 + ) AS assigned_enabled_count + FROM ip_addresses a + ORDER BY a.label COLLATE NOCASE, a.id + """ + ).fetchall() + + con.close() + return [dict(row) for row in rows] + + +def get_ip_address(address_id: int) -> dict | None: + run_migrations() + con = get_connection() + + row = con.execute( + """ + SELECT + id, + label, + ip_cidr, + note, + COALESCE(default_methods, 'WRITE') AS default_methods, + COALESCE(is_enabled, 1) AS is_enabled, + created_at, + updated_at + FROM ip_addresses + WHERE id = ? + """, + (address_id,), + ).fetchone() + + con.close() + return dict(row) if row else None + + +def find_ip_address_by_cidr(ip_cidr: str) -> dict | None: + run_migrations() + con = get_connection() + + row = con.execute( + """ + SELECT id, label, ip_cidr + FROM ip_addresses + WHERE ip_cidr = ? + ORDER BY id + LIMIT 1 + """, + (ip_cidr,), + ).fetchone() + + con.close() + return dict(row) if row else None + + +def create_ip_address(label: str, ip_cidr: str, note: str, default_methods: str, is_enabled: bool) -> int: + run_migrations() + con = get_connection() + + con.execute( + """ + INSERT INTO ip_addresses (label, ip_cidr, note, default_methods, is_enabled, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + """, + (label, ip_cidr, note, default_methods, 1 if is_enabled else 0), + ) + address_id = int(con.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]) + + con.commit() + con.close() + return address_id + + +def update_ip_address(address_id: int, label: str, ip_cidr: str, note: str, default_methods: str, is_enabled: bool) -> int: + """Uloží adresu a propíše novou IP/CIDR i popis do všech jejích přiřazení. + + Vrací počet přiřazených pravidel, kterých se propsání dotklo — kvůli hlášce pro uživatele + (změna adresy se projeví ve všech službách, kde je přiřazená). + """ + run_migrations() + con = get_connection() + + con.execute( + """ + UPDATE ip_addresses + SET label = ?, + ip_cidr = ?, + note = ?, + default_methods = ?, + is_enabled = ?, + updated_at = CURRENT_TIMESTAMP + WHERE id = ? + """, + (label, ip_cidr, note, default_methods, 1 if is_enabled else 0, address_id), + ) + + # Přiřazení drží kopii IP/CIDR (kvůli generátoru Caddyfile), takže se musí přepsat spolu s + # katalogem. Vypnutá adresa vypne i všechna svá pravidla, zapnutí je nechává na uživateli. + cursor = con.execute( + """ + UPDATE app_ip_access_rules + SET ip_cidr = ?, + description = ?, + is_enabled = CASE WHEN ? = 0 THEN 0 ELSE COALESCE(is_enabled, 1) END, + updated_at = CURRENT_TIMESTAMP + WHERE address_id = ? + """, + (ip_cidr, label, 1 if is_enabled else 0, address_id), + ) + affected = cursor.rowcount or 0 + + con.commit() + con.close() + return affected + + +def delete_ip_address(address_id: int) -> int: + """Smaže adresu z katalogu i všechna její přiřazení. Vrací počet smazaných přiřazení.""" + run_migrations() + con = get_connection() + + cursor = con.execute("DELETE FROM app_ip_access_rules WHERE address_id = ?", (address_id,)) + affected = cursor.rowcount or 0 + con.execute("DELETE FROM ip_addresses WHERE id = ?", (address_id,)) + + con.commit() + con.close() + return affected + + +def get_assignments() -> list[dict]: + """Všechna přiřazení z katalogu (bez ručních pravidel) napříč službami.""" + run_migrations() + con = get_connection() + + rows = con.execute( + """ + SELECT + r.id, + r.app_id, + r.address_id, + r.ip_cidr, + r.methods, + COALESCE(r.is_enabled, 1) AS is_enabled, + a.label, + COALESCE(app.name, r.app_id) AS app_name + FROM app_ip_access_rules r + JOIN ip_addresses a ON a.id = r.address_id + LEFT JOIN apps app ON app.id = r.app_id + ORDER BY a.label COLLATE NOCASE, r.app_id + """ + ).fetchall() + + con.close() + return [dict(row) for row in rows] + + +def get_manual_rules() -> list[dict]: + """Pravidla zadaná ručně v detailu služby (bez vazby na katalog) — kandidáti na import.""" + run_migrations() + con = get_connection() + + rows = con.execute( + """ + SELECT + r.id, + r.app_id, + r.ip_cidr, + r.methods, + r.description, + COALESCE(r.is_enabled, 1) AS is_enabled, + COALESCE(app.name, r.app_id) AS app_name + FROM app_ip_access_rules r + LEFT JOIN apps app ON app.id = r.app_id + WHERE r.address_id IS NULL + ORDER BY r.ip_cidr, r.app_id + """ + ).fetchall() + + con.close() + return [dict(row) for row in rows] + + +def _sync_assignments(con, pairs: dict[tuple[int, str], tuple[str, bool]], scope_column: str, scope_value) -> dict[str, int]: + """Srovná přiřazení v jednom rozsahu (jedna adresa, nebo jedna služba) s požadovaným stavem. + + ``pairs`` mapuje (address_id, app_id) -> (methods, is_enabled) pro přiřazení, která mají po + uložení existovat. Cokoli navázaného na katalog v daném rozsahu a chybějícího v ``pairs`` se + smaže. Vrací počty created/updated/deleted pro hlášku a audit. + """ + existing_rows = con.execute( + f""" + SELECT id, address_id, app_id, methods, COALESCE(is_enabled, 1) AS is_enabled + FROM app_ip_access_rules + WHERE address_id IS NOT NULL AND {scope_column} = ? + """, + (scope_value,), + ).fetchall() + + existing = {(int(row["address_id"]), row["app_id"]): dict(row) for row in existing_rows} + counts = {"created": 0, "updated": 0, "deleted": 0} + + for key, row in existing.items(): + if key not in pairs: + con.execute("DELETE FROM app_ip_access_rules WHERE id = ?", (row["id"],)) + counts["deleted"] += 1 + + for (address_id, app_id), (methods, is_enabled) in pairs.items(): + address = con.execute( + "SELECT label, ip_cidr, COALESCE(is_enabled, 1) AS is_enabled FROM ip_addresses WHERE id = ?", + (address_id,), + ).fetchone() + if not address: + # Adresa mezitím zmizela (paralelní smazání) – přiřazení nemá kam ukazovat, přeskoč. + continue + + # Vypnutá adresa nesmí vzniknout jako aktivní pravidlo, i kdyby to formulář poslal. + effective_enabled = bool(is_enabled) and bool(address["is_enabled"]) + row = existing.get((address_id, app_id)) + + if row is None: + con.execute( + """ + INSERT INTO app_ip_access_rules ( + app_id, address_id, ip_cidr, methods, description, is_enabled, created_at, updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + """, + (app_id, address_id, address["ip_cidr"], methods, address["label"], 1 if effective_enabled else 0), + ) + counts["created"] += 1 + elif row["methods"] != methods or bool(row["is_enabled"]) != effective_enabled: + con.execute( + """ + UPDATE app_ip_access_rules + SET ip_cidr = ?, + methods = ?, + description = ?, + is_enabled = ?, + updated_at = CURRENT_TIMESTAMP + WHERE id = ? + """, + (address["ip_cidr"], methods, address["label"], 1 if effective_enabled else 0, row["id"]), + ) + counts["updated"] += 1 + + return counts + + +def set_address_assignments(address_id: int, assignments: dict[str, tuple[str, bool]]) -> dict[str, int]: + """Nastaví, do kterých služeb je adresa přiřazená. ``assignments``: app_id -> (methods, enabled).""" + run_migrations() + con = get_connection() + + pairs = {(address_id, app_id): value for app_id, value in assignments.items()} + counts = _sync_assignments(con, pairs, "address_id", address_id) + + con.commit() + con.close() + return counts + + +def set_app_assignments(app_id: str, assignments: dict[int, tuple[str, bool]]) -> dict[str, int]: + """Nastaví, které adresy má služba přiřazené. ``assignments``: address_id -> (methods, enabled).""" + run_migrations() + con = get_connection() + + pairs = {(address_id, app_id): value for address_id, value in assignments.items()} + counts = _sync_assignments(con, pairs, "app_id", app_id) + + con.commit() + con.close() + return counts + + +def import_manual_rules(rule_ids: list[int]) -> dict[str, int]: + """Naváže vybraná ruční pravidla na katalog (adresu podle IP/CIDR najde, nebo založí). + + Vrací počty vytvořených adres a navázaných pravidel. + """ + run_migrations() + con = get_connection() + + counts = {"addresses_created": 0, "rules_linked": 0} + for rule_id in rule_ids: + rule = con.execute( + """ + SELECT id, ip_cidr, methods, description + FROM app_ip_access_rules + WHERE id = ? AND address_id IS NULL + """, + (rule_id,), + ).fetchone() + if not rule: + continue + + address = con.execute( + "SELECT id, label FROM ip_addresses WHERE ip_cidr = ? ORDER BY id LIMIT 1", + (rule["ip_cidr"],), + ).fetchone() + + if address: + address_id = int(address["id"]) + label = address["label"] + else: + label = (rule["description"] or "").strip() or rule["ip_cidr"] + con.execute( + """ + INSERT INTO ip_addresses (label, ip_cidr, note, default_methods, is_enabled, created_at, updated_at) + VALUES (?, ?, ?, ?, 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + """, + (label, rule["ip_cidr"], "Importováno z pravidla v detailu služby.", rule["methods"] or "WRITE"), + ) + address_id = int(con.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]) + counts["addresses_created"] += 1 + + con.execute( + """ + UPDATE app_ip_access_rules + SET address_id = ?, + description = ?, + updated_at = CURRENT_TIMESTAMP + WHERE id = ? + """, + (address_id, label, rule["id"]), + ) + counts["rules_linked"] += 1 + + con.commit() + con.close() + return counts diff --git a/app/db/migrations.py b/app/db/migrations.py index 2711fb9..40ade51 100644 --- a/app/db/migrations.py +++ b/app/db/migrations.py @@ -285,5 +285,28 @@ def run_migrations(): ) con.execute("CREATE INDEX IF NOT EXISTS idx_app_ip_access_rules_app_id ON app_ip_access_rules(app_id)") + # Pojmenovaný katalog IP adres. Jedna adresa (např. "Klient X – kancelář") se přiřazuje do N + # služeb; přiřazení zůstává řádkem v app_ip_access_rules (zdroj pravdy pro generátor Caddyfile), + # jen s vazbou address_id na katalog. Ručně zadaná pravidla mají address_id = NULL. + con.execute( + """ + CREATE TABLE IF NOT EXISTS ip_addresses ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + label TEXT NOT NULL, + ip_cidr TEXT NOT NULL, + note TEXT, + default_methods TEXT NOT NULL DEFAULT 'WRITE', + is_enabled INTEGER NOT NULL DEFAULT 1, + created_at TEXT DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + con.execute("CREATE INDEX IF NOT EXISTS idx_ip_addresses_label ON ip_addresses(label)") + + if "address_id" not in _table_columns(con, "app_ip_access_rules"): + con.execute("ALTER TABLE app_ip_access_rules ADD COLUMN address_id INTEGER") + con.execute("CREATE INDEX IF NOT EXISTS idx_app_ip_access_rules_address_id ON app_ip_access_rules(address_id)") + con.commit() con.close() diff --git a/app/main.py b/app/main.py index a4e24f7..dee8463 100644 --- a/app/main.py +++ b/app/main.py @@ -6,7 +6,7 @@ from starlette.middleware.sessions import SessionMiddleware from .config import read_env_bool, read_env_value from .logging_config import configure_logging -from .routes import alerting, apps, audit, auth, backups, catalog, deployments, developers, environment, health, incidents, jobs, logs, operations, runtime, scheduled_scripts, users, workers +from .routes import alerting, apps, audit, auth, backups, catalog, deployments, developers, environment, health, incidents, ip_access, jobs, logs, operations, runtime, scheduled_scripts, users, workers def create_app() -> FastAPI: @@ -42,6 +42,7 @@ def create_app() -> FastAPI: app.include_router(audit.router) app.include_router(runtime.router) app.include_router(environment.router) + app.include_router(ip_access.router) return app diff --git a/app/routes/apps.py b/app/routes/apps.py index 2637304..4a0a171 100644 --- a/app/routes/apps.py +++ b/app/routes/apps.py @@ -759,11 +759,27 @@ def app_detail(app_id: str, request: Request, message: str = "", error: str = "" row_class = "" if rule_enabled else " rule-disabled" enabled_checked = bool_checked(rule_enabled) ip_label = html.escape(rule.get("ip_cidr", "") or "") + # Pravidla navázaná na katalog IP adres drží IP/CIDR i popis podle katalogu, aby se obě + # místa nerozešla – tady jsou proto jen ke čtení a mění se v centrální správě IP Access. + address_id = rule.get("address_id") + if address_id: + address_label = html.escape(rule.get("address_label") or rule.get("description") or "", quote=True) + source_cell = ( + f'{address_label}' + ) + ip_cell = f'' + description_cell = f'' + else: + source_cell = 'ručně' + ip_cell = f'' + description_cell = f'' ip_access_rule_rows += f""" - + {ip_cell} - + {description_cell} + {source_cell}
@@ -776,7 +792,7 @@ def app_detail(app_id: str, request: Request, message: str = "", error: str = "" """ if not ip_access_rule_rows: - ip_access_rule_rows = 'Zatím nejsou evidovaná žádná IP access pravidla.' + ip_access_rule_rows = 'Zatím nejsou evidovaná žádná IP access pravidla.' 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. +

+ @@ -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""" + + + + + + + + + + """ + + if not rows: + rows = '' + + 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í. +

+
IP/CIDR Methods DescriptionZdroj Enabled Actions
{assigned} služeb +
+ +
+ +
+
Katalog je zatím prázdný — přidejte první IP adresu níže.
+ + + + + + + + + + {rows} +
NázevIP/CIDRVýchozí metodyPoznámkaStavPřiřazenoAkce
+ +
+

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 adresaPř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} +
ImportSlužbaIP/CIDRMetodyPopisAktivní
+
+ +
+
+
+ """ + + +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." + ), + ) 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":