Skip to content

Commit 4d8d09c

Browse files
kgrubbkgrubb
committed
fix(detectors): correct srcIpsByKey fold in egress anomaly
## Fixed - foldCurrent no longer relied on Map#set chaining to obtain the Set for each dedupe key. ## Changed - enrichEgressEvidence returns evidence built with object spreads only. - Insight engine marks posted findings for dedupe with forEach instead of for...of.
1 parent aa1c9e7 commit 4d8d09c

2 files changed

Lines changed: 122 additions & 78 deletions

File tree

‎src/detectors/egressAnomaly.ts‎

Lines changed: 112 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,102 @@ import type { EgressAggRow } from '../opensearch/queries/index.js';
33
import { egressDedupeKey, ipv6GlobalUnicastPrefix64 } from '../util/egressDedupeKey.js';
44
import type { Finding } from './types.js';
55

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+
6102
export function detectEgressAnomalies(opts: {
7103
window: { from: string; to: string };
8104
current: EgressAggRow[];
@@ -12,81 +108,24 @@ export function detectEgressAnomalies(opts: {
12108
currentMinutes: number;
13109
}): Finding[] {
14110
const expectedScale = opts.currentMinutes / opts.baselineMinutes;
111+
const baselineByKey = foldBaseline(opts.baseline);
112+
const { bytesByKey, maxRowByKey, srcIpsByKey } = foldCurrent(opts.current);
15113

16-
const baselineByKey = new Map<string, number>();
17-
for (const row of opts.baseline) {
18-
if (!row.srcIp) continue;
19-
const k = egressDedupeKey(row.srcIp);
20-
baselineByKey.set(k, (baselineByKey.get(k) ?? 0) + row.bytes);
21-
}
22-
23-
const bytesByKey = new Map<string, number>();
24-
const maxRowByKey = new Map<string, EgressAggRow>();
25-
const srcIpsByKey = new Map<string, Set<string>>();
26-
for (const row of opts.current) {
27-
if (!row.srcIp) continue;
28-
const k = egressDedupeKey(row.srcIp);
29-
bytesByKey.set(k, (bytesByKey.get(k) ?? 0) + row.bytes);
30-
const prev = maxRowByKey.get(k);
31-
if (!prev || row.bytes > prev.bytes) maxRowByKey.set(k, row);
32-
const set = srcIpsByKey.get(k) ?? new Set<string>();
33-
set.add(row.srcIp);
34-
srcIpsByKey.set(k, set);
35-
}
36-
37-
const findings: Finding[] = [];
38-
for (const [key, totalBytes] of bytesByKey) {
114+
return [...bytesByKey.entries()].flatMap(([key, totalBytes]) => {
39115
const sampleRow = maxRowByKey.get(key);
40-
if (!sampleRow?.srcIp) continue;
41-
42-
const baselineBytes = baselineByKey.get(key) ?? 0;
43-
const expectedBytes = baselineBytes * expectedScale;
44-
const threshold = Math.max(opts.thresholds.egressMinBytes, expectedBytes * opts.thresholds.egressMultiplier);
45-
if (totalBytes <= threshold) continue;
46-
47-
const ratio = expectedBytes > 0 ? totalBytes / expectedBytes : Number.POSITIVE_INFINITY;
48-
const severity: Finding['severity'] =
49-
totalBytes > threshold * 5 ? 'high' : totalBytes > threshold * 2 ? 'medium' : 'low';
50-
51-
const id = `egress:${key}`;
52-
const p64 = ipv6GlobalUnicastPrefix64(sampleRow.srcIp);
53-
const title = p64
54-
? `Unusual egress from IPv6 /64 ${p64}`
55-
: `Unusual egress from ${sampleRow.srcIp}`;
56-
const summary =
57-
expectedBytes > 0
58-
? p64
59-
? `IPv6 /64 ${p64}: ${totalBytes.toLocaleString()} bytes vs expected ~${Math.round(expectedBytes).toLocaleString()} (${ratio.toFixed(1)}x); top host ${sampleRow.srcIp}.`
60-
: `${sampleRow.srcIp} transferred ${Math.round(totalBytes).toLocaleString()} bytes vs expected ~${Math.round(expectedBytes).toLocaleString()} bytes (${ratio.toFixed(1)}x).`
61-
: p64
62-
? `IPv6 /64 ${p64}: ${totalBytes.toLocaleString()} bytes (no baseline for comparison); top host ${sampleRow.srcIp}.`
63-
: `${sampleRow.srcIp} transferred ${Math.round(totalBytes).toLocaleString()} bytes (no baseline for comparison).`;
64-
65-
const contributingSrcIps = [...(srcIpsByKey.get(key) ?? new Set())].sort();
66-
67-
findings.push({
68-
id,
69-
kind: 'egress_anomaly' as const,
70-
severity,
71-
title,
72-
summary,
73-
evidence: {
74-
egressKey: key,
75-
srcIp: sampleRow.srcIp,
76-
contributingSrcIps,
77-
bytes: totalBytes,
78-
expectedBytes,
79-
baselineBytes,
80-
baselineMinutes: opts.baselineMinutes,
81-
currentMinutes: opts.currentMinutes,
82-
thresholds: {
83-
egressMultiplier: opts.thresholds.egressMultiplier,
84-
egressMinBytes: opts.thresholds.egressMinBytes,
85-
},
86-
},
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,
87127
window: opts.window,
88128
});
89-
}
90-
91-
return findings;
129+
return f ? [f] : [];
130+
});
92131
}

‎src/insights/enrichEgressEvidence.ts‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -95,11 +95,16 @@ export async function enrichEgressFinding(opts: {
9595
}))
9696
: undefined;
9797

98-
const ev: Record<string, unknown> = { ...finding.evidence, topDestinations, topDstPorts };
99-
if (topClientNamespaces?.length) ev['topClientNamespaces'] = topClientNamespaces;
100-
if (topClientPods?.length) ev['topClientPods'] = topClientPods;
101-
102-
return { ...finding, evidence: ev };
98+
return {
99+
...finding,
100+
evidence: {
101+
...finding.evidence,
102+
topDestinations,
103+
topDstPorts,
104+
...(topClientNamespaces?.length ? { topClientNamespaces } : {}),
105+
...(topClientPods?.length ? { topClientPods } : {}),
106+
},
107+
};
103108
}
104109

105110
export async function enrichInsightsEgressBatch(opts: {

0 commit comments

Comments
 (0)