-- ============================================================================ -- ANUBIS CRAWLER RULES — Labeling des http_logs + pipeline ML -- Architecture : -- anubis_ua_rules (table) → dict_anubis_ua (REGEXP_TREE) -- anubis_ip_rules (table) → dict_anubis_ip (IP_TRIE) -- http_logs : +anubis_bot_name, +anubis_bot_action -- mv_http_logs : reconstruit avec enrichissement Anubis -- view_ai_features_1h : +anubis_bot_name, +anubis_bot_action (via dictGet) -- ml_detected_anomalies : +anubis_bot_name, +anubis_bot_action -- ml_all_scores : +anubis_bot_name, +anubis_bot_action -- ============================================================================ -- ---------------------------------------------------------------------------- -- 1. TABLE SOURCE — règles User-Agent (pour dictionnaire REGEXP_TREE) -- -- Format attendu par ClickHouse regexp_tree (v23.5+) : -- id UInt64 : identifiant unique -- parent_id UInt64 : 0 = racine, sinon id du parent (héritage d'attributs) -- regexp String : expression régulière (re2/vectorscan) -- keys Array(String) : noms des attributs, ex. ['bot_name', 'action'] -- values Array(String) : valeurs correspondantes -- -- Hiérarchie utilisée pour la priorité : -- Règles génériques DENY (parent_id=0) → enfants ALLOW spécifiques -- Exemple : ai-crawlers-training (parent) → openai-gptbot (enfant) -- Quand l'UA correspond à enfant ET parent, c'est le nom de l'enfant qui -- est retourné (l'enfant hérite ET surcharge les attributs du parent). -- ---------------------------------------------------------------------------- CREATE TABLE IF NOT EXISTS ja4_processing.anubis_ua_rules ( id UInt64, parent_id UInt64, regexp String, keys Array(String), values Array(String) ) ENGINE = ReplacingMergeTree() ORDER BY id; -- ---------------------------------------------------------------------------- -- 2. TABLE SOURCE — règles IP/CIDR (pour dictionnaire IP_TRIE) -- -- Colonnes requises par dict_anubis_ip et mv_http_logs : -- rule_id : identifiant de règle, croisé avec dict_anubis_ua pour -- la logique UA+IP (même rule_id → match combiné) -- has_ua : 1 si la règle possède aussi une regex UA (croisement nécessaire) -- category : catégorie Anubis (bots, crawlers, clients, policies…) -- ---------------------------------------------------------------------------- CREATE TABLE IF NOT EXISTS ja4_processing.anubis_ip_rules ( prefix String, bot_name LowCardinality(String), action LowCardinality(String), rule_id UInt64, has_ua UInt8, category LowCardinality(String) ) ENGINE = ReplacingMergeTree() ORDER BY prefix; -- ---------------------------------------------------------------------------- -- 3. DICTIONNAIRE UA — REGEXP_TREE -- dictGet('ja4_processing.dict_anubis_ua', 'bot_name', header_user_agent) -- -- Le PRIMARY KEY est 'regexp' (String) — requis par ClickHouse 26.x. -- Connexion interne (HOST localhost PORT 9000) pour éviter deadlock HTTP. -- Remplacer 'admin' et le mot de passe par les credentials ClickHouse. -- ---------------------------------------------------------------------------- DROP DICTIONARY IF EXISTS ja4_processing.dict_anubis_ua; CREATE DICTIONARY ja4_processing.dict_anubis_ua ( regexp String, bot_name String, action String ) PRIMARY KEY regexp SOURCE(CLICKHOUSE(HOST 'localhost' PORT 9000 USER 'admin' PASSWORD 'CHANGE_ME' DB 'ja4_processing' TABLE 'anubis_ua_rules')) LAYOUT(REGEXP_TREE) LIFETIME(MIN 300 MAX 600); -- ---------------------------------------------------------------------------- -- 4. DICTIONNAIRE IP — IP_TRIE -- dictGetOrDefault('ja4_processing.dict_anubis_ip', 'bot_name', toIPv6(src_ip), '') -- Connexion interne (HOST localhost PORT 9000) — même raison que dict_anubis_ua. -- ---------------------------------------------------------------------------- DROP DICTIONARY IF EXISTS ja4_processing.dict_anubis_ip; CREATE DICTIONARY ja4_processing.dict_anubis_ip ( prefix String, bot_name String, action String, rule_id UInt64, has_ua UInt8, category String ) PRIMARY KEY prefix SOURCE(CLICKHOUSE(HOST 'localhost' PORT 9000 USER 'admin' PASSWORD 'CHANGE_ME' DB 'ja4_processing' TABLE 'anubis_ip_rules')) LAYOUT(IP_TRIE()) LIFETIME(MIN 300 MAX 600); -- ---------------------------------------------------------------------------- -- 5. TABLE SOURCE — règles ASN (pour dictionnaire Flat) -- Alimentée par botPolicies.yaml via fetch_rules.py → insert_asn_rules() -- ---------------------------------------------------------------------------- CREATE TABLE IF NOT EXISTS ja4_processing.anubis_asn_rules ( asn UInt32, bot_name LowCardinality(String), action LowCardinality(String), category LowCardinality(String) ) ENGINE = ReplacingMergeTree() ORDER BY asn; -- ---------------------------------------------------------------------------- -- 6. TABLE SOURCE — règles pays ISO-3166 (pour dictionnaire Flat) -- Alimentée par botPolicies.yaml via fetch_rules.py → insert_country_rules() -- ---------------------------------------------------------------------------- CREATE TABLE IF NOT EXISTS ja4_processing.anubis_country_rules ( country_code LowCardinality(String), bot_name LowCardinality(String), action LowCardinality(String), category LowCardinality(String) ) ENGINE = ReplacingMergeTree() ORDER BY country_code; -- ---------------------------------------------------------------------------- -- 7. DICTIONNAIRE ASN — Flat -- dictGetOrDefault('ja4_processing.dict_anubis_asn', 'bot_name', src_asn, '') -- ---------------------------------------------------------------------------- DROP DICTIONARY IF EXISTS ja4_processing.dict_anubis_asn; CREATE DICTIONARY ja4_processing.dict_anubis_asn ( asn UInt32, bot_name String, action String, category String ) PRIMARY KEY asn SOURCE(CLICKHOUSE(HOST 'localhost' PORT 9000 USER 'admin' PASSWORD 'CHANGE_ME' DB 'ja4_processing' TABLE 'anubis_asn_rules')) LAYOUT(FLAT()) LIFETIME(MIN 300 MAX 600); -- ---------------------------------------------------------------------------- -- 8. DICTIONNAIRE PAYS — Flat -- dictGetOrDefault('ja4_processing.dict_anubis_country', 'bot_name', src_country_code, '') -- ---------------------------------------------------------------------------- DROP DICTIONARY IF EXISTS ja4_processing.dict_anubis_country; CREATE DICTIONARY ja4_processing.dict_anubis_country ( country_code String, bot_name String, action String, category String ) PRIMARY KEY country_code SOURCE(CLICKHOUSE(HOST 'localhost' PORT 9000 USER 'admin' PASSWORD 'CHANGE_ME' DB 'ja4_processing' TABLE 'anubis_country_rules')) LAYOUT(FLAT()) LIFETIME(MIN 300 MAX 600); -- ---------------------------------------------------------------------------- -- 9. AJOUT DES COLONNES ANUBIS dans http_logs -- Idempotent : ne plante pas si déjà présentes -- ---------------------------------------------------------------------------- ALTER TABLE ja4_logs.http_logs ADD COLUMN IF NOT EXISTS anubis_bot_name LowCardinality(String) DEFAULT '', ADD COLUMN IF NOT EXISTS anubis_bot_action LowCardinality(String) DEFAULT '', ADD COLUMN IF NOT EXISTS anubis_bot_category LowCardinality(String) DEFAULT ''; -- ---------------------------------------------------------------------------- -- 10. RECONSTRUCTION DE mv_http_logs avec enrichissement Anubis -- Logique de priorisation : -- 1. UA regex (plus informatif — identifie le bot précis) -- 2. IP/CIDR (fallback — identifie le réseau cloud) -- ---------------------------------------------------------------------------- DROP VIEW IF EXISTS ja4_logs.mv_http_logs; CREATE MATERIALIZED VIEW ja4_logs.mv_http_logs TO ja4_logs.http_logs ( `time` DateTime, `log_date` Date, `src_ip` IPv4, `src_port` UInt16, `src_asn` UInt32, `src_country_code` String, `dst_ip` IPv4, `dst_port` UInt16, `src_as_name` String, `src_org` String, `src_domain` String, `method` String, `scheme` String, `host` String, `path` String, `query` String, `http_version` String, `orphan_side` String, `correlated` UInt8, `keepalives` UInt16, `a_timestamp` UInt64, `b_timestamp` UInt64, `conn_id` String, `ip_meta_df` UInt8, `ip_meta_id` UInt16, `ip_meta_total_length` UInt16, `ip_meta_ttl` UInt8, `tcp_meta_options` String, `tcp_meta_window_size` UInt32, `tcp_meta_mss` UInt16, `tcp_meta_window_scale` UInt8, `syn_to_clienthello_ms` Int32, `tls_version` String, `tls_sni` String, `tls_alpn` String, `ja3` String, `ja3_hash` String, `ja4` String, `client_headers` String, `header_user_agent` String, `header_accept` String, `header_accept_encoding` String, `header_accept_language` String, `header_content_type` String, `header_x_request_id` String, `header_x_trace_id` String, `header_x_forwarded_for` String, `header_sec_ch_ua` String, `header_sec_ch_ua_mobile` String, `header_sec_ch_ua_platform` String, `header_sec_fetch_dest` String, `header_sec_fetch_mode` String, `header_sec_fetch_site` String, `anubis_bot_name` String, `anubis_bot_action` String ) AS SELECT parseDateTimeBestEffort(coalesce(JSONExtractString(raw_json, 'time'), '1970-01-01T00:00:00Z')) AS time, toDate(time) AS log_date, toIPv4(coalesce(JSONExtractString(raw_json, 'src_ip'), '0.0.0.0')) AS src_ip, toUInt16(coalesce(JSONExtractUInt(raw_json, 'src_port'), 0)) AS src_port, dictGetOrDefault('ja4_processing.dict_iplocate_asn', 'asn', toIPv6(src_ip), toUInt32(0)) AS src_asn, dictGetOrDefault('ja4_processing.dict_iplocate_asn', 'country_code', toIPv6(src_ip), '') AS src_country_code, toIPv4(coalesce(JSONExtractString(raw_json, 'dst_ip'), '0.0.0.0')) AS dst_ip, toUInt16(coalesce(JSONExtractUInt(raw_json, 'dst_port'), 0)) AS dst_port, dictGetOrDefault('ja4_processing.dict_iplocate_asn', 'name', toIPv6(src_ip), '') AS src_as_name, '' AS src_org, '' AS src_domain, coalesce(JSONExtractString(raw_json, 'method'), '') AS method, coalesce(JSONExtractString(raw_json, 'scheme'), '') AS scheme, coalesce(JSONExtractString(raw_json, 'host'), '') AS host, coalesce(JSONExtractString(raw_json, 'path'), '') AS path, coalesce(JSONExtractString(raw_json, 'query'), '') AS query, coalesce(JSONExtractString(raw_json, 'http_version'), '') AS http_version, coalesce(JSONExtractString(raw_json, 'orphan_side'), '') AS orphan_side, toUInt8(coalesce(JSONExtractBool(raw_json, 'correlated'), 0)) AS correlated, toUInt16(coalesce(JSONExtractUInt(raw_json, 'keepalives'), 0)) AS keepalives, coalesce(JSONExtractUInt(raw_json, 'a_timestamp'), 0) AS a_timestamp, coalesce(JSONExtractUInt(raw_json, 'b_timestamp'), 0) AS b_timestamp, coalesce(JSONExtractString(raw_json, 'conn_id'), '') AS conn_id, toUInt8(coalesce(JSONExtractBool(raw_json, 'ip_meta_df'), 0)) AS ip_meta_df, toUInt16(coalesce(JSONExtractUInt(raw_json, 'ip_meta_id'), 0)) AS ip_meta_id, toUInt16(coalesce(JSONExtractUInt(raw_json, 'ip_meta_total_length'), 0)) AS ip_meta_total_length, toUInt8(coalesce(JSONExtractUInt(raw_json, 'ip_meta_ttl'), 0)) AS ip_meta_ttl, coalesce(JSONExtractString(raw_json, 'tcp_meta_options'), '') AS tcp_meta_options, toUInt32(coalesce(JSONExtractUInt(raw_json, 'tcp_meta_window_size'), 0)) AS tcp_meta_window_size, toUInt16(coalesce(JSONExtractUInt(raw_json, 'tcp_meta_mss'), 0)) AS tcp_meta_mss, toUInt8(coalesce(JSONExtractUInt(raw_json, 'tcp_meta_window_scale'), 0)) AS tcp_meta_window_scale, toInt32(coalesce(JSONExtractInt(raw_json, 'syn_to_clienthello_ms'), 0)) AS syn_to_clienthello_ms, coalesce(JSONExtractString(raw_json, 'tls_version'), '') AS tls_version, coalesce(JSONExtractString(raw_json, 'tls_sni'), '') AS tls_sni, coalesce(JSONExtractString(raw_json, 'tls_alpn'), '') AS tls_alpn, coalesce(JSONExtractString(raw_json, 'ja3'), '') AS ja3, coalesce(JSONExtractString(raw_json, 'ja3_hash'), '') AS ja3_hash, coalesce(JSONExtractString(raw_json, 'ja4'), '') AS ja4, coalesce(JSONExtractString(raw_json, 'client_headers'), '') AS client_headers, coalesce(JSONExtractString(raw_json, 'header_User-Agent'), '') AS header_user_agent, coalesce(JSONExtractString(raw_json, 'header_Accept'), '') AS header_accept, coalesce(JSONExtractString(raw_json, 'header_Accept-Encoding'), '') AS header_accept_encoding, coalesce(JSONExtractString(raw_json, 'header_Accept-Language'), '') AS header_accept_language, coalesce(JSONExtractString(raw_json, 'header_Content-Type'), '') AS header_content_type, coalesce(JSONExtractString(raw_json, 'header_X-Request-Id'), '') AS header_x_request_id, coalesce(JSONExtractString(raw_json, 'header_X-Trace-Id'), '') AS header_x_trace_id, coalesce(JSONExtractString(raw_json, 'header_X-Forwarded-For'), '') AS header_x_forwarded_for, coalesce(JSONExtractString(raw_json, 'header_Sec-CH-UA'), '') AS header_sec_ch_ua, coalesce(JSONExtractString(raw_json, 'header_Sec-CH-UA-Mobile'), '') AS header_sec_ch_ua_mobile, coalesce(JSONExtractString(raw_json, 'header_Sec-CH-UA-Platform'), '') AS header_sec_ch_ua_platform, coalesce(JSONExtractString(raw_json, 'header_Sec-Fetch-Dest'), '') AS header_sec_fetch_dest, coalesce(JSONExtractString(raw_json, 'header_Sec-Fetch-Mode'), '') AS header_sec_fetch_mode, coalesce(JSONExtractString(raw_json, 'header_Sec-Fetch-Site'), '') AS header_sec_fetch_site, -- ── Enrichissement Anubis ──────────────────────────────────────────────── -- Priorité : UA regex > IP/CIDR (UA identifie précisément le bot) COALESCE( nullIf(dictGet('ja4_processing.dict_anubis_ua', 'bot_name', coalesce(JSONExtractString(raw_json, 'header_User-Agent'), '')), ''), nullIf(dictGetOrDefault('ja4_processing.dict_anubis_ip', 'bot_name', toIPv6(toIPv4(coalesce(JSONExtractString(raw_json, 'src_ip'), '0.0.0.0'))), ''), ''), '' ) AS anubis_bot_name, COALESCE( nullIf(dictGet('ja4_processing.dict_anubis_ua', 'action', coalesce(JSONExtractString(raw_json, 'header_User-Agent'), '')), ''), nullIf(dictGetOrDefault('ja4_processing.dict_anubis_ip', 'action', toIPv6(toIPv4(coalesce(JSONExtractString(raw_json, 'src_ip'), '0.0.0.0'))), ''), ''), '' ) AS anubis_bot_action FROM ja4_logs.http_logs_raw; -- ============================================================================ -- INTÉGRATION ML — Propagation Anubis vers le pipeline bot_detector -- ============================================================================ -- ---------------------------------------------------------------------------- -- 11. COLONNES ANUBIS dans ml_detected_anomalies -- ---------------------------------------------------------------------------- ALTER TABLE ja4_processing.ml_detected_anomalies ADD COLUMN IF NOT EXISTS anubis_bot_name LowCardinality(String) DEFAULT '', ADD COLUMN IF NOT EXISTS anubis_bot_action LowCardinality(String) DEFAULT '', ADD COLUMN IF NOT EXISTS anubis_bot_category LowCardinality(String) DEFAULT ''; -- ---------------------------------------------------------------------------- -- 12. COLONNES ANUBIS dans ml_all_scores -- ---------------------------------------------------------------------------- ALTER TABLE ja4_processing.ml_all_scores ADD COLUMN IF NOT EXISTS anubis_bot_name LowCardinality(String) DEFAULT '', ADD COLUMN IF NOT EXISTS anubis_bot_action LowCardinality(String) DEFAULT '', ADD COLUMN IF NOT EXISTS anubis_bot_category LowCardinality(String) DEFAULT ''; -- ---------------------------------------------------------------------------- -- 13. VIEW view_ai_features_1h — Enrichissement Anubis -- Ajoute anubis_bot_name et anubis_bot_action via dictGet. -- Priorité : UA regex (first_ua → dict_anubis_ua) > IP/CIDR (src_ip → dict_anubis_ip) -- Voir le fichier complet dans /tmp/update_view_ai_features.sql ou recréer -- avec CREATE OR REPLACE VIEW après avoir appliqué les étapes précédentes. -- ---------------------------------------------------------------------------- -- NOTE : Exécuter le contenu de /tmp/update_view_ai_features.sql ici (trop long). -- Ou lancer depuis le repo : psql -f bot_detector/anubis/view_ai_features_anubis.sql