
===== src/ande_ban/database.py =====
from __future__ import annotations

import os
import sqlite3
from pathlib import Path

SCHEMA = """
PRAGMA journal_mode=WAL;
PRAGMA foreign_keys=ON;
CREATE TABLE IF NOT EXISTS events (
    id INTEGER PRIMARY KEY,
    occurred_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
    event_type TEXT NOT NULL CHECK(event_type IN ('ban','unban','pardon')),
    ip TEXT NOT NULL,
    jail TEXT NOT NULL,
    failures INTEGER NOT NULL DEFAULT 0,
    country TEXT,
    country_name TEXT,
    asn INTEGER,
    organization TEXT,
    geo_provider TEXT,
    offense_number INTEGER NOT NULL DEFAULT 0,
    duration_seconds INTEGER,
    permanent INTEGER NOT NULL DEFAULT 0,
    mode TEXT NOT NULL,
    reason TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_events_ip ON events(ip);
CREATE INDEX IF NOT EXISTS idx_events_time ON events(occurred_at);
CREATE INDEX IF NOT EXISTS idx_events_country ON events(country);
CREATE INDEX IF NOT EXISTS idx_events_type_time
    ON events(event_type, occurred_at);
CREATE INDEX IF NOT EXISTS idx_events_ip_type
    ON events(ip, event_type);
CREATE INDEX IF NOT EXISTS idx_events_jail_type
    ON events(jail, event_type);
CREATE INDEX IF NOT EXISTS idx_events_country_type
    ON events(country, event_type);
CREATE TABLE IF NOT EXISTS active_bans (
    jail TEXT NOT NULL,
    ip TEXT NOT NULL,
    country TEXT,
    country_name TEXT,
    asn INTEGER,
    organization TEXT,
    first_seen TEXT NOT NULL,
    last_seen TEXT NOT NULL,
    source TEXT NOT NULL,
    PRIMARY KEY(jail,ip)
);
CREATE INDEX IF NOT EXISTS idx_active_country ON active_bans(country);
CREATE TABLE IF NOT EXISTS jail_status (
    jail TEXT PRIMARY KEY,
    currently_failed INTEGER NOT NULL DEFAULT 0,
    total_failed INTEGER NOT NULL DEFAULT 0,
    currently_banned INTEGER NOT NULL DEFAULT 0,
    total_banned INTEGER NOT NULL DEFAULT 0,
    last_sync TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS permanent_blacklist (
    ip TEXT PRIMARY KEY,
    country TEXT,
    jail TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
    reason TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS metadata (
    key TEXT PRIMARY KEY,
    value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS sync_runs (
    id INTEGER PRIMARY KEY,
    started_at TEXT NOT NULL,
    finished_at TEXT NOT NULL,
    duration_ms INTEGER NOT NULL,
    status TEXT NOT NULL CHECK(status IN ('success','failure')),
    jail_count INTEGER NOT NULL DEFAULT 0,
    active_bans INTEGER NOT NULL DEFAULT 0,
    events_added INTEGER NOT NULL DEFAULT 0,
    error TEXT
);
CREATE INDEX IF NOT EXISTS idx_sync_runs_started
    ON sync_runs(started_at DESC);
CREATE INDEX IF NOT EXISTS idx_sync_runs_status_started
    ON sync_runs(status, started_at DESC);
"""


def _columns(conn: sqlite3.Connection, table: str) -> set[str]:
    return {row[1] for row in conn.execute(f"PRAGMA table_info({table})")}


def _add_columns(
    conn: sqlite3.Connection,
    table: str,
    additions: dict[str, str],
) -> None:
    columns = _columns(conn, table)

    for name, sql_type in additions.items():
        if name not in columns:
            conn.execute(
                f"ALTER TABLE {table} ADD COLUMN {name} {sql_type}"
            )


def _migrate(conn: sqlite3.Connection) -> None:
    _add_columns(
        conn,
        "active_bans",
        {
            "country_name": "TEXT",
            "asn": "INTEGER",
            "organization": "TEXT",
        },
    )

    _add_columns(
        conn,
        "events",
        {
            "country_name": "TEXT",
            "asn": "INTEGER",
            "organization": "TEXT",
            "geo_provider": "TEXT",
        },
    )

    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_active_asn "
        "ON active_bans(asn)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_events_asn_type "
        "ON events(asn, event_type)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_events_org_type "
        "ON events(organization, event_type)"
    )


def _secure_database_permissions(path: Path) -> None:
    """Keep the database private while allowing the ande-ban group access.

    Installation creates /var/lib/ande-ban as a setgid directory owned by the
    ande-ban group. Root-run Fail2Ban actions therefore inherit that group.
    chmod may legitimately fail for a non-owner opening an existing database,
    so permission repair is best-effort here and enforced by the installer.
    """
    try:
        os.chmod(path, 0o660)
    except (FileNotFoundError, PermissionError):
        pass


def connect(path: Path) -> sqlite3.Connection:
    path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(path, timeout=10)
    _secure_database_permissions(path)
    conn.row_factory = sqlite3.Row
    conn.executescript(SCHEMA)
    _migrate(conn)
    return conn


def offense_count(conn: sqlite3.Connection, ip: str) -> int:
    row = conn.execute(
        "SELECT COUNT(*) AS n FROM events WHERE ip=? AND event_type='ban'", (ip,)
    ).fetchone()
    return int(row["n"])

===== src/ande_ban/events.py =====
from __future__ import annotations

import ipaddress

from .config import Settings
from .database import offense_count
from .storage import get_backend
from .geoip import lookup_geo
from .policy import decide


def record_event(
    settings: Settings,
    event_type: str,
    ip: str,
    jail: str,
    failures: int = 0,
) -> dict:
    ip = str(ipaddress.ip_address(ip))
    jail = jail.strip() or "unknown"

    geo = lookup_geo(
        ip=ip,
        country_database=settings.geoip_database,
        asn_database=settings.geoip_asn_database,
        provider=settings.geoip_provider,
    )

    with get_backend(None, settings.database).connect() as conn:
        previous = offense_count(conn, ip)
        number = previous + 1 if event_type == "ban" else previous

        decision = decide(
            settings,
            ip,
            geo.country,
            number,
        )

        conn.execute(
            """
            INSERT INTO events (
                event_type,
                ip,
                jail,
                failures,
                country,
                country_name,
                asn,
                organization,
                geo_provider,
                offense_number,
                duration_seconds,
                permanent,
                mode,
                reason
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                event_type,
                ip,
                jail,
                failures,
                geo.country,
                geo.country_name,
                geo.asn,
                geo.organization,
                geo.provider,
                number,
                decision.duration_seconds,
                int(decision.permanent),
                settings.mode,
                decision.reason,
            ),
        )

        if (
            event_type == "ban"
            and decision.permanent
            and not decision.trusted
        ):
            conn.execute(
                """
                INSERT INTO permanent_blacklist (
                    ip,
                    country,
                    jail,
                    reason
                )
                VALUES (?, ?, ?, ?)
                ON CONFLICT(ip) DO UPDATE SET
                    country = excluded.country,
                    jail = excluded.jail,
                    reason = excluded.reason
                """,
                (
                    ip,
                    geo.country,
                    jail,
                    decision.reason,
                ),
            )

    return {
        "ip": ip,
        "jail": jail,
        "country": geo.country,
        "country_name": geo.country_name,
        "asn": geo.asn,
        "organization": geo.organization,
        "geo_provider": geo.provider,
        "offense": number,
        "duration_seconds": decision.duration_seconds,
        "permanent": decision.permanent,
        "trusted": decision.trusted,
        "mode": settings.mode,
        "reason": decision.reason,
    }

===== src/ande_ban/campaigns.py =====
"""Historical attack-campaign detection for ANDE-Ban.

Campaigns are currently grouped by organization and ASN. This module contains
only deterministic local analysis and does not perform external reputation
lookups or enforcement actions.
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
import hashlib
import sqlite3
from typing import Any


ACTIVE_WINDOW = timedelta(hours=24)
RECENT_WINDOW = timedelta(days=7)

THREAT_LOW_MAX = 24
THREAT_MODERATE_MAX = 49
THREAT_HIGH_MAX = 74


def _parse_timestamp(value: Any) -> datetime | None:
    """Parse an ANDE-Ban UTC timestamp safely."""

    if value is None:
        return None

    text = str(value).strip()
    if not text:
        return None

    try:
        parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError:
        return None

    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)

    return parsed.astimezone(timezone.utc)


def _campaign_status(
    last_seen: datetime | None,
    now: datetime,
) -> str:
    """Classify campaign recency."""

    if last_seen is None:
        return "unknown"

    age = now - last_seen

    if age <= ACTIVE_WINDOW:
        return "active"

    if age <= RECENT_WINDOW:
        return "recent"

    return "dormant"


def _campaign_id(organization: str, asn: int) -> str:
    """Build a stable, non-secret identifier for a campaign."""

    identity = f"{organization.strip().lower()}|{asn}"
    digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()
    return digest[:12]


def _bounded_score(value: int, maximum_value: int, points: int) -> int:
    """Convert an observed value into a bounded score component."""

    if value <= 0 or maximum_value <= 0 or points <= 0:
        return 0

    return min(
        points,
        round((value / maximum_value) * points),
    )


def _threat_level(score: int) -> str:
    """Return the display threat level for a confidence score."""

    if score <= THREAT_LOW_MAX:
        return "low"

    if score <= THREAT_MODERATE_MAX:
        return "moderate"

    if score <= THREAT_HIGH_MAX:
        return "high"

    return "critical"


def calculate_campaign_confidence(
    *,
    attacks: int,
    unique_ips: int,
    attacks_24h: int,
    targeted_jails: int,
    countries: int,
    duration_seconds: int,
) -> tuple[int, str]:
    """Calculate a deterministic campaign confidence score.

    The score reflects locally observed campaign breadth, activity,
    persistence, and targeting. It is not an external reputation score.
    """

    duration_days = duration_seconds / 86400 if duration_seconds > 0 else 0

    score = sum(
        (
            _bounded_score(attacks, 50, 25),
            _bounded_score(unique_ips, 10, 20),
            _bounded_score(attacks_24h, 20, 20),
            _bounded_score(targeted_jails, 5, 15),
            _bounded_score(countries, 5, 10),
            _bounded_score(round(duration_days), 30, 10),
        )
    )

    score = max(0, min(100, score))

    return score, _threat_level(score)


def get_campaigns(
    conn: sqlite3.Connection,
    *,
    limit: int = 20,
    now: datetime | None = None,
) -> list[dict[str, Any]]:
    """Return organization-and-ASN attack campaigns.

    A group qualifies when it contains either:

    * two or more unique attacking IP addresses; or
    * five or more historical ban events.

    No firewall or blacklist changes are made by this function.
    """

    if limit < 1:
        return []

    current_time = now or datetime.now(timezone.utc)
    current_time = current_time.astimezone(timezone.utc)
    activity_cutoff = current_time - ACTIVE_WINDOW

    rows = conn.execute(
        """
        SELECT
            organization,
            asn,
            COUNT(*) AS attacks,
            COUNT(DISTINCT ip) AS unique_ips,
            COUNT(DISTINCT country) AS countries,
            COUNT(DISTINCT jail) AS targeted_jails,
            MIN(occurred_at) AS first_seen,
            MAX(occurred_at) AS last_seen,
            SUM(
                CASE
                    WHEN julianday(occurred_at) >= julianday(?)
                    THEN 1
                    ELSE 0
                END
            ) AS attacks_24h
        FROM events
        WHERE event_type = 'ban'
          AND organization IS NOT NULL
          AND TRIM(organization) != ''
          AND asn IS NOT NULL
        GROUP BY organization, asn
        HAVING COUNT(DISTINCT ip) >= 2
            OR COUNT(*) >= 5
        ORDER BY
            attacks_24h DESC,
            attacks DESC,
            unique_ips DESC,
            organization,
            asn
        LIMIT ?
        """,
        (
            activity_cutoff.isoformat(),
            limit,
        ),
    ).fetchall()

    columns = [description[0] for description in conn.execute(
        """
        SELECT
            organization,
            asn,
            COUNT(*) AS attacks,
            COUNT(DISTINCT ip) AS unique_ips,
            COUNT(DISTINCT country) AS countries,
            COUNT(DISTINCT jail) AS targeted_jails,
            MIN(occurred_at) AS first_seen,
            MAX(occurred_at) AS last_seen,
            0 AS attacks_24h
        FROM events
        WHERE 0
        GROUP BY organization, asn
        """
    ).description]

    campaigns: list[dict[str, Any]] = []

    for raw_row in rows:
        row = dict(zip(columns, raw_row))

        organization = str(row["organization"])
        asn = int(row["asn"])

        first_seen = _parse_timestamp(row.get("first_seen"))
        last_seen = _parse_timestamp(row.get("last_seen"))

        duration_seconds = 0
        if first_seen is not None and last_seen is not None:
            duration_seconds = max(
                0,
                int((last_seen - first_seen).total_seconds()),
            )

        attacks = int(row.get("attacks") or 0)
        unique_ips = int(row.get("unique_ips") or 0)
        countries = int(row.get("countries") or 0)
        targeted_jails = int(row.get("targeted_jails") or 0)
        attacks_24h = int(row.get("attacks_24h") or 0)

        confidence_score, threat_level = (
            calculate_campaign_confidence(
                attacks=attacks,
                unique_ips=unique_ips,
                attacks_24h=attacks_24h,
                targeted_jails=targeted_jails,
                countries=countries,
                duration_seconds=duration_seconds,
            )
        )

        campaigns.append(
            {
                "campaign_id": _campaign_id(organization, asn),
                "organization": organization,
                "asn": asn,
                "attacks": attacks,
                "unique_ips": unique_ips,
                "countries": countries,
                "targeted_jails": targeted_jails,
                "first_seen": row.get("first_seen"),
                "last_seen": row.get("last_seen"),
                "duration_seconds": duration_seconds,
                "attacks_24h": attacks_24h,
                "status": _campaign_status(last_seen, current_time),
                "confidence_score": confidence_score,
                "threat_level": threat_level,
            }
        )

    campaigns.sort(
        key=lambda campaign: (
            -campaign["confidence_score"],
            -campaign["attacks_24h"],
            -campaign["attacks"],
            campaign["organization"].lower(),
            campaign["asn"],
        )
    )

    return campaigns

===== src/ande_ban/analytics.py =====
from __future__ import annotations

import sqlite3
from datetime import datetime, timedelta, timezone
from typing import Any


ANALYTICS_INDEXES = (
    """
    CREATE INDEX IF NOT EXISTS idx_events_type_time
    ON events(event_type, occurred_at)
    """,
    """
    CREATE INDEX IF NOT EXISTS idx_events_ip_type
    ON events(ip, event_type)
    """,
    """
    CREATE INDEX IF NOT EXISTS idx_events_jail_type
    ON events(jail, event_type)
    """,
    """
    CREATE INDEX IF NOT EXISTS idx_events_country_type
    ON events(country, event_type)
    """,
    """
    CREATE INDEX IF NOT EXISTS idx_events_asn_type
    ON events(asn, event_type)
    """,
)


from .campaigns import get_campaigns

def ensure_analytics_indexes(conn: sqlite3.Connection) -> None:
    for statement in ANALYTICS_INDEXES:
        conn.execute(statement)


def _rows(
    conn: sqlite3.Connection,
    query: str,
    parameters: tuple[Any, ...] = (),
) -> list[dict[str, Any]]:
    return [
        dict(row)
        for row in conn.execute(query, parameters).fetchall()
    ]


def _threat_score(
    bans: int,
    jails: int,
    permanent: bool,
    recent: bool,
) -> int:
    score = 0

    # Repeated bans contribute up to 40 points.
    score += min(max(bans, 0) * 4, 40)

    # Activity across multiple services contributes up to 20 points.
    score += min(max(jails - 1, 0) * 10, 20)

    # Permanent blacklist placement is a strong signal.
    if permanent:
        score += 20

    # Activity during the last 24 hours raises current urgency.
    if recent:
        score += 20

    return min(score, 100)


def _threat_level(score: int) -> str:
    if score >= 80:
        return "critical"
    if score >= 60:
        return "high"
    if score >= 35:
        return "elevated"
    return "observed"


def _activity_24h(conn: sqlite3.Connection) -> list[dict[str, Any]]:
    raw = {
        row["hour"]: int(row["attacks"])
        for row in conn.execute(
            """
            SELECT
                strftime(
                    '%Y-%m-%dT%H:00:00Z',
                    occurred_at
                ) AS hour,
                COUNT(*) AS attacks
            FROM events
            WHERE event_type = 'ban'
              AND julianday(occurred_at)
                  >= julianday('now', '-24 hours')
            GROUP BY hour
            ORDER BY hour
            """
        )
        if row["hour"]
    }

    now = datetime.now(timezone.utc).replace(
        minute=0,
        second=0,
        microsecond=0,
    )

    result = []

    for offset in range(23, -1, -1):
        hour = now - timedelta(hours=offset)
        key = hour.strftime("%Y-%m-%dT%H:00:00Z")

        result.append(
            {
                "hour": key,
                "label": hour.strftime("%H:00"),
                "attacks": raw.get(key, 0),
            }
        )

    return result


def get_analytics(conn: sqlite3.Connection) -> dict[str, Any]:
    ensure_analytics_indexes(conn)

    historical_summary_row = conn.execute(
        """
        SELECT
            COUNT(*) AS ban_events,
            COUNT(DISTINCT ip) AS unique_ips,
            COUNT(DISTINCT asn) AS unique_asns,
            COUNT(DISTINCT organization) AS unique_organizations,
            COUNT(DISTINCT country) AS unique_countries
        FROM events
        WHERE event_type = 'ban'
        """
    ).fetchone()

    historical_summary = {
        "ban_events": historical_summary_row["ban_events"],
        "unique_ips": historical_summary_row["unique_ips"],
        "unique_asns": historical_summary_row["unique_asns"],
        "unique_organizations": historical_summary_row[
            "unique_organizations"
        ],
        "unique_countries": historical_summary_row["unique_countries"],
        "repeat_offenders": conn.execute(
            """
            SELECT COUNT(*)
            FROM (
                SELECT ip
                FROM events
                WHERE event_type = 'ban'
                GROUP BY ip
                HAVING COUNT(*) > 1
            )
            """
        ).fetchone()[0],
        "permanent_candidates": conn.execute(
            "SELECT COUNT(*) FROM permanent_blacklist"
        ).fetchone()[0],
    }

    top_countries = _rows(
        conn,
        """
        SELECT
            COALESCE(country, 'Unknown') AS country,
            COALESCE(
                MAX(NULLIF(country_name, '')),
                country,
                'Unknown'
            ) AS country_name,
            COUNT(*) AS attacks,
            COUNT(DISTINCT ip) AS unique_ips,
            COUNT(DISTINCT asn) AS unique_asns,
            COUNT(DISTINCT organization) AS unique_organizations,
            COUNT(DISTINCT jail) AS targeted_jails,
            MIN(occurred_at) AS first_seen,
            MAX(occurred_at) AS last_seen
        FROM events
        WHERE event_type = 'ban'
        GROUP BY COALESCE(country, 'Unknown')
        ORDER BY attacks DESC, unique_ips DESC, country_name
        LIMIT 10
        """,
    )

    top_asns = _rows(
        conn,
        """
        SELECT
            asn,
            COALESCE(
                MAX(NULLIF(organization, '')),
                'Unknown organization'
            ) AS organization,
            COUNT(*) AS attacks,
            COUNT(DISTINCT ip) AS unique_ips,
            COUNT(DISTINCT jail) AS targeted_jails,
            MIN(occurred_at) AS first_seen,
            MAX(occurred_at) AS last_seen
        FROM events
        WHERE event_type = 'ban'
          AND asn IS NOT NULL
        GROUP BY asn
        ORDER BY attacks DESC, unique_ips DESC, asn
        LIMIT 10
        """,
    )

    top_organizations = _rows(
        conn,
        """
        SELECT
            organization,
            COUNT(DISTINCT asn) AS asns,
            COUNT(*) AS attacks,
            COUNT(DISTINCT ip) AS unique_ips,
            COUNT(DISTINCT country) AS countries,
            COUNT(DISTINCT jail) AS jails,
            MIN(occurred_at) AS first_seen,
            MAX(occurred_at) AS last_seen
        FROM events
        WHERE event_type = 'ban'
          AND organization IS NOT NULL
          AND TRIM(organization) != ''
        GROUP BY organization
        ORDER BY attacks DESC, unique_ips DESC, organization
        LIMIT 10
        """,
    )

    campaigns = get_campaigns(conn)

    repeat_offenders = _rows(
        conn,
        """
        SELECT
            e.ip,
            COUNT(*) AS bans,
            COUNT(DISTINCT e.jail) AS jails,
            MIN(e.occurred_at) AS first_seen,
            MAX(e.occurred_at) AS last_seen,
            MAX(CASE WHEN pb.ip IS NOT NULL THEN 1 ELSE 0 END)
                AS permanent,
            MAX(CASE WHEN ab.ip IS NOT NULL THEN 1 ELSE 0 END)
                AS currently_active
        FROM events e
        LEFT JOIN permanent_blacklist pb
          ON pb.ip = e.ip
        LEFT JOIN active_bans ab
          ON ab.ip = e.ip
        WHERE e.event_type = 'ban'
        GROUP BY e.ip
        HAVING COUNT(*) > 1
        ORDER BY bans DESC, last_seen DESC
        LIMIT 20
        """,
    )

    now = datetime.now(timezone.utc)
    recent_threshold = now - timedelta(hours=24)

    for offender in repeat_offenders:
        last_seen_text = offender.get("last_seen")
        recent = False

        if last_seen_text:
            try:
                last_seen = datetime.fromisoformat(
                    str(last_seen_text).replace("Z", "+00:00")
                )
                recent = last_seen >= recent_threshold
            except ValueError:
                recent = False

        score = _threat_score(
            bans=int(offender.get("bans") or 0),
            jails=int(offender.get("jails") or 0),
            permanent=bool(offender.get("permanent")),
            recent=recent,
        )

        offender["threat_score"] = score
        offender["threat_level"] = _threat_level(score)
        offender["recent_24h"] = recent

    repeat_offenders.sort(
        key=lambda item: (
            int(item["threat_score"]),
            int(item["bans"]),
        ),
        reverse=True,
    )

    top_jails = _rows(
        conn,
        """
        SELECT
            jail,
            COUNT(*) AS bans,
            COUNT(DISTINCT ip) AS unique_ips,
            MAX(occurred_at) AS last_attack
        FROM events
        WHERE event_type = 'ban'
        GROUP BY jail
        ORDER BY bans DESC, jail
        """,
    )

    activity = _activity_24h(conn)
    attacks_24h = sum(int(item["attacks"]) for item in activity)
    peak = max(activity, key=lambda item: int(item["attacks"]))

    repeat_count = conn.execute(
        """
        SELECT COUNT(*)
        FROM (
            SELECT ip
            FROM events
            WHERE event_type = 'ban'
            GROUP BY ip
            HAVING COUNT(*) > 1
        )
        """
    ).fetchone()[0]

    active_asns = conn.execute(
        """
        SELECT COUNT(DISTINCT asn)
        FROM active_bans
        WHERE asn IS NOT NULL
        """
    ).fetchone()[0]

    return {
        "generated_at": now.isoformat().replace("+00:00", "Z"),
        "summary": {
            "attacks_24h": attacks_24h,
            "repeat_offenders": repeat_count,
            "active_asns": active_asns,
            "peak_hour": peak["hour"],
            "peak_hour_label": peak["label"],
            "peak_hour_attacks": peak["attacks"],
        },
        "historical_summary": historical_summary,
        "top_countries": top_countries,
        "top_asns": top_asns,
        "top_organizations": top_organizations,
        "campaigns": campaigns,
        "repeat_offenders": repeat_offenders,
        "top_jails": top_jails,
        "activity_24h": activity,
        "analytics_scope": "historical_events",
    }

ANALYTICS_RANGES = {
    "24h": timedelta(hours=24),
    "7d": timedelta(days=7),
    "30d": timedelta(days=30),
}


def get_range_analytics(
    conn: sqlite3.Connection,
    range_key: str = "24h",
    *,
    limit: int = 10,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Return bounded analytics for a supported rolling time range."""
    if range_key not in ANALYTICS_RANGES:
        raise ValueError(f"unsupported analytics range: {range_key}")

    ensure_analytics_indexes(conn)
    limit = min(max(int(limit), 1), 25)
    generated = now or datetime.now(timezone.utc)
    since = generated - ANALYTICS_RANGES[range_key]
    since_text = since.isoformat().replace("+00:00", "Z")

    summary_row = conn.execute(
        """
        SELECT COUNT(*) AS bans,
               COUNT(DISTINCT ip) AS unique_ips,
               COUNT(DISTINCT country) AS countries,
               COUNT(DISTINCT asn) AS asns,
               COUNT(DISTINCT jail) AS jails
          FROM events
         WHERE event_type='ban' AND occurred_at >= ?
        """,
        (since_text,),
    ).fetchone()

    def grouped(dimension: str, label_sql: str) -> list[dict[str, Any]]:
        allowed = {"ip", "country", "asn", "jail"}
        if dimension not in allowed:
            raise ValueError("invalid analytics dimension")
        return _rows(
            conn,
            f"""
            SELECT {dimension} AS key,
                   {label_sql} AS label,
                   COUNT(*) AS attacks,
                   COUNT(DISTINCT ip) AS unique_ips,
                   COUNT(DISTINCT jail) AS jails,
                   MAX(occurred_at) AS last_seen
              FROM events
             WHERE event_type='ban'
               AND occurred_at >= ?
               AND {dimension} IS NOT NULL
             GROUP BY {dimension}
             ORDER BY attacks DESC, last_seen DESC
             LIMIT ?
            """,
            (since_text, limit),
        )

    top_ips = grouped("ip", "ip")
    top_countries = grouped(
        "country",
        "COALESCE(MAX(NULLIF(country_name,'')), country, 'Unknown')",
    )
    top_asns = grouped(
        "asn",
        "CASE WHEN MAX(NULLIF(organization,'')) IS NULL "
        "THEN 'AS' || asn ELSE 'AS' || asn || ' — ' || "
        "MAX(NULLIF(organization,'')) END",
    )
    top_jails = grouped("jail", "jail")

    bucket_format = "%Y-%m-%dT%H:00:00Z" if range_key == "24h" else "%Y-%m-%dT00:00:00Z"
    bucket_label = "%H:00" if range_key == "24h" else "%b %d"
    raw_activity = {
        row["bucket"]: int(row["attacks"])
        for row in conn.execute(
            f"""
            SELECT strftime('{bucket_format}', occurred_at) AS bucket,
                   COUNT(*) AS attacks
              FROM events
             WHERE event_type='ban' AND occurred_at >= ?
             GROUP BY bucket
             ORDER BY bucket
            """,
            (since_text,),
        )
        if row["bucket"]
    }
    points = 24 if range_key == "24h" else (7 if range_key == "7d" else 30)
    step = timedelta(hours=1) if range_key == "24h" else timedelta(days=1)
    anchor = generated.replace(
        minute=0, second=0, microsecond=0
    ) if range_key == "24h" else generated.replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    activity = []
    for offset in range(points - 1, -1, -1):
        point = anchor - (step * offset)
        key = point.strftime(bucket_format)
        activity.append({
            "bucket": key,
            "label": point.strftime(bucket_label),
            "attacks": raw_activity.get(key, 0),
        })

    sync_row = conn.execute(
        """
        SELECT COUNT(*) AS runs,
               SUM(CASE WHEN status='failure' THEN 1 ELSE 0 END) AS failures,
               AVG(duration_ms) AS average_duration_ms,
               MAX(duration_ms) AS maximum_duration_ms
          FROM sync_runs
         WHERE started_at >= ?
        """,
        (since_text,),
    ).fetchone()

    return {
        "generated_at": generated.isoformat().replace("+00:00", "Z"),
        "range": range_key,
        "since": since_text,
        "limit": limit,
        "summary": {key: int(summary_row[key] or 0) for key in summary_row.keys()},
        "activity": activity,
        "top_ips": top_ips,
        "top_countries": top_countries,
        "top_asns": top_asns,
        "top_jails": top_jails,
        "sync": {
            "runs": int(sync_row["runs"] or 0),
            "failures": int(sync_row["failures"] or 0),
            "average_duration_ms": round(sync_row["average_duration_ms"] or 0),
            "maximum_duration_ms": int(sync_row["maximum_duration_ms"] or 0),
        },
    }

===== src/ande_ban/sync.py =====
from __future__ import annotations

import re
import subprocess
from datetime import datetime, timezone
from time import perf_counter

from .config import Settings
from .database import connect
from .geoip import lookup_geo


def _run(*args: str) -> str:
    process = subprocess.run(
        ["/usr/bin/fail2ban-client", *args],
        capture_output=True,
        text=True,
        timeout=10,
        check=False,
    )
    if process.returncode != 0:
        raise RuntimeError((process.stderr or process.stdout).strip())
    return process.stdout.strip()


def _jails() -> list[str]:
    output = _run("status")
    for line in output.splitlines():
        if "Jail list:" in line:
            return [
                item.strip()
                for item in line.split("Jail list:", 1)[1].split(",")
                if item.strip()
            ]
    return []


def _status_number(status: str, label: str) -> int:
    match = re.search(
        rf"^.*?{re.escape(label)}:\s*(\d+)\s*$",
        status,
        flags=re.MULTILINE | re.IGNORECASE,
    )
    return int(match.group(1)) if match else 0


def _utc_now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def _store_sync_run(
    settings: Settings,
    *,
    started_at: str,
    finished_at: str,
    duration_ms: int,
    status: str,
    jail_count: int,
    active_bans: int,
    events_added: int,
    error: str | None = None,
) -> None:
    with connect(settings.database) as conn:
        conn.execute(
            """INSERT INTO sync_runs(
                   started_at, finished_at, duration_ms, status,
                   jail_count, active_bans, events_added, error)
               VALUES(?,?,?,?,?,?,?,?)""",
            (
                started_at,
                finished_at,
                duration_ms,
                status,
                jail_count,
                active_bans,
                events_added,
                error,
            ),
        )
        conn.execute(
            """DELETE FROM sync_runs
               WHERE id NOT IN (
                   SELECT id FROM sync_runs ORDER BY id DESC LIMIT ?
               )""",
            (settings.sync_run_retention,),
        )


def sync_fail2ban(settings: Settings) -> dict:
    started_at = _utc_now()
    started_clock = perf_counter()
    jail_count = 0
    active_bans = 0
    events_added = 0

    try:
        jails = _jails()
        jail_count = len(jails)
        seen: set[tuple[str, str]] = set()
        statistics: list[dict] = []

        with connect(settings.database) as conn:
            event_count_before = conn.execute(
                "SELECT COUNT(*) FROM events"
            ).fetchone()[0]

            for jail in jails:
                status = _run("status", jail)
                currently_failed = _status_number(status, "Currently failed")
                total_failed = _status_number(status, "Total failed")
                currently_banned = _status_number(status, "Currently banned")
                total_banned = _status_number(status, "Total banned")
                banned_ips = _run("get", jail, "banip").split()

                conn.execute(
                    """INSERT INTO jail_status(
                           jail, currently_failed, total_failed,
                           currently_banned, total_banned, last_sync)
                       VALUES(?,?,?,?,?,?)
                       ON CONFLICT(jail) DO UPDATE SET
                           currently_failed=excluded.currently_failed,
                           total_failed=excluded.total_failed,
                           currently_banned=excluded.currently_banned,
                           total_banned=excluded.total_banned,
                           last_sync=excluded.last_sync""",
                    (
                        jail,
                        currently_failed,
                        total_failed,
                        currently_banned,
                        total_banned,
                        started_at,
                    ),
                )

                statistics.append(
                    {
                        "jail": jail,
                        "currently_failed": currently_failed,
                        "total_failed": total_failed,
                        "currently_banned": currently_banned,
                        "total_banned": total_banned,
                        "banned_ips": banned_ips,
                    }
                )

                for ip in banned_ips:
                    seen.add((jail, ip))
                    geo = lookup_geo(
                        ip,
                        settings.geoip_database,
                        settings.geoip_asn_database,
                        settings.geoip_provider,
                    )
                    conn.execute(
                        """INSERT INTO active_bans(
                               jail, ip, country, country_name, asn,
                               organization, first_seen, last_seen, source)
                           VALUES(?,?,?,?,?,?,?,?, 'fail2ban-sync')
                           ON CONFLICT(jail,ip) DO UPDATE SET
                               country=COALESCE(excluded.country,active_bans.country),
                               country_name=COALESCE(excluded.country_name,active_bans.country_name),
                               asn=COALESCE(excluded.asn,active_bans.asn),
                               organization=COALESCE(excluded.organization,active_bans.organization),
                               last_seen=excluded.last_seen""",
                        (
                            jail,
                            ip,
                            geo.country,
                            geo.country_name,
                            geo.asn,
                            geo.organization,
                            started_at,
                            started_at,
                        ),
                    )

            rows = conn.execute(
                "SELECT jail,ip FROM active_bans WHERE source='fail2ban-sync'"
            ).fetchall()
            for row in rows:
                if (row["jail"], row["ip"]) not in seen:
                    conn.execute(
                        "DELETE FROM active_bans WHERE jail=? AND ip=?",
                        (row["jail"], row["ip"]),
                    )

            if jails:
                placeholders = ",".join("?" for _ in jails)
                conn.execute(
                    f"DELETE FROM jail_status WHERE jail NOT IN ({placeholders})",
                    jails,
                )
            else:
                conn.execute("DELETE FROM jail_status")

            conn.execute(
                """INSERT INTO metadata(key,value) VALUES('last_successful_sync',?)
                   ON CONFLICT(key) DO UPDATE SET value=excluded.value""",
                (started_at,),
            )
            event_count_after = conn.execute(
                "SELECT COUNT(*) FROM events"
            ).fetchone()[0]
            events_added = max(0, event_count_after - event_count_before)

        active_bans = len(seen)
        finished_at = _utc_now()
        duration_ms = max(0, round((perf_counter() - started_clock) * 1000))
        _store_sync_run(
            settings,
            started_at=started_at,
            finished_at=finished_at,
            duration_ms=duration_ms,
            status="success",
            jail_count=jail_count,
            active_bans=active_bans,
            events_added=events_added,
        )
        return {
            "status": "success",
            "started_at": started_at,
            "finished_at": finished_at,
            "duration_ms": duration_ms,
            "jails": statistics,
            "jail_count": jail_count,
            "active_bans": active_bans,
            "events_added": events_added,
        }
    except Exception as exc:
        finished_at = _utc_now()
        duration_ms = max(0, round((perf_counter() - started_clock) * 1000))
        try:
            _store_sync_run(
                settings,
                started_at=started_at,
                finished_at=finished_at,
                duration_ms=duration_ms,
                status="failure",
                jail_count=jail_count,
                active_bans=active_bans,
                events_added=events_added,
                error=str(exc)[:2000],
            )
        except Exception:
            pass
        raise

===== src/ande_ban/administration.py =====
from __future__ import annotations

import json
import os
import platform
import shutil
import sqlite3
import subprocess
import tarfile
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable

from . import __version__
from .config import Settings
from .database import connect
from .diagnostics import run_diagnostics
from .maintenance import validate_configuration, upgrade_check

_ALLOWED_SERVICES = {
    "web": "ande-ban-web.service",
    "sync": "ande-ban-sync.service",
    "sync-timer": "ande-ban-sync.timer",
    "geoip": "ande-ban-geoip-update.service",
    "geoip-timer": "ande-ban-geoip-update.timer",
}


def database_statistics(settings: Settings) -> dict[str, Any]:
    db = settings.database
    with connect(db) as conn:
        page_count = int(conn.execute("PRAGMA page_count").fetchone()[0])
        page_size = int(conn.execute("PRAGMA page_size").fetchone()[0])
        freelist = int(conn.execute("PRAGMA freelist_count").fetchone()[0])
        quick_check = str(conn.execute("PRAGMA quick_check").fetchone()[0])
        counts = {}
        for table in ("events", "active_bans", "jail_status", "sync_runs", "permanent_blacklist"):
            counts[table] = int(conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0])
        indexes = int(conn.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='index'").fetchone()[0])
    return {
        "path": str(db),
        "size_bytes": db.stat().st_size if db.exists() else 0,
        "page_count": page_count,
        "page_size": page_size,
        "freelist_pages": freelist,
        "estimated_free_bytes": freelist * page_size,
        "quick_check": quick_check,
        "indexes": indexes,
        "counts": counts,
    }


def maintain_database(settings: Settings, *, analyze: bool = True, vacuum: bool = False, force: bool = False) -> dict[str, Any]:
    if vacuum and not force:
        raise ValueError("VACUUM requires --force because it rewrites the database file")
    before = database_statistics(settings)
    with connect(settings.database) as conn:
        if analyze:
            conn.execute("ANALYZE")
        if vacuum:
            conn.execute("VACUUM")
        conn.commit()
    after = database_statistics(settings)
    return {"status": "ok", "analyzed": analyze, "vacuumed": vacuum, "before": before, "after": after}


def recent_logs(
    service: str,
    *,
    lines: int = 100,
    runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
) -> dict[str, Any]:
    if service not in _ALLOWED_SERVICES:
        raise ValueError(f"service must be one of: {', '.join(sorted(_ALLOWED_SERVICES))}")
    lines = min(max(int(lines), 1), 1000)
    unit = _ALLOWED_SERVICES[service]
    command = ["journalctl", "--no-pager", "--output=short-iso", "-n", str(lines), "-u", unit]
    result = runner(command, capture_output=True, text=True, timeout=10, check=False)
    return {
        "service": service,
        "unit": unit,
        "lines_requested": lines,
        "returncode": result.returncode,
        "output": result.stdout.rstrip(),
        "error": result.stderr.rstrip() or None,
    }


def list_backups(directory: str | Path, *, limit: int = 50) -> dict[str, Any]:
    root = Path(directory).expanduser().resolve()
    limit = min(max(int(limit), 1), 250)
    items = []
    if root.is_dir():
        candidates = sorted(root.glob("ande-ban*.tar.gz"), key=lambda p: p.stat().st_mtime, reverse=True)
        for path in candidates[:limit]:
            stat = path.stat()
            items.append({
                "name": path.name,
                "path": str(path),
                "size_bytes": stat.st_size,
                "modified_at": datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat().replace("+00:00", "Z"),
                "mode": oct(stat.st_mode & 0o777),
            })
    return {"directory": str(root), "backups": items}


def _redacted_config(config_path: Path) -> str:
    sensitive = ("password", "secret", "token", "api_key", "apikey", "license")
    output = []
    for line in config_path.read_text(errors="replace").splitlines():
        stripped = line.strip().lower()
        if "=" in line and any(key in stripped.split("=", 1)[0] for key in sensitive):
            output.append(f"{line.split('=', 1)[0]}= <redacted>")
        else:
            output.append(line)
    return "\n".join(output) + "\n"


def create_diagnostic_bundle(
    settings: Settings,
    config_path: str | Path,
    output: str | Path | None = None,
    *,
    log_runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
) -> dict[str, Any]:
    config_path = Path(config_path)
    if output is None:
        stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        output = Path.cwd() / f"ande-ban-diagnostics-{stamp}.tar.gz"
    output = Path(output).expanduser().resolve()
    output.parent.mkdir(parents=True, exist_ok=True)
    created_at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
    with tempfile.TemporaryDirectory(prefix="ande-ban-diagnostics-") as td:
        root = Path(td) / "ande-ban-diagnostics"
        root.mkdir()
        reports = {
            "manifest.json": {
                "format_version": 1,
                "created_at": created_at,
                "ande_ban_version": __version__,
                "python_version": platform.python_version(),
                "platform": platform.platform(),
            },
            "doctor.json": run_diagnostics(settings, config_path),
            "configuration-check.json": validate_configuration(config_path),
            "upgrade-check.json": upgrade_check(settings, config_path),
            "database-statistics.json": database_statistics(settings),
        }
        for filename, payload in reports.items():
            (root / filename).write_text(json.dumps(payload, indent=2) + "\n")
        if config_path.is_file():
            (root / "ande-ban.redacted.ini").write_text(_redacted_config(config_path))
        logs_dir = root / "logs"
        logs_dir.mkdir()
        for key in ("web", "sync", "sync-timer", "geoip-timer"):
            try:
                report = recent_logs(key, lines=200, runner=log_runner)
            except Exception as exc:
                report = {"service": key, "error": str(exc), "output": ""}
            (logs_dir / f"{key}.json").write_text(json.dumps(report, indent=2) + "\n")
        with tarfile.open(output, "w:gz") as tf:
            tf.add(root, arcname=root.name)
    os.chmod(output, 0o600)
    return {"status": "ok", "bundle": str(output), "created_at": created_at}

===== src/ande_ban/maintenance.py =====
from __future__ import annotations

import hashlib
import json
import os
import shutil
import sqlite3
import tarfile
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from .config import Settings, load_settings
from .database import connect

FORMAT_VERSION = 1


def validate_configuration(path: str | Path) -> dict[str, Any]:
    config_path = Path(path)
    checks: list[dict[str, str]] = []
    if not config_path.is_file():
        return {"status": "failed", "checks": [{"name": "configuration", "status": "fail", "message": f"not found: {config_path}"}]}
    try:
        settings = load_settings(config_path)
        checks.append({"name": "syntax", "status": "ok", "message": "configuration parsed successfully"})
    except Exception as exc:
        return {"status": "failed", "checks": [{"name": "syntax", "status": "fail", "message": str(exc)}]}
    if not (1 <= settings.listen_port <= 65535):
        checks.append({"name": "listen_port", "status": "fail", "message": "must be between 1 and 65535"})
    else:
        checks.append({"name": "listen_port", "status": "ok", "message": str(settings.listen_port)})
    if settings.listen_host not in {"127.0.0.1", "::1", "localhost"}:
        checks.append({"name": "listen_host", "status": "warn", "message": f"dashboard is exposed on {settings.listen_host}; use a trusted reverse proxy and authentication"})
    else:
        checks.append({"name": "listen_host", "status": "ok", "message": settings.listen_host})
    checks.append({"name": "mode", "status": "ok", "message": settings.mode})
    status = "failed" if any(c["status"] == "fail" for c in checks) else ("warning" if any(c["status"] == "warn" for c in checks) else "ok")
    return {"status": status, "checks": checks}


def _sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def create_backup(settings: Settings, config_path: str | Path, output: str | Path | None = None) -> dict[str, Any]:
    config_path = Path(config_path)
    if output is None:
        stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        output = Path.cwd() / f"ande-ban-backup-{stamp}.tar.gz"
    output = Path(output).resolve()
    output.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix="ande-ban-backup-") as td:
        root = Path(td) / "ande-ban-backup"
        root.mkdir()
        db_copy = root / "events.db"
        with connect(settings.database) as source, sqlite3.connect(db_copy) as target:
            source.backup(target)
        shutil.copy2(config_path, root / "ande-ban.ini")
        manifest = {
            "format_version": FORMAT_VERSION,
            "created_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
            "database_sha256": _sha256(db_copy),
            "configuration_sha256": _sha256(root / "ande-ban.ini"),
        }
        (root / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n")
        with tarfile.open(output, "w:gz") as tf:
            tf.add(root, arcname="ande-ban-backup")
    os.chmod(output, 0o600)
    return {"status": "ok", "backup": str(output), "manifest": manifest}


def _safe_members(tf: tarfile.TarFile) -> None:
    expected = {"ande-ban-backup", "ande-ban-backup/events.db", "ande-ban-backup/ande-ban.ini", "ande-ban-backup/manifest.json"}
    names = {m.name.rstrip("/") for m in tf.getmembers()}
    if not expected.issubset(names):
        raise ValueError("backup archive is missing required files")
    for member in tf.getmembers():
        p = Path(member.name)
        if p.is_absolute() or ".." in p.parts or member.issym() or member.islnk():
            raise ValueError("unsafe backup archive member")


def inspect_backup(archive: str | Path) -> dict[str, Any]:
    archive = Path(archive)
    with tempfile.TemporaryDirectory(prefix="ande-ban-inspect-") as td, tarfile.open(archive, "r:gz") as tf:
        _safe_members(tf)
        tf.extractall(td, filter="data")
        root = Path(td) / "ande-ban-backup"
        manifest = json.loads((root / "manifest.json").read_text())
        if manifest.get("format_version") != FORMAT_VERSION:
            raise ValueError("unsupported backup format")
        if _sha256(root / "events.db") != manifest.get("database_sha256"):
            raise ValueError("database checksum mismatch")
        if _sha256(root / "ande-ban.ini") != manifest.get("configuration_sha256"):
            raise ValueError("configuration checksum mismatch")
        with sqlite3.connect(root / "events.db") as conn:
            integrity = conn.execute("PRAGMA quick_check").fetchone()[0]
            if integrity != "ok":
                raise ValueError(f"database integrity check failed: {integrity}")
            events = conn.execute("SELECT COUNT(*) FROM events").fetchone()[0]
        return {"status": "ok", "archive": str(archive.resolve()), "events": events, "manifest": manifest}


def restore_backup(settings: Settings, config_path: str | Path, archive: str | Path, force: bool = False) -> dict[str, Any]:
    if not force:
        raise ValueError("restore requires --force")
    inspection = inspect_backup(archive)
    safety = create_backup(settings, config_path, Path(archive).resolve().parent / "pre-restore-safety-backup.tar.gz")
    with tempfile.TemporaryDirectory(prefix="ande-ban-restore-") as td, tarfile.open(archive, "r:gz") as tf:
        _safe_members(tf)
        tf.extractall(td, filter="data")
        root = Path(td) / "ande-ban-backup"
        settings.database.parent.mkdir(parents=True, exist_ok=True)
        tmp_db = settings.database.with_suffix(settings.database.suffix + ".restore")
        shutil.copy2(root / "events.db", tmp_db)
        os.chmod(tmp_db, 0o660)
        for suffix in ("-wal", "-shm"):
            Path(str(settings.database) + suffix).unlink(missing_ok=True)
        os.replace(tmp_db, settings.database)
        shutil.copy2(root / "ande-ban.ini", config_path)
        os.chmod(config_path, 0o640)
    return {"status": "ok", "restored": inspection, "safety_backup": safety["backup"]}


def upgrade_check(settings: Settings, config_path: str | Path) -> dict[str, Any]:
    config = validate_configuration(config_path)
    checks = list(config["checks"])
    try:
        with connect(settings.database) as conn:
            integrity = conn.execute("PRAGMA quick_check").fetchone()[0]
            sync_runs = conn.execute("SELECT COUNT(*) FROM sync_runs").fetchone()[0]
        checks.append({"name": "database migration", "status": "ok" if integrity == "ok" else "fail", "message": f"quick_check={integrity}; sync_runs={sync_runs}"})
    except Exception as exc:
        checks.append({"name": "database migration", "status": "fail", "message": str(exc)})
    try:
        test_dir = settings.database.parent
        writable = os.access(test_dir, os.W_OK)
        checks.append({"name": "runtime directory", "status": "ok" if writable else "fail", "message": f"{test_dir} {'writable' if writable else 'not writable'}"})
    except Exception as exc:
        checks.append({"name": "runtime directory", "status": "fail", "message": str(exc)})
    status = "failed" if any(c["status"] == "fail" for c in checks) else ("warning" if any(c["status"] == "warn" for c in checks) else "ok")
    return {"status": status, "checks": checks}

===== src/ande_ban/diagnostics.py =====
from __future__ import annotations

import os
import shutil
import sqlite3
import stat
import subprocess
from datetime import datetime, timezone
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Callable

from .config import Settings


@dataclass(frozen=True)
class DiagnosticCheck:
    name: str
    status: str
    message: str
    required: bool = True

    @property
    def ok(self) -> bool:
        return self.status == "ok"


Runner = Callable[..., subprocess.CompletedProcess[str]]
Finder = Callable[[str], str | None]


def _mode(path: Path) -> str:
    return oct(stat.S_IMODE(path.stat().st_mode))


def _path_check(name: str, path: Path, *, required: bool = True) -> DiagnosticCheck:
    if not path.exists():
        status = "fail" if required else "warn"
        return DiagnosticCheck(name, status, f"not found: {path}", required)
    if not os.access(path, os.R_OK):
        status = "fail" if required else "warn"
        return DiagnosticCheck(name, status, f"not readable: {path}", required)
    return DiagnosticCheck(name, "ok", f"{path} (mode {_mode(path)})", required)


def _command_check(
    name: str,
    command: list[str],
    *,
    runner: Runner,
    required: bool = True,
    finder: Finder = shutil.which,
) -> DiagnosticCheck:
    executable = finder(command[0])
    if executable is None:
        status = "fail" if required else "warn"
        return DiagnosticCheck(name, status, f"command not found: {command[0]}", required)
    try:
        result = runner(command, capture_output=True, text=True, timeout=10, check=False)
    except (OSError, subprocess.SubprocessError) as exc:
        status = "fail" if required else "warn"
        return DiagnosticCheck(name, status, str(exc), required)
    output = (result.stdout or result.stderr).strip()
    if result.returncode == 0:
        return DiagnosticCheck(name, "ok", output or "available", required)
    status = "fail" if required else "warn"
    return DiagnosticCheck(name, status, output or f"exit status {result.returncode}", required)



def _parse_utc(value: str) -> datetime:
    normalized = value.strip().replace("Z", "+00:00")
    parsed = datetime.fromisoformat(normalized)
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def _sync_freshness_check(
    database: Path,
    stale_after_seconds: int,
    *,
    now: datetime | None = None,
) -> DiagnosticCheck:
    current = now or datetime.now(timezone.utc)
    try:
        uri = f"file:{database.resolve()}?mode=ro"
        with sqlite3.connect(uri, uri=True, timeout=5) as conn:
            row = conn.execute(
                "SELECT value FROM metadata WHERE key='last_successful_sync'"
            ).fetchone()
        if not row or not row[0]:
            return DiagnosticCheck(
                "synchronization freshness",
                "warn",
                "no successful synchronization has been recorded",
                False,
            )
        synced = _parse_utc(str(row[0]))
        age = max(0, int((current - synced).total_seconds()))
        if age > stale_after_seconds:
            return DiagnosticCheck(
                "synchronization freshness",
                "warn",
                f"stale: last sync {age}s ago (threshold {stale_after_seconds}s)",
                False,
            )
        return DiagnosticCheck(
            "synchronization freshness",
            "ok",
            f"last sync {age}s ago",
            False,
        )
    except (sqlite3.Error, OSError, ValueError) as exc:
        return DiagnosticCheck(
            "synchronization freshness", "warn", str(exc), False
        )

def run_diagnostics(
    settings: Settings,
    config_path: str | Path,
    *,
    runner: Runner = subprocess.run,
    finder: Finder = shutil.which,
    now: datetime | None = None,
) -> dict[str, object]:
    checks: list[DiagnosticCheck] = []
    config = Path(config_path)
    checks.append(_path_check("configuration", config))

    database = settings.database
    if database.exists():
        checks.append(_path_check("database file", database))
        try:
            uri = f"file:{database.resolve()}?mode=ro"
            with sqlite3.connect(uri, uri=True, timeout=5) as conn:
                integrity = conn.execute("PRAGMA quick_check").fetchone()[0]
                event_count = conn.execute("SELECT COUNT(*) FROM events").fetchone()[0]
            if integrity == "ok":
                checks.append(
                    DiagnosticCheck(
                        "database integrity",
                        "ok",
                        f"quick_check=ok; events={event_count}",
                    )
                )
            else:
                checks.append(DiagnosticCheck("database integrity", "fail", str(integrity)))
        except (sqlite3.Error, OSError) as exc:
            checks.append(DiagnosticCheck("database integrity", "fail", str(exc)))
    else:
        parent = database.parent
        writable = parent.exists() and os.access(parent, os.W_OK | os.X_OK)
        status = "ok" if writable else "fail"
        message = f"database will be created in {parent}" if writable else f"parent is not writable: {parent}"
        checks.append(DiagnosticCheck("database location", status, message))

    if database.exists():
        checks.append(
            _sync_freshness_check(
                database, getattr(settings, "sync_stale_after_seconds", 180), now=now
            )
        )

    checks.append(
        _path_check(
            "GeoIP country database",
            settings.geoip_database,
            required=getattr(settings, "geoip_provider", "auto") != "none",
        )
    )
    checks.append(
        _path_check(
            "GeoIP ASN database",
            settings.geoip_asn_database,
            required=False,
        )
    )

    checks.append(
        _command_check(
            "Fail2Ban",
            ["fail2ban-client", "ping"],
            runner=runner,
            finder=finder,
        )
    )
    checks.append(
        _command_check(
            "dashboard service",
            ["systemctl", "is-active", "ande-ban-web.service"],
            runner=runner,
            finder=finder,
        )
    )
    checks.append(
        _command_check(
            "synchronization timer",
            ["systemctl", "is-active", "ande-ban-sync.timer"],
            runner=runner,
            finder=finder,
        )
    )
    checks.append(
        _command_check(
            "GeoIP update timer",
            ["systemctl", "is-active", "ande-ban-geoip-update.timer"],
            runner=runner,
            required=False,
            finder=finder,
        )
    )

    failed = sum(1 for check in checks if check.status == "fail")
    warnings = sum(1 for check in checks if check.status == "warn")
    return {
        "status": "ok" if failed == 0 else "fail",
        "failed": failed,
        "warnings": warnings,
        "checks": [asdict(check) for check in checks],
    }

===== src/ande_ban/enrich_history.py =====
from __future__ import annotations

import argparse
from dataclasses import dataclass
from pathlib import Path

from .config import Settings, load_settings
from .database import connect
from .geoip import lookup_geo


@dataclass
class EnrichmentStats:
    examined: int = 0
    found: int = 0
    updated: int = 0
    unchanged: int = 0
    failed: int = 0


def enrich_history(
    settings: Settings,
    *,
    dry_run: bool = False,
    limit: int | None = None,
    batch_size: int = 500,
) -> EnrichmentStats:
    """
    Enrich historical event rows that are missing GeoIP or ASN metadata.

    Existing non-NULL values are preserved. The operation is safe to rerun.
    """

    if batch_size < 1:
        raise ValueError("batch_size must be at least 1")

    if limit is not None and limit < 1:
        raise ValueError("limit must be at least 1")

    stats = EnrichmentStats()
    last_id = 0

    while True:
        remaining = None if limit is None else limit - stats.examined
        if remaining is not None and remaining <= 0:
            break

        fetch_size = batch_size
        if remaining is not None:
            fetch_size = min(fetch_size, remaining)

        with connect(settings.database) as conn:
            rows = conn.execute(
                """
                SELECT
                    id,
                    ip,
                    country,
                    country_name,
                    asn,
                    organization,
                    geo_provider
                FROM events
                WHERE id > ?
                  AND geo_provider IS NULL
                ORDER BY id
                LIMIT ?
                """,
                (last_id, fetch_size),
            ).fetchall()

        if not rows:
            break

        for row in rows:
            last_id = row["id"]
            stats.examined += 1

            try:
                geo = lookup_geo(
                    ip=row["ip"],
                    country_database=settings.geoip_database,
                    asn_database=settings.geoip_asn_database,
                    provider=settings.geoip_provider,
                )
            except Exception as exc:
                stats.failed += 1
                print(
                    f"[error] event={row['id']} ip={row['ip']} "
                    f"{type(exc).__name__}: {exc}"
                )
                continue

            if geo.found:
                stats.found += 1

            values = {
                "country": row["country"] or geo.country,
                "country_name": row["country_name"] or geo.country_name,
                "asn": row["asn"] if row["asn"] is not None else geo.asn,
                "organization": row["organization"] or geo.organization,
                "geo_provider": row["geo_provider"] or geo.provider,
            }

            changed = any(
                values[column] != row[column]
                for column in values
            )

            if not changed:
                stats.unchanged += 1
                continue

            if dry_run:
                stats.updated += 1
                print(
                    f"[dry-run] event={row['id']} ip={row['ip']} "
                    f"country={values['country'] or '-'} "
                    f"asn={values['asn'] or '-'} "
                    f"organization={values['organization'] or '-'} "
                    f"provider={values['geo_provider'] or '-'}"
                )
                continue

            with connect(settings.database) as conn:
                conn.execute(
                    """
                    UPDATE events
                    SET
                        country = ?,
                        country_name = ?,
                        asn = ?,
                        organization = ?,
                        geo_provider = ?
                    WHERE id = ?
                    """,
                    (
                        values["country"],
                        values["country_name"],
                        values["asn"],
                        values["organization"],
                        values["geo_provider"],
                        row["id"],
                    ),
                )

            stats.updated += 1

    return stats


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Backfill GeoIP and ASN data for historical ANDE-Ban events."
    )
    parser.add_argument(
        "--config",
        default="/etc/ande-ban/ande-ban.ini",
        help="Path to the ANDE-Ban configuration file.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Show proposed changes without updating the database.",
    )
    parser.add_argument(
        "--limit",
        type=int,
        help="Maximum number of historical rows to examine.",
    )
    parser.add_argument(
        "--batch-size",
        type=int,
        default=500,
        help="Number of rows fetched per batch. Default: 500.",
    )
    return parser


def main() -> int:
    args = _build_parser().parse_args()
    settings = load_settings(Path(args.config))

    stats = enrich_history(
        settings,
        dry_run=args.dry_run,
        limit=args.limit,
        batch_size=args.batch_size,
    )

    print()
    print("Historical enrichment complete")
    print(f"  Examined:  {stats.examined}")
    print(f"  Geo found: {stats.found}")
    print(f"  Updated:   {stats.updated}")
    print(f"  Unchanged: {stats.unchanged}")
    print(f"  Failed:    {stats.failed}")
    print(f"  Dry run:   {'yes' if args.dry_run else 'no'}")

    return 1 if stats.failed else 0


if __name__ == "__main__":
    raise SystemExit(main())

===== src/ande_ban/webapp.py =====
from __future__ import annotations

import argparse
import json
import os
import shutil
import sqlite3
import subprocess
from pathlib import Path
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

from . import __version__
from .analytics import ANALYTICS_RANGES, get_analytics, get_range_analytics
from .administration import database_statistics, list_backups, recent_logs
from .api import error as api_error, success as api_success
from .metrics import collect_system_metrics
from .config import load_settings
from .database import connect
from .diagnostics import run_diagnostics

PACKAGE_DIR = Path(__file__).resolve().parent
TEMPLATE_DIR = PACKAGE_DIR / "templates"
STATIC_DIR = PACKAGE_DIR / "static"

HTML = (TEMPLATE_DIR / "index.html").read_bytes()
STYLE = (STATIC_DIR / "style.css").read_bytes()
SCRIPT = (STATIC_DIR / "app.js").read_bytes()

STATIC_RESPONSES = {
    "/": ("text/html; charset=utf-8", HTML),
    "/static/style.css": ("text/css; charset=utf-8", STYLE),
    "/static/app.js": (
        "application/javascript; charset=utf-8",
        SCRIPT,
    ),
}


def _service_state(name: str) -> str:
    try:
        p = subprocess.run(
            ["/usr/bin/systemctl", "is-active", name],
            capture_output=True,
            text=True,
            timeout=3,
            check=False,
        )
        return p.stdout.strip() or "unknown"
    except (OSError, subprocess.SubprocessError):
        return "unknown"


def handler_factory(
    settings,
    config_path="/etc/ande-ban/ande-ban.ini",
    *,
    diagnostics_runner=subprocess.run,
    diagnostics_finder=shutil.which,
):
    class Handler(BaseHTTPRequestHandler):
        def send_json(self, payload, status=200):
            data = json.dumps(payload).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Cache-Control", "no-store")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            if self.command != "HEAD":
                self.wfile.write(data)

        def send_api(self, data, status=200, *, metadata=None):
            self.send_json(api_success(data, metadata=metadata), status)

        def send_api_error(self, code, message, status=400, *, details=None):
            self.send_json(api_error(code, message, details=details), status)

        def do_HEAD(self):
            self.do_GET()

        def do_GET(self):
            parsed = urlparse(self.path)
            path = parsed.path
            static_response = STATIC_RESPONSES.get(path)
            if static_response:
                content_type, data = static_response
                self.send_response(200)
                self.send_header("Content-Type", content_type)
                self.send_header("Cache-Control", "no-store")
                self.send_header("Content-Length", str(len(data)))
                self.end_headers()
                if self.command != "HEAD":
                    self.wfile.write(data)
                return

            if path == "/api/v1/version":
                self.send_api({"version": __version__})
                return

            if path == "/api/version":
                self.send_json({"version": __version__})
                return

            if path == "/api/v1/admin/database":
                try:
                    self.send_api(database_statistics(settings))
                except (sqlite3.Error, OSError) as exc:
                    self.send_api_error("DATABASE_UNAVAILABLE", str(exc), 503)
                return

            if path == "/api/v1/admin/backups":
                self.send_api(list_backups("/var/backups/ande-ban"))
                return

            if path == "/api/v1/admin/logs":
                values = parse_qs(parsed.query)
                service = values.get("service", ["web"])[0]
                try:
                    lines = int(values.get("lines", ["100"])[0])
                    self.send_api(recent_logs(service, lines=lines, runner=diagnostics_runner))
                except ValueError as exc:
                    self.send_api_error("INVALID_ADMIN_REQUEST", str(exc), 400)
                except (OSError, subprocess.SubprocessError) as exc:
                    self.send_api_error("LOGS_UNAVAILABLE", str(exc), 503)
                return

            if path in {"/api/health", "/api/v1/health"}:
                report = run_diagnostics(
                    settings,
                    config_path,
                    runner=diagnostics_runner,
                    finder=diagnostics_finder,
                )
                try:
                    with sqlite3.connect(
                        f"file:{settings.database.resolve()}?mode=ro",
                        uri=True,
                        timeout=5,
                    ) as health_conn:
                        last = health_conn.execute(
                            "SELECT value FROM metadata WHERE key='last_successful_sync'"
                        ).fetchone()
                except (sqlite3.Error, OSError):
                    last = None
                try:
                    db_size = os.path.getsize(settings.database)
                except OSError:
                    db_size = 0
                check_map = {check["name"]: check for check in report["checks"]}
                report.update(
                    {
                        "version": __version__,
                        "fail2ban": check_map.get("Fail2Ban", {}).get("message", "unknown"),
                        "sync_timer": check_map.get("synchronization timer", {}).get("message", "unknown"),
                        "mode": settings.mode,
                        "last_successful_sync": last[0] if last else None,
                        "database": str(settings.database),
                        "database_size_bytes": db_size,
                        "geoip_country_ready": settings.geoip_database.exists(),
                        "geoip_asn_ready": settings.geoip_asn_database.exists(),
                    }
                )
                if path == "/api/v1/health":
                    self.send_api(report)
                else:
                    self.send_json(report)
                return

            with connect(settings.database) as conn:
                if path == "/api/v1/metrics":
                    self.send_api(collect_system_metrics(conn, settings))
                    return
                if path == "/api/v1/analytics/trends":
                    values = parse_qs(parsed.query)
                    range_key = values.get("range", ["24h"])[0]
                    try:
                        limit = int(values.get("limit", ["10"])[0])
                    except ValueError:
                        self.send_api_error("INVALID_LIMIT", "limit must be an integer.", 400)
                        return
                    if range_key not in ANALYTICS_RANGES:
                        self.send_api_error(
                            "INVALID_RANGE",
                            "range must be one of 24h, 7d, or 30d.",
                            400,
                        )
                        return
                    self.send_api(get_range_analytics(conn, range_key, limit=limit))
                    return
                legacy_path = "/api" + path.removeprefix("/api/v1") if path.startswith("/api/v1/") else path
                use_envelope = path.startswith("/api/v1/")
                if legacy_path == "/api/summary":
                    event = conn.execute(
                        """SELECT COUNT(*) n, COUNT(DISTINCT ip) u,
                                  COUNT(DISTINCT country) co
                           FROM events WHERE event_type='ban'"""
                    ).fetchone()
                    summary_payload = {
                        "active_bans": conn.execute("SELECT COUNT(*) FROM active_bans").fetchone()[0],
                        "ban_events": event["n"],
                        "unique_ips": event["u"],
                        "countries": conn.execute("SELECT COUNT(DISTINCT country) FROM active_bans WHERE country IS NOT NULL").fetchone()[0],
                        "permanent": conn.execute("SELECT COUNT(*) FROM permanent_blacklist").fetchone()[0],
                        "jails": conn.execute("SELECT COUNT(*) FROM jail_status").fetchone()[0],
                        "mode": settings.mode,
                    }
                    (self.send_api if use_envelope else self.send_json)(summary_payload)
                    return
                if legacy_path == "/api/analytics":
                    (self.send_api if use_envelope else self.send_json)(get_analytics(conn))
                    return
                if legacy_path == "/api/sync-runs":
                    values = parse_qs(parsed.query)
                    try:
                        limit = int(values.get("limit", ["50"])[0])
                    except ValueError:
                        limit = 50
                    limit = min(max(limit, 1), 250)
                    rows = [
                        dict(row)
                        for row in conn.execute(
                            "SELECT * FROM sync_runs ORDER BY id DESC LIMIT ?",
                            (limit,),
                        )
                    ]
                    aggregate = conn.execute(
                        """SELECT
                               COUNT(*) AS total_runs,
                               SUM(CASE WHEN status='failure' THEN 1 ELSE 0 END) AS failures,
                               AVG(duration_ms) AS average_duration_ms,
                               MAX(started_at) AS latest_started_at
                           FROM sync_runs"""
                    ).fetchone()
                    sync_payload = {
                        "summary": {
                            "total_runs": aggregate["total_runs"] or 0,
                            "failures": aggregate["failures"] or 0,
                            "average_duration_ms": round(aggregate["average_duration_ms"] or 0),
                            "latest_started_at": aggregate["latest_started_at"],
                        },
                        "runs": rows,
                    }
                    (self.send_api if use_envelope else self.send_json)(sync_payload)
                    return
                queries = {
                    "/api/jails": "SELECT * FROM jail_status ORDER BY jail",
                    "/api/active": "SELECT * FROM active_bans ORDER BY last_seen DESC LIMIT 1000",
                    "/api/countries": """SELECT country,country_name,COUNT(*) count
                                           FROM active_bans GROUP BY country,country_name
                                           ORDER BY count DESC""",
                    "/api/events": "SELECT * FROM events ORDER BY id DESC LIMIT 250",
                }
                query = queries.get(legacy_path)
                if query:
                    (self.send_api if use_envelope else self.send_json)([dict(row) for row in conn.execute(query)])
                    return
            if path.startswith("/api/v1/"):
                self.send_api_error("NOT_FOUND", "API endpoint not found.", 404)
            else:
                self.send_json({"error": "not found"}, 404)

        def log_message(self, *args):
            pass

    return Handler


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", default="/etc/ande-ban/ande-ban.ini")
    args = parser.parse_args()
    settings = load_settings(args.config)
    ThreadingHTTPServer(
        (settings.listen_host, settings.listen_port), handler_factory(settings, args.config)
    ).serve_forever()


if __name__ == "__main__":
    main()

===== src/ande_ban/cli.py =====
from __future__ import annotations

import argparse
import json
from .config import load_settings
from .administration import (
    create_diagnostic_bundle, database_statistics, list_backups,
    maintain_database, recent_logs,
)
from .database import connect
from .diagnostics import run_diagnostics
from .events import record_event
from .maintenance import create_backup, inspect_backup, restore_backup, upgrade_check, validate_configuration
from .sync import sync_fail2ban


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(prog="ande-ban")
    p.add_argument("--config", default="/etc/ande-ban/ande-ban.ini")
    sub = p.add_subparsers(dest="command", required=True)
    for name in ("ban", "unban"):
        q = sub.add_parser(name)
        q.add_argument("--ip", required=True)
        q.add_argument("--jail", required=True)
        q.add_argument("--failures", type=int, default=0)
    sub.add_parser("status")
    sync = sub.add_parser("sync")
    sync.add_argument("--json", action="store_true", dest="json_output")
    doctor = sub.add_parser("doctor", help="run read-only operational diagnostics")
    doctor.add_argument("--json", action="store_true", dest="json_output")
    config = sub.add_parser("config", help="configuration operations")
    config_sub = config.add_subparsers(dest="config_command", required=True)
    config_check = config_sub.add_parser("check", help="validate configuration")
    config_check.add_argument("--json", action="store_true", dest="json_output")
    backup = sub.add_parser("backup", help="create a consistent configuration and database backup")
    backup.add_argument("--output")
    backup.add_argument("--json", action="store_true", dest="json_output")
    inspect = sub.add_parser("backup-inspect", help="validate and inspect a backup archive")
    inspect.add_argument("archive")
    inspect.add_argument("--json", action="store_true", dest="json_output")
    restore = sub.add_parser("restore", help="restore configuration and database from a validated backup")
    restore.add_argument("archive")
    restore.add_argument("--force", action="store_true")
    restore.add_argument("--json", action="store_true", dest="json_output")
    upgrade = sub.add_parser("upgrade-check", help="run pre-upgrade compatibility checks")
    upgrade.add_argument("--json", action="store_true", dest="json_output")
    dbstats = sub.add_parser("database-stats", help="show SQLite database statistics")
    dbstats.add_argument("--json", action="store_true", dest="json_output")
    dbmaint = sub.add_parser("database-maintain", help="run guarded SQLite maintenance")
    dbmaint.add_argument("--no-analyze", action="store_true")
    dbmaint.add_argument("--vacuum", action="store_true")
    dbmaint.add_argument("--force", action="store_true")
    dbmaint.add_argument("--json", action="store_true", dest="json_output")
    logs = sub.add_parser("logs", help="show recent ANDE-Ban systemd logs")
    logs.add_argument("--service", default="web", choices=("web", "sync", "sync-timer", "geoip", "geoip-timer"))
    logs.add_argument("--lines", type=int, default=100)
    logs.add_argument("--json", action="store_true", dest="json_output")
    backups = sub.add_parser("backups", help="list ANDE-Ban backup archives")
    backups.add_argument("--directory", default="/var/backups/ande-ban")
    backups.add_argument("--limit", type=int, default=50)
    backups.add_argument("--json", action="store_true", dest="json_output")
    bundle = sub.add_parser("diagnostic-bundle", help="create a redacted support bundle")
    bundle.add_argument("--output")
    bundle.add_argument("--json", action="store_true", dest="json_output")
    h = sub.add_parser("history")
    h.add_argument("--ip")
    h.add_argument("--limit", type=int, default=50)
    return p


def main() -> None:
    args = build_parser().parse_args()
    if args.command == "config":
        report = validate_configuration(args.config)
        if args.json_output:
            print(json.dumps(report, indent=2))
        else:
            for check in report["checks"]:
                print(f"[{check['status'].upper()}] {check['name']}: {check['message']}")
            print(f"Summary: {report['status'].upper()}")
        raise SystemExit(0 if report["status"] == "ok" else 1)
    if args.command == "backup-inspect":
        report = inspect_backup(args.archive)
        print(json.dumps(report, indent=2) if args.json_output else f"Backup valid: {report['archive']} ({report['events']} events)")
        return
    settings = load_settings(args.config)
    if args.command in {"ban", "unban"}:
        print(json.dumps(record_event(settings, args.command, args.ip, args.jail, args.failures)))
        return
    if args.command == "sync":
        result = sync_fail2ban(settings)
        if args.json_output:
            print(json.dumps(result, indent=2))
        else:
            print(
                "Sync complete: "
                f"{result['active_bans']} active bans across "
                f"{result['jail_count']} jail"
                f"{'s' if result['jail_count'] != 1 else ''}; "
                f"{result['events_added']} events added; "
                f"{result['duration_ms'] / 1000:.2f} seconds"
            )
        return
    if args.command == "backup":
        report = create_backup(settings, args.config, args.output)
        print(json.dumps(report, indent=2) if args.json_output else f"Backup created: {report['backup']}")
        return
    if args.command == "restore":
        report = restore_backup(settings, args.config, args.archive, args.force)
        print(json.dumps(report, indent=2) if args.json_output else f"Restore complete; safety backup: {report['safety_backup']}")
        return
    if args.command == "upgrade-check":
        report = upgrade_check(settings, args.config)
        if args.json_output:
            print(json.dumps(report, indent=2))
        else:
            for check in report["checks"]:
                print(f"[{check['status'].upper()}] {check['name']}: {check['message']}")
            print(f"Summary: {report['status'].upper()}")
        raise SystemExit(0 if report["status"] == "ok" else 1)
    if args.command == "database-stats":
        report = database_statistics(settings)
        if args.json_output:
            print(json.dumps(report, indent=2))
        else:
            print(f"Database: {report['path']}")
            print(f"Integrity: {report['quick_check']}")
            print(f"Size: {report['size_bytes']} bytes; free estimate: {report['estimated_free_bytes']} bytes")
            for name, count in report["counts"].items():
                print(f"{name}: {count}")
        return
    if args.command == "database-maintain":
        report = maintain_database(settings, analyze=not args.no_analyze, vacuum=args.vacuum, force=args.force)
        print(json.dumps(report, indent=2) if args.json_output else f"Database maintenance complete: analyzed={report['analyzed']} vacuumed={report['vacuumed']}")
        return
    if args.command == "logs":
        report = recent_logs(args.service, lines=args.lines)
        print(json.dumps(report, indent=2) if args.json_output else report["output"])
        raise SystemExit(0 if report["returncode"] == 0 else 1)
    if args.command == "backups":
        report = list_backups(args.directory, limit=args.limit)
        if args.json_output:
            print(json.dumps(report, indent=2))
        else:
            print(f"Backup directory: {report['directory']}")
            for item in report["backups"]:
                print(f"{item['modified_at']} {item['size_bytes']:>10} {item['mode']} {item['name']}")
        return
    if args.command == "diagnostic-bundle":
        report = create_diagnostic_bundle(settings, args.config, args.output)
        print(json.dumps(report, indent=2) if args.json_output else f"Diagnostic bundle created: {report['bundle']}")
        return
    if args.command == "doctor":
        report = run_diagnostics(settings, args.config)
        if args.json_output:
            print(json.dumps(report, indent=2))
        else:
            for check in report["checks"]:
                marker = {"ok": "PASS", "warn": "WARN", "fail": "FAIL"}[check["status"]]
                print(f"[{marker}] {check['name']}: {check['message']}")
            print(
                f"Summary: {report['status'].upper()} "
                f"({report['failed']} failed, {report['warnings']} warnings)"
            )
        raise SystemExit(0 if report["status"] == "ok" else 1)
    with connect(settings.database) as conn:
        if args.command == "status":
            result = {
                "events": conn.execute("SELECT COUNT(*) FROM events").fetchone()[0],
                "bans": conn.execute("SELECT COUNT(*) FROM events WHERE event_type='ban'").fetchone()[0],
                "permanent": conn.execute("SELECT COUNT(*) FROM permanent_blacklist").fetchone()[0],
                "mode": settings.mode,
            }
            print(json.dumps(result, indent=2))
        else:
            if args.ip:
                rows = conn.execute("SELECT * FROM events WHERE ip=? ORDER BY id DESC LIMIT ?", (args.ip, args.limit))
            else:
                rows = conn.execute("SELECT * FROM events ORDER BY id DESC LIMIT ?", (args.limit,))
            print(json.dumps([dict(r) for r in rows], indent=2))

if __name__ == "__main__":
    main()
