"""Retrieval-augmented Q&A over the stored log events. Offline path (always available): keyword/TF-IDF ranking over rows fetched from the DB + a rule-based Thai summary. Optional path: when OPENROUTER_API_KEY is set, the retrieved context is sent to an LLM for a richer answer. Any LLM error falls back to the local summary so the system never hard-fails. """ from __future__ import annotations import math from collections import Counter, defaultdict from typing import Any import httpx from .classifier import CATEGORY_META, detect_rule, expand_query, row_text, tokenize from .config import get_settings from .db import search_rows async def retrieve(query: str, limit: int = 8) -> list[dict[str, Any]]: """Keyword retrieval: pull candidate rows then TF-IDF rank in-process.""" terms, category_terms = expand_query(query) # Pull a candidate pool (filtered by the raw query when possible). pool = await search_rows(query, limit=600) if not pool: pool = await search_rows("", limit=600) if not terms or not pool: return [_as_hit(r, 1.0) for r in pool[:limit]] # Build a tiny TF-IDF index over the candidate pool. docs = [Counter(tokenize(row_text(r))) for r in pool] doc_freq: Counter[str] = Counter() for d in docs: doc_freq.update(d.keys()) total = max(len(docs), 1) idf = {t: math.log((1 + total) / (1 + df)) + 1 for t, df in doc_freq.items()} query_counts = Counter(terms) scores: defaultdict[int, float] = defaultdict(float) for idx, doc in enumerate(docs): for term, qtf in query_counts.items(): tf = doc.get(term, 0) if tf: scores[idx] += (1 + math.log(tf)) * idf.get(term, 1.0) * qtf if pool[idx].get("category") in category_terms: scores[idx] += 12 if query.lower() in row_text(pool[idx]).lower(): scores[idx] += 8 ranked = sorted(scores.items(), key=lambda kv: kv[1], reverse=True)[:limit] return [_as_hit(pool[idx], round(score, 3)) for idx, score in ranked] def _as_hit(row: dict[str, Any], score: float) -> dict[str, Any]: return {"score": score, "row": row, "rule": detect_rule(row), "retrieval": "keyword"} def answer_local(question: str, hits: list[dict[str, Any]]) -> dict[str, Any]: rows = [h["row"] for h in hits] by_cat = Counter(r.get("category") for r in rows) by_sev = Counter(r.get("severity") for r in rows) if not rows: answer = ("## ยังไม่พบหลักฐาน\n\n" "- ลองถามด้วย IP, category เช่น `brute_force`, `c2_beacon`\n" "- หรืออัปโหลด/ส่ง mock log ก่อนแล้วถามใหม่") else: top = rows[0] meta = CATEGORY_META.get(top.get("category"), CATEGORY_META["benign"]) cats = ", ".join(c for c, _ in by_cat.most_common(3) if c) answer = ( f"## สรุปความเสี่ยง\n\n" f"- พบหลักฐานที่เกี่ยวข้อง **{len(rows)} รายการ**\n" f"- กลุ่มเด่น: **{cats}**\n" f"- รายการสำคัญสุด: **{meta['label']}** severity=`{top.get('severity')}` source=`{top.get('source')}`\n" f"- เวลา: `{top.get('ts')}`\n\n" f"## เหตุผล\n\n- {', '.join(detect_rule(top)['reasons'])}\n\n" f"## Next action\n\n- {meta['runbook']}" ) return {"answer": answer, "summary": {"categories": dict(by_cat), "severities": dict(by_sev)}} def build_context(hits: list[dict[str, Any]]) -> str: lines = [] for i, hit in enumerate(hits, 1): row = hit["row"] meta = CATEGORY_META.get(row.get("category"), CATEGORY_META["benign"]) lines.append("\n".join([ f"[{i}] score={hit['score']}", f"time={row.get('ts')} source={row.get('source')} severity={row.get('severity')}", f"category={row.get('category')} label={meta['label']} mitre={meta['mitre']}", f"src_ip={row.get('src_ip', '-')} dst_ip={row.get('dst_ip', '-')}", f"message={row.get('message')}", f"rule_reason={'; '.join(hit['rule']['reasons'])}", f"runbook={meta['runbook']}", ])) return "\n\n".join(lines) async def call_llm(question: str, context: str, model: str = "") -> str: settings = get_settings() if not settings.openrouter_api_key: raise RuntimeError("OPENROUTER_API_KEY is not set") payload = { "model": model or settings.openrouter_model, "messages": [ {"role": "system", "content": ( "You are a defensive security log analyst. Answer in Thai. " "Use only the provided retrieved logs and runbook context. " "Format as concise Markdown with sections: สรุป, หลักฐาน, ความเสี่ยง, Next action. " "Use short bullet points and inline code for IPs, users, MITRE IDs, and log fields. " "Do not provide offensive instructions.")}, {"role": "user", "content": f"คำถาม: {question}\n\nRetrieved context:\n{context}"}, ], "temperature": 0.2, "max_tokens": 700, } async with httpx.AsyncClient(timeout=45) as client: resp = await client.post( "https://openrouter.ai/api/v1/chat/completions", headers={"Authorization": f"Bearer {settings.openrouter_api_key}", "Content-Type": "application/json"}, json=payload, ) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"] async def ask(question: str, limit: int = 8, use_llm: bool = False, model: str = "") -> dict[str, Any]: hits = await retrieve(question, limit) local = answer_local(question, hits) answer, mode, llm_error = local["answer"], "local", None if use_llm and get_settings().llm_ready: try: answer = await call_llm(question, build_context(hits), model) mode = "openrouter" except Exception as exc: # keep the demo alive on rate-limit/errors llm_error = str(exc) return { "answer": answer, "mode": mode, "llm_error": llm_error, "retrieval": hits[0]["retrieval"] if hits else "none", "local_summary": local["summary"], "hits": hits, }