Comment créer une suite de sécurité avec l'IA à partir d'outils open source
Traduit de l'original en espagnol. Lire en espagnol
Notre configuration se compose d’un serveur Linux qui exécute plusieurs outils de sécurité open source : fail2ban, ClamAV, rkhunter, maldet et lynis. Chaque outil écrit ses logs dans un répertoire commun, avec un script bash qui agrège les événements de fail2ban, de SSH et d’autres sources. Pour donner du sens à ces données, nous utilisons un script Python qui fait appel à Azure OpenAI via la bibliothèque LangChain.
Disclaimer :
Ceci n’est qu’un exemple. Si vous souhaitez l’utiliser en production, analysez d’abord vos besoins spécifiques, testez de manière approfondie et adaptez le code aux besoins et aux politiques de votre organisation. N’utilisez pas ce code tel quel sans l’avoir relu et adapté à votre environnement. Veillez à bien protéger les fichiers de configuration, les identifiants et les rapports générés.
Que sont ces outils et pourquoi les exécuter ensemble ?
Dans cet exemple, nous combinons plusieurs outils open source largement reconnus pour la sécurité des serveurs Linux :
- fail2ban : surveille les logs de services comme SSH pour détecter les tentatives de connexion échouées ou les attaques par force brute, et bloque automatiquement les IP fautives via des règles de pare-feu.
- ClamAV : un antivirus open source qui analyse les fichiers à la recherche de malwares, de virus et de chevaux de Troie connus.
- rkhunter (Rootkit Hunter) : recherche des rootkits, des portes dérobées et des exploits locaux en examinant les fichiers système, les binaires et les configurations suspectes.
- maldet (Linux Malware Detect) : spécialisé dans la détection des malwares connus et des menaces web sous Linux, particulièrement utile pour les serveurs web mutualisés.
- lynis : un outil d’audit de sécurité et de conformité qui analyse la configuration, recherche les vulnérabilités et fournit des recommandations pour renforcer le système.
Pourquoi les exécuter ensemble ?
Chacun de ces outils couvre un aspect différent de la sécurité du système : de la détection d’intrusions et du blocage des accès non autorisés à l’analyse des malwares et des rootkits, en passant par l’audit de la configuration. En les exécutant tous ensemble et en combinant leurs résultats, on obtient une vision beaucoup plus complète et approfondie de l’état de sécurité du serveur. De plus, centraliser les logs et les analyser avec l’intelligence artificielle permet de corréler les événements, de découvrir des schémas qui pourraient passer inaperçus et de prioriser plus efficacement les actions de réponse.
1. Chargement de l’environnement et de la configuration
Le script commence par charger les variables d’environnement depuis un fichier .env (s’il existe) ou depuis l’environnement du système. Il vérifie ensuite que toutes les variables requises sont présentes et s’arrête s’il en manque une.
Pratique recommandée :
Assurez-vous que le fichier .env et tout fichier de configuration sont correctement protégés et accessibles uniquement à l’utilisateur concerné. Envisagez de chiffrer les fichiers qui contiennent des identifiants.
def load_environment():
"""Load environment variables from dotenv file and check required ones."""
if Path(ENV_FILE).exists():
load_dotenv(dotenv_path=ENV_FILE)
logging.info(f"Loaded environment from {ENV_FILE}")
else:
logging.warning(f".env file not found at {ENV_FILE}; relying on current environment.")
missing_vars = [var for var in REQUIRED_ENV if not os.getenv(var)]
if missing_vars:
logging.error(f"Missing required environment variables: {', '.join(missing_vars)}")
sys.exit(1)2. Vérification et création des répertoires
Il s’assure que les répertoires de logs, de sortie et d’archives existent, et les crée si nécessaire. Si le répertoire de logs est absent, le script s’arrête.
Important :
Vérifiez les permissions des répertoires et cloisonnez les accès pour empêcher des utilisateurs non autorisés de manipuler les fichiers de log et les rapports.
def ensure_directories():
"""Ensure log, output, and archive directories exist."""
if not LOG_DIR.is_dir():
logging.error(f"Log directory not found: {LOG_DIR}")
sys.exit(1)
OUT_DIR.mkdir(parents=True, exist_ok=True)
ARCHIVE_DIR.mkdir(parents=True, exist_ok=True)3. Authentification auprès d’Azure
La fonction get_azure_ad_token obtient un token d’accès auprès d’Azure Active Directory à l’aide de client credentials.
Remarque :
Assurez-vous que les autorisations dans Azure sont limitées au strict nécessaire pour votre cas d’usage. Les identifiants utilisés doivent être stockés et transmis de manière sécurisée.
get_azure_ad_token():
"""Acquire Azure AD Token using client credentials."""
token_url = f"https://login.microsoftonline.com/{os.environ['AZURE_TENANT_ID']}/oauth2/v2.0/token"
token_data = {
"grant_type": "client_credentials",
"client_id": os.environ['AZURE_CLIENT_ID'],
"client_secret": os.environ['AZURE_CLIENT_SECRET'],
"scope": os.environ.get('AZURE_OPENAI_SCOPE', 'https://cognitiveservices.azure.com/.default'),
}
response = requests.post(token_url, data=token_data)
response.raise_for_status()
return response.json()["access_token"]4. Chargement des prompts pour l’IA
Le script charge deux prompts depuis des fichiers texte : l’un pour l’analyse individuelle de chaque log, l’autre pour le rapport final agrégé.
Important :
Le contenu des prompts détermine la qualité et l’orientation des rapports. Relisez et ajustez les prompts selon le niveau technique et les besoins de votre organisation.
load_prompt(prompt_file: Path) -> str:
"""Load a prompt from a file."""
if not prompt_file.is_file():
logging.error(f"{prompt_file} not found!")
sys.exit(1)
with prompt_file.open("r", encoding="utf-8") as f:
return f.read()5. Collecte des logs
Il récupère tous les fichiers .log et .txt du répertoire de logs pour les analyser.
Remarque :
Assurez-vous que seuls les logs attendus se trouvent dans le répertoire, afin d’éviter d’analyser des fichiers sans intérêt ou potentiellement malveillants.
collect_log_files() -> list:
"""Collect log files (*.log, *.txt) from LOG_DIR."""
log_files = list(LOG_DIR.glob("*.log")) + list(LOG_DIR.glob("*.txt"))
return log_files6. Analyse IA de chaque log
Il lit chaque log, l’envoie à l’IA avec le prompt individuel et enregistre le résultat.
Avertissement :
L’IA peut commettre des erreurs d’interprétation ou ne pas détecter des menaces avancées. Elle ne remplace pas la supervision et la vérification humaines dans les contextes critiques.
log_file in log_files:
try:
with log_file.open("r", encoding="utf-8") as f:
log_content = f.read()
messages = [
SystemMessage(content=per_log_system_prompt),
HumanMessage(content=log_content)
]
response = llm(messages)
report_text = response.content if hasattr(response, 'content') else str(response)
individual_reports.append({
"log_file": log_file.name,
"report": report_text
})
except Exception as e:
logging.error(f"Error processing {log_file}: {e}")7. Enregistrement des rapports individuels
Il enregistre les rapports individuels générés par l’IA dans un fichier texte.
Bonne pratique :
Relisez régulièrement les rapports et versionnez-les si votre organisation l’exige.
individual_reports_txt = ""
for report in individual_reports:
individual_reports_txt += f"===== Report for {report['log_file']} =====\n{report['report']}\n\n"
try:
with REPORT_TXT.open("w", encoding="utf-8") as f:
f.write(individual_reports_txt)
except Exception as e:
logging.error(f"Error writing individual reports: {e}")8. Rapport final agrégé par l’IA
L’IA reçoit tous les rapports individuels et produit une analyse ou une synthèse globale.
Important :
Validez les résultats de l’IA et tenez compte de ses limites. Utilisez la sortie comme une aide, et non comme une décision automatique définitive.
reports_content = "\n".join(
f"Report for {r['log_file']}:\n{r['report']}" for r in individual_reports
)
final_messages = [
SystemMessage(content=final_system_prompt),
HumanMessage(content=reports_content)
]
try:
final_response = llm(final_messages)
final_report = final_response.content if hasattr(final_response, 'content') else str(final_response)
except Exception as e:
logging.error(f"Error during final aggregation OpenAI call: {e}")
final_report = "Aggregation failed due to error."9. Enregistrement et permissions des rapports
Le rapport final et les rapports individuels sont stockés dans des fichiers .txt et .json, avec les permissions appropriées.
Recommandation :
Envisagez de chiffrer les rapports s’ils contiennent des informations sensibles et configurez des sauvegardes automatiques dans un environnement sécurisé.
try:
with REPORT_TXT.open("a", encoding="utf-8") as f:
f.write("\n===== FINAL AGGREGATED REPORT =====\n")
f.write(final_report)
with REPORT_JSON.open("w", encoding="utf-8") as f:
json.dump({
"individual_reports": individual_reports,
"final_report": final_report
}, f, indent=2)
except Exception as e:
logging.error(f"Error saving final report: {e}")
def set_permissions():
"""Set secure permissions on report files."""
try:
REPORT_JSON.chmod(0o640)
REPORT_TXT.chmod(0o640)
except Exception as e:
logging.error(f"Error setting permissions on report files: {e}")10. Envoi automatique par e-mail (SMTP)
Il envoie le rapport de sécurité par e-mail via SMTP.
Avertissement :
Utilisez des mots de passe d’application et des connexions TLS/SSL. N’utilisez pas de comptes personnels et n’exposez pas d’identifiants dans des fichiers non sécurisés.
def send_email_report(report_path: Path):
"""Send the TXT report as the email body using smtplib."""
SMTP_SERVER = os.getenv('SMTP_SERVER')
SMTP_PORT = int(os.getenv('SMTP_PORT', '587'))
SMTP_USERNAME = os.getenv('SMTP_USERNAME')
SMTP_PASSWORD = os.getenv('SMTP_PASSWORD')
if not SMTP_SERVER or not MAIL_TO or not MAIL_FROM:
logging.error("SMTP configuration or email addresses missing.")
logging.info(f"Reports saved at:\n JSON: {REPORT_JSON}\n Text: {REPORT_TXT}")
sys.exit(3)
try:
with report_path.open("r", encoding="utf-8") as f:
report_content = f.read()
msg = EmailMessage()
msg['Subject'] = MAIL_SUBJECT
msg['From'] = MAIL_FROM
msg['To'] = MAIL_TO
msg.set_content(report_content)
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
server.starttls()
if SMTP_USERNAME and SMTP_PASSWORD:
server.login(SMTP_USERNAME, SMTP_PASSWORD)
server.send_message(msg)
logging.info(f"Security report emailed to {MAIL_TO} via SMTP")
except Exception as e:
logging.error(f"Failed to email report via SMTP: {e}")11. Archivage et nettoyage
Tous les logs et rapports sont compressés dans un fichier ZIP, et les anciens fichiers sont supprimés selon la politique de rétention.
Important :
Vérifiez régulièrement l’espace disque et l’état des archives, et surveillez les performances du système, surtout si le volume de logs est élevé.
def archive_and_cleanup():
"""Archive logs and reports, and clean up old files."""
zip_name = f"security-logs-{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}.zip"
logs_to_archive = list(LOG_DIR.glob("*.txt")) + list(LOG_DIR.glob("*.log")) + [REPORT_JSON]
files_to_archive = [f for f in logs_to_archive if f.is_file()]
archive_path = ARCHIVE_DIR / zip_name
if files_to_archive:
try:
with zipfile.ZipFile(archive_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for file in files_to_archive:
zipf.write(file, arcname=file.name)
for file in files_to_archive:
file.unlink()
logging.info(f"Archived and removed: {' '.join(str(f) for f in files_to_archive)}")
except Exception as e:
logging.error(f"Error archiving files: {e}")
else:
logging.info(f"No .txt, .log, or .json files to archive in {LOG_DIR}.")
# Remove old archives
now = datetime.datetime.now().timestamp()
for f in ARCHIVE_DIR.glob("*.zip"):
try:
if f.stat().st_mtime < now - ARCHIVE_RETENTION_DAYS * 86400:
f.unlink()
except Exception as e:
logging.error(f"Error deleting archive {f}: {e}")
# Remove other files/directories in LOG_DIR except .log_offsets
for f in LOG_DIR.iterdir():
if f.name == '.log_offsets':
continue
try:
if f.is_dir():
shutil.rmtree(f)
else:
f.unlink()
except Exception as e:
logging.error(f"Error removing {f}: {e}")12. System prompts utilisés
// MAIN LLM - This will handle the final report
/*
Act as a senior security analyst.
You are given output from an AI analyst.
Logs are from Fail2ban, ClamAV, RKHunter, Lynis, Maldet,
and SSH authentication logs from /var/log/security-scans.
Tasks:
Vulnerabilities: Identify misconfigurations, outdated practices,
open attack surfaces, or missing controls in the system.
Provide severity, verbatim evidence, and remediation steps for each finding.
Detections: Extract concrete security events, such as repeated failed SSH attempts,
or banned IP address activity.
Include first_seen, last_seen, counts, and impacted entities as evidence.
Patterns & Analysis: Summarize attacker behavior patterns,
such as targeted usernames and source IP ranges observed.
Include defenses triggered, like bans, lockouts, and possible evasion indicators.
Executive Summary: Provide a concise, non-technical summary of risk posture,
including urgent recommended actions if found.
Appendices:
a. List Indicators of Compromise (IPs, users, keys)
and note any data coverage limitations encountered in analysis.
b. Output a single JSON object containing:
{
“executive_summary”: “…”,
“detections”: [
{
“type”: “…”,
“details”: “…”,
“evidence”: [“…”],
“first_seen”: “…”,
“last_seen”: “…”,
“count”: 0,
“severity”: “low|medium|high|critical”
}
],
“vulnerabilities”: [
{
“issue”: “…”,
“severity”: “…”,
“evidence”: [“…”],
“remediation”: “…”
}
],
“patterns_analysis”: {
“attacker_behaviors”: “…”,
“defenses”: “…”,
“timeline”: “…”,
“notable_patterns”: “…”
},
“ioc”: {
“ips”: [“…”],
“users”: [“…”],
“other”: [“…”]
},
“limitations”: “…”
}
c. Use only concrete evidence from logs
and quote relevant lines in the evidence arrays.
*/
// JUNIOR LLM - This will handle individual log analysis
/*
You are a Linux security log analyst.
Review a single Linux security log entry for issues.
Identify security issues, anomalies, or suspicious activities present in the log.
Summarize findings concisely and clearly for further review.
Focus on:
All relevant security events or indicators in the log entry.
Potential impact or risk level identified from the analysis.
Actionable insights or next steps, if applicable, for remediating issues.
Do not include the full log entry in your response.
Avoid unnecessary verbosity to keep the summary focused and clear.
Your summary will be used by senior analysts for review and aggregation.
Format response as:
Findings:
Insights:
Recommendations (if any):
*/Exemple : script Bash pour créer des logs d’activité personnalisés à analyser
- Adaptez
/var/log/fail2ban.log, etc. aux chemins de logs de votre distribution si nécessaire. - Par souci de confidentialité, évitez de l’exécuter ou de le partager sur des systèmes contenant des données sensibles ou réglementées sans examen préalable.
#!/usr/bin/env bash
set -euo pipefail
# Set output directory in a generic temp/logs path
OUT_DIR="/tmp/security-scans"
TS="$(date +'%Y%m%d-%H%M%S')"
mkdir -p "$OUT_DIR"
# Helper to safely run commands that might not exist
have() { command -v "$1" >/dev/null 2>&1; }
# -----------------------------
# Fail2ban: jails and bans (last 48h from fail2ban.log)
# -----------------------------
if have fail2ban-client && [ -r /var/log/fail2ban.log ]; then
{
echo "== Fail2ban bans (last 48 hours) =="
awk -v d="$(date --date='-48 hours' '+%Y-%m-%d %H:%M:%S')" \
'$0 >= d && /Ban /' /var/log/fail2ban.log
} > "$OUT_DIR/fail2ban-bans-activity-$TS.txt" 2>&1
fi
# -----------------------------
# SSH: successful and failed logins (last 48h)
# -----------------------------
AUTH_FILE=""
for cand in /var/log/auth.log /var/log/secure; do
if [ -r "$cand" ]; then AUTH_FILE="$cand"; break; fi
done
if [ -n "$AUTH_FILE" ]; then
{
echo "== SSH successful logins (last 48h) =="
for day in $(date --date='-0 days' '+%b %e') $(date --date='-1 days' '+%b %e'); do
awk -v d="$day" '$0 ~ d' "$AUTH_FILE" | grep -E "sshd\[.*\]: Accepted (password|publickey)" || true
done
echo
echo "== SSH failed logins (last 48h) =="
for day in $(date --date='-0 days' '+%b %e') $(date --date='-1 days' '+%b %e'); do
awk -v d="$day" '$0 ~ d' "$AUTH_FILE" | grep -E "sshd\[.*\]: Failed password" || true
done
echo
echo "== SSH invalid users (last 48h) =="
for day in $(date --date='-0 days' '+%b %e') $(date --date='-1 days' '+%b %e'); do
awk -v d="$day" '$0 ~ d' "$AUTH_FILE" | grep -E "sshd\[.*\]: Invalid user" || true
done
echo
echo "== SSH root or sudo session opens (last 48h) =="
for day in $(date --date='-0 days' '+%b %e') $(date --date='-1 days' '+%b %e'); do
awk -v d="$day" '$0 ~ d' "$AUTH_FILE" | grep -E "sshd\[.*\]: (pam_unix|session opened|Accepted .+ for root)" || true
done
} > "$OUT_DIR/ssh-logins-$TS.txt" 2>&1
# Compact summaries
{
echo "== Summary (counts, last 48h) =="
echo -n "Accepted logins: "
count=0
for day in $(date --date='-0 days' '+%b %e') $(date --date='-1 days' '+%b %e'); do
count=$((count + $(awk -v d="$day" '$0 ~ d' "$AUTH_FILE" | grep -E "sshd\[.*\]: Accepted (password|publickey)" | wc -l)))
done
echo "$count"
echo -n "Failed passwords: "
count=0
for day in $(date --date='-0 days' '+%b %e') $(date --date='-1 days' '+%b %e'); do
count=$((count + $(awk -v d="$day" '$0 ~ d' "$AUTH_FILE" | grep -E "sshd\[.*\]: Failed password" | wc -l)))
done
echo "$count"
echo -n "Invalid users: "
count=0
for day in $(date --date='-0 days' '+%b %e') $(date --date='-1 days' '+%b %e'); do
count=$((count + $(awk -v d="$day" '$0 ~ d' "$AUTH_FILE" | grep -E "sshd\[.*\]: Invalid user" | wc -l)))
done
echo "$count"
} > "$OUT_DIR/ssh-summary-$TS.txt" 2>&1
fi
# -----------------------------
# Permissions
# -----------------------------
chmod 640 "$OUT_DIR"/*"$TS".txt || true
echo "Security scans written to: $OUT_DIR (timestamp $TS)"Code complet
#!/usr/bin/env python3
import os
import sys
import glob
import shutil
import socket
import datetime
import zipfile
import requests
import json
import logging
from dotenv import load_dotenv
from langchain_openai import AzureChatOpenAI
from langchain.schema import SystemMessage, HumanMessage
from pathlib import Path
import smtplib
from email.message import EmailMessage
# ============================
# Constants & Configuration
# ============================
ARCHIVE_RETENTION_DAYS = 30
LOG_DIR = Path(os.getenv('LOG_DIR', '/var/log/security'))
OUT_DIR = Path(os.getenv('OUT_DIR', '/var/log/security'))
ARCHIVE_DIR = Path(os.getenv('ARCHIVE_DIR', '/var/log/archived'))
ENV_FILE = os.environ.get('ENV_FILE', './.env')
SYSTEM_PROMPT_FILE = Path('system_prompt.txt')
JUNIOR_SYSTEM_PROMPT_FILE = Path('junior_system_prompt.txt')
TIMESTAMP = datetime.datetime.now().strftime('%Y%m%d-%H%M%S')
REPORT_JSON = OUT_DIR / f'security-report-{TIMESTAMP}.json'
REPORT_TXT = OUT_DIR / f'security-report-{TIMESTAMP}.txt'
MAIL_TO = os.getenv('MAIL_TO', 'info@email.com')
MAIL_SUBJECT = os.getenv('MAIL_SUBJECT', f'Security Report {TIMESTAMP}')
MAIL_FROM = os.getenv('MAIL_FROM', f"security-reports@{socket.getfqdn() if hasattr(socket, 'getfqdn') else socket.gethostname()}")
REQUIRED_ENV = [
'AZURE_OPENAI_ENDPOINT',
'AZURE_OPENAI_DEPLOYMENT',
'API_VERSION',
'AZURE_TENANT_ID',
'AZURE_CLIENT_ID',
'AZURE_CLIENT_SECRET',
]
# ============================
# Logging Setup
# ============================
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[logging.StreamHandler(sys.stderr)]
)
# ============================
# Helper Functions
# ============================
def load_environment():
"""Load environment variables from dotenv file and check required ones."""
if Path(ENV_FILE).exists():
load_dotenv(dotenv_path=ENV_FILE)
logging.info(f"Loaded environment from {ENV_FILE}")
else:
logging.warning(f".env file not found at {ENV_FILE}; relying on current environment.")
missing_vars = [var for var in REQUIRED_ENV if not os.getenv(var)]
if missing_vars:
logging.error(f"Missing required environment variables: {', '.join(missing_vars)}")
sys.exit(1)
def ensure_directories():
"""Ensure log, output, and archive directories exist."""
if not LOG_DIR.is_dir():
logging.error(f"Log directory not found: {LOG_DIR}")
sys.exit(1)
OUT_DIR.mkdir(parents=True, exist_ok=True)
ARCHIVE_DIR.mkdir(parents=True, exist_ok=True)
def get_azure_ad_token():
"""Acquire Azure AD Token using client credentials."""
token_url = f"https://login.microsoftonline.com/{os.environ['AZURE_TENANT_ID']}/oauth2/v2.0/token"
token_data = {
"grant_type": "client_credentials",
"client_id": os.environ['AZURE_CLIENT_ID'],
"client_secret": os.environ['AZURE_CLIENT_SECRET'],
"scope": os.environ.get('AZURE_OPENAI_SCOPE', 'https://cognitiveservices.azure.com/.default'),
}
response = requests.post(token_url, data=token_data)
response.raise_for_status()
return response.json()["access_token"]
def load_prompt(prompt_file: Path) -> str:
"""Load a prompt from a file."""
if not prompt_file.is_file():
logging.error(f"{prompt_file} not found!")
sys.exit(1)
with prompt_file.open("r", encoding="utf-8") as f:
return f.read()
def collect_log_files() -> list:
"""Collect log files (*.log, *.txt) from LOG_DIR."""
log_files = list(LOG_DIR.glob("*.log")) + list(LOG_DIR.glob("*.txt"))
return log_files
def send_email_report(report_path: Path):
"""Send the TXT report as the email body using smtplib."""
SMTP_SERVER = os.getenv('SMTP_SERVER')
SMTP_PORT = int(os.getenv('SMTP_PORT', '587'))
SMTP_USERNAME = os.getenv('SMTP_USERNAME')
SMTP_PASSWORD = os.getenv('SMTP_PASSWORD')
if not SMTP_SERVER or not MAIL_TO or not MAIL_FROM:
logging.error("SMTP configuration or email addresses missing.")
logging.info(f"Reports saved at:\n JSON: {REPORT_JSON}\n Text: {REPORT_TXT}")
sys.exit(3)
try:
with report_path.open("r", encoding="utf-8") as f:
report_content = f.read()
msg = EmailMessage()
msg['Subject'] = MAIL_SUBJECT
msg['From'] = MAIL_FROM
msg['To'] = MAIL_TO
msg.set_content(report_content)
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
server.starttls()
if SMTP_USERNAME and SMTP_PASSWORD:
server.login(SMTP_USERNAME, SMTP_PASSWORD)
server.send_message(msg)
logging.info(f"Security report emailed to {MAIL_TO} via SMTP")
except Exception as e:
logging.error(f"Failed to email report via SMTP: {e}")
def archive_and_cleanup():
"""Archive logs and reports, and clean up old files."""
zip_name = f"security-logs-{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}.zip"
logs_to_archive = list(LOG_DIR.glob("*.txt")) + list(LOG_DIR.glob("*.log")) + [REPORT_JSON]
files_to_archive = [f for f in logs_to_archive if f.is_file()]
archive_path = ARCHIVE_DIR / zip_name
if files_to_archive:
try:
with zipfile.ZipFile(archive_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for file in files_to_archive:
zipf.write(file, arcname=file.name)
for file in files_to_archive:
file.unlink()
logging.info(f"Archived and removed: {' '.join(str(f) for f in files_to_archive)}")
except Exception as e:
logging.error(f"Error archiving files: {e}")
else:
logging.info(f"No .txt, .log, or .json files to archive in {LOG_DIR}.")
# Remove old archives
now = datetime.datetime.now().timestamp()
for f in ARCHIVE_DIR.glob("*.zip"):
try:
if f.stat().st_mtime < now - ARCHIVE_RETENTION_DAYS * 86400:
f.unlink()
except Exception as e:
logging.error(f"Error deleting archive {f}: {e}")
# Remove other files/directories in LOG_DIR except .log_offsets
for f in LOG_DIR.iterdir():
if f.name == '.log_offsets':
continue
try:
if f.is_dir():
shutil.rmtree(f)
else:
f.unlink()
except Exception as e:
logging.error(f"Error removing {f}: {e}")
def set_permissions():
"""Set secure permissions on report files."""
try:
REPORT_JSON.chmod(0o640)
REPORT_TXT.chmod(0o640)
except Exception as e:
logging.error(f"Error setting permissions on report files: {e}")
# ============================
# Main Functionality
# ============================
def main():
load_environment()
ensure_directories()
# Prepare LangChain AzureChatOpenAI
llm = AzureChatOpenAI(
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT"],
api_version=os.environ["API_VERSION"],
azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
azure_ad_token_provider=get_azure_ad_token,
temperature=0.7,
)
# Load prompts
final_system_prompt = load_prompt(SYSTEM_PROMPT_FILE)
per_log_system_prompt = load_prompt(JUNIOR_SYSTEM_PROMPT_FILE)
# Collect log files
log_files = collect_log_files()
individual_reports = []
# Query OpenAI for each log file
for log_file in log_files:
try:
with log_file.open("r", encoding="utf-8") as f:
log_content = f.read()
messages = [
SystemMessage(content=per_log_system_prompt),
HumanMessage(content=log_content)
]
response = llm(messages)
report_text = response.content if hasattr(response, 'content') else str(response)
individual_reports.append({
"log_file": log_file.name,
"report": report_text
})
except Exception as e:
logging.error(f"Error processing {log_file}: {e}")
# Save individual reports
individual_reports_txt = ""
for report in individual_reports:
individual_reports_txt += f"===== Report for {report['log_file']} =====\n{report['report']}\n\n"
try:
with REPORT_TXT.open("w", encoding="utf-8") as f:
f.write(individual_reports_txt)
except Exception as e:
logging.error(f"Error writing individual reports: {e}")
# Final aggregation OpenAI call
reports_content = "\n".join(
f"Report for {r['log_file']}:\n{r['report']}" for r in individual_reports
)
final_messages = [
SystemMessage(content=final_system_prompt),
HumanMessage(content=reports_content)
]
try:
final_response = llm(final_messages)
final_report = final_response.content if hasattr(final_response, 'content') else str(final_response)
except Exception as e:
logging.error(f"Error during final aggregation OpenAI call: {e}")
final_report = "Aggregation failed due to error."
# Append final aggregated report
try:
with REPORT_TXT.open("a", encoding="utf-8") as f:
f.write("\n===== FINAL AGGREGATED REPORT =====\n")
f.write(final_report)
with REPORT_JSON.open("w", encoding="utf-8") as f:
json.dump({
"individual_reports": individual_reports,
"final_report": final_report
}, f, indent=2)
except Exception as e:
logging.error(f"Error saving final report: {e}")
set_permissions()
send_email_report(REPORT_TXT)
archive_and_cleanup()
if __name__ == "__main__":
main()Résumé
En orchestrant des outils open source et en s’appuyant sur un grand modèle de langage (LLM) pour analyser les logs et en faire des rapports, ce script transforme un ensemble d’utilitaires en une plateforme de sécurité propulsée par l’IA. L’approche automatise la revue des logs, met les constats en contexte et tient les administrateurs informés, le tout avec un minimum d’intervention manuelle.
Cependant, n’oubliez pas :
- L’IA peut passer à côté de menaces avancées et doit être une aide, pas l’unique moyen de défense.
- Protégez les fichiers et les identifiants.
- Validez et relisez les rapports.
- Surveillez le système et ajustez les paramètres selon la charge et le volume des logs.
Cette architecture est un exemple concret de la façon dont l’IA peut renforcer la gestion d’infrastructure, en apportant clarté, efficacité et informations exploitables dans le monde bruyant des logs de sécurité.
AI Platform Engineer & Full-Stack Developer
Building Enterprise Integrations & Automations
- Ce qu'est vraiment un LLM — et comment le mesurer avec Artificial Analysis - 12 septembre 2026
- AI SDLC : un recueil de bonnes pratiques pour développer avec l'IA - 19 juin 2026
- Agentic Patterns et les multiples visages du RAG - 25 mai 2026