diff --git a/ROADMAP.md b/ROADMAP.md index af5e156..24e98db 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -58,11 +58,11 @@ Acceptance: an analyst can open an incident, see an ordered multi-stream timelin Goal: reduce repeat noise without hiding genuinely new behavior. -- [ ] Show active feedback rules, expiry, scope, author, and matching findings in the UI. -- [ ] Support feedback scope choices: exact value, entity and field, entire entity, or stream-wide rule. -- [ ] Require an expiry for broad suppressions and warn before a broad rule is saved. -- [ ] Track detector precision using confirmed versus false-positive feedback. -- [ ] Include feedback-derived precision and suppression counts in metrics and diagnostics. +- [x] Show active feedback rules, expiry, scope, author, and matching findings in the UI. +- [x] Support feedback scope choices: exact value, entity and field, entire entity, or stream-wide rule. +- [x] Require an expiry for broad suppressions and warn before a broad rule is saved. +- [x] Track detector precision using confirmed versus false-positive feedback. +- [x] Include feedback-derived precision and suppression counts in metrics and diagnostics. Acceptance: a false-positive decision suppresses only the intended pattern and all suppressions are auditable and reversible. diff --git a/src/fgai/anomaly.py b/src/fgai/anomaly.py index 42a3765..b755b0c 100644 --- a/src/fgai/anomaly.py +++ b/src/fgai/anomaly.py @@ -5,6 +5,7 @@ from collections import Counter, defaultdict from datetime import datetime from statistics import mean, pstdev +from .feedback import SUPPRESSING_STATUSES from .logs import THREAT_ACTIONS, event_score, is_utm_event from .models import AnomalyFinding, LogEvent @@ -220,7 +221,8 @@ def detect_source_anomalies( for deviation in field_deviations.get(src_ip, []): score += int(deviation.get("score", 0)) - reasons.append(str(deviation.get("reason", "stream field baseline deviation"))) + if deviation.get("feedback") not in SUPPRESSING_STATUSES: + reasons.append(str(deviation.get("reason", "stream field baseline deviation"))) if not reasons: continue diff --git a/src/fgai/dashboard.py b/src/fgai/dashboard.py index 42ebb54..9e96fa7 100644 --- a/src/fgai/dashboard.py +++ b/src/fgai/dashboard.py @@ -185,7 +185,7 @@ HTML = """

Events and Anomalies

Baseline and Stream Health

Operator Guidance

Waiting for monitor data.

Correlation Map

Waiting for correlated entities.

Investigation Incidents

Anomalies

Recommendations

AI Assessment

LLM assessment disabled.
-

Triage Queue

Field Baseline Deviations

Related Activity Across Sources

Block Candidates

Threat Intelligence

Policy Findings

+

Triage Queue

Field Baseline Deviations

Active Feedback Rules

Related Activity Across Sources

Block Candidates

Threat Intelligence

Policy Findings

Diagnostics

Recommended Stream Profiles

Waiting for observed stream data.

Installed Ollama Models

Loading local Ollama models.

Runtime Configuration

How To Use SignalScope

1. Normal workflow

  1. Settings: connect Graylog MCP and enable streams.
  2. Settings: apply recommended profiles for missing streams.
  3. Diagnostics: confirm raw samples, aggregate counts, and profile readiness.
  4. Overview: use Operator Guidance, incidents, trends, and correlation map.
  5. Findings: review only high-signal deviations first, then mark decisions.

2. Stream health

  • ready: profile exists, events are arriving, and baseline fields are ready.
  • learning: profile exists and events arrive, but baseline age or buckets are still too low.
  • missing_profile: stream is enabled but no profile exists. Apply or edit one.
  • no_events: stream is enabled but the current poll has no raw sample events.
  • partial_fetch: Graylog returned only part of the requested raw sample.

3. Ready fields

0/8 means 8 profile fields are tracked but none are mature enough yet. A field needs at least 12 baseline buckets and the configured Baseline training days before it is ready.

During learning, treat findings as signals to tune profiles, not as final alerts.

4. Profiles

  • Entity fields define who or what behavior is tracked, such as user, host, source IP, or application actor.
  • Baseline fields define the changing behavior to learn, such as action, event ID, service, URL, status, or counters.
  • Relationships learn pairs such as username to srcip or host to process.
  • Detectors add burst checks for auth failures, DNS queries, and deny actions.

5. Findings

  • Start with Triage Queue and incidents, not raw long tables.
  • Open evidence details before confirming a finding.
  • Use Expected for known behavior, False positive for bad signal, Confirmed for real investigation items.
  • Use expiry when a behavior is expected only temporarily.

6. High EPS / MCP

  • Use aggregate or auto fetch mode for high EPS streams.
  • Keep raw samples small enough for context; aggregate counts represent the full window.
  • Sample capped is normal in aggregate mode. Truncated raw mode means you may miss context.
  • If MCP is stale, the UI shows cached status so you can still inspect previous findings.

7. Ollama

  • Dashboard assessment summarizes current evidence.
  • Profile advisor maps unknown/custom fields and suggests relationships.
  • Ollama advice is constrained to fields discovered from Graylog; unknown fields are rejected.

8. Baseline DB size

  • Baseline DB size is the local SQLite behavior baseline.
  • Retention deletes old rows, but SQLite only returns disk space after manual VACUUM.
  • If the DB grows fast, lower bucket retention, value retention, or max values per entity field in Settings.
  • Stop the service before running signalscope baseline-maintenance --vacuum.

9. What to fix first

  1. No streams enabled.
  2. Enabled streams with missing profiles.
  3. Enabled streams with zero raw events.
  4. Profiles stuck at 0 ready fields after the training window.
  5. Too many repeated findings without review feedback.
  6. Baseline DB growing without maintenance.
@@ -216,6 +216,8 @@ function graylogEvidenceLink(query, configuration, label='Open in Graylog') { } const tableSort = {}; const uiCache = {correlations: [], fieldRows: [], correlationHitboxes: [], selectedCorrelationKey: '', entityLabels: {}}; +let pendingReview = null; +let lastReviewAuthor = ''; window.availableStreams = []; window.streamSelection = {}; window.configuredStreams = []; @@ -793,11 +795,76 @@ async function refresh() { document.getElementById('fieldDeviations').innerHTML = `${fieldRowsCached ? '

No current field deviations in this poll; showing cached findings from the last non-empty poll.

' : ''}` + table(fieldRows, [ {label:'Entity', key:'entity', render:r => entityCell(r)}, {label:'Stream', key:'stream_title'}, {label:'Detector', key:'detector'}, {label:'Field', key:'field'}, {label:'Score', key:'score'}, {label:'Baseline', render:r => esc(`${r.confidence || '-'}; ${r.baseline_age_days ?? 0}d; ${r.baseline_samples || 0} samples`)}, {label:'Review', render:r => esc(r.feedback || 'unreviewed')}, {label:'Evidence', render:r => { const weighted = r.weight && r.weight !== 1 ? `; weighted ${r.base_score ?? r.score} x ${r.weight}` : ''; const summary=esc(`${r.reason}; current ${r.current ?? '-'} vs baseline ${r.baseline ?? '-'}; scope ${r.baseline_scope ?? '-'}${weighted}; values: ${(r.sample_values || []).join(', ') || '-'}`); const events=(r.sample_events || []).map(item => `${esc(`${item.timestamp} | ${item.source} -> ${item.destination} | ${item.action} ${item.service} | ${item.value} | ${item.message}`)}${item.graylog_query ? `
${esc(item.graylog_query)}${graylogEvidenceLink(item.graylog_query, configuration)}` : ''}`).join('
'); const id=`deviation:${r.entity}:${r.stream_id}:${r.field}:${r.value || ''}`; return events ? `
${summary}

${events}

` : summary; }}, {label:'Review action', render:r => `
`} ], 'field-deviations'); - document.querySelectorAll('.feedback').forEach(button => button.addEventListener('click', async () => { - const note = prompt('Review note (optional):') || ''; - const days = prompt('Expiry in days (0 = no expiry):', '0') || '0'; - const response = await fetch('/api/feedback', {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify({status:button.dataset.status, entity:button.dataset.entity, stream_id:button.dataset.stream, field:button.dataset.field, value:button.dataset.value, note, expires_at: Number(days) > 0 ? Math.floor(Date.now()/1000) + Number(days) * 86400 : 0})}); - document.getElementById('feedbackNotice').textContent = response.ok ? 'Review saved. The matching pattern will be labeled on the next refresh.' : 'Could not save review.'; + const reviewForm = document.getElementById('reviewForm'); + const reviewScopeSelect = document.getElementById('reviewScope'); + const reviewScopeWarning = document.getElementById('reviewScopeWarning'); + const updateReviewScopeWarning = () => { + const broad = ['entity', 'stream'].includes(reviewScopeSelect.value); + reviewScopeWarning.hidden = !broad; + }; + document.querySelectorAll('.feedback').forEach(button => button.addEventListener('click', () => { + pendingReview = {status: button.dataset.status, entity: button.dataset.entity, stream_id: button.dataset.stream, field: button.dataset.field, value: button.dataset.value}; + reviewScopeSelect.value = pendingReview.value ? 'value' : 'field'; + document.getElementById('reviewNote').value = ''; + document.getElementById('reviewExpiryDays').value = '0'; + document.getElementById('reviewAuthor').value = lastReviewAuthor; + updateReviewScopeWarning(); + reviewForm.hidden = false; + reviewForm.scrollIntoView({block: 'nearest'}); + })); + reviewScopeSelect.onchange = updateReviewScopeWarning; + document.getElementById('reviewCancel').onclick = () => { + pendingReview = null; + reviewForm.hidden = true; + }; + document.getElementById('reviewSubmit').onclick = async () => { + if (!pendingReview) return; + const scope = reviewScopeSelect.value; + const note = document.getElementById('reviewNote').value || ''; + const author = document.getElementById('reviewAuthor').value || ''; + const days = Number(document.getElementById('reviewExpiryDays').value || '0'); + const broad = scope === 'entity' || scope === 'stream'; + if (broad && days <= 0) { + reviewScopeWarning.hidden = false; + return; + } + if (broad) { + const target = scope === 'entity' ? `entity "${pendingReview.entity}"` : `stream "${pendingReview.stream_id}"`; + if (!confirm(`This will suppress ALL findings for ${target} across every field until it expires. Continue?`)) return; + } + lastReviewAuthor = author; + const payload = {status: pendingReview.status, scope, entity: pendingReview.entity, stream_id: pendingReview.stream_id, field: pendingReview.field, value: pendingReview.value, note, author, expires_at: days > 0 ? Math.floor(Date.now()/1000) + days * 86400 : 0}; + const response = await fetch('/api/feedback', {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify(payload)}); + if (response.ok) { + document.getElementById('feedbackNotice').textContent = 'Review saved. The matching pattern will be labeled on the next refresh.'; + pendingReview = null; + reviewForm.hidden = true; + refresh(); + } else { + const error = await response.json().catch(() => ({})); + document.getElementById('feedbackNotice').textContent = `Could not save review: ${error.error || 'invalid request'}.`; + } + }; + const feedbackRules = (data.feedback || []).slice().sort((left, right) => Number(right.created_at || 0) - Number(left.created_at || 0)); + document.getElementById('feedbackRuleSummary').textContent = `${feedbackRules.length} active rule(s).`; + document.getElementById('feedbackRules').innerHTML = table(feedbackRules, [ + {label:'Scope', key:'scope'}, + {label:'Entity', render:r => esc(r.entity || '-')}, + {label:'Stream', render:r => esc(r.stream_id || '-')}, + {label:'Field', render:r => esc(r.field || '-')}, + {label:'Value', render:r => esc(r.value || '-')}, + {label:'Status', key:'status'}, + {label:'Author', render:r => esc(r.author || '-')}, + {label:'Note', render:r => esc(r.note || '-')}, + {label:'Expiry', render:r => r.expires_at ? esc(new Date(Number(r.expires_at) * 1000).toLocaleString()) : 'never'}, + {label:'Created', render:r => esc(new Date(Number(r.created_at || 0) * 1000).toLocaleString())}, + {label:'Matching findings', render:r => String(rawFieldRows.filter(item => item.feedback_rule_id === r.id).length)}, + {label:'Action', render:r => ``} + ], 'feedback-rules'); + document.querySelectorAll('.revoke-feedback-rule').forEach(button => button.addEventListener('click', async () => { + if (!confirm('Revoke this feedback rule? Matching findings will reappear on the next refresh.')) return; + const response = await fetch('/api/feedback/revoke', {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify({id: button.dataset.id})}); + document.getElementById('feedbackNotice').textContent = response.ok ? 'Rule revoked.' : 'Could not revoke rule.'; refresh(); })); const relatedGroups = compactRelatedActivity(relatedRows); @@ -836,6 +903,20 @@ async function refresh() { '

Entities

' + table(context.source_profiles || [], [{label:'Entity', key:'entity', render:r => entityCell(r)}, {label:'Events', key:'events'}, {label:'UTM', key:'utm_events'}, {label:'Deny', key:'deny_or_threat_actions'}, {label:'Destinations', key:'distinct_destinations'}, {label:'Actions', render:r => esc((r.top_actions || []).join(', '))}], 'entities') + '

Profile Baseline Readiness

' + table(profileReadiness, [{label:'Profile', key:'profile_name'}, {label:'Stream', key:'stream_title'}, {label:'Field', key:'field'}, {label:'Buckets', key:'buckets'}, {label:'Age days', key:'age_days'}, {label:'Training days', key:'training_days'}, {label:'Ready', key:'ready', render:r => r.ready ? 'ready' : 'learning'}, {label:'Action', render:r => r.stream_id ? `` : '-'}], 'profile-readiness') + '

Data Quality

' + table([quality], [{label:'Events', key:'events'}, {label:'Timestamp coverage', render:r => `${r.timestamp_coverage || 0}%`}, {label:'Source coverage', render:r => `${r.source_coverage || 0}%`}, {label:'Truncated streams', render:r => esc((r.truncated_streams || []).join(', ') || 'none')}]) + + '

Feedback Rules

' + table( + ['value', 'field', 'entity', 'stream'].map(scope => ({scope, count: feedbackRules.filter(item => (item.scope || 'value') === scope).length})), + [{label:'Scope', key:'scope'}, {label:'Active rules', key:'count'}] + ) + `

${rawFieldRows.filter(item => ['expected', 'false_positive'].includes(item.feedback || '')).length} finding(s) suppressed by feedback this poll.

` + + '

Detector Precision

' + table( + Object.entries(rawFieldRows.reduce((acc, item) => { + if (!['confirmed', 'false_positive'].includes(item.feedback || '')) return acc; + const key = item.detector || 'unknown'; + acc[key] = acc[key] || {confirmed: 0, false_positive: 0}; + acc[key][item.feedback] += 1; + return acc; + }, {})).map(([detector, counts]) => ({detector, confirmed: counts.confirmed, false_positive: counts.false_positive, precision: (counts.confirmed + counts.false_positive) ? `${(counts.confirmed / (counts.confirmed + counts.false_positive) * 100).toFixed(1)}%` : '-'})), + [{label:'Detector', key:'detector'}, {label:'Confirmed', key:'confirmed'}, {label:'False positive', key:'false_positive'}, {label:'Precision', key:'precision'}] + ) + '

Security Event Samples

' + table(context.security_event_samples || [], [{label:'Entity', key:'entity', render:r => entityCell(r)}, {label:'Type', key:'type'}, {label:'Action', key:'action'}, {label:'Severity', key:'severity'}, {label:'Destination', key:'dst'}, {label:'Service', key:'service'}]) + '

Top Sources

' + table(d.top_source_ips || [], [{label:'Value', key:'value', render:r => entityCell({entity:r.value})}, {label:'Count', key:'count'}]) + '

Top Destinations

' + table(d.top_destination_ips || [], [{label:'Value', key:'value'}, {label:'Count', key:'count'}]) + @@ -1283,6 +1364,13 @@ def serve_dashboard(host: str, port: int, status_file: str, *, image_dir: str | except (ValueError, json.JSONDecodeError) as exc: self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8")) return + if self.path == "/api/feedback/revoke" and self._is_loopback_client(): + try: + payload = json.loads(self.rfile.read(min(int(self.headers.get("Content-Length", "0")), 16_384)).decode("utf-8")) + self._send(200, "application/json", json.dumps(FeedbackStore().revoke(str(payload.get("id", "")))).encode("utf-8")) + except (ValueError, json.JSONDecodeError) as exc: + self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8")) + return if self.path == "/api/feedback/clear" and self._is_loopback_client(): try: payload = json.loads(self.rfile.read(min(int(self.headers.get("Content-Length", "0")), 16_384)).decode("utf-8")) diff --git a/src/fgai/feedback.py b/src/fgai/feedback.py index b8dd901..738e201 100644 --- a/src/fgai/feedback.py +++ b/src/fgai/feedback.py @@ -1,9 +1,77 @@ from __future__ import annotations +import hashlib import json import time from pathlib import Path +VALID_STATUSES = {"false_positive", "expected", "confirmed"} +VALID_SCOPES = {"value", "field", "entity", "stream"} +SCOPE_PRIORITY = {"value": 0, "field": 1, "entity": 2, "stream": 3} +SUPPRESSING_STATUSES = {"false_positive", "expected"} + +_SCOPE_ID_FIELDS = { + "value": ("stream_id", "entity", "field", "value"), + "field": ("stream_id", "entity", "field"), + "entity": ("entity",), + "stream": ("stream_id",), +} + + +def _resolve_scope(item: dict[str, object]) -> str: + scope = str(item.get("scope") or "").lower() + if scope: + return scope + return "value" if str(item.get("value", "")) else "field" + + +def _entry_id(scope: str, entry: dict[str, object]) -> str: + fields = _SCOPE_ID_FIELDS[scope] + payload = {"scope": scope, **{key: entry.get(key, "") for key in fields}} + return hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()[:16] + + +def _normalize(item: dict[str, object]) -> dict[str, object]: + scope = _resolve_scope(item) + if scope not in VALID_SCOPES: + raise ValueError("invalid feedback scope") + status = str(item.get("status", "")).lower() + if status not in VALID_STATUSES: + raise ValueError("invalid feedback status") + stream_id = str(item.get("stream_id", "")) + entity = str(item.get("entity", "")) + field = str(item.get("field", "")) + value = str(item.get("value", "")) + if scope == "stream" and not stream_id: + raise ValueError("stream-scoped feedback requires a stream_id") + if scope == "entity" and not entity: + raise ValueError("entity-scoped feedback requires an entity") + if scope in {"value", "field"} and not (entity and stream_id and field): + raise ValueError("field-scoped feedback requires entity, stream_id, and field") + expires_at = int(item.get("expires_at", 0) or 0) + if scope in {"entity", "stream"} and expires_at <= 0: + raise ValueError("broad-scope feedback rules require an expiry") + if scope == "stream": + entity = field = value = "" + elif scope == "entity": + stream_id = field = value = "" + elif scope == "field": + value = "" + entry = { + "stream_id": stream_id, + "entity": entity, + "field": field, + "value": value, + "scope": scope, + "status": status, + "note": str(item.get("note", "")), + "author": str(item.get("author", "")), + "created_at": int(time.time()), + "expires_at": expires_at, + } + entry["id"] = _entry_id(scope, entry) + return entry + class FeedbackStore: def __init__(self, path: str = "state/signalscope-feedback.json") -> None: @@ -15,17 +83,37 @@ class FeedbackStore: except (FileNotFoundError, json.JSONDecodeError): items = [] now = int(time.time()) - return [item for item in items if not item.get("expires_at") or int(item["expires_at"]) > now] + result = [] + for item in items: + if not isinstance(item, dict): + continue + if item.get("expires_at") and int(item["expires_at"]) <= now: + continue + scope = _resolve_scope(item) + item = dict(item) + item["scope"] = scope + item.setdefault("author", "") + item.setdefault("id", _entry_id(scope, item)) + result.append(item) + return result def add(self, item: dict[str, object]) -> dict[str, object]: - status = str(item.get("status", "")).lower() - if status not in {"false_positive", "expected", "confirmed"}: - raise ValueError("invalid feedback status") - entry = {"stream_id": str(item.get("stream_id", "")), "entity": str(item.get("entity", "")), "field": str(item.get("field", "")), "value": str(item.get("value", "")), "status": status, "note": str(item.get("note", "")), "created_at": int(time.time()), "expires_at": int(item.get("expires_at", 0) or 0)} - entries = [value for value in self.entries() if (value.get("stream_id"), value.get("entity"), value.get("field"), value.get("value", "")) != (entry["stream_id"], entry["entity"], entry["field"], entry["value"])] - entries.append(entry); self.path.parent.mkdir(parents=True, exist_ok=True); self.path.write_text(json.dumps(entries, indent=2), encoding="utf-8") + entry = _normalize(item) + entries = [value for value in self.entries() if value.get("id") != entry["id"]] + entries.append(entry) + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text(json.dumps(entries, indent=2), encoding="utf-8") return entry + def revoke(self, entry_id: str) -> dict[str, object]: + entries = self.entries() + before = len(entries) + entries = [item for item in entries if item.get("id") != entry_id] + removed = before - len(entries) + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text(json.dumps(entries, indent=2), encoding="utf-8") + return {"removed": removed, "remaining": len(entries), "id": entry_id} + def clear(self, *, status: str | None = None) -> dict[str, object]: entries = self.entries() if status is None: @@ -33,7 +121,7 @@ class FeedbackStore: entries = [] else: status = status.lower() - if status not in {"false_positive", "expected", "confirmed"}: + if status not in VALID_STATUSES: raise ValueError("invalid feedback status") before = len(entries) entries = [item for item in entries if str(item.get("status", "")).lower() != status] @@ -41,3 +129,27 @@ class FeedbackStore: self.path.parent.mkdir(parents=True, exist_ok=True) self.path.write_text(json.dumps(entries, indent=2), encoding="utf-8") return {"removed": removed, "remaining": len(entries), "status": status or "all"} + + +def feedback_matches(entry: dict[str, object], *, entity: str, stream_id: str, field: str, value: str) -> bool: + scope = str(entry.get("scope") or _resolve_scope(entry)) + if scope == "stream": + return str(entry.get("stream_id", "")) == stream_id + if scope == "entity": + return str(entry.get("entity", "")) == entity + if scope == "field": + return ( + str(entry.get("entity", "")) == entity + and str(entry.get("stream_id", "")) == stream_id + and str(entry.get("field", "")) == field + ) + return ( + str(entry.get("entity", "")) == entity + and str(entry.get("stream_id", "")) == stream_id + and str(entry.get("field", "")) == field + and str(entry.get("value", "")) == value + ) + + +def select_match(candidates: list[dict[str, object]]) -> dict[str, object]: + return min(candidates, key=lambda item: SCOPE_PRIORITY.get(str(item.get("scope") or _resolve_scope(item)), 1)) diff --git a/src/fgai/incidents.py b/src/fgai/incidents.py index bfb2679..97996fe 100644 --- a/src/fgai/incidents.py +++ b/src/fgai/incidents.py @@ -7,6 +7,7 @@ from collections import defaultdict from pathlib import Path from .entities import entity_type +from .feedback import SUPPRESSING_STATUSES from .models import AnomalyFinding @@ -29,7 +30,7 @@ def build_incidents(anomalies: list[AnomalyFinding], field_deviations: dict[str, for entity, deviations in field_deviations.items(): group = groups[entity] group["fields"].extend(deviations) - active = [item for item in deviations if item.get("feedback") not in {"expected", "false_positive"}] + active = [item for item in deviations if item.get("feedback") not in SUPPRESSING_STATUSES] group["score"] = min(100, int(group["score"]) + min(30, sum(int(item.get("score", 0)) for item in active))) group["evidence"].extend(str(item.get("reason", "")) for item in active) for item in active: diff --git a/src/fgai/metrics.py b/src/fgai/metrics.py index 9d33ece..82addd6 100644 --- a/src/fgai/metrics.py +++ b/src/fgai/metrics.py @@ -1,5 +1,7 @@ from __future__ import annotations +from .feedback import SUPPRESSING_STATUSES, VALID_SCOPES + def prometheus_metrics(status: dict[str, object]) -> str: summary = status.get("summary", {}) if isinstance(status.get("summary"), dict) else {} @@ -8,6 +10,9 @@ def prometheus_metrics(status: dict[str, object]) -> str: capabilities = status.get("capabilities", {}) if isinstance(status.get("capabilities"), dict) else {} mcp = capabilities.get("graylog_mcp", {}) if isinstance(capabilities.get("graylog_mcp"), dict) else {} incidents = status.get("incidents", []) if isinstance(status.get("incidents"), list) else [] + feedback = status.get("feedback", []) if isinstance(status.get("feedback"), list) else [] + deviations = status.get("field_deviations", {}) if isinstance(status.get("field_deviations"), dict) else {} + all_deviations = [item for items in deviations.values() if isinstance(items, list) for item in items if isinstance(item, dict)] lines = ["# HELP fgai_events_total Events in the latest analysis window.", "# TYPE fgai_events_total gauge"] for key in ("total", "utm", "threat_actions", "critical_or_high"): lines.append(f'fgai_events_total{{kind="{key}"}} {int(summary.get(key, 0) or 0)}') @@ -20,4 +25,26 @@ def prometheus_metrics(status: dict[str, object]) -> str: lines += ["# HELP signalscope_incidents Open incident candidates by severity.", "# TYPE signalscope_incidents gauge"] for severity in ("critical", "high", "medium", "low"): lines.append(f'signalscope_incidents{{severity="{severity}"}} {sum(1 for item in incidents if isinstance(item, dict) and item.get("severity") == severity)}') + lines += ["# HELP fgai_feedback_rules Active feedback rules by scope.", "# TYPE fgai_feedback_rules gauge"] + for scope in sorted(VALID_SCOPES): + lines.append(f'fgai_feedback_rules{{scope="{scope}"}} {sum(1 for item in feedback if isinstance(item, dict) and item.get("scope", "value") == scope)}') + suppressed = sum(1 for item in all_deviations if item.get("feedback") in SUPPRESSING_STATUSES) + lines += ["# HELP fgai_feedback_suppressed_findings Findings suppressed by feedback in the latest poll.", "# TYPE fgai_feedback_suppressed_findings gauge", f'fgai_feedback_suppressed_findings {suppressed}'] + precision: dict[str, dict[str, int]] = {} + for item in all_deviations: + verdict = item.get("feedback") + if verdict not in {"confirmed", "false_positive"}: + continue + detector = str(item.get("detector", "unknown")) + counts = precision.setdefault(detector, {"confirmed": 0, "false_positive": 0}) + counts[verdict] += 1 + lines += ["# HELP fgai_detector_feedback_total Confirmed vs false-positive feedback by detector.", "# TYPE fgai_detector_feedback_total gauge"] + for detector, counts in sorted(precision.items()): + for verdict in ("confirmed", "false_positive"): + lines.append(f'fgai_detector_feedback_total{{detector="{detector}",verdict="{verdict}"}} {counts[verdict]}') + lines += ["# HELP fgai_detector_precision Confirmed / (confirmed + false_positive) by detector.", "# TYPE fgai_detector_precision gauge"] + for detector, counts in sorted(precision.items()): + denominator = counts["confirmed"] + counts["false_positive"] + if denominator: + lines.append(f'fgai_detector_precision{{detector="{detector}"}} {counts["confirmed"] / denominator:.4f}') return "\n".join(lines) + "\n" diff --git a/src/fgai/monitor.py b/src/fgai/monitor.py index e8abbfd..c58e49b 100644 --- a/src/fgai/monitor.py +++ b/src/fgai/monitor.py @@ -12,7 +12,7 @@ from .baseline import BaselineStore from .config import ConfigStore from .correlation import correlate_source_ips from .event_context import build_event_context -from .feedback import FeedbackStore +from .feedback import FeedbackStore, SUPPRESSING_STATUSES, feedback_matches, select_match from .graylog_aggregate import GraylogAggregateSource from .graylog_mcp import GraylogMcpClient from .graylog_source import GraylogStreamSource @@ -383,16 +383,18 @@ def build_status( feedback = FeedbackStore().entries() for entity, deviations in field_deviations.items(): for deviation in deviations: - match = next(( + stream_id = str(deviation.get("stream_id", "")) + field = str(deviation.get("field", "")) + value = str(deviation.get("value", "")) + candidates = [ item for item in feedback - if item.get("entity") == entity - and item.get("stream_id") == deviation.get("stream_id") - and item.get("field") == deviation.get("field") - and (not item.get("value") or item.get("value") == deviation.get("value", "")) - ), None) - if match: + if feedback_matches(item, entity=entity, stream_id=stream_id, field=field, value=value) + ] + if candidates: + match = select_match(candidates) deviation["feedback"] = match["status"] - if match["status"] in {"false_positive", "expected"}: + deviation["feedback_rule_id"] = match.get("id", "") + if match["status"] in SUPPRESSING_STATUSES: deviation["score"] = 0 anomalies = detect_source_anomalies(events, limit=anomaly_limit, baselines=profiles, field_deviations=field_deviations) baseline_events = baseline.ingest(events) if baseline else 0 diff --git a/src/fgai/triage.py b/src/fgai/triage.py index ed56dd8..2827abf 100644 --- a/src/fgai/triage.py +++ b/src/fgai/triage.py @@ -2,6 +2,8 @@ from __future__ import annotations from collections import Counter, defaultdict +from .feedback import SUPPRESSING_STATUSES + def _severity(score: int) -> str: return "critical" if score >= 85 else "high" if score >= 60 else "medium" if score >= 35 else "low" @@ -56,7 +58,7 @@ def build_triage_queue( group["deviations"] = int(group["deviations"]) + 1 group["streams"].add(str(item.get("stream_name") or item.get("stream_title") or item.get("stream_id", ""))) group["detectors"][str(item.get("detector", "unknown"))] += 1 - if status in {"expected", "false_positive"}: + if status in SUPPRESSING_STATUSES: continue score = int(item.get("score", 0) or 0) active_scores.append(score) diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py index 5beaef7..bf3e119 100644 --- a/tests/test_dashboard.py +++ b/tests/test_dashboard.py @@ -76,6 +76,22 @@ class DashboardTests(unittest.TestCase): self.assertIn("/api/feedback/clear", source) self.assertIn("/api/review-state/clear", source) + def test_dashboard_exposes_feedback_revoke_endpoint(self): + source = inspect.getsource(dashboard) + self.assertIn("/api/feedback/revoke", source) + self.assertIn(".revoke(", source) + + def test_dashboard_has_active_feedback_rules_panel(self): + self.assertIn('id="feedbackRules"', HTML) + self.assertIn("revoke-feedback-rule", HTML) + self.assertIn('id="feedbackRuleSummary"', HTML) + + def test_dashboard_has_scoped_review_form(self): + self.assertIn('id="reviewScope"', HTML) + self.assertIn('value="entity"', HTML) + self.assertIn('value="stream"', HTML) + self.assertIn("Broad-scope rules must expire", HTML) + def test_dashboard_does_not_clear_streams_when_picker_is_unloaded(self): self.assertIn("const streamValues = Object.values(window.streamSelection || {})", HTML) self.assertIn("if (streamValues.length)", HTML) diff --git a/tests/test_feedback.py b/tests/test_feedback.py index 6e7536a..fbeafb8 100644 --- a/tests/test_feedback.py +++ b/tests/test_feedback.py @@ -1,8 +1,9 @@ import tempfile +import time import unittest from pathlib import Path -from fgai.feedback import FeedbackStore +from fgai.feedback import FeedbackStore, feedback_matches, select_match class FeedbackTests(unittest.TestCase): @@ -20,6 +21,74 @@ class FeedbackTests(unittest.TestCase): self.assertEqual(all_items, {"removed": 1, "remaining": 0, "status": "all"}) self.assertEqual(store.entries(), []) + def test_feedback_scope_defaults_and_ids_are_stable(self): + with tempfile.TemporaryDirectory() as directory: + store = FeedbackStore(str(Path(directory) / "feedback.json")) + no_value = store.add({"status": "expected", "entity": "alice", "stream_id": "windows", "field": "username"}) + self.assertEqual(no_value["scope"], "field") + + with_value = store.add({"status": "expected", "entity": "alice", "stream_id": "windows", "field": "username", "value": "bob"}) + self.assertEqual(with_value["scope"], "value") + self.assertEqual(len(store.entries()), 2) + + again = store.add({"status": "confirmed", "entity": "alice", "stream_id": "windows", "field": "username"}) + self.assertEqual(again["id"], no_value["id"]) + self.assertEqual(len(store.entries()), 2) + self.assertEqual(next(item for item in store.entries() if item["id"] == no_value["id"]).get("status"), "confirmed") + + def test_feedback_broad_scope_requires_expiry(self): + with tempfile.TemporaryDirectory() as directory: + store = FeedbackStore(str(Path(directory) / "feedback.json")) + with self.assertRaises(ValueError): + store.add({"status": "expected", "scope": "entity", "entity": "alice"}) + entry = store.add({"status": "expected", "scope": "entity", "entity": "alice", "expires_at": int(time.time()) + 3600}) + self.assertEqual(entry["scope"], "entity") + self.assertEqual(entry["stream_id"], "") + + def test_feedback_stream_scope_requires_stream_id(self): + with tempfile.TemporaryDirectory() as directory: + store = FeedbackStore(str(Path(directory) / "feedback.json")) + with self.assertRaises(ValueError): + store.add({"status": "expected", "scope": "stream", "expires_at": int(time.time()) + 3600}) + entry = store.add({"status": "expected", "scope": "stream", "stream_id": "vpn", "expires_at": int(time.time()) + 3600}) + self.assertEqual(entry["entity"], "") + + def test_feedback_entity_scope_requires_entity(self): + with tempfile.TemporaryDirectory() as directory: + store = FeedbackStore(str(Path(directory) / "feedback.json")) + with self.assertRaises(ValueError): + store.add({"status": "expected", "scope": "entity", "expires_at": int(time.time()) + 3600}) + + def test_feedback_revoke_removes_single_rule_by_id(self): + with tempfile.TemporaryDirectory() as directory: + store = FeedbackStore(str(Path(directory) / "feedback.json")) + one = store.add({"status": "expected", "entity": "alice", "stream_id": "windows", "field": "username"}) + store.add({"status": "confirmed", "entity": "bob", "stream_id": "vpn", "field": "username"}) + + result = store.revoke(one["id"]) + self.assertEqual(result, {"removed": 1, "remaining": 1, "id": one["id"]}) + self.assertEqual(len(store.entries()), 1) + + missing = store.revoke("does-not-exist") + self.assertEqual(missing, {"removed": 0, "remaining": 1, "id": "does-not-exist"}) + + def test_feedback_matches_respects_scope_and_precedence(self): + value_rule = {"scope": "value", "entity": "alice", "stream_id": "windows", "field": "username", "value": "10.0.0.1", "status": "confirmed"} + field_rule = {"scope": "field", "entity": "alice", "stream_id": "windows", "field": "username", "status": "expected"} + entity_rule = {"scope": "entity", "entity": "alice", "status": "expected"} + stream_rule = {"scope": "stream", "stream_id": "windows", "status": "expected"} + + self.assertTrue(feedback_matches(stream_rule, entity="alice", stream_id="windows", field="username", value="10.0.0.1")) + self.assertFalse(feedback_matches(stream_rule, entity="alice", stream_id="vpn", field="username", value="10.0.0.1")) + self.assertTrue(feedback_matches(entity_rule, entity="alice", stream_id="vpn", field="hostname", value="x")) + self.assertFalse(feedback_matches(field_rule, entity="alice", stream_id="windows", field="hostname", value="x")) + self.assertTrue(feedback_matches(value_rule, entity="alice", stream_id="windows", field="username", value="10.0.0.1")) + self.assertFalse(feedback_matches(value_rule, entity="alice", stream_id="windows", field="username", value="10.0.0.2")) + + candidates = [stream_rule, entity_rule, field_rule, value_rule] + self.assertEqual(select_match(candidates), value_rule) + self.assertEqual(select_match([stream_rule, entity_rule]), entity_rule) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 2052fbc..9f40e86 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -8,3 +8,45 @@ class MetricsTests(unittest.TestCase): output = prometheus_metrics({"summary": {"total": 10}, "anomaly_summary": {"high": 2}, "baseline": {"sources_ready": 3}, "capabilities": {"graylog_mcp": {"status": "connected", "events_fetched": 9}}}) self.assertIn('fgai_events_total{kind="total"} 10', output) self.assertIn("fgai_graylog_mcp_connected 1", output) + + def test_feedback_rules_counted_by_scope(self): + status = { + "feedback": [ + {"scope": "entity", "status": "expected"}, + {"scope": "value", "status": "confirmed"}, + ], + "field_deviations": {}, + } + text = prometheus_metrics(status) + self.assertIn('fgai_feedback_rules{scope="entity"} 1', text) + self.assertIn('fgai_feedback_rules{scope="value"} 1', text) + self.assertIn('fgai_feedback_rules{scope="field"} 0', text) + self.assertIn('fgai_feedback_rules{scope="stream"} 0', text) + + def test_suppressed_findings_and_detector_precision(self): + status = { + "feedback": [], + "field_deviations": { + "alice": [ + {"detector": "rare_value", "feedback": "confirmed"}, + {"detector": "rare_value", "feedback": "false_positive"}, + {"detector": "rare_value", "feedback": "expected"}, + {"detector": "auth_failure_burst", "feedback": "unreviewed"}, + ] + }, + } + text = prometheus_metrics(status) + self.assertIn("fgai_feedback_suppressed_findings 2", text) + self.assertIn('fgai_detector_feedback_total{detector="rare_value",verdict="confirmed"} 1', text) + self.assertIn('fgai_detector_feedback_total{detector="rare_value",verdict="false_positive"} 1', text) + self.assertIn('fgai_detector_precision{detector="rare_value"} 0.5000', text) + self.assertNotIn('detector="auth_failure_burst"', text) + + def test_metrics_handle_empty_status(self): + text = prometheus_metrics({}) + self.assertIn('fgai_feedback_rules{scope="value"} 0', text) + self.assertIn("fgai_feedback_suppressed_findings 0", text) + + +if __name__ == "__main__": + unittest.main()