Como criar uma suíte de segurança com IA a partir de ferramentas open source
Traduzido do original em espanhol. Ler em espanhol
Nossa configuração consiste em um servidor Linux executando várias ferramentas open source de segurança: fail2ban, ClamAV, rkhunter, maldet e lynis. Cada ferramenta grava sua saída de logs em um diretório comum, junto com um script em bash que agrega eventos do fail2ban, do SSH e de outras fontes. Para dar sentido a esses dados, usamos um script em Python que emprega o Azure OpenAI por meio da biblioteca LangChain.
Disclaimer:
Este é apenas um exemplo. Se quiser usá-lo em um ambiente de produção, primeiro analise seus requisitos específicos, faça testes exaustivos e adapte o código às necessidades e políticas da sua organização. Não use este código como está sem antes revisá-lo e ajustá-lo ao seu ambiente. Proteja com cuidado os arquivos de configuração, as credenciais e os relatórios gerados.
O que são essas ferramentas e por que executá-las juntas?
Neste exemplo, combinamos várias ferramentas open source amplamente reconhecidas para a segurança de servidores Linux:
- fail2ban: Monitora os logs de serviços como o SSH para detectar tentativas de acesso malsucedidas ou ataques de força bruta e bloqueia automaticamente os IPs ofensivos por meio de regras de firewall.
- ClamAV: Um antivírus open source que analisa arquivos em busca de malware, vírus e trojans conhecidos.
- rkhunter (Rootkit Hunter): Procura rootkits, backdoors e exploits locais examinando arquivos de sistema, binários e configurações suspeitas.
- maldet (Linux Malware Detect): Especializado em detectar malware conhecido e ameaças web em ambientes Linux, especialmente útil para servidores web compartilhados.
- lynis: Uma ferramenta de auditoria de segurança e conformidade que analisa a configuração, procura vulnerabilidades e oferece recomendações para reforçar o sistema.
Por que executá-las juntas?
Cada uma dessas ferramentas cobre um aspecto diferente da segurança do sistema: da detecção de intrusões e do bloqueio de acessos não autorizados até a análise de malware, rootkits e auditorias de configuração. Ao executar todas juntas e combinar seus resultados, obtemos uma visão muito mais completa e profunda do estado de segurança do servidor. Além disso, centralizar os logs e analisá-los com inteligência artificial permite correlacionar eventos, descobrir padrões que poderiam passar despercebidos e priorizar ações de resposta de forma mais eficiente.
1. Carregamento do ambiente e configuração
O script começa carregando as variáveis de ambiente de um arquivo .env (se existir) ou do próprio ambiente do sistema. Depois verifica se todas as variáveis necessárias estão presentes, interrompendo a execução se faltar alguma.
Prática recomendada:
Garanta que o arquivo .env e qualquer arquivo de configuração estejam bem protegidos e acessíveis apenas ao usuário necessário. Considere criptografar os arquivos que contêm credenciais.
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. Verificação e criação de diretórios
Garante que os diretórios de logs, saída e arquivamento existam, criando-os se necessário. Se o diretório de logs não existir, o script é interrompido.
Importante:
Verifique as permissões dos diretórios e segmente o acesso para evitar que usuários não autorizados manipulem os arquivos de log e os relatórios.
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. Autenticação no Azure
A função get_azure_ad_token obtém um token de acesso do Azure Active Directory usando client credentials.
Nota:
Garanta que as permissões no Azure sejam as mínimas necessárias para o seu caso de uso. As credenciais usadas devem ser armazenadas e transmitidas de forma segura.
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. Carregamento dos prompts para a IA
O script carrega dois prompts a partir de arquivos de texto: um para a análise individual de cada log e outro para o relatório final agregado.
Importante:
O conteúdo dos prompts determina a qualidade e o foco dos relatórios. Revise e ajuste os prompts de acordo com o nível técnico e as necessidades da sua organização.
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. Coleta dos logs
Coleta todos os arquivos .log e .txt do diretório de logs para análise posterior.
Nota:
Garanta que apenas os logs esperados estejam no diretório, para evitar analisar arquivos irrelevantes ou potencialmente maliciosos.
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. Análise de IA de cada log
Lê cada log, envia-o à IA junto com o prompt individual e armazena o resultado.
Aviso:
A IA pode cometer erros de interpretação ou não detectar ameaças avançadas. Ela não substitui a supervisão e a verificação humanas em contextos críticos.
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. Gravação dos relatórios individuais
Grava os relatórios individuais gerados pela IA em um arquivo de texto.
Boa prática:
Revise os relatórios periodicamente e mantenha controle de versões se sua organização exigir.
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. Relatório final agregado pela IA
A IA recebe todos os relatórios individuais e gera uma análise ou resumo global.
Importante:
Valide os resultados da IA e leve em conta suas limitações. Considere usar a saída como apoio, e não como decisão automática final.
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. Gravação e permissões dos relatórios
O relatório final e os individuais são armazenados em arquivos .txt e .json, com as permissões adequadas.
Recomendação:
Considere criptografar os relatórios se contiverem informações sensíveis e configure backups automáticos em um ambiente seguro.
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. Envio automático por e-mail (SMTP)
Envia o relatório de segurança por e-mail usando SMTP.
Aviso:
Garanta o uso de senhas de aplicativo e conexões TLS/SSL. Não use contas pessoais nem exponha credenciais em arquivos inseguros.
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. Arquivamento e limpeza
Todos os logs e relatórios são compactados em um arquivo ZIP, e os arquivos antigos são excluídos de acordo com a política de retenção.
Importante:
Verifique regularmente o espaço em disco e o estado dos arquivos arquivados, e monitore o desempenho do sistema, especialmente se o volume de logs for alto.
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 utilizados
// 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):
*/Exemplo: script Bash para criar logs personalizados de atividade para análise
- Troque
/var/log/fail2ban.logetc. pelos caminhos de log da sua distribuição, se necessário. - Por privacidade, evite executar ou compartilhar em sistemas com dados sensíveis ou regulados sem a revisão adequada.
#!/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)"Código completo
#!/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()Resumo
Ao orquestrar ferramentas open source e aproveitar um modelo de linguagem (LLM) para a análise e o relatório dos logs, este script transforma um conjunto de utilitários em uma plataforma de segurança com IA. A abordagem automatiza a revisão dos logs, contextualiza as descobertas e mantém os administradores informados, tudo com o mínimo de intervenção manual.
No entanto, lembre-se:
- A IA pode não detectar ameaças avançadas e deve ser uma ajuda, não o único método de defesa.
- Proteja os arquivos e as credenciais.
- Valide e revise os relatórios.
- Monitore o sistema e ajuste os parâmetros de acordo com a carga e o tamanho dos logs.
Esta arquitetura é um exemplo prático de como a IA pode potencializar a gestão de infraestrutura, trazendo clareza, eficiência e inteligência acionável ao mundo ruidoso dos logs de segurança.
AI Platform Engineer & Full-Stack Developer
Building Enterprise Integrations & Automations
- O que é realmente um LLM — e como medi-lo com o Artificial Analysis - 12 de setembro de 2026
- AI SDLC: um compêndio de boas práticas para desenvolver com IA - 19 de junho de 2026
- Agentic Patterns e as muitas faces do RAG - 25 de maio de 2026