"""Tidy an OpenAI-compatible base URL so common shorthands just work.

import re

from urllib.parse import urlparse

url = (url or "").strip()

if not url:

return ""

if not re.match(r"^https?://", url, re.IGNORECASE):

url = "http://" + url

url = url.rstrip("/")

if urlparse(url).path in ("", "/"):

url = url + "/v1"

"""AI-powered network analysis, vulnerability interpretation, and insights."""

def __init__(self, shared_data):

self.shared_data = shared_data

self.logger = Logger(name="AIService", level=logging.INFO)

self.env_manager = EnvManager()

cfg = shared_data.config

# Configuration

self.enabled = cfg.get("ai_enabled", False)

self.model = cfg.get("ai_model", "gpt-5.4-nano")

# Self-hosted / OpenAI-compatible endpoint support (issue #462).

# When ai_base_url is set (e.g. an Ollama/LocalAI/vLLM/LM Studio server

# on the LAN or mesh), Ragnar talks to it as a thin HTTP client and

# uses the Chat Completions API — local servers implement /v1/chat/

# completions, not OpenAI's proprietary Responses API. Empty base URL

# keeps the default OpenAI Responses behavior unchanged.

self.base_url = normalize_base_url(cfg.get("ai_base_url", ""))

self.use_chat_api = self._resolve_use_chat_api()

# OpenAI-cloud fallback for when a self-hosted endpoint goes away

# (issue #462). Built lazily the first time it's needed.

self._fallback_client = None

self._fallback_model = None

self.fallback_active = False # True while running on the cloud fallback

# These must remain for backward compatibility (but not used)

self.max_tokens = cfg.get("ai_max_tokens")

self.temperature = cfg.get("ai_temperature")

self.temperature_supported = True # will disable on first failure

self.vulnerability_summaries = cfg.get("ai_vulnerability_summaries", True)

self.network_insights = cfg.get("ai_network_insights", True)

self.api_token = self.env_manager.get_token()

# Cache

self.cache = {}

self.cache_ttl = 3600 # 1 hour (3600 seconds) - reduce token consumption

# Client initialization

self.client = None

self.initialization_error = None

self._initialize_client()

# ===================================================================

# INITIALIZATION

# ===================================================================

def _resolve_use_chat_api(self):

"""Decide which API dialect to speak.

style = str(self.shared_data.config.get("ai_api_style", "auto") or "auto").strip().lower()

if style == "chat":

return True

if style == "responses":

return False

return bool(self.base_url)

def _initialize_client(self):

if not self.enabled:

return

# A real key is mandatory for OpenAI's cloud, but self-hosted endpoints

# usually accept any non-empty string — the SDK still requires one, so

# fall back to a harmless placeholder when the operator left it blank.

if not self.api_token and not self.base_url:

self.initialization_error = "No OpenAI API key found."

self.logger.warning(self.initialization_error)

return

if OpenAI is None:

self.client = None

self.initialization_error = _OPENAI_MISSING_MSG

self.logger.warning(self.initialization_error)

return

try:

# Bound every request: the SDK default is 600s, so a single stuck or

# rate-limited call (common when several field units share one API

# key and each fans out concurrent dashboard calls) would otherwise

# hang for minutes — well past any client timeout — instead of

# failing fast and letting the rest of the insights render. One retry

# keeps the worst-case bounded (~40s/call) so the background insights

# run resolves quickly even on a slow board.

client_kwargs = {

"api_key": self.api_token or "ragnar-local",

"timeout": 20.0,

"max_retries": 1,

}

if self.base_url:

client_kwargs["base_url"] = self.base_url

self.client = OpenAI(**client_kwargs)

self.initialization_error = None

endpoint = self.base_url or "OpenAI"

api_dialect = "chat.completions" if self.use_chat_api else "responses"

self.logger.info(

f"AI Service initialized (model: {self.model}, endpoint: {endpoint}, api: {api_dialect})"

)

except Exception as exc:

self.client = None

self.initialization_error = f"OpenAI client initialization failed: {exc}"

self.logger.error(self.initialization_error)

def reload_token(self) -> bool:

"""Refresh the API token from disk and reinitialize the OpenAI client."""

# Keep enabled flag + endpoint config synced with latest config intent

if hasattr(self.shared_data, "config"):

cfg = self.shared_data.config

self.enabled = cfg.get("ai_enabled", self.enabled)

self.model = cfg.get("ai_model", self.model)

self.base_url = normalize_base_url(cfg.get("ai_base_url", ""))

self.use_chat_api = self._resolve_use_chat_api()

# Force the cloud fallback to rebuild with the latest token/model.

self._fallback_client = None

self.fallback_active = False

self.api_token = self.env_manager.get_token()

self.client = None

self.initialization_error = None

if not self.enabled:

self.logger.info("AI service disabled in config; skipping token reload.")

return False

if not self.api_token and not self.base_url:

self.logger.warning("AI token reload requested but no token present in environment.")

self.initialization_error = "No OpenAI API key found."

return False

self._initialize_client()

success = self.client is not None and self.initialization_error is None

if success:

self.logger.info("AI service reloaded with updated token.")

else:

if self.initialization_error:

self.logger.error(

f"AI service failed to reinitialize after token reload: {self.initialization_error}"

)

else:

self.logger.error("AI service failed to reinitialize after token reload.")

return success

def complete(self, user_msg: str, system_msg: Optional[str] = None) -> Optional[str]:

"""One-shot completion returning the model's raw text (or None).

if system_msg is None:

system_msg = ("You are a security automation assistant. Output exactly one "

"JSON object and nothing else — no prose, no markdown fences.")

return self._ask(system_msg, user_msg)

# ===================================================================

# UTILITY HELPERS

# ===================================================================

def is_enabled(self):

"""Return True when the service is enabled and the client is ready."""

return self.ensure_ready()

def ensure_ready(self):

"""Lazily initialize the OpenAI client if configuration says AI is enabled."""

# Sync enabled state with config in case it changed

if hasattr(self.shared_data, "config"):

self.enabled = self.shared_data.config.get("ai_enabled", self.enabled)

if not self.enabled:

return False

# Already initialized and healthy

if self.client is not None and self.initialization_error is None:

return True

# Don't keep retrying when we've already recorded a permanent failure

# But allow retry if token was added after initial failure

if self.initialization_error and self.api_token:

# Clear error and retry if we have a token now

self.initialization_error = None

# Refresh token from disk if we don't have one yet

if not self.api_token:

self.api_token = self.env_manager.get_token()

# A token is only mandatory for the OpenAI cloud; a self-hosted base URL

# can run keyless.

if not self.api_token and not self.base_url:

self.initialization_error = "No OpenAI API key found."

self.logger.warning(self.initialization_error)

return False

self._initialize_client()

return self.client is not None and self.initialization_error is None

def _cache_key(self, name: str, content: Any):

import hashlib

raw = f"{name}:{json.dumps(content, sort_keys=True)}"

return hashlib.md5(raw.encode()).hexdigest()

def _cache_get(self, key: str):

item = self.cache.get(key)

if not item:

return None

if time.time() - item["timestamp"] > self.cache_ttl:

del self.cache[key]

return None

return item["value"]

def _cache_set(self, key: str, value: Any):

self.cache[key] = {"timestamp": time.time(), "value": value}

# ===================================================================

# CORE GPT-5 CALL — NEW RESPONSES API

# ===================================================================

def _ask(self, system_msg: str, user_msg: str) -> Optional[str]:

"""

if not self.is_enabled():

return None

if self.client is None:

self.logger.error("AI client unavailable despite service being enabled.")

return None

# Self-hosted / OpenAI-compatible servers speak Chat Completions, not

# the proprietary Responses API — route them separately.

if self.use_chat_api:

return self._ask_chat(system_msg, user_msg)

return self._ask_responses(system_msg, user_msg)

def _ask_responses(self, system_msg, user_msg, client=None, model=None):

"""OpenAI Responses-API call (GPT-5 family). Reused for both the primary

client = client or self.client

model = model or self.model

payload = {

"model": model,

"input": [

{"role": "system", "content": system_msg},

{"role": "user", "content": user_msg},

],

"reasoning": {"effort": "low"},

"text": {"verbosity": "low"},

}

# Include temperature ONLY if still marked supported

if self.temperature_supported and self.temperature is not None:

payload["temperature"] = self.temperature

# FIRST ATTEMPT

try:

result = client.responses.create(**payload)

return self._extract_output(result)

except Exception as e:

error_text = str(e).lower()

# Handle GPT-5 "temperature unsupported" case

if "temperature" in error_text and "unsupported" in error_text:

self.temperature_supported = False

self.logger.warning(

"Model reported temperature as unsupported — retrying without it."

)

payload.pop("temperature", None)

# SECOND ATTEMPT WITHOUT TEMPERATURE

try:

result = client.responses.create(**payload)

return self._extract_output(result)

except Exception as e2:

self.logger.error(f"Retry after removing temperature failed: {e2}")

return None

self.logger.error(f"OpenAI call failed: {e}")

return None

def _ask_chat(self, system_msg: str, user_msg: str) -> Optional[str]:

"""Chat Completions call for self-hosted OpenAI-compatible endpoints.

payload = {

"model": self.model,

"messages": [

{"role": "system", "content": system_msg},

{"role": "user", "content": user_msg},

],

}

if self.temperature_supported and self.temperature is not None:

payload["temperature"] = self.temperature

try:

result = self.client.chat.completions.create(**payload)

except Exception as e:

error_text = str(e).lower()

if "temperature" in error_text and payload.pop("temperature", None) is not None:

self.temperature_supported = False

self.logger.warning("Endpoint rejected temperature — retrying without it.")

try:

result = self.client.chat.completions.create(**payload)

except Exception as e2:

return self._maybe_fallback(system_msg, user_msg, e2)

else:

return self._maybe_fallback(system_msg, user_msg, e)

# A successful self-hosted call clears any prior fallback state.

self.fallback_active = False

# Usage fields on OpenAI-compatible servers use prompt/completion naming.

usage = getattr(result, "usage", None)

if usage is not None:

self.logger.info(

"AI Tokens → input:{} output:{} total:{}".format(

getattr(usage, "prompt_tokens", "?"),

getattr(usage, "completion_tokens", "?"),

getattr(usage, "total_tokens", "?"),

)

)

try:

content = result.choices[0].message.content

return content.strip() if content else None

except (AttributeError, IndexError, TypeError):

self.logger.error("Self-hosted AI response had no message content.")

return None

def _maybe_fallback(self, system_msg, user_msg, error):

"""Retry a failed self-hosted call on OpenAI's cloud when the endpoint

if self._should_fallback(error):

fb = self._ensure_fallback_client()

if fb is not None:

self.logger.warning(

f"Self-hosted endpoint unavailable ({error}); falling back to "

f"OpenAI cloud (model: {self._fallback_model})."

)

self.fallback_active = True

return self._ask_responses(

system_msg, user_msg, client=fb, model=self._fallback_model

)

self.logger.error(f"Self-hosted AI call failed: {error}")

return None

def _should_fallback(self, error):

"""True when the error looks like the endpoint is unreachable / timed

if type(error).__name__ in (

"APIConnectionError", "APITimeoutError", "InternalServerError",

"ConnectionError", "Timeout", "ConnectTimeout", "ReadTimeout",

):

return True

msg = str(error).lower()

return any(s in msg for s in (

"connection", "timed out", "timeout", "refused", "unreachable",

"failed to establish", "max retries", "connect", "temporarily unavailable",

))

def _ensure_fallback_client(self):

"""Lazily build an OpenAI-cloud client used only for fallback. Requires a

if not self.base_url:

return None # already on the cloud — no separate fallback needed

if self._fallback_client is not None:

return self._fallback_client

if OpenAI is None:

return None # SDK absent — no cloud fallback possible

token = self.api_token or self.env_manager.get_token()

if not token:

return None

try:

self._fallback_client = OpenAI(api_key=token, timeout=20.0, max_retries=1)

self._fallback_model = (

str(self.shared_data.config.get("ai_fallback_model", "") or "").strip()

or "gpt-5.4-nano"

)

self.logger.info(f"AI cloud fallback ready (model: {self._fallback_model}).")

return self._fallback_client

except Exception as e:

self.logger.warning(f"Could not initialize AI fallback client: {e}")

return None

def _extract_output(self, result):

"""Extract output text and log token usage."""

if hasattr(result, "usage"):

u = result.usage

self.logger.info(

f"AI Tokens → input:{u.input_tokens} output:{u.output_tokens} total:{u.total_tokens}"

)

try:

return result.output_text.strip()

except:

return None

# ===================================================================

# NETWORK SUMMARY

# ===================================================================

def analyze_network_summary(self, network_data):

if not self.is_enabled() or not self.network_insights:

return None

key = self._cache_key("summary", network_data)

cached = self._cache_get(key)

if cached:

return cached

system = (

"You are Ragnar, a witty cybersecurity Viking AI. "

"Provide concise, aggressive but clear summaries."

)

user = f"""

resp = self._ask(system, user)

if resp:

self._cache_set(key, resp)

return resp

# ===================================================================

# VULNERABILITY ANALYSIS

# ===================================================================

def analyze_vulnerabilities(self, vulnerabilities: List[Dict]):

if not self.is_enabled() or not self.vulnerability_summaries:

return None

key = self._cache_key("vuln_analysis", {"count": len(vulnerabilities)})

cached = self._cache_get(key)

if cached:

return cached

limited = vulnerabilities[:10]

data_json = json.dumps(limited, indent=2)

system = (

"You are Ragnar, an elite vulnerability hunter. "

"Structure your analysis clearly with sections and bullet points. "

"Use markdown-style formatting for readability."

)

user = f"""

resp = self._ask(system, user)

if resp:

self._cache_set(key, resp)

return resp

# ===================================================================

# PCAP ANALYSIS (general network fault diagnosis)

# ===================================================================

def analyze_pcap(self, data: Dict):

"""Interpret a PCAP-analysis summary (from network_diagnostics) and tell

if not self.is_enabled():

return None

wifi = (data or {}).get('wifi') or {}

summary = (data or {}).get('summary') or {}

expert = (data or {}).get('expert') or {}

protos = (data or {}).get('protocols') or []

# Key the cache on what actually drives the verdict, not just packet count

# + Wi-Fi fields — otherwise two unrelated non-Wi-Fi captures with the same

# packet count collide and get served each other's analysis. The expert

# findings + protocol mix uniquely fingerprint what the AI will say.

key = self._cache_key("pcap_analysis", {

"pkts": summary.get('packets'),

"bytes": summary.get('data_size'),

"dur": summary.get('duration_s'),

"exp": (expert.get('errors'), expert.get('warnings'), expert.get('notes')),

"items": [(i.get('protocol'), i.get('summary'), i.get('count'))

for i in (expert.get('items') or [])[:10]],

"protos": [(p.get('proto'), p.get('bytes')) for p in protos[:10]],

"wifi": {

"deauth": (wifi.get('deauth') or {}).get('total'),

"disassoc": (wifi.get('disassoc') or {}).get('total'),

"retry": wifi.get('retry_pct'),

} if wifi else None,

})

cached = self._cache_get(key)

if cached:

return cached

# Keep the prompt compact: send the summary, expert findings, and (if

# present) the Wi-Fi event breakdown that actually explains drops.

payload = {

'summary': summary,

'expert': (data or {}).get('expert'),

'top_protocols': ((data or {}).get('protocols') or [])[:15],

'top_talkers': ((data or {}).get('talkers') or [])[:8],

'wifi': wifi or None,

}

data_json = json.dumps(payload, indent=2, default=str)[:6000]

is_wifi = bool(wifi.get('is_wifi'))

system = (

"You are a senior network engineer triaging a packet capture. Decide whether the "

"network is HEALTHY or has a real problem, and if so diagnose the ROOT CAUSE from "

"the evidence in plain language a field tech can act on. Reason across the whole "

"stack: L2 (ARP/broadcast/multicast storms, duplicate IPs, spanning-tree churn), "

"TCP health (retransmissions, resets, zero-windows, dup-ACKs, out-of-order, "

"high RTT), DNS and DHCP failures/latency, TLS/handshake errors, and — for "

"802.11 captures — deauth/disassoc reason codes and retry rates that drive "

"client drops. "

"CALIBRATE SEVERITY AGAINST VOLUME: a small number of RSTs, duplicate ACKs, "

"retransmissions, and normal FIN-based connection closes is EXPECTED, especially "

"for ordinary web/TLS browsing — judge by RATES relative to total packets/bytes "

"and to the number of conversations, not by raw counts. Normal connection teardown "

"(FIN / 'connection closing') is NOT a fault, and a few RSTs are how clients "

"routinely close sessions. Do not describe the network as 'unstable' or 'degraded' "

"unless the error rate is clearly disproportionate to the traffic. If the capture "

"looks healthy, SAY SO plainly (make that the Verdict) rather than manufacturing a "

"problem. "

"Be specific and quote the counts/codes. Do not invent data not present in the "

"summary. When you cite a Wireshark/tshark display filter or CLI command, use only "

"valid, real syntax (e.g. tcp.analysis.retransmission, tcp.flags.reset==1, "

"tcp.analysis.duplicate_ack) and parenthesize mixed AND/OR filters correctly; if "

"unsure of the exact filter, describe what to match in words rather than inventing "

"a field name. Use short markdown sections and bullet points."

)

focus = (

"This is a Wi-Fi / access-point capture: pay special attention to WHY CLIENTS "

"ARE DROPPING / DISCONNECTING (deauth/disassoc reason codes, retry %, AP "

"behavior), but still flag any other network problems you see." if is_wifi else

"Diagnose the most impactful problems in this capture across every layer — "

"not just one — and rank them by how much they hurt the network. If nothing "

"rises above normal background noise, say the capture looks healthy."

)

user = f"""{focus}

resp = self._ask(system, user)

if resp:

self._cache_set(key, resp)

return resp

def analyze_signal(self, question: str, context: Dict, facts: Dict = None):

"""RF-analyst assistant for the Signal Analyzer (SigMF IQ captures).

if not self.is_enabled():

return None

merged = dict(context or {})

if facts:

merged["measured"] = facts

ctx_json = json.dumps(merged, indent=2, default=str)[:7000]

q = (question or "").strip() or "What is this signal, and how would I decode it?"

system = (

"You are a senior RF / SIGINT and signal-reverse-engineering analyst helping a user "

"at a software-defined-radio workbench. They have captured raw IQ (a SigMF recording "

"from an RTL-SDR, ~24 MHz-1.7 GHz, receive-only) and are looking at an on-device "

"analyzer: spectrogram, PSD, envelope, automatic burst detection, an OOK/AM & FSK/FM "

"demodulator that recovers a bitstream, a modulation classifier, and a protocol tool "

"(preamble / repeated-frame alignment / CRC scan). "

"Your job: interpret the measurements and TEACH the user how a professional would "

"proceed - what the signal likely is, which band/service it belongs to, what "

"modulation and symbol rate the numbers imply, how to set the demod (centre offset, "

"bandwidth, OOK vs FSK), how to read the recovered bits / frame structure / CRC, and "

"the single best NEXT step. "

"IMPORTANT HONESTY RULES: the analyzer's dB values are RELATIVE (uncalibrated), not "

"absolute dBm - never present them as calibrated power. LoRa / chirp-spread and most "

"digital mesh (Meshtastic) CANNOT be demodulated here - it is energy/occupancy only; "

"say so. Rolling-code remotes and unlisted protocols won't be 'named' by rtl_433 - "

"explain what the bits still tell you (fixed vs rolling fields). Do NOT invent values "

"that aren't in the provided context; if something needed isn't there, tell the user "

"which analyzer control to use to get it. Reference the analyzer's own controls by "

"name (Zoom/Measure, Demodulate, BW, Frames, Decode, Bin/Hex/Manchester). Be concise "

"and practical; use short markdown sections/bullets. This is receive-only analysis.\n\n"

"TAKING ACTIONS: you can DRIVE the analyzer. When a concrete next step would help, END "

"your reply with a fenced ```json block: {\"actions\":[ ... ]} — the user gets one-click "

"buttons to run them (nothing runs without their click). Use ONLY these actions and only "

"with values grounded in the data shown:\n"

" {\"action\":\"tune\",\"f_mhz\":<centre>,\"bw_khz\":<width>,\"mode\":\"ook|fsk\"} — mark + zoom a signal and set demod BW\n"

" {\"action\":\"demod\",\"mode\":\"ook|fsk\",\"bw_khz\":<width>} — demodulate to bits\n"

" {\"action\":\"classify\"} — auto modulation classification\n"

" {\"action\":\"frames\",\"line\":\"raw|manchester|nrzi\"} — preamble/frame/CRC analysis\n"

" {\"action\":\"decode433\"} — run rtl_433 to name known ISM devices\n"

" {\"action\":\"zoom\",\"t0\":<s>,\"t1\":<s>,\"f0_mhz\":<lo>,\"f1_mhz\":<hi>}\n"

" {\"action\":\"reset\"} — reset zoom\n"

"Propose at most 4 actions, in the order to try them. Omit the block entirely if there's "

"nothing concrete to do. Keep the readable answer ABOVE the block; put no prose after it."

)

user = (

"Here is the current analyzer state for the selected capture (JSON):\n\n"

+ ctx_json

+ "\n\nUser question:\n" + q[:2000]

+ "\n\nAnswer as their RF analyst: interpret what's measured, be specific about the "

"numbers present, and end with a concrete next step using the analyzer's controls."

)

return self._ask(system, user)

# ===================================================================

# ATTACK VECTOR IDENTIFICATION

# ===================================================================

def identify_network_weaknesses(self, network_data: Dict, findings: List[Dict]):

if not self.is_enabled():

return None

key = self._cache_key("weakness", {

"targets": network_data.get("target_count"),

"findings": len(findings),

})

cached = self._cache_get(key)

if cached:

return cached

sample = json.dumps(findings[:5], indent=2)

system = (

"You are Ragnar, a penetration strategist. "

"Structure attack vector analysis clearly with numbered attack paths. "

"Use markdown formatting for readability."

)

user = f"""

resp = self._ask(system, user)

if resp:

self._cache_set(key, resp)

return resp

# ===================================================================

# WI-FI CONNECTION ANALYSIS (professional tone, no persona)

# ===================================================================

def analyze_wifi_connection(self, context: Dict):

"""Assess the user's current Wi-Fi connection and RF environment from a

if not self.is_enabled():

return None

connected = (context or {}).get("connected")

bt = (context or {}).get("bt")

zb = (context or {}).get("zigbee")

key = self._cache_key("wifi_conn", {

"bssid": (connected or {}).get("bssid"),

"signal": (connected or {}).get("signal_dbm"),

"cochannel": (context or {}).get("co_channel_count"),

"aps": (context or {}).get("ap_total"),

# Include the non-Wi-Fi 2.4 GHz overlays so enabling/refreshing them

# busts the cache instead of serving a Wi-Fi-only verdict.

"bt": (bt or {}).get("device_count") if bt else None,

"zb": (zb or {}).get("device_count") if zb else None,

})

cached = self._cache_get(key)

if cached:

return cached

# Which non-Wi-Fi 2.4 GHz sources are present shapes what we ask for.

overlays = []

if bt:

overlays.append("Bluetooth/BLE")

if zb:

overlays.append("Zigbee / 802.15.4")

overlay_line = (

(" The data also includes non-Wi-Fi 2.4 GHz emitters discovered "

"alongside the Wi-Fi scan (" + " and ".join(overlays) + "), under "

"the 'bt' and/or 'zigbee' keys, with an estimated per-Wi-Fi-channel "

"pressure for channels 1/6/11/13. Treat these as coexistence load on "

"the 2.4 GHz band: factor them into 2.4 GHz channel advice and, when "

"they are significant, recommend moving affected clients to 5/6 GHz. "

"The pressure figures are heuristic activity estimates, not measured "

"energy — say so and do not overstate them.") if overlays else ""

)

data_json = json.dumps(context, indent=2, default=str)[:6000]

system = (

"You are a professional wireless network engineer. Analyze the user's "

"current Wi-Fi connection and the surrounding RF environment from the "

"scan data provided, and give clear, actionable recommendations to fix "

"or improve it. Use a neutral, professional tone — no personas, no "

"slang, no jokes. Base every statement strictly on the data provided; "

"do not invent SSIDs, channels or values that are not present. If the "

"user is not associated to a network, say so and analyze the "

"environment instead. Use short markdown sections and bullet points."

+ overlay_line

)

user = f"""Analyze this Wi-Fi connection and RF environment.

resp = self._ask(system, user)

if resp:

self._cache_set(key, resp)

return resp

# ===================================================================

# SECURITY POSTURE (Watchtower + Wi-Fi Defense passive monitoring)

# ===================================================================

def analyze_security_posture(self, posture: Dict):

"""Summarize the passive-monitoring posture from Watchtower's aggregated

if not self.is_enabled():

return None

key = self._cache_key("posture", {

"total": (posture or {}).get("total_alerts"),

"worst": (posture or {}).get("worst"),

"by_sev": (posture or {}).get("by_severity"),

})

cached = self._cache_get(key)

if cached:

return cached

data_json = json.dumps(posture, indent=2, default=str)[:5000]

system = (

"You are a professional security operations analyst. Summarize the "

"current passive-monitoring posture from the alert data. Watchtower "

"aggregates Ragnar's passive watchers, including the Wi-Fi Defense "

"family (wifiwatch, legacywatch, wpswatch) and wired monitors "

"(ndpwatch, arp_guard, …). Use a neutral, professional tone. Base "

"every statement on the data; do not invent alerts. Be concise."

)

user = f"""Summarize this passive security-monitoring posture.

resp = self._ask(system, user)

if resp:

self._cache_set(key, resp)

return resp

# ===================================================================

# WI-FI DEFENSE (WIDS + airtime + client-isolation, one read)

# ===================================================================

def analyze_wifi_defense(self, context: Dict):

"""One professional read across the three WiFi Defense modules for a

if not self.is_enabled():

return None

wids = (context or {}).get('wids') or {}

airtime = (context or {}).get('airtime') or {}

isolation = (context or {}).get('isolation') or {}

key = self._cache_key("wifidef", {

"threat": wids.get('threat'),

"dets": len(wids.get('detections') or []),

"at": len(airtime.get('findings') or []),

"iso": [(b.get('bssid'), b.get('verdict')) for b in (isolation.get('bss') or [])[:12]],

"frames": wids.get('frames'),

})

cached = self._cache_get(key)

if cached:

return cached

present = []

if wids:

present.append("WIDS")

if airtime:

present.append("airtime/link-quality")

if isolation:

present.append("client-isolation")

data_json = json.dumps(context, indent=2, default=str)[:6500]

system = (

"You are a professional wireless security operations analyst reviewing "

"a single passive 802.11 monitor-mode capture through three lenses: an "