server.py

import asyncio
import contextlib
import functools
import ssl
import struct
import os
import mysql.connector
import json
import hashlib
import re
import time
from datetime import datetime
from dotenv import load_dotenv
import pathlib

ENV_PATH = pathlib.Path(".env")
if ENV_PATH.exists():
    load_dotenv(dotenv_path=ENV_PATH)


debug = True


SERVER_IP = os.getenv('INGEST_BIND', '0.0.0.0')
SERVER_PORT = int(os.getenv('INGEST_PORT', '5001'))
BUFFER_SIZE = 16 * 1024

# Telemetry over TLS, on a second port beside the plaintext one.
#
# A second port rather than a flag on the first, because the two cannot share
# one: the existing protocol opens with a 4-byte big-endian length, so its
# first byte is always 0x00, and a TLS ClientHello opens with 0x16. They are
# trivially distinguishable - and distinguishing them means reading a byte
# that then cannot be handed back to the TLS layer, so one port can serve one
# of them and not both.
#
# Both listeners run by default, on purpose. Cutting straight to TLS would
# blackhole every agent built before this, and telemetry going quiet is the
# exact failure this codebase keeps finding: a host that stopped reporting and
# a host with nothing to report render identically. So the plaintext listener
# stays, counts who is still using it, and says so - and
# INGEST_TLS_REQUIRED=1 closes it once the fleet has moved.
INGEST_TLS_PORT = int(os.getenv('INGEST_TLS_PORT', '5011'))
INGEST_TLS_REQUIRED = os.getenv('INGEST_TLS_REQUIRED', '0').strip().lower() in ('1', 'true', 'yes', 'on')

#: Agents seen on the plaintext listener since this process started. Not a
#: counter of connections - a set of names, because "nine agents are still in
#: the clear" is the number an operator can act on, and 40,000 is not.
PLAINTEXT_AGENTS: set[str] = set()

DB_HOST = os.getenv('DB_HOST', '127.0.0.1')
DB_USER = os.getenv('DB_USER', 'root')
DB_PASSWORD = os.getenv('DB_PASSWORD', 'my-secret-pw')
DB_PORT = int(os.getenv('DB_PORT', '3306'))

#: Tables that hold the *current* state of a host rather than a history of
#: events. Each batch replaces the contents: the agent re-sends the whole
#: picture, and the previous picture is no longer true.
#:
#: "The agent re-sends the whole picture" is the load-bearing half of that
#: sentence, and it is a claim about the collector, not a preference. Snapshot
#: semantics are only correct for a collector that re-reports everything every
#: cycle; against one that reports deltas, DELETE-then-insert throws away
#: everything the delta did not mention.
#:
#: `critical_files` used to be here and is the reason the paragraph exists.
#: `check_permissions` writes a row only when a file's owner, group or mode
#: has actually changed - it is a change log, and the agent even keeps a
#: UNIQUE index on its `dup_fp` to guarantee that. So three permission changes
#: meant the server deleted the 1,200 rows it held and stored three, and after
#: a server-side database reset the table stayed empty for ever: there was
#: nothing new to send, and by design a sent row is never re-offered. Six of
#: the seven tables the agent deduplicates locally were already in
#: DEDUP_TABLES; this was the one that was not.
SNAPSHOT_TABLES = {
    "resource_usage", "disk_usage",
    "network_connections", "hardware_inventory", "docker_containers",
}

#: Tables where the same event must never be stored twice.
#:
#: Disjoint from SNAPSHOT_TABLES, and that is not a style preference - the two
#: mechanisms cancel each other out exactly. A snapshot table is emptied
#: before the batch is written; a deduplicated row is skipped when its
#: fingerprint has been seen before. Put a table in both and the sequence is:
#:
#:   batch 1  delete nothing, fingerprints are new, 50 rows written
#:   batch 2  delete all 50, every fingerprint already known, 0 rows written
#:   batch 3+ the same
#:
#: which leaves the table permanently empty after the second batch, while the
#: agent's log reads `network_connections sent (50 rows)` every cycle. The
#: console showed a host with no network connections and no hardware, the
#: agent showed it shipping both, and both were telling the truth.
#:
#: `_assert_ingest_modes_are_exclusive` below makes adding a table to both a
#: startup failure rather than a silently empty tab.
DEDUP_TABLES = {
    "critical_files", "portscan_result", "packages",
    "vulnerabilities_report", "siem_events", "events_alert", "soar_actions",
    "fim_data", "registry_logs", "process_events", "security_audit",
    "software_inventory", "network_inventory"
}


def _assert_ingest_modes_are_exclusive() -> None:
    """A table is a snapshot or it is deduplicated, never both.

    Checked at import so the failure is a server that will not start, rather
    than a tab that is empty for reasons nobody can see from either side.
    """
    both = SNAPSHOT_TABLES & DEDUP_TABLES
    if both:
        raise RuntimeError(
            f"{sorted(both)} are both snapshot and deduplicated tables. A "
            f"snapshot is emptied before each batch and a duplicate row is "
            f"skipped, so together they leave the table permanently empty "
            f"after the second batch while the agent reports sending rows."
        )


_assert_ingest_modes_are_exclusive()

ALLOWED_TABLES = {
    "critical_files", "portscan_result", "resource_usage",
    "disk_usage", "packages", "vulnerabilities_report",
    "siem_events", "events_alert", "soar_actions",
    "fim_data", "registry_logs", "network_connections", "process_events", "hardware_inventory", "security_audit", "docker_containers",
    "software_inventory", "network_inventory"
}

RABBITMQ_URL = os.getenv("RABBITMQ_URL", "amqp://guest:guest@rabbitmq/")

# RECENT_AI_TASKS / AI_DEDUP_WINDOW used to live here: a process-local dict
# with a 30-second window, wiped wholesale at 1000 entries. Deduplication
# therefore stopped working precisely when volume made it matter, divided by
# the Sanic worker count, and reset on every restart. Replaced by the
# database-backed counter in core/triage.py.

# asyncio holds only a *weak* reference to a running task, so a fire-and-forget
# `create_task` can be garbage collected before it finishes. On the ingest path
# that means telemetry silently not reaching the AI queue or the search index,
# with nothing anywhere to say it happened.
_background_tasks: set = set()


def _spawn(coro, *, label: str = "background"):
    """Fire-and-forget a coroutine while keeping it alive until it finishes."""
    task = asyncio.create_task(coro)
    _background_tasks.add(task)
    task.add_done_callback(_background_tasks.discard)

    def _report(t: asyncio.Task):
        if not t.cancelled() and t.exception() is not None:
            print(f"[!] {label} task failed: {t.exception()}", flush=True)

    task.add_done_callback(_report)
    return task


def connect_db(db_name):
    return mysql.connector.connect(
        host=DB_HOST,
        user=DB_USER,
        password=DB_PASSWORD,
        port=DB_PORT,
        database=db_name,
    )

def _sanitize_db_name(agent: str) -> str:
    safe = re.sub(r'[^A-Za-z0-9_]', '_', agent or 'agent')
    safe = safe.strip('_') or 'agent'
    return f"{safe}_db"

def create_agent_db_if_not_exists(agent: str) -> str:
    db_name = _sanitize_db_name(agent)
    conn = mysql.connector.connect(
        host=DB_HOST,
        user=DB_USER,
        password=DB_PASSWORD,
        port=DB_PORT
    )
    cursor = conn.cursor()
    try:
        cursor.execute(f"CREATE DATABASE IF NOT EXISTS `{db_name}`")
        conn.commit()
    finally:
        cursor.close()
        conn.close()
    return db_name

def _strip_sql_comments(sql: str) -> str:
    """Remove `--` comments, leaving string literals alone.

    Before splitting, not after, and that ordering is the whole point. This
    used to drop whole comment *lines* once the file had already been cut on
    `;` - so a semicolon inside a comment cut the statement it was part of:

        agent_name VARCHAR(255) NULL,  -- NULL means global; a value scopes …

    `soar_notification_templates` was therefore never created on any agent
    database. Both halves came back as syntax errors on every ingest, and the
    one that read `near 'a value scopes the row to one device'` was the
    comment being executed as SQL.

    Quotes are tracked because `--` can legitimately appear inside a literal,
    and stripping there would corrupt data rather than a comment.
    """
    out: list[str] = []
    quote = ""
    i = 0
    while i < len(sql):
        ch = sql[i]
        if quote:
            out.append(ch)
            if ch == "\\" and i + 1 < len(sql):        # escaped char
                out.append(sql[i + 1])
                i += 2
                continue
            if ch == quote:
                quote = ""
            i += 1
            continue
        if ch in ("'", '"', "`"):
            quote = ch
            out.append(ch)
            i += 1
            continue
        if ch == "-" and sql.startswith("--", i):
            newline = sql.find("\n", i)
            if newline == -1:
                break
            i = newline                               # keep the newline
            continue
        out.append(ch)
        i += 1
    return "".join(out)


def _split_sql_statements(sql: str) -> list[str]:
    """Split a schema file into statements, with comments removed first.

    Deliberately duplicated from the agent's `modules/db.py`: the agent ships
    as a standalone binary and cannot import from here. Both had the same
    trap, in different forms - see the note at the call site.
    """
    return [chunk.strip() for chunk in _strip_sql_comments(sql).split(';')
            if chunk.strip()]


AUTOMATION_STATUSES = ("pending", "active", "paused", "completed", "failed", "cancelled")


def _migrate_automation_status(cursor, db_name: str) -> None:
    """Widen `automations.status` to include 'cancelled', at most once.

    Kept out of init.sql and guarded, because that file is re-executed on
    every call to create_tables_if_not_exist. An unconditional
    `ALTER TABLE ... MODIFY` there took a metadata lock on `automations` each
    time; with one leaked connection holding an open transaction on the table,
    the ALTER queued behind it - and a *waiting* DDL in MySQL blocks every
    reader that arrives after it, not just writers.

    The result was total: agents polling /automations/pending got no answer
    at all, and a 500 ten seconds later when Sanic gave up. Killing the
    blocked ALTER did not help, because the next call re-issued it.

    Reading information_schema costs nothing and takes no lock, so the normal
    path - the column is already correct - never touches the table.
    """
    cursor.execute(
        "SELECT column_type FROM information_schema.columns "
        "WHERE table_schema = %s AND table_name = 'automations' "
        "AND column_name = 'status'",
        (db_name,),
    )
    row = cursor.fetchone()
    if not row:
        return                      # no table yet; init.sql creates it correct
    if "cancelled" in str(row[0]):
        return                      # already migrated

    values = ",".join(f"'{v}'" for v in AUTOMATION_STATUSES)
    try:
        # Never wait on a metadata lock. If the table is busy, skipping is
        # correct: the next start tries again, and blocking here would take
        # the polling endpoints down with it.
        cursor.execute("SET SESSION lock_wait_timeout = 3")
        cursor.execute(
            f"ALTER TABLE automations "
            f"MODIFY status ENUM({values}) NOT NULL DEFAULT 'pending'"
        )
        print(f"[+] {db_name}: automations.status widened to include 'cancelled'")
    except mysql.connector.Error as e:
        print(f"[!] {db_name}: could not widen automations.status ({e}); "
              f"will retry on the next start")


#: Columns holding Fernet ciphertext that shipped too narrow to hold it.
#:
#: `network_connections.local_addr` was VARCHAR(64). A token is over a hundred
#: characters whatever the plaintext, so every batch failed on its first row
#: with `1406 Data too long` - and because that table is a snapshot, the
#: DELETE had already run. The tab was empty for days while the agent's log
#: read `network_connections sent (50 rows)` every cycle.
#:
#: `tests/test_encrypted_column_widths.py` compares the two declarations that
#: have to agree, so the next one fails in CI rather than as an empty tab.
_WIDEN_FOR_CIPHERTEXT = (
    ("network_connections", "local_addr", "VARCHAR(255)"),
    ("network_connections", "remote_addr", "VARCHAR(255)"),
    ("hardware_inventory", "serial_number", "VARCHAR(255)"),
    ("events_alert", "score", "VARCHAR(255)"),
)


def _migrate_encrypted_widths(cursor, db_name: str) -> None:
    """Widen encrypted columns on databases that already exist.

    `CREATE TABLE IF NOT EXISTS` leaves an older table exactly as it was, so
    fixing the schema file only helps deployments made after today.
    """
    cursor.execute(
        "SELECT table_name, column_name, character_maximum_length "
        "FROM information_schema.columns WHERE table_schema = %s",
        (db_name,),
    )
    widths = {(str(t).lower(), str(c).lower()): m for t, c, m in cursor.fetchall()}

    for table, column, declaration in _WIDEN_FOR_CIPHERTEXT:
        current = widths.get((table.lower(), column.lower()), None)
        # Absent (no such column) or already wide enough: nothing to do.
        if current is None or int(current) >= 200:
            continue
        try:
            cursor.execute("SET SESSION lock_wait_timeout = 3")
            # Backticks directly, as everywhere else in this file. The names
            # come from the constant above, not from anything an agent sends.
            cursor.execute(
                f"ALTER TABLE `{table}` MODIFY COLUMN `{column}` {declaration} NULL")
            print(f"[+] {db_name}: {table}.{column} widened to hold ciphertext",
                  flush=True)
        except Exception as e:
            # Deliberately not just `mysql.connector.Error`. The first version
            # of this called a helper that does not exist in this module, so it
            # raised NameError - which that clause does not catch, on a path
            # only reached when there is work to do. The migration died
            # silently every time it mattered and never when it did not.
            print(f"[!] {db_name}: could not widen {table}.{column} "
                  f"({type(e).__name__}: {e}); encrypted rows will keep being "
                  f"rejected", flush=True)


def _migrate_agent_info(cursor, db_name: str) -> None:
    """Bring an older agent_info up to what the check-in INSERT writes.

    These were four ALTERs run unconditionally: `os_info` here, written as
    `ADD COLUMN IF NOT EXISTS` - which is MariaDB, so against MySQL it was a
    syntax error every time and the column never appeared on any deployment
    predating it - and hostname, mac_address and reported_ip inside
    `update_agent_info`, behind `except Exception: pass`, on *every agent
    check-in*. On a database that already had them, which is all of them,
    that is three failed statements and three metadata locks per check-in,
    at whatever rate the fleet reports in.
    """
    wanted = (
        ("reported_ip", "ADD COLUMN reported_ip VARCHAR(45) NULL"),
        ("os_info", "ADD COLUMN os_info VARCHAR(255) NULL"),
        ("hostname", "ADD COLUMN hostname VARCHAR(255) NULL"),
        ("mac_address", "ADD COLUMN mac_address VARCHAR(48) NULL"),
    )
    cursor.execute(
        "SELECT column_name FROM information_schema.columns "
        "WHERE table_schema = %s AND table_name = 'agent_info'",
        (db_name,),
    )
    present = {str(r[0]).lower() for r in cursor.fetchall()}
    if not present:
        return                      # no table yet; the CREATE above made it right

    pending = [ddl for name, ddl in wanted if name not in present]
    if not pending:
        return

    try:
        cursor.execute("SET SESSION lock_wait_timeout = 3")
        cursor.execute(f"ALTER TABLE agent_info {', '.join(pending)}")
        print(f"[+] {db_name}: agent_info gained {len(pending)} column(s)")
    except mysql.connector.Error as e:
        print(f"[!] {db_name}: could not migrate agent_info ({e}); "
              f"will retry on the next check-in")


def _migrate_soar_actions(cursor, db_name: str) -> None:
    """Make `expires_at` and `resolved_at` nullable, at most once.

    A pair of MODIFYs at the bottom of init.sql, described there as
    idempotent. They were - re-running them raised nothing - which is why it
    took so long to notice they were not free: MySQL takes a metadata lock for
    a MODIFY whether or not the type changes, and this file runs on every
    ingest. A waiting DDL blocks every reader that arrives behind it.
    """
    cursor.execute(
        "SELECT column_name, is_nullable FROM information_schema.columns "
        "WHERE table_schema = %s AND table_name = 'soar_actions' "
        "AND column_name IN ('expires_at', 'resolved_at')",
        (db_name,),
    )
    rows = cursor.fetchall()
    if not rows:
        return                      # no table yet; init.sql creates it correct
    if all(str(r[1]).upper() == "YES" for r in rows):
        return                      # already nullable

    try:
        cursor.execute("SET SESSION lock_wait_timeout = 3")
        cursor.execute(
            "ALTER TABLE soar_actions "
            "MODIFY COLUMN expires_at TIMESTAMP NULL, "
            "MODIFY COLUMN resolved_at TIMESTAMP NULL"
        )
        print(f"[+] {db_name}: soar_actions timestamps made nullable")
    except mysql.connector.Error as e:
        print(f"[!] {db_name}: could not relax soar_actions timestamps ({e}); "
              f"will retry on the next ingest")


def _migrate_siem_events(cursor, db_name: str) -> None:
    """Add `severity` and `techniques` to an older siem_events, at most once.

    These were four bare ALTERs at the bottom of init.sql. On every database
    that already had them - which is every database, since the CREATE above
    declares them - each one failed, and create_tables_if_not_exist printed

        [!] SQL Execution Error: 1060 (42S21): Duplicate column name 'severity'
        [!] SQL Execution Error: 1061 (42000): Duplicate key name 'idx_siem_sev'

    on every ingest. Two costs, and the quieter one is the worse: the noise
    buried real schema errors in the same log, and each attempt took a
    metadata lock on siem_events - the table every dashboard query reads -
    at ingest frequency, which is the shape of failure `_migrate_automation_status`
    documents above.

    information_schema answers both questions without touching the table, so
    a database that is already correct does nothing at all.
    """
    cursor.execute(
        "SELECT column_name FROM information_schema.columns "
        "WHERE table_schema = %s AND table_name = 'siem_events'",
        (db_name,),
    )
    present = {str(r[0]).lower() for r in cursor.fetchall()}
    if not present:
        return                      # no table yet; init.sql creates it correct

    cursor.execute(
        "SELECT index_name FROM information_schema.statistics "
        "WHERE table_schema = %s AND table_name = 'siem_events'",
        (db_name,),
    )
    indexes = {str(r[0]).lower() for r in cursor.fetchall()}

    pending = []
    for column, ddl in (
        ("severity", "ADD COLUMN severity VARCHAR(16) NULL"),
        ("techniques", "ADD COLUMN techniques VARCHAR(255) NULL"),
    ):
        if column not in present:
            pending.append(ddl)
    for index, ddl in (
        ("idx_siem_sev", "ADD KEY idx_siem_sev (severity)"),
        ("idx_siem_tech", "ADD KEY idx_siem_tech (techniques)"),
    ):
        if index not in indexes:
            pending.append(ddl)
    if not pending:
        return

    try:
        # One ALTER for all of it: MySQL rebuilds the table per statement, and
        # this is the largest table there is. Never wait on the lock - the
        # next ingest tries again, and blocking here stalls the readers.
        cursor.execute("SET SESSION lock_wait_timeout = 3")
        cursor.execute(f"ALTER TABLE siem_events {', '.join(pending)}")
        print(f"[+] {db_name}: siem_events migrated ({len(pending)} change(s))")
    except mysql.connector.Error as e:
        print(f"[!] {db_name}: could not migrate siem_events ({e}); "
              f"will retry on the next ingest")


# One engine for the whole fleet, built once. The per-host engine lives in the
# agent (core.correlation.default_engine); this one only sees shapes that
# span machines, which no agent can.
#
# Optional on purpose: a broken import here must not stop ingestion. Losing
# correlation costs detections, and refusing to accept telemetry costs all of
# them.
try:
    from core import telemetry_crypto
    from core.correlation import fleet_engine
    from core.sigma_loader import agent_event_fields
    _FLEET = fleet_engine()
except Exception as _corr_err:                        # pragma: no cover
    print(f"[Correlation] disabled: {_corr_err}", flush=True)
    _FLEET = None
    agent_event_fields = None


def correlate_across_hosts(agent, table, item, cursor):
    """Feed one stored event to the fleet engine; store whatever fired.

    A fired window becomes an `events_alert` row rather than a label on the
    event that completed it. It is a finding about the estate, and the fifth
    failed logon is no more interesting than the first.

    Returns the rows it wrote so the caller can put them through the AI queue.
    They have to go: `events_alert` normally reaches the workers because the
    ingest loop publishes what it inserts, and rows written from inside that
    loop are not items the loop iterates. Without this a correlation finding
    lands in the database, appears in the alerts view, and never becomes an AI
    insight - which was the state that made the model look blind to password
    spray when the platform had in fact detected it.

    Failures are contained: this runs inside the ingest transaction, and a
    correlation bug must cost a detection rather than the telemetry.
    """
    if _FLEET is None or table != "siem_events":
        return []

    written = []
    try:
        # Decrypted here and nowhere else: the row written to the database
        # and the message published to the broker both stay encrypted. A
        # correlation window counting ciphertext counts nothing, because
        # Fernet's random IV makes every encryption of the same text different.
        plain = telemetry_crypto.decrypt_item(table, item)
        fields = agent_event_fields(plain.get("message") or "")
        fields["agent"] = agent
        for found in _FLEET.observe(fields):
            # `dup_fp` comes from the detection's own sequence, not a hash of
            # the text. Two firings of the same window say the same words, so
            # a content hash would let the existing dedup swallow the second
            # as a repeat - the opposite of what a spray resuming after its
            # cooldown should do.
            row = {
                "source": f"Correlation/{found.rule}",
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "severity": found.severity,
                "categories": ",".join(found.techniques),
                "message": found.summary(),
                "dup_fp": hashlib.sha256(
                    f"{found.rule}|{found.group}|{found.seq}".encode()
                ).hexdigest(),
            }
            cursor.execute(
                "INSERT INTO events_alert "
                "(source, `timestamp`, severity, categories, message, dup_fp) "
                "VALUES (%(source)s, %(timestamp)s, %(severity)s, "
                "%(categories)s, %(message)s, %(dup_fp)s)", row)
            written.append(row)
            print(f"[Correlation] {found.title}: {found.detail}", flush=True)
    except Exception as e:
        print(f"[Correlation] failed for {agent}/{table}: {e}", flush=True)
    return written


#: Tables whose collector re-reports its whole list every cycle, so a row
#: removed here comes back within one interval - ten minutes for the inventory
#: pair, one package scan for `packages`.
#:
#: This is the entire reason the repair below is allowed to delete anything.
#: The change-log tables - `fim_data`, `critical_files`, `registry_logs`,
#: `soar_actions` - look superficially similar and must not be touched: the
#: agent marks a row sent exactly once and never offers it again, so a row
#: deleted there is gone for good.
_RE_REPORTED_TABLES = ("packages", "software_inventory", "network_inventory")

#: Named rather than dated, so it is obvious what it undoes.
_DEDUP_REPAIR = "collapse-rows-stored-without-a-usable-fingerprint"

#: Rows removed per table per ingest. Small enough that a pass is never the
#: reason a send times out - see the note in the repair about why that matters
#: more than finishing quickly.
_REPAIR_CHUNK = 5000


def _repair_rows_stored_without_a_fingerprint(cursor, db_name: str) -> None:
    """Clear out what accumulated while deduplication could not match anything.

    `compute_fingerprint` hashed the row as it arrived, which made every
    fingerprint unique by construction - see its docstring. Fixing it stops the
    pile growing; it does not shrink the pile, and nothing else ever will. On
    one host that pile was 249,980 `packages` rows for 167 packages and 161,157
    `software_inventory` rows for 313 programs, and the console faithfully
    listed all of them. That is what the user sees, so the fix is not finished
    without this.

    Two different situations, because two different things are wrong with the
    stored rows:

      - rows that carry a `dup_fp` (`packages` always did; the server was
        throwing it away) collapse onto it, keeping the lowest id so the
        earliest `created_at` survives - that is the "first seen" date, and it
        is worth more than the newest copy;

      - rows that carry none cannot be told apart at all. `software_inventory`
        and `network_inventory` had no fingerprint until this change, so there
        is nothing to group on and they are deleted outright. They are back,
        correctly, one row per program, at the next inventory cycle.

    Then `ingest_fingerprint` is rebuilt from the survivors. Its existing
    contents are row hashes that can never be produced again - a quarter of a
    million dead entries that would also let every surviving row be stored a
    second time on the next send.

    **A few thousand rows per call, not all of them.** `create_tables_if_not_exist`
    runs on the ingest path, so a DELETE across 400,000 rows holds up the send
    that triggered it. The agent would time out, retry, and start the repair
    again from the beginning - and if one pass can never finish inside one
    timeout, that is not a slow repair, it is a repair that never happens while
    every ingest pays for it. So each call removes a bounded chunk, the
    fingerprints are rebuilt only once a table has nothing left to remove, and
    the marker is written only when every table has converged. Ingest is
    frequent; 400,000 rows are gone within the hour.

    The guard is therefore a primary-key lookup, and the marker is written
    after the work rather than before: an interrupted repair resumes instead of
    recording itself as done.

    **Bounded to rows that existed when it started**, which is not a detail.
    An agent that has not been rebuilt still sends these tables with no
    `dup_fp`, so an unbounded "delete what has no fingerprint" deletes each
    fresh batch as it lands: the table stays empty, the console reads "this
    host has no software", and the repair never converges so it never stops.
    The high-water mark is taken once, per table, and stored - so old rows go,
    new ones stay whatever they look like, and upgrading the agent is what
    fixes the duplicates rather than a condition for seeing any data at all.
    """
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS ingest_migration (
            name       VARCHAR(64) NOT NULL PRIMARY KEY,
            value      BIGINT NULL,
            applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        ) ENGINE=InnoDB
    """)
    # An earlier build of this created the table without `value`.
    cursor.execute(
        "SELECT column_name FROM information_schema.columns "
        "WHERE table_schema = %s AND table_name = 'ingest_migration'",
        (db_name,),
    )
    if "value" not in {str(r[0]).lower() for r in cursor.fetchall()}:
        cursor.execute("ALTER TABLE ingest_migration ADD COLUMN value BIGINT NULL")

    cursor.execute("SELECT 1 FROM ingest_migration WHERE name = %s", (_DEDUP_REPAIR,))
    if cursor.fetchone():
        return

    cursor.execute(
        "SELECT table_name FROM information_schema.tables WHERE table_schema = %s",
        (db_name,),
    )
    present = {str(r[0]).lower() for r in cursor.fetchall()}

    outstanding = 0

    for table in _RE_REPORTED_TABLES:
        if table not in present:
            continue

        # The newest row when this table was first looked at. Recorded rather
        # than recomputed, because recomputing it each pass would sweep in
        # everything that arrived in between - which is the whole failure this
        # bound exists to prevent.
        mark_name = f"{_DEDUP_REPAIR}:{table}"[:64]
        cursor.execute("SELECT value FROM ingest_migration WHERE name = %s",
                       (mark_name,))
        row = cursor.fetchone()
        if row is None:
            cursor.execute(f"SELECT COALESCE(MAX(id), 0) FROM `{table}`")
            high_water = int((cursor.fetchone() or [0])[0] or 0)
            cursor.execute(
                "INSERT IGNORE INTO ingest_migration (name, value) VALUES (%s, %s)",
                (mark_name, high_water))
        else:
            high_water = int(row[0] or 0)

        # Keep the lowest id per fingerprint. The derived table is not
        # decoration: MySQL refuses a subquery that reads the table being
        # deleted from unless it is materialised through one.
        cursor.execute(
            f"DELETE FROM `{table}` "
            f"WHERE id <= %s AND dup_fp IS NOT NULL AND id NOT IN ("
            f"    SELECT keep_id FROM ("
            f"        SELECT MIN(id) AS keep_id FROM `{table}` "
            f"        WHERE dup_fp IS NOT NULL GROUP BY dup_fp"
            f"    ) AS keep"
            f") LIMIT {_REPAIR_CHUNK}",
            (high_water,)
        )
        collapsed = cursor.rowcount

        cursor.execute(
            f"DELETE FROM `{table}` WHERE id <= %s AND dup_fp IS NULL "
            f"LIMIT {_REPAIR_CHUNK}",
            (high_water,))
        dropped = cursor.rowcount

        if collapsed or dropped:
            outstanding += collapsed + dropped
            print(f"[+] {db_name}: {table} - {collapsed} duplicate rows "
                  f"collapsed, {dropped} without a fingerprint removed "
                  f"(the collector re-reports those)", flush=True)
            continue

        # Nothing left to remove here, so what remains is the final set and
        # its fingerprints can be rebuilt. Doing this while rows were still
        # being deleted would record fingerprints for rows about to go.
        cursor.execute(
            "DELETE FROM ingest_fingerprint WHERE table_name = %s", (table,))
        cursor.execute(
            "INSERT IGNORE INTO ingest_fingerprint (table_name, fp) "
            f"SELECT %s, dup_fp FROM `{table}` WHERE dup_fp IS NOT NULL",
            (table,),
        )

    if outstanding:
        return                      # more to do; the next ingest continues it

    cursor.execute(
        "INSERT IGNORE INTO ingest_migration (name) VALUES (%s)", (_DEDUP_REPAIR,))


def create_tables_if_not_exist(db_name):
    conn = connect_db(db_name)
    cursor = conn.cursor()
    try:
        # Resolved against this file, not the working directory. It used to be
        # a bare relative "init.sql" whose FileNotFoundError was swallowed, so
        # running the server from anywhere but the repo root created no tables
        # at all and said nothing about it.
        schema_path = pathlib.Path(__file__).resolve().parent / "db" / "init.sql"
        try:
            sql = schema_path.read_text(encoding="utf-8")
        except FileNotFoundError:
            # Loud: every agent table comes from this file.
            print(f"[!] Schema {schema_path} is missing — agent tables will not "
                  f"be created. This is not recoverable at runtime.")
            sql = ""
        # Comment lines are stripped rather than the chunk being skipped when
        # it starts with one. Splitting on ';' leaves a statement's leading
        # comment attached to it, so "skip chunks beginning with --" silently
        # discards every documented statement - that bug cost 19 of 52
        # statements in the agent's copy of this loop. Here the risk was the
        # milder one: a chunk that is only a comment was sent to MySQL and
        # produced a syntax error on every startup.
        for statement in _split_sql_statements(sql):
            try:
                cursor.execute(statement)
            except mysql.connector.Error as e:
                print(f"[!] SQL Execution Error: {e}")

        # Guarded, and after the CREATEs so the tables exist on a fresh DB.
        try:
            _migrate_automation_status(cursor, db_name)
        except mysql.connector.Error as e:
            print(f"[!] automations.status migration skipped: {e}")

        try:
            _migrate_siem_events(cursor, db_name)
        except mysql.connector.Error as e:
            print(f"[!] siem_events migration skipped: {e}")

        try:
            _migrate_soar_actions(cursor, db_name)
        except mysql.connector.Error as e:
            print(f"[!] soar_actions migration skipped: {e}")

        try:
            _migrate_encrypted_widths(cursor, db_name)
        except mysql.connector.Error as e:
            print(f"[!] encrypted-width migration skipped: {e}")

        try:
            # Stamped once per incarnation of this database. INSERT IGNORE, so
            # a database that already has an identity keeps it and only a
            # freshly created one gets a new value - which is exactly the
            # event the agent needs to hear about.
            cursor.execute(
                "INSERT IGNORE INTO ingest_epoch (id, epoch) VALUES (1, UUID())")
        except mysql.connector.Error as e:
            print(f"[!] could not stamp the ingest epoch: {e}")

        try:
            # Every column update_agent_info's INSERT names is declared here.
            #
            # hostname, mac_address and reported_ip used to be added by ALTERs
            # in that function instead, which meant a fresh database depended
            # on those running before its first check-in - and once they are
            # guarded, a guard that correctly does nothing would have broken
            # it. A migration is for databases that already exist; the CREATE
            # has to stand on its own.
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS agent_info (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    agent_name VARCHAR(255) NOT NULL,
                    public_ip VARCHAR(45),
                    reported_ip VARCHAR(45) NULL,
                    os_info VARCHAR(255) NULL,
                    hostname VARCHAR(255) NULL,
                    mac_address VARCHAR(48) NULL,
                    last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    UNIQUE KEY unique_agent (agent_name)
                ) ENGINE=InnoDB
            """)
            _migrate_agent_info(cursor, db_name)
        except mysql.connector.Error as e:
            print(f"[!] Agent info table creation/alter error: {e}")

        try:
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS ingest_fingerprint (
                    table_name VARCHAR(64) NOT NULL,
                    fp CHAR(64) NOT NULL,
                    first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    PRIMARY KEY (table_name, fp)
                ) ENGINE=InnoDB
            """)
        except mysql.connector.Error as e:
            print(f"[!] Fingerprint table creation error: {e}")

        try:
            # After the CREATE above, which it writes into.
            _repair_rows_stored_without_a_fingerprint(cursor, db_name)
        except mysql.connector.Error as e:
            print(f"[!] duplicate-row repair skipped: {e}")

        conn.commit()
    finally:
        cursor.close()
        conn.close()

def _json_default(o):
    if isinstance(o, datetime):
        return o.isoformat()
    return str(o)

def compute_fingerprint(table: str, item: dict) -> str:
    """What "the same row" means, for deduplication.

    The agent's `dup_fp` first, and this is the whole fix. Hashing the row as
    it arrives cannot work and never could:

      - encrypted columns arrive as `enc::gAAAA...`, and Fernet uses a random
        IV, so the same plaintext encrypts differently every time;
      - `timestamp` and `created_at` are set per collection cycle, so even a
        plaintext table changes on every send.

    Either one makes the hash unique by construction, so nothing was ever
    recognised as already held. Measured on one host: `packages` had 167
    distinct agent fingerprints and 249,980 rows stored, with an
    `ingest_fingerprint` entry for every single one. `software_inventory` was
    161,157 rows for 313 distinct programs — around 515 copies each, all
    faithfully recorded as new.

    `Sentora/modules/enc_db.content_fingerprint` exists precisely because of
    this and says so in its own docstring: the agent computes the value over
    plaintext, before encryption, because the server cannot. It just was not
    being used at this end.

    The row hash remains as the fallback for a table whose producer sets no
    `dup_fp`. It still cannot deduplicate anything with a timestamp — that is
    what `test_a_dedup_table_carries_a_fingerprint_from_the_agent` is for.
    """
    supplied = item.get("dup_fp")
    if isinstance(supplied, str) and supplied.strip():
        return supplied.strip()

    clean = {k: v for k, v in item.items() if k not in ("id", "sent")}
    blob = json.dumps(clean, sort_keys=True, separators=(",", ":"), default=_json_default).encode("utf-8")
    return hashlib.sha256(table.encode() + b"|" + blob).hexdigest()

# Defined in core/triage.py so the defensive sweep in app.py computes exactly
# the same value - see the note there.
from core.triage import compute_ai_fingerprint  # noqa: E402,F401

def update_agent_info(agent: str, public_ip: str, os_info: str = None, hostname: str = None,
                      mac_address: str = None, reported_ip: str = None):
    """Record where this agent is, from both vantage points.

    `public_ip` is what we observed the connection coming from; `reported_ip`
    is what the agent worked out about itself. They are usually the same, and
    where they differ neither one is simply right:

      - Cloud VM behind NAT: the agent sees 172.31.x, we see the elastic IP.
        Ours is the one worth showing an operator.
      - Server in Docker on the same machine as the agent: the connection
        arrives from the bridge gateway (172.18.0.1), which is a *router*, not
        the agent. The agent's own address is the only one that describes a
        host.

    Both are kept because they answer different questions. This used to be
    about which one to dial - overwriting one with the other threw away the
    address that worked, and VNC and SOAR dispatch failed with `Connection
    refused` against an IP that never had a listener on it. Nothing is dialled
    now, and the pair is worth more than it was rather than less:
    `reported_ip` is where the endpoint believes it sits, and `public_ip` is
    where its packets actually arrive from - the one thing recorded here that
    a compromised agent cannot forge, because it is a property of the TCP
    connection rather than of anything the agent said.
    """
    db_name = create_agent_db_if_not_exists(agent)
    create_tables_if_not_exist(db_name)
    conn = connect_db(db_name)
    cursor = conn.cursor()
    try:
        # The columns are declared in the CREATE and any older table was
        # migrated by `_migrate_agent_info`, both inside the
        # `create_tables_if_not_exist` above. Three unconditional ALTERs used
        # to sit here instead, running on every single check-in.
        cursor.execute("""
            INSERT INTO agent_info (agent_name, public_ip, reported_ip, os_info, hostname, mac_address, last_seen)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE
            public_ip = VALUES(public_ip),
            reported_ip = COALESCE(VALUES(reported_ip), reported_ip),
            os_info = VALUES(os_info),
            hostname = COALESCE(VALUES(hostname), hostname),
            mac_address = COALESCE(VALUES(mac_address), mac_address),
            last_seen = VALUES(last_seen)
        """, (agent, public_ip, reported_ip, os_info, hostname, mac_address, datetime.now()))
        conn.commit()
    except Exception as e:
        print(f"! Error updating agent info: {e}")
    finally:
        cursor.close()
        conn.close()

from core import mq as mq_utils
from core import triage
import core.opensearch as os_utils

async def publish_to_ai_queue(agent: str, table: str, item: dict):
    """Publish a log entry to RabbitMQ for both automation and defensive (SOAR) analysis"""
    await mq_utils.publish_to_queue(mq_utils.AI_AUTOMATION, agent, table, item)

    if table == "events_alert":
        await mq_utils.publish_to_queue(mq_utils.AI_SOAR, agent, table, item)
        print(f"[AI-Trigger] Published {table} task for {agent} to SOAR (Defensive) queue.")

def _parse_os_info_tail(os_info: str | None):
    """Agent appends |HOST=<hostname>|MAC=<addr> to OS_INFO so the server
    can recover the real machine identifiers without changing the TCP
    wire format. Returns (clean_os_info, hostname, mac_address)."""
    if not os_info or "|" not in os_info:
        return os_info, None, None
    parts = os_info.split("|")
    base = parts[0]
    hostname = mac = None
    for p in parts[1:]:
        if p.startswith("HOST="):
            hostname = p[5:].strip() or None
        elif p.startswith("MAC="):
            mac = p[4:].strip() or None
    return base, hostname, mac


def _ingest_epoch(cursor) -> str:
    """Which incarnation of this agent's database the server is on.

    Returned with every receipt so the agent can notice a reset. `sent` lives
    on the agent and is its only record of what the server holds; when this
    database is recreated the agent has no way to learn that, and by design it
    never re-offers a row it marked sent. Tables that produce rows constantly
    refill on their own, so the damage lands exactly on the quiet ones - a
    host held 93 port-scan rows against a server holding none, with nothing
    wrong at either end.

    Empty on failure rather than raising. An unknown epoch has to read as "no
    information", because the alternative - the agent treating a hiccup as a
    reset - would re-ship its whole history on a bad query.
    """
    try:
        cursor.execute("SELECT epoch FROM ingest_epoch WHERE id = 1")
        row = cursor.fetchone()
        return row[0] if row else ""
    except Exception:
        return ""


async def insert_data(agent: str, table: str, data: list, public_ip: str = None, os_info: str = None,
                      reported_ip: str = None) -> dict:
    """Store a batch and report what actually happened to it.

    The return value is the point. This used to return None and swallow every
    insertion failure into a debug line, which was fine for a caller that
    ignored it - and the caller ignored it because the wire protocol had no
    way to say anything back. The agent therefore marked rows sent the moment
    `sendall` returned, and a server that accepted a batch and stored none of
    it was indistinguishable from one that stored all of it.

    That is not hypothetical: four tables were permanently empty for days
    while the agent logged "sent (50 rows)" every cycle, because the column
    was too narrow for the ciphertext and every INSERT raised 1406 into this
    `except`.

    So: a receipt, and an `error` that survives to the other end.
    """
    db_name = create_agent_db_if_not_exists(agent)
    create_tables_if_not_exist(db_name)

    clean_os, hostname, mac_address = _parse_os_info_tail(os_info)
    if public_ip or clean_os:
        update_agent_info(agent, public_ip, clean_os, hostname=hostname,
                          mac_address=mac_address, reported_ip=reported_ip)

    conn = connect_db(db_name)
    cursor = conn.cursor()
    try:
        if table not in ALLOWED_TABLES:
            print(f"[!] Unknown table '{table}' received. Skipping.")
            return {"stored": 0, "error": f"unknown table {table!r}"}

        # A snapshot replaces what was there. Named rather than written out
        # here, because the same list existing twice is how it came to
        # overlap with DEDUP_TABLES in the first place.
        if table in SNAPSHOT_TABLES:
            cursor.execute(f"DELETE FROM `{table}`")

        stored = 0
        duplicates = 0
        rejected = 0
        first_error = None

        for item in data:
            item = dict(item)
            item.pop("id", None)
            item["sent"] = False

            if table in DEDUP_TABLES:
                fp = compute_fingerprint(table, item)
                cursor.execute(
                    "INSERT IGNORE INTO ingest_fingerprint (table_name, fp) VALUES (%s, %s)",
                    (table, fp)
                )
                if cursor.rowcount == 0:
                    # Already held. Counted as accepted rather than stored:
                    # the agent must not resend it, and reporting it as
                    # stored would make dedup look like data loss.
                    duplicates += 1
                    continue

            keys = ', '.join(f"`{k}`" for k in item.keys())
            values = ', '.join(['%s'] * len(item))
            sql = f"INSERT INTO `{table}` ({keys}) VALUES ({values})"
            if table in SNAPSHOT_TABLES:
                # Last write wins, which is what "snapshot" means.
                #
                # The agent appends a row per entity per collection cycle and
                # ships them in batches of fifty, so one batch routinely
                # carries the same container or disk from several cycles. The
                # server's DELETE clears the previous picture but not the
                # repeats inside the batch itself, and `docker_containers` has
                # UNIQUE(container_id) - so the second copy raised 1062, the
                # whole batch rolled back, and the table stayed empty while
                # the agent logged "docker_containers sent (8 rows)".
                updates = ', '.join(f"`{k}`=VALUES(`{k}`)" for k in item.keys())
                sql += f" ON DUPLICATE KEY UPDATE {updates}"

            try:
                cursor.execute(sql, list(item.values()))
                stored += 1
            except Exception as row_error:
                # Per row, because one unstorable row used to take the other
                # forty-nine with it. A batch is not a transaction as far as
                # the fleet is concerned: it is fifty independent facts, and
                # losing all of them because one is malformed is the worst
                # available outcome.
                rejected += 1
                if first_error is None:
                    first_error = f"{type(row_error).__name__}: {row_error}"

                # Take the fingerprint back out.
                #
                # It was written a few lines above, before the row it stands
                # for existed. While a failure rolled the whole batch back
                # that was harmless - the fingerprint went with it. Now that a
                # batch commits despite a rejected row, leaving it behind
                # marks the row as already-seen for ever: it is skipped as a
                # duplicate on every future attempt, so it can never be stored
                # even once the reason it failed is fixed.
                #
                # `registry_logs` showed exactly this - three fingerprints
                # held against a table that did not exist, for three rows that
                # would have stayed unstorable after the table was created.
                if table in DEDUP_TABLES:
                    try:
                        cursor.execute(
                            "DELETE FROM ingest_fingerprint "
                            "WHERE table_name=%s AND fp=%s", (table, fp))
                    except Exception as cleanup_error:
                        print(f"[!] {agent}/{table}: could not release the "
                              f"fingerprint for a rejected row: "
                              f"{cleanup_error}", flush=True)
                continue

            for _alert in correlate_across_hosts(agent, table, item, cursor):
                _spawn(publish_to_ai_queue(agent, "events_alert", _alert),
                       label=f"ai-publish {agent}/correlation")

            if table in {"siem_events", "events_alert"}:
                # Gate 1 — severity floor. Cheap, and happens before anything
                # touches the dedup table.
                send, why = triage.passes_severity(item)
                if not send:
                    # Recorded in the agent's database, not a module counter:
                    # this process is not the one that serves the stats
                    # endpoint, so an in-memory tally would always read zero
                    # there — an answer-shaped number measuring nothing.
                    triage.record_drop(cursor, item.get("severity") or item.get("level") or "UNKNOWN")
                    if debug:
                        print(f"[Triage] skipped {agent}/{table}: {why}")
                else:
                    # Gate 2 — deduplication, counted in the database so it
                    # survives restarts and is shared across Sanic workers.
                    ai_fp = compute_ai_fingerprint(table, item)
                    try:
                        cursor.execute(triage.DEDUP_DDL)
                        is_new, seen = triage.record_occurrence(cursor, table, ai_fp)
                    except Exception as dedup_err:
                        # Fail open: a broken counter must not stop analysis.
                        if debug:
                            print(f"[Triage] dedup unavailable ({dedup_err}), analysing anyway")
                        is_new, seen = True, 1

                    if is_new:
                        # This is the whole AI pipeline's entry point. The task
                        # was previously assigned to a local that went out of
                        # scope immediately, leaving only asyncio's weak
                        # reference — so an event could be dropped before it
                        # ever reached the queue, while the log line below
                        # claimed it had been sent.
                        item["_ai_fingerprint"] = ai_fp
                        _spawn(publish_to_ai_queue(agent, table, item),
                               label=f"ai-publish {agent}/{table}")
                    elif debug:
                        print(f"[Triage] duplicate {agent}/{table} (x{seen}), "
                              f"attached to the existing verdict")

            _spawn(os_utils.index_log(agent, table, item),
                   label=f"index {agent}/{table}")

        conn.commit()
        if rejected:
            # Reported, not hidden behind a successful-looking receipt. The
            # agent does not retry these - a row the server cannot store will
            # not become storable on the next attempt - so this line is the
            # only trace they ever leave.
            print(f"[!] {agent}/{table}: stored {stored}, rejected {rejected} "
                  f"({first_error})", flush=True)
        return {"stored": stored, "duplicates": duplicates,
                "rejected": rejected, "error": None,
                "epoch": _ingest_epoch(cursor)}
    except Exception as e:
        # Printed unconditionally, not behind `debug`. A batch that fails to
        # store is the single most important thing this process can say, and
        # it spent that failure inside an `if debug` that is off in
        # production.
        print(f"[!] {agent}/{table}: data insertion failed: "
              f"{type(e).__name__}: {e}", flush=True)
        try:
            conn.rollback()
        except Exception:
            pass
        return {"stored": 0, "duplicates": 0, "rejected": 0,
                "error": f"{type(e).__name__}: {e}"}
    finally:
        cursor.close()
        conn.close()


async def recv_all(reader, length):
    data = b''
    while len(data) < length:
        more = await reader.read(length - len(data))
        if not more:
            raise EOFError(f"Expected {length} bytes but received {len(data)} bytes.")
        data += more
    return data


def _own_gateways() -> set[str]:
    """Router addresses on this machine's own interfaces.

    Read from the kernel's routing table, so this costs nothing and asks
    nobody. In a container the default route points at the Docker bridge
    gateway - which is exactly the address that turns up as the "peer" when an
    agent on the host connects to a server in a container.
    """
    gateways: set[str] = set()
    try:
        with open("/proc/net/route", "r", encoding="ascii") as fh:
            next(fh, None)                      # header
            for line in fh:
                parts = line.split()
                if len(parts) < 3:
                    continue
                raw = parts[2]                  # gateway, little-endian hex
                if raw == "00000000":
                    continue
                try:
                    packed = int(raw, 16).to_bytes(4, "little")
                    gateways.add(".".join(str(b) for b in packed))
                except Exception:
                    continue
    except Exception:
        # Not Linux, or no procfs. Losing the refinement is acceptable -
        # refusing telemetry over it is not - so this returns what it has.
        # Not logged: it runs on every connection, and on a platform without
        # procfs it would fail on every one of them.
        return gateways
    return gateways


def peer_is_a_local_router(host: str, gateways: set[str]) -> bool:
    """Whether this peer address is a router on our side rather than a host.

    When the peer we see is our own default gateway, the connection has been
    translated on the way in and the source address replaced by the router's.
    It tells us where the packets entered our network, not where the agent is,
    so it is not an answer to "which machine is this".
    """
    return bool(host) and host in gateways


def observed_peer_ip(writer) -> str | None:
    """The address this connection actually came from, when it tells us more
    than the agent already did.

    Two properties, and only the first is why it was written. It is usually
    the more public address: on a cloud VM the agent sees 172.31.x and we see
    the elastic IP, and the elastic IP is the one worth showing an operator.
    The second matters more now - it is the only address here the agent does
    not supply, so it is the only one a compromised endpoint cannot lie about.

    Returns None when the peer is loopback or link-local, because that means
    the agent is on this host or reaching us through something local, and its
    own idea of its address is the better one.

    Private ranges are *kept*. On a flat corporate LAN 10.x is the real
    address of the machine and there is nothing more public to have.
    """
    try:
        peer = writer.get_extra_info("peername")
    except Exception:
        return None
    if not peer:
        return None

    host = peer[0] if isinstance(peer, (tuple, list)) else str(peer)
    if not host:
        return None

    # IPv4-mapped IPv6, which is what a dual-stack listener reports.
    if host.startswith("::ffff:"):
        host = host[len("::ffff:"):]

    try:
        import ipaddress
        addr = ipaddress.ip_address(host)
    except ValueError:
        return None
    if addr.is_loopback or addr.is_link_local or addr.is_unspecified:
        return None

    # Our own router is not an agent.
    #
    # The rule above - "what we observed beats what we were told" - assumes
    # the translation happens on the *agent's* side, which is the cloud-VM
    # case it was written for. With the server in a container on the agent's
    # own machine it is the other way round: the connection is NAT'd on the
    # way in and arrives from the bridge gateway, 172.18.0.1. That address is
    # a router, not a host, and recording it overwrote the one address that
    # described a machine - which is how the console came to show a Docker
    # gateway as a host's "Primary IP", and, back when the server dialled
    # agents, why every call to that endpoint was refused.
    #
    # Decided from our own routing table, so it needs no configuration and
    # asks nothing outside.
    if peer_is_a_local_router(host, _own_gateways()):
        return None

    return host


async def handle_client(reader, writer, encrypted: bool = True):
    """One batch from one agent.

    `encrypted` says which listener the connection arrived on. It is recorded
    rather than acted on: refusing here would make a legacy agent's telemetry
    vanish with no explanation anywhere, which is the failure mode this whole
    module keeps being bitten by. INGEST_TLS_REQUIRED refuses it earlier, by
    not opening the plaintext port at all - a connection refused is a symptom
    an operator can see.
    """
    try:
        raw_len = await recv_all(reader, 4)
        (agent_name_len,) = struct.unpack('!I', raw_len)
        agent_name = (await recv_all(reader, agent_name_len)).decode('utf-8')

        if not encrypted and agent_name not in PLAINTEXT_AGENTS:
            PLAINTEXT_AGENTS.add(agent_name)
            print(f"[!] {agent_name} is sending telemetry in the clear on "
                  f"port {SERVER_PORT}. Its logs, hostnames and process "
                  f"names cross the network unencrypted. Rebuild the agent "
                  f"against an https:// server URL to move it to "
                  f"{INGEST_TLS_PORT}; set INGEST_TLS_REQUIRED=1 to stop "
                  f"accepting this once the fleet has moved.", flush=True)

        raw_ip_len = await recv_all(reader, 4)
        (ip_len,) = struct.unpack('!I', raw_ip_len)
        claimed_ip = (await recv_all(reader, ip_len)).decode('utf-8')

        # What we observed beats what we were told.
        #
        # The agent works its own address out from a UDP route lookup, which
        # is the right thing to do - it contacts nothing and works air-gapped.
        # On a cloud VM behind NAT it returns the private address: an EC2 host
        # reported 172.31.42.49 while the world saw 16.171.42.197, and the
        # console labelled the private one "Primary IP".
        #
        # The peer address of an established TCP connection is the one thing
        # here that cannot be wrong, and cannot be forged by the agent either.
        public_ip = observed_peer_ip(writer) or claimed_ip

        raw_os_len = await recv_all(reader, 4)
        (os_len,) = struct.unpack('!I', raw_os_len)
        os_info = (await recv_all(reader, os_len)).decode('utf-8')

        raw_fname_len = await recv_all(reader, 4)
        (fname_len,) = struct.unpack('!I', raw_fname_len)
        fname = (await recv_all(reader, fname_len)).decode('utf-8')

        raw_fsize = await recv_all(reader, 8)
        (fsize,) = struct.unpack('!Q', raw_fsize)

        data_bytes = await recv_all(reader, fsize)
        try:
            data = json.loads(data_bytes.decode('utf-8'))
        except Exception as e:
            if debug:
                print(f"[ERROR] JSON decode failed from {agent_name}@{public_ip}: {e}")
                return

        receipt = await insert_data(agent_name, fname.replace(".json", ""), data,
                                    public_ip, os_info, reported_ip=claimed_ip)
        await _send_receipt(writer, receipt)
        if debug:
            print(f"[INFO] Data received from {agent_name}@{public_ip} ({os_info}) - File: {fname}, Size: {fsize} bytes")

    except Exception as e:
        if debug:
            print(f"[ERROR] Client handling failed: {e}")
    finally:
        writer.close()
        await writer.wait_closed()


async def _send_receipt(writer, receipt) -> None:
    """Tell the agent what became of the batch it just sent.

    The ingest protocol had no reply at all, which is why the agent marked
    rows sent on `sendall` returning - the strongest thing it could know, and
    much weaker than it read. This adds the only frame that lets the agent
    distinguish "stored" from "accepted by a socket".

    Appended rather than versioned, and safe in both directions. An agent
    built before this reads nothing and closes; the write lands in a socket
    buffer nobody drains and is discarded when the connection goes, so the
    failure mode for an old fleet is exactly the behaviour it has today.
    """
    if receipt is None:
        return
    try:
        blob = json.dumps(receipt).encode("utf-8")
        writer.write(struct.pack("!I", len(blob)) + blob)
        await writer.drain()
    except Exception as e:
        # An agent that hung up before reading is the ordinary case for any
        # build older than this frame, so this is not worth a line unless
        # somebody is debugging.
        if debug:
            print(f"[INFO] receipt not delivered: {type(e).__name__}: {e}")


def _ingest_tls_context():
    """The server-side TLS context for the ingest listener, or None.

    Shares `core.tls` with the console rather than growing a second pair of
    certificate settings. One identity for the deployment is also the only
    arrangement an agent can be told to trust in one step.

    This process reads the certificate and never creates one. `app` owns
    generation, the way it already owns the Fernet key, and mounts the
    directory read-only here. Two processes generating independently would
    each end up holding a certificate the other does not, and an agent that
    verified the console would then fail against ingest for no visible
    reason - a host that goes quiet, which is the exact symptom this codebase
    keeps chasing.
    """
    from core import tls as product_tls

    if not product_tls.tls_enabled():
        return None

    config = product_tls.existing()
    if not config:
        # Not an error worth a traceback: on a first boot this process
        # usually wins the race with `app` by a few seconds. `restart: always`
        # brings it back, and the message says what it is waiting for rather
        # than looking like a broken configuration.
        raise RuntimeError(
            "no certificate at certs/server.crt yet. `app` generates it on "
            "first boot and this process reads it; if `app` is running and "
            "this persists, check that both mount the same certs volume."
        )

    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    context.load_cert_chain(certfile=config["cert"], keyfile=config["key"])
    return context


async def main():
    servers = []

    tls_context = None
    try:
        tls_context = _ingest_tls_context()
    except Exception as e:
        # Same rule as the console: asking for TLS and not getting it is a
        # startup failure. Telemetry is the more sensitive of the two streams
        # - it carries the process names, paths and log lines of every host -
        # so falling back to the plaintext port would be the worse half of
        # the deployment silently continuing.
        print(f"[!] ingest TLS was requested and cannot be served: {e}", flush=True)
        raise SystemExit(2)

    if tls_context is not None:
        servers.append(await asyncio.start_server(
            functools.partial(handle_client, encrypted=True),
            SERVER_IP, INGEST_TLS_PORT, ssl=tls_context,
            reuse_address=True, reuse_port=False,
        ))
        print(f"[*] TLS ingest listening on {SERVER_IP}:{INGEST_TLS_PORT}", flush=True)
    elif INGEST_TLS_REQUIRED:
        print("[!] INGEST_TLS_REQUIRED=1 but TLS_ENABLED is not set, so there "
              "is no encrypted listener to require. Set TLS_ENABLED=1.",
              flush=True)
        raise SystemExit(2)

    if INGEST_TLS_REQUIRED:
        print(f"[*] Plaintext ingest on {SERVER_PORT} is closed "
              f"(INGEST_TLS_REQUIRED=1). Agents that have not been rebuilt "
              f"will be refused rather than silently accepted.", flush=True)
    else:
        servers.append(await asyncio.start_server(
            functools.partial(handle_client, encrypted=False),
            SERVER_IP, SERVER_PORT,
            reuse_address=True, reuse_port=False,
        ))
        warning = ("in the clear" if tls_context is None else
                   "in the clear, for agents not yet rebuilt")
        print(f"[*] TCP ingest listening on {SERVER_IP}:{SERVER_PORT} ({warning})",
              flush=True)

    async with contextlib.AsyncExitStack() as stack:
        for server in servers:
            await stack.enter_async_context(server)
        await asyncio.gather(*(s.serve_forever() for s in servers))

if __name__ == "__main__":
    asyncio.run(main())