lorentz
9ca2ccf1c7
- OwnReportRow now selects campaign_id::text; match branch no-ops (created: false, no UPDATE) when the tiered match resolves to the report's own current campaign_id via a sibling row - Closes CAMP-02 gap / CR-01: /analyze can be re-run indefinitely without inflating campaigns.report_count
400 lines
16 KiB
TypeScript
400 lines
16 KiB
TypeScript
/**
|
|
* Campaign Grouping Service
|
|
*
|
|
* Shared grouping core called by all 3 trigger sites: the webhook path
|
|
* (`webhook-service.ts`'s `triggerPhishingDetection()`), the cron sweep
|
|
* (`phishing-sweep-service.ts`), and the on-demand
|
|
* `POST /api/phishing/tickets/{id}/analyze` route. One deterministic,
|
|
* testable function computes a tiered campaign key (CAMP-01) and
|
|
* finds-or-creates a `campaigns` row (CAMP-02) — no duplicated matching
|
|
* logic between callers, mirroring `phishing-detector.ts`'s shared-core
|
|
* architecture.
|
|
*
|
|
* D-07 limitation (load-bearing, stated explicitly): `parseAndStoreMessage`
|
|
* (the only writer of `messages`/`indicators` rows — Phase 16) is not wired
|
|
* into the automatic webhook/cron path this phase. That means the automatic
|
|
* path only ever has `reports`/`contacts` data available, so Tier 1
|
|
* (Message-ID) and Tier 2 (attachment-hash/URL-domain) can only ever match
|
|
* for a report that has already been through an explicit `/analyze` call at
|
|
* least once. Until then, automatic grouping effectively only reaches
|
|
* Tier 3 (sender + normalized subject + client + 24h window).
|
|
*/
|
|
|
|
import { postgresClient } from './postgres-client';
|
|
|
|
// =============================================================================
|
|
// Pure logic — tier-key helpers (mirrors matchesPhishingPatterns / computePhishingContentHash)
|
|
// =============================================================================
|
|
|
|
/**
|
|
* D-03: strip leading Re:/Fwd:/Fw: (repeated, case-insensitive), lowercase, trim.
|
|
*/
|
|
export function normalizeSubject(subject: string | null): string {
|
|
let s = (subject ?? '').trim();
|
|
const prefixRe = /^(re|fwd|fw):\s*/i;
|
|
while (prefixRe.test(s)) {
|
|
s = s.replace(prefixRe, '').trim();
|
|
}
|
|
return s.toLowerCase();
|
|
}
|
|
|
|
/**
|
|
* Pitfall 4: indicators.value for indicator_type='url' is a bare URL string,
|
|
* not a domain — extract at read time, guarded (malformed/relative URLs are
|
|
* possible in real-world phishing emails).
|
|
*/
|
|
export function extractUrlDomain(url: string): string | null {
|
|
try {
|
|
return new URL(url).hostname || null;
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// Orchestration — groupReportIntoCampaign (mirrors detectPhishingTicket)
|
|
// =============================================================================
|
|
|
|
export interface GroupReportResult {
|
|
campaignId: string;
|
|
groupMethod: 'message_id' | 'attachment_or_url' | 'sender_subject_client';
|
|
created: boolean;
|
|
}
|
|
|
|
interface OwnReportRow {
|
|
title: string | null;
|
|
requester_contact_id: number | null;
|
|
company_id: number | null;
|
|
created_at: string;
|
|
campaign_id: string | null;
|
|
}
|
|
|
|
interface OwnMessageRow {
|
|
id: string;
|
|
message_id: string | null;
|
|
}
|
|
|
|
interface IndicatorRow {
|
|
indicator_type: string;
|
|
value: string;
|
|
}
|
|
|
|
interface CandidateIndicatorRow extends IndicatorRow {
|
|
message_id: string;
|
|
}
|
|
|
|
interface Tier2CandidateRow {
|
|
report_id: string;
|
|
campaign_id: string;
|
|
title: string | null;
|
|
message_id: string;
|
|
}
|
|
|
|
/** Builds a `message_id:...` key — the strongest available signal (Tier 1). */
|
|
function computeTier1Key(messageId: string | null): string | null {
|
|
return messageId ? `message_id:${messageId}` : null;
|
|
}
|
|
|
|
/** Builds an `attachment_or_url:...` key from own indicators + subject + sender (Tier 2). */
|
|
function computeTier2Key(
|
|
attachmentHashes: string[],
|
|
urlDomains: string[],
|
|
normalizedSubject: string,
|
|
senderValue: string | null
|
|
): string | null {
|
|
if (!senderValue || !normalizedSubject) return null;
|
|
if (attachmentHashes.length === 0 && urlDomains.length === 0) return null;
|
|
const keyParts = [...attachmentHashes, ...urlDomains].sort();
|
|
return `attachment_or_url:${keyParts.join(',')}:${normalizedSubject}:${senderValue}`;
|
|
}
|
|
|
|
/** Builds a `sender_subject_client:...` key from sender + subject + company (Tier 3). */
|
|
function computeTier3Key(
|
|
requesterContactId: number | null,
|
|
normalizedSubject: string,
|
|
companyId: number | null
|
|
): string | null {
|
|
if (!requesterContactId || !normalizedSubject || !companyId) return null;
|
|
return `sender_subject_client:${requesterContactId}:${normalizedSubject}:${companyId}`;
|
|
}
|
|
|
|
/**
|
|
* Shared entry point called by the webhook path, the cron sweep, and the
|
|
* on-demand `/analyze` route. Looks up the report's own signal (message,
|
|
* indicators, requester/company), tries Tier 1 -> Tier 2 -> Tier 3 matching
|
|
* queries in order inside a single transaction (Pitfall 2 — `campaigns.
|
|
* campaign_key` has no UNIQUE constraint), and either links the report to an
|
|
* existing campaign (bumping report_count/last_seen_at, CAMP-02) or inserts a
|
|
* new campaigns row.
|
|
*
|
|
* D-08: `opts.skipIfAlreadyGrouped` short-circuits (returns null) when the
|
|
* report is already linked — used by the automatic webhook/cron path to
|
|
* avoid redundant work on every re-fire/sweep pass. The `/analyze` route
|
|
* omits this option so it always re-runs full tiered matching (it may
|
|
* upgrade a Tier-3-only report to Tier 1 after `parseAndStoreMessage` has
|
|
* just populated `messages`/`indicators` for the first time).
|
|
*
|
|
* D-04: never merges campaigns — attaches to the first/best tiered match
|
|
* found, or creates a new campaign. Multi-campaign-merge logic is explicitly
|
|
* out of scope this phase.
|
|
*/
|
|
export async function groupReportIntoCampaign(
|
|
reportId: string,
|
|
opts?: { skipIfAlreadyGrouped?: boolean }
|
|
): Promise<GroupReportResult | null> {
|
|
try {
|
|
if (opts?.skipIfAlreadyGrouped) {
|
|
const existing = await postgresClient.query<{ campaign_id: string | null }>(
|
|
`SELECT campaign_id::text AS campaign_id FROM reports WHERE id = $1`,
|
|
[reportId]
|
|
);
|
|
if (existing.rows[0]?.campaign_id) {
|
|
return null; // already grouped, short-circuit (D-08)
|
|
}
|
|
}
|
|
|
|
return await postgresClient.transaction(async (client) => {
|
|
const ownReportRes = await client.query<OwnReportRow>(
|
|
`SELECT title, requester_contact_id, company_id, created_at, campaign_id::text AS campaign_id
|
|
FROM reports
|
|
WHERE id = $1`,
|
|
[reportId]
|
|
);
|
|
const ownReport = ownReportRes.rows[0];
|
|
if (!ownReport) {
|
|
throw new Error(`groupReportIntoCampaign: report ${reportId} not found`);
|
|
}
|
|
|
|
const normalizedSubject = normalizeSubject(ownReport.title);
|
|
|
|
// Own messages row, if `parseAndStoreMessage` has already run for this
|
|
// report (D-07 — not the case for the automatic webhook/cron path
|
|
// until an explicit /analyze call has happened at least once).
|
|
const ownMessageRes = await client.query<OwnMessageRow>(
|
|
`SELECT id::text AS id, message_id
|
|
FROM messages
|
|
WHERE report_id = $1
|
|
LIMIT 1`,
|
|
[reportId]
|
|
);
|
|
const ownMessage = ownMessageRes.rows[0] ?? null;
|
|
|
|
let matchCampaignId: string | null = null;
|
|
let matchGroupMethod: GroupReportResult['groupMethod'] | null = null;
|
|
|
|
// ---------------------------------------------------------------------
|
|
// Tier 1: Message-ID. Self-exclusion (`r.id != $2`) is REQUIRED — without
|
|
// it this query would trivially match the report's own messages row
|
|
// against itself, double-incrementing its own already-linked campaign
|
|
// on every re-run (plan-checker finding).
|
|
// ---------------------------------------------------------------------
|
|
if (ownMessage?.message_id) {
|
|
const tier1 = await client.query<{ campaign_id: string }>(
|
|
`SELECT r.campaign_id::text AS campaign_id
|
|
FROM messages m
|
|
JOIN reports r ON r.id = m.report_id
|
|
WHERE m.message_id = $1
|
|
AND r.campaign_id IS NOT NULL
|
|
AND r.id != $2
|
|
ORDER BY r.created_at ASC
|
|
LIMIT 1`,
|
|
[ownMessage.message_id, reportId]
|
|
);
|
|
if (tier1.rows[0]?.campaign_id) {
|
|
matchCampaignId = tier1.rows[0].campaign_id;
|
|
matchGroupMethod = 'message_id';
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------
|
|
// Tier 2: attachment-hash / URL-domain + normalized subject + sender,
|
|
// within a 24h window (D-02). Only reachable if the own report has its
|
|
// own indicators (i.e. parseAndStoreMessage already ran for it).
|
|
// Self-exclusion (`r.id != $1`) required for the same reason as Tier 1.
|
|
// ---------------------------------------------------------------------
|
|
let ownAttachmentHashes: string[] = [];
|
|
let ownUrlDomains: string[] = [];
|
|
let ownSenderValue: string | null = null;
|
|
|
|
if (!matchCampaignId && ownMessage) {
|
|
const ownIndicatorsRes = await client.query<IndicatorRow>(
|
|
`SELECT indicator_type, value
|
|
FROM indicators
|
|
WHERE message_id = $1`,
|
|
[ownMessage.id]
|
|
);
|
|
ownAttachmentHashes = ownIndicatorsRes.rows
|
|
.filter((i) => i.indicator_type === 'attachment_hash')
|
|
.map((i) => i.value);
|
|
ownUrlDomains = ownIndicatorsRes.rows
|
|
.filter((i) => i.indicator_type === 'url')
|
|
.map((i) => extractUrlDomain(i.value))
|
|
.filter((d): d is string => d !== null);
|
|
ownSenderValue =
|
|
ownIndicatorsRes.rows.find((i) => i.indicator_type === 'sender')?.value ?? null;
|
|
|
|
if (
|
|
(ownAttachmentHashes.length > 0 || ownUrlDomains.length > 0) &&
|
|
ownSenderValue &&
|
|
normalizedSubject
|
|
) {
|
|
const tier2Candidates = await client.query<Tier2CandidateRow>(
|
|
`SELECT r.id::text AS report_id, r.campaign_id::text AS campaign_id, r.title,
|
|
m.id::text AS message_id
|
|
FROM messages m
|
|
JOIN reports r ON r.id = m.report_id
|
|
WHERE r.campaign_id IS NOT NULL
|
|
AND r.id != $1
|
|
AND r.created_at BETWEEN $2::timestamptz - INTERVAL '24 hours'
|
|
AND $2::timestamptz + INTERVAL '24 hours'`,
|
|
[reportId, ownReport.created_at]
|
|
);
|
|
|
|
if (tier2Candidates.rows.length > 0) {
|
|
const candidateMessageIds = tier2Candidates.rows.map((c) => c.message_id);
|
|
const candidateIndicatorsRes = await client.query<CandidateIndicatorRow>(
|
|
`SELECT message_id::text AS message_id, indicator_type, value
|
|
FROM indicators
|
|
WHERE message_id = ANY($1::uuid[])`,
|
|
[candidateMessageIds]
|
|
);
|
|
const indicatorsByMessage = new Map<string, CandidateIndicatorRow[]>();
|
|
for (const ind of candidateIndicatorsRes.rows) {
|
|
const arr = indicatorsByMessage.get(ind.message_id) ?? [];
|
|
arr.push(ind);
|
|
indicatorsByMessage.set(ind.message_id, arr);
|
|
}
|
|
|
|
for (const candidate of tier2Candidates.rows) {
|
|
if (normalizeSubject(candidate.title) !== normalizedSubject) continue;
|
|
const candidateIndicators = indicatorsByMessage.get(candidate.message_id) ?? [];
|
|
const candidateSender = candidateIndicators.find(
|
|
(i) => i.indicator_type === 'sender'
|
|
)?.value;
|
|
if (candidateSender !== ownSenderValue) continue;
|
|
const candidateHashes = candidateIndicators
|
|
.filter((i) => i.indicator_type === 'attachment_hash')
|
|
.map((i) => i.value);
|
|
const candidateDomains = candidateIndicators
|
|
.filter((i) => i.indicator_type === 'url')
|
|
.map((i) => extractUrlDomain(i.value))
|
|
.filter((d): d is string => d !== null);
|
|
const hashMatch = ownAttachmentHashes.some((h) => candidateHashes.includes(h));
|
|
const domainMatch = ownUrlDomains.some((d) => candidateDomains.includes(d));
|
|
if (hashMatch || domainMatch) {
|
|
matchCampaignId = candidate.campaign_id;
|
|
matchGroupMethod = 'attachment_or_url';
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------
|
|
// Tier 3: sender + normalizeSubject(title) + client + 24h window
|
|
// (D-02). Joins reports.requester_contact_id -> contacts (Pitfall 5 —
|
|
// NOT `contact_id`). Self-exclusion (`r.id != $3`) required for the
|
|
// same reason as Tiers 1-2.
|
|
// ---------------------------------------------------------------------
|
|
if (!matchCampaignId && ownReport.requester_contact_id && ownReport.company_id && normalizedSubject) {
|
|
const tier3 = await client.query<{ campaign_id: string; title: string | null }>(
|
|
`SELECT r.campaign_id::text AS campaign_id, r.title
|
|
FROM reports r
|
|
JOIN contacts c ON c.id = r.requester_contact_id
|
|
WHERE r.requester_contact_id = $1
|
|
AND r.company_id = $2
|
|
AND r.campaign_id IS NOT NULL
|
|
AND r.id != $3
|
|
AND r.created_at BETWEEN $4::timestamptz - INTERVAL '24 hours'
|
|
AND $4::timestamptz + INTERVAL '24 hours'
|
|
ORDER BY r.created_at ASC`,
|
|
[ownReport.requester_contact_id, ownReport.company_id, reportId, ownReport.created_at]
|
|
);
|
|
const match = tier3.rows.find((r) => normalizeSubject(r.title) === normalizedSubject);
|
|
if (match) {
|
|
matchCampaignId = match.campaign_id;
|
|
matchGroupMethod = 'sender_subject_client';
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------
|
|
// Find-or-create against `campaigns` (CAMP-02).
|
|
// ---------------------------------------------------------------------
|
|
if (matchCampaignId && matchGroupMethod) {
|
|
// CR-01/CAMP-02 no-op guard: tier queries only exclude the report's
|
|
// OWN row (`r.id != $x`), not its siblings already in the same
|
|
// campaign. Since `/analyze` re-runs grouping unconditionally
|
|
// (D-08), a report already linked to campaign X can re-match X via
|
|
// a sibling row on every repeat call, inflating `report_count`
|
|
// without bound. If the match resolves to the report's own current
|
|
// campaign, it's already correctly linked — skip both UPDATEs.
|
|
if (matchCampaignId === ownReport.campaign_id) {
|
|
return { campaignId: matchCampaignId, groupMethod: matchGroupMethod, created: false };
|
|
}
|
|
await client.query(
|
|
`UPDATE campaigns
|
|
SET report_count = report_count + 1, last_seen_at = NOW(), updated_at = NOW()
|
|
WHERE id = $1`,
|
|
[matchCampaignId]
|
|
);
|
|
await client.query(
|
|
`UPDATE reports SET campaign_id = $1, updated_at = NOW() WHERE id = $2`,
|
|
[matchCampaignId, reportId]
|
|
);
|
|
return { campaignId: matchCampaignId, groupMethod: matchGroupMethod, created: false };
|
|
}
|
|
|
|
// No match on any tier — create a new campaign. Prefer the strongest
|
|
// available key (Tier 1 > Tier 2 > Tier 3) so future duplicates of THIS
|
|
// report can match it; do NOT set/transition campaigns.status (leave
|
|
// migration default 'open' — Phase 19/20's concern).
|
|
const tier1Key = computeTier1Key(ownMessage?.message_id ?? null);
|
|
const tier2Key = computeTier2Key(
|
|
ownAttachmentHashes,
|
|
ownUrlDomains,
|
|
normalizedSubject,
|
|
ownSenderValue
|
|
);
|
|
const tier3Key = computeTier3Key(
|
|
ownReport.requester_contact_id,
|
|
normalizedSubject,
|
|
ownReport.company_id
|
|
);
|
|
|
|
let newCampaignKey: string;
|
|
let newGroupMethod: GroupReportResult['groupMethod'];
|
|
if (tier1Key) {
|
|
newCampaignKey = tier1Key;
|
|
newGroupMethod = 'message_id';
|
|
} else if (tier2Key) {
|
|
newCampaignKey = tier2Key;
|
|
newGroupMethod = 'attachment_or_url';
|
|
} else if (tier3Key) {
|
|
newCampaignKey = tier3Key;
|
|
newGroupMethod = 'sender_subject_client';
|
|
} else {
|
|
newCampaignKey = `report:${reportId}`;
|
|
newGroupMethod = 'sender_subject_client';
|
|
}
|
|
|
|
const newCampaign = await client.query<{ id: string }>(
|
|
`INSERT INTO campaigns (campaign_key, group_method, first_seen_at, last_seen_at, report_count)
|
|
VALUES ($1, $2, NOW(), NOW(), 1)
|
|
RETURNING id::text AS id`,
|
|
[newCampaignKey, newGroupMethod]
|
|
);
|
|
const newCampaignId = newCampaign.rows[0].id;
|
|
|
|
await client.query(
|
|
`UPDATE reports SET campaign_id = $1, updated_at = NOW() WHERE id = $2`,
|
|
[newCampaignId, reportId]
|
|
);
|
|
|
|
return { campaignId: newCampaignId, groupMethod: newGroupMethod, created: true };
|
|
});
|
|
} catch (error) {
|
|
console.error('[CAMPAIGN-GROUPING] Failed to group report into campaign', reportId, error);
|
|
throw error;
|
|
}
|
|
}
|