79 lines
3.0 KiB
Python
79 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Crée une Saved Query "Sans VPN" réutilisable dans OpenSearch Dashboards (Wazuh).
|
|
|
|
Objectif : permettre d'exclure en un clic les alertes WireGuard/VPN (handshakes,
|
|
timeouts, roaming, reconnexions — rules 100800-100826) de la vue native
|
|
"Security Events" / "Threat Hunting", qui liste par défaut TOUTES les alertes
|
|
sans filtre de groupe. Le dashboard dédié "VPN - Activite WireGuard"
|
|
(create-dashboard-vpn.py) continue lui d'afficher ces événements normalement :
|
|
cette query ne supprime rien de l'index, elle filtre uniquement l'affichage.
|
|
|
|
Usage:
|
|
docker cp create-saved-query-no-vpn.py wazuh_dashboard:/tmp/
|
|
docker exec wazuh_dashboard python3 /tmp/create-saved-query-no-vpn.py
|
|
|
|
Une fois créée, ouvrir "Security Events" > barre de recherche > icône
|
|
"Saved Queries" (dossier) > sélectionner "Sans VPN".
|
|
"""
|
|
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
import ssl
|
|
import base64
|
|
import sys
|
|
import os
|
|
|
|
# ── Configuration ────────────────────────────────────────────────────────────
|
|
DASHBOARD_HOST = "https://localhost:5601"
|
|
DASHBOARD_USER = "kibanaserver"
|
|
DASHBOARD_PASS = os.environ.get("DASHBOARD_PASSWORD", "E9Jpr6586kQ3wYrCS2!")
|
|
|
|
QUERY_ID = "no-vpn-alerts"
|
|
QUERY_TITLE = "Sans VPN"
|
|
QUERY_DESC = "Exclut les alertes WireGuard/VPN (handshakes, timeouts, roaming) de la vue Security Events."
|
|
QUERY_KUERY = "NOT rule.groups: wireguard"
|
|
|
|
# ── Client HTTP ───────────────────────────────────────────────────────────────
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
_auth = base64.b64encode(f"{DASHBOARD_USER}:{DASHBOARD_PASS}".encode()).decode()
|
|
_headers = {
|
|
"Content-Type": "application/json",
|
|
"osd-xsrf": "true",
|
|
"Authorization": f"Basic {_auth}",
|
|
}
|
|
|
|
def api(method, path, body=None):
|
|
data = json.dumps(body).encode() if body else None
|
|
req = urllib.request.Request(DASHBOARD_HOST + path, data=data, headers=_headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, context=ctx, timeout=15) as r:
|
|
return json.loads(r.read())
|
|
except urllib.error.HTTPError as e:
|
|
return {"error": e.code, "msg": e.read().decode()[:300]}
|
|
|
|
def main():
|
|
print("=== Creation Saved Query: Sans VPN ===\n")
|
|
body = {
|
|
"attributes": {
|
|
"title": QUERY_TITLE,
|
|
"description": QUERY_DESC,
|
|
"query": {"language": "kuery", "query": QUERY_KUERY},
|
|
"filters": [],
|
|
}
|
|
}
|
|
r = api("POST", f"/api/saved_objects/query/{QUERY_ID}?overwrite=true", body)
|
|
if "id" in r:
|
|
print(f" OK Saved Query '{QUERY_TITLE}' créée ({r['id']})")
|
|
print(" Utilisation : Security Events > barre de recherche > icone dossier (Saved Queries) > 'Sans VPN'")
|
|
else:
|
|
print(f" ERR {r}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|