docs: add sql/init.sql + update README ClickHouse schema
- feat: sql/init.sql — initialisation complète ClickHouse (db, tables, MV, users) - feat: table http_logs mise à jour avec tous les champs réels du JSON corrélé - ajout tcp_meta_mss (UInt16), tcp_meta_window_scale (UInt8), tls_alpn (LowCardinality) - ajout keepalives, a_timestamp, b_timestamp, conn_id, ip_meta_id, ip_meta_total_length - ajout tous les header_* manquants (x_request_id, x_trace_id, sec_fetch_*, etc.) - correction types: ip_meta_id/ip_meta_total_length UInt32 → UInt16 - feat: vue matérialisée complète avec coalesce() sur tous les champs - docs: README schema section remplacée par référence à sql/init.sql + tableau des colonnes Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
toto
141
README.md
141
README.md
@ -198,103 +198,44 @@ Les orphelins A (sans B correspondant) sont émis avec `"correlated": false, "or
|
||||
|
||||
## Schema ClickHouse
|
||||
|
||||
### Setup complet
|
||||
Le fichier `sql/init.sql` contient le schéma complet prêt à l'emploi.
|
||||
|
||||
```sql
|
||||
-- Base de données
|
||||
CREATE DATABASE IF NOT EXISTS mabase_prod;
|
||||
|
||||
-- Table brute (cible des inserts du service)
|
||||
CREATE TABLE mabase_prod.http_logs_raw
|
||||
(
|
||||
raw_json String,
|
||||
ingest_time DateTime DEFAULT now()
|
||||
)
|
||||
ENGINE = MergeTree
|
||||
PARTITION BY toDate(ingest_time)
|
||||
ORDER BY ingest_time;
|
||||
|
||||
-- Table parsée
|
||||
CREATE TABLE mabase_prod.http_logs
|
||||
(
|
||||
time DateTime,
|
||||
log_date Date DEFAULT toDate(time),
|
||||
src_ip IPv4, src_port UInt16,
|
||||
dst_ip IPv4, dst_port UInt16,
|
||||
method LowCardinality(String),
|
||||
scheme LowCardinality(String),
|
||||
host LowCardinality(String),
|
||||
path String, query String,
|
||||
http_version LowCardinality(String),
|
||||
orphan_side LowCardinality(String),
|
||||
correlated UInt8,
|
||||
tls_version LowCardinality(String),
|
||||
tls_sni LowCardinality(String),
|
||||
ja3 String, ja3_hash String, ja4 String,
|
||||
header_user_agent String, header_accept String,
|
||||
header_accept_encoding String, header_accept_language 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,
|
||||
ip_meta_ttl UInt8, ip_meta_df UInt8,
|
||||
tcp_meta_window_size UInt32, tcp_meta_options LowCardinality(String),
|
||||
syn_to_clienthello_ms Int32
|
||||
)
|
||||
ENGINE = MergeTree
|
||||
PARTITION BY log_date
|
||||
ORDER BY (time, src_ip, dst_ip, ja4);
|
||||
|
||||
-- Vue matérialisée RAW → http_logs
|
||||
CREATE MATERIALIZED VIEW mabase_prod.mv_http_logs
|
||||
TO mabase_prod.http_logs 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(JSONExtractUInt(raw_json,'src_port')) AS src_port,
|
||||
toIPv4(coalesce(JSONExtractString(raw_json,'dst_ip'),'0.0.0.0')) AS dst_ip,
|
||||
toUInt16(JSONExtractUInt(raw_json,'dst_port')) AS dst_port,
|
||||
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(JSONExtractBool(raw_json,'correlated')) AS correlated,
|
||||
coalesce(JSONExtractString(raw_json,'tls_version'),'') AS tls_version,
|
||||
coalesce(JSONExtractString(raw_json,'tls_sni'),'') AS tls_sni,
|
||||
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,'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_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,
|
||||
toUInt8(JSONExtractUInt(raw_json,'ip_meta_ttl')) AS ip_meta_ttl,
|
||||
toUInt8(JSONExtractBool(raw_json,'ip_meta_df')) AS ip_meta_df,
|
||||
toUInt32(JSONExtractUInt(raw_json,'tcp_meta_window_size')) AS tcp_meta_window_size,
|
||||
coalesce(JSONExtractString(raw_json,'tcp_meta_options'),'') AS tcp_meta_options,
|
||||
toInt32(JSONExtractInt(raw_json,'syn_to_clienthello_ms')) AS syn_to_clienthello_ms
|
||||
FROM mabase_prod.http_logs_raw;
|
||||
```bash
|
||||
clickhouse-client --multiquery < sql/init.sql
|
||||
```
|
||||
|
||||
### Architecture des tables
|
||||
|
||||
```
|
||||
http_logs_raw ← inserts du service (raw_json String)
|
||||
│
|
||||
└─ mv_http_logs ← vue matérialisée (parse JSON → colonnes typées)
|
||||
│
|
||||
▼
|
||||
http_logs ← table requêtable par les analystes
|
||||
```
|
||||
|
||||
### Table `http_logs` — colonnes
|
||||
|
||||
| Groupe | Colonnes |
|
||||
|---|---|
|
||||
| Temporel | `time` DateTime, `log_date` Date |
|
||||
| Réseau | `src_ip` IPv4, `src_port` UInt16, `dst_ip` IPv4, `dst_port` UInt16 |
|
||||
| HTTP | `method`, `scheme`, `host`, `path`, `query`, `http_version` (LowCardinality) |
|
||||
| Corrélation | `orphan_side`, `correlated` UInt8, `keepalives` UInt16, `a_timestamp`/`b_timestamp` UInt64, `conn_id` |
|
||||
| IP meta | `ip_meta_df` UInt8, `ip_meta_id` UInt16, `ip_meta_total_length` UInt16, `ip_meta_ttl` UInt8 |
|
||||
| TCP meta | `tcp_meta_options`, `tcp_meta_window_size` UInt32, `tcp_meta_mss` UInt16, `tcp_meta_window_scale` UInt8, `syn_to_clienthello_ms` Int32 |
|
||||
| TLS / fingerprint | `tls_version`, `tls_sni`, `tls_alpn` (LowCardinality), `ja3`, `ja3_hash`, `ja4` |
|
||||
| En-têtes HTTP | `header_user_agent`, `header_accept`, `header_accept_encoding`, `header_accept_language`, `header_x_request_id`, `header_x_trace_id`, `header_x_forwarded_for`, `header_sec_ch_ua*`, `header_sec_fetch_*` |
|
||||
|
||||
### Utilisateurs et permissions
|
||||
|
||||
```sql
|
||||
CREATE USER IF NOT EXISTS data_writer IDENTIFIED WITH plaintext_password BY 'MotDePasse';
|
||||
CREATE USER IF NOT EXISTS analyst IDENTIFIED WITH plaintext_password BY 'MotDePasseAnalyst';
|
||||
-- data_writer : INSERT sur http_logs_raw uniquement (compte du service)
|
||||
GRANT INSERT ON mabase_prod.http_logs_raw TO data_writer;
|
||||
GRANT SELECT ON mabase_prod.http_logs_raw TO data_writer;
|
||||
|
||||
GRANT INSERT(raw_json) ON mabase_prod.http_logs_raw TO data_writer;
|
||||
GRANT SELECT(raw_json) ON mabase_prod.http_logs_raw TO data_writer;
|
||||
-- analyst : lecture sur la table parsée
|
||||
GRANT SELECT ON mabase_prod.http_logs TO analyst;
|
||||
```
|
||||
|
||||
@ -302,14 +243,16 @@ GRANT SELECT ON mabase_prod.http_logs TO analyst;
|
||||
|
||||
```sql
|
||||
-- Données brutes reçues
|
||||
SELECT count(*), min(ingest_time), max(ingest_time) FROM http_logs_raw;
|
||||
SELECT count(*), min(ingest_time), max(ingest_time) FROM mabase_prod.http_logs_raw;
|
||||
|
||||
-- Données parsées par la vue matérialisée
|
||||
SELECT count(*), min(time), max(time) FROM http_logs;
|
||||
SELECT count(*), min(time), max(time) FROM mabase_prod.http_logs;
|
||||
|
||||
-- Derniers logs
|
||||
-- Derniers logs corrélés
|
||||
SELECT time, src_ip, dst_ip, method, host, path, ja4
|
||||
FROM http_logs ORDER BY time DESC LIMIT 10;
|
||||
FROM mabase_prod.http_logs
|
||||
WHERE correlated = 1
|
||||
ORDER BY time DESC LIMIT 10;
|
||||
```
|
||||
|
||||
## Signaux
|
||||
@ -453,16 +396,18 @@ python3 scripts/test-correlation-advanced.py --all
|
||||
|
||||
### ClickHouse : erreurs d'insertion
|
||||
|
||||
- **`No such column`** : vérifier que la table utilise la colonne unique `raw_json` (pas de colonnes séparées)
|
||||
- **`ACCESS_DENIED`** : `GRANT INSERT(raw_json) ON mabase_prod.http_logs_raw TO data_writer;`
|
||||
- **`No such column`** : vérifier que la table `http_logs_raw` utilise la colonne unique `raw_json` (pas de colonnes séparées)
|
||||
- **`ACCESS_DENIED`** : `GRANT INSERT ON mabase_prod.http_logs_raw TO data_writer;`
|
||||
- Les erreurs de flush sont loggées en ERROR dans les logs du service
|
||||
|
||||
### Vue matérialisée vide
|
||||
|
||||
Si `http_logs_raw` a des données mais `http_logs` est vide :
|
||||
```sql
|
||||
SHOW CREATE TABLE mv_http_logs;
|
||||
GRANT SELECT(raw_json) ON mabase_prod.http_logs_raw TO data_writer;
|
||||
-- Vérifier la vue
|
||||
SHOW CREATE TABLE mabase_prod.mv_http_logs;
|
||||
-- Vérifier les permissions (la MV s'exécute sous le compte du service)
|
||||
GRANT SELECT ON mabase_prod.http_logs_raw TO data_writer;
|
||||
```
|
||||
|
||||
### Sockets Unix : permission denied
|
||||
|
||||
174
sql/init.sql
Normal file
174
sql/init.sql
Normal file
@ -0,0 +1,174 @@
|
||||
-- =============================================================================
|
||||
-- logcorrelator - Initialisation ClickHouse
|
||||
-- =============================================================================
|
||||
-- Ce fichier crée la base de données, les tables, la vue matérialisée
|
||||
-- et les utilisateurs nécessaires au fonctionnement de logcorrelator.
|
||||
--
|
||||
-- Usage :
|
||||
-- clickhouse-client --multiquery < sql/init.sql
|
||||
-- =============================================================================
|
||||
|
||||
-- -----------------------------------------------------------------------------
|
||||
-- Base de données
|
||||
-- -----------------------------------------------------------------------------
|
||||
CREATE DATABASE IF NOT EXISTS mabase_prod;
|
||||
|
||||
-- -----------------------------------------------------------------------------
|
||||
-- Table brute : cible directe des inserts du service
|
||||
-- Le service n'insère que dans cette table (colonne raw_json).
|
||||
-- -----------------------------------------------------------------------------
|
||||
CREATE TABLE IF NOT EXISTS mabase_prod.http_logs_raw
|
||||
(
|
||||
`raw_json` String,
|
||||
`ingest_time` DateTime DEFAULT now()
|
||||
)
|
||||
ENGINE = MergeTree
|
||||
PARTITION BY toDate(ingest_time)
|
||||
ORDER BY ingest_time
|
||||
SETTINGS index_granularity = 8192;
|
||||
|
||||
-- -----------------------------------------------------------------------------
|
||||
-- Table parsée : alimentée automatiquement par la vue matérialisée
|
||||
-- -----------------------------------------------------------------------------
|
||||
CREATE TABLE IF NOT EXISTS mabase_prod.http_logs
|
||||
(
|
||||
-- Temporel
|
||||
`time` DateTime,
|
||||
`log_date` Date DEFAULT toDate(time),
|
||||
|
||||
-- Réseau
|
||||
`src_ip` IPv4,
|
||||
`src_port` UInt16,
|
||||
`dst_ip` IPv4,
|
||||
`dst_port` UInt16,
|
||||
|
||||
-- HTTP
|
||||
`method` LowCardinality(String),
|
||||
`scheme` LowCardinality(String),
|
||||
`host` LowCardinality(String),
|
||||
`path` String,
|
||||
`query` String,
|
||||
`http_version` LowCardinality(String),
|
||||
|
||||
-- Corrélation
|
||||
`orphan_side` LowCardinality(String),
|
||||
`correlated` UInt8,
|
||||
`keepalives` UInt16,
|
||||
`a_timestamp` UInt64,
|
||||
`b_timestamp` UInt64,
|
||||
`conn_id` String,
|
||||
|
||||
-- Métadonnées IP
|
||||
`ip_meta_df` UInt8,
|
||||
`ip_meta_id` UInt16,
|
||||
`ip_meta_total_length` UInt16,
|
||||
`ip_meta_ttl` UInt8,
|
||||
|
||||
-- Métadonnées TCP
|
||||
`tcp_meta_options` LowCardinality(String),
|
||||
`tcp_meta_window_size` UInt32,
|
||||
`tcp_meta_mss` UInt16,
|
||||
`tcp_meta_window_scale` UInt8,
|
||||
`syn_to_clienthello_ms` Int32,
|
||||
|
||||
-- TLS / fingerprint
|
||||
`tls_version` LowCardinality(String),
|
||||
`tls_sni` LowCardinality(String),
|
||||
`tls_alpn` LowCardinality(String),
|
||||
`ja3` String,
|
||||
`ja3_hash` String,
|
||||
`ja4` String,
|
||||
|
||||
-- En-têtes HTTP
|
||||
`header_user_agent` String,
|
||||
`header_accept` String,
|
||||
`header_accept_encoding` String,
|
||||
`header_accept_language` 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
|
||||
)
|
||||
ENGINE = MergeTree
|
||||
PARTITION BY log_date
|
||||
ORDER BY (time, src_ip, dst_ip, ja4)
|
||||
SETTINGS index_granularity = 8192;
|
||||
|
||||
-- -----------------------------------------------------------------------------
|
||||
-- Vue matérialisée : parse le JSON de http_logs_raw vers http_logs
|
||||
-- -----------------------------------------------------------------------------
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS mabase_prod.mv_http_logs
|
||||
TO mabase_prod.http_logs
|
||||
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,
|
||||
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,
|
||||
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, '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_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
|
||||
FROM mabase_prod.http_logs_raw;
|
||||
|
||||
-- -----------------------------------------------------------------------------
|
||||
-- Utilisateurs et permissions
|
||||
-- -----------------------------------------------------------------------------
|
||||
CREATE USER IF NOT EXISTS data_writer IDENTIFIED WITH plaintext_password BY 'ChangeMe';
|
||||
CREATE USER IF NOT EXISTS analyst IDENTIFIED WITH plaintext_password BY 'ChangeMe';
|
||||
|
||||
-- data_writer : INSERT uniquement sur la table brute
|
||||
GRANT INSERT ON mabase_prod.http_logs_raw TO data_writer;
|
||||
GRANT SELECT ON mabase_prod.http_logs_raw TO data_writer;
|
||||
|
||||
-- analyst : lecture sur la table parsée
|
||||
GRANT SELECT ON mabase_prod.http_logs TO analyst;
|
||||
|
||||
-- -----------------------------------------------------------------------------
|
||||
-- Vérifications post-installation
|
||||
-- -----------------------------------------------------------------------------
|
||||
-- SELECT count(*), min(ingest_time), max(ingest_time) FROM mabase_prod.http_logs_raw;
|
||||
-- SELECT count(*), min(time), max(time) FROM mabase_prod.http_logs;
|
||||
-- SELECT time, src_ip, dst_ip, method, host, path, ja4 FROM mabase_prod.http_logs ORDER BY time DESC LIMIT 10;
|
||||
Reference in New Issue
Block a user