ja4-platform/docs/shared/python-ja4common.md

python-ja4common

Bibliothèque Python partagée pour la plateforme ja4, fournissant un client ClickHouse singleton et une configuration centralisée via pydantic-settings.

Utilisée par : bot-detector (via ja4_common.clickhouse)

Note : le dashboard n'utilise pas ja4_common. Il possède son propre client clickhouse-connect léger dans backend/database.py.

---

Informations du package

Clé	Valeur
Nom du package	ja4-common
Version	0.1.0
Python	≥ 3.11
Dépendances	clickhouse-connect >= 0.8.0, pydantic-settings >= 2.1.0
Build system	setuptools >= 68 + wheel

---

ClickHouseSettings

Modèle pydantic-settings qui lit la configuration depuis les variables d'environnement et les fichiers .env.

Champs

Champ	Type	Défaut	Variable d'env	Description
CLICKHOUSE_HOST	str	"clickhouse"	CLICKHOUSE_HOST	Nom d'hôte du serveur ClickHouse
CLICKHOUSE_PORT	int	8123	CLICKHOUSE_PORT	Port de l'API HTTP ClickHouse
CLICKHOUSE_DB	str	"ja4_processing"	CLICKHOUSE_DB	Base de données par défaut (rétro-compatibilité)
CLICKHOUSE_DB_PROCESSING	str	"ja4_processing"	CLICKHOUSE_DB_PROCESSING	Base de données ML, agrégations, dictionnaires
CLICKHOUSE_DB_LOGS	str	"ja4_logs"	CLICKHOUSE_DB_LOGS	Base de données des logs HTTP bruts
CLICKHOUSE_USER	str	"admin"	CLICKHOUSE_USER	Utilisateur pour l'authentification
CLICKHOUSE_PASSWORD	str	""	CLICKHOUSE_PASSWORD	Mot de passe pour l'authentification

Schéma dual-database

La plateforme utilise deux bases de données ClickHouse :

• CLICKHOUSE_DB_PROCESSING (ja4_processing) : tables ML (ml_detected_anomalies, ml_all_scores), agrégations (agg_*), dictionnaires (dict_*), feedback SOC, audit
• CLICKHOUSE_DB_LOGS (ja4_logs) : http_logs_raw, http_logs, vues matérialisées

Des références croisées existent entre les deux bases — les vues matérialisées de l'une lisent dans l'autre. Utiliser toujours des noms de tables pleinement qualifiés :

from ja4_common.settings import settings

query = f"SELECT ... FROM {settings.CLICKHOUSE_DB_PROCESSING}.ml_detected_anomalies ..."
query = f"SELECT ... FROM {settings.CLICKHOUSE_DB_LOGS}.http_logs ..."

Ne jamais coder en dur les noms de bases dans les requêtes.

Sources de configuration

Les paramètres sont chargés par ordre de priorité :

1. Variables d'environnement (priorité la plus haute)
2. Fichier .env dans le répertoire courant
3. Valeurs par défaut (priorité la plus basse)

Les noms de variables d'environnement sont sensibles à la casse (ex. CLICKHOUSE_HOST, pas clickhouse_host).

Utilisation

from ja4_common.settings import settings

print(settings.CLICKHOUSE_HOST)           # "clickhouse" ou depuis l'env
print(settings.CLICKHOUSE_PORT)           # 8123 ou depuis l'env
print(settings.CLICKHOUSE_DB_PROCESSING)  # "ja4_processing" ou depuis l'env
print(settings.CLICKHOUSE_DB_LOGS)        # "ja4_logs" ou depuis l'env

Le singleton settings est créé au niveau du module à l'import.

---

ClickHouseClient

Encapsule clickhouse_connect avec reconnexion automatique et une API épurée.

Méthodes

Méthode	Signature	Description
connect	connect() → Client	Retourne le client clickhouse_connect sous-jacent, crée ou reconnecte si nécessaire
_ping	_ping() → bool	Vérifie la connexion via client.ping(), retourne False en cas d'exception
query	query(query: str, params: dict = None)	Exécute une requête SELECT, retourne le résultat
command	command(query: str, params: dict = None)	Exécute une commande DDL/DML (CREATE, INSERT, etc.)
insert	insert(table: str, data, column_names=None)	Insertion en masse dans une table
close	close()	Ferme la connexion et libère les ressources

Reconnexion automatique

La méthode connect() reconnecte automatiquement si la connexion est perdue :

def connect(self):
    if self._client is None or not self._ping():
        self._client = clickhouse_connect.get_client(
            host=settings.CLICKHOUSE_HOST,
            port=settings.CLICKHOUSE_PORT,
            database=settings.CLICKHOUSE_DB,
            user=settings.CLICKHOUSE_USER,
            password=settings.CLICKHOUSE_PASSWORD,
            connect_timeout=10,
        )
    return self._client

Exemple d'utilisation

from ja4_common.clickhouse import get_client

client = get_client()

# Requête SELECT avec noms de tables pleinement qualifiés
result = client.query(
    f"SELECT count() FROM {settings.CLICKHOUSE_DB_LOGS}.http_logs "
    "WHERE src_ip = {ip:String}",
    {"ip": "203.0.113.42"},
)
print(result.result_rows)

# INSERT
client.insert(
    "audit_logs",
    [[datetime.now(), "analyst1", "investigate", "ip", "203.0.113.42"]],
    column_names=["timestamp", "user_name", "action", "entity_type", "entity_id"],
)

# Commande DDL
client.command("OPTIMIZE TABLE http_logs FINAL")

---

Singleton get_client()

La fonction get_client() fournit un singleton de ClickHouseClient au niveau du module :

from ja4_common.clickhouse import get_client

# Le premier appel crée le client
client1 = get_client()

# Les appels suivants retournent la même instance
client2 = get_client()
assert client1 is client2

Implémentation

_client: Optional[ClickHouseClient] = None

def get_client() -> ClickHouseClient:
    global _client
    if _client is None:
        _client = ClickHouseClient()
    return _client

Architecture à deux niveaux de singleton :

• get_client() → singleton pour ClickHouseClient
• settings dans settings.py → singleton pour ClickHouseSettings

---

Exports du package

Le __init__.py n'exporte qu'une chaîne de version :

"""JA4 Common — shared utilities for the JA4 security suite."""
__version__ = "0.1.0"

Les consommateurs doivent importer directement depuis les sous-modules :

from ja4_common.clickhouse import get_client
from ja4_common.settings import settings

---

Intégration dans un nouveau service

1. Ajouter la dépendance

Dans le requirements.txt du service :

ja4-common @ file:///app/shared/python/ja4_common

Ou dans pyproject.toml :

[project]
dependencies = [
    "ja4-common",
]

2. Configuration Docker

# Copier la bibliothèque partagée
COPY shared/python/ja4_common /app/shared/python/ja4_common
RUN pip install /app/shared/python/ja4_common

# Copier le code du service
COPY services/mon-service /app/services/mon-service

3. Utiliser dans le code

from ja4_common.clickhouse import get_client
from ja4_common.settings import settings

# Accéder à la configuration
print(f"Connexion à {settings.CLICKHOUSE_HOST}:{settings.CLICKHOUSE_PORT}")
print(f"Base logs : {settings.CLICKHOUSE_DB_LOGS}")
print(f"Base processing : {settings.CLICKHOUSE_DB_PROCESSING}")

# Utiliser le client avec noms pleinement qualifiés
db = get_client()
result = db.query(
    f"SELECT count() FROM {settings.CLICKHOUSE_DB_PROCESSING}.ml_detected_anomalies"
)

4. Configuration d'environnement

Créer un fichier .env ou définir les variables d'environnement :

CLICKHOUSE_HOST=clickhouse.example.com
CLICKHOUSE_PORT=8123
CLICKHOUSE_DB=ja4_processing
CLICKHOUSE_DB_PROCESSING=ja4_processing
CLICKHOUSE_DB_LOGS=ja4_logs
CLICKHOUSE_USER=data_writer
CLICKHOUSE_PASSWORD=secret

---

Tests : simuler le client

Avec unittest.mock

from unittest.mock import MagicMock, patch
from ja4_common.clickhouse import ClickHouseClient

def test_mon_service():
    mock_client = MagicMock(spec=ClickHouseClient)
    mock_client.query.return_value = MagicMock(result_rows=[(42,)])

    with patch("ja4_common.clickhouse._client", mock_client):
        from ja4_common.clickhouse import get_client
        client = get_client()
        result = client.query("SELECT count() FROM http_logs")
        assert result.result_rows == [(42,)]

Surcharger les paramètres en test

from ja4_common.settings import ClickHouseSettings

# Créer des paramètres personnalisés pour les tests
test_settings = ClickHouseSettings(
    CLICKHOUSE_HOST="localhost",
    CLICKHOUSE_PORT=8123,
    CLICKHOUSE_DB="test_db",
    CLICKHOUSE_DB_PROCESSING="test_processing",
    CLICKHOUSE_DB_LOGS="test_logs",
    CLICKHOUSE_USER="test_user",
    CLICKHOUSE_PASSWORD="test_pass",
)

---

Fichiers sources

Fichier	Description
ja4_common/__init__.py	Docstring du package et __version__
ja4_common/settings.py	Modèle pydantic-settings ClickHouseSettings et singleton settings
ja4_common/clickhouse.py	Classe ClickHouseClient et singleton get_client()
pyproject.toml	Métadonnées du package et dépendances
tests/test_settings.py	Tests unitaires pour ClickHouseSettings
tests/test_clickhouse.py	Tests unitaires pour ClickHouseClient