ES

Cómo crear una suite de seguridad con IA a partir de herramientas open source

Nuestra configuración consiste en un servidor Linux ejecutando varias herramientas open source de seguridad: fail2ban, ClamAV, rkhunter, maldet y lynis. Cada herramienta escribe su salida de logs en un directorio común, junto con un script en bash que agrega eventos de fail2ban, SSH y otras fuentes. Para dar sentido a estos datos, utilizamos un script en Python que emplea Azure OpenAI usando la librería Langchain.

Disclaimer:
Este es solo un ejemplo. Si deseas usarlo en un entorno de producción, primero analiza tus requerimientos específicos, realiza pruebas exhaustivas y adapta el código según las necesidades y políticas de tu organización. No uses este código tal cual sin antes revisarlo y ajustarlo a tu entorno. Asegúrate de proteger cuidadosamente archivos de configuración, credenciales y reportes generados.

iStock AI Generator

¿Qué son estas herramientas y por qué ejecutarlas juntas?

En este ejemplo, combinamos varias herramientas open source ampliamente reconocidas para la seguridad en servidores Linux:

  • fail2ban: Monitorea los logs de servicios como SSH para detectar intentos de acceso fallidos o ataques de fuerza bruta, y automáticamente bloquea las IPs ofensivas mediante reglas de firewall.
  • ClamAV: Un antivirus open source que escanea archivos en busca de malware, virus y troyanos conocidos.
  • rkhunter (Rootkit Hunter): Busca rootkits, puertas traseras y exploits locales examinando archivos de sistema, binarios y configuraciones sospechosas.
  • maldet (Linux Malware Detect): Especializado en detectar malware conocido y amenazas web en entornos Linux, especialmente útil para servidores web compartidos.
  • lynis: Una herramienta de auditoría de seguridad y cumplimiento que realiza análisis de configuración, busca vulnerabilidades y entrega recomendaciones para reforzar el sistema.

¿Por qué ejecutarlas juntas?
Cada una de estas herramientas aborda un aspecto diferente de la seguridad del sistema: desde la detección de intrusiones y bloqueo de accesos no autorizados, hasta el análisis de malware, rootkits y auditorías de configuración. Al ejecutar todas juntas y combinar sus resultados, obtenemos una visión mucho más completa y profunda del estado de seguridad del servidor. Además, centralizar los logs y analizarlos con inteligencia artificial permite correlacionar eventos, descubrir patrones que podrían pasar desapercibidos y priorizar acciones de respuesta de forma más eficiente.

1. Carga de entorno y configuración

El script comienza cargando las variables de entorno desde un archivo .env (si existe) o desde el propio entorno del sistema. Luego verifica que todas las variables requeridas estén presentes, deteniendo la ejecución si falta alguna.

Práctica recomendada:
Asegúrate de que el archivo .env y cualquier archivo de configuración estén correctamente protegidos y accesibles solo para el usuario necesario. Considera cifrar los archivos que contengan credenciales.

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. Comprobación y creación de directorios

Asegura que los directorios de logs, salida y archivo existan, creándolos si es necesario. Si falta el directorio de logs, el script se detiene.

Importante:
Verifica permisos de directorio y segmenta el acceso para evitar que usuarios no autorizados puedan manipular los archivos de log y reportes.

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. Autenticación con Azure

La función get_azure_ad_token obtiene un token de acceso desde Azure Active Directory usando credenciales de cliente.

Nota:
Asegúrate de que los permisos en Azure sean los mínimos necesarios para tu caso de uso. Las credenciales usadas deben almacenarse y transmitirse 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. Carga de prompts para la IA

El script carga dos prompts desde archivos de texto: uno para el análisis individual de cada log y otro para el informe final agregado.

Importante:
El contenido de los prompts determinará la calidad y el enfoque de los reportes. Revisa y ajusta los prompts según el nivel técnico y las necesidades de tu organización.

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. Recopilación de logs

Recoge todos los archivos .log y .txt del directorio de logs para su posterior análisis.

Nota:
Asegúrate de que solo los logs esperados se encuentren en el directorio, para evitar analizar archivos irrelevantes o 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_files

6. Análisis IA por cada log

Lee cada log, lo envía a la IA junto con el prompt individual y almacena el resultado.

Advertencia:
La IA puede cometer errores de interpretación o no detectar amenazas avanzadas. No reemplaza la supervisión y verificación humana en 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. Guardado de informes individuales

Guarda los reportes individuales generados por la IA en un archivo de texto.

Mejor práctica:
Revisa los reportes periódicamente y mantén un control de versiones si tu organización lo requiere.

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. Informe final agregado por IA

La IA recibe todos los reportes individuales y genera un análisis o resumen global.

Importante:
Valida los resultados de la IA y ten en cuenta sus limitaciones. Considera usar la salida como apoyo y no como decisión 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. Guardado y permisos de reportes

El reporte final y los individuales se almacenan en archivos .txt y .json, asegurando los permisos adecuados.

Recomendación:
Considera cifrar los reportes si contienen información sensible y configura backups automáticos en un entorno 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. Envío automático por correo electrónico (SMTP)

Envía el informe de seguridad por correo usando SMTP.

Advertencia:
Asegúrate de usar contraseñas de aplicación y conexiones TLS/SSL. No uses cuentas personales ni expongas credenciales en archivos 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. Archivado y limpieza

Todos los logs y reportes son comprimidos en un archivo ZIP y los archivos antiguos se eliminan según la política de retención.

Importante:
Verifica regularmente el espacio en disco, el estado de los archivos archivados y monitorea el rendimiento del sistema, especialmente si el volumen de logs es 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):
*/

Ejemplo: Script Bash para crear logs personalizados de actividad para análisis

  • Cambia /var/log/fail2ban.log, etc. Según las rutas de logs de tu distribución si es necesario.
  • Por privacidad, evita ejecutar o compartir en sistemas con datos sensibles o regulados sin la revisión adecuada.
#!/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()

Resumen

Al orquestar herramientas open source y aprovechar un modelo de lenguaje (LLM) para el análisis y reporte de logs, este script transforma un conjunto de utilidades en una plataforma de seguridad potenciada por IA. El enfoque automatiza la revisión de logs, contextualiza los hallazgos y mantiene informados a los administradores, todo ello con mínima intervención manual.

Sin embargo, recuerda:

  • La IA puede no detectar amenazas avanzadas y debe ser una ayuda, no el único método de defensa.
  • Protege los archivos y credenciales.
  • Valida y revisa los reportes.
  • Monitorea el sistema y ajusta parámetros según la carga y el tamaño de los logs.

Esta arquitectura es un ejemplo práctico de cómo la IA puede potenciar la gestión de infraestructura, aportando claridad, eficiencia e inteligencia accionable en el ruidoso mundo de los logs de seguridad.

Maximiliano Díaz Doglia

AI Platform Engineer & Full-Stack Developer
Building Enterprise Integrations & Automations