#!/usr/bin/env python3 """ daemon.py — Autonomous Wazuh-LLM Incident Response Daemon Runs in a continuous loop, watching /var/ossec/logs/alerts/alerts.json for new Wazuh alerts (rule level 7-12). Also detects SSH brute-force patterns: 5+ failed SSH logins from the same source IP within a 5-minute window. For every triggered alert the daemon automatically: 1. Classify — Ollama wazuh-llama -> incident_type, severity, IOCs 2. Show cmds — deterministic table -> which Cisco show commands to run 3. GET state — RESTCONF GET calls -> current device state (JSON) 4. Fix LLM — domain LoRA adapter -> CLI fix commands 5. Apply fix — RESTCONF PATCH/PUT -> push changes to device 6. Log result — managed_incidents.jsonl + managed_incidents.log Usage: python daemon.py python daemon.py --alerts /var/ossec/logs/alerts/alerts.json python daemon.py --poll 3 --ssh-threshold 5 --ssh-window 300 --min-level 7 python daemon.py --dry-run # classify + plan, but don't PATCH devices """ import argparse import base64 import gc import json import logging import os import re import signal import sys import time from collections import defaultdict from datetime import datetime, timezone, timedelta from pathlib import Path import requests import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # Heavy ML imports — loaded lazily when first alert fires _torch = None _PeftModel = None _AutoModelForCausalLM = None _AutoTokenizer = None def _load_ml(): global _torch, _PeftModel, _AutoModelForCausalLM, _AutoTokenizer if _torch is not None: return import torch from peft import PeftModel from transformers import AutoModelForCausalLM, AutoTokenizer _torch = torch _PeftModel = PeftModel _AutoModelForCausalLM = AutoModelForCausalLM _AutoTokenizer = AutoTokenizer # ───────────────────────────────────────────────────────────────────────────── # PATHS & CONSTANTS # ───────────────────────────────────────────────────────────────────────────── WORK_DIR = Path(__file__).parent # project-root/backend/ PROJECT_ROOT = WORK_DIR.parent # project-root/ BASE_MODEL = str(PROJECT_ROOT / "models" / "base" / "Hermes-3-Llama-3.1-8B") DEFAULT_ALERTS_FILE = Path("/var/ossec/logs/alerts/alerts.json") SHOW_OUTPUTS_DIR = PROJECT_ROOT / "show_outputs" MANAGED_LOG_JSONL = PROJECT_ROOT / "managed_incidents.jsonl" MANAGED_LOG_READABLE = PROJECT_ROOT / "managed_incidents.log" OLLAMA_URL = "http://localhost:11434/api/generate" WAZUH_MODEL = "wazuh-llama" ROUTER_USER = "admin" ROUTER_PASS = "cisco123!" RESTCONF_PORT = 443 RESTCONF_GET_TIMEOUT = 15 # seconds per GET call RESTCONF_PATCH_TIMEOUT = 20 # seconds per PATCH call # ───────────────────────────────────────────────────────────────────────────── # LOGGING # ───────────────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler(str(PROJECT_ROOT / "daemon.log"), encoding="utf-8"), ], ) log = logging.getLogger(__name__) # ───────────────────────────────────────────────────────────────────────────── # DEVICE MAPS (same as pipeline.py) # ───────────────────────────────────────────────────────────────────────────── SOURCE_IP_DEVICE_MAP: dict[str, str] = { "1.1.1.1": "R1", "2.2.2.2": "R2", "3.3.3.3": "R3", "4.4.4.4": "R4", "5.5.5.5": "R5", "10.10.10.10": "R1", "10.10.10.1": "cisco-router-02", "192.168.100.2": "SW1", "192.168.100.3": "SW2", "192.168.100.21": "ACCESS-SW1", "192.168.100.22": "ACCESS-SW2", "192.168.40.20": "dhcp-test-client", "192.168.40.21": "dhcp-test-client", "192.168.40.22": "dhcp-test-client", "198.51.100.23": "external-ssh-scanner", "203.0.113.45": "external-port-scan", "185.199.108.12": "malicious-server", "192.0.2.111": "rdp-bruteforce-source", } DEVICE_MGMT_IP: dict[str, str] = { "R1": "10.10.10.10", "R2": "2.2.2.2", "R3": "3.3.3.3", "R4": "4.4.4.4", "SW1": "192.168.100.2", "SW2": "192.168.100.3", "ACCESS-SW1": "192.168.100.21", "ACCESS-SW2": "192.168.100.22", "cisco-router-02": "10.10.10.1", } UNMANAGED = { "unknown", "external-ssh-scanner", "external-port-scan", "malicious-server", "rdp-bruteforce-source", "dhcp-test-client", } _I = PROJECT_ROOT / "models" / "incidents" LORA_PATHS: dict[tuple, Path] = { ("bgp", 1): _I / "bgp" / "lora_llm_bgp1", ("bgp", 2): _I / "bgp" / "lora_llm_bgp2", ("bgp", 3): _I / "bgp" / "lora_llm_bgp3", ("ospf", 1): _I / "ospf" / "ospf1", ("ospf", 2): _I / "ospf" / "ospf2", ("ospf", 3): _I / "ospf" / "ospf3", ("ospf", 4): _I / "ospf" / "ospf4", ("sec", 1): _I / "security" / "lora_llm_sec1", ("sec", 2): _I / "security" / "lora_llm_sec2", ("sec", 3): _I / "security" / "lora_llm_sec3", ("switch", 1): _I / "switch" / "lora_llm_switch1", ("switch", 2): _I / "switch" / "lora_llm_switch2", ("switch", 3): _I / "switch" / "lora_llm_switch3", ("service", 1): _I / "service" / "lora_llm_service1", ("service", 2): _I / "service" / "lora_llm_service2", ("sys", 1): _I / "sys" / "lora_llm_sys1", ("sys", 2): _I / "sys" / "lora_llm_sys2", ("sys", 3): _I / "sys" / "lora_llm_sys3", } DOMAIN_INSTRUCTIONS: dict[str, str] = { "bgp": ( "Analyze multi-device BGP incidents and output ONLY CLI FIX COMMANDS.\n" "Do not provide explanation. Always modify the device in the Wazuh alert.\n" "Rules:\n" "- Keepalive MUST be >= 30 seconds.\n" "- NEVER use keepalive value 10.\n" "- Keepalive MUST be <= hold/3.\n" "- For hold timer expiration, timers MUST exactly match peer timers." ), "ospf": ( "Analyze a multi-device OSPF adjacency issue and generate ONLY CLI FIX COMMANDS.\n" "Do not provide explanation. Do not add comments. Only output the commands." ), "sec": ( "Analyze security policy incidents and output ONLY CLI FIX COMMANDS.\n" "Do not provide explanation.\n" "INCIDENT-SPECIFIC FIX RULES (MANDATORY):\n" "1. acl_blocking_legitimate_traffic: MUST add a PERMIT; MUST NOT keep DENY for the flow.\n" "2. acl_misconfiguration: Correct protocol, wildcard, or direction.\n" "3. excessive_deny_entries: Refine/replace broad DENYs. NEVER 'permit ip any any'.\n" "4. DNS = UDP/53 (TCP/53 only for zone transfers).\n" "5. NEVER invert policy logic." ), "switch": ( "Analyze switching/L2 incidents and output ONLY CLI FIX COMMANDS.\n" "Do not provide explanation. Output only valid Cisco IOS/IOS-XE commands." ), "service": ( "Analyze network service incidents (DHCP, DNS, NTP) and output ONLY CLI FIX COMMANDS.\n" "Do not provide explanation. Output only the commands." ), "sys": ( "Given a system-health incident and device outputs, generate ONLY VALID " "Cisco IOS/IOS-XE CONFIGURATION commands that immediately mitigate the incident.\n" "GLOBAL RULES:\n" "1. Output ONLY complete, syntactically correct Cisco IOS/IOS-XE config commands.\n" "2. Do NOT output show, debug, clear, reload, write memory, or comments.\n" "3. HIGH CPU: disable console logging, reduce buffered logging, " "disable unused HTTP/HTTPS services.\n" "4. HIGH MEMORY: REQUIRED - no ip http server / no ip http secure-server.\n" "5. PROCESS CRASH: FIRST command MUST be: " "exception crashinfo file bootflash:crashinfo" ), } # ───────────────────────────────────────────────────────────────────────────── # SHOW COMMAND RULES (same as pipeline.py) # ───────────────────────────────────────────────────────────────────────────── _SHOW_RULES: list[tuple[tuple[str, ...], list[str]]] = [ # OSPF (("ospf_neighbor_down", "ospf_neighbour_down", "neighbor_down", "neighbour_down", "adj_down", "adjacency_down", "full_to_down", "full to down"), ["show ip ospf neighbor", "show ip ospf interface", "show ip ospf database", "show ip route"]), (("ospf_stuck_init", "stuck_init", "ospf_init", "ospf_2way", "ospf_2-way"), ["show ip ospf neighbor", "show ip ospf interface", "show ip ospf database"]), (("ospf_exstart", "exstart_exchange_stuck", "exstart", "exchange_stuck", "ospf_exchange"), ["show ip ospf neighbor", "show ip ospf interface", "show ip ospf database"]), (("ospf_mtu_mismatch", "mtu_mismatch", "mtu mismatch"), ["show ip ospf neighbor", "show ip ospf interface", "show interfaces"]), (("ospf_area_mismatch", "area_mismatch", "area mismatch", "wrong_area", "ospf_area"), ["show ip ospf neighbor", "show ip ospf interface", "show ip ospf database"]), (("ospf_auth_mismatch", "auth_mismatch", "authentication_mismatch", "ospf_authentication"), ["show ip ospf neighbor", "show ip ospf interface", "show running-config"]), (("ospf_hello_dead_mismatch", "hello_dead_mismatch", "hello_mismatch", "dead_mismatch", "ospf_hello", "ospf_dead"), ["show ip ospf neighbor", "show ip ospf interface"]), (("ospf_network_type_mismatch", "network_type_mismatch", "ospf_network_type"), ["show ip ospf neighbor", "show ip ospf interface"]), (("ospf_lsa_flood", "lsa_flood", "lsa flood", "ospf_flood"), ["show ip ospf", "show ip ospf neighbor", "show interfaces"]), (("ospf_lsdb_inconsistency", "lsdb_inconsistency", "lsdb inconsistency"), ["show ip ospf database", "show ip ospf neighbor"]), (("ospf_redistribution_issue", "redistribution_issue", "ospf_redistrib", "ospf redistrib", "ospf_route_redistribution"), ["show ip route", "show running-config", "show ip ospf"]), (("ospf",), ["show ip ospf neighbor", "show ip ospf interface", "show ip ospf database", "show ip route"]), # BGP (("bgp_session_flap", "session_flap", "bgp_flap", "bgp flap"), ["show ip bgp summary", "show ip bgp neighbors", "show logging", "show interfaces"]), (("bgp_hold_timer_expiration", "hold_timer_expiration", "hold_timer_expired", "hold timer expir", "hold_timer", "hold timer"), ["show ip bgp summary", "show ip bgp neighbors", "show logging"]), (("bgp_neighborship_reset", "neighborship_reset", "bgp_reset", "bgp reset", "bgp_neighbor_reset"), ["show ip bgp summary", "show ip bgp neighbors", "show logging"]), (("bgp_prefix_limit_exceeded", "prefix_limit_exceeded", "prefix_limit", "prefix limit"), ["show ip bgp summary", "show ip bgp neighbors", "show logging"]), (("bgp_route_leak_suspected", "route_leak_suspected", "route_leak", "route leak", "bgp_community"), ["show ip bgp summary", "show ip bgp neighbors", "show running-config", "show logging"]), (("bgp_path_selection_incorrect", "path_selection_incorrect", "path_selection", "path selection", "bgp_bestpath", "bgp bestpath"), ["show ip bgp summary", "show ip bgp neighbors", "show ip bgp", "show logging"]), (("bgp_missing_routes_in_rib", "missing_routes_in_rib", "missing_routes", "missing routes", "bgp_rib", "routes_missing"), ["show ip bgp", "show ip route", "show ip bgp summary", "show logging"]), (("bgp_next_hop_self_issue", "next_hop_self_issue", "next_hop_self", "next-hop-self", "bgp_next_hop"), ["show ip bgp neighbors", "show ip bgp", "show ip route", "show running-config"]), (("bgp_afi_safi_mismatch", "afi_safi_mismatch", "afi_safi", "afi-safi", "bgp_ipv6", "bgp_address_family"), ["show ip bgp summary", "show ip bgp neighbors", "show running-config"]), (("bgp",), ["show ip bgp summary", "show ip bgp neighbors", "show logging"]), # Security (("acl_blocking_legitimate_traffic", "acl_blocking_legit", "acl_blocking", "blocking_legitimate", "block_legit"), ["show access-lists", "show ip access-lists", "show interfaces", "show ip interface brief"]), (("acl_misconfiguration", "acl_misconfig", "acl_wrong", "access_list_misconfiguration"), ["show access-lists", "show ip access-lists", "show running-config"]), (("excessive_deny_entries", "excessive_deny", "excessive_denies", "too_many_denies", "deny_entries"), ["show access-lists", "show ip access-lists"]), (("acl_block", "acl_drop", "access_list", "access-list", "access_control_list"), ["show access-lists", "show ip access-lists", "show interfaces"]), (("firewall_conn_exhaust", "conn_exhaust", "connection_exhaustion", "connection_table_full", "conn_table"), ["show interfaces", "show ip nat statistics", "show ip nat translations"]), (("nat_translation_failure", "nat_failure", "nat_translation", "nat_issue", "nat_error"), ["show ip nat statistics", "show ip nat translations"]), (("ip_spoofing", "spoofing", "ip_spoof"), ["show interfaces", "show ip arp", "show logging"]), (("ssh_brute_force", "ssh_bruteforce", "brute_force", "bruteforce", "brute force"), ["show logging", "show ip access-lists"]), (("failed_login", "login_failure", "login_failures", "authentication_failure"), ["show logging"]), (("snmp_bruteforce", "snmp_brute_force", "snmp_attack", "snmp brute"), ["show logging", "show running-config"]), (("port_scan", "port scan", "port_scanning", "network_scan"), ["show logging", "show ip access-lists", "show interfaces"]), (("security", "firewall", "intrusion", "malware", "attack"), ["show access-lists", "show logging", "show interfaces"]), # Switching / L2 (("mac_flapping", "mac_flap", "mac flapping", "mac flap"), ["show interfaces status", "show mac address-table", "show spanning-tree", "show logging"]), (("stp_topology_change", "stp_change", "spanning_tree_topology", "topology_change", "spanning tree topology"), ["show spanning-tree", "show interfaces status", "show logging"]), (("port_errdisable", "errdisable", "err_disable", "err-disable", "port err-disable"), ["show interfaces status", "show logging"]), (("native_vlan_mismatch", "native_vlan", "native vlan mismatch"), ["show interfaces", "show vlan"]), (("trunk_negotiation_failure", "trunk_negotiation", "trunk_mismatch"), ["show interfaces", "show vlan"]), (("vlan_mismatch",), ["show interfaces status", "show vlan", "show interfaces"]), (("lldp_cdp_inconsistency", "lldp_inconsistency", "cdp_inconsistency", "lldp cdp"), ["show cdp neighbors", "show lldp neighbors"]), (("storm_control",), ["show interfaces", "show logging"]), (("broadcast_storm", "broadcast storm", "multicast_storm"), ["show interfaces", "show logging"]), (("switch", "vlan", "trunk", "spanning", "layer2", "layer 2", "l2", "mac_address", "mac address"), ["show interfaces status", "show spanning-tree", "show vlan"]), # Services (("dhcp_conflict",), ["show ip dhcp conflict", "show ip dhcp binding", "show ip arp"]), (("dhcp_starvation", "dhcp starvation", "dhcp_pool_exhausted"), ["show ip dhcp pool", "show ip dhcp binding", "show ip dhcp server statistics"]), (("ip_conflict", "ip conflict", "duplicate_ip", "duplicate ip"), ["show ip arp", "show ip dhcp conflict"]), (("dhcp",), ["show ip dhcp pool", "show ip dhcp conflict", "show ip dhcp binding"]), (("dns_issue", "dns_issues", "dns_failure", "dns failure", "dns_resolution"), ["show ip route", "show running-config"]), (("ntp_unsync", "ntp_sync_issue", "ntp_issue", "ntp issue", "ntp_drift", "clock_skew"), ["show ntp status", "show ntp associations"]), (("arp_spoofing", "arp_spoof", "arp spoof", "gratuitous_arp"), ["show ip arp", "show mac address-table", "show logging"]), (("dns", "domain_name", "name_server", "resolver"), ["show ip route", "show running-config"]), (("ntp", "time_sync", "clock"), ["show ntp status", "show ntp associations"]), (("arp",), ["show ip arp", "show mac address-table"]), # System health (("high_cpu", "cpu_high", "cpu_utilization", "cpu spike", "high cpu"), ["show processes cpu", "show processes cpu history", "show logging", "show interfaces"]), (("high_memory", "memory_high", "memory_exhaustion", "memory_leak", "out_of_memory", "oom", "high memory"), ["show processes memory", "show version", "show logging"]), (("process_crash", "process crash", "crashinfo", "crash"), ["show version", "show logging", "show processes memory"]), (("traceback",), ["show logging", "show version"]), (("interface_flap", "interface flap", "link_flap", "link flap", "int_flap"), ["show interfaces", "show logging", "show ip route"]), (("duplex_mismatch", "duplex mismatch", "speed_mismatch", "speed mismatch"), ["show interfaces", "show running-config"]), (("reload", "reboot", "scheduled_reload"), ["show version", "show logging"]), (("environment", "temperature", "fan_fail", "power_supply"), ["show environment", "show version"]), (("cpu", "memory", "system", "hardware", "health"), ["show processes cpu", "show processes memory", "show logging", "show version"]), ] _DEFAULT_SHOW_CMDS = ["show interfaces", "show ip route", "show logging", "show version"] def _get_show_commands(incident_type: str) -> list[str]: key = incident_type.strip().lower() for keywords, cmds in _SHOW_RULES: if any(kw in key for kw in keywords): return list(cmds) return list(_DEFAULT_SHOW_CMDS) # ───────────────────────────────────────────────────────────────────────────── # RESTCONF GET RULES (same as pipeline.py) # ───────────────────────────────────────────────────────────────────────────── _RESTCONF_RULES: list[tuple[str, str]] = [ ("show ip ospf neighbor detail", "/restconf/data/Cisco-IOS-XE-ospf-oper:ospf-oper-data/ospf-state"), ("show ip ospf neighbor", "/restconf/data/Cisco-IOS-XE-ospf-oper:ospf-oper-data/ospf-state"), ("show ip ospf interface", "/restconf/data/Cisco-IOS-XE-ospf-oper:ospf-oper-data/ospf-state"), ("show ip ospf database", "/restconf/data/Cisco-IOS-XE-ospf-oper:ospf-oper-data/ospf-state"), ("show ip ospf", "/restconf/data/Cisco-IOS-XE-ospf-oper:ospf-oper-data/ospf-state"), ("show ip bgp summary", "/restconf/data/Cisco-IOS-XE-bgp-oper:bgp-state-data/bgp-route-vrfs"), ("show ip bgp neighbors", "/restconf/data/Cisco-IOS-XE-bgp-oper:bgp-state-data/neighbors"), ("show ip bgp", "/restconf/data/Cisco-IOS-XE-bgp-oper:bgp-state-data"), ("show interfaces status", "/restconf/data/Cisco-IOS-XE-interfaces-oper:interfaces"), ("show interfaces", "/restconf/data/Cisco-IOS-XE-interfaces-oper:interfaces"), ("show ip interface brief", "/restconf/data/Cisco-IOS-XE-interfaces-oper:interfaces"), ("show ip route", "/restconf/data/Cisco-IOS-XE-ip-route-oper:ip-route-data"), ("show ip access-lists", "/restconf/data/Cisco-IOS-XE-acl-oper:access-lists"), ("show access-lists", "/restconf/data/Cisco-IOS-XE-acl-oper:access-lists"), ("show port-security", "/restconf/data/Cisco-IOS-XE-port-security-oper:port-security"), ("show vlan", "/restconf/data/Cisco-IOS-XE-vlan-oper:vlans"), ("show spanning-tree", "/restconf/data/Cisco-IOS-XE-spanning-tree-oper:stp-details"), ("show mac address-table", "/restconf/data/Cisco-IOS-XE-matm-oper:matm-oper-data"), ("show etherchannel", "/restconf/data/Cisco-IOS-XE-lacp-oper:lacp"), ("show ip dhcp binding", "/restconf/data/Cisco-IOS-XE-dhcp-oper:dhcp-oper-data/client-bindings"), ("show ip dhcp conflict", "/restconf/data/Cisco-IOS-XE-dhcp-oper:dhcp-oper-data"), ("show ip dhcp pool", "/restconf/data/Cisco-IOS-XE-dhcp-oper:dhcp-oper-data"), ("show ip dhcp server statistics", "/restconf/data/Cisco-IOS-XE-dhcp-oper:dhcp-oper-data"), ("show ip dhcp", "/restconf/data/Cisco-IOS-XE-dhcp-oper:dhcp-oper-data"), ("show ip arp", "/restconf/data/Cisco-IOS-XE-arp-oper:arp-data"), ("show cdp neighbors", "/restconf/data/Cisco-IOS-XE-cdp-oper:cdp-neighbor-details"), ("show lldp neighbors", "/restconf/data/Cisco-IOS-XE-lldp-oper:lldp-entries"), ("show ntp associations", "/restconf/data/Cisco-IOS-XE-ntp-oper:ntp-oper-data"), ("show ntp status", "/restconf/data/Cisco-IOS-XE-ntp-oper:ntp-oper-data"), ("show processes cpu history", "/restconf/data/Cisco-IOS-XE-process-cpu-oper:cpu-usage"), ("show processes cpu", "/restconf/data/Cisco-IOS-XE-process-cpu-oper:cpu-usage"), ("show processes memory", "/restconf/data/Cisco-IOS-XE-process-memory-oper:memory-usage-processes"), ("show logging", "/restconf/data/Cisco-IOS-XE-native:native/logging"), ("show ip nat translations", "/restconf/data/Cisco-IOS-XE-nat-oper:nat-data"), ("show ip nat statistics", "/restconf/data/Cisco-IOS-XE-nat-oper:nat-data"), ("show running-config", "/restconf/data/Cisco-IOS-XE-native:native"), ("show version", "/restconf/data/Cisco-IOS-XE-native:native/version"), ("show environment", "/restconf/data/Cisco-IOS-XE-environment-oper:environment-sensors"), ] def _resolve_restconf_path(cmd: str) -> str | None: c = cmd.strip().lower() for prefix, path in _RESTCONF_RULES: if c == prefix or c.startswith(prefix + " ") or c.startswith(prefix): return path return None # ───────────────────────────────────────────────────────────────────────────── # RESTCONF AUTH # ───────────────────────────────────────────────────────────────────────────── def _auth_header() -> str: return "Basic " + base64.b64encode( f"{ROUTER_USER}:{ROUTER_PASS}".encode() ).decode() def _restconf_headers() -> dict: return { "Accept": "application/yang-data+json", "Content-Type": "application/yang-data+json", "Authorization": _auth_header(), } # ───────────────────────────────────────────────────────────────────────────── # RESTCONF EXECUTOR # ───────────────────────────────────────────────────────────────────────────── def restconf_get(mgmt_ip: str, yang_path: str) -> dict: """ Execute a RESTCONF GET. Returns {"ok": bool, "status": int|None, "data": dict|None, "error": str|None} """ url = f"https://{mgmt_ip}:{RESTCONF_PORT}{yang_path}" try: resp = requests.get( url, headers=_restconf_headers(), verify=False, timeout=RESTCONF_GET_TIMEOUT, ) try: data = resp.json() except Exception: data = {"raw": resp.text[:2000]} return {"ok": resp.ok, "status": resp.status_code, "url": url, "data": data} except requests.RequestException as exc: return {"ok": False, "status": None, "url": url, "error": str(exc)} def restconf_patch(url: str, body: dict) -> dict: """ Execute a RESTCONF PATCH. Returns {"ok": bool, "status": int|None, "error": str|None} """ if body is None: return {"ok": False, "status": None, "error": "No YANG body — apply manually via CLI"} try: resp = requests.patch( url, headers=_restconf_headers(), json=body, verify=False, timeout=RESTCONF_PATCH_TIMEOUT, ) return {"ok": resp.ok, "status": resp.status_code, "response": resp.text[:500] if not resp.ok else "OK"} except requests.RequestException as exc: return {"ok": False, "status": None, "error": str(exc)} def collect_device_state(device: str, show_cmds: list[str]) -> dict[str, dict]: """ Run RESTCONF GETs for all show commands. Returns {show_command: response_dict} """ mgmt_ip = DEVICE_MGMT_IP.get(device, device) results: dict[str, dict] = {} seen_paths: set[str] = set() for cmd in show_cmds: path = _resolve_restconf_path(cmd) if path is None: log.warning(" No RESTCONF path for: %s", cmd) results[cmd] = {"ok": False, "error": "No RESTCONF path found"} continue if path in seen_paths: log.debug(" Skipping duplicate path: %s", path) continue seen_paths.add(path) log.info(" GET %s -> %s", cmd, path) result = restconf_get(mgmt_ip, path) results[cmd] = result log.info(" -> HTTP %s ok=%s", result.get("status"), result.get("ok")) return results # ───────────────────────────────────────────────────────────────────────────── # WAZUH CLASSIFIER (same as pipeline.py) # ───────────────────────────────────────────────────────────────────────────── _WAZUH_SYSTEM = ( "You are a cybersecurity analyst specializing in Wazuh and Cisco network alerts. " "Analyze the log and return ONLY valid JSON with these fields: " "incident_type (e.g. ospf_neighbor_down, bgp_session_flap, acl_block, " "dhcp_starvation, high_cpu, mac_flapping, brute_force, port_scan, etc.), " "classification (benign/suspicious/malicious), " "severity (low/medium/high/critical), " "source_ip (string or null), destination_ip (string or null), " "iocs (list of strings), " "recommended_actions (list of strings), " "explanation (1-2 sentences). " "If data is missing infer from context or set null." ) _IP_RE = re.compile(r"([0-9]+(?:\.[0-9]+){3})") _CISCO_SRC = re.compile(r"\[Source:\s*([0-9]+(?:\.[0-9]+){3})\]") def _parse_json_safe(text: str) -> dict | None: text = text.strip() try: return json.loads(text) except Exception: pass s, e = text.find("{"), text.rfind("}") if s != -1 and e != -1: try: return json.loads(text[s:e + 1]) except Exception: pass return None def call_wazuh_llm(log_text: str) -> dict | None: payload = { "model": WAZUH_MODEL, "prompt": f"<|system|>\n{_WAZUH_SYSTEM}\n\n<|user|>\nLog: {log_text}\n<|assistant|>\n", "options": {"num_predict": 400, "stop": ["<|user|>", "<|system|>"]}, } try: with requests.post(OLLAMA_URL, json=payload, stream=True, timeout=60) as resp: if resp.status_code != 200: return None output = "" for raw in resp.iter_lines(): if not raw: continue try: chunk = json.loads(raw.decode("utf-8")) except Exception: output += raw.decode("utf-8", errors="ignore") continue output += chunk.get("response", "") if chunk.get("done"): break return _parse_json_safe(output) except requests.RequestException: return None def _fallback_classify(alert_raw: dict) -> dict: desc = alert_raw.get("rule", {}).get("description", "unknown") return { "incident_type": desc.lower().replace(" ", "_"), "classification": "suspicious", "severity": "high", "explanation": desc, "source_ip": None, "destination_ip": None, "iocs": [], "recommended_actions": [], } def _extract_device(alert_raw: dict) -> str: """Extract device name from a raw Wazuh alert dict.""" full_log = alert_raw.get("full_log") or alert_raw.get("decoded") or "" m = _CISCO_SRC.search(full_log) if m: return SOURCE_IP_DEVICE_MAP.get(m.group(1), "unknown") m2 = _IP_RE.search(full_log) if m2: return SOURCE_IP_DEVICE_MAP.get(m2.group(1), "unknown") agent_ip = alert_raw.get("agent", {}).get("ip", "") return SOURCE_IP_DEVICE_MAP.get(agent_ip, "unknown") # ───────────────────────────────────────────────────────────────────────────── # DOMAIN ROUTING (same as pipeline.py) # ───────────────────────────────────────────────────────────────────────────── def classify_domain(incident_type: str, description: str = "") -> tuple[str, int]: combined = ((incident_type or "") + " " + (description or "")).lower() if "bgp" in combined: if any(k in combined for k in ("session_flap", "session flap", "hold_timer", "hold timer", "neighborship", "keepalive")): return "bgp", 1 if any(k in combined for k in ("routing_policy", "prefix", "community")): return "bgp", 2 return "bgp", 3 if "ospf" in combined: if any(k in combined for k in ("neighbor_down", "neighbor down", "adjacency", "adj_down", "full to down")): return "ospf", 1 if any(k in combined for k in ("stuck", "exstart", "init", "exchange", "2way")): return "ospf", 2 if any(k in combined for k in ("area", "auth", "mismatch", "hello", "dead")): return "ospf", 3 if any(k in combined for k in ("lsa", "lsdb", "flood", "redistrib")): return "ospf", 4 return "ospf", 1 if any(k in combined for k in ("acl", "access_list", "firewall", "security", "brute_force", "brute force", "port_scan", "malware", "intrusion", "login_failure", "deny")): if any(k in combined for k in ("blocking", "legitimate", "legit")): return "sec", 1 if any(k in combined for k in ("misconfiguration", "misconfig", "wrong")): return "sec", 2 if any(k in combined for k in ("excessive", "too_many", "deny_entries")): return "sec", 3 return "sec", 1 if any(k in combined for k in ("mac_flap", "stp", "spanning", "vlan", "trunk", "err_disable", "switch", "l2", "layer2")): if any(k in combined for k in ("mac_flap", "stp", "spanning", "err_disable")): return "switch", 1 if "port" in combined or "duplex" in combined: return "switch", 2 if "vlan" in combined or "trunk" in combined: return "switch", 3 return "switch", 1 if any(k in combined for k in ("dhcp", "dns", "ntp", "service", "ip_conflict")): if any(k in combined for k in ("dhcp", "ip_conflict", "starvation")): return "service", 1 return "service", 2 if any(k in combined for k in ("cpu", "memory", "crash", "process", "interface_flap", "duplex", "mismatch", "system", "reload", "reboot", "traceback")): if any(k in combined for k in ("cpu", "memory", "crash", "process", "traceback")): return "sys", 1 if any(k in combined for k in ("interface_flap", "duplex", "mismatch", "flap")): return "sys", 2 return "sys", 3 return "sys", 1 # ───────────────────────────────────────────────────────────────────────────── # CLI FIX -> RESTCONF PATCH (same as run_fix.py) # ───────────────────────────────────────────────────────────────────────────── _CTX_PAT = [ re.compile(r"^interface\s+(\S+)", re.I), re.compile(r"^router\s+(ospf|bgp|eigrp|isis|rip)\s+(\S+)", re.I), re.compile(r"^ip\s+access-list\s+(extended|standard)\s+(\S+)", re.I), re.compile(r"^ip\s+dhcp\s+pool\s+(\S+)", re.I), ] def _parse_cli_blocks(fix_text: str) -> list[dict]: blocks: list[dict] = [] ctx = "global" lines: list[str] = [] for raw in fix_text.splitlines(): s = raw.strip() if not s: continue if s == "!": if lines: blocks.append({"context": ctx, "lines": list(lines)}) lines = [] ctx = "global" continue if not raw.startswith((" ", "\t")) and any(p.match(s) for p in _CTX_PAT): if lines: blocks.append({"context": ctx, "lines": list(lines)}) lines = [] ctx = s continue lines.append(s) if lines: blocks.append({"context": ctx, "lines": list(lines)}) return blocks def _url_enc(slot: str) -> str: return slot.replace("/", "%2F").replace(".", "%2E") def _iface_parts(iface: str) -> tuple[str, str]: m = re.match(r"([A-Za-z]+)([\d/\.]+)", iface.strip()) return (m.group(1), m.group(2)) if m else (iface, "0") def _yang_interface(itype: str, islot: str, lines: list[str]) -> dict: inner: dict = {"name": islot} for line in lines: l = line.strip().lower() if l in ("no shutdown", "no shut"): inner["shutdown"] = False elif l in ("shutdown", "shut"): inner["shutdown"] = True m = re.match(r"(?:no\s+)?ip\s+ospf\s+hello-interval\s+(\d+)", l) if m: inner.setdefault("ip", {}).setdefault("Cisco-IOS-XE-ospf:ospf", {})["hello-interval"] = int(m.group(1)) m = re.match(r"(?:no\s+)?ip\s+ospf\s+dead-interval\s+(\d+)", l) if m: inner.setdefault("ip", {}).setdefault("Cisco-IOS-XE-ospf:ospf", {})["dead-interval"] = int(m.group(1)) m = re.match(r"ip\s+ospf\s+(\d+)\s+area\s+(\S+)", l) if m: ospf = inner.setdefault("ip", {}).setdefault("Cisco-IOS-XE-ospf:ospf", {}) ospf["process-id"] = int(m.group(1)); ospf["area"] = m.group(2) m = re.match(r"ip\s+ospf\s+network\s+(\S+)", l) if m: inner.setdefault("ip", {}).setdefault("Cisco-IOS-XE-ospf:ospf", {})["network"] = {"network-type": m.group(1)} if re.match(r"ip\s+ospf\s+mtu-ignore", l): inner.setdefault("ip", {}).setdefault("Cisco-IOS-XE-ospf:ospf", {})["mtu-ignore"] = True m = re.match(r"switchport\s+mode\s+(\S+)", l) if m: (inner.setdefault("Cisco-IOS-XE-switch:switchport-conf", {}) .setdefault("switchport", {}).setdefault("mode", {}))[m.group(1)] = {} m = re.match(r"switchport\s+access\s+vlan\s+(\d+)", l) if m: (inner.setdefault("Cisco-IOS-XE-switch:switchport-conf", {}) .setdefault("switchport", {}).setdefault("access", {}))["vlan"] = {"vlan": int(m.group(1))} m = re.match(r"switchport\s+trunk\s+native\s+vlan\s+(\d+)", l) if m: (inner.setdefault("Cisco-IOS-XE-switch:switchport-conf", {}) .setdefault("switchport", {}).setdefault("trunk", {}))["native"] = {"vlan": {"vlan-id": int(m.group(1))}} if re.match(r"spanning-tree\s+portfast", l): inner.setdefault("Cisco-IOS-XE-spanning-tree:spanning-tree", {})["portfast"] = {} m = re.match(r"duplex\s+(\S+)", l) if m: inner["duplex"] = {"Cisco-IOS-XE-ethernet:duplex-enum": m.group(1)} m = re.match(r"speed\s+(\d+)", l) if m: inner["speed"] = {"Cisco-IOS-XE-ethernet:value": int(m.group(1))} m = re.match(r"ip\s+address\s+(\S+)\s+(\S+)", l) if m: inner.setdefault("ip", {}).setdefault("address", {})["primary"] = { "address": m.group(1), "mask": m.group(2)} return {f"Cisco-IOS-XE-native:{itype}": [inner]} def _yang_ospf(pid: str, lines: list[str]) -> dict: inner: dict = {"id": int(pid) if pid.isdigit() else pid} for line in lines: l = line.strip().lower() m = re.match(r"area\s+(\S+)\s+authentication(\s+message-digest)?", l) if m: inner.setdefault("area", []).append({ "area-id": m.group(1), "authentication": {"message-digest": {}} if m.group(2) else {} }) m = re.match(r"network\s+(\S+)\s+(\S+)\s+area\s+(\S+)", l) if m: inner.setdefault("network", []).append( {"ip": m.group(1), "mask": m.group(2), "area": m.group(3)}) m = re.match(r"router-id\s+(\S+)", l) if m: inner["router-id"] = m.group(1) return {"Cisco-IOS-XE-native:ospf": [inner]} def _yang_bgp(asn: str, lines: list[str]) -> dict: inner: dict = {"id": int(asn) if asn.isdigit() else asn} neighbors: dict[str, dict] = {} for line in lines: l = line.strip().lower() m = re.match(r"neighbor\s+(\S+)\s+timers\s+(\d+)\s+(\d+)", l) if m: neighbors.setdefault(m.group(1), {"id": m.group(1)})["timers"] = { "keepalive": int(m.group(2)), "holdtime": int(m.group(3))} m = re.match(r"neighbor\s+(\S+)\s+maximum-prefix\s+(\d+)", l) if m: neighbors.setdefault(m.group(1), {"id": m.group(1)})["maximum-prefix"] = int(m.group(2)) m = re.match(r"neighbor\s+(\S+)\s+next-hop-self", l) if m: neighbors.setdefault(m.group(1), {"id": m.group(1)})["next-hop-self"] = {} m = re.match(r"neighbor\s+(\S+)\s+remote-as\s+(\d+)", l) if m: neighbors.setdefault(m.group(1), {"id": m.group(1)})["remote-as"] = int(m.group(2)) if neighbors: inner["neighbor"] = list(neighbors.values()) return {"Cisco-IOS-XE-native:bgp": [inner]} def _yang_acl(acl_type: str, acl_name: str, lines: list[str]) -> dict: entries = [] seq = 10 for line in lines: m = re.match(r"(permit|deny)\s+(.+)", line.strip(), re.I) if m: entries.append({"sequence": seq, "action": m.group(1).lower(), "rule": m.group(2)}) seq += 10 return {f"Cisco-IOS-XE-native:{acl_type}": [{"name": acl_name, "access-list-seq-rule": entries}]} def _yang_global(lines: list[str]) -> list[dict]: ops = [] for line in lines: l = line.strip().lower() m = re.match(r"ntp\s+server\s+(\S+)", l) if m: ops.append({"suffix": "/ntp", "body": { "Cisco-IOS-XE-native:ntp": {"server": {"server-list": [{"ip-address": m.group(1)}]}}}}) m = re.match(r"ip\s+route\s+(\S+)\s+(\S+)\s+(\S+)", l) if m: ops.append({"suffix": "/ip/route", "body": { "Cisco-IOS-XE-native:route": {"ip-route-interface-forwarding-list": [{ "prefix": m.group(1), "mask": m.group(2), "fwd-list": [{"fwd": m.group(3)}]}]}}}) if re.match(r"no\s+ip\s+http\s+secure-server", l): ops.append({"suffix": "/ip/http", "body": {"Cisco-IOS-XE-native:http": {"secure-server": False}}}) elif re.match(r"no\s+ip\s+http\s+server", l): ops.append({"suffix": "/ip/http", "body": {"Cisco-IOS-XE-native:http": {"server": False}}}) m = re.match(r"logging\s+buffered\s+(\d+)", l) if m: ops.append({"suffix": "/logging", "body": { "Cisco-IOS-XE-native:logging": {"buffered": {"size": int(m.group(1))}}}}) if re.match(r"no\s+logging\s+console", l): ops.append({"suffix": "/logging", "body": {"Cisco-IOS-XE-native:logging": {"console": False}}}) m = re.match(r"exception\s+crashinfo\s+(?:file\s+)?(.+)", l) if m: ops.append({"suffix": "/exception", "body": { "Cisco-IOS-XE-native:exception": {"crashinfo": {"filepath": m.group(1).strip()}}}}) return ops def build_restconf_patch_ops(device: str, fix_text: str) -> list[dict]: """Convert CLI fix commands to RESTCONF PATCH operations with YANG bodies.""" mgmt_ip = DEVICE_MGMT_IP.get(device, device) base_url = f"https://{mgmt_ip}:{RESTCONF_PORT}/restconf/data/Cisco-IOS-XE-native:native" ops = [] for block in _parse_cli_blocks(fix_text): ctx = block["context"] lines = block["lines"] if not lines: continue cl = ctx.strip().lower() m = re.match(r"interface\s+(\S+)", cl) if m: itype, islot = _iface_parts(m.group(1)) ops.append({"method": "PATCH", "url": f"{base_url}/interface/{itype}={_url_enc(islot)}", "body": _yang_interface(itype, islot, lines), "cli_context": ctx, "cli_commands": lines}) continue m = re.match(r"router\s+ospf\s+(\S+)", cl) if m: ops.append({"method": "PATCH", "url": f"{base_url}/router/ospf={m.group(1)}", "body": _yang_ospf(m.group(1), lines), "cli_context": ctx, "cli_commands": lines}) continue m = re.match(r"router\s+bgp\s+(\S+)", cl) if m: ops.append({"method": "PATCH", "url": f"{base_url}/router/bgp={m.group(1)}", "body": _yang_bgp(m.group(1), lines), "cli_context": ctx, "cli_commands": lines}) continue m = re.match(r"ip\s+access-list\s+(extended|standard)\s+(\S+)", cl) if m: ops.append({"method": "PATCH", "url": f"{base_url}/ip/access-list/{m.group(1)}={m.group(2)}", "body": _yang_acl(m.group(1), m.group(2), lines), "cli_context": ctx, "cli_commands": lines}) continue if cl == "global": for gop in _yang_global(lines): ops.append({"method": "PATCH", "url": base_url + gop["suffix"], "body": gop["body"], "cli_context": "global", "cli_commands": lines}) continue ops.append({"method": "PATCH", "url": base_url, "body": None, "cli_context": ctx, "cli_commands": lines, "note": "Context not mapped - apply manually"}) return ops # ───────────────────────────────────────────────────────────────────────────── # LoRA MODEL RUNNER (same pattern as run_fix.py FixRunner) # ───────────────────────────────────────────────────────────────────────────── class FixRunner: def __init__(self, base_path: str): self._base_path = base_path self._tokenizer = None self._base_model = None self._active = None self._model = None def _ensure_base(self): _load_ml() if self._tokenizer is None: log.info("Loading base tokenizer...") self._tokenizer = _AutoTokenizer.from_pretrained(self._base_path) if self._base_model is None: log.info("Loading base model (fp16)...") self._base_model = _AutoModelForCausalLM.from_pretrained( self._base_path, torch_dtype=_torch.float16, device_map="auto" ) def _swap(self, lora_path: Path): key = str(lora_path) if self._active == key and self._model is not None: return if self._model is not None: del self._model; self._model = None if self._base_model is not None: del self._base_model; self._base_model = None gc.collect() _torch.cuda.empty_cache() self._ensure_base() log.info("Loading LoRA adapter: %s", lora_path.name) self._model = _PeftModel.from_pretrained(self._base_model, key) self._model.eval() self._active = key def run(self, lora_path: Path, prompt: str, max_new_tokens: int = 300) -> str: self._swap(lora_path) inputs = self._tokenizer(prompt, return_tensors="pt").to("cuda") with _torch.no_grad(): out = self._model.generate( **inputs, max_new_tokens=max_new_tokens, temperature=0.2, do_sample=False, repetition_penalty=1.05, ) return self._tokenizer.decode(out[0], skip_special_tokens=True) def build_fix_prompt(domain: str, alert_info: dict, device_state: dict[str, dict]) -> str: instruction = DOMAIN_INSTRUCTIONS.get(domain, DOMAIN_INSTRUCTIONS["sys"]) device = alert_info.get("device", "DEVICE") lines = [ "### Instruction:", instruction, "", "### Wazuh alert:", json.dumps(alert_info, indent=2), "", "### Device information:", f"\n#### DEVICE {device} ####", ] for cmd, result in device_state.items(): data = result.get("data") or result.get("error", "No data") lines.append(f"\n# {cmd}\n{json.dumps(data, indent=2) if isinstance(data, dict) else str(data)}") lines.append("\n### Response (CLI FIX COMMANDS ONLY):\n") return "\n".join(lines) # ───────────────────────────────────────────────────────────────────────────── # SSH BRUTE FORCE DETECTOR # ───────────────────────────────────────────────────────────────────────────── _SSH_KEYWORDS = ( "sshd", "ssh", "failed password", "invalid user", "authentication failure", "failed login", ) _SSH_RULE_GROUPS = {"authentication_failed", "sshd", "ssh_auth"} class SSHBruteForceDetector: """ Track SSH login failures per source IP. Fires when the same IP accumulates >= threshold failures within window_secs. """ def __init__(self, threshold: int = 5, window_secs: int = 300): self.threshold = threshold self.window_secs = window_secs self._attempts: dict[str, list[float]] = defaultdict(list) def _is_ssh_failure(self, alert_raw: dict) -> bool: rule = alert_raw.get("rule", {}) groups = set(rule.get("groups", [])) desc = rule.get("description", "").lower() full = (alert_raw.get("full_log") or "").lower() return (groups & _SSH_RULE_GROUPS or any(kw in desc for kw in _SSH_KEYWORDS) or any(kw in full for kw in _SSH_KEYWORDS)) def _src_ip(self, alert_raw: dict) -> str | None: data = alert_raw.get("data", {}) or {} ip = data.get("srcip") or data.get("src_ip") if ip: return ip full = alert_raw.get("full_log") or "" m = _CISCO_SRC.search(full) or _IP_RE.search(full) return m.group(1) if m else None def feed(self, alert_raw: dict) -> tuple[bool, str | None, int]: """ Feed one alert. Returns (triggered, src_ip, count_so_far). """ if not self._is_ssh_failure(alert_raw): return False, None, 0 ip = self._src_ip(alert_raw) if not ip: return False, None, 0 now = time.monotonic() cutoff = now - self.window_secs self._attempts[ip] = [t for t in self._attempts[ip] if t >= cutoff] self._attempts[ip].append(now) count = len(self._attempts[ip]) if count >= self.threshold: self._attempts[ip] = [] # reset after detection return True, ip, count return False, ip, count # ───────────────────────────────────────────────────────────────────────────── # INCIDENT LOGGER # ───────────────────────────────────────────────────────────────────────────── def _now_iso() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def log_managed_incident(record: dict): """Append to managed_incidents.jsonl and write a human-readable summary line.""" MANAGED_LOG_JSONL.parent.mkdir(parents=True, exist_ok=True) with MANAGED_LOG_JSONL.open("a", encoding="utf-8") as fh: fh.write(json.dumps(record, ensure_ascii=False) + "\n") # Human-readable single-line summary patches_ok = sum(1 for p in record.get("patch_results", []) if p.get("ok")) patches_tot = len(record.get("patch_results", [])) status = record.get("status", "unknown") line = ( f"[{record['managed_at']}] #{record['alert_index']:04d} " f"device={record['device']:<14s} incident={record['incident_type']:<30s} " f"domain={record['domain']}-{record['sub_id']} " f"patches={patches_ok}/{patches_tot} status={status}\n" f" trigger : {record.get('trigger', '?')}\n" f" fix : {record.get('fix_commands_summary', '(see jsonl)')}\n" f" duration : {record.get('duration_seconds', 0):.1f}s\n" f" folder : {record.get('alert_folder', '-')}\n" + "-" * 80 + "\n" ) with MANAGED_LOG_READABLE.open("a", encoding="utf-8") as fh: fh.write(line) log.info("Incident logged -> %s [%s]", record["incident_type"], status) # ───────────────────────────────────────────────────────────────────────────── # CORE ALERT PROCESSOR # ───────────────────────────────────────────────────────────────────────────── _safe_re = re.compile(r"[^\w\-]") def _safe(s: str) -> str: return _safe_re.sub("_", s) def process_alert( alert_raw: dict, alert_index: int, runner: FixRunner, trigger: str = "level_alert", dry_run: bool = False, ) -> dict: """ Full autonomous pipeline for one alert. Returns the managed_incident record dict. """ t_start = time.monotonic() ts = _now_iso() log.info("\n" + "=" * 65) log.info(" Processing alert #%d trigger=%s", alert_index, trigger) log.info("=" * 65) # ── Step 1: Extract device ─────────────────────────────────────────────── device = _extract_device(alert_raw) rule = alert_raw.get("rule", {}) log.info("[1] Device: %s Rule: %s", device, rule.get("description", "?")) # ── Step 2: Classify with Ollama ───────────────────────────────────────── log.info("[2] Calling Ollama classifier...") full_log = alert_raw.get("full_log") or alert_raw.get("decoded") or "" wazuh_result = call_wazuh_llm(full_log) if not wazuh_result: log.warning("[2] Ollama unavailable — using rule fallback.") wazuh_result = _fallback_classify(alert_raw) incident_type = wazuh_result.get("incident_type", "unknown") severity = wazuh_result.get("severity", "high") explanation = wazuh_result.get("explanation", "") log.info("[2] incident_type=%s severity=%s", incident_type, severity) # Build the alert info dict passed to domain LLM alert_info = { **wazuh_result, "device": device, "rule": rule.get("description", ""), "timestamp": alert_raw.get("timestamp", ts), "agent": alert_raw.get("agent", {}), "trigger": trigger, } # Skip unmanaged devices early if device in UNMANAGED: log.warning("[2] Device '%s' is unmanaged — skipping RESTCONF/LLM.", device) rec = { "managed_at": ts, "alert_index": alert_index, "trigger": trigger, "device": device, "incident_type": incident_type, "severity": severity, "domain": "n/a", "sub_id": 0, "show_commands": [], "restconf_get_results": {}, "fix_commands": "SKIPPED (unmanaged device)", "fix_commands_summary": "skipped", "patch_results": [], "status": "skipped_unmanaged", "duration_seconds": round(time.monotonic() - t_start, 1), "alert_folder": None, "wazuh_classification": wazuh_result, } log_managed_incident(rec) return rec # ── Step 3: Show commands ──────────────────────────────────────────────── show_cmds = _get_show_commands(incident_type) log.info("[3] Show commands (%d): %s", len(show_cmds), show_cmds) # ── Step 4: Create alert folder ────────────────────────────────────────── folder_name = f"alert_{alert_index:03d}_{_safe(device)}_{_safe(incident_type)}" alert_folder = SHOW_OUTPUTS_DIR / folder_name alert_folder.mkdir(parents=True, exist_ok=True) # ── Step 5: Execute RESTCONF GETs ──────────────────────────────────────── log.info("[5] Executing RESTCONF GETs for %s ...", device) device_state = collect_device_state(device, show_cmds) # Save GET results to folder for audit (alert_folder / "alert_info.json").write_text( json.dumps(alert_info, indent=2, ensure_ascii=False), encoding="utf-8") (alert_folder / "restconf_get_results.json").write_text( json.dumps(device_state, indent=2, ensure_ascii=False), encoding="utf-8") # ── Step 6: Domain routing + LoRA fix ─────────────────────────────────── domain, sub_id = classify_domain(incident_type, explanation) lora_path = LORA_PATHS.get((domain, sub_id)) lora_rel = os.path.relpath(lora_path, PROJECT_ROOT) if lora_path else "n/a" log.info("[6] Domain: %s-%d LoRA: %s", domain, sub_id, lora_rel) fix_commands = "" if not lora_path or not lora_path.exists(): log.error("[6] LoRA adapter not found: %s", lora_path) fix_commands = f"[ERROR] LoRA adapter not found: {lora_path}" else: prompt = build_fix_prompt(domain, alert_info, device_state) raw_out = runner.run(lora_path, prompt) parts = re.split(r"### Response \(CLI FIX COMMANDS ONLY\):", raw_out, flags=re.I) fix_commands = parts[-1].strip() if len(parts) > 1 else raw_out.strip() log.info("[6] Fix commands:\n%s", fix_commands) (alert_folder / "fix_commands.txt").write_text(fix_commands, encoding="utf-8") # ── Step 7: Build RESTCONF PATCH operations ────────────────────────────── log.info("[7] Converting CLI fix to RESTCONF PATCH ops ...") patch_ops = build_restconf_patch_ops(device, fix_commands) (alert_folder / "restconf_fix_commands.json").write_text( json.dumps(patch_ops, indent=2, ensure_ascii=False), encoding="utf-8") # ── Step 8: Apply RESTCONF PATCHes ─────────────────────────────────────── patch_results = [] if dry_run: log.info("[8] DRY RUN — skipping %d PATCH operation(s).", len(patch_ops)) for op in patch_ops: patch_results.append({"url": op["url"], "dry_run": True, "ok": None}) else: log.info("[8] Applying %d RESTCONF PATCH operation(s) ...", len(patch_ops)) for op in patch_ops: result = restconf_patch(op["url"], op.get("body")) patch_results.append({"url": op["url"], **result}) flag = "OK" if result.get("ok") else f"FAIL({result.get('status')})" log.info(" PATCH %s -> %s", op["url"], flag) (alert_folder / "patch_results.json").write_text( json.dumps(patch_results, indent=2, ensure_ascii=False), encoding="utf-8") # ── Step 9: Log managed incident ───────────────────────────────────────── ok_count = sum(1 for r in patch_results if r.get("ok") is True) tot_count = len(patch_results) if dry_run: status = "dry_run" elif tot_count == 0: status = "no_patches" elif ok_count == tot_count: status = "fixed" elif ok_count > 0: status = "partial" else: status = "failed" summary_lines = fix_commands.splitlines() fix_summary = " | ".join(summary_lines[:3]) + ("..." if len(summary_lines) > 3 else "") record = { "managed_at": ts, "alert_index": alert_index, "trigger": trigger, "device": device, "incident_type": incident_type, "severity": severity, "domain": domain, "sub_id": sub_id, "lora_used": lora_rel, "show_commands": show_cmds, "restconf_get_results": { cmd: {"ok": r.get("ok"), "status": r.get("status")} for cmd, r in device_state.items() }, "fix_commands": fix_commands, "fix_commands_summary": fix_summary, "restconf_patch_count": tot_count, "patch_results": patch_results, "status": status, "duration_seconds": round(time.monotonic() - t_start, 1), "alert_folder": str(alert_folder), "wazuh_classification": wazuh_result, "raw_alert_rule": rule.get("description", ""), "raw_alert_level": rule.get("level"), } log_managed_incident(record) log.info( "[DONE] Alert #%d: %s/%s -> %s patches=%d/%d %.1fs", alert_index, device, incident_type, status, ok_count, tot_count, record["duration_seconds"], ) return record # ───────────────────────────────────────────────────────────────────────────── # ALERT FILE WATCHER (tail -f style) # ───────────────────────────────────────────────────────────────────────────── class AlertTailer: """ Efficiently tails a growing JSON-lines file. Seeks to EOF on first open (skips historical alerts), then yields newly appended lines as they arrive. """ def __init__(self, path: Path, min_level: int, max_level: int): self.path = path self.min_level = min_level self.max_level = max_level self._fh = None self._pos = 0 def _open(self): if self._fh is None: if not self.path.exists(): return False self._fh = self.path.open("r", encoding="utf-8", errors="replace") self._fh.seek(0, 2) # seek to end — only process NEW alerts self._pos = self._fh.tell() log.info("Watching: %s (from position %d)", self.path, self._pos) return True def _reopen_if_rotated(self): """Detect log rotation: if file shrank, re-open from the beginning.""" try: current_size = self.path.stat().st_size except FileNotFoundError: return if current_size < self._pos: log.info("Log rotation detected — re-opening %s", self.path) self._fh.close() self._fh = None self._pos = 0 self._open() def poll(self) -> list[dict]: """Return any new valid alerts since last poll.""" if not self._open(): return [] self._reopen_if_rotated() new_alerts = [] while True: line = self._fh.readline() if not line: break self._pos = self._fh.tell() line = line.strip() if not line: continue try: data = json.loads(line) except json.JSONDecodeError: continue level = data.get("rule", {}).get("level", 0) if self.min_level <= level <= self.max_level: new_alerts.append(data) return new_alerts def close(self): if self._fh: self._fh.close() self._fh = None # ───────────────────────────────────────────────────────────────────────────── # DAEMON # ───────────────────────────────────────────────────────────────────────────── class IncidentDaemon: def __init__(self, alerts_file: Path, poll_secs: int, min_level: int, max_level: int, ssh_threshold: int, ssh_window: int, dry_run: bool): self.tailer = AlertTailer(alerts_file, min_level, max_level) self.ssh_detector = SSHBruteForceDetector(ssh_threshold, ssh_window) self.runner = FixRunner(BASE_MODEL) self.poll_secs = poll_secs self.dry_run = dry_run self._running = True self._alert_index = 0 self._processed = 0 SHOW_OUTPUTS_DIR.mkdir(parents=True, exist_ok=True) def _next_index(self) -> int: self._alert_index += 1 return self._alert_index def _handle(self, alert_raw: dict, trigger: str): idx = self._next_index() try: process_alert( alert_raw=alert_raw, alert_index=idx, runner=self.runner, trigger=trigger, dry_run=self.dry_run, ) self._processed += 1 except Exception as exc: log.exception("Unhandled error processing alert #%d: %s", idx, exc) def run(self): log.info("=" * 65) log.info(" Incident Response Daemon started") log.info(" Alerts file : %s", self.tailer.path) log.info(" Poll every : %ds", self.poll_secs) log.info(" Level range : %d-%d", self.tailer.min_level, self.tailer.max_level) log.info(" SSH trigger : %d failures / %ds window", self.ssh_detector.threshold, self.ssh_detector.window_secs) log.info(" Dry run : %s", self.dry_run) log.info(" Output log : %s", MANAGED_LOG_JSONL) log.info("=" * 65) while self._running: new_alerts = self.tailer.poll() for alert_raw in new_alerts: level = alert_raw.get("rule", {}).get("level", 0) # SSH brute-force pattern check (runs for every alert regardless of level) triggered, src_ip, count = self.ssh_detector.feed(alert_raw) if triggered: log.warning( "SSH brute-force pattern detected: %d failures from %s within %ds", count, src_ip, self.ssh_detector.window_secs, ) # Synthesise a brute-force alert record ssh_alert = dict(alert_raw) ssh_alert.setdefault("rule", {})["description"] = ( f"SSH brute-force: {count} failures from {src_ip}" ) self._handle(ssh_alert, trigger=f"ssh_brute_force:{src_ip}:{count}") # Standard level-based alert elif self.tailer.min_level <= level <= self.tailer.max_level: log.info( "New alert: level=%d rule=%s", level, alert_raw.get("rule", {}).get("description", "?"), ) self._handle(alert_raw, trigger=f"level{level}_alert") if not new_alerts: time.sleep(self.poll_secs) def stop(self): log.info("Daemon shutting down (processed %d alerts).", self._processed) self._running = False self.tailer.close() # ───────────────────────────────────────────────────────────────────────────── # ENTRY POINT # ───────────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Autonomous Wazuh-LLM Incident Response Daemon" ) parser.add_argument( "--alerts", default=str(DEFAULT_ALERTS_FILE), help=f"Wazuh alerts JSON-lines file (default: {DEFAULT_ALERTS_FILE})" ) parser.add_argument( "--poll", type=int, default=3, help="File poll interval in seconds (default: 3)" ) parser.add_argument( "--min-level", type=int, default=7, help="Minimum Wazuh rule level to process (default: 7)" ) parser.add_argument( "--max-level", type=int, default=12, help="Maximum Wazuh rule level to process (default: 12)" ) parser.add_argument( "--ssh-threshold", type=int, default=5, help="SSH failures from same IP to trigger brute-force alert (default: 5)" ) parser.add_argument( "--ssh-window", type=int, default=300, help="SSH brute-force detection window in seconds (default: 300)" ) parser.add_argument( "--dry-run", action="store_true", help="Classify and generate fix commands but do NOT apply RESTCONF PATCHes" ) args = parser.parse_args() alerts_path = Path(args.alerts) if not alerts_path.parent.exists(): log.warning( "Alerts directory does not exist: %s\n" " Daemon will wait until the file appears.", alerts_path.parent ) daemon = IncidentDaemon( alerts_file = alerts_path, poll_secs = args.poll, min_level = args.min_level, max_level = args.max_level, ssh_threshold = args.ssh_threshold, ssh_window = args.ssh_window, dry_run = args.dry_run, ) def _signal_handler(sig, _frame): print(f"\nSignal {sig} received — stopping daemon...") daemon.stop() sys.exit(0) signal.signal(signal.SIGINT, _signal_handler) signal.signal(signal.SIGTERM, _signal_handler) daemon.run() if __name__ == "__main__": main()