| """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 |
|
|
| |
| 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"} |
|
|
| _F = re.I | re.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)['\"]" |
| 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+" |
| r"|=\s*\(\s*select\s*[\d(]" |
| r"|\b(and|or|xor|&&)\s+\d+\s*[=<>]+\s*\d+\s*(--|#|;|$)" |
| r"|\band\s+'s'\s*=\s*'s'\s*;?\s*$" |
| |
| |
| 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" |
| |
| |
| 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), |
| } |
| |
| 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*\]" |
| |
| |
| r"|\"(user(name)?|login|e-?mail)\"\s*:\s*\{\s*\"\$(ne\"\s*:\s*(null|\"\")\s*\}" |
| r"|gte?\"\s*:\s*\"\"\s*\}|regex\"\s*:\s*\"(\^|\.\*))", flags=_F), |
| } |
| |
| _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"} |
| LANG_KIND = {"sql": "sql", "cql": "sql", "graph": "sql"} |
|
|
| |
| 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*:)?") |
|
|
|
|
| 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]") |
| _LC_HASH = re.compile(r"(?:#|--)[^\r\n]*[\r\n]") |
| _HASH_COMMENT = ("", "mysql", "clickhouse") |
| |
| |
| |
| _BREAK = re.compile(r"'s'|'") |
| _TERM = re.compile(r"--|#|/\*") |
| _TERM_LIT = re.compile(r"--|#|/\*|;") |
| |
| |
| _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) |
| 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: |
| raw = raw[:768] + " " + raw[-256:] |
| 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)) |
| |
| |
| 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): |
| 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"): |
| |
| 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: |
| |
| |
| 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'": |
| 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(): |
| 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(): |
| continue |
| if t.group() == "/*" and last_close >= t.end(): |
| 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]: |
| 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 = {} |
| self.known_shapes: dict = {} |
| self.vol_ref: dict = {} |
| self.net_ref: dict = {} |
| self.vol = {"rows": _Decay(tau_s), "bytes": _Decay(tau_s)} |
| |
| self.vol_long = {"rows": _Decay(3600.0), "bytes": _Decay(3600.0)} |
| self.vol_ref_long: dict = {} |
| |
| self.tvol, self.tvol_long = _Decay(tau_s), _Decay(3600.0) |
| self.tvol_ref: dict = {} |
| self.tvol_ref_long: dict = {} |
| self.fails = _Decay(60.0) |
| self.exempt_shapes = set(exempt_shapes or ()) |
| self.log: list = [] |
| self.counts = {"events": 0, "alarms": 0} |
|
|
| |
| 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(): |
| ref = self.vol_ref.setdefault((db, m), {}) |
| 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]))) |
|
|
| 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 |
| if ref is None or len(ref) < 200: |
| return max(base, 2.0) |
| if len(ref) < math.ceil(1 / a): |
| 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 |
|
|
| |
| 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"]) |
| |
| |
| |
| 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: |
| 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 |
| if new_client and public and serious: |
| sev = 1.0 |
| 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 |
|
|
|
|
| |
| _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() |
|
|
|
|
| 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} |
|
|
