appfactory-portal/app/db/ip_access.py

"""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