Skip to content

Commit 4eb9e64

Browse files
kgrubbkgrubb
authored
feat(llm,insights): OpenWebUI probe, clearer errors, high-signal insights (#19)
## Added - Startup LLM probe for bare LLM_BASE_URL: try /api/v1/models then /v1/models, pin chat/completions to the first working API root; skip when base already has /v1 or /api/v1. - Egress enrichment for proactive posts: top destinations, dst ports, optional client namespace/pod when field caps resolve them. - IPv6 global-unicast /64 grouping for egress dedupe; contributingSrcIps on affected findings. - Tests for egress dedupe key, enrichment, poll selection, and summarizeFindings JSON contract. ## Changed - LLM HTTP errors log configured model and resolved chat URL. - Helm: no placeholder LLM_MODEL default; ConfigMap omits LLM_MODEL when unset. - Insight poller: no rare_destination; medium/high only; max 3 findings per post; dedupe.mark only after successful post. - summarizeFindings strict JSON {post,text}; parse errors or post:false skip chat (fail-closed, no fallback post for insights). - Slack summarizer prompt uses enrichment when present; README documents probe and LLM_MODEL examples. ## Fixed - Bare OpenWebUI bases that only answer on /api/v1 no longer stay on /v1 after a 400 from the wrong route. - Egress detector fold accumulates src IPs per dedupe key correctly. - Scheduled insight spam from low severity, poller rare-destination noise, and duplicate IPv6 host lines.
1 parent c45b9a3 commit 4eb9e64

21 files changed

Lines changed: 801 additions & 250 deletions

‎README.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -132,9 +132,9 @@ Kaytoo is configured via environment variables. For a minimal example, see `.env
132132

133133
| Variable | Required | Default | Notes |
134134
| --- | --- | --- | --- |
135-
| `LLM_BASE_URL` | yes | - | Example: `http://openwebui:3000` or `http://ollama-proxy:11434`. |
135+
| `LLM_BASE_URL` | yes | - | Example: `http://openwebui:3000` or `http://ollama-proxy:11434`. If the path is bare (no `/v1` or `/api/v1`), Kaytoo probes `/api/v1/models` then `/v1/models` once at startup and pins the responder. To skip the probe, include the version path (e.g. `http://openwebui:3000/api/v1`). |
136136
| `LLM_API_KEY` | yes | - | Some self-hosted backends accept an empty string. |
137-
| `LLM_MODEL` | no | `gpt-5.4-codex` | - |
137+
| `LLM_MODEL` | no | `gpt-5.4-codex` | Must match a name your backend exposes (`/api/v1/models` or `/v1/models`). E.g. `gpt-4o`, `llama3.1`, `unsloth/Qwen3.6-35B-A3B-GGUF:UD-Q4_K_XL`. A wrong name typically produces a LiteLLM `'NoneType' object has no attribute 'startswith'` 400. |
138138

139139
**Conversation agent (chat adapters: Slack / Matrix / Mattermost)**
140140

‎helm/kaytoo/templates/configmap.yaml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,9 @@ data:
2020
{{- end }}
2121

2222
LLM_BASE_URL: {{ .Values.config.llm.baseUrl | quote }}
23+
{{- if .Values.config.llm.model }}
2324
LLM_MODEL: {{ .Values.config.llm.model | quote }}
25+
{{- end }}
2426

2527
LOG_LEVEL: {{ .Values.logging.level | quote }}
2628

‎helm/kaytoo/values.yaml‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,10 @@ config:
5050
mcpUrl: ""
5151
llm:
5252
baseUrl: ""
53-
model: "gpt-5.3-codex"
53+
# Required. Use the exact model name your OpenAI-compatible backend exposes,
54+
# e.g. "gpt-4o" for OpenAI, "llama3.1" for Ollama,
55+
# or "unsloth/Qwen3.6-35B-A3B-GGUF:UD-Q4_K_XL" for OpenWebUI/LiteLLM.
56+
model: ""
5457

5558
secrets:
5659
existingSecret: ""

‎src/config.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@ const configSchema = z
8686
llm: z.object({
8787
baseUrl: z.string().url('LLM_BASE_URL must be a valid URL'),
8888
apiKey: z.string().min(1, 'LLM_API_KEY is required'),
89-
model: z.string().min(1).default('gpt-5.3-codex'),
89+
model: z.string().min(1).default('gpt-5.4-codex'),
9090
}),
9191
behavior: z.object({
9292
pollIntervalSeconds: z.string().default('300').pipe(intFromString),
@@ -273,7 +273,7 @@ export function getConfig(env: NodeJS.ProcessEnv = process.env, opts?: GetConfig
273273
llm: {
274274
baseUrl: env.LLM_BASE_URL,
275275
apiKey: env.LLM_API_KEY,
276-
model: env.LLM_MODEL ?? 'gpt-5.4-codex',
276+
model: env.LLM_MODEL,
277277
},
278278
behavior: {},
279279
thresholds: {},

‎src/detectors/egressAnomaly.ts‎

Lines changed: 115 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,104 @@
11
import type { KaytooConfig } from '../config.js';
22
import type { EgressAggRow } from '../opensearch/queries/index.js';
3+
import { egressDedupeKey, ipv6GlobalUnicastPrefix64 } from '../util/egressDedupeKey.js';
34
import type { Finding } from './types.js';
45

6+
type CurrentAcc = {
7+
bytesByKey: Map<string, number>;
8+
maxRowByKey: Map<string, EgressAggRow>;
9+
srcIpsByKey: Map<string, Set<string>>;
10+
};
11+
12+
function foldBaseline(rows: EgressAggRow[]): Map<string, number> {
13+
return rows.reduce((m, row) => {
14+
if (!row.srcIp) return m;
15+
const k = egressDedupeKey(row.srcIp);
16+
m.set(k, (m.get(k) ?? 0) + row.bytes);
17+
return m;
18+
}, new Map<string, number>());
19+
}
20+
21+
function foldCurrent(rows: EgressAggRow[]): CurrentAcc {
22+
return rows.reduce(
23+
(acc, row) => {
24+
if (!row.srcIp) return acc;
25+
const k = egressDedupeKey(row.srcIp);
26+
acc.bytesByKey.set(k, (acc.bytesByKey.get(k) ?? 0) + row.bytes);
27+
const prev = acc.maxRowByKey.get(k);
28+
if (!prev || row.bytes > prev.bytes) acc.maxRowByKey.set(k, row);
29+
const ips = acc.srcIpsByKey.get(k);
30+
if (ips) ips.add(row.srcIp);
31+
else acc.srcIpsByKey.set(k, new Set([row.srcIp]));
32+
return acc;
33+
},
34+
{
35+
bytesByKey: new Map<string, number>(),
36+
maxRowByKey: new Map<string, EgressAggRow>(),
37+
srcIpsByKey: new Map<string, Set<string>>(),
38+
} satisfies CurrentAcc,
39+
);
40+
}
41+
42+
function egressFindingForKey(opts: {
43+
key: string;
44+
totalBytes: number;
45+
sampleRow: EgressAggRow;
46+
srcIpsByKey: Map<string, Set<string>>;
47+
baselineByKey: Map<string, number>;
48+
expectedScale: number;
49+
thresholds: KaytooConfig['thresholds'];
50+
baselineMinutes: number;
51+
currentMinutes: number;
52+
window: { from: string; to: string };
53+
}): Finding | null {
54+
const { key, totalBytes, sampleRow, srcIpsByKey, baselineByKey, expectedScale, thresholds, baselineMinutes, currentMinutes, window } = opts;
55+
const baselineBytes = baselineByKey.get(key) ?? 0;
56+
const expectedBytes = baselineBytes * expectedScale;
57+
const threshold = Math.max(thresholds.egressMinBytes, expectedBytes * thresholds.egressMultiplier);
58+
if (totalBytes <= threshold) return null;
59+
60+
const ratio = expectedBytes > 0 ? totalBytes / expectedBytes : Number.POSITIVE_INFINITY;
61+
const severity: Finding['severity'] =
62+
totalBytes > threshold * 5 ? 'high' : totalBytes > threshold * 2 ? 'medium' : 'low';
63+
64+
const id = `egress:${key}`;
65+
const p64 = ipv6GlobalUnicastPrefix64(sampleRow.srcIp);
66+
const title = p64 ? `Unusual egress from IPv6 /64 ${p64}` : `Unusual egress from ${sampleRow.srcIp}`;
67+
const summary =
68+
expectedBytes > 0
69+
? p64
70+
? `IPv6 /64 ${p64}: ${totalBytes.toLocaleString()} bytes vs expected ~${Math.round(expectedBytes).toLocaleString()} (${ratio.toFixed(1)}x); top host ${sampleRow.srcIp}.`
71+
: `${sampleRow.srcIp} transferred ${Math.round(totalBytes).toLocaleString()} bytes vs expected ~${Math.round(expectedBytes).toLocaleString()} bytes (${ratio.toFixed(1)}x).`
72+
: p64
73+
? `IPv6 /64 ${p64}: ${totalBytes.toLocaleString()} bytes (no baseline for comparison); top host ${sampleRow.srcIp}.`
74+
: `${sampleRow.srcIp} transferred ${Math.round(totalBytes).toLocaleString()} bytes (no baseline for comparison).`;
75+
76+
const contributingSrcIps = [...(srcIpsByKey.get(key) ?? new Set())].sort();
77+
78+
return {
79+
id,
80+
kind: 'egress_anomaly' as const,
81+
severity,
82+
title,
83+
summary,
84+
evidence: {
85+
egressKey: key,
86+
srcIp: sampleRow.srcIp,
87+
contributingSrcIps,
88+
bytes: totalBytes,
89+
expectedBytes,
90+
baselineBytes,
91+
baselineMinutes,
92+
currentMinutes,
93+
thresholds: {
94+
egressMultiplier: thresholds.egressMultiplier,
95+
egressMinBytes: thresholds.egressMinBytes,
96+
},
97+
},
98+
window,
99+
};
100+
}
101+
5102
export function detectEgressAnomalies(opts: {
6103
window: { from: string; to: string };
7104
current: EgressAggRow[];
@@ -10,45 +107,25 @@ export function detectEgressAnomalies(opts: {
10107
baselineMinutes: number;
11108
currentMinutes: number;
12109
}): Finding[] {
13-
const baselineByIp = new Map(opts.baseline.map((row) => [row.srcIp, row.bytes] as const));
14110
const expectedScale = opts.currentMinutes / opts.baselineMinutes;
111+
const baselineByKey = foldBaseline(opts.baseline);
112+
const { bytesByKey, maxRowByKey, srcIpsByKey } = foldCurrent(opts.current);
15113

16-
return opts.current.flatMap((row) => {
17-
if (!row.srcIp) return [];
18-
const baselineBytes = baselineByIp.get(row.srcIp) ?? 0;
19-
const expectedBytes = baselineBytes * expectedScale;
20-
const threshold = Math.max(opts.thresholds.egressMinBytes, expectedBytes * opts.thresholds.egressMultiplier);
21-
if (row.bytes <= threshold) return [];
22-
23-
const ratio = expectedBytes > 0 ? row.bytes / expectedBytes : Number.POSITIVE_INFINITY;
24-
const severity: Finding['severity'] =
25-
row.bytes > threshold * 5 ? 'high' : row.bytes > threshold * 2 ? 'medium' : 'low';
26-
27-
return [
28-
{
29-
// Stable across polls so DedupeStore suppresses repeat LLM posts for the same source IP.
30-
id: `egress:${row.srcIp}`,
31-
kind: 'egress_anomaly' as const,
32-
severity,
33-
title: `Unusual egress from ${row.srcIp}`,
34-
summary:
35-
expectedBytes > 0
36-
? `${row.srcIp} transferred ${Math.round(row.bytes).toLocaleString()} bytes vs expected ~${Math.round(expectedBytes).toLocaleString()} bytes (${ratio.toFixed(1)}x).`
37-
: `${row.srcIp} transferred ${Math.round(row.bytes).toLocaleString()} bytes (no baseline for comparison).`,
38-
evidence: {
39-
srcIp: row.srcIp,
40-
bytes: row.bytes,
41-
expectedBytes,
42-
baselineBytes,
43-
baselineMinutes: opts.baselineMinutes,
44-
currentMinutes: opts.currentMinutes,
45-
thresholds: {
46-
egressMultiplier: opts.thresholds.egressMultiplier,
47-
egressMinBytes: opts.thresholds.egressMinBytes,
48-
},
49-
},
50-
window: opts.window,
51-
},
52-
];
114+
return [...bytesByKey.entries()].flatMap(([key, totalBytes]) => {
115+
const sampleRow = maxRowByKey.get(key);
116+
if (!sampleRow?.srcIp) return [];
117+
const f = egressFindingForKey({
118+
key,
119+
totalBytes,
120+
sampleRow,
121+
srcIpsByKey,
122+
baselineByKey,
123+
expectedScale,
124+
thresholds: opts.thresholds,
125+
baselineMinutes: opts.baselineMinutes,
126+
currentMinutes: opts.currentMinutes,
127+
window: opts.window,
128+
});
129+
return f ? [f] : [];
53130
});
54131
}

‎src/insights/engine.ts‎

Lines changed: 32 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -5,17 +5,11 @@ import { getLogger, logErr } from '../logging/logger.js';
55
import { runWithLogContextAsync } from '../logging/context.js';
66
import { createSearchClient } from '../search/client.js';
77
import { waitForOpenSearchFieldMapping } from '../opensearch/waitForFieldMapping.js';
8-
import {
9-
queryPortscanCandidates,
10-
queryRareDestinationsSignificantTerms,
11-
queryTopEgressBySource,
12-
} from '../opensearch/queries/index.js';
8+
import { queryPortscanCandidates, queryTopEgressBySource } from '../opensearch/queries/index.js';
139
import { detectEgressAnomalies } from '../detectors/egressAnomaly.js';
1410
import { detectPortScans } from '../detectors/portScan.js';
15-
import { detectRareDestinations } from '../detectors/rareDest.js';
1611
import type { Finding } from '../detectors/types.js';
1712
import { createOpenAiCompatClient } from '../llm/openaiCompat.js';
18-
import { formatFindingsFallback } from '../notify/format.js';
1913
import type { InsightSink } from '../notify/insightSink.js';
2014
import { DedupeStore } from '../state/dedupe.js';
2115
import { thrownMessage } from '../util/guards.js';
@@ -25,7 +19,8 @@ import {
2519
fetchOpenSearchAlertingFindings,
2620
type DetectionFetchResult,
2721
} from './opensearchDetections.js';
28-
import { findingSeverityRank, shouldSkipHeuristicPoll } from './pollUtils.js';
22+
import { selectNovelInsightPostBatch, shouldSkipHeuristicPoll } from './pollUtils.js';
23+
import { enrichInsightsEgressBatch } from './enrichEgressEvidence.js';
2924

3025
function detectionFetchFailure(e: unknown): DetectionFetchResult {
3126
return { ok: false, findings: [], warning: thrownMessage(e) };
@@ -113,14 +108,12 @@ export async function startInsightEngine(opts: { config: KaytooConfig; insightSi
113108
const currentMinutes = 15;
114109
const baselineMinutes = 24 * 60;
115110
const portscanMinutes = 5;
116-
const backgroundMinutes = 7 * 24 * 60;
117111

118112
const currentWindow = windowRelative({ to: now, minutesBack: currentMinutes });
119113
const baselineWindow = windowRelative({ to: now, minutesBack: baselineMinutes });
120114
const portscanWindow = windowRelative({ to: now, minutesBack: portscanMinutes });
121-
const backgroundWindow = windowRelative({ to: now, minutesBack: backgroundMinutes });
122115

123-
const [currentEgress, baselineEgress, portscanRows, rareDestRows] = await Promise.all([
116+
const [currentEgress, baselineEgress, portscanRows] = await Promise.all([
124117
queryTopEgressBySource({
125118
client,
126119
index: config.search.indexPattern,
@@ -142,14 +135,6 @@ export async function startInsightEngine(opts: { config: KaytooConfig; insightSi
142135
window: portscanWindow,
143136
size: 50,
144137
}),
145-
queryRareDestinationsSignificantTerms({
146-
client,
147-
index: config.search.indexPattern,
148-
fields,
149-
window: currentWindow,
150-
backgroundWindow,
151-
size: 10,
152-
}).catch(() => []),
153138
]);
154139

155140
const findings: Finding[] = [
@@ -166,19 +151,15 @@ export async function startInsightEngine(opts: { config: KaytooConfig; insightSi
166151
rows: portscanRows,
167152
thresholds: config.thresholds,
168153
}),
169-
...detectRareDestinations({
170-
window: currentWindow,
171-
rows: rareDestRows,
172-
}),
173-
].sort((a, b) => findingSeverityRank(b.severity) - findingSeverityRank(a.severity));
154+
];
174155

175156
const novel = findings.filter((f) => !dedupe.has(f.id));
176157
if (novel.length === 0) {
177158
log.debug('no new findings this poll');
178159
return;
179160
}
180161

181-
await postFindings(novel);
162+
await postFindings(findings);
182163
} catch (e) {
183164
log.error({ ...logErr(e) }, 'poll failed');
184165
} finally {
@@ -189,26 +170,40 @@ export async function startInsightEngine(opts: { config: KaytooConfig; insightSi
189170
}
190171

191172
async function postFindings(findings: Finding[]): Promise<void> {
192-
const novel = findings.filter((f) => !dedupe.has(f.id));
193-
if (novel.length === 0) return;
173+
const toPost = selectNovelInsightPostBatch(findings, dedupe);
174+
if (toPost.length === 0) return;
175+
176+
const toSummarize = await enrichInsightsEgressBatch({
177+
client,
178+
index: config.search.indexPattern,
179+
fields,
180+
findings: toPost,
181+
log,
182+
});
194183

195-
const text = await llm
196-
.summarizeFindings({ channelStyle: 'slack', findings: novel })
197-
.then((r) => r.text)
198-
.catch((e) => {
199-
log.warn({ ...logErr(e), findingCount: novel.length }, 'LLM summarization failed, using fallback');
200-
return formatFindingsFallback(novel);
201-
});
184+
const summary = await llm.summarizeFindings({ channelStyle: 'slack', findings: toSummarize }).catch((e) => {
185+
log.warn({ ...logErr(e), findingCount: toPost.length }, 'LLM summarization failed; skipping proactive post');
186+
return null;
187+
});
188+
189+
if (summary === null) return;
190+
191+
if (!summary.post || !summary.text.trim()) {
192+
log.debug({ findingCount: toPost.length, post: summary.post }, 'LLM declined proactive insight post');
193+
return;
194+
}
195+
196+
const text = summary.text.trim();
202197

203198
try {
204199
await opts.insightSink.postInsight(text);
205200
} catch {
206201
// Notifier already logged the cause; record outcome only and skip dedupe so the next poll retries.
207-
log.warn({ findingCount: novel.length, output: config.output }, 'post findings failed');
202+
log.warn({ findingCount: toPost.length, output: config.output }, 'post findings failed');
208203
return;
209204
}
210-
for (const f of novel) dedupe.mark(f.id);
211-
log.info({ findingCount: novel.length, output: config.output }, 'posted findings');
205+
toPost.forEach((f) => dedupe.mark(f.id));
206+
log.info({ findingCount: toPost.length, output: config.output }, 'posted findings');
212207
}
213208

214209
await pollOnce();
@@ -228,4 +223,3 @@ function rateLimitedWarn(log: Logger, map: Map<string, number>, key: string, msg
228223
log.warn({ degradedKey: key, degradedMsg: msg }, 'insights degraded');
229224
map.set(key, now + 10 * 60_000);
230225
}
231-

0 commit comments

Comments
 (0)