APIs de SIEM: Automatizar Búsquedas en Splunk y Elasticsearch

Por qué automatizar las búsquedas en el SIEM

El SIEM es el sistema nervioso central del SOC. Splunk, Elasticsearch (ELK/Elastic Security), Microsoft Sentinel y otros ingieren millones de eventos diarios de firewalls, endpoints, servidores y aplicaciones. Los analistas buscan amenazas manualmente escribiendo queries, revisando dashboards y respondiendo alertas.

El problema: las búsquedas manuales no escalan. Un threat hunter necesita ejecutar la misma query en ventanas de tiempo distintas, cruzar resultados con feeds de IOCs, generar informes y repetir el proceso para cada hipótesis. Con la API del SIEM, todo esto se programa una vez y se ejecuta automáticamente.

Este artículo cubre las dos APIs más utilizadas: Splunk REST API y Elasticsearch API, con ejemplos prácticos de threat hunting automatizado.

Splunk REST API

Splunk expone una API REST completa en el puerto 8089 (management). Cada búsqueda se gestiona como un "search job": se crea, se espera a que termine y se recogen los resultados.

Autenticación y configuración

"""
splunk_client.py — Cliente Python para Splunk REST API
"""
import time
import requests
from dataclasses import dataclass
from urllib3.exceptions import InsecureRequestWarning

# Splunk usa certificados self-signed por defecto
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)


@dataclass
class SplunkConfig:
    host: str
    port: int = 8089
    username: str = ""
    password: str = ""
    token: str = ""   # Bearer token (preferible)
    verify_ssl: bool = False


class SplunkClient:
    """Cliente para Splunk REST API con manejo de search jobs."""

    def __init__(self, config: SplunkConfig):
        self.base_url = f"https://{config.host}:{config.port}"
        self.verify = config.verify_ssl
        self.session = requests.Session()

        if config.token:
            self.session.headers["Authorization"] = (
                f"Bearer {config.token}"
            )
        else:
            # Autenticacion por usuario/password
            resp = self.session.post(
                f"{self.base_url}/services/auth/login",
                data={
                    "username": config.username,
                    "password": config.password,
                    "output_mode": "json",
                },
                verify=self.verify,
            )
            resp.raise_for_status()
            session_key = resp.json()["sessionKey"]
            self.session.headers["Authorization"] = (
                f"Splunk {session_key}"
            )

    def search(
        self,
        query: str,
        earliest: str = "-24h",
        latest: str = "now",
        max_results: int = 1000,
        timeout_seconds: int = 300,
    ) -> list[dict]:
        """Ejecutar una busqueda SPL y devolver resultados."""
        # Asegurar que la query empieza con 'search'
        if not query.strip().startswith("search") and \
           not query.strip().startswith("|"):
            query = f"search {query}"

        # Crear search job
        resp = self.session.post(
            f"{self.base_url}/services/search/jobs",
            data={
                "search": query,
                "earliest_time": earliest,
                "latest_time": latest,
                "output_mode": "json",
                "max_count": max_results,
            },
            verify=self.verify,
        )
        resp.raise_for_status()
        job_id = resp.json()["sid"]

        # Esperar a que termine
        start_time = time.time()
        while True:
            elapsed = time.time() - start_time
            if elapsed > timeout_seconds:
                # Cancelar el job
                self.session.delete(
                    f"{self.base_url}/services/search/jobs/{job_id}",
                    verify=self.verify,
                )
                raise TimeoutError(
                    f"Search job {job_id} exceeded "
                    f"{timeout_seconds}s timeout"
                )

            status_resp = self.session.get(
                f"{self.base_url}/services/search/jobs/{job_id}",
                params={"output_mode": "json"},
                verify=self.verify,
            )
            status = status_resp.json()["entry"][0]["content"]

            if status["isDone"]:
                break

            time.sleep(2)

        # Recoger resultados
        results_resp = self.session.get(
            f"{self.base_url}/services/search/jobs/{job_id}/results",
            params={
                "output_mode": "json",
                "count": max_results,
            },
            verify=self.verify,
        )
        results_resp.raise_for_status()

        return results_resp.json().get("results", [])

Búsquedas de threat hunting con Splunk

"""
splunk_hunting.py — Queries de threat hunting automatizadas
"""

HUNTING_QUERIES = {
    "suspicious_powershell": {
        "description": "PowerShell con argumentos codificados en Base64",
        "query": (
            'search index=windows sourcetype=WinEventLog:Security '
            'EventCode=4688 '
            'New_Process_Name="*powershell.exe" '
            '(CommandLine="*-enc*" OR CommandLine="*-EncodedCommand*" '
            'OR CommandLine="*-e *" OR CommandLine="*FromBase64*") '
            '| stats count by Computer, Account_Name, CommandLine '
            '| sort -count'
        ),
        "severity": "high",
    },
    "unusual_outbound_connections": {
        "description": "Conexiones salientes a puertos no estandar",
        "query": (
            'search index=network sourcetype=firewall action=allowed '
            'dest_port!=80 dest_port!=443 dest_port!=53 '
            'dest_port!=22 dest_port!=25 dest_port!=123 '
            'direction=outbound '
            '| stats count dc(dest_ip) as unique_dests '
            'by src_ip dest_port '
            '| where count > 100 OR unique_dests > 10 '
            '| sort -count'
        ),
        "severity": "medium",
    },
    "failed_login_brute_force": {
        "description": "Multiples intentos fallidos de login",
        "query": (
            'search index=windows sourcetype=WinEventLog:Security '
            'EventCode=4625 '
            '| stats count dc(TargetUserName) as unique_users '
            'by IpAddress '
            '| where count > 50 OR unique_users > 5 '
            '| sort -count'
        ),
        "severity": "medium",
    },
    "dns_tunneling": {
        "description": "Posible exfiltracion via DNS (queries largas)",
        "query": (
            'search index=dns '
            '| eval query_length=len(query) '
            '| where query_length > 50 '
            '| stats count avg(query_length) as avg_len '
            'by src_ip query_type '
            '| where count > 100 AND avg_len > 60 '
            '| sort -count'
        ),
        "severity": "high",
    },
}


def run_hunting_campaign(
    client: SplunkClient,
    timerange: str = "-24h",
) -> dict[str, list[dict]]:
    """Ejecutar todas las queries de hunting y recopilar resultados."""
    results = {}

    for hunt_name, hunt_config in HUNTING_QUERIES.items():
        print(
            f"[*] Running: {hunt_name} "
            f"- {hunt_config['description']}"
        )
        try:
            hits = client.search(
                hunt_config["query"],
                earliest=timerange,
            )
            results[hunt_name] = {
                "description": hunt_config["description"],
                "severity": hunt_config["severity"],
                "hit_count": len(hits),
                "results": hits,
            }
            if hits:
                print(
                    f"    [!] {len(hits)} results "
                    f"(severity: {hunt_config['severity']})"
                )
            else:
                print("    [OK] No results")
        except Exception as e:
            print(f"    [ERROR] {e}")
            results[hunt_name] = {
                "error": str(e),
                "hit_count": 0,
            }

    return results

Elasticsearch API

Elasticsearch (la base de datos detrás de ELK y Elastic Security) expone una API REST en el puerto 9200. Las búsquedas usan el formato Query DSL (JSON).

Cliente Python para Elasticsearch

"""
elastic_client.py — Cliente Python para Elasticsearch
"""
from elasticsearch import Elasticsearch
from datetime import datetime, timedelta
from dataclasses import dataclass


@dataclass
class ElasticConfig:
    hosts: list[str]
    api_key: str = ""
    username: str = ""
    password: str = ""
    verify_certs: bool = True
    ca_certs: str = ""


def create_elastic_client(config: ElasticConfig) -> Elasticsearch:
    """Crear cliente Elasticsearch con autenticacion."""
    kwargs = {
        "hosts": config.hosts,
        "verify_certs": config.verify_certs,
    }

    if config.api_key:
        kwargs["api_key"] = config.api_key
    elif config.username:
        kwargs["basic_auth"] = (
            config.username, config.password
        )

    if config.ca_certs:
        kwargs["ca_certs"] = config.ca_certs

    return Elasticsearch(**kwargs)


class ElasticHunter:
    """Clase para threat hunting en Elasticsearch."""

    def __init__(self, client: Elasticsearch):
        self.es = client

    def search(
        self,
        index: str,
        query: dict,
        size: int = 1000,
        sort: list = None,
        aggs: dict = None,
        time_field: str = "@timestamp",
        hours_back: int = 24,
    ) -> dict:
        """Busqueda con filtro temporal automatico."""
        now = datetime.utcnow()
        time_from = now - timedelta(hours=hours_back)

        # Envolver la query en un filtro temporal
        body = {
            "query": {
                "bool": {
                    "must": [query],
                    "filter": [
                        {
                            "range": {
                                time_field: {
                                    "gte": time_from.isoformat(),
                                    "lte": now.isoformat(),
                                }
                            }
                        }
                    ],
                }
            },
            "size": size,
        }

        if sort:
            body["sort"] = sort
        if aggs:
            body["aggs"] = aggs

        return self.es.search(index=index, body=body)

    def count(
        self, index: str, query: dict, hours_back: int = 24
    ) -> int:
        """Contar documentos que coinciden con la query."""
        now = datetime.utcnow()
        time_from = now - timedelta(hours=hours_back)

        body = {
            "query": {
                "bool": {
                    "must": [query],
                    "filter": [
                        {
                            "range": {
                                "@timestamp": {
                                    "gte": time_from.isoformat(),
                                }
                            }
                        }
                    ],
                }
            }
        }

        result = self.es.count(index=index, body=body)
        return result["count"]

Queries de threat hunting en Elasticsearch

"""
elastic_hunting.py — Queries de hunting para Elasticsearch
"""

ELASTIC_HUNTS = {
    "process_injection_sysmon": {
        "description": "Posible process injection (Sysmon EventID 8)",
        "index": "winlogbeat-*",
        "query": {
            "bool": {
                "must": [
                    {"match": {"event.code": "8"}},
                    {"match": {"winlog.event_data.SourceImage": "*"}}
                ],
                "must_not": [
                    {
                        "terms": {
                            "winlog.event_data.SourceImage.keyword": [
                                "C:\\Windows\\System32\\csrss.exe",
                                "C:\\Windows\\System32\\lsass.exe",
                                "C:\\Windows\\System32\\services.exe",
                            ]
                        }
                    }
                ],
            }
        },
        "aggs": {
            "by_source": {
                "terms": {
                    "field": (
                        "winlog.event_data.SourceImage.keyword"
                    ),
                    "size": 20,
                }
            }
        },
    },
    "rare_parent_child": {
        "description": "Procesos padre-hijo inusuales",
        "index": "winlogbeat-*",
        "query": {
            "bool": {
                "must": [
                    {"match": {"event.code": "1"}},
                ],
            }
        },
        "aggs": {
            "parent_child_pairs": {
                "composite": {
                    "size": 50,
                    "sources": [
                        {
                            "parent": {
                                "terms": {
                                    "field": "process.parent.name"
                                }
                            }
                        },
                        {
                            "child": {
                                "terms": {
                                    "field": "process.name"
                                }
                            }
                        },
                    ],
                }
            }
        },
    },
    "lateral_movement_smb": {
        "description": "Movimiento lateral via SMB/admin shares",
        "index": "packetbeat-*",
        "query": {
            "bool": {
                "must": [
                    {"match": {"network.protocol": "smb"}},
                    {"match": {"destination.port": 445}},
                ],
                "must_not": [
                    {
                        "terms": {
                            "source.ip": [
                                "10.0.0.1",  # DC
                                "10.0.0.2",  # File server
                            ]
                        }
                    }
                ],
            }
        },
        "aggs": {
            "by_source": {
                "terms": {
                    "field": "source.ip",
                    "size": 20,
                },
                "aggs": {
                    "unique_dests": {
                        "cardinality": {
                            "field": "destination.ip"
                        }
                    }
                },
            }
        },
    },
}


def run_elastic_hunting(
    hunter: ElasticHunter,
    hours_back: int = 24,
) -> dict:
    """Ejecutar campana de hunting en Elasticsearch."""
    results = {}

    for hunt_name, config in ELASTIC_HUNTS.items():
        print(f"[*] Running: {hunt_name} - {config['description']}")
        try:
            resp = hunter.search(
                index=config["index"],
                query=config["query"],
                aggs=config.get("aggs"),
                hours_back=hours_back,
            )

            hit_count = resp["hits"]["total"]["value"]
            results[hunt_name] = {
                "description": config["description"],
                "total_hits": hit_count,
                "hits": [
                    h["_source"] for h in resp["hits"]["hits"][:10]
                ],
                "aggregations": resp.get("aggregations", {}),
            }

            if hit_count > 0:
                print(f"    [!] {hit_count} results found")
            else:
                print("    [OK] No results")

        except Exception as e:
            print(f"    [ERROR] {e}")
            results[hunt_name] = {"error": str(e)}

    return results

Búsquedas programadas

Una de las ventajas clave de la automatización es la ejecución programada. En lugar de que un analista ejecute queries manualmente cada mañana, un script las ejecuta cada hora y alerta solo cuando encuentra algo.

"""
scheduled_hunting.py — Hunting automatizado con scheduling
"""
import json
import schedule
import time
from datetime import datetime
from pathlib import Path


class HuntingScheduler:
    """Programador de campanas de hunting automatizadas."""

    def __init__(
        self,
        splunk_client=None,
        elastic_hunter=None,
        output_dir: str = "./hunting_results",
        alert_callback=None,
    ):
        self.splunk = splunk_client
        self.elastic = elastic_hunter
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
        self.alert_callback = alert_callback

    def run_and_report(
        self, platform: str, hours_back: int = 1
    ):
        """Ejecutar hunting y generar reporte."""
        timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")

        if platform == "splunk" and self.splunk:
            results = run_hunting_campaign(
                self.splunk, f"-{hours_back}h"
            )
        elif platform == "elastic" and self.elastic:
            results = run_elastic_hunting(
                self.elastic, hours_back
            )
        else:
            print(f"[ERROR] Platform {platform} not configured")
            return

        # Guardar resultados
        output_file = (
            self.output_dir / f"hunt_{platform}_{timestamp}.json"
        )
        with open(output_file, "w") as f:
            json.dump(results, f, indent=2, default=str)

        # Alertar si hay resultados de severidad alta
        for hunt_name, hunt_result in results.items():
            hit_count = hunt_result.get(
                "hit_count",
                hunt_result.get("total_hits", 0)
            )
            if hit_count > 0 and self.alert_callback:
                self.alert_callback(
                    hunt_name=hunt_name,
                    hit_count=hit_count,
                    details=hunt_result,
                )

        print(f"[+] Results saved to {output_file}")


def start_scheduled_hunting(scheduler: HuntingScheduler):
    """Iniciar hunting programado."""
    # Cada hora: hunting rapido (ultima hora)
    schedule.every(1).hours.do(
        scheduler.run_and_report, "splunk", 1
    )
    schedule.every(1).hours.do(
        scheduler.run_and_report, "elastic", 1
    )

    # Cada 6 horas: hunting profundo (ultimas 6 horas)
    schedule.every(6).hours.do(
        scheduler.run_and_report, "splunk", 6
    )

    # Diario: hunting amplio (ultimas 24 horas)
    schedule.every().day.at("06:00").do(
        scheduler.run_and_report, "elastic", 24
    )

    print("[*] Hunting scheduler started")
    print("    Hourly: quick hunts (1h window)")
    print("    Every 6h: deep hunts (6h window)")
    print("    Daily 06:00: broad hunts (24h window)")

    while True:
        schedule.run_pending()
        time.sleep(60)

Pipeline de enriquecimiento de alertas

Cuando el SIEM genera una alerta, el analista necesita contexto adicional. La API permite automatizar el enriquecimiento: buscar eventos relacionados, consultar IOCs y agregar información de threat intelligence.

"""
alert_enrichment.py — Enriquecimiento automatico de alertas SIEM
"""
from dataclasses import dataclass


@dataclass
class EnrichedAlert:
    original_alert: dict
    related_events: list[dict]
    threat_intel: dict
    risk_score: int
    recommended_action: str


def enrich_alert(
    alert: dict,
    splunk_client: "SplunkClient" = None,
    elastic_hunter: "ElasticHunter" = None,
) -> EnrichedAlert:
    """Enriquecer una alerta con contexto adicional del SIEM."""
    enriched = EnrichedAlert(
        original_alert=alert,
        related_events=[],
        threat_intel={},
        risk_score=0,
        recommended_action="investigate",
    )

    src_ip = alert.get("src_ip", "")
    dest_ip = alert.get("dest_ip", "")
    username = alert.get("user", "")
    hostname = alert.get("host", "")

    # Buscar otros eventos del mismo origen en la ultima hora
    if src_ip and splunk_client:
        related = splunk_client.search(
            f'search index=* src_ip="{src_ip}" '
            f'| stats count by sourcetype, action '
            f'| sort -count',
            earliest="-1h",
        )
        enriched.related_events.extend(related)

    # Buscar intentos de login del mismo usuario
    if username and splunk_client:
        login_events = splunk_client.search(
            f'search index=windows EventCode=4624 OR EventCode=4625 '
            f'TargetUserName="{username}" '
            f'| stats count by EventCode, IpAddress '
            f'| sort -count',
            earliest="-24h",
        )
        enriched.threat_intel["login_activity"] = login_events

    # Buscar actividad del host
    if hostname and elastic_hunter:
        host_activity = elastic_hunter.search(
            index="winlogbeat-*",
            query={
                "match": {"host.name": hostname}
            },
            aggs={
                "by_event": {
                    "terms": {
                        "field": "event.code",
                        "size": 20,
                    }
                }
            },
            hours_back=1,
        )
        enriched.threat_intel["host_events"] = (
            host_activity.get("aggregations", {})
        )

    # Calcular risk score basado en contexto
    event_count = len(enriched.related_events)
    if event_count > 50:
        enriched.risk_score += 30
    elif event_count > 20:
        enriched.risk_score += 15

    # Multiples IPs destino = posible scan o lateral movement
    unique_dests = set()
    for event in enriched.related_events:
        if "dest_ip" in event:
            unique_dests.add(event["dest_ip"])
    if len(unique_dests) > 5:
        enriched.risk_score += 25

    # Recomendacion basada en score
    if enriched.risk_score > 50:
        enriched.recommended_action = "escalate_to_n2"
    elif enriched.risk_score > 25:
        enriched.recommended_action = "investigate"
    else:
        enriched.recommended_action = "monitor"

    return enriched

Extracción de datos para dashboards

Otro caso de uso frecuente: extraer métricas del SIEM para dashboards personalizados fuera de la plataforma (Grafana, un portal interno, informes ejecutivos).

"""
dashboard_data.py — Extraccion de metricas para dashboards
"""
from datetime import datetime, timedelta


def get_soc_metrics(
    client: "SplunkClient",
    days: int = 7,
) -> dict:
    """Extraer metricas SOC para dashboard ejecutivo."""
    metrics = {}

    # Total de alertas por dia
    alerts = client.search(
        'search index=notable '
        '| timechart span=1d count by urgency',
        earliest=f"-{days}d",
    )
    metrics["alerts_by_day"] = alerts

    # MTTR (Mean Time To Respond)
    mttr = client.search(
        'search index=notable status=closed '
        '| eval response_time = '
        'round((info_max_time - info_min_time) / 60, 1) '
        '| stats avg(response_time) as avg_mttr '
        'median(response_time) as median_mttr '
        'p95(response_time) as p95_mttr',
        earliest=f"-{days}d",
    )
    metrics["mttr"] = mttr

    # Top 10 reglas que mas alertan
    top_rules = client.search(
        'search index=notable '
        '| stats count by search_name '
        '| sort -count | head 10',
        earliest=f"-{days}d",
    )
    metrics["top_rules"] = top_rules

    # Alertas por analista asignado
    by_analyst = client.search(
        'search index=notable status=* '
        '| stats count by owner, status '
        '| sort -count',
        earliest=f"-{days}d",
    )
    metrics["by_analyst"] = by_analyst

    return metrics

Construir un framework de hunting

Cuando tienes decenas de queries de hunting, necesitas un framework que las organice, las ejecute y gestione los resultados de forma estructurada.

"""
hunting_framework.py — Framework minimo de hunting automatizado
"""
import json
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from pathlib import Path


class HuntStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"


class HuntSeverity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass
class HuntDefinition:
    """Definicion de una query de hunting."""
    name: str
    description: str
    mitre_technique: str    # T1xxx
    platform: str           # splunk | elastic
    query: str | dict       # SPL string o Elastic DSL dict
    severity: HuntSeverity
    index: str = ""
    tags: list[str] = field(default_factory=list)
    false_positive_notes: str = ""
    response_action: str = ""


@dataclass
class HuntResult:
    """Resultado de una ejecucion de hunting."""
    hunt_name: str
    status: HuntStatus
    timestamp: str
    hit_count: int = 0
    results: list[dict] = field(default_factory=list)
    duration_seconds: float = 0
    error: str = ""


class HuntingFramework:
    """Framework para gestionar campanas de hunting."""

    def __init__(
        self,
        splunk_client=None,
        elastic_hunter=None,
        results_dir: str = "./hunt_results",
    ):
        self.splunk = splunk_client
        self.elastic = elastic_hunter
        self.hunts: dict[str, HuntDefinition] = {}
        self.results_dir = Path(results_dir)
        self.results_dir.mkdir(exist_ok=True)

    def register_hunt(self, hunt: HuntDefinition):
        """Registrar una query de hunting."""
        self.hunts[hunt.name] = hunt

    def run_hunt(
        self, name: str, hours_back: int = 24
    ) -> HuntResult:
        """Ejecutar una query de hunting individual."""
        hunt = self.hunts.get(name)
        if not hunt:
            return HuntResult(
                hunt_name=name,
                status=HuntStatus.FAILED,
                timestamp=datetime.utcnow().isoformat(),
                error=f"Hunt '{name}' not registered",
            )

        start = datetime.utcnow()

        try:
            if hunt.platform == "splunk" and self.splunk:
                results = self.splunk.search(
                    hunt.query,
                    earliest=f"-{hours_back}h",
                )
                hit_count = len(results)
            elif hunt.platform == "elastic" and self.elastic:
                resp = self.elastic.search(
                    index=hunt.index,
                    query=hunt.query,
                    hours_back=hours_back,
                )
                results = [
                    h["_source"]
                    for h in resp["hits"]["hits"]
                ]
                hit_count = resp["hits"]["total"]["value"]
            else:
                raise ValueError(
                    f"Platform {hunt.platform} not configured"
                )

            duration = (
                datetime.utcnow() - start
            ).total_seconds()

            return HuntResult(
                hunt_name=name,
                status=HuntStatus.COMPLETED,
                timestamp=start.isoformat(),
                hit_count=hit_count,
                results=results[:100],
                duration_seconds=duration,
            )

        except Exception as e:
            return HuntResult(
                hunt_name=name,
                status=HuntStatus.FAILED,
                timestamp=start.isoformat(),
                error=str(e),
            )

    def run_all(self, hours_back: int = 24) -> list[HuntResult]:
        """Ejecutar todas las queries registradas."""
        all_results = []
        for name in self.hunts:
            print(f"[*] Running hunt: {name}")
            result = self.run_hunt(name, hours_back)
            all_results.append(result)

            status = "OK" if result.hit_count == 0 else (
                f"ALERT: {result.hit_count} hits"
            )
            print(
                f"    [{result.status.value}] {status} "
                f"({result.duration_seconds:.1f}s)"
            )

        # Guardar resumen
        summary_file = (
            self.results_dir
            / f"campaign_{datetime.utcnow():%Y%m%d_%H%M%S}.json"
        )
        summary = [
            {
                "name": r.hunt_name,
                "status": r.status.value,
                "hits": r.hit_count,
                "duration": r.duration_seconds,
            }
            for r in all_results
        ]
        with open(summary_file, "w") as f:
            json.dump(summary, f, indent=2)

        return all_results

Registrar hunts desde YAML

Para que el equipo de threat hunting pueda añadir queries sin modificar código Python:

# hunts/powershell_encoded.yaml
name: suspicious_powershell_encoded
description: "PowerShell con comandos codificados en Base64"
mitre_technique: T1059.001
platform: splunk
severity: high
tags:
  - execution
  - powershell
  - living-off-the-land
query: >
  search index=windows sourcetype=WinEventLog:Security
  EventCode=4688
  New_Process_Name="*powershell.exe"
  (CommandLine="*-enc*" OR CommandLine="*-EncodedCommand*")
  | stats count by Computer, Account_Name, CommandLine
  | sort -count
false_positive_notes: >
  Algunos productos legitimos usan PowerShell encoded
  (SCCM, Intune). Verificar el contenido decodificado.
response_action: >
  Decodificar el comando Base64, verificar si el script
  es conocido. Si es desconocido, aislar el endpoint.

Recursos

• Splunk REST API Reference: documentación oficial de la API REST de Splunk, incluyendo search jobs, alertas y configuración. Disponible en docs.splunk.com.
• Elasticsearch Python Client (elasticsearch-py): cliente oficial de Python para Elasticsearch con soporte para búsquedas, agregaciones y gestión de índices. PyPI: elasticsearch.
• Splunk SDK for Python: SDK oficial que simplifica la interacción con la API de Splunk, incluyendo manejo automático de search jobs y autenticación. PyPI: splunk-sdk.
• Sigma Rules (SigmaHQ): repositorio de reglas de detección genéricas que pueden convertirse a SPL (Splunk) o Elasticsearch Query DSL con herramientas como sigma-cli.
• MITRE ATT&CK Navigator: herramienta para mapear cobertura de detección contra la matriz ATT&CK. Útil para identificar gaps en las queries de hunting.
• Elastic Security Detection Rules: repositorio oficial de Elastic con reglas de detección en formato TOML, convertibles a queries de Elasticsearch. GitHub: elastic/detection-rules.
• Hunting ELK (HELK): plataforma open source de hunting basada en ELK con notebooks Jupyter integrados. Referencia para arquitectura de hunting automatizado.