Blame for runtime/python/dibaguard/shield/database.py

"""Database protection: understands database engines, their query languages and their logs, and guards them.

Engines: PostgreSQL, MySQL/MariaDB, MSSQL, Oracle, CockroachDB, ClickHouse (SQL); MongoDB, CouchDB (document);
Cassandra/ScyllaDB (CQL); Redis, Memcached, etcd (key-value); Elasticsearch/OpenSearch (search); Neo4j (graph);
InfluxDB (time series); Kafka, ZooKeeper (infrastructure). Each is known by its ports (flows and the posture audit)
and, when its query/audit log is available, by its query language.

    g = DbGuard(DbNet.load("models/dbnet.npz"))
    g.calibrate(normal_events)      # the applications' normal workload: who talks to which DB, which query shapes, how much
    out = g.score(events)           # per event: alarm, families, severity, evidence, whether the client is a known app server
    events come from the adapters: parse_postgres / parse_mysql / parse_mongo / parse_cassandra / parse_redis / parse_es

Three independent views of every query, combined:
  signatures   precise, named: RCE from the database (COPY ... TO PROGRAM, xp_cmdshell, CONFIG SET dir, $where ...),
               privilege escalation, destruction / ransom, injection (judged on the query with its string literals
               removed: a payload *inside* a properly quoted value is data; only one that changes the structure is an
               injection), credential / catalog recon, unbounded reads
  DbNet        the learned tier (dibaguard/dbnet.py): one small network over the query's structure for every language,
               thresholded per database by conformal calibration on that database's own normal queries
  behaviour    per client: authentication-failure bursts, rows read far above the busiest normal client, clients that
               never talked to this database, query shapes the applications never run
Shapes the applications run in their normal workload are trusted for the soft families (injection by the model alone,
recon, unbounded): an injection is by definition a structure the application did not write.

The database's clients are mostly the organisation's own application servers. Blocking one at the network level
takes the application down, so for a known app client the verdict is an alert with the exact remediation to run on
the database (kill the session, revoke), and the Layer attributes the attack to the web source that caused it
(cross-layer correlation with HTTP alarms); only unknown / public clients are blocked outright.
"""
from __future__ import annotations

import ipaddress
import json
import math
import re
import time
from collections import OrderedDict

import numpy as np

from dibaguard.shield.rx import Gap, Sig

# port -> (engine, kind)
DB_PORTS = {
    5432: ("postgres", "sql"), 3306: ("mysql", "sql"), 33060: ("mysql", "sql"), 1433: ("mssql", "sql"), 1521: ("oracle", "sql"),
    26257: ("cockroach", "sql"), 8123: ("clickhouse", "sql"), 9000: ("clickhouse", "sql"), 27017: ("mongodb", "document"),
    27018: ("mongodb", "document"), 27019: ("mongodb", "document"), 5984: ("couchdb", "document"), 9042: ("cassandra", "cql"),
    9142: ("cassandra", "cql"), 7000: ("cassandra-internode", "cql"), 7001: ("cassandra-internode", "cql"), 7199: ("cassandra-jmx", "cql"),
    6379: ("redis", "kv"), 6380: ("redis", "kv"), 11211: ("memcached", "kv"), 2379: ("etcd", "kv"), 9200: ("elasticsearch", "search"),
    9300: ("elasticsearch", "search"), 7474: ("neo4j", "graph"), 7687: ("neo4j", "graph"), 8086: ("influxdb", "tsdb"),
    9092: ("kafka", "infra"), 2181: ("zookeeper", "infra"), 28015: ("rethinkdb", "document"), 8529: ("arangodb", "document"),
}
ENGINES = sorted({v[0] for v in DB_PORTS.values()})
_KIND = {"postgres": "sql", "mysql": "sql", "mssql": "sql", "oracle": "sql", "cockroach": "sql", "clickhouse": "sql", "sqlite": "sql",
         "cassandra": "cql", "neo4j": "graph"}           # languages with single-quoted string literals

_F = re.I | re.S
# judged on the structural view (string literals -> 's', comments -> space, versioned comments unwrapped)
# Branches shaped "X .* Y" (a keyword, anything, another keyword) are rx.Gap matchers rather than regex branches: a
# backtracking regex rescans the query from every X on crafted input (tens of ms per KB, the A22 tier judges six
# query contexts per value); a Gap gives the same answer in one pass. Sig(...).search works like a regex's.
SIGS = {
    "rce": Sig(
        r"\blo_(import|export)\s*\(|\bpg_(read_file|read_binary_file|ls_dir|write_file|stat_file)\s*\("
        r"|\bxp_(cmdshell|dirtree|fileexist)\b"
        r"|\bsp_oa(create|method)\b|\bopen(rowset|datasource)\s*\(|\bsys_(exec|eval)\s*\(|\bload\s+data\s+(local\s+)?infile\b|\bsoname\b"
        r"|\binto\s+(out|dump)file\b|\bload_file\s*\(|\bdbms_(java|scheduler)\b|\butl_(http|file|tcp|inaddr)\b|\bhttpuritype\b|\$where\b|\$function\b"
        r"|\$accumulator\b|\bmapreduce\b|\brunprogram\s*\(|^\s*config\s+set\s+(dir|dbfilename)\b|^\s*(slaveof|replicaof)\s+(?!no\s+one)\S+"
        r"|redis\.p?call\s*\(\s*['\"](config|slaveof|replicaof|module|debug)['\"]"          # the same, wrapped in EVAL / FUNCTION"
        r"|^\s*module\s+load\b|^\s*system\.exec\b|\bpackage\.loadlib\b|\bdblink(_exec|_connect)?\s*\("
        r"|\bjava\.lang\.|\bgetruntime\s*\(|\bapoc\.(load|cypher\.run\w*|import)\w*|\binsert\s+into\s+function\b|\bfrom\s+(url|file|s3|hdfs)\s*\(",
        Gap([r"\bcopy\b", r"\b(to|from)\s+program\b"], None, _F),
        Gap([r"\bcreate\s+(or\s+replace\s+)?function\b", r"\blanguage\s+(c|plpythonu|plpython3u|plperlu|plsh|java|javascript)\b"], None, _F),
        Gap([r"\bcreate\s+extension\b", r"\b(dblink|plpython\w*|adminpack)\b"], None, _F), flags=_F),
    "privilege": Sig(
        r"\bgrant\s+all\b"
        r"|\bset\s+global\b|\bsp_addsrvrolemember\b"
        r"|\balter\s+server\s+role\s+sysadmin\b|\bupdate\s+(pg_authid|mysql\.user)\b|\bset\s+role\s+postgres\b|\balter\s+system\s+set\b"
        r""
        r"|^\s*acl\s+setuser\b.*(\+@all|allcommands|~\*|nopass)|^\s*config\s+set\s+(requirepass|protected-mode)\b|\bgrant\s+all\s+permissions\b"
        r"|redis\.p?call\s*\(\s*['\"]acl['\"]\s*,\s*['\"]setuser"
        r"|/_security/(user|role)/\S+\s*\{|\bdbms\.security\.createuser\b|\bgrant\s+role\s+admin\b|\bcreate\s+login\b",
        Gap([r"\balter\s+(user|role)\b", r"\b(superuser|createrole|bypassrls|replication)\b"], None, _F),
        Gap([r"\bgrant\b", r"\bto\s+public\b"], None, _F),
        Gap([r"\bcreate\s+(user|role)\b", r"\bsuperuser\b"], None, _F),
        Gap([r"\bgrant\b", r"\bwith\s+grant\s+option\b"], None, _F),
        Gap([r"\b(createuser|grantrolestouser|updateuser)\b", r"\b(root|useradminanydatabase|readwriteanydatabase|dbowner|clusteradmin|__system)\b"], None, _F), flags=_F),
    "destructive": Sig(
        r"\bdrop\s+(database|schema|table|keyspace|collection)\b|\btruncate\b|^\s*delete\s+from\s+[\w.\"`\[\]]+\s*;?\s*$"
        r"|^\s*delete\s+from\s+\S+\s+where\s+(?P<d1>\d+)\s*=\s*(?P=d1)\s*;?\s*$|^\s*update\s+[\w.\"`\[\]]+\s+set\s+(?:(?!\bwhere\b)[^;])*;?\s*$"
        r"|\bdropdatabase\b|\"drop\"\s*:\s*\"|\.drop\s*\(\s*\)|\b(deletemany|remove)\s*\(\s*\{\s*\}\s*\)|\"deletes\"\s*:\s*\[\s*\{\s*\"q\"\s*:\s*\{\s*\}"
        r"|^\s*(flushall|flushdb)\b|^\s*shutdown\b|^\s*debug\s+segfault\b"
        r"|redis\.p?call\s*\(\s*['\"](flushall|flushdb|shutdown)['\"]"
        r"|^\s*delete\s+/[\w*,.-]+/?\s*$|\bmatch\s*\(\s*\w+\s*\)\s*(detach\s+)?delete\b",
        Gap([r"\"q\"\s*:\s*\{\s*\}\s*,\s*\"u\"\s*:", r"\"multi\"\s*:\s*true"], None, _F),
        Gap([r"_delete_by_query", r"match_all"], None, _F), flags=_F),
    "injection": Sig(
        r"(?:\bor|\|\|)\s*\(?\s*(\d+|'s'|0x[0-9a-f]+)\s*(=|like|>|<)\s*(\d+|'s'|0x[0-9a-f]+)|(?:\bor|\|\|)\s+\(?\s*(?P<t1>[\w'()]+)\s*=\s*(?P=t1)(?![\w(])"
        r"|\bor\s+(true|not\s+0)\b|\(\s*select\s+(cast|convert)\s*\("
        r"|(?:\bor|\band|\bxor|\|\||&&)\s*\(?\s*\d+\s*[-*+/]\s*\d+\s*=\s*\d+"           # arithmetic probes: or 5*5=25, and 3-1=2
        r"|=\s*\(\s*select\s*[\d(]"                                                       # a scalar subquery standing in for a value
        r"|\b(and|or|xor|&&)\s+\d+\s*[=<>]+\s*\d+\s*(--|#|;|$)"                           # boolean probe ending the statement: and 1=2 --
        r"|\band\s+'s'\s*=\s*'s'\s*;?\s*$"                                                # string boolean probe: x' and '1'='1
        # the rest of the always-true class: between / in / is true / regexp / rlike / div, bare 'or 1', xor, a
        # function of constants - whatever spelling a boolean-blind probe or an auth bypass uses
        r"|(?:\bor|\bxor|\|\|)\s*\(?\s*(-?\d+|'s'|true|false|null)\s+(not\s+)?(between\s+\S+\s+and\s+\S+|in\s*\(|is\s+(not\s+)?(true|false|null|unknown)\b"
        r"|r?like\s|regexp\s|sounds\s+like\s|div\s|mod\s|<=>)"
        r"|(?:\bor|\bxor|\|\|)\s+(-?[1-9]\d*|true|'s')\s*(;|--|#|\)|$)|\bxor\s+(-?\d+|'s'|true)\b"
        r"|(?:\bor|\bxor)\s+(greatest|least|if|ifnull|coalesce|abs|sign|length|ascii|char|concat)\s*\(\s*(-?\d+|'s')\s*(,\s*(-?\d+|'s')\s*)*\)"
        r"|;\s*(and|or|xor)\s"                                                            # no statement starts with AND: broken by a payload"
        # UNION SELECT as an attack writes it - column padding / probes, catalogs or credentials, a trailing comment -
        # not the report query that legitimately unions two selects
        r"|\bunion(\s+all|\s+distinct)?\s+select\s+(null\b|\d|0x|concat|char\s*\(|@@|version\s*\(|user\s*\(|database\s*\(|group_concat|'s')"
        r""
        r"|;\s*(drop|truncate|delete|insert|update|create|alter|exec|execute|shutdown|grant|copy|declare|waitfor)\b|\b(pg_sleep|sleep|benchmark|randomblob)\s*\("
        r"|\bwaitfor\s+delay\b|\bdbms_(pipe\.receive_message|lock\.sleep)\b|\b(extractvalue|updatexml)\s*\(|\bcast\s*\(\s*\(\s*select\b"
        r"|\bconvert\s*\(\s*int\s*,\s*\(\s*select\b|\bexp\s*\(\s*~|\bfloor\s*\(\s*rand\s*\(|\b(order|group)\s+by\s+(\d+\s*,\s*){0,9}\d+\s*(--|#)"
        r"|\b(ascii|substring|substr|mid|ord)\s*\(\s*\(\s*select\b|\bif\s*\(\s*\d+\s*=\s*\d+\s*,\s*sleep"
        r"|\b(ascii|substring|substr|mid|ord|length)\s*\(\s*(@@\w+|(database|user|current_user|version|schema)\s*\(\s*\))|\border\s+by\s+\d+\s+order\s+by\b",
        Gap([r"\bunion(\s+all|\s+distinct)?\s+select\b", r"\b(from\s+(information_schema|mysql\.|pg_|sys\.|master\.|users|members|accounts|auth_user)"
                r"|password|passwd|authentication_string|pass\b)"], ";", _F),
        Gap([r"\bunion(\s+all|\s+distinct)?\s+select\b", r"(--|#)\s*-?\s*$"], ";", _F), flags=_F),
    "recon": Sig(
        r"\bpg_(shadow|authid|hba_file_rules)\b|\brolpassword\b|\bmysql\.user\b|\bauthentication_string\b|\bsys\.(sql_logins|server_principals|syslogins)\b"
        r"|\bsystem_auth\.\w+|\bsalted_hash\b|\blistdatabases\b|\bgetcmdlineopts\b|\busersinfo\b|\bsystem\.users\b|\bshowprivileges\b"
        r"|\bdb\.getusers\s*\(|^\s*config\s+get\b|^\s*(client\s+list|acl\s+list)\b|^\s*keys\s+\S*(pass|secret|token)|/_cat/(indices|aliases|shards)|/_security/user\b"
        r"|/_nodes/settings|include_defaults|@@(datadir|secure_file_priv|hostname)\b|\bshow\s+grants\b|\bis_srvrolemember\b|\bsysdatabases\b"
        r"|\binformation_schema\.user_privileges\b|\bgroup_concat\s*\(\s*(table|column)_name\b|\blist\s+(all\s+)?(permissions|users|roles)\b"
        r"|\bdbms\.security\.listusers\b|\btable_schema\s+not\s+in\b",
        Gap([r"\bpassword_hash\b", r"\bsys\."], None, _F), flags=_F),
    "unbounded": Sig(
        r"^\s*select\s+\*\s+from\s+(?!system|pg_|information_schema|sys\.|dual|mysql\.)[\w.\"`\[\]]+\s*;?\s*$|\ballow\s+filtering\b|\bfind\s*\(\s*(\{\s*\})?\s*\)|\"filter\"\s*:\s*\{\s*\}"
        r"|^\s*keys\s+\*\s*$|^\s*scan\s+\d+\s+match\s+\*\s|count\s+\d{5,}|\bsize\s*[=:]\s*\d{4,}|\bscroll\s*=|\blimit\s+\d{6,}|\bnext\s+\d{6,}\s+rows\b"
        r"|\bcopy\s+\w+\s+to\s+stdout\b|\"\$out\"|\bsql_no_cache\b", flags=_F),
}
# patterns that need the literal itself (judged on the raw text)
RAW = {
    "rce": Sig(r"\bsp_configure\s+'(xp_cmdshell|ole automation procedures)'|\bload\s+csv\s+from\s+'file:|\b(url|file|s3)\s*\(\s*'(https?://(169\.254|metadata)|/)"
                      r"",
                Gap([r"\"(script|source|inline|body|map|reduce|init|accumulate)\"\s*:\s*\"", r"(java\.lang|getruntime|runtime\.exec|runprogram|processbuilder)"], "\"", _F), flags=_F),
    "destructive": Sig(r"\bcreate\s+(table|database)\s+\W?\w*(readme|read_me|recover|warning|please_read)",
                Gap([r"\b(btc|bitcoin|monero|xmr)\b", r"\b(recover|backed\s+up|pay|send)\b"], None, _F),
                Gap([r"\b(recover|backed\s+up|pay|send)\b", r"\b(btc|bitcoin)\b"], None, _F), flags=_F),
    "injection": Sig(r"\"(pass\w*|pwd|token|secret|api_?key)\"\s*:\s*\{\s*\"\$(ne|gt|gte|lt|regex|exists|nin|in|not)\"|\b(pass\w*|pwd|token)\s*:\s*\{\s*\$(ne|gt|regex|exists)\b"
                            r"|\"\$or\"\s*:\s*\[\s*\{\s*\}|\"\$expr\"\s*:\s*\{\s*\"\$eq\"\s*:\s*\[\s*(?P<e1>\d+)\s*,\s*(?P=e1)\s*\]"
                            # an identity field (who logs in) matched by a catch-all operator: {"username": {"$ne": null}},
                            # {"email": {"$regex": "^admin"}}, {"login": {"$gt": ""}} - an application asks for a value there
                            r"|\"(user(name)?|login|e-?mail)\"\s*:\s*\{\s*\"\$(ne\"\s*:\s*(null|\"\")\s*\}"
                            r"|gte?\"\s*:\s*\"\"\s*\}|regex\"\s*:\s*\"(\^|\.\*))", flags=_F),
}
# words at least one of which every match of a family's signatures contains (checked before the regex runs)
_NEED = {
    "rce": ("program", "lo_", "pg_", "language", "xp_", "sp_", "openrowset", "opendatasource", "sys_", "infile", "soname", "outfile",
            "dumpfile", "load_file", "dbms_", "utl_", "httpuritype", "$where", "$function", "$accumulator", "mapreduce", "runprogram",
            "config", "slaveof", "replicaof", "module", "system.exec", "loadlib", "dblink", "extension", "java.lang", "getruntime", "apoc",
            "function", "url", "file", "s3", "hdfs", "redis.", "script", "source", "inline", "body", "map", "reduce", "init", "accumulate",
            "load csv", "processbuilder"),
    "privilege": ("alter", "grant", "create", "set global", "sp_addsrvrolemember", "sysadmin", "pg_authid", "mysql.user", "set role",
                  "createuser", "grantrolestouser", "updateuser", "acl", "requirepass", "protected-mode", "_security", "dbms.security"),
    "destructive": ("drop", "truncate", "delete", "update", "remove", "deletes", "\"q\"", "flushall", "flushdb", "shutdown", "debug",
                    "_delete_by_query", "match", "btc", "bitcoin", "monero", "xmr", "recover", "backed", "readme", "read_me", "warning",
                    "please_read", "redis."),
    "injection": (" or", "or ", "||", "xor", "and", "&&", "union", ";", "sleep", "benchmark", "randomblob", "waitfor", "dbms_", "extractvalue",
                  "updatexml", "cast", "convert", "exp", "rand", "order", "group", "ascii", "substr", "mid", "ord", "length", "if", "select",
                  "$", "pass", "pwd", "token", "secret"),
    "recon": ("pg_", "rolpassword", "mysql.user", "authentication_string", "sys.", "password_hash", "system_auth", "salted_hash",
              "listdatabases", "getcmdlineopts", "usersinfo", "system.users", "showprivileges", "getusers", "config", "client", "acl",
              "keys", "_cat", "_security", "_nodes", "include_defaults", "@@", "grants", "is_srvrolemember", "sysdatabases",
              "information_schema", "group_concat", "list", "dbms.security", "table_schema"),
    "unbounded": ("select", "filtering", "find", "filter", "keys", "scan", "count", "size", "scroll", "limit", "next", "copy", "$out",
                  "sql_no_cache"),
}
SEVERITY = {"rce": 1.0, "privilege": 0.95, "destructive": 0.95, "injection": 0.9, "exfil": 0.9, "bruteforce": 0.7, "new_client": 0.5,
            "recon": 0.5, "unbounded": 0.4}
SOFT = {"injection", "recon", "unbounded"}           # families a known application query shape can explain
LANG_KIND = {"sql": "sql", "cql": "sql", "graph": "sql"}

# remediation an operator (or an automation hook) runs on the database itself - DibaGuard never holds DB credentials
REMEDIATION = {
    "postgres": "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE client_addr = '{client}';  -- then REVOKE / fix pg_hba.conf",
    "mysql": "SELECT CONCAT('KILL ', id, ';') FROM information_schema.processlist WHERE host LIKE '{client}%';  -- then REVOKE",
    "mssql": "SELECT session_id FROM sys.dm_exec_sessions WHERE host_name = '{client}';  -- KILL <session_id>; disable xp_cmdshell",
    "oracle": "SELECT sid, serial# FROM v$session WHERE machine = '{client}';  -- ALTER SYSTEM KILL SESSION 'sid,serial#' IMMEDIATE;",
    "mongodb": "db.currentOp({{'client': /^{client}/}}).inprog.forEach(op => db.killOp(op.opid));  // then fix roles / net.bindIp",
    "cassandra": "nodetool disablebinary on exposed nodes; revoke the role; review system_auth; audit with nodetool getauditlog",
    "redis": "CLIENT KILL ADDR {client}:*  (or CLIENT KILL TYPE normal); then rename-command CONFIG/FLUSHALL/SLAVEOF, set ACLs / requirepass",
    "elasticsearch": "revoke the API key / user; set xpack.security.enabled=true; restore from snapshot if indices were deleted",
    "clickhouse": "KILL QUERY WHERE client_hostname = '{client}'; restrict the user's networks in users.xml",
    "neo4j": "CALL dbms.listConnections() YIELD connectionId, clientAddress WHERE clientAddress STARTS WITH '{client}' CALL dbms.killConnection(connectionId)",
}

_BATCH = re.compile(r"\s*begin\s+(unlogged\s+|counter\s+)?batch\b", re.I)
_BATCH_SEP = re.compile(r";\s*(?=(insert|update|delete|apply)\b)", re.I)
_KV = {"redis", "memcached", "etcd"}
_MONGO_OP = re.compile(r'\{\s*"\$(ne|gt|gte|lt|lte|regex|exists|nin|in|where|expr|not|nor|or|and|options|function|elemMatch|all|size|type|mod)"'
                       r'\s*:\s*("(?:[^"\\]|\\.)*"|[^{}\[\]]*|\[[^\]]*\])\s*\}')
_KV_RISKY = {"config", "slaveof", "replicaof", "module", "acl", "eval", "evalsha", "flushall", "flushdb", "shutdown", "debug", "keys", "scan",
             "client", "info", "system.exec", "migrate", "script", "function"}
_JSONSTR = re.compile(r"\"((?:[^\"\\]|\\.)*)\"(\s*:)?")          # every JSON string in order (keys and values paired right)


def _jsonval(m: re.Match) -> str:
    """a JSON key or an $operator stays; a string value is just data"""
    return m.group(0) if m.group(2) or m.group(1).startswith("$") else '"s"'
_SQ = re.compile(r"'(?:[^'\\]|\\.|'')*'")
_VC = re.compile(r"/\*!\d*\s*(.*?)\*/", re.S)
_CM = re.compile(r"/\*.*?\*/", re.S)
_STR = re.compile(r"'(?:[^'\\]|\\.)*'|\"(?:[^\"\\]|\\.)*\"")
_NUM = re.compile(r"\b\d+(\.\d+)?\b")
_MIX = re.compile(r"(?<![\w$])\d[a-z0-9_-]*[a-z][a-z0-9_-]*|\b[0-9a-f]{8,}\b|\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b")
_KVKEY = re.compile(r"(?<=[:/])[^\s:/\"']+(?=\s|$)")
_WS = re.compile(r"\s+")
_LC_DASH = re.compile(r"--[^\r\n]*[\r\n]")                   # a line comment the newline ends (every SQL dialect)
_LC_HASH = re.compile(r"(?:#|--)[^\r\n]*[\r\n]")             # ... and MySQL's (ClickHouse's) '#' comment
_HASH_COMMENT = ("", "mysql", "clickhouse")
# the comment-cut fallback of classify (_comment_cut): a literal / lone quote, a comment marker, and what may stand
# between them - nothing (or closing parentheses), or a SQL tail: a connective or clause keyword, an operator, or a
# statement after ';'
_BREAK = re.compile(r"'s'|'")
_TERM = re.compile(r"--|#|/\*")
_TERM_LIT = re.compile(r"--|#|/\*|;")          # in a literal context ';' ends the statement the payload broke into, too
# string literals where a backslash escapes nothing (standard SQL: PostgreSQL, SQLite, MSSQL, Oracle, MySQL with
# NO_BACKSLASH_ESCAPES): there admin\' closes the literal that MySQL would keep open
_SQ_STD = re.compile(r"'(?:[^']|'')*'")
_NOTHING = re.compile(r"[\s)]*")
_SQL_LEAD = re.compile(r"[\s)]*(?:(?:or|and|xor|not|union|select|having|group|order|limit|procedure|into|from|where|like|rlike|regexp|in|is"
                       r"|between|div|mod|sounds|collate|case|exists|glob|notnull|isnull)(?![a-z0-9_])|\|\||&&|[=<>!+*/%^~|&,?:@]|-(?!-)"
                       r"|;\s*(?:select|insert|update|delete|drop|create|alter|exec|execute|declare|waitfor|shutdown|grant|truncate|set"
                       r"|show|call|load|with|begin|handler|do)(?![a-z0-9_]))", re.I)


def shape(query: str) -> str:
    """normalise a query to its shape (literals -> ?, numbers -> 0, whitespace collapsed): what the application *does*,
    independent of the values - the same idea as pg_stat_statements / the MongoDB query shape"""
    q = _STR.sub("?", query)
    q = _MIX.sub("0", q.lower())
    q = _NUM.sub("0", q)
    q = _KVKEY.sub("?", q)                            # key-value commands: the key's prefix is the shape, its id is a value
    return _WS.sub(" ", q).strip()[:2000]


def views(query: str, engine: str = "") -> tuple[str, str, str]:
    """(raw, literals stripped, structural). Values are data in every language: single-quoted literals in SQL / CQL /
    Cypher, JSON string values (keys and $operators kept) in document and search requests, and the arguments of
    ordinary key-value commands - only the risky ones (CONFIG, SLAVEOF, EVAL, ...) are read with their arguments."""
    raw = str(query)
    if len(raw) > 1024:                      # bounded work per query (an attacker controls the length): the head and the
        raw = raw[:768] + " " + raw[-256:]   # tail, where an injected payload lands - padding in front cannot push it out
    if engine in _KV:
        toks = raw.split()
        s = raw if toks and toks[0].lower() in _KV_RISKY else (toks[0] if toks else "")
        return raw, s, _WS.sub(" ", s)
    if _KIND.get(engine, "sql" if not engine else None) is None:
        s = _WS.sub(" ", _JSONSTR.sub(_jsonval, raw))
        return raw, s, s
    s1 = _SQ.sub("'s'", raw)
    s2 = _CM.sub(" ", _VC.sub(r" \1 ", s1))
    # line comments that a newline ends go before whitespace is folded: "union#x<LF>select" is UNION SELECT, and once
    # the newline is a space the comment would swallow the rest (V-02). '#' comments only where the dialect has them
    if "\n" in s2 or "\r" in s2:
        s2 = (_LC_HASH if engine in _HASH_COMMENT else _LC_DASH).sub(" ", s2)
    s2 = _WS.sub(" ", s2).strip()
    return raw, s1, s2


def classify(query: str, engine: str = "", literal: bool | None = None) -> list[str]:
    """families a single query shows, most severe first. literal: how the comment-cut fallback reads the query - None
    (a query from a log: _comment_cut), True (the A22 tier placed a value inside a string literal the query opened, so
    the query's first quote is the application's: _literal_cut), False (a value placed outside any literal, the numeric
    contexts: no fallback - a quote there opens a literal, it closes none)."""
    raw, s1, s2 = views(query, engine)
    if engine == "cassandra" and _BATCH.match(s2):            # a CQL batch is a legitimate run of statements
        s2 = _BATCH_SEP.sub(" ", s2)
    sql_like = _KIND.get(engine, "sql" if not engine else None) is not None
    doc = engine not in _KV and not sql_like
    low_s2, low_raw = s2.lower(), raw.lower()
    hits = []
    for f in ("rce", "privilege", "destructive", "injection", "recon", "unbounded"):
        # a family's regex runs only when one of its words is present (bounded cost on long, crafted queries)
        if not any(k in low_s2 or k in low_raw for k in _NEED[f]):
            continue
        if (SIGS[f].search(raw if (doc and f == "privilege") else s2) and (f != "injection" or sql_like)) or (f in RAW and RAW[f].search(raw)):
            hits.append(f)
    if "injection" not in hits and sql_like and literal is not False:
        if literal:
            # the break-out is judged as MySQL reads backslashes and as standard SQL does (x\' or json(id); breaks out of
            # a PostgreSQL / SQLite literal)
            cut = _literal_cut(s1) or ("\\'" in raw and _literal_cut(_SQ_STD.sub("'s'", raw)))
        else:
            cut = _comment_cut(s1)
        if cut:
            hits.append("injection")
    return hits


def _literal_cut(s1: str) -> bool:
    """The fallback for a value the A22 tier placed in a string literal (V-18): the query's first literal is the one the
    payload closed, and what follows it up to the first comment or ';' must be nothing (admin'--, x';, x')#) or start
    with SQL (x' or name is not null--, x' and password glob password;, ' notnull --). Prose breaks out with letters:
    "It's great; works fine" leaves "s great" before the ';', which starts no SQL. The application's closing quote is
    always after the cut. Known residue: prose quoting a comment marker ("lines starting with '#' are skipped") has the
    shape of admin'#' and stays an injection."""
    m = _BREAK.search(s1)
    if m is None or m.group() != "'s'":           # the application's literal never closed: no break-out
        return False
    b = m.end()
    last_close = s1.rfind("*/")
    for t in _TERM_LIT.finditer(s1, b):
        if t.group() == "/*" and last_close >= t.end():    # a closed /* */ cuts nothing: the next one decides
            continue
        lead = s1[b:t.start()]
        return bool(_NOTHING.fullmatch(lead) or _SQL_LEAD.match(lead))
    return False


def _comment_cut(s1: str) -> bool:
    """The fallback of `classify` (V-18): a literal closed by the payload, then a comment that hides a quote after it -
    the application's own closing quote - with nothing but SQL between the break-out and the comment: admin'--,
    x')#, x' or name is not null--, 1' /*. Real SQL structure is required: a contraction or a possessive puts *letters*
    right after its quote ("It's great -- 5 stars", "I'm happy -- 5 stars"), which starts no SQL, and a comment that
    hides no quote ("Best 'smart' -- love it" in a numeric context) cuts nothing. Linear: each live stretch (from a
    literal or a lone quote to the next quote) is scanned once. Known residue: prose quoting a comment marker in a
    string context ("lines starting with '#' are skipped") has exactly the shape of admin'#' and stays an injection."""
    if "'" not in s1:
        return False
    last_q = s1.rfind("'")
    last_close = s1.rfind("*/")
    toks = list(_BREAK.finditer(s1))
    for i, m in enumerate(toks):
        b = m.end()
        e = toks[i + 1].start() if i + 1 < len(toks) else len(s1)
        t = _TERM.search(s1, b, e)
        if t is None or last_q < t.end():                       # no comment here, or it hides no quote
            continue
        if t.group() == "/*" and last_close >= t.end():           # a closed /* */ cuts nothing
            continue
        lead = s1[b:t.start()]
        if _NOTHING.fullmatch(lead) or _SQL_LEAD.match(lead):
            return True
    return False


def _is_public(ip: str) -> bool:
    try:
        a = ipaddress.ip_address(ip)
        return not (a.is_private or a.is_loopback or a.is_link_local)
    except ValueError:
        return False


_FROM = re.compile(r"\bfrom\s+((?:[`\"\[]?[\w$]+[`\"\]]?\.)?[`\"\[]?[\w$]+)", re.I)
_COLL = re.compile(r'"(?:find|aggregate|count|distinct|collection)"\s*:\s*"([\w.$-]+)"|\bdb\.([\w$]+)\.\w+\(')


def table_of(query: str, engine: str = "") -> str:
    """the first table / collection a read names ('' when none): volumes are judged per table too - a dump of one
    sensitive table through the application's own connection is a small share of the application's total rows"""
    q = str(query)[:4096]
    m = _FROM.search(q)
    if m:
        return re.sub(r'[`"\[\]]', "", m.group(1)).lower()
    m = _COLL.search(q)
    return (m.group(1) or m.group(2)).lower() if m else ""


class _Decay:
    """per-key exponentially decayed counters (bounded)"""

    def __init__(self, tau_s: float, max_keys: int = 200_000):
        self.tau, self.max = tau_s, max_keys
        self.d: "OrderedDict[object, list]" = OrderedDict()

    def add(self, key, t: float, v: float) -> float:
        st = self.d.get(key)
        if st is None:
            st = [0.0, t]
            self.d[key] = st
            if len(self.d) > self.max:
                self.d.popitem(last=False)
        else:
            self.d.move_to_end(key)
            st[0] *= math.exp(-max(0.0, t - st[1]) / self.tau)
            if t < st[1]:                       # a late (older) event: its weight at the key's clock, never a rewind
                v *= math.exp(-(st[1] - t) / self.tau)
        st[0] += v
        st[1] = max(st[1], t)
        return st[0]

    def value(self, key, t: float) -> float:
        st = self.d.get(key)
        return 0.0 if st is None else st[0] * math.exp(-max(0.0, t - st[1]) / self.tau)


class DbGuard:
    def __init__(self, net=None, alpha: float = 1e-3, tau_s: float = 300.0, auth_fail_floor: float = 8.0, exempt_shapes: set | None = None,
                 learn_new_clients: bool = False):
        self.net, self.alpha, self.tau = net, alpha, tau_s
        self.auth_fail_floor = auth_fail_floor
        self.learn_new_clients = learn_new_clients
        self.known_clients: dict = {}            # db -> clients seen in the normal workload (the application servers)
        self.known_shapes: dict = {}             # db -> query shapes the applications run
        self.vol_ref: dict = {}                  # (db, measure) -> {client: its busiest decayed-volume minute in calibration}
        self.net_ref: dict = {}                  # db -> sorted DbNet scores of that database's own normal queries
        self.vol = {"rows": _Decay(tau_s), "bytes": _Decay(tau_s)}
        # the slow dump: an hour-long horizon, so reading just under the 5-minute bar for hours still adds up
        self.vol_long = {"rows": _Decay(3600.0), "bytes": _Decay(3600.0)}
        self.vol_ref_long: dict = {}             # (db, measure) -> {client: its largest hour-decayed volume in calibration}
        # rows per (database, table, client), both horizons, and each client's largest calibrated volume per table
        self.tvol, self.tvol_long = _Decay(tau_s), _Decay(3600.0)
        self.tvol_ref: dict = {}                 # (db, table) -> {client: largest ~5-min volume in calibration}
        self.tvol_ref_long: dict = {}            # (db, table) -> {client: largest ~1-h volume in calibration}
        self.fails = _Decay(60.0)
        self.exempt_shapes = set(exempt_shapes or ())   # operator-approved shapes (e.g. a nightly TRUNCATE of a staging table)
        self.log: list = []
        self.counts = {"events": 0, "alarms": 0}

    # -- calibration on the applications' normal workload
    def calibrate(self, events: list[dict]) -> dict:
        peaks: dict = {}
        dec = {m: _Decay(self.tau) for m in self.vol}
        dec_l = {m: _Decay(3600.0) for m in self.vol}
        tdec, tdec_l = _Decay(self.tau), _Decay(3600.0)
        qs, dbs = [], []
        for e in sorted(events, key=lambda x: x["t"]):
            db = e.get("db") or e["engine"]
            self.known_clients.setdefault(db, set()).add(e["client"])
            if e.get("query"):
                self.known_shapes.setdefault(db, set()).add(shape(e["query"]))
                qs.append((e["query"], e["engine"]))
                dbs.append(db)
            for m in self.vol:
                if e.get(m):
                    r = dec[m].add((db, e["client"]), e["t"], float(e[m]))
                    k = (db, m, e["client"], int(e["t"] // 60))
                    peaks[k] = max(peaks.get(k, 0.0), r)
                    rl = dec_l[m].add((db, e["client"]), e["t"], float(e[m]))
                    refl = self.vol_ref_long.setdefault((db, m), {})
                    refl[e["client"]] = max(refl.get(e["client"], 0.0), rl)
            tb = table_of(e.get("query") or "", e["engine"]) if e.get("rows") else ""
            if tb:
                for d, ref in ((tdec, self.tvol_ref), (tdec_l, self.tvol_ref_long)):
                    v = d.add((db, tb, e["client"]), e["t"], float(e["rows"]))
                    r = ref.setdefault((db, tb), {})
                    r[e["client"]] = max(r.get(e["client"], 0.0), v)
        for (db, m, c, _), v in peaks.items():          # each client's own busiest minute (a backup host reads millions,
            ref = self.vol_ref.setdefault((db, m), {})   # an app pod thousands: one shared threshold would hide every dump)
            ref[c] = max(ref.get(c, 0.0), v)
        if self.net is not None and qs:
            s, _ = self.net.score([q for q, _ in qs], [e for _, e in qs])
            for db in set(dbs):
                self.net_ref[db] = np.sort(np.concatenate([self.net_ref.get(db, np.zeros(0)), s[np.asarray(dbs) == db]]))
        return {"databases": len(self.known_clients), "clients": {db: len(c) for db, c in self.known_clients.items()},
                "shapes": {db: len(s) for db, s in self.known_shapes.items()}, "dbnet_reference": {db: len(v) for db, v in self.net_ref.items()}}

    def _q(self, ref, alpha: float) -> float:
        k = min(len(ref) - 1, int(math.ceil((len(ref) + 1) * (1 - alpha))) - 1)
        return float(ref[k]) + 1e-5 * max(1.0, abs(float(ref[k])))     # equal up to float32 rounding is equal

    def net_threshold(self, db: str) -> float:
        """conformal: the (1-alpha) quantile of this database's own normal queries; the shipped global threshold (and
        never below it) until the database has enough of its own"""
        base = float(getattr(self.net, "thr", 0.0) or 0.0)
        ref = self.net_ref.get(db)
        a = self.alpha / 10.0                    # per query, not per source-minute: a database sees thousands a minute
        if ref is None or len(ref) < 200:
            return max(base, 2.0)
        if len(ref) < math.ceil(1 / a):          # too few for that quantile: the highest normal score is the bar
            return max(base, float(ref[-1]) + 0.5)
        return max(base, self._q(ref, a))

    def _vol_thr(self, db: str, m: str, client: str) -> float:
        """3x the client's own busiest calibrated minute; a client without history is held to a typical app client
        (the median of the known clients' busiest minutes)"""
        ref = self.vol_ref.get((db, m))
        if not ref:
            return math.inf
        floor = 1000.0 if m == "rows" else 5e6
        own = ref.get(client)
        if own is not None:
            return max(3.0 * own, floor)
        return max(3.0 * float(np.median(list(ref.values()))), floor)

    def _vol_thr_long(self, db: str, m: str, client: str) -> float:
        ref = self.vol_ref_long.get((db, m))
        if not ref:
            return math.inf
        floor = 5_000.0 if m == "rows" else 2.5e7
        own = ref.get(client)
        if own is not None:
            return max(3.0 * own, floor)
        return max(3.0 * float(np.median(list(ref.values()))), floor)

    def _peer_volume(self, db: str, m: str, client: str, t: float, long: bool = False, table: str | None = None) -> float:
        """the current median volume of this database's other active application clients (on one table, with `table`):
        a sales surge lifts every app server together; a dump makes one client stand out from its peers (0 when there
        are no peers to compare). Only the database's known clients are looked up (tens), never the whole table."""
        if table is None:
            dec, tau = (self.vol_long[m], 3600.0) if long else (self.vol[m], self.tau)
        else:
            dec, tau = (self.tvol_long, 3600.0) if long else (self.tvol, self.tau)
        cur = []
        for c in self.known_clients.get(db, ()):
            st = dec.d.get((db, c) if table is None else (db, table, c))
            if c != client and st is not None and t - st[1] < tau:
                cur.append(st[0] * math.exp(-max(0.0, t - st[1]) / tau))
        return float(np.median(cur)) if len(cur) >= 2 else 0.0

    def _table_volume(self, db: str, table: str, client: str, t: float, rows: float) -> str | None:
        """a client reading one table far beyond its own calibrated volume on that table (a paged dump of `customers`
        through the application's connection: 400 rows a minute is +13% of the app's rows, 5x its reads of that table)"""
        out = None
        for dec, refs, floor, span in ((self.tvol, self.tvol_ref, 1000.0, f"~{self.tau / 60:.0f} min"), (self.tvol_long, self.tvol_ref_long, 5000.0, "~1 h")):
            v = dec.add((db, table, client), t, rows)
            own = refs.get((db, table), {}).get(client, 0.0)
            if out is None and v > max(3.0 * own, own + floor) and v > 5.0 * self._peer_volume(db, "rows", client, t, dec is self.tvol_long, table):
                out = (f"{v:,.0f} rows of {table} read within {span}: over 3x (and {floor:,.0f} rows above) this client's largest "
                       f"normal volume on that table ({own:,.0f})")
        return out

    # -- scoring
    def score(self, events: list[dict]) -> list[dict]:
        events = sorted(events, key=lambda x: x["t"])
        ms = np.full(len(events), -np.inf)
        mf = np.full(len(events), "", dtype=object)
        qi = [i for i, e in enumerate(events) if e.get("query")]
        if self.net is not None and qi:
            s, f = self.net.score([events[i]["query"] for i in qi], [events[i]["engine"] for i in qi])
            ms[qi], mf[qi] = s, f
        out = []
        for i, e in enumerate(events):
            self.counts["events"] += 1
            db = e.get("db") or e["engine"]
            fams, evidence, learned = [], [], False
            q = e.get("query") or ""
            shp = shape(q) if q else ""
            known_shape = shp in self.known_shapes.get(db, set())
            if q and shp not in self.exempt_shapes:
                hits = classify(q, e["engine"])
                # operator injection by shape: the application sends a plain value for this field; an {"$op": ...}
                # object in its place (the classic {"$ne": null}, {"$gt": ""}, {"$regex": ".*"}) changes the query's
                # shape into one the application never runs - whatever the field is called
                if e["engine"] in ("mongodb", "couchdb") and not known_shape and "injection" not in hits:
                    red = shape(_MONGO_OP.sub('"?"', q))
                    if red != shp and red in self.known_shapes.get(db, set()):
                        hits.append("injection")
                if hits:
                    evidence.append("signature: " + ", ".join(hits) + " | " + q[:240])
                thr = self.net_threshold(db)
                if ms[i] > thr and mf[i] not in hits:
                    learned = True
                    hits.append(mf[i])
                    evidence.append(f"DbNet: {mf[i]} (score {ms[i]:.1f} > threshold {thr:.1f})" + ("" if evidence else " | " + q[:240]))
                if known_shape:          # a structure the applications themselves run: not injected, not a recon / dump
                    hits = [h for h in hits if h not in SOFT]
                fams += hits
            status = e.get("status", "ok")
            if status == "auth_fail":
                f = self.fails.add((db, e["client"]), e["t"], 1.0)
                if f >= self.auth_fail_floor:
                    fams.append("bruteforce")
                    evidence.append(f"{f:.0f} failed logins within ~1 min (user {e.get('user', '?')})")
            for m in self.vol:
                if e.get(m):
                    r = self.vol[m].add((db, e["client"]), e["t"], float(e[m]))
                    rl = self.vol_long[m].add((db, e["client"]), e["t"], float(e[m]))
                    if r > self._vol_thr(db, m, e["client"]) and r > 5.0 * self._peer_volume(db, m, e["client"], e["t"]):
                        fams.append("exfil")
                        evidence.append(f"{r:,.0f} {m} read within ~{self.tau / 60:.0f} min: over 3x this client's (or a typical app client's) busiest normal minute on {db}")
                    elif rl > self._vol_thr_long(db, m, e["client"]) and rl > 5.0 * self._peer_volume(db, m, e["client"], e["t"], long=True):
                        fams.append("exfil")
                        evidence.append(f"{rl:,.0f} {m} read within ~1 h: over 3x this client's (or a typical app client's) largest normal hour on {db} (slow dump)")
            tb = table_of(q, e["engine"]) if q and e.get("rows") else ""
            if tb and self.tvol_ref:
                why = self._table_volume(db, tb, e["client"], e["t"], float(e["rows"]))
                if why and "exfil" not in fams:
                    fams.append("exfil")
                    evidence.append(why)
            known = self.known_clients.get(db)
            app_client = known is not None and e["client"] in known
            new_client = known is not None and not app_client
            public = _is_public(e["client"])
            if new_client and status != "auth_fail":
                fams.append("new_client")
                evidence.append(f"client {e['client']} never talked to {db} in the normal workload" + (" (public address)" if public else ""))
                if self.learn_new_clients and not (set(fams) - {"new_client"}):
                    known.add(e["client"])
            fams = sorted(set(fams), key=lambda x: -SEVERITY.get(x, 0.5))
            serious = [x for x in fams if x not in ("new_client", "unbounded")]
            alarm = bool(serious) or (new_client and public)
            sev = SEVERITY.get(fams[0], 0.5) if fams else 0.0
            if learned and fams and fams[0] == mf[i] and len(serious) == 1:
                sev *= 0.8                                   # the model alone: strong, but one notch below a named signature
            if new_client and public and serious:
                sev = 1.0                                    # an unknown internet host doing anything serious to a database
            v = {"t": e["t"], "engine": e["engine"], "db": db, "client": e["client"], "user": e.get("user"), "alarm": alarm,
                 "families": fams, "severity": round(sev, 2), "score": int(round(100 * sev)) if alarm else 0, "app_client": app_client,
                 "public": public, "learned_only": learned and len(serious) == 1 and serious[0] == mf[i], "evidence": evidence,
                 "dbnet": round(float(ms[i]), 2) if np.isfinite(ms[i]) else None,
                 "remediation": REMEDIATION.get(e["engine"], "").format(client=e["client"]) if alarm else ""}
            out.append(v)
            if alarm:
                self.counts["alarms"] += 1
                self.log.append(v)
                if len(self.log) > 50_000:
                    del self.log[:10_000]
        return out


# ---------------------------------------------------------------- log adapters (one event dict per line)
_PG = re.compile(r"(?P<ts>\d{4}-\d\d-\d\d[ T]\d\d:\d\d:\d\d(?:\.\d+)?)\S*\s.*?(?:host=|client=|@)?(?P<client>\d{1,3}(?:\.\d{1,3}){3}).*?"
                 r"(?:(?P<fail>password authentication failed|no pg_hba\.conf entry)|(?:statement|execute \S+):\s*(?P<q>.*))", re.I)


def _ts(s: str) -> float:
    import datetime as dt
    s = s.replace("T", " ").split("+")[0].rstrip("Z")
    for fmt in ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S"):
        try:
            return dt.datetime.strptime(s, fmt).replace(tzinfo=dt.timezone.utc).timestamp()
        except ValueError:
            continue
    return time.time()


def parse_postgres(line: str) -> dict | None:
    """postgres stderr/csv log with log_line_prefix including %h (client), log_statement and auth failures"""
    m = _PG.search(line)
    if not m:
        return None
    user = re.search(r"user[= ]\"?(\w+)", line)
    return {"t": _ts(m.group("ts")), "engine": "postgres", "client": m.group("client"), "user": user.group(1) if user else None,
            "query": m.group("q") or "", "status": "auth_fail" if m.group("fail") else "ok"}


_MYSQL_THREADS: "OrderedDict[str, tuple]" = OrderedDict()     # general-log thread id -> (user, client) from its Connect line


def parse_mysql(line: str) -> dict | None:
    """MySQL general log (Connect / Query / Execute, the thread's client remembered from its Connect line) and
    'Access denied for user' lines"""
    m = re.search(r"(?P<ts>\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(?:\.\d+)?)Z?\s+(?P<tid>\d+)\s+(?P<cmd>Connect|Query|Execute|Quit)\s*(?P<rest>.*)", line)
    if m:
        rest, tid = m.group("rest"), m.group("tid")
        if m.group("cmd") == "Connect":
            c = re.search(r"'(\w+)'@'([\w.:-]+)'", rest) or re.search(r"(\w+)@([\w.:-]+)", rest)
            user, client = (c.group(1), c.group(2)) if c else (None, "?")
            if "Access denied" not in rest:
                _MYSQL_THREADS[tid] = (user, client)
                if len(_MYSQL_THREADS) > 100_000:
                    _MYSQL_THREADS.popitem(last=False)
            return {"t": _ts(m.group("ts")), "engine": "mysql", "client": client, "user": user, "query": "",
                    "status": "auth_fail" if "Access denied" in rest else "ok"}
        if m.group("cmd") == "Quit":
            _MYSQL_THREADS.pop(tid, None)
            return None
        user, client = _MYSQL_THREADS.get(tid, (None, "?"))
        return {"t": _ts(m.group("ts")), "engine": "mysql", "client": client, "user": user, "query": rest, "status": "ok"}
    d = re.search(r"Access denied for user '(\w+)'@'([\d.]+)'", line)
    if d:
        return {"t": time.time(), "engine": "mysql", "client": d.group(2), "user": d.group(1), "query": "", "status": "auth_fail"}
    return None


def parse_mongo(line: str) -> dict | None:
    """MongoDB 4.4+ structured JSON log: COMMAND / slow query entries and 'Authentication failed'"""
    try:
        j = json.loads(line)
    except ValueError:
        return None
    attr = j.get("attr", {}) or {}
    t = j.get("t", {})
    ts = _ts(t.get("$date", "")) if isinstance(t, dict) else time.time()
    remote = str(attr.get("remote") or attr.get("client") or "?").rsplit(":", 1)[0]
    if "Authentication failed" in str(j.get("msg", "")):
        u = attr.get("user")
        return {"t": ts, "engine": "mongodb", "client": remote, "user": u.get("user") if isinstance(u, dict) else u, "query": "", "status": "auth_fail"}
    cmd = attr.get("command")
    if cmd is None:
        return None
    return {"t": ts, "engine": "mongodb", "client": remote, "user": None, "db": (attr.get("ns", "").split(".")[0] or None),
            "query": json.dumps(cmd, separators=(",", ":")), "status": "ok", "rows": int(attr.get("nreturned", 0) or 0)}


def parse_cassandra(line: str) -> dict | None:
    """Cassandra 4 audit log: user:..|host:..|source:/ip|...|operation:<cql>"""
    if "operation:" not in line and "type:LOGIN_ERROR" not in line:
        return None
    src = re.search(r"source:/?([\d.]+)", line)
    user = re.search(r"user:(\w+)", line)
    ts = re.search(r"timestamp:(\d{10,13})", line)
    t = (int(ts.group(1)) / (1000 if len(ts.group(1)) == 13 else 1)) if ts else time.time()
    fail = "type:LOGIN_ERROR" in line or "type:UNAUTHORIZED_ATTEMPT" in line
    return {"t": t, "engine": "cassandra", "client": src.group(1) if src else "?", "user": user.group(1) if user else None,
            "query": "" if fail or "operation:" not in line else line.split("operation:", 1)[1].strip(), "status": "auth_fail" if fail else "ok"}


def parse_redis(line: str) -> dict | None:
    """redis-cli MONITOR line: 1700000000.123456 [0 1.2.3.4:5555] "CONFIG" "SET" "dir" "/var/www" """
    m = re.match(r'\s*(\d+\.\d+)\s+\[\d+\s+([\d.]+):\d+\]\s+(.*)', line)
    if not m:
        return None
    parts = re.findall(r'"((?:[^"\\]|\\.)*)"', m.group(3))
    return {"t": float(m.group(1)), "engine": "redis", "client": m.group(2), "user": None, "query": " ".join(parts), "status": "ok"}


def parse_es(line: str) -> dict | None:
    """Elasticsearch audit log (JSON): request.method + url.path (+ body when enabled)"""
    try:
        j = json.loads(line)
    except ValueError:
        return None
    method = j.get("request.method") or (j.get("request") or {}).get("method")
    path = j.get("url.path") or (j.get("url") or {}).get("path")
    if not path:
        return None
    ip = j["origin.address"].rsplit(":", 1)[0] if isinstance(j.get("origin.address"), str) else "?"
    fail = j.get("event.action") in ("authentication_failed", "access_denied")
    ts = j.get("@timestamp") or j.get("timestamp")
    return {"t": _ts(ts) if isinstance(ts, str) else time.time(), "engine": "elasticsearch", "client": ip, "user": j.get("user.name"),
            "query": f"{(method or 'get').lower()} {path} {j.get('request.body', '')}".strip(), "status": "auth_fail" if fail else "ok"}


PARSERS = {"postgres": parse_postgres, "mysql": parse_mysql, "mongodb": parse_mongo, "cassandra": parse_cassandra, "redis": parse_redis,
           "elasticsearch": parse_es}