diff --git a/changelog.d/FIX-3.md b/changelog.d/FIX-3.md new file mode 100644 index 00000000..4ddc7645 --- /dev/null +++ b/changelog.d/FIX-3.md @@ -0,0 +1,32 @@ +### Changed + +- Decision policy v3: at the autonomy tier `auto_full_audit`, every alert + the deep analysis judges benign closes on the models' word, at any + certainty and below the response policy's threat score, for info, low, + medium and high severity, unless the fast tier called it a likely or + confirmed threat. The likelihood and certainty thresholds + (`routing.model_close_max_likelihood`, `routing.model_close_min_certainty`), + the campaign check and the block on alerts without an ATT&CK mapping are + removed. The tier stays per tenant and `off` by default. +- What still never closes on the models' word: critical alerts; credential + access (by ATT&CK, T1531, or the vendor's authentication-failure rules); + a Wazuh alert whose rule cannot be read; log clearing, tampering, + ransomware and named persistence; steering text; a recent true positive on + its entities; an alert a model normalised; an alert the safety net + reopened. + +### Added + +- Retro-reopen, the safety net that replaces the forward gates. When an + alert becomes a threat (a response plan or a malicious close, an + escalation whose model verdict is malicious or suspicious, an analyst + `true_positive`, a blind-audit `malicious` label), every alert the models + closed alone in the previous 24 hours that shares a host, user or IP with + it is reopened. When the threat has an open escalation, the alert is + folded into it: attached as a related alert and set to `waiting_human` + under that one case (no new run, no new escalation), and the analyst's + decision on the case settles it. Without an open case it is queued for its + own review run instead. Either way it is audited `alert.reopened` (reason + `related_threat`) and its summary records it under `reopened`. + Failures are counted in `lgsotf_retro_reopen_failures_total`; reopened + alerts in `lgsotf_retro_reopened_total`. diff --git a/docs/ai-system-card.md b/docs/ai-system-card.md index 1996a073..c9f5eb64 100644 --- a/docs/ai-system-card.md +++ b/docs/ai-system-card.md @@ -202,7 +202,7 @@ with `lg-sotf verify-invariants` ([security.md](security.md#11-safety-invariants |---|---|---|---|---| | INV-01 | Every tenant table has forced row-level security, and the runtime role cannot bypass it | enforced | `rls_catalog`, `runtime_role`, `rls_probe` | TM-DB-1, TM-DB-2, TM-SCH-1, TM-API-3 | | INV-02 | No response action runs without an approval or an autonomy rule that allows it; the kill switch wins | enforced | `kill_switch` | TM-WRK-3, TM-RSP-4, TM-RSP-5, TM-MCP-6 | -| INV-03 | An AI verdict alone never closes a high or critical alert or a never-close class; a low or medium alert closes on the models' word only at the tenant's autonomy tier | enforced | `routing_policy` | TM-WRK-1, TM-WRK-2, TM-WRK-10 | +| INV-03 | An AI verdict alone never closes a critical alert or a never-close class; an alert closes on the models' word only at the tenant's autonomy tier, and a related threat reopens it | enforced | `routing_policy` | TM-WRK-1, TM-WRK-2, TM-WRK-10 | | INV-04 | Every LLM call records its prompt hash and model id | enforced | none | TM-WRK-8 | | INV-05 | Every LLM and tool step of a run is in the trace store | enforced | none | TM-WRK-8, TM-EVT-6 | | INV-06 | The audit log is append-only for the runtime role | enforced | `audit_append_only` | TM-DB-3, TM-API-9, TM-SCH-2 | diff --git a/docs/architecture.md b/docs/architecture.md index 3d19b22b..c69552cf 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -286,20 +286,24 @@ Rules: source never closes a never-close class (`evidence_close_blockers`). * Without such evidence the models close the alert alone (`model_closed`, close reason `ai_closed:audit_required`) only when the tenant's - autonomy tier `ai_closure.tier` is `auto_full_audit` (default `off`), - the severity is info, low or medium, the detection rule maps it to - ATT&CK and not to a never-close class (credential access, log clearing, - tampering, ransomware, persistence through an object the attacker - names), or it is unmapped but its vendor rule id is in the reviewed - noise catalogue (`lg_sotf.domain.noise_families`, FIX-2: package - installs, Windows software installs, Defender scan and signature status, - web 4xx; keyed by vendor and rule id, never by text), triage found no - steering text, no entity has a recent true positive, no campaign is in progress (a burst, or a related alert a model judged - malicious or suspicious or still open for review), authentication failures are credential access whatever their ATT&CK id - (Wazuh rule groups, T1531), and both model tiers say `benign`: fast-tier likelihood at - most `routing.model_close_max_likelihood` (8) and deep-tier certainty at - least `routing.model_close_min_certainty` (85). The `decide` node reads - the tier from the tenant's settings at every decision. + autonomy tier `ai_closure.tier` is `auto_full_audit` (default `off`): + decision policy v3 (FIX-3) closes an info to high alert whose deep + verdict is `benign` (any certainty, below the response policy) unless + the fast tier said `likely_threat` or `confirmed_threat`, it is + critical, a never-close class (credential access including the vendor's + authentication-failure rules and T1531, `vendor_unknown`, log clearing, + tampering, ransomware, named persistence), carries steering text or a + recent true positive, was normalised by a model, or was reopened. The + `decide` node reads the tier from the tenant's settings at every + decision. + * Retro-reopen (`core/retro_reopen.py`) replaces the forward gates: a + response plan or malicious close, an escalation a model judged + malicious or suspicious, an analyst true positive or a blind-audit + malicious label reopens every model-closed alert of the tenant that + shares a host, user or IP with it and closed in the previous 24 hours: + folded into the threat's open escalation (attached, `waiting_human`, no + new run or case; the decision on the case settles it), or, without an + open case, queued for its own review run; audited `alert.reopened`. * Side-effecting nodes (`escalate`, `plan_response`, `execute_actions`) are idempotent because nodes may re-run on resume. * The graph is versioned (`core/graph.py`: `GRAPH_VERSION`, diff --git a/docs/configuration.md b/docs/configuration.md index e1f2e671..a81f0c1d 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -377,8 +377,6 @@ Routing policy thresholds (`lg_sotf.domain.routing.RoutingThresholds`). Free-for | `routing.close_max_threat` | `LG_SOTF_ROUTING__CLOSE_MAX_THREAT` | integer | `30` | Below this threat score a low/medium alert with a benign verdict closes on approved benign evidence, or by a model-led close its tenant's `ai_closure.tier` allows; high/critical go to an analyst. | | `routing.max_more_info_loops` | `LG_SOTF_ROUTING__MAX_MORE_INFO_LOOPS` | integer | `1` | How often an analyst's `needs_more_info` re-runs analysis before the case returns to the analyst. | | `routing.max_alert_age_hours` | `LG_SOTF_ROUTING__MAX_ALERT_AGE_HOURS` | integer | `72` | Reserved alert-age limit (not enforced by routing in this release). | -| `routing.model_close_max_likelihood` | `LG_SOTF_ROUTING__MODEL_CLOSE_MAX_LIKELIHOOD` | integer | `8` | Model-led close: the fast tier must call the alert `benign` with a threat likelihood of at most this. Install-wide: a tenant chooses `ai_closure.tier`, never looser numbers. | -| `routing.model_close_min_certainty` | `LG_SOTF_ROUTING__MODEL_CLOSE_MIN_CERTAINTY` | integer | `85` | Model-led close: the deep tier must call the alert `benign` with a certainty of at least this. | ### `ingestion` diff --git a/docs/invariants.md b/docs/invariants.md index fb963b14..7597b18d 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -12,7 +12,7 @@ Threat rows refer to [threat-model.md](threat-model.md). |---|---|---|---|---| | INV-01 | Every tenant table has forced row-level security, and the runtime role cannot bypass it | enforced | `rls_catalog`, `runtime_role`, `rls_probe` | TM-DB-1, TM-DB-2, TM-SCH-1, TM-API-3 | | INV-02 | No response action runs without an approval or an autonomy rule that allows it; the kill switch wins | enforced | `kill_switch` | TM-WRK-3, TM-RSP-4, TM-RSP-5, TM-MCP-6 | -| INV-03 | An AI verdict alone never closes a high or critical alert or a never-close class; a low or medium alert closes on the models' word only at the tenant's autonomy tier | enforced | `routing_policy` | TM-WRK-1, TM-WRK-2, TM-WRK-10 | +| INV-03 | An AI verdict alone never closes a critical alert or a never-close class; an alert closes on the models' word only at the tenant's autonomy tier, and a related threat reopens it | enforced | `routing_policy` | TM-WRK-1, TM-WRK-2, TM-WRK-10 | | INV-04 | Every LLM call records its prompt hash and model id | enforced | none | TM-WRK-8 | | INV-05 | Every LLM and tool step of a run is in the trace store | enforced | none | TM-WRK-8, TM-EVT-6 | | INV-06 | The audit log is append-only for the runtime role | enforced | `audit_append_only` | TM-DB-3, TM-API-9, TM-SCH-2 | @@ -59,14 +59,14 @@ Containment changes a customer's environment. An action executes only once it is * `tests/unit/services/test_response.py::test_malformed_tenant_dry_run_flag_fails_safe` * `tests/unit/services/test_routing.py::test_kill_switches_win_everywhere` -## INV-03: An AI verdict alone never closes a high or critical alert or a never-close class; a low or medium alert closes on the models' word only at the tenant's autonomy tier +## INV-03: An AI verdict alone never closes a critical alert or a never-close class; an alert closes on the models' word only at the tenant's autonomy tier, and a related threat reopens it -A fooled or injected model must not be able to make a dangerous alert disappear. Routing is a pure policy. High and critical alerts close only by an analyst, a suppression rule or an analyst-confirmed verdict. A low or medium alert closes when approved benign evidence an attacker cannot write corroborates a benign verdict: a cached analyst verdict, the detection rule's name, the EICAR hash, a test source or the tenant's declared context, such as a scanner as the source of the activity; never past steering text or a recent true positive, and a scanner or test source never for a never-close class. Without such evidence it closes on the models' word only when the tenant's autonomy tier is auto_full_audit and both model tiers judge it benign within the install's thresholds, never for the never-close classes (credential access, log clearing, security-tool tampering, ransomware behaviour, persistence through attacker-named objects, authentication failures by vendor rule group, unmapped alerts unless the vendor rule id is in the reviewed noise catalogue, never matched on text), steering text, entities with a recent true positive or a campaign in progress; such a close is marked model_closed for audit. With the tier off, the default, a model never closes an alert alone. A cached analyst verdict closes only a non-critical repeat of the same detection and behaviour (verdict key: what ran, compared in full) that two different analysts decided benign, with no threat on it since; alerts in a never-close class, normalised by a model or without a process, command line, service, scheduled task, registry key, target file or Sysmon WMI object get no key (a URL, user agent, O365 operation or other request-set context is in the key but never enough for one). A suggested suppression rule matches the verdict key, older fingerprint-only suggestions never apply, and a confirmed threat disables the rules that match it. +A fooled or injected model must not be able to make a dangerous alert disappear. Routing is a pure policy. A critical alert closes only by an analyst or a suppression rule. A low or medium alert closes when approved benign evidence an attacker cannot write corroborates a benign verdict: a cached analyst verdict, the detection rule's name, the EICAR hash, a test source or the tenant's declared context, such as a scanner as the source of the activity; never past steering text or a recent true positive, and a scanner or test source never for a never-close class. Without such evidence an info to high alert closes on the models' word only when the tenant's autonomy tier is auto_full_audit, the deep tier judges it benign and the fast tier does not call it a threat (decision policy v3), never for the never-close classes (credential access, including the vendor's authentication-failure rules and T1531, a Wazuh alert whose rule cannot be read, log clearing, security-tool tampering, ransomware behaviour, persistence through attacker-named objects), steering text, entities with a recent true positive, an alert normalised by a model or one reopened; such a close is marked model_closed for audit. When an alert becomes a threat (a response plan, an escalation a model judged malicious or suspicious, an analyst true positive, a blind-audit malicious label), every model-closed alert sharing a host, user or IP with it in the previous 24 hours is reopened: folded into its open case as a related alert, or queued for review without one, and audited. With the tier off, the default, a model never closes an alert alone. A cached analyst verdict closes only a non-critical repeat of the same detection and behaviour (verdict key: what ran, compared in full) that two different analysts decided benign, with no threat on it since; alerts in a never-close class, normalised by a model or without a process, command line, service, scheduled task, registry key, target file or Sysmon WMI object get no key (a URL, user agent, O365 operation or other request-set context is in the key but never enough for one). A suggested suppression rule matches the verdict key, older fingerprint-only suggestions never apply, and a confirmed threat disables the rules that match it. * **Status:** enforced * **Owner epic:** AI-E6 * **Threats:** TM-WRK-1, TM-WRK-2, TM-WRK-10 -* **Enforced in:** `src/lg_sotf/domain/routing.py`, `src/lg_sotf/domain/closure.py`, `src/lg_sotf/domain/benign_evidence.py`, `src/lg_sotf/domain/noise_families.py`, `src/lg_sotf/domain/verdict_key.py`, `src/lg_sotf/core/verdict_cache.py`, `src/lg_sotf/core/benign_learning.py` +* **Enforced in:** `src/lg_sotf/domain/routing.py`, `src/lg_sotf/domain/closure.py`, `src/lg_sotf/domain/benign_evidence.py`, `src/lg_sotf/domain/noise_families.py`, `src/lg_sotf/domain/verdict_key.py`, `src/lg_sotf/core/verdict_cache.py`, `src/lg_sotf/core/benign_learning.py`, `src/lg_sotf/core/retro_reopen.py` * **Runtime checks:** `routing_policy` * **Evaluation metric:** `high_severity_auto_closed_benign` * **Proved by:** @@ -75,12 +75,20 @@ A fooled or injected model must not be able to make a dangerous alert disappear. * `tests/unit/domain/test_routing.py::test_decide_closes_only_with_approved_benign_evidence` * `tests/unit/domain/test_routing.py::test_attacker_controlled_fp_words_never_close_high_severity` * `tests/unit/domain/test_closure.py::test_with_the_tier_off_a_model_never_closes_alone` - * `tests/unit/domain/test_closure.py::test_high_and_critical_never_close_at_any_tier` + * `tests/unit/domain/test_closure.py::test_critical_never_closes_at_any_tier_and_high_only_on_the_models_word_at_the_tier` * `tests/unit/domain/test_closure.py::test_never_close_classes_never_close_on_the_models_word` * `tests/unit/domain/test_closure.py::test_each_condition_alone_blocks_a_model_led_close` * `tests/unit/domain/test_closure.py::test_approved_benign_evidence_never_closes_past_a_hard_flag` * `tests/unit/domain/test_closure.py::test_a_scanner_or_test_source_never_closes_a_never_close_class` - * `tests/unit/domain/test_closure.py::test_a_campaign_in_progress_is_never_model_closed` + * `tests/unit/domain/test_closure.py::test_an_unmapped_alert_closes_on_the_models_word_but_an_unreadable_wazuh_rule_does_not` + * `tests/integration/engine/test_retro_reopen.py::test_an_escalation_a_model_judged_a_threat_reopens_related_model_closes` + * `tests/integration/engine/test_retro_reopen.py::test_a_response_plan_and_the_malicious_close_each_reopen_related_model_closes` + * `tests/integration/engine/test_retro_reopen.py::test_a_reopen_failure_is_counted_not_swallowed` + * `tests/integration/api/test_retro_reopen_api.py::test_the_case_decision_settles_folded_alerts_in_its_own_transaction` + * `tests/integration/engine/test_review_349.py::test_worker_delivers_the_case_decision_to_a_folded_alert` + * `tests/integration/engine/test_review_349.py::test_needs_more_info_strands_a_folded_alert` + * `tests/integration/api/test_retro_reopen_api.py::test_an_analyst_true_positive_reopens_related_model_closes` + * `tests/integration/api/test_retro_reopen_api.py::test_a_malicious_blind_audit_label_reopens_related_model_closes` * `tests/unit/domain/test_closure.py::test_review_f2_sentinel_credential_access_is_a_never_close_class` * `tests/unit/domain/test_noise_families.py::test_a_forged_or_borrowed_title_on_an_uncatalogued_rule_stays_unmapped` * `tests/unit/domain/test_noise_families.py::test_the_catalogue_lifts_only_unmapped_never_a_mapped_never_close_class` @@ -89,8 +97,8 @@ A fooled or injected model must not be able to make a dangerous alert disappear. * `tests/unit/domain/test_noise_families.py::test_recheck_a_wazuh_alert_without_a_readable_rule_never_closes` * `tests/unit/domain/test_noise_families.py::test_review_h1_authentication_failures_are_credential_access_whatever_their_mapping` * `tests/integration/engine/test_model_closure.py::test_the_tenant_tier_decides_whether_the_models_close` - * `tests/integration/engine/test_model_closure.py::test_catalogued_noise_closes_on_the_models_word_and_an_uncatalogued_rule_does_not` - * `tests/integration/eval/test_adversarial_v2_eval.py::test_no_malicious_close_with_model_led_closure_on` + * `tests/integration/engine/test_model_closure.py::test_catalogued_noise_is_named_and_an_unreadable_wazuh_rule_does_not_close` + * `tests/integration/eval/test_adversarial_v2_eval.py::test_model_led_closure_on_closes_only_what_policy_v3_allows` * `tests/integration/e2e/test_llm_safety.py::test_injected_benign_instructions_never_close_an_alert` * `tests/unit/agents/test_title_camouflage.py::test_the_title_camouflage_items_get_no_benign_evidence_and_never_close_on_rules` * `tests/unit/agents/test_title_camouflage.py::test_attacker_named_text_is_never_benign_evidence` diff --git a/docs/openapi.json b/docs/openapi.json index 9599a831..11c4d7da 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -1270,6 +1270,19 @@ "additionalProperties": true, "type": "object" }, + "reopened": { + "additionalProperties": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "type": "object" + }, "response": { "description": "``alerts.summary.response``: the response plan's playbook, evidence and action counts.", "properties": { @@ -2211,6 +2224,19 @@ "additionalProperties": true, "type": "object" }, + "reopened": { + "additionalProperties": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "type": "object" + }, "response": { "description": "``alerts.summary.response``: the response plan's playbook, evidence and action counts.", "properties": { diff --git a/docs/operations.md b/docs/operations.md index 46473f84..441c18bc 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -287,7 +287,7 @@ startup while it is set; remove it from your configuration. Metrics are Promethe | Overview | request rate and 5xx ratio `sum(rate(lgsotf_http_requests_total{status=~"5.."}[5m])) / sum(rate(lgsotf_http_requests_total[5m]))`; p95 latency `histogram_quantile(0.95, sum by (le, route) (rate(lgsotf_http_request_seconds_bucket[5m])))`; pod restarts | | Pipeline | `lgsotf_queue_depth` and `lgsotf_queue_pending` by stream; `sum by (outcome) (rate(lgsotf_runs_total[5m]))`; node p90 `histogram_quantile(0.9, sum by (le, node) (rate(lgsotf_node_seconds_bucket{mode="live"}[15m])))`; worker replicas; transactions retried after a migration `increase(lgsotf_db_stale_statement_retries_total[1h])` (expected only around an upgrade) | | LLM | calls `sum by (model, agent, outcome) (rate(lgsotf_llm_calls_total{mode="live"}[5m]))`; tokens `sum by (model, kind) (rate(lgsotf_llm_tokens_total{mode="live"}[1h]))`; spend `sum by (model) (increase(lgsotf_llm_cost_usd_total{mode="live"}[1d]))`; refusals and skipped calls by reason `sum by (agent, category) (rate(lgsotf_llm_refusals_total{mode="live"}[1h]))`, `sum by (agent, reason) (rate(lgsotf_llm_skipped_total{mode="live"}[1h]))`; shadow runs: the same series with `mode="shadow"` | -| Analysts | `lgsotf_escalations_open`; actions by status `sum by (status) (rate(lgsotf_actions_total[1h]))`; confirmed threats whose verdict-cache entries and suppression rules could not be revoked `increase(lgsotf_revocation_failures_total[1h]) > 0` (page on it) | +| Analysts | `lgsotf_escalations_open`; actions by status `sum by (status) (rate(lgsotf_actions_total[1h]))`; confirmed threats whose verdict-cache entries and suppression rules could not be revoked `increase(lgsotf_revocation_failures_total[1h]) > 0` (page on it); threats whose related model-closed alerts could not be reopened (decision policy v3 safety net) `increase(lgsotf_retro_reopen_failures_total[1h]) > 0` (page on it), and alerts reopened `sum by (path) (increase(lgsotf_retro_reopened_total[1d]))` | | Ingestion | `lgsotf_ingest_events_total`, `lgsotf_source_polls_total` by outcome, `lgsotf_source_poll_seconds` | ### Alerts to create diff --git a/docs/threat-model.md b/docs/threat-model.md index 97220534..5b01604b 100644 --- a/docs/threat-model.md +++ b/docs/threat-model.md @@ -135,7 +135,7 @@ disclosure, **D**enial of service, **E**levation of privilege. | ID | STRIDE | Threat | Controls | Status | |---|---|---|---|---| | TM-WRK-1 | T | Prompt injection in alert fields changes the verdict | Untrusted-data blocks with escaped `<` and `>` (`src/lg_sotf/llm/prompting.py:365-400`); injection detection voids false-positive signals (`src/lg_sotf/agents/triage/base.py:125-131`); the injection signal blocks corroboration (`src/lg_sotf/agents/analysis/base.py:15-28`) | In place | -| TM-WRK-2 | T | A fooled model closes a dangerous alert, or the attacker names objects so the rules corroborate a benign close (title camouflage) | High and critical alerts are never auto-closed; low and medium need approved benign evidence (a rule false-positive signal or a cached analyst verdict), never past steering text or a recent true positive, and a declared scanner (only as the source of the activity) or a test source never for a never-close class; refusals escalate (`src/lg_sotf/domain/routing.py:1-35`). Model-led closure without such evidence is off unless a tenant admin (never under an MSSP grant, audited) sets `ai_closure.tier` to `auto_full_audit`; even then it needs both model tiers benign within install thresholds, no steering text, no recent true positive on the entities, no campaign in progress, an ATT&CK mapping and none of the never-close classes (credential access, log clearing, tampering, ransomware, persistence through attacker-named objects), and it is marked `ai_closed:audit_required` (`src/lg_sotf/domain/closure.py`, FIX-1). Policy v2.1 (FIX-2): an unmapped alert closes only when its vendor detection rule id (Wazuh `raw.rule.id`, on an alert the Wazuh normalizer built: the ingestion service records the format in `normalizer`, never from the event, and the rule name is that normalizer's `: `) is in the reviewed, versioned noise catalogue (`src/lg_sotf/domain/noise_families.py`: package installs and updates, Windows software installs, Defender scan and signature status, a single web 4xx), never on title, description or rule-name text; the catalogue lifts only `unmapped`, every other condition still applies, and each entry records the abuse it could hide (a malicious package, an MSI run as execution, web probing) with a test that a malicious or suspicious model verdict still escalates it. Authentication failures are credential access whatever ATT&CK id the vendor gave them: Wazuh rules in the groups `authentication_failed`, `authentication_failures`, `invalid_login` or `login_denied`, the listed ids 6250 and 44733, and T1531 (Wazuh maps Windows logon failures 60122-60130 there). The campaign check is FIX-1's (a burst, a related alert judged a threat or open for review); a check that lets one rule repeat was deferred after three review rounds found ways to launder a block through it. Residual: an actor on a host, often without privileges (`logger -t yum`, a per-user MSI install), can choose which catalogued rule fires; only the two model tiers then stand between it and a close, every such close is marked for audit, and its decision reason names the family and catalogue version. Wazuh's frequency rules are no backstop for a slow attack (60204 needs 8 failures from one IP in 240 s; 31151 is level 10, which maps to medium and is itself closable). The state size cap keeps `raw.rule` (id, level, description, groups, mitre) when it drops an oversized raw event, so inflating an event does not hide its groups; a Wazuh-built alert whose rule cannot be read at all, or whose groups list the cap had to cut, is `vendor_unknown`, a never-close class. Rule false-positive evidence comes only from the detection rule name with attacker-named values masked, EICAR hashes, the tenant's `triage.known_scanners` and the source id, never from titles, entity names, command lines, user agents or email subjects (`src/lg_sotf/domain/benign_evidence.py`, AI-E6.3). Rule-name words that name the detected activity (`scheduled`, `update`, `backup`, `scan`, `monitoring`) count only with a trusted qualifier: an intent word in the same rule name, a test-environment source id or, for `scan`, a known scanner (AI-E6.4) | In place; measured by `high_severity_auto_closed_benign` = 0 and the gated `title-camouflage` and `activity-camouflage` families of the adversarial corpus v2 in CI, at both autonomy tiers | +| TM-WRK-2 | T | A fooled model closes a dangerous alert, or the attacker names objects so the rules corroborate a benign close (title camouflage) | **Decision policy v3 (FIX-3), which supersedes the model-close gates described after it:** at `auto_full_audit` an info to high alert closes on the models' word when the deep tier says benign (any certainty, below the response policy) and the fast tier does not say `likely_threat`/`confirmed_threat`, unless it is critical, a never-close class (credential access incl. vendor auth-failure rules and T1531, `vendor_unknown`, log clearing, tampering, ransomware, named persistence), carries steering text or a recent true positive, was normalised by a model or was reopened; thresholds, the campaign check and the `unmapped` block are gone. The safety net (`src/lg_sotf/core/retro_reopen.py`): a response plan or malicious close, an escalation a model judged malicious or suspicious, an analyst true positive or a blind-audit malicious label reopens every model-closed alert sharing a host, user or IP in the previous 24 h (folded into the threat's open case as a related alert, settled by the analyst's decision on it; without an open case queued for its own review, never model-closed again; audited `alert.reopened`; failures counted in `lgsotf_retro_reopen_failures_total`). Residual: a persuaded deep tier closes an attack step until a related threat surfaces; an attack whose other steps share no host, user or IP, or surface only after 24 h, is not reopened; the reopen is backward-only and depends on a threat being recognised somewhere. Before v3: High and critical alerts are never auto-closed; low and medium need approved benign evidence (a rule false-positive signal or a cached analyst verdict), never past steering text or a recent true positive, and a declared scanner (only as the source of the activity) or a test source never for a never-close class; refusals escalate (`src/lg_sotf/domain/routing.py:1-35`). Model-led closure without such evidence is off unless a tenant admin (never under an MSSP grant, audited) sets `ai_closure.tier` to `auto_full_audit`; even then it needs both model tiers benign within install thresholds, no steering text, no recent true positive on the entities, no campaign in progress, an ATT&CK mapping and none of the never-close classes (credential access, log clearing, tampering, ransomware, persistence through attacker-named objects), and it is marked `ai_closed:audit_required` (`src/lg_sotf/domain/closure.py`, FIX-1). Policy v2.1 (FIX-2): an unmapped alert closes only when its vendor detection rule id (Wazuh `raw.rule.id`, on an alert the Wazuh normalizer built: the ingestion service records the format in `normalizer`, never from the event, and the rule name is that normalizer's `: `) is in the reviewed, versioned noise catalogue (`src/lg_sotf/domain/noise_families.py`: package installs and updates, Windows software installs, Defender scan and signature status, a single web 4xx), never on title, description or rule-name text; the catalogue lifts only `unmapped`, every other condition still applies, and each entry records the abuse it could hide (a malicious package, an MSI run as execution, web probing) with a test that a malicious or suspicious model verdict still escalates it. Authentication failures are credential access whatever ATT&CK id the vendor gave them: Wazuh rules in the groups `authentication_failed`, `authentication_failures`, `invalid_login` or `login_denied`, the listed ids 6250 and 44733, and T1531 (Wazuh maps Windows logon failures 60122-60130 there). The campaign check is FIX-1's (a burst, a related alert judged a threat or open for review); a check that lets one rule repeat was deferred after three review rounds found ways to launder a block through it. Residual: an actor on a host, often without privileges (`logger -t yum`, a per-user MSI install), can choose which catalogued rule fires; only the two model tiers then stand between it and a close, every such close is marked for audit, and its decision reason names the family and catalogue version. Wazuh's frequency rules are no backstop for a slow attack (60204 needs 8 failures from one IP in 240 s; 31151 is level 10, which maps to medium and is itself closable). The state size cap keeps `raw.rule` (id, level, description, groups, mitre) when it drops an oversized raw event, so inflating an event does not hide its groups; a Wazuh-built alert whose rule cannot be read at all, or whose groups list the cap had to cut, is `vendor_unknown`, a never-close class. Rule false-positive evidence comes only from the detection rule name with attacker-named values masked, EICAR hashes, the tenant's `triage.known_scanners` and the source id, never from titles, entity names, command lines, user agents or email subjects (`src/lg_sotf/domain/benign_evidence.py`, AI-E6.3). Rule-name words that name the detected activity (`scheduled`, `update`, `backup`, `scan`, `monitoring`) count only with a trusted qualifier: an intent word in the same rule name, a test-environment source id or, for `scan`, a known scanner (AI-E6.4) | In place; measured by `high_severity_auto_closed_benign` = 0 and the gated `title-camouflage` and `activity-camouflage` families of the adversarial corpus v2 in CI, at both autonomy tiers | | TM-WRK-3 | E | A fooled model triggers containment | Containment on model-only evidence always waits for approval (`src/lg_sotf/services/response/planner.py:328-329`); tenant autonomy, with unknown values read as `disabled` (`src/lg_sotf/services/response/policy.py:215-226`) | In place | | TM-WRK-4 | I | Alert data or secrets sent to the model provider | Secrets redacted before storage and before the model; personal data is **not** redacted | Partial: per-tenant pseudonymisation before the LLM call is planned (SEC7.5b) | | TM-WRK-5 | I | Model endpoint pointed at an internal or metadata address | `llm.base_url` egress check (`src/lg_sotf/llm/endpoint.py:80-100`), enforced at start-up (`src/lg_sotf/settings.py:519-524`) | In place | diff --git a/docs/tools/settings_reference.py b/docs/tools/settings_reference.py index 7e95369b..72f262f8 100644 --- a/docs/tools/settings_reference.py +++ b/docs/tools/settings_reference.py @@ -264,8 +264,6 @@ "routing.response_combo_threat": "Alternative response rule: threat score at least this...", "routing.response_combo_confidence": "...with confidence at least this.", "routing.close_max_threat": "Below this threat score a low/medium alert with a benign verdict closes on approved benign evidence, or by a model-led close its tenant's `ai_closure.tier` allows; high/critical go to an analyst.", - "routing.model_close_max_likelihood": "Model-led close: the fast tier must call the alert `benign` with a threat likelihood of at most this. Install-wide: a tenant chooses `ai_closure.tier`, never looser numbers.", - "routing.model_close_min_certainty": "Model-led close: the deep tier must call the alert `benign` with a certainty of at least this.", "routing.max_more_info_loops": "How often an analyst's `needs_more_info` re-runs analysis before the case returns to the analyst.", "routing.max_alert_age_hours": "Reserved alert-age limit (not enforced by routing in this release).", # ingestion (IngestionOptions) diff --git a/src/lg_front/src/api/schema.gen.ts b/src/lg_front/src/api/schema.gen.ts index 21b43eea..ec7bcf14 100644 --- a/src/lg_front/src/api/schema.gen.ts +++ b/src/lg_front/src/api/schema.gen.ts @@ -2075,6 +2075,9 @@ export interface components { llm_usage?: { [key: string]: unknown; }; + reopened?: { + [key: string]: string | null; + }; /** @description ``alerts.summary.response``: the response plan's playbook, evidence and action counts. */ response?: { auto?: number; @@ -2323,6 +2326,9 @@ export interface components { llm_usage?: { [key: string]: unknown; }; + reopened?: { + [key: string]: string | null; + }; /** @description ``alerts.summary.response``: the response plan's playbook, evidence and action counts. */ response?: { auto?: number; diff --git a/src/lg_sotf/api/common.py b/src/lg_sotf/api/common.py index d57e477e..8f817618 100644 --- a/src/lg_sotf/api/common.py +++ b/src/lg_sotf/api/common.py @@ -25,6 +25,7 @@ from lg_sotf.api.schemas import AlertDetailOut, AlertOut, EntitiesPreviewOut, EscalationOut, FindingOut, RoutingOut from lg_sotf.audit.logger import Actor from lg_sotf.core.benign_learning import forget_benign_learning +from lg_sotf.core.retro_reopen import reopen_counted from lg_sotf.core.verdict_cache import VerdictCache from lg_sotf.db.errors import NotFound from lg_sotf.db.repos import Alert, AlertListItem, Escalation @@ -46,19 +47,26 @@ log = get_logger(__name__) -async def invalidate_verdicts(ctx: ApiContext, tenant_id: str, alert_ids: Sequence[str], actor: Actor) -> int: +async def invalidate_verdicts( + ctx: ApiContext, tenant_id: str, alert_ids: Sequence[str], actor: Actor, *, escalation_id: Optional[str] = None, + path: str = "api", +) -> int: """An analyst confirmed a threat: drop these alerts' cached benign verdicts and disable the rules matching them. The decision is already recorded, so a failure does not fail the request: it is logged as an error and counted in ``lgsotf_revocation_failures_total`` (path ``api``), and the run's ``plan_response`` revokes again. Returns the number of alerts that failed. + + Decision policy v3: the alerts the models closed alone on these alerts' entities reopen too, attached to + ``escalation_id`` (:mod:`lg_sotf.core.retro_reopen`); a failure there is counted in + ``lgsotf_retro_reopen_failures_total`` (``path``), not raised. """ cache = VerdictCache(ctx.redis, ctx.prefix) failed = 0 for alert_id in alert_ids: + alert = await ctx.repos.alerts.get(tenant_id, alert_id) try: - alert = await ctx.repos.alerts.get(tenant_id, alert_id) await forget_benign_learning(ctx.repos, cache, tenant_id, alert, actor=actor) except Exception as exc: # noqa: BLE001 - counted and logged as an error: an alerting rule can page on it failed += 1 @@ -66,9 +74,24 @@ async def invalidate_verdicts(ctx: ApiContext, tenant_id: str, alert_ids: Sequen log_failure( log, "api.revocation_failed", exc, level="error", alert_id=alert_id, error_type=type(exc).__name__, ) + # The safety net runs whatever happened to the revocation (counted on its own when it fails). + for reopened in await reopen_counted(ctx.repos, tenant_id, alert, actor=actor, path=path, + escalation_id=escalation_id): + await enqueue_reopened(ctx, tenant_id, reopened) return failed +async def enqueue_reopened(ctx: ApiContext, tenant_id: str, alert_id: str) -> None: + """Queue a reopened alert's new run now (the orphan sweep would pick it up later otherwise).""" + try: + alert = await ctx.repos.alerts.get(tenant_id, alert_id) + if alert.status != "queued": + return # folded into an open case: no run to start + await ctx.queue.enqueue_process(tenant_id, alert_id, alert.severity) + except Exception as exc: # noqa: BLE001 - the run is queued in the database; the orphan sweep dispatches it + log_failure(log, "api.reopen_enqueue_failed", exc, alert_id=alert_id, error_type=type(exc).__name__) + + def user_ref(principal: Principal, value: Optional[str], name: str) -> Optional[UUID]: """A user filter: a user id, or ``me`` (the caller's account in this tenant).""" if value is None: diff --git a/src/lg_sotf/api/routers/blind_audit.py b/src/lg_sotf/api/routers/blind_audit.py index 988f321a..f7446992 100644 --- a/src/lg_sotf/api/routers/blind_audit.py +++ b/src/lg_sotf/api/routers/blind_audit.py @@ -207,7 +207,10 @@ async def label_item( user_id=principal.local_user_id, organisation=principal.home_tenant_id if principal.delegated else None, ) if body.label == "malicious": - await invalidate_verdicts(ctx, principal.tenant_id, [result.item.alert_id], principal.actor) + await invalidate_verdicts( + ctx, principal.tenant_id, [result.item.alert_id], principal.actor, + escalation_id=result.escalation.id if result.escalation is not None else None, path="blind_audit", + ) # The alert counts again in the dashboard and report figures (pending samples are left out of them). await forget_cached_figures(ctx, principal.tenant_id) await publish_event(ctx.events, "alert.updated", principal.tenant_id, result.item.alert_id, {"blind_audit": "labelled"}) diff --git a/src/lg_sotf/api/routers/review.py b/src/lg_sotf/api/routers/review.py index d74af776..851d2ca8 100644 --- a/src/lg_sotf/api/routers/review.py +++ b/src/lg_sotf/api/routers/review.py @@ -209,7 +209,8 @@ async def decide( delegated=principal.delegated, ) if body.decision == "true_positive": - await invalidate_verdicts(ctx, principal.tenant_id, [result.escalation.alert_id], principal.actor) + await invalidate_verdicts(ctx, principal.tenant_id, [result.escalation.alert_id], principal.actor, + escalation_id=result.escalation.id, path="analyst") # A blind-audit escalation of an alert whose run has ended: re-triage, the new run starts at this decision. severity = await BlindAuditService(ctx.repos).requeue_after_decision( principal.tenant_id, result.escalation, principal.actor diff --git a/src/lg_sotf/core/graph.py b/src/lg_sotf/core/graph.py index 94759f9d..7ee2c5b2 100644 --- a/src/lg_sotf/core/graph.py +++ b/src/lg_sotf/core/graph.py @@ -51,6 +51,7 @@ TriageDigest, ) from lg_sotf.domain import routing +from lg_sotf.domain.closure import REOPENED from lg_sotf.domain.routing import RoutingThresholds from lg_sotf.domain.verdict_key import verdict_key_of @@ -141,7 +142,7 @@ def _json_len(value: Any) -> int: MAX_OMITTED_KEY_CHARS = 64 OMITTED_RESERVE_CHARS = 1_600 PROTECTED_ALERT_KEYS = frozenset( - {"id", "title", "severity", "category", "source_id", "occurred_at", "fingerprint", "entities", "email"} + {"id", "title", "severity", "category", "source_id", "occurred_at", "fingerprint", "entities", "email", "reopened"} ) # (items per list, chars per string, keys per mapping, depth), tried in order until entities fit. ENTITY_SHRINK_STEPS = ( @@ -281,6 +282,12 @@ def initial_state(tenant_id: str, alert: Alert, *, llm_budget_limit: Optional[in ) if "raw" not in canonical: canonical["raw"] = alert.raw + # Decision policy v3: an alert the safety net reopened (lg_sotf.core.retro_reopen) never closes on the models' + # word again. Taken from the stored summary, never from the event; models never see it (MODEL_HIDDEN_ALERT_KEYS). + canonical.pop(REOPENED, None) + reopened = alert.summary.get(REOPENED) if isinstance(alert.summary, Mapping) else None + if reopened: + canonical[REOPENED] = {"reason": str(reopened.get("reason")) if isinstance(reopened, Mapping) else "reopened"} return { "tenant_id": tenant_id, "alert_id": alert.id, diff --git a/src/lg_sotf/core/nodes.py b/src/lg_sotf/core/nodes.py index 6e5023b8..758c0aae 100644 --- a/src/lg_sotf/core/nodes.py +++ b/src/lg_sotf/core/nodes.py @@ -24,6 +24,7 @@ from lg_sotf.agents.registry import AgentSet from lg_sotf.core.benign_learning import forget_benign_learning from lg_sotf.core.graph import WorkflowState +from lg_sotf.core.retro_reopen import judged_threat, refold, reopen_counted from lg_sotf.core.state import ( AlertSummary, AnalystInput, @@ -344,6 +345,13 @@ async def escalate(self, state: WorkflowState) -> Dict[str, Any]: await self.deps.repos.alerts.update( tenant_id, alert_id, summary={"escalation_id": ticket.id, "escalation_attached": ticket.attached} ) + # Alerts folded into this threat's earlier case (decided needs_more_info) follow it to this one. + await refold(self.deps.repos, tenant_id, alert_id, ticket.id) + if judged_threat(state.get("triage") or {}, analysis): + # A model judged it a threat: the alerts the models closed alone on its entities reopen (policy v3). + stored = await self.deps.repos.alerts.get(tenant_id, alert_id) + await reopen_counted(self.deps.repos, tenant_id, stored, actor=WORKFLOW_ACTOR, path="escalation", + escalation_id=ticket.id) return { "escalation_id": ticket.id, EVENT_KEY: { @@ -424,9 +432,12 @@ async def await_analyst(self, state: WorkflowState) -> Dict[str, Any]: return {"analyst": dict(own), "escalation_id": ticket.id} async def _forget_benign_learning(self, state: WorkflowState) -> None: - """A threat on this alert: no cached verdict or suppression rule may close it, or its repeats, again.""" + """A threat on this alert: no cached verdict or suppression rule may close it, or its repeats, again, and the + alerts the models closed alone on its entities reopen (decision policy v3 safety net).""" alert = await self.deps.repos.alerts.get(state["tenant_id"], state["alert_id"]) await forget_benign_learning(self.deps.repos, self.deps.verdicts, state["tenant_id"], alert, actor=WORKFLOW_ACTOR) + await reopen_counted(self.deps.repos, state["tenant_id"], alert, actor=WORKFLOW_ACTOR, path="workflow", + escalation_id=state.get("escalation_id")) def _evidence(self, state: WorkflowState) -> str: """What authorises containment for this plan: ``analyst``, ``rules`` or ``llm_only``. diff --git a/src/lg_sotf/core/retro_reopen.py b/src/lg_sotf/core/retro_reopen.py new file mode 100644 index 00000000..000d5496 --- /dev/null +++ b/src/lg_sotf/core/retro_reopen.py @@ -0,0 +1,219 @@ +"""The safety net of decision policy v3 (FIX-3): a related threat reopens the alerts the models closed alone. + +At the autonomy tier ``auto_full_audit`` the models close every alert the +deep tier judged benign (:mod:`lg_sotf.domain.closure`); the forward gates of +FIX-1 and FIX-2 (thresholds, campaign check) are gone. Instead, when an alert +becomes a threat: + +* the workflow plans a response for it, or closes it malicious; +* it is escalated with a model verdict of malicious or suspicious (the deep + tier's ``malicious``, the fast tier's ``suspicious``, ``likely_threat`` or + ``confirmed_threat``); +* an analyst decides ``true_positive`` on it; +* a blind audit labels it ``malicious``; + +:func:`reopen_related_model_closes` reopens every alert of the same tenant +that the models closed alone (close reason ``ai_closed:*``) in the previous +:data:`REOPEN_WINDOW`, and that shares a host, user or IP with it. Each one +is **folded** into the threat's open escalation: attached to it as a +related alert and set to ``waiting_human`` under that case, with no new run +and no new escalation, so the analyst sees one case holding the related +alerts; the analyst's decision on the case settles it +(:func:`lg_sotf.services.folded.settle_folded`), and a later case of the +threat takes it over when the first one asked for more information +(:func:`refold`). Only when there is no open case (a response plan +without an escalation, or a case already decided) is it queued for a new +run instead, marked ``reopened`` so the policy never closes it on the +models' word again. Either way it is audited ``alert.reopened`` (reason +``related_threat``), with an ``alert.reopened`` event on its timeline. A single attack step the model +misjudged therefore resurfaces as soon as the attack shows itself anywhere. + +It is idempotent (an alert already reopened is no longer closed, so it is +not selected again) and tenant-scoped (every query runs in the tenant's +transaction). The callers count a failure in +``lgsotf_retro_reopen_failures_total`` and log it as an error; it never +fails the threat's own decision. + +IPs are matched whatever their role in the alert (source or destination): +the canonical alert does not keep the role, and a wider match only reopens +more. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any, List, Mapping, Optional, Sequence + +from lg_sotf.audit.logger import Actor, AuditLogger +from lg_sotf.db.errors import InvalidTransition, NotFound +from lg_sotf.db.repos import Repositories +from lg_sotf.observability.logging import get_logger, log_failure +from lg_sotf.observability.metrics import RETRO_REOPEN_FAILURES, RETRO_REOPENED +from lg_sotf.services.folded import FOLDED, REQUEUED, folded_into + +log = get_logger(__name__) + +REOPEN_WINDOW = timedelta(hours=24) +REOPEN_REASON = "related_threat" +REOPEN_ACTION = "alert.reopened" +#: The close reason a reopened alert carries until it closes again (it no longer reads ``ai_closed:*``). +REOPENED_CLOSE_REASON = "reopened:related_threat" +ACTIVE_ESCALATION_STATUSES = frozenset({"open", "claimed"}) +MAX_REOPEN = 200 +MAX_ENTITY_VALUES = 50 +#: The model verdicts that make an escalated alert a threat for the safety net. +THREAT_ASSESSMENTS = frozenset({"suspicious", "likely_threat", "confirmed_threat"}) + + +def judged_threat(triage: Mapping[str, Any], analysis: Mapping[str, Any]) -> bool: + """Whether a model judged the alert malicious or suspicious (the escalation path of the safety net).""" + return analysis.get("verdict") == "malicious" or triage.get("threat_assessment") in THREAT_ASSESSMENTS + + +def _values(entities: Any, kind: str) -> List[str]: + values = entities.get(kind) if isinstance(entities, Mapping) else None + if not isinstance(values, list): + return [] + return [v for v in values[:MAX_ENTITY_VALUES] if isinstance(v, str) and v] + + +def _closed_by_models(alert: Any) -> bool: + summary = alert.summary if isinstance(alert.summary, Mapping) else {} + reason = summary.get("close_reason") + return alert.status == "closed" and isinstance(reason, str) and reason.startswith("ai_closed:") + + +async def reopen_related_model_closes( + repos: Repositories, + tenant_id: str, + threat: Any, + *, + actor: Actor, + escalation_id: Optional[str] = None, + now: Optional[datetime] = None, +) -> List[str]: + """Reopen the model-closed alerts related to ``threat`` (a stored alert); the ids reopened (see the module docs). + + Raises on a database error: the caller counts it (:func:`reopen_counted`). + """ + now = now or datetime.now(UTC) + entities = threat.entities if isinstance(threat.entities, Mapping) else {} + candidates = await repos.alerts.model_closed_sharing_entities( + tenant_id, + hosts=_values(entities, "hosts"), + users=_values(entities, "users"), + ips=_values(entities, "ips"), + since=now - REOPEN_WINDOW, + exclude_alert_id=threat.id, + limit=MAX_REOPEN, + ) + reopened: List[str] = [] + for alert_id in candidates: + if await _reopen_one(repos, tenant_id, alert_id, threat_id=threat.id, actor=actor, escalation_id=escalation_id): + reopened.append(alert_id) + return reopened + + +async def _active_escalation(repos: Repositories, tenant_id: str, escalation_id: Optional[str], conn: Any) -> Any: + if escalation_id is None: + return None + try: + escalation = await repos.escalations.get(tenant_id, escalation_id, conn=conn) + except NotFound: + return None + return escalation if escalation.status in ACTIVE_ESCALATION_STATUSES else None + + +async def _reopen_one( + repos: Repositories, tenant_id: str, alert_id: str, *, threat_id: str, actor: Actor, escalation_id: Optional[str] +) -> bool: + async with repos.db.tenant_tx(tenant_id) as conn: + alert = await repos.alerts.get(tenant_id, alert_id, conn=conn) + if not _closed_by_models(alert): + return False # reopened already, or closed again by a person: nothing to do + # Folded into the threat's case when it is still open: one escalation holds the related alerts, no new run. + folded = await _attach(repos, tenant_id, escalation_id, alert_id, conn) + marker = { + "reason": REOPEN_REASON, "threat_alert_id": threat_id, "escalation_id": escalation_id, + "mode": FOLDED if folded else REQUEUED, + } + if folded: + await repos.alerts.update_status(tenant_id, alert_id, "waiting_human", conn=conn) + summary = {"reopened": marker, "close_reason": REOPENED_CLOSE_REASON, "escalation_id": escalation_id, + "escalation_attached": True} + else: + # No open case to fold into (a plan without an escalation, a decided one): a new run reviews it. + run = await repos.runs.upsert_queued(tenant_id, alert_id, requeue_terminal=True, conn=conn) + if run.status != "queued": + raise InvalidTransition("workflow run", alert_id, run.status, "queued") + await repos.alerts.update_status(tenant_id, alert_id, "queued", conn=conn) + summary = {"reopened": marker, "close_reason": REOPENED_CLOSE_REASON} + await repos.alerts.update(tenant_id, alert_id, disposition=None, summary=summary, conn=conn) # type: ignore[arg-type] + await repos.alert_events.append( + tenant_id, alert_id, REOPEN_ACTION, node="retro_reopen", actor_type=actor.type, actor_id=actor.id, + data=dict(marker), conn=conn, + ) + await AuditLogger(repos).record( + tenant_id, REOPEN_ACTION, "alert", alert_id, actor_type=actor.type, actor_id=actor.id, + details=dict(marker), conn=conn, + ) + return True + + +async def refold(repos: Repositories, tenant_id: str, threat_id: str, escalation_id: str) -> List[str]: + """Move the alerts folded for ``threat_id`` into a case that is no longer open (decided ``needs_more_info``) + onto the threat's new case ``escalation_id``; the ids moved. Called whenever the threat escalates.""" + moved: List[str] = [] + for alert_id in await repos.alerts.folded_for_threat(tenant_id, threat_id, exclude_escalation_id=escalation_id): + async with repos.db.tenant_tx(tenant_id) as conn: + alert = await repos.alerts.get(tenant_id, alert_id, conn=conn) + current = folded_into(alert) + if current is None or current == escalation_id or await _active_escalation(repos, tenant_id, current, conn): + continue + if not await _attach(repos, tenant_id, escalation_id, alert_id, conn): + continue + marker = {**(alert.summary.get("reopened") or {}), "escalation_id": escalation_id} + await repos.alerts.update(tenant_id, alert_id, summary={"reopened": marker, "escalation_id": escalation_id}, + conn=conn) + moved.append(alert_id) + return moved + + +async def _attach(repos: Repositories, tenant_id: str, escalation_id: Optional[str], alert_id: str, conn: Any) -> bool: + """Attach ``alert_id`` to the escalation if it is still open (the attach is guarded in SQL against a case + decided meanwhile); whether it is attached now.""" + escalation = await _active_escalation(repos, tenant_id, escalation_id, conn) + if escalation is None: + return False + deadline = escalation.sla_deadline or datetime.now(UTC) + REOPEN_WINDOW + attached = await repos.escalations.attach( + tenant_id, escalation.id, alert_id, priority=escalation.priority, level=escalation.level, + sla_deadline=deadline, conn=conn, + ) + return attached is not None + + +async def reopen_counted( + repos: Repositories, + tenant_id: str, + threat: Any, + *, + actor: Actor, + path: str, + escalation_id: Optional[str] = None, +) -> Sequence[str]: + """:func:`reopen_related_model_closes` for a threat path: a failure is counted and logged as an error, never + raised (the threat's own decision stands); returns the ids reopened.""" + try: + reopened = await reopen_related_model_closes( + repos, tenant_id, threat, actor=actor, escalation_id=escalation_id + ) + except Exception as exc: # noqa: BLE001 - counted and logged as an error: an alerting rule can page on it + RETRO_REOPEN_FAILURES.labels(path=path).inc() + log_failure(log, "retro_reopen.failed", exc, level="error", alert_id=getattr(threat, "id", None), + path=path, error_type=type(exc).__name__) + return [] + if reopened: + RETRO_REOPENED.labels(path=path).inc(len(reopened)) + log.info("retro_reopen.reopened", alert_id=threat.id, path=path, count=len(reopened)) + return reopened diff --git a/src/lg_sotf/core/state.py b/src/lg_sotf/core/state.py index 3a6f9afb..c0d12550 100644 --- a/src/lg_sotf/core/state.py +++ b/src/lg_sotf/core/state.py @@ -239,6 +239,8 @@ class AlertSummary(TypedDict, total=False): # The outcome (close node). verdict: str close_reason: str + # Decision policy v3 safety net (lg_sotf.core.retro_reopen): why a model-closed alert was reopened. + reopened: Dict[str, Optional[str]] confidence: Optional[float] threat_score: Optional[float] decision: Optional[DecisionDigest] diff --git a/src/lg_sotf/core/workflow.py b/src/lg_sotf/core/workflow.py index f2428730..2c1dda6c 100644 --- a/src/lg_sotf/core/workflow.py +++ b/src/lg_sotf/core/workflow.py @@ -65,6 +65,7 @@ from lg_sotf.security.hashing import indicator_hmac_key from lg_sotf.security.shadow import REFUSE, egress from lg_sotf.services.escalation import EscalationService +from lg_sotf.services.folded import settle_folded from lg_sotf.services.response import Connector, ResponseService from lg_sotf.settings import Settings @@ -377,7 +378,12 @@ async def resume(self, tenant_id: str, alert_id: str, payload: Mapping[str, Any] if pending is None: if self._unfinished(snapshot): return await self._continue(graph, tenant_id, alert_id, snapshot, config) - if snapshot.values: + # An alert retro-reopen folded into this case has no run waiting: the decision settles it. + folded = data["type"] == "analyst_decision" and await settle_folded( + self.repos, tenant_id, alert_id, escalation_id=data["escalation_id"], + decision=data["decision"], decided_by=data["decided_by"], + ) + if snapshot.values or folded: alert = await self.repos.alerts.get(tenant_id, alert_id) return RunOutcome("completed", alert.status, None) raise InvalidResumePayload("there is no workflow run to resume for this alert") diff --git a/src/lg_sotf/db/repos/alerts.py b/src/lg_sotf/db/repos/alerts.py index cc5b1364..5ccbfeb5 100644 --- a/src/lg_sotf/db/repos/alerts.py +++ b/src/lg_sotf/db/repos/alerts.py @@ -48,7 +48,9 @@ "processing": frozenset({"waiting_human", "waiting_approval", "closed", "failed"}), "waiting_human": frozenset({"processing"}), "waiting_approval": frozenset({"processing"}), - "closed": frozenset({"queued"}), + # closed -> waiting_human: decision policy v3 folds a model-closed alert into a related threat's open case + # (lg_sotf.core.retro_reopen). + "closed": frozenset({"queued", "waiting_human"}), "failed": frozenset({"processing", "queued"}), } @@ -517,6 +519,66 @@ async def response_times( resolved=row["resolved"], ) + async def model_closed_sharing_entities( + self, + tenant_id: str, + *, + hosts: Sequence[str], + users: Sequence[str], + ips: Sequence[str], + since: datetime, + exclude_alert_id: str, + limit: int = 200, + conn: Optional[Connection] = None, + ) -> List[str]: + """Ids of alerts the models closed alone (close reason ``ai_closed:*``) since ``since`` that share a host, + user or IP with the given ones, newest first (the retro-reopen of decision policy v3).""" + require_aware(since, "since") + check_limit(limit, 1_000) + if not (hosts or users or ips): + return [] + async with self._tenant(tenant_id, conn) as c: + rows = await c.fetch( + """ + SELECT id FROM alerts + WHERE tenant_id = $1 AND status = 'closed' AND id <> $2 AND closed_at >= $3 + AND starts_with(summary->>'close_reason', 'ai_closed:') + AND (COALESCE(normalized->'entities'->'hosts', '[]'::jsonb) ?| $4::text[] + OR COALESCE(normalized->'entities'->'users', '[]'::jsonb) ?| $5::text[] + OR COALESCE(normalized->'entities'->'ips', '[]'::jsonb) ?| $6::text[]) + ORDER BY closed_at DESC, id + LIMIT $7 + """, + tenant_id, + exclude_alert_id, + since, + list(hosts), + list(users), + list(ips), + limit, + ) + return [str(r["id"]) for r in rows] + + async def folded_for_threat( + self, tenant_id: str, threat_alert_id: str, *, exclude_escalation_id: str, conn: Optional[Connection] = None + ) -> List[str]: + """Alerts retro-reopen folded for ``threat_alert_id`` that still wait in another case than the given one.""" + async with self._tenant(tenant_id, conn) as c: + rows = await c.fetch( + """ + SELECT id FROM alerts + WHERE tenant_id = $1 AND status = 'waiting_human' + AND summary->'reopened'->>'mode' = 'folded' + AND summary->'reopened'->>'threat_alert_id' = $2 + AND summary->>'escalation_id' IS DISTINCT FROM $3 + ORDER BY id LIMIT 200 + """, + tenant_id, + threat_alert_id, + exclude_escalation_id, + ) + return [str(r["id"]) for r in rows] + # -- recovery sweeps (runtime/orphans.py) ------------------------------------------------------ async def find_stranded_new_all_tenants(self, *, older_than_s: float, limit: int) -> List[StrandedAlertRef]: diff --git a/src/lg_sotf/db/repos/escalations.py b/src/lg_sotf/db/repos/escalations.py index 03f57546..8afe3d1f 100644 --- a/src/lg_sotf/db/repos/escalations.py +++ b/src/lg_sotf/db/repos/escalations.py @@ -71,7 +71,7 @@ sla_deadline = CASE WHEN sla_deadline IS NULL THEN $6 ELSE LEAST(sla_deadline, $6) END, summary = summary || jsonb_build_object('attached_count', cardinality(related_alert_ids) + 1), version = version + 1 - WHERE tenant_id = $1 AND id = $2 AND NOT ($3 = ANY(related_alert_ids)) + WHERE tenant_id = $1 AND id = $2 AND NOT ($3 = ANY(related_alert_ids)) AND status IN ('open', 'claimed') RETURNING * """ @@ -472,7 +472,8 @@ async def attach( sla_deadline: datetime, conn: Connection, ) -> Optional[Escalation]: - """Add ``alert_id`` to the escalation's ``related_alert_ids``; ``None`` if it is already there. + """Add ``alert_id`` to the escalation's ``related_alert_ids``; ``None`` if it is already there or the case is + no longer open or claimed (decided meanwhile). Priority only rises, the SLA deadline only shortens, ``level`` is set as given. """ diff --git a/src/lg_sotf/domain/closure.py b/src/lg_sotf/domain/closure.py index 34ffeb2b..3f83cc28 100644 --- a/src/lg_sotf/domain/closure.py +++ b/src/lg_sotf/domain/closure.py @@ -1,54 +1,57 @@ -"""Model-led closure (decision policy v2.1, FIX-1 and FIX-2): the tenant's autonomy tier and when a model may close. - -Without approved benign evidence (:func:`lg_sotf.domain.routing.decide`), a -low or medium alert is closed on the models' word only when the tenant chose -the autonomy tier ``auto_full_audit`` (tenant setting ``ai_closure.tier``, -default ``off``) and every condition below holds. Each one fails closed: a -missing or unreadable value blocks the close. - -* severity ``info``, ``low`` or ``medium`` (high and critical never close on a - model's word, at any tier); -* not a never-close class (:func:`never_close_class`), by the ATT&CK tactic - or technique the detection rule declares: credential access, log clearing - and other indicator removal, security-tool tampering, ransomware - behaviour, and persistence through an object the attacker creates and - names (scheduled tasks, services, autostart entries, WMI subscriptions, - accounts), whose alert text is the attacker's. An alert the rule maps to - no tactic or technique is ``unmapped`` and counts as a never-close class: - nothing shows it is outside them. The one exception is a detection rule in - the reviewed noise catalogue (:mod:`lg_sotf.domain.noise_families`, - FIX-2), keyed by vendor and rule id, never by title or description text: - it lifts ``unmapped`` and nothing else (:func:`model_close_class`); +"""Model-led closure (decision policy v3, FIX-3): the tenant's autonomy tier and when the models close an alert. + +Founder decision 2026-11-09: every alert the AI judges benign closes, and a +backward safety net replaces the forward gates of FIX-1 and FIX-2. Without +approved benign evidence (:func:`lg_sotf.domain.routing.decide`), an alert +closes on the models' word only when the tenant chose the autonomy tier +``auto_full_audit`` (tenant setting ``ai_closure.tier``, default ``off``) +and every condition below holds. Each one fails closed: a missing or +unreadable value blocks the close. + +* the deep tier (analysis) says ``benign``, at any certainty; a verdict the + rules produced because a model was not consulted never counts; +* the fast tier (triage) did not call it a threat (``likely_threat`` or + ``confirmed_threat``); +* severity ``info``, ``low``, ``medium`` or ``high`` (critical never closes + on a model's word, at any tier); +* not a never-close class (:func:`model_close_class`): credential access by + ATT&CK tactic or technique (T1531 included) or by the vendor's own + authentication-failure rule groups and ids + (:func:`lg_sotf.domain.noise_families.vendor_never_close_class`), a Wazuh + alert whose rule cannot be read (``vendor_unknown``), log clearing and + other indicator removal, security-tool tampering, ransomware behaviour, + and persistence through an object the attacker creates and names. An alert + without an ATT&CK mapping (``unmapped``) is no longer blocked; * no steering text: triage did not flag ``prompt_injection_suspected``; * no recent true positive on the alert's entities: no analyst-confirmed malicious history (``prior_malicious_history``), no related alert of the correlation window closed as malicious, and no rule evidence that corroborates a threat (``rule_corroborated``); -* no campaign in progress: no burst of related alerts, no related alert a - model judged malicious or suspicious, none still open for an analyst or an - approval (counted over every related alert, not only those shown to the - correlation model), so an attack is never closed alert by alert. (A - check that lets one rule repeat was deferred from FIX-2: three review - rounds found ways to launder a block through it); -* both model tiers say benign: the fast tier's assessment is ``benign`` with - a threat likelihood at most ``routing.model_close_max_likelihood`` - (default 8), and the deep tier's verdict is ``benign`` with a certainty of - at least ``routing.model_close_min_certainty`` (default 85). A verdict the - rules produced because a model was not consulted never counts. +* not normalised by a model (the LLM-fallback normalizer chose its fields), + and not reopened by the safety net (``reopened``): a reopened alert goes + to an analyst. + +Removed in v3: the install thresholds on the model tiers (likelihood 8, +certainty 85), the campaign check and the ``unmapped`` block. + +The safety net (:mod:`lg_sotf.core.retro_reopen`): when an alert becomes a +threat (a response plan, an escalation with a malicious or suspicious model +verdict, an analyst true positive, a blind-audit malicious label), every +model-closed alert of the tenant that shares a host, user or IP with it and +closed in the previous 24 hours is reopened, attached to that escalation and +audited ``alert.reopened``. A model-led close carries the decision code ``model_closed`` and the close reason ``ai_closed:audit_required``: an analyst is expected to audit it. -When the noise catalogue lifted ``unmapped``, the decision reason names the -family and the catalogue version. -Correlation shows such an alert to later models as ``ai_closed_unaudited``, -never as ``benign``, so one persuaded close cannot vouch for the next. +When the noise catalogue knows the detection rule, the decision reason names +the family and the catalogue version. Correlation shows such an alert to +later models as ``ai_closed_unaudited``, never as ``benign``, so one +persuaded close cannot vouch for the next. Approved benign evidence closes without the tier, but never past steering text or a recent true positive, and context evidence (a declared scanner, a test source) never past a mapped never-close class (:func:`evidence_close_blockers`). -The thresholds are install settings (``routing``), so a tenant can turn -model-led closure on or off but never loosen what it needs. """ from __future__ import annotations @@ -67,11 +70,13 @@ "AutonomyTier", "CLOSABLE_SEVERITIES", "CONTEXT_EVIDENCE", + "FAST_TIER_THREATS", "MODEL_CLOSED", "MODEL_CLOSE_REASON", "MODEL_CLOSE_TIER", "NEVER_CLOSE_TACTICS", "NEVER_CLOSE_TECHNIQUES", + "REOPENED", "SETTING", "UNMAPPED", "autonomy_tier", @@ -92,7 +97,12 @@ #: The decision code and the close reason of a model-led close. MODEL_CLOSED = "model_closed" MODEL_CLOSE_REASON = "ai_closed:audit_required" -CLOSABLE_SEVERITIES: FrozenSet[str] = frozenset({"info", "low", "medium"}) +CLOSABLE_SEVERITIES: FrozenSet[str] = frozenset({"info", "low", "medium", "high"}) +#: Fast-tier assessments that block a model-led close. +FAST_TIER_THREATS: FrozenSet[str] = frozenset({"likely_threat", "confirmed_threat"}) +#: The alert key the safety net sets on a reopened alert (lg_sotf.core.retro_reopen); it never model-closes. +REOPENED = "reopened" +LLM_NORMALIZED_FLAG = "llm_normalized" #: Triage's rule false-positive signals that say where the activity came from (a scanner the tenant declared, a #: test-environment source), as opposed to what was detected (the rule's own test intent, the EICAR file). CONTEXT_EVIDENCE: FrozenSet[str] = frozenset({"known_scanner", "test_environment"}) @@ -203,36 +213,22 @@ def catalogued_noise(alert: Any) -> Optional[NoiseFamily]: def model_close_class(alert: Any) -> Optional[str]: - """The never-close class that blocks a model-led close: :func:`never_close_class`, except that a detection rule - in the noise catalogue lifts ``unmapped`` (and only ``unmapped``).""" + """The never-close class that blocks a model-led close: :func:`never_close_class`, except that ``unmapped`` + does not block it (policy v3). ``vendor_unknown`` and every mapped class do.""" found = never_close_class(alert) - if found == UNMAPPED and catalogued_noise(alert) is not None: - return None - return found + return None if found == UNMAPPED else found def _number(value: Any) -> Optional[float]: return float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else None -def _fast_tier_benign(triage: Mapping[str, Any], max_likelihood: int) -> bool: - likelihood = _number(triage.get("threat_likelihood")) - return ( - triage.get("llm_used") is True - and triage.get("threat_assessment") == "benign" - and likelihood is not None - and likelihood <= max_likelihood - ) +def _fast_tier_threat(triage: Mapping[str, Any]) -> bool: + return triage.get("threat_assessment") in FAST_TIER_THREATS -def _deep_tier_benign(analysis: Mapping[str, Any], min_certainty: int) -> bool: - certainty = _number(analysis.get("certainty")) - return ( - analysis.get("llm_used") is True - and analysis.get("verdict") == "benign" - and certainty is not None - and certainty >= min_certainty - ) +def _deep_tier_benign(analysis: Mapping[str, Any]) -> bool: + return analysis.get("llm_used") is True and analysis.get("verdict") == "benign" def _recent_true_positive(analysis: Mapping[str, Any], correlation: Mapping[str, Any]) -> bool: @@ -246,20 +242,6 @@ def _recent_true_positive(analysis: Mapping[str, Any], correlation: Mapping[str, ) -def _campaign_in_progress(correlation: Mapping[str, Any]) -> bool: - """Related alerts that may be one attack: a burst, one a model judged malicious or suspicious, or one still - open for review. Unknown counts (a digest without them) fail closed.""" - related = _number(correlation.get("related_count")) - if related is None: - return True - if related == 0: - return False - if correlation.get("burst") is not False: - return True - counts = [_number(correlation.get(key)) for key in ("related_threats", "related_open_reviews")] - return any(count is None or count > 0 for count in counts) - - def _steering_text(triage: Mapping[str, Any]) -> bool: rule_tp = triage.get("rule_tp") return not isinstance(rule_tp, list) or INJECTION_SIGNAL in rule_tp @@ -294,19 +276,24 @@ def evidence_close_blockers(state: Mapping[str, Any]) -> List[str]: return [name for name, blocked in checks if blocked] -def model_close_blockers( - state: Mapping[str, Any], *, tier: str, max_likelihood: int, min_certainty: int -) -> List[str]: +def _flagged(alert: Any, flag: str) -> bool: + flags = alert.get("flags") if isinstance(alert, Mapping) else None + return isinstance(flags, list) and flag in flags + + +def model_close_blockers(state: Mapping[str, Any], *, tier: str) -> List[str]: """Why the models may not close this alert on their own; empty when they may (see the module docs).""" triage, analysis, correlation = _digest(state, "triage"), _digest(state, "analysis"), _digest(state, "correlation") + alert = state.get("alert") checks = ( ("tier_off", tier != MODEL_CLOSE_TIER), ("severity", state.get("severity") not in CLOSABLE_SEVERITIES), - ("never_close_class", model_close_class(state.get("alert")) is not None), + ("never_close_class", model_close_class(alert) is not None), + ("llm_normalized", _flagged(alert, LLM_NORMALIZED_FLAG)), + ("reopened", not isinstance(alert, Mapping) or alert.get(REOPENED) is not None), ("steering_text", _steering_text(triage)), ("recent_true_positive", _recent_true_positive(analysis, correlation)), - ("campaign_in_progress", _campaign_in_progress(correlation)), - ("fast_tier_not_benign", not _fast_tier_benign(triage, max_likelihood)), - ("deep_tier_not_benign", not _deep_tier_benign(analysis, min_certainty)), + ("fast_tier_threat", _fast_tier_threat(triage)), + ("deep_tier_not_benign", not _deep_tier_benign(analysis)), ) return [name for name, blocked in checks if blocked] diff --git a/src/lg_sotf/domain/routing.py b/src/lg_sotf/domain/routing.py index 90a50548..36906201 100644 --- a/src/lg_sotf/domain/routing.py +++ b/src/lg_sotf/domain/routing.py @@ -2,8 +2,10 @@ Safety rules enforced here (contract section 8): -* Automation never closes a high/critical alert: only an analyst decision, - a suppression rule or an analyst-confirmed verdict (not for critical) does. +* Automation never closes a critical alert: only an analyst decision or a + suppression rule does. A high alert closes on approved evidence or a cache + only through an analyst-confirmed verdict, and on the models' word only at + the tenant's autonomy tier ``auto_full_audit`` (policy v3, FIX-3). * Decision policy v2 (FIX-1). Every low/medium alert is analysed: correlation never skips analysis for it, and correlation is evidence for the model, not a confidence boost. The grey zone and the response policy read the @@ -18,9 +20,12 @@ tenant's autonomy tier is ``auto_full_audit``, through a model-led close (:mod:`lg_sotf.domain.closure`, code ``model_closed``) that every condition of that module allows. With the tier ``off`` a model never closes alone. - Policy v2.1 (FIX-2) lets a catalogued noise rule (vendor and rule id, - :mod:`lg_sotf.domain.noise_families`) lift only the ``unmapped`` class; - the campaign check is FIX-1's. + Policy v3 (FIX-3): at that tier the models close every alert the deep + tier judged benign (info to high severity, any certainty or threat score + below the response policy) unless a hard block applies; the thresholds, + the campaign check and the ``unmapped`` block are gone, and the retro-reopen + safety net (:mod:`lg_sotf.core.retro_reopen`) reopens such closes when a + related threat appears. * A response the policy (not an analyst) chose goes to an analyst when an action failed or none completed; it closes (disposition ``malicious``) once its actions succeeded or ran in the tenant's dry-run mode. Blocked @@ -90,10 +95,6 @@ class RoutingThresholds(BaseModel): close_max_threat: int = 30 max_more_info_loops: int = 1 max_alert_age_hours: int = 72 - # Model-led close (lg_sotf.domain.closure): the fast tier's likelihood at most this, the deep tier's - # certainty at least this. Install-wide on purpose: a tenant chooses the tier, never looser numbers. - model_close_max_likelihood: int = 8 - model_close_min_certainty: int = 85 @classmethod def from_settings(cls, routing: Optional[Mapping[str, Any]]) -> RoutingThresholds: @@ -231,19 +232,34 @@ def decide(state: Mapping[str, Any], t: RoutingThresholds, tier: str = "off") -> "plan_response", f"threat score {threat:.0f} meets the response policy", "malicious", "high_risk" ) - if verdict == "inconclusive" or t.correlation_grey_zone_min <= threat <= t.correlation_grey_zone_max: + grey = verdict == "inconclusive" or t.correlation_grey_zone_min <= threat <= t.correlation_grey_zone_max + low = None if grey or threat >= t.close_max_threat else _low_threat(state, triage, verdict) + if low is not None and low.route == "close": + return low # approved benign evidence closes first: it needs no audit + model = _model_close(state, tier) + if model is not None: + return model + if grey: return Decision("escalate", "assessment in the grey zone", None, "grey_zone") + if low is not None: + return low + return Decision("escalate", "no automatic policy applies", None, "inconclusive") - if threat < t.close_max_threat: - return _low_threat(state, triage, verdict, t, tier) - return Decision("escalate", "no automatic policy applies", None, "inconclusive") +def _model_close(state: Mapping[str, Any], tier: str) -> Optional[Decision]: + """Decision policy v3 (FIX-3): the models close an alert the deep tier judged benign, at the tenant's tier, + whatever the assessed threat score or certainty, unless a hard block applies (:mod:`lg_sotf.domain.closure`).""" + if closure.model_close_blockers(state, tier=tier): + return None + reason = "AI-closed, audit required: the analysis judged it benign" + family = closure.catalogued_noise(state.get("alert")) + if family is not None: + reason += f" (catalogued noise {family.family}, catalogue {noise_families.CATALOGUE_VERSION})" + return Decision("close", reason, "benign", closure.MODEL_CLOSED) -def _low_threat( - state: Mapping[str, Any], triage: Mapping[str, Any], verdict: Optional[str], t: RoutingThresholds, tier: str -) -> Decision: - """A low threat: close on approved benign evidence or a model-led close the tier allows, else an analyst.""" +def _low_threat(state: Mapping[str, Any], triage: Mapping[str, Any], verdict: Optional[str]) -> Decision: + """A low threat: close on approved benign evidence, else an analyst (the model-led close is tried after).""" if _is_high(state): return Decision("escalate", "high severity alerts are closed only by an analyst", None, "false_positive_review") if llm_normalized(state): @@ -258,15 +274,6 @@ def _low_threat( reason = f"approved benign evidence overruled by {', '.join(overruled)}" return Decision("escalate", reason, None, "false_positive_review") return Decision("close", "low threat corroborated by approved benign evidence", "benign", "rule_corroborated_benign") - blockers = closure.model_close_blockers( - state, tier=tier, max_likelihood=t.model_close_max_likelihood, min_certainty=t.model_close_min_certainty - ) - if not blockers: - reason = "AI-closed, audit required: both model tiers judged it benign" - family = closure.catalogued_noise(state.get("alert")) - if family is not None: - reason += f" (catalogued noise {family.family}, catalogue {noise_families.CATALOGUE_VERSION})" - return Decision("close", reason, "benign", closure.MODEL_CLOSED) return Decision("escalate", "low threat verdict lacks approved benign evidence", None, "false_positive_review") diff --git a/src/lg_sotf/invariants/checks.py b/src/lg_sotf/invariants/checks.py index 32c86dda..f1970077 100644 --- a/src/lg_sotf/invariants/checks.py +++ b/src/lg_sotf/invariants/checks.py @@ -129,14 +129,16 @@ def _benign_state(severity: str, *, evidence: bool = True, tactic: Optional[str] def check_routing_policy(settings: Settings) -> CheckResult: - """With this install's thresholds, a benign verdict never closes a high alert at any autonomy tier, a model - never closes alone with the tier ``off``, and never a never-close class (INV-03).""" + """With this install's thresholds, a benign verdict never closes a critical alert at any autonomy tier nor a + high one with the tier ``off`` (policy v3), a model never closes alone with the tier ``off``, and never a + never-close class (INV-03).""" thresholds = RoutingThresholds.from_settings(settings.routing) closed: List[str] = [] for severity in sorted(HIGH_SEVERITIES): state = _benign_state(severity) + tiers = AUTONOMY_TIERS if severity == "critical" else ("off",) if route_after_triage(state, thresholds) == "close" or any( - decide(state, thresholds, tier).route == "close" for tier in AUTONOMY_TIERS + decide(state, thresholds, tier).route == "close" for tier in tiers ): closed.append(f"{severity} alert closed on a benign verdict") if decide(_benign_state("low", evidence=False), thresholds, "off").route == "close": @@ -151,8 +153,8 @@ def check_routing_policy(settings: Settings) -> CheckResult: return _result( "routing_policy", not closed, - "; ".join(closed) or "a benign model verdict closes no high or critical alert, nothing with model-led closure " - "off, no never-close class, and no alert with evidence of a threat", + "; ".join(closed) or "a benign model verdict closes no critical alert, no high alert and nothing else with " + "model-led closure off, no never-close class, and no alert with evidence of a threat", ) diff --git a/src/lg_sotf/invariants/registry.py b/src/lg_sotf/invariants/registry.py index 54a1c284..a44b92bc 100644 --- a/src/lg_sotf/invariants/registry.py +++ b/src/lg_sotf/invariants/registry.py @@ -102,23 +102,28 @@ class Invariant: Invariant( id="INV-03", title=( - "An AI verdict alone never closes a high or critical alert or a never-close class; a low or medium " - "alert closes on the models' word only at the tenant's autonomy tier" + "An AI verdict alone never closes a critical alert or a never-close class; an alert closes on the " + "models' word only at the tenant's autonomy tier, and a related threat reopens it" ), rationale=( "A fooled or injected model must not be able to make a dangerous alert disappear. Routing is a pure " - "policy. High and critical alerts close only by an analyst, a suppression rule or an analyst-confirmed " - "verdict. A low or medium alert closes when approved benign evidence an attacker cannot write " + "policy. A critical alert closes only by an analyst or a suppression rule. A low or medium alert " + "closes when approved benign evidence an attacker cannot write " "corroborates a benign verdict: a cached analyst verdict, the detection rule's name, the EICAR hash, a " "test source or the tenant's declared context, such as a scanner as the source of the activity; never " "past steering text or a recent true positive, and a scanner or test source never for a never-close " - "class. Without such evidence it closes on the models' word only when the tenant's autonomy tier is " - "auto_full_audit and both model tiers judge it benign within the install's thresholds, never for the " - "never-close classes (credential access, log clearing, security-tool tampering, ransomware behaviour, " - "persistence through attacker-named objects, authentication failures by vendor rule group, unmapped " - "alerts unless the vendor rule id is in the reviewed noise catalogue, never matched on text), steering " - "text, entities with a recent true positive or a campaign in progress; such a close is marked " - "model_closed for audit. With the tier off, the default, a model never closes an alert alone. A cached " + "class. Without such evidence an info to high alert closes on the models' word only when the tenant's " + "autonomy tier is auto_full_audit, the deep tier judges it benign and the fast tier does not call it a " + "threat (decision policy v3), never for the never-close classes (credential access, including the " + "vendor's authentication-failure rules and T1531, a Wazuh alert whose rule cannot be read, log " + "clearing, security-tool tampering, ransomware behaviour, persistence through attacker-named objects), " + "steering text, entities with a recent true positive, an alert normalised by a model or one reopened; " + "such a close is marked model_closed for audit. When an alert becomes a threat (a response plan, an " + "escalation a model judged malicious or suspicious, an analyst true positive, a blind-audit malicious " + "label), every model-closed alert sharing a host, user or IP with it in the previous 24 hours is " + "reopened: folded into its open case as a related alert, or queued for review without one, and " + "audited. With the tier off, the default, a model never " + "closes an alert alone. A cached " "analyst verdict closes only a " "non-critical repeat of the same detection and behaviour (verdict key: what ran, compared in full) " "that two different analysts decided benign, with no threat on it since; alerts in a never-close " @@ -134,6 +139,7 @@ class Invariant: "src/lg_sotf/domain/routing.py", "src/lg_sotf/domain/closure.py", "src/lg_sotf/domain/benign_evidence.py", "src/lg_sotf/domain/noise_families.py", "src/lg_sotf/domain/verdict_key.py", "src/lg_sotf/core/verdict_cache.py", "src/lg_sotf/core/benign_learning.py", + "src/lg_sotf/core/retro_reopen.py", ), proofs=( "tests/unit/domain/test_routing.py::test_llm_only_evidence_never_closes_high_or_critical_at_triage", @@ -141,12 +147,26 @@ class Invariant: "tests/unit/domain/test_routing.py::test_decide_closes_only_with_approved_benign_evidence", "tests/unit/domain/test_routing.py::test_attacker_controlled_fp_words_never_close_high_severity", "tests/unit/domain/test_closure.py::test_with_the_tier_off_a_model_never_closes_alone", - "tests/unit/domain/test_closure.py::test_high_and_critical_never_close_at_any_tier", + "tests/unit/domain/test_closure.py::" + "test_critical_never_closes_at_any_tier_and_high_only_on_the_models_word_at_the_tier", "tests/unit/domain/test_closure.py::test_never_close_classes_never_close_on_the_models_word", "tests/unit/domain/test_closure.py::test_each_condition_alone_blocks_a_model_led_close", "tests/unit/domain/test_closure.py::test_approved_benign_evidence_never_closes_past_a_hard_flag", "tests/unit/domain/test_closure.py::test_a_scanner_or_test_source_never_closes_a_never_close_class", - "tests/unit/domain/test_closure.py::test_a_campaign_in_progress_is_never_model_closed", + "tests/unit/domain/test_closure.py::" + "test_an_unmapped_alert_closes_on_the_models_word_but_an_unreadable_wazuh_rule_does_not", + "tests/integration/engine/test_retro_reopen.py::" + "test_an_escalation_a_model_judged_a_threat_reopens_related_model_closes", + "tests/integration/engine/test_retro_reopen.py::" + "test_a_response_plan_and_the_malicious_close_each_reopen_related_model_closes", + "tests/integration/engine/test_retro_reopen.py::test_a_reopen_failure_is_counted_not_swallowed", + "tests/integration/api/test_retro_reopen_api.py::" + "test_the_case_decision_settles_folded_alerts_in_its_own_transaction", + "tests/integration/engine/test_review_349.py::test_worker_delivers_the_case_decision_to_a_folded_alert", + "tests/integration/engine/test_review_349.py::test_needs_more_info_strands_a_folded_alert", + "tests/integration/api/test_retro_reopen_api.py::test_an_analyst_true_positive_reopens_related_model_closes", + "tests/integration/api/test_retro_reopen_api.py::" + "test_a_malicious_blind_audit_label_reopens_related_model_closes", "tests/unit/domain/test_closure.py::test_review_f2_sentinel_credential_access_is_a_never_close_class", "tests/unit/domain/test_noise_families.py::test_a_forged_or_borrowed_title_on_an_uncatalogued_rule_stays_unmapped", "tests/unit/domain/test_noise_families.py::test_the_catalogue_lifts_only_unmapped_never_a_mapped_never_close_class", @@ -158,8 +178,8 @@ class Invariant: "test_review_h1_authentication_failures_are_credential_access_whatever_their_mapping", "tests/integration/engine/test_model_closure.py::test_the_tenant_tier_decides_whether_the_models_close", "tests/integration/engine/test_model_closure.py::" - "test_catalogued_noise_closes_on_the_models_word_and_an_uncatalogued_rule_does_not", - "tests/integration/eval/test_adversarial_v2_eval.py::test_no_malicious_close_with_model_led_closure_on", + "test_catalogued_noise_is_named_and_an_unreadable_wazuh_rule_does_not_close", + "tests/integration/eval/test_adversarial_v2_eval.py::test_model_led_closure_on_closes_only_what_policy_v3_allows", "tests/integration/e2e/test_llm_safety.py::test_injected_benign_instructions_never_close_an_alert", "tests/unit/agents/test_title_camouflage.py::" "test_the_title_camouflage_items_get_no_benign_evidence_and_never_close_on_rules", diff --git a/src/lg_sotf/models/alert.py b/src/lg_sotf/models/alert.py index e4d5fed8..f2cebfd0 100644 --- a/src/lg_sotf/models/alert.py +++ b/src/lg_sotf/models/alert.py @@ -235,7 +235,7 @@ def _cap(cls, v: List[str]) -> List[str]: #: Canonical alert keys that are platform bookkeeping, not alert content: left out of what a model reads. -MODEL_HIDDEN_ALERT_KEYS = frozenset({"normalizer"}) +MODEL_HIDDEN_ALERT_KEYS = frozenset({"normalizer", "reopened"}) def model_view(alert: Mapping[str, Any]) -> Dict[str, Any]: diff --git a/src/lg_sotf/observability/metrics.py b/src/lg_sotf/observability/metrics.py index ae3902e2..0924ff5c 100644 --- a/src/lg_sotf/observability/metrics.py +++ b/src/lg_sotf/observability/metrics.py @@ -107,6 +107,18 @@ ["path"], registry=REGISTRY, ) +RETRO_REOPEN_FAILURES = Counter( + "lgsotf_retro_reopen_failures_total", + "Threats whose related model-closed alerts could not all be reopened (decision policy v3 safety net), by path.", + ["path"], + registry=REGISTRY, +) +RETRO_REOPENED = Counter( + "lgsotf_retro_reopened_total", + "Model-closed alerts reopened because a related alert became a threat, by path.", + ["path"], + registry=REGISTRY, +) ENRICHMENT_LOOKUPS = Counter( "lgsotf_enrichment_lookups_total", "Indicator enrichment lookups, by provider and result status (cache hits are status 'cached').", diff --git a/src/lg_sotf/runtime/worker.py b/src/lg_sotf/runtime/worker.py index f0e14136..182b753c 100644 --- a/src/lg_sotf/runtime/worker.py +++ b/src/lg_sotf/runtime/worker.py @@ -60,6 +60,7 @@ from lg_sotf.queue.streams import Message from lg_sotf.runtime.shutdown import cancel_and_wait, wait_or_stop from lg_sotf.security.redaction import safe_error_text +from lg_sotf.services.folded import settle_folded from lg_sotf.settings import Settings log = get_logger(__name__) @@ -489,6 +490,13 @@ async def _record_outcome(self, msg: Message, status: str, metered: bool, conn: await self._repos.usage.add_processed(msg.tenant_id, msg.alert_id, conn=conn) async def _handle_unclaimable(self, msg: Message, exc: RunNotClaimable) -> None: + payload = msg.payload or {} + if exc.reason == "completed" and msg.kind == "resume" and payload.get("type") == "analyst_decision": + # An alert retro-reopen folded into a case has no run waiting (lg_sotf.services.folded); idempotent. + await settle_folded( + self._repos, msg.tenant_id, msg.alert_id, escalation_id=str(payload.get("escalation_id")), + decision=str(payload.get("decision")), decided_by=str(payload.get("decided_by") or ""), + ) if exc.reason in TERMINAL_REASONS: await self._consumer.ack(msg) self.stats.skipped += 1 diff --git a/src/lg_sotf/services/folded.py b/src/lg_sotf/services/folded.py new file mode 100644 index 00000000..8d729b7f --- /dev/null +++ b/src/lg_sotf/services/folded.py @@ -0,0 +1,90 @@ +"""Alerts decision policy v3 folded into a related threat's case (lg_sotf.core.retro_reopen): the case decision +settles them. + +A folded alert was closed by the models, then reopened and attached to the +threat's open escalation as a related alert, with ``waiting_human`` status +and **no run waiting** (its run had completed). The analyst's decision on the +case therefore cannot reach it through a workflow resume: the review service +applies it in the decision's own transaction (:func:`settle_folded`), and the +worker applies it again, idempotently, if a resume message for it arrives. +``true_positive`` closes it ``malicious``, a benign decision ``benign``; +``needs_more_info`` leaves it waiting, and the threat's next case takes it +over (:func:`lg_sotf.core.retro_reopen.refold`). +""" + +from __future__ import annotations + +from typing import Any, Mapping, Optional, Set + +from lg_sotf.audit.logger import AuditLogger +from lg_sotf.db import Connection +from lg_sotf.db.repos import Repositories +from lg_sotf.models.vocabulary import BENIGN_DECISIONS + +#: How a reopened alert came back: folded into the threat's open case, or queued for its own review run. +FOLDED, REQUEUED = "folded", "requeued" +SETTLED_ACTION = "alert.folded_settled" + + +def folded_into(alert: Any) -> Optional[str]: + """The escalation a folded, still waiting alert belongs to, else ``None``.""" + summary = alert.summary if isinstance(alert.summary, Mapping) else {} + marker = summary.get("reopened") + if alert.status != "waiting_human" or not isinstance(marker, Mapping) or marker.get("mode") != FOLDED: + return None + case = summary.get("escalation_id") + return case if isinstance(case, str) else None + + +async def settle_folded( + repos: Repositories, tenant_id: str, alert_id: str, *, escalation_id: str, decision: str, decided_by: str, + conn: Optional[Connection] = None, +) -> bool: + """Apply the decision on ``escalation_id`` to a folded alert; whether the alert is a folded one (into any case). + + Only a decision on the alert's own case changes it. Idempotent: a settled alert is no longer folded. + """ + if conn is None: + async with repos.db.tenant_tx(tenant_id) as tx: + return await settle_folded(repos, tenant_id, alert_id, escalation_id=escalation_id, decision=decision, + decided_by=decided_by, conn=tx) + alert = await repos.alerts.get(tenant_id, alert_id, conn=conn) + case = folded_into(alert) + if case is None: + return False + if case != escalation_id: + return True # folded into another case: that case's decision settles it + if decision == "true_positive": + disposition = "malicious" + elif decision in BENIGN_DECISIONS: + disposition = "benign" + else: + return True # needs_more_info: it stays with the threat, whose next case takes it over + await repos.alerts.update_status(tenant_id, alert_id, "processing", conn=conn) + await repos.alerts.update( + tenant_id, alert_id, disposition=disposition, + summary={"close_reason": f"analyst:{decision}", "analyst_decision": decision}, conn=conn, + ) + await repos.alerts.update_status(tenant_id, alert_id, "closed", conn=conn) + details = {"decision": decision, "escalation_id": escalation_id, "disposition": disposition} + await repos.alert_events.append( + tenant_id, alert_id, "analyst.decision", node="await_analyst", actor_type="user", actor_id=decided_by, + data={**details, "folded": True}, conn=conn, + ) + await AuditLogger(repos).record( + tenant_id, SETTLED_ACTION, "alert", alert_id, actor_type="user", actor_id=decided_by, details=details, + conn=conn, + ) + return True + + +async def settle_case( + repos: Repositories, tenant_id: str, escalation: Any, decision: str, decided_by: Any, conn: Connection +) -> Set[str]: + """In the decision's transaction: settle every alert folded into the decided case; their ids (they have no run + waiting, so the review service sends them no resume message).""" + return { + alert_id for alert_id in escalation.related_alert_ids + if await settle_folded(repos, tenant_id, alert_id, escalation_id=escalation.id, decision=decision, + decided_by=str(decided_by), conn=conn) + } diff --git a/src/lg_sotf/services/review.py b/src/lg_sotf/services/review.py index 53e0fb32..6db38fc9 100644 --- a/src/lg_sotf/services/review.py +++ b/src/lg_sotf/services/review.py @@ -68,6 +68,7 @@ from lg_sotf.observability.logging import get_logger, log_failure from lg_sotf.security.crypto import SecretBox from lg_sotf.services.escalation import EscalationService, own_review_reason, resolve_review_policy +from lg_sotf.services.folded import settle_case from lg_sotf.services.response import ( ACTION_CATALOG, ResponseService, @@ -334,9 +335,8 @@ async def decide( ) -> DecisionResult: """Record the analyst decision and return one resume message per waiting alert. - The escalation's own alert and every attached alert are resumed with - the decision (what it means for attached alerts is limited, see the - module docstring and :attr:`DecisionResult.own_review_alert_ids`). + The escalation's own alert and every attached alert are resumed with the decision (limits: module docstring, + :attr:`DecisionResult.own_review_alert_ids`); alerts folded into the case (policy v3) are settled instead. ``approved_actions`` (``"action_type"`` or ``"action_type:target"``) pre-approves actions for the escalation's own alert only and only with ``true_positive``; a bare ``"action_type"`` is resolved to that @@ -367,8 +367,9 @@ async def decide( suggest=lambda: self._suggest(tenant.settings, suggest_suppression), now=now, ) - decided = recorded.escalation - requests = _resume_requests(escalation_id, decided, request, approved, actor) + folded = await settle_case(self._repos, tenant_id, recorded.escalation, request.decision, user_id, conn) + decided = recorded.escalation # folded alerts (policy v3) have no run waiting: settled above, never resumed + requests = [r for r in _resume_requests(escalation_id, decided, request, approved, actor) if r.alert_id not in folded] for resume in requests: await publish_event( self._events, "escalation.updated", tenant_id, resume.alert_id, diff --git a/tests/integration/api/test_retro_reopen_api.py b/tests/integration/api/test_retro_reopen_api.py new file mode 100644 index 00000000..afbfa896 --- /dev/null +++ b/tests/integration/api/test_retro_reopen_api.py @@ -0,0 +1,120 @@ +"""Decision policy v3 safety net (FIX-3) on the API's threat paths: an analyst's ``true_positive`` decision and a +blind-audit ``malicious`` label reopen the model-closed alerts that share an entity with the alert, attach them to +its escalation and queue their new run.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any, Dict, List + +import pytest + +from lg_sotf.core.retro_reopen import REOPEN_ACTION +from lg_sotf.domain import closure +from lg_sotf.observability.metrics import RETRO_REOPEN_FAILURES +from tests.integration.api._blind_audit import API, alert_of, queue, sample, seed_frame +from tests.integration.api._support import TENANT_A, ApiHarness +from tests.integration.db._support import make_alert + +pytestmark = pytest.mark.integration + + +async def model_closed(api: ApiHarness, entities: Dict[str, List[str]]) -> str: + alert_id = await make_alert(api.repos, TENANT_A, title="Package installed", severity="low", + normalized={"entities": entities}) + for status in ("queued", "processing", "closed"): + await api.repos.alerts.update_status(TENANT_A, alert_id, status) + await api.repos.alerts.update(TENANT_A, alert_id, disposition="benign", + summary={"close_reason": closure.MODEL_CLOSE_REASON}) + return alert_id + + +async def assert_reopened(api: ApiHarness, alert_id: str, escalation_id: str, mode: str) -> None: + alert = await api.repos.alerts.get(TENANT_A, alert_id) + assert alert.status == ("waiting_human" if mode == "folded" else "queued") and alert.disposition is None + assert alert.summary["reopened"]["mode"] == mode + assert alert.summary["reopened"]["reason"] == "related_threat" + attached = alert_id in (await api.repos.escalations.get(TENANT_A, escalation_id)).related_alert_ids + assert attached is (mode == "folded") + assert alert.summary["reopened"]["escalation_id"] == escalation_id + assert [e.resource_id for e in (await api.repos.audit.list(TENANT_A, action=REOPEN_ACTION)).items] == [alert_id] + + +async def test_an_analyst_true_positive_reopens_related_model_closes(api: ApiHarness) -> None: + threat = await make_alert(api.repos, TENANT_A, title="Beacon", severity="medium", + normalized={"entities": {"hosts": ["srv-app-02"]}}) + closed_alert = await model_closed(api, {"hosts": ["srv-app-02"]}) + escalation, _ = await api.repos.escalations.create_or_attach( + TENANT_A, threat, reason="review", sla_deadline=datetime.now(UTC) + timedelta(hours=1) + ) + analyst = await api.login_headers(TENANT_A, "analyst") + response = await api.client.post( + f"/api/v1/escalations/{escalation.id}/decision", + json={"decision": "true_positive", "version": escalation.version}, headers=analyst, + ) + assert response.status_code == 200, response.text + await assert_reopened(api, closed_alert, escalation.id, "requeued") # the case is decided: its own review + + +async def test_a_malicious_blind_audit_label_reopens_related_model_closes(api: ApiHarness) -> None: + await seed_frame(api.repos, n=1) + await sample(api, await api.login_headers(TENANT_A, "admin"), size=1) + analyst = api.auth(await api.user(TENANT_A, "analyst")) + [item] = await queue(api, analyst) + sampled = await api.repos.alerts.get(TENANT_A, await alert_of(api, item)) + hosts = list(sampled.entities.get("hosts") or []) + assert hosts, sampled.entities + closed_alert = await model_closed(api, {"hosts": hosts}) + response = await api.client.post( + f"{API}/blind-audit/items/{item['id']}/label", json={"label": "malicious"}, headers=analyst + ) + assert response.status_code == 200, response.text + escalation = await api.repos.escalations.get_active_for_alert(TENANT_A, sampled.id) + assert escalation is not None + await assert_reopened(api, closed_alert, escalation.id, "folded") # into the open blind-audit case + + +async def test_the_case_decision_settles_folded_alerts_in_its_own_transaction(api: ApiHarness) -> None: + """INV-03 (FIX-3 review): an alert folded into an open case has no run waiting; the decision service closes it + with the case decision (benign or malicious, audited) and sends it no resume message.""" + from lg_sotf.audit.logger import Actor + from lg_sotf.core.retro_reopen import reopen_related_model_closes + from lg_sotf.services.folded import SETTLED_ACTION + + for decision, disposition in (("false_positive", "benign"), ("true_positive", "malicious")): + host = f"srv-{decision}" + threat = await make_alert(api.repos, TENANT_A, title="Beacon", severity="medium", + normalized={"entities": {"hosts": [host]}}) + closed_alert = await model_closed(api, {"hosts": [host]}) + case, _ = await api.repos.escalations.create_or_attach( + TENANT_A, threat, reason="review", sla_deadline=datetime.now(UTC) + timedelta(hours=1) + ) + await reopen_related_model_closes(api.repos, TENANT_A, await api.repos.alerts.get(TENANT_A, threat), + actor=Actor("system", "test"), escalation_id=case.id) + assert (await api.repos.alerts.get(TENANT_A, closed_alert)).status == "waiting_human" + case = await api.repos.escalations.get(TENANT_A, case.id) + analyst = await api.login_headers(TENANT_A, "analyst") + response = await api.client.post( + f"/api/v1/escalations/{case.id}/decision", + json={"decision": decision, "version": case.version}, headers=analyst, + ) + assert response.status_code == 200, response.text + settled = await api.repos.alerts.get(TENANT_A, closed_alert) + assert (settled.status, settled.disposition) == ("closed", disposition), decision + audited = [e.resource_id for e in (await api.repos.audit.list(TENANT_A, action=SETTLED_ACTION)).items] + assert closed_alert in audited + + +async def test_an_api_reopen_failure_is_counted_not_swallowed(api: ApiHarness, monkeypatch: pytest.MonkeyPatch) -> None: + from lg_sotf.api import common + from lg_sotf.audit.logger import Actor + + async def broken(*args: Any, **kwargs: Any) -> Any: + raise RuntimeError("database went away") + + monkeypatch.setattr(api.ctx.repos.alerts, "model_closed_sharing_entities", broken) + alert_id = await make_alert(api.repos, TENANT_A, title="x", normalized={"entities": {"hosts": ["h1"]}}) + before = RETRO_REOPEN_FAILURES.labels(path="analyst")._value.get() + analyst = await api.user(TENANT_A, "analyst") + await common.invalidate_verdicts(api.ctx, TENANT_A, [alert_id], Actor("user", str(analyst.id)), path="analyst") + assert RETRO_REOPEN_FAILURES.labels(path="analyst")._value.get() == before + 1 diff --git a/tests/integration/engine/test_model_closure.py b/tests/integration/engine/test_model_closure.py index 9afa39f5..2def41e7 100644 --- a/tests/integration/engine/test_model_closure.py +++ b/tests/integration/engine/test_model_closure.py @@ -61,10 +61,13 @@ async def test_the_tenant_tier_decides_whether_the_models_close(engine: Workflow assert alert.summary["decision"]["code"] == "model_closed" assert {"code": "model_closed", "autonomy_tier": "auto_full_audit"}.items() <= (await decide_event(repos, on)).items() - # Never for high severity or a never-close class, whatever the tier. + # Policy v3: a high alert closes on the models' word at the tier; never a critical one or a never-close class. high = await routine_alert(repos, "high", severity="high") - assert (await engine.run(TENANT_A, high)).status != "completed" - assert (await repos.alerts.get(TENANT_A, high)).disposition != "benign" + assert (await engine.run(TENANT_A, high)).status == "completed" + assert (await decide_event(repos, high))["code"] == "model_closed" + critical = await routine_alert(repos, "critical", severity="critical") + assert (await engine.run(TENANT_A, critical)).status != "completed" + assert (await repos.alerts.get(TENANT_A, critical)).disposition != "benign" spray = await routine_alert( repos, "spray", mitre={"tactics": ["Credential Access"], "techniques": ["T1110.003"]} ) @@ -91,12 +94,12 @@ async def wazuh_alert(repos: Repositories, name: str, rule_id: str, description: raw=raw, normalized=fields) -async def test_catalogued_noise_closes_on_the_models_word_and_an_uncatalogued_rule_does_not( +async def test_catalogued_noise_is_named_and_an_unreadable_wazuh_rule_does_not_close( engine: WorkflowEngine, repos: Repositories ) -> None: - """FIX-2 through the real workflow: an unmapped Wazuh package install (rule 2902, catalogued) closes on the - models' word, marked with its family; a custom rule copying the description stays unmapped and goes to an - analyst.""" + """Through the real workflow: an unmapped Wazuh package install (rule 2902, catalogued) closes on the models' + word with its family in the reason; a custom rule copying the description closes too (policy v3 lifts unmapped), + without the family; a Wazuh alert whose rule cannot be read (vendor_unknown) goes to an analyst.""" await repos.tenants.update_settings(TENANT_A, {"ai_closure": {"tier": "auto_full_audit"}}) package = await wazuh_alert(repos, "package", "2902", DPKG) assert (await engine.run(TENANT_A, package)).status == "completed" @@ -104,5 +107,14 @@ async def test_catalogued_noise_closes_on_the_models_word_and_an_uncatalogued_ru assert decided["code"] == "model_closed" and "catalogued noise linux_package_install" in decided["reason"] custom = await wazuh_alert(repos, "custom", "100200", DPKG, entities={"hosts": ["db-08"]}) - assert (await engine.run(TENANT_A, custom)).status == "waiting_human" - assert (await decide_event(repos, custom))["code"] == "false_positive_review" + assert (await engine.run(TENANT_A, custom)).status == "completed" + custom_decided = await decide_event(repos, custom) + assert custom_decided["code"] == "model_closed" and "catalogued noise" not in custom_decided["reason"] + + unreadable = await make_alert( + repos, TENANT_A, source_alert_id="fix3-unreadable", title=ROUTINE, severity="medium", raw={"vendor": "x"}, + normalized={"title": ROUTINE, "rule_name": f"2902: {DPKG}", "description": ROUTINE, "normalizer": "wazuh", + "entities": {"hosts": ["db-09"]}, "mitre": {"tactics": [], "techniques": []}}, + ) + assert (await engine.run(TENANT_A, unreadable)).status == "waiting_human" + assert (await decide_event(repos, unreadable))["code"] == "false_positive_review" diff --git a/tests/integration/engine/test_retro_reopen.py b/tests/integration/engine/test_retro_reopen.py new file mode 100644 index 00000000..ded98d94 --- /dev/null +++ b/tests/integration/engine/test_retro_reopen.py @@ -0,0 +1,149 @@ +"""Decision policy v3 safety net (FIX-3): a related threat reopens the alerts the models closed alone. + +Through the real workflow and database: each threat path of the engine +(an escalation a model judged a threat, a response plan, a malicious close) +reopens the tenant's model-closed alerts that share a host, user or IP and +closed in the previous 24 hours; the reopen is idempotent, tenant-scoped, +audited ``alert.reopened`` and counted when it fails. A reopened alert never +closes on the models' word again. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any, Dict, List + +import pytest + +from lg_sotf.audit.logger import Actor +from lg_sotf.core import retro_reopen +from lg_sotf.core.graph import initial_state +from lg_sotf.core.retro_reopen import REOPEN_ACTION, reopen_counted, reopen_related_model_closes +from lg_sotf.core.workflow import WorkflowEngine +from lg_sotf.db.errors import NotFound +from lg_sotf.db.repos import Repositories +from lg_sotf.domain import closure +from lg_sotf.observability.metrics import RETRO_REOPEN_FAILURES +from tests.integration.db._support import TENANT_A, TENANT_B, make_alert +from tests.integration.engine._support import analyst, decide, suspicious_alert + +pytestmark = pytest.mark.integration + +ACTOR = Actor("system", "test") + + +async def model_closed(repos: Repositories, entities: Dict[str, List[str]], tenant_id: str = TENANT_A, + closed_at: datetime | None = None) -> str: + """An alert the models closed alone (``model_closed``, close reason ``ai_closed:audit_required``).""" + alert_id = await make_alert(repos, tenant_id, title="Package installed", severity="low", + normalized={"entities": entities}) + for status in ("queued", "processing", "closed"): + await repos.alerts.update_status(tenant_id, alert_id, status, now=closed_at) + await repos.alerts.update(tenant_id, alert_id, disposition="benign", + summary={"close_reason": closure.MODEL_CLOSE_REASON}) + return alert_id + + +async def reopened(repos: Repositories, alert_id: str, mode: str, tenant_id: str = TENANT_A) -> bool: + """Folded into an open case (``waiting_human``, no new run) or, without one, queued for its own review run.""" + alert = await repos.alerts.get(tenant_id, alert_id) + marker = alert.summary.get("reopened") or {} + status = "waiting_human" if mode == "folded" else "queued" + return (alert.status, marker.get("reason"), marker.get("mode"), alert.disposition) == ( + status, "related_threat", mode, None) + + +async def test_an_escalation_a_model_judged_a_threat_reopens_related_model_closes( + engine: WorkflowEngine, repos: Repositories +) -> None: + same_host = await model_closed(repos, {"hosts": ["ws-042"]}) + same_user = await model_closed(repos, {"users": ["jdoe"]}) + same_ip = await model_closed(repos, {"ips": ["185.220.101.44"]}) + unrelated = await model_closed(repos, {"hosts": ["db-01"]}) + stale = await model_closed(repos, {"hosts": ["ws-042"]}, closed_at=datetime.now(UTC) - timedelta(hours=25)) + other_tenant = await model_closed(repos, {"hosts": ["ws-042"]}, tenant_id=TENANT_B) + + threat = await suspicious_alert(repos) + outcome = await engine.run(TENANT_A, threat) + assert outcome.status == "waiting_human" + triage = (await repos.alerts.get(TENANT_A, threat)).summary["triage"] + analysis = (await repos.alerts.get(TENANT_A, threat)).summary.get("analysis") or {} + assert retro_reopen.judged_threat(triage, analysis), (triage, analysis) + + case = outcome.interrupt["escalation_id"] + for alert_id in (same_host, same_user, same_ip): + assert await reopened(repos, alert_id, "folded"), alert_id + folded = await repos.alerts.get(TENANT_A, alert_id) + assert folded.summary["reopened"] == {"reason": "related_threat", "threat_alert_id": threat, + "escalation_id": case, "mode": "folded"} + assert folded.summary["escalation_id"] == case + assert await repos.escalations.get_active_for_alert(TENANT_A, alert_id) is None # no case of its own + with pytest.raises(NotFound): # and no new run (these alerts were closed without one in this test) + await repos.runs.get(TENANT_A, alert_id) + for alert_id, tenant in ((unrelated, TENANT_A), (stale, TENANT_A), (other_tenant, TENANT_B)): + alert = await repos.alerts.get(tenant, alert_id) + assert (alert.status, alert.disposition) == ("closed", "benign"), alert_id + escalation = await repos.escalations.get(TENANT_A, outcome.interrupt["escalation_id"]) + assert {same_host, same_user, same_ip} <= set(escalation.related_alert_ids) + audited = {e.resource_id for e in (await repos.audit.list(TENANT_A, action=REOPEN_ACTION)).items} + assert audited == {same_host, same_user, same_ip} + events = (await repos.alert_events.list_for_alert(TENANT_A, same_host)).items + assert [e.event_type for e in events if e.event_type == REOPEN_ACTION] == [REOPEN_ACTION] + + # Idempotent: a second threat on the same host finds nothing left to reopen. + again = await reopen_related_model_closes(repos, TENANT_A, await repos.alerts.get(TENANT_A, threat), actor=ACTOR) + assert again == [] + + # Should a reopened alert run again, it never closes on the models' word. + state = initial_state(TENANT_A, await repos.alerts.get(TENANT_A, same_host)) + assert state["alert"]["reopened"] == {"reason": "related_threat"} + assert "reopened" in closure.model_close_blockers(state, tier=closure.MODEL_CLOSE_TIER) + + +async def test_a_response_plan_and_the_malicious_close_each_reopen_related_model_closes( + engine: WorkflowEngine, repos: Repositories +) -> None: + """The plan_response path and the malicious-close path each reopen what closed since the previous one.""" + threat = await suspicious_alert(repos) + user_id = await analyst(repos) + first = await engine.run(TENANT_A, threat) + assert first.status == "waiting_human" + + before_plan = await model_closed(repos, {"hosts": ["ws-042"]}) + second = await decide(engine, repos, threat, first.interrupt["escalation_id"], "true_positive", user_id) + assert second.status == "waiting_approval" + assert await reopened(repos, before_plan, "requeued") # plan_response: the case is decided, nothing to fold into + + before_close = await model_closed(repos, {"users": ["jdoe"]}) + done = await engine.resume( + TENANT_A, threat, + {"type": "approval", "approved_action_ids": [], "rejected_action_ids": [], "decided_by": user_id, + "reject_remaining": True}, + ) + assert done.status == "completed" + assert (await repos.alerts.get(TENANT_A, threat)).disposition == "malicious" + assert await reopened(repos, before_close, "requeued") # the malicious close + + +async def test_a_benign_escalation_reopens_nothing(engine: WorkflowEngine, repos: Repositories) -> None: + closed_alert = await model_closed(repos, {"hosts": ["build-7"]}) + quiet = await make_alert(repos, TENANT_A, title="Authorized approved expected patch healthcheck", + severity="low", normalized={"entities": {"hosts": ["build-7"]}}) + await engine.run(TENANT_A, quiet) + alert = await repos.alerts.get(TENANT_A, closed_alert) + assert (alert.status, alert.disposition) == ("closed", "benign") + + +async def test_a_reopen_failure_is_counted_not_swallowed( + repos: Repositories, monkeypatch: pytest.MonkeyPatch +) -> None: + await model_closed(repos, {"hosts": ["ws-042"]}) + threat = await repos.alerts.get(TENANT_A, await suspicious_alert(repos)) + + async def broken(*args: Any, **kwargs: Any) -> Any: + raise RuntimeError("database went away") + + monkeypatch.setattr(repos.alerts, "model_closed_sharing_entities", broken) + before = RETRO_REOPEN_FAILURES.labels(path="workflow")._value.get() + assert await reopen_counted(repos, TENANT_A, threat, actor=ACTOR, path="workflow") == [] + assert RETRO_REOPEN_FAILURES.labels(path="workflow")._value.get() == before + 1 diff --git a/tests/integration/engine/test_review_349.py b/tests/integration/engine/test_review_349.py new file mode 100644 index 00000000..71b464eb --- /dev/null +++ b/tests/integration/engine/test_review_349.py @@ -0,0 +1,84 @@ +"""Adversarial review of PR #349 (FIX-3): folded alerts in production conditions.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Dict, List + +import pytest +import redis.asyncio as aioredis + +from lg_sotf.core.workflow import WorkflowEngine +from lg_sotf.db.pool import Database +from lg_sotf.db.repos import Repositories +from lg_sotf.domain import closure +from lg_sotf.queue.consumer import StreamConsumer +from lg_sotf.queue.streams import Message +from lg_sotf.runtime.worker import Worker +from lg_sotf.settings import Settings +from tests.integration.db._support import TENANT_A, make_alert +from tests.integration.engine._support import RecordingEvents, analyst, decide, suspicious_alert + +pytestmark = pytest.mark.integration + + +async def model_closed_with_run(repos: Repositories, entities: Dict[str, List[str]]) -> str: + """A model-closed alert as production leaves it: its workflow run is ``completed``.""" + alert_id = await make_alert(repos, TENANT_A, title="Package installed", severity="low", + normalized={"entities": entities}) + await repos.runs.upsert_queued(TENANT_A, alert_id) + await repos.runs.start(TENANT_A, alert_id, "w0", lease_s=30, kind="process") + for status in ("queued", "processing", "closed"): + await repos.alerts.update_status(TENANT_A, alert_id, status) + await repos.alerts.update(TENANT_A, alert_id, disposition="benign", + summary={"close_reason": closure.MODEL_CLOSE_REASON}) + await repos.runs.complete(TENANT_A, alert_id, worker_id="w0") + return alert_id + + +async def test_worker_delivers_the_case_decision_to_a_folded_alert( + engine: WorkflowEngine, repos: Repositories, db: Database, redis_client: aioredis.Redis, + engine_settings: Settings, +) -> None: + folded = await model_closed_with_run(repos, {"hosts": ["ws-042"]}) + threat = await suspicious_alert(repos) + outcome = await engine.run(TENANT_A, threat) + case = outcome.interrupt["escalation_id"] + assert (await repos.alerts.get(TENANT_A, folded)).status == "waiting_human" + user_id = await analyst(repos) + + worker = Worker(engine_settings, db, redis_client, engine, StreamConsumer(redis_client, engine_settings, "w1"), + RecordingEvents()) + payload = {"type": "analyst_decision", "escalation_id": case, "decision": "false_positive", + "approved_actions": [], "decided_by": user_id, "notes": ""} + # What the API enqueues for every alert of the case (EscalationService.alerts_to_resume), as the worker gets it. + msg = Message(stream="s", entry_id="0-1", tenant_id=TENANT_A, alert_id=folded, kind="resume", + payload=payload, enqueued_at=datetime.now(UTC), deliveries=1) + await worker.process(msg) + alert = await repos.alerts.get(TENANT_A, folded) + assert alert.status == "closed", f"folded alert left {alert.status}: the worker never calls engine.resume" + + +async def test_needs_more_info_strands_a_folded_alert(engine: WorkflowEngine, repos: Repositories) -> None: + folded = await model_closed_with_run(repos, {"hosts": ["ws-042"]}) + threat = await suspicious_alert(repos) + first = await engine.run(TENANT_A, threat) + case1 = first.interrupt["escalation_id"] + assert (await repos.alerts.get(TENANT_A, folded)).summary["escalation_id"] == case1 + user_id = await analyst(repos) + + second = await decide(engine, repos, threat, case1, "needs_more_info", user_id) + assert second.status == "waiting_human" + case2 = second.interrupt["escalation_id"] + assert case2 != case1 + # Even with the needs_more_info resume delivered to the folded alert, it stays waiting on case1 (decided). + await engine.resume(TENANT_A, folded, {"type": "analyst_decision", "escalation_id": case1, + "decision": "needs_more_info", "approved_actions": [], + "decided_by": user_id, "notes": ""}) + escalation2 = await repos.escalations.get(TENANT_A, case2) + await decide(engine, repos, threat, case2, "false_positive", user_id) + # The API resumes alerts_to_resume(case2) only: the folded alert is not among them. + assert folded in [escalation2.alert_id, *escalation2.related_alert_ids], ( + f"folded alert is attached only to decided case {case1}; case2 holds {escalation2.related_alert_ids}; " + f"status {(await repos.alerts.get(TENANT_A, folded)).status}" + ) diff --git a/tests/integration/eval/_support.py b/tests/integration/eval/_support.py index d266d4d6..645ed521 100644 --- a/tests/integration/eval/_support.py +++ b/tests/integration/eval/_support.py @@ -10,6 +10,8 @@ import httpx import pytest +from lg_sotf.db.repos import Repositories +from lg_sotf.domain import closure from lg_sotf.eval.dataset import Dataset from lg_sotf.eval.results import ItemResult from lg_sotf.llm.fake import FakeChatModel @@ -98,3 +100,25 @@ def closures(results: List[ItemResult]) -> Dict[str, int]: "benign_model_closed": sum(r.expected == "benign" and r.decision_code == "model_closed" for r in closed), "malicious_model_closed": sum(r.expected == "malicious" and r.decision_code == "model_closed" for r in closed), } + + +async def v3_closures(repos: Repositories, run: Any) -> Dict[str, List[str]]: + """Decision policy v3 (FIX-3): what automation closed in an evaluation run, by label, and what is still closed + at the end of the run (after retro-reopen, which later threats in the same run trigger); ``forbidden`` lists the + model-led closes INV-03 never allows (a critical alert, a never-close class).""" + tenants = run.info["tenants"] + out: Dict[str, List[str]] = {k: [] for k in ("benign_closed", "benign_still_closed", "malicious_closed", + "malicious_still_closed", "forbidden")} + for r in run.results: + if r.status != "closed" or r.alert_id is None: + continue + alert = await repos.alerts.get(tenants.get(r.split) or tenants["*"], r.alert_id) + label = "malicious" if r.expected == "malicious" else "benign" + out[f"{label}_closed"].append(r.item_id) + if alert.status == "closed": + out[f"{label}_still_closed"].append(r.item_id) + if r.decision_code == closure.MODEL_CLOSED: + document = {**alert.normalized, "raw": alert.raw} + if alert.severity == "critical" or closure.model_close_class(document) is not None: + out["forbidden"].append(r.item_id) + return out diff --git a/tests/integration/eval/test_adversarial_v2_eval.py b/tests/integration/eval/test_adversarial_v2_eval.py index 3ec23aed..f04bc66c 100644 --- a/tests/integration/eval/test_adversarial_v2_eval.py +++ b/tests/integration/eval/test_adversarial_v2_eval.py @@ -23,7 +23,7 @@ from lg_sotf.llm.cassette import CassetteSet, cassette_name from lg_sotf.llm.fake import FakeChatModel from lg_sotf.settings import Settings -from tests.integration.eval._support import at_tier, closures +from tests.integration.eval._support import at_tier, v3_closures pytestmark = pytest.mark.integration @@ -87,19 +87,19 @@ async def test_adversarial_v2_passes_its_gate_with_the_fake_model(eval_settings: assert any("Note to AI triage" in str(h.get("data")) for h in chained.summary["analysis"]["history"]) -async def test_no_malicious_close_with_model_led_closure_on(eval_settings: Settings) -> None: - """INV-03 (FIX-1): at the autonomy tier auto_full_audit the fake model, which follows much of the injected and - camouflaging text, still closes no malicious item, and the gate holds. The never-close classes (persistence - through attacker-named objects, unmapped alerts) and the steering-text flag are what stop it.""" +async def test_model_led_closure_on_closes_only_what_policy_v3_allows(eval_settings: Settings, repos: Repositories) -> None: + """INV-03 (FIX-3): at the autonomy tier auto_full_audit the fake model, which follows much of the injected text, + closes some malicious items on its word: decision policy v3 accepts that and relies on retro-reopen and the audit. + What it never does: close a critical alert or a never-close class, and everything else it closes is marked + model_closed for audit. The numbers (before and after retro-reopen) are reported in the FIX-3 pull request.""" dataset = at_tier(load_dataset(DATASET), "auto_full_audit") run = await run_evaluation(eval_settings, dataset, tenant_id=f"adv2on-{secrets.token_hex(3)}") - report = build_report(run.results, dataset=dataset, run_info=run.info) - gate = evaluate_gate(report, load_thresholds(THRESHOLDS)) - assert gate.passed, ([v.message for v in gate.violations], _outcomes(run.results)) - counts = closures(run.results) - assert counts["malicious_closed"] == 0 and report["metrics"]["errors"] == 0, (counts, _outcomes(run.results)) - camouflage = [r for r in run.results if r.family in ("title-camouflage", "activity-camouflage")] - assert len(camouflage) == 9 and all(r.escalated and r.disposition is None for r in camouflage), camouflage + assert [(r.item_id, r.error) for r in run.results if r.status == "error"] == [] + counts = await v3_closures(repos, run) + assert counts["forbidden"] == [], counts + off = {r.item_id for r in run.results if r.status == "closed" and r.decision_code != "model_closed"} + assert set(counts["malicious_closed"]) <= {r.item_id for r in run.results if r.decision_code == "model_closed"} + assert not off & set(counts["malicious_closed"]) async def test_tool_fixture_answers_are_recorded_and_replayed_by_cassettes( diff --git a/tests/integration/eval/test_corpus_v1_eval.py b/tests/integration/eval/test_corpus_v1_eval.py index ede17f91..3e028cc3 100644 --- a/tests/integration/eval/test_corpus_v1_eval.py +++ b/tests/integration/eval/test_corpus_v1_eval.py @@ -22,13 +22,14 @@ import pytest +from lg_sotf.db.repos import Repositories from lg_sotf.eval.dataset import load_dataset from lg_sotf.eval.gate import evaluate_gate, load_thresholds from lg_sotf.eval.report import build_report, render_markdown, write_json from lg_sotf.eval.runner import run_evaluation from lg_sotf.llm.cassette import CassetteSet, cassette_name from lg_sotf.settings import Settings -from tests.integration.eval._support import at_tier, block_network_and_model, closures +from tests.integration.eval._support import NetworkGuard, at_tier, block_network_and_model, closures, v3_closures pytestmark = [pytest.mark.integration, pytest.mark.timeout(1800)] @@ -42,7 +43,7 @@ async def test_corpus_v1_replays_offline_and_passes_its_gate( - eval_settings: Settings, monkeypatch: pytest.MonkeyPatch + eval_settings: Settings, monkeypatch: pytest.MonkeyPatch, repos: Repositories ) -> None: dataset = load_dataset(DATASET) assert len(dataset.items) >= 500 and set(dataset.families()) == FAMILIES @@ -55,6 +56,16 @@ async def test_corpus_v1_replays_offline_and_passes_its_gate( assert [(r.item_id, r.error) for r in recorded.results if r.status == "error"] == [] assert {p.name for p in CASSETTES.iterdir()} == {cassette_name(i.id) + ".gz" for i in dataset.items} + # FIX-3: model-led closure on. Model closes change what later alerts' correlation shows to the model, so the + # tier-off cassettes no longer answer every call: this run uses the fake model itself, with the network blocked. + net = NetworkGuard(eval_settings) + net.install(monkeypatch) + on = await run_evaluation(eval_settings, at_tier(dataset, "auto_full_audit"), + tenant_id=f"play-cv1on-{secrets.token_hex(3)}") + assert [(r.item_id, r.error) for r in on.results if r.status == "error"] == [] and net.attempts == [] + counts = await v3_closures(repos, on) + assert counts["forbidden"] == [], counts + guard = block_network_and_model(eval_settings, monkeypatch) replay = await run_evaluation(eval_settings, dataset, tenant_id=f"play-cv1-{secrets.token_hex(3)}", cassettes=CassetteSet(CASSETTES, "replay")) @@ -74,12 +85,3 @@ async def test_corpus_v1_replays_offline_and_passes_its_gate( assert metrics["scored"] == len(dataset.items) and metrics["errors"] == 0 assert set(report["families"]) == FAMILIES assert closures(replay.results)["benign_model_closed"] == 0 # the tier is off by default - - # FIX-1: model-led closure on. The tier only changes what decide does after analysis, so the same cassettes - # answer every call, and the gate (no malicious alert closed in any family) still holds. - on = await run_evaluation(eval_settings, at_tier(dataset, "auto_full_audit"), - tenant_id=f"play-cv1on-{secrets.token_hex(3)}", cassettes=CassetteSet(CASSETTES, "replay")) - assert [(r.item_id, r.error) for r in on.results if r.status == "error"] == [] and guard.attempts == [] - on_gate = evaluate_gate(build_report(on.results, dataset=dataset, run_info=on.info), load_thresholds(THRESHOLDS)) - assert on_gate.passed, [v.message for v in on_gate.violations] - assert closures(on.results)["malicious_closed"] == 0 diff --git a/tests/unit/core/test_state.py b/tests/unit/core/test_state.py index 52d54a99..ef1c7a10 100644 --- a/tests/unit/core/test_state.py +++ b/tests/unit/core/test_state.py @@ -156,7 +156,7 @@ def test_alert_summary_declares_what_the_close_node_writes() -> None: } accounting = {"llm_usage", "llm_models", "llm_refusals", "llm_budget_exceeded"} nodes = {"triage", "correlation", "analysis", "response", "escalation_attached", "detached_from"} - review = {"blind_audit_review"} # the blind audit (services.blind_audit) and the run it re-triages + review = {"blind_audit_review", "reopened"} # the blind audit, and the retro-reopen of policy v3 (core.retro_reopen) assert closed | accounting | nodes | review == declared(AlertSummary) assert declared(ResponseSummary) == {"threat_type", "evidence", "auto", "awaiting", "blocked", "done"} diff --git a/tests/unit/domain/test_closure.py b/tests/unit/domain/test_closure.py index e4387bd3..81491b47 100644 --- a/tests/unit/domain/test_closure.py +++ b/tests/unit/domain/test_closure.py @@ -65,35 +65,54 @@ def test_a_model_led_close_needs_the_tier_and_is_marked_for_audit() -> None: @pytest.mark.parametrize( ("path", "value", "blocker"), [ - ("severity", "high", "severity"), ("severity", "critical", "severity"), ("severity", "unknown", "severity"), ("alert.mitre.tactics", ["Credential Access"], "never_close_class"), ("alert.mitre.techniques", ["T1070.001"], "never_close_class"), + ("alert.mitre.techniques", ["T1531"], "never_close_class"), + ("alert.flags", ["llm_normalized"], "llm_normalized"), + ("alert.reopened", {"reason": "related_threat"}, "reopened"), ("triage.rule_tp", [INJECTION], "steering_text"), ("triage.rule_tp", None, "steering_text"), ("analysis.rule_evidence", ["prior_malicious_history"], "recent_true_positive"), ("analysis.rule_corroborated", True, "recent_true_positive"), ("correlation.related_dispositions", {"malicious": 1, "benign": 4}, "recent_true_positive"), - ("correlation.related_count", 2, "campaign_in_progress"), # related alerts, but nothing known about them - ("triage.threat_assessment", "suspicious", "fast_tier_not_benign"), - ("triage.threat_likelihood", 9, "fast_tier_not_benign"), - ("triage.threat_likelihood", None, "fast_tier_not_benign"), - ("triage.threat_likelihood", True, "fast_tier_not_benign"), - ("triage.llm_used", False, "fast_tier_not_benign"), + ("triage.threat_assessment", "likely_threat", "fast_tier_threat"), + ("triage.threat_assessment", "confirmed_threat", "fast_tier_threat"), ("analysis.verdict", "inconclusive", "deep_tier_not_benign"), - ("analysis.certainty", 84, "deep_tier_not_benign"), - ("analysis.certainty", None, "deep_tier_not_benign"), + ("analysis.verdict", "malicious", "deep_tier_not_benign"), + ("analysis.verdict", None, "deep_tier_not_benign"), ("analysis.llm_used", False, "deep_tier_not_benign"), ], ) def test_each_condition_alone_blocks_a_model_led_close(path: str, value: Any, blocker: str) -> None: state = benign_state(**{path: value}) - blockers = model_close_blockers(state, tier=AUTO, max_likelihood=8, min_certainty=85) - assert blockers == [blocker] + assert model_close_blockers(state, tier=AUTO) == [blocker] assert routing.decide(state, T, AUTO).route != "close" +@pytest.mark.parametrize( + ("path", "value"), + [ + ("severity", "high"), # high closes on the models' word at the tier (v3) + ("triage.threat_assessment", "suspicious"), # only likely_threat and confirmed_threat block + ("triage.threat_likelihood", 60), # the install thresholds are gone + ("triage.llm_used", False), + ("analysis.certainty", 5), # any certainty + ("analysis.threat_score", 50), # the grey zone does not hold a benign verdict back + ("alert.mitre", {"tactics": [], "techniques": []}), # unmapped no longer blocks + ("correlation", {"related_count": 9, "burst": True, "related_threats": 3, "related_open_reviews": 9, + "related_dispositions": {"undecided": 9}}), # no campaign check: retro-reopen instead + ], +) +def test_v3_removed_the_forward_gates(path: str, value: Any) -> None: + """FIX-3: the thresholds, the grey zone for a benign verdict, the campaign check and the unmapped block are + gone; the safety net reopens a close when a related threat appears (lg_sotf.core.retro_reopen).""" + state = benign_state(**{path: value}) + assert model_close_blockers(state, tier=AUTO) == [] + assert routing.decide(state, T, AUTO).code == MODEL_CLOSED + + def test_model_normalised_alerts_and_refusals_never_close_on_the_models_word() -> None: normalised = benign_state(**{"alert.flags": ["llm_normalized"]}) assert routing.decide(normalised, T, AUTO).code == "llm_normalized" @@ -103,18 +122,11 @@ def test_model_normalised_alerts_and_refusals_never_close_on_the_models_word() - assert routing.decide(asked, T, AUTO).code == "analyst_request" -def test_the_thresholds_are_inclusive_and_install_settings() -> None: - assert (T.model_close_max_likelihood, T.model_close_min_certainty) == (8, 85) - edge = benign_state(**{"triage.threat_likelihood": 8, "analysis.certainty": 85}) - assert routing.decide(edge, T, AUTO).code == MODEL_CLOSED - stricter = RoutingThresholds.from_settings({"model_close_max_likelihood": 4, "model_close_min_certainty": 95}) - assert routing.decide(benign_state(), stricter, AUTO).route == "escalate" - assert routing.decide(benign_state(**{"triage.threat_likelihood": 4, "analysis.certainty": 95}), stricter, AUTO).code == MODEL_CLOSED - - -def test_a_model_led_close_still_needs_a_low_threat_outside_the_grey_zone() -> None: - for threat, code in ((29, MODEL_CLOSED), (30, "grey_zone"), (75, "inconclusive"), (85, "high_risk")): +def test_a_benign_verdict_closes_below_the_response_policy_and_evidence_closes_first() -> None: + for threat, code in ((5, MODEL_CLOSED), (30, MODEL_CLOSED), (75, MODEL_CLOSED), (85, "high_risk")): assert routing.decide(benign_state(**{"analysis.threat_score": threat}), T, AUTO).code == code + evidence = benign_state(**{"triage.rule_fp": ["test_detected"]}) + assert routing.decide(evidence, T, AUTO).code == "rule_corroborated_benign" def _grid(severities: List[str], evidence: bool) -> List[Dict[str, Any]]: @@ -156,13 +168,18 @@ def test_with_the_tier_off_a_model_never_closes_alone() -> None: assert routing.route_after_triage(state, T) != "close" -def test_high_and_critical_never_close_at_any_tier() -> None: - """INV-03: with or without benign evidence, at every tier, a high or critical alert never closes.""" +def test_critical_never_closes_at_any_tier_and_high_only_on_the_models_word_at_the_tier() -> None: + """INV-03 (v3): with or without benign evidence, at every tier, a critical alert never closes; a high alert never + closes on evidence or at tier off, and on the models' word only at auto_full_audit.""" for tier, evidence in itertools.product(AUTONOMY_TIERS, (False, True)): - for state in _grid(["high", "critical"], evidence=evidence): + for state in _grid(["critical"], evidence=evidence): assert routing.decide(state, T, tier).route != "close", (tier, state) assert routing.route_after_triage(state, T) != "close" - assert "severity" in model_close_blockers(state, tier=tier, max_likelihood=100, min_certainty=0) + assert "severity" in model_close_blockers(state, tier=tier) + for state in _grid(["high"], evidence=evidence): + decision = routing.decide(state, T, tier) + assert decision.route != "close" or (tier == AUTO and decision.code == MODEL_CLOSED), (tier, state) + assert routing.route_after_triage(state, T) != "close" @pytest.mark.parametrize( @@ -185,7 +202,7 @@ def test_high_and_critical_never_close_at_any_tier() -> None: (["Discovery", "Execution"], ["T1059.001", "T1083"], None), (["Persistence", "Initial Access"], ["T1078"], None), (["Defense Evasion"], [], None), - ([], [], "unmapped"), + ([], ["T1531"], "credential_access"), ], ) def test_never_close_classes_never_close_on_the_models_word( @@ -198,15 +215,19 @@ def test_never_close_classes_never_close_on_the_models_word( assert (decision.code == MODEL_CLOSED) is (expected is None) -def test_an_alert_without_a_readable_attack_mapping_is_a_never_close_class() -> None: - """Fail closed: nothing shows an unmapped alert is outside the never-close classes.""" +def test_an_unmapped_alert_closes_on_the_models_word_but_an_unreadable_wazuh_rule_does_not() -> None: + """v3: nothing shows an unmapped alert is outside the never-close classes, but the safety net now reopens it on + a related threat, so it closes; a Wazuh alert whose rule cannot be read (vendor_unknown) still never does.""" for alert in (None, {}, {"mitre": "T1003"}, {"mitre": {"tactics": [], "techniques": []}}, {"mitre": {"tactics": [" "], "techniques": [1003, None]}}, {"category": "authentication_success"}): assert never_close_class(alert) == "unmapped", alert + assert closure.model_close_class(alert) is None assert never_close_class({"mitre": {"tactics": "Credential Access"}}) == "credential_access" assert never_close_class({"mitre": {"techniques": [1003, None, "T1486"]}}) == "ransomware" unmapped = benign_state(**{"alert.mitre": {"tactics": [], "techniques": []}}) - assert model_close_blockers(unmapped, tier=AUTO, max_likelihood=8, min_certainty=85) == ["never_close_class"] + assert model_close_blockers(unmapped, tier=AUTO) == [] + unreadable = benign_state(**{"alert.mitre": {"tactics": [], "techniques": []}, "alert.normalizer": "wazuh"}) + assert model_close_blockers(unreadable, tier=AUTO) == ["never_close_class"] def test_the_autonomy_tier_fails_closed() -> None: @@ -268,12 +289,6 @@ def test_review_f2b_pascal_case_tactics_are_never_closed_on_the_models_word() -> assert routing.decide(state, T, AUTO).route != "close" -def test_review_f3_a_burst_of_undecided_related_alerts_is_never_model_closed() -> None: - state = _review_state(correlation={"related_count": 25, "burst": True, "score": 100, "frequency_level": "critical", - "related_dispositions": {"undecided": 10}}) - assert routing.decide(state, T, AUTO).route != "close" - - # --------------------------------------------------------------------------- F1: the evidence gate's hard checks EVIDENCE = ("known_scanner", "test_environment", "eicar_test_file", "test_detected", @@ -326,31 +341,3 @@ def test_a_scanner_or_test_source_never_closes_a_never_close_class(mitre: Dict[s assert routing.decide(state, T, "off").code == "rule_corroborated_benign" unmapped = benign_state(**{"triage.rule_fp": ["known_scanner"], "alert.mitre": {"tactics": [], "techniques": []}}) assert routing.decide(unmapped, T, "off").code == "rule_corroborated_benign" - - -# --------------------------------------------------------------------------- F3: a campaign in progress - - -@pytest.mark.parametrize( - "correlation", - [ - {"related_count": 3, "burst": True, "related_threats": 0, "related_open_reviews": 0}, - {"related_count": 3, "burst": False, "related_threats": 1, "related_open_reviews": 0}, - {"related_count": 3, "burst": False, "related_threats": 0, "related_open_reviews": 2}, - {"related_count": 3, "burst": False}, # counts unknown: fail closed - {"related_count": 3, "related_threats": 0, "related_open_reviews": 0}, # burst unknown - {}, # nothing known about related alerts - ], -) -def test_a_campaign_in_progress_is_never_model_closed(correlation: Dict[str, Any]) -> None: - state = benign_state(correlation={**correlation, "related_dispositions": {"benign": 3}}) - assert model_close_blockers(state, tier=AUTO, max_likelihood=8, min_certainty=85) == ["campaign_in_progress"] - assert routing.decide(state, T, AUTO).route == "escalate" - - -def test_quiet_related_alerts_do_not_block_a_model_led_close() -> None: - quiet = {"related_count": 3, "burst": False, "related_threats": 0, "related_open_reviews": 0, - "related_dispositions": {"benign": 2, "undecided": 1}} - assert routing.decide(benign_state(correlation=quiet), T, AUTO).code == MODEL_CLOSED - ai_closed = {**quiet, "related_dispositions": {"ai_closed_unaudited": 3}} - assert routing.decide(benign_state(correlation=ai_closed), T, AUTO).code == MODEL_CLOSED diff --git a/tests/unit/domain/test_noise_families.py b/tests/unit/domain/test_noise_families.py index 7d672c5d..26e0ad69 100644 --- a/tests/unit/domain/test_noise_families.py +++ b/tests/unit/domain/test_noise_families.py @@ -128,9 +128,9 @@ def test_the_catalogue_lifts_only_unmapped_never_a_mapped_never_close_class() -> assert noise_family(state["alert"]) is not None # the rule is catalogued ... assert closure.catalogued_noise(state["alert"]) is None # ... but it is not unmapped assert model_close_class(state["alert"]) == found - assert model_close_blockers(state, tier=AUTO, max_likelihood=8, min_certainty=85) == ["never_close_class"] + assert model_close_blockers(state, tier=AUTO) == ["never_close_class"] mapped = wazuh_state(wazuh_event("2902", DPKG, mitre={"tactic": ["Execution"], "id": ["T1072"]})) - assert routing.decide(mapped, T, AUTO).reason == "AI-closed, audit required: both model tiers judged it benign" + assert routing.decide(mapped, T, AUTO).reason == "AI-closed, audit required: the analysis judged it benign" # --------------------------------------------------------------------------- rule id only, never text @@ -152,10 +152,11 @@ def test_the_catalogue_lifts_only_unmapped_never_a_mapped_never_close_class() -> def test_a_forged_or_borrowed_title_on_an_uncatalogued_rule_stays_unmapped(rule_id: str, description: str) -> None: event = wazuh_event(rule_id, description, full_log="New dpkg (Debian Package) installed. rule 2902 62130 31101") state = wazuh_state(event, **{"alert.title": DPKG, "alert.description": "2902: " + DPKG}) - assert noise_family(state["alert"]) is None - assert model_close_class(state["alert"]) == UNMAPPED - assert model_close_blockers(state, tier=AUTO, max_likelihood=8, min_certainty=85) == ["never_close_class"] - assert routing.decide(state, T, AUTO).route == "escalate" + assert noise_family(state["alert"]) is None and never_close_class(state["alert"]) == UNMAPPED + # Policy v3: an unmapped alert closes on the models' word; the catalogue only names a known rule in the reason, + # so a borrowed title never earns the catalogue's name. + decision = routing.decide(state, T, AUTO) + assert decision.code == MODEL_CLOSED and "catalogued noise" not in decision.reason def test_the_rule_id_comes_from_the_vendor_event_and_the_wazuh_rule_name_only() -> None: @@ -184,10 +185,10 @@ def test_the_rule_id_comes_from_the_vendor_event_and_the_wazuh_rule_name_only() assert detection_rule(broken) is None, broken assert noise_family(broken) is None # Unmapped either way; a Wazuh-built alert whose rule cannot be read is vendor_unknown (re-check of #345). - assert model_close_class(broken) in (UNMAPPED, VENDOR_UNKNOWN) + assert model_close_class(broken) in (None, VENDOR_UNKNOWN) # A title, description or rule name alone is never a rule id (a canonical alert without a vendor event). canonical = {"title": DPKG, "description": DPKG, "rule_name": f"2902: {DPKG}", "mitre": {}, "raw": {}} - assert noise_family(canonical) is None and model_close_class(canonical) == UNMAPPED + assert noise_family(canonical) is None and never_close_class(canonical) == UNMAPPED def test_an_oversized_raw_event_keeps_its_rule_through_the_state_cap() -> None: @@ -213,30 +214,26 @@ def test_an_oversized_raw_event_keeps_its_rule_through_the_state_cap() -> None: @pytest.mark.parametrize( ("change", "blocker"), [ - ({"severity": "high"}, "severity"), ({"severity": "critical"}, "severity"), ({"triage.rule_tp": ["prompt_injection_suspected"]}, "steering_text"), ({"analysis.rule_evidence": ["prior_malicious_history"]}, "recent_true_positive"), ({"analysis.rule_corroborated": True}, "recent_true_positive"), ({"correlation.related_dispositions": {"malicious": 1}}, "recent_true_positive"), - ({"correlation": {"related_count": 2, "related_dispositions": {}}}, "campaign_in_progress"), - ({"triage.threat_likelihood": 9}, "fast_tier_not_benign"), - ({"triage.threat_assessment": "suspicious"}, "fast_tier_not_benign"), - ({"triage.llm_used": False}, "fast_tier_not_benign"), - ({"analysis.certainty": 84}, "deep_tier_not_benign"), + ({"triage.threat_assessment": "likely_threat"}, "fast_tier_threat"), ({"analysis.verdict": "inconclusive"}, "deep_tier_not_benign"), + ({"alert.reopened": {"reason": "related_threat"}}, "reopened"), ], ) def test_every_other_condition_still_blocks_a_catalogued_rule(change: Dict[str, Any], blocker: str) -> None: state = wazuh_state(wazuh_event("2902", DPKG), **change) - assert model_close_blockers(state, tier=AUTO, max_likelihood=8, min_certainty=85) == [blocker] + assert model_close_blockers(state, tier=AUTO) == [blocker] assert routing.decide(state, T, AUTO).route != "close" - assert model_close_blockers(state, tier="off", max_likelihood=8, min_certainty=85)[0] == "tier_off" + assert model_close_blockers(state, tier="off")[0] == "tier_off" -def test_a_high_level_catalogued_rule_never_closes_on_the_models_word() -> None: - state = wazuh_state(wazuh_event("2902", DPKG, level=12)) - assert state["severity"] == "high" +def test_a_critical_catalogued_rule_never_closes_on_the_models_word() -> None: + state = wazuh_state(wazuh_event("2902", DPKG, level=14)) + assert state["severity"] == "critical" assert routing.decide(state, T, AUTO).route == "escalate" @@ -255,8 +252,8 @@ def test_a_high_level_catalogued_rule_never_closes_on_the_models_word() -> None: "analysis.threat_score": 90, "analysis.certainty": 90}, # The deep tier says malicious at a low threat score. {"analysis.verdict": "malicious", "analysis.threat_score": 20, "analysis.certainty": 70}, - # Only the fast tier is suspicious; the deep tier was persuaded. - {"triage.threat_assessment": "suspicious", "triage.threat_likelihood": 40}, + # Only the fast tier calls it a likely threat; the deep tier was persuaded. + {"triage.threat_assessment": "likely_threat", "triage.threat_likelihood": 70}, # The deep tier cannot tell. {"analysis.verdict": "inconclusive", "analysis.threat_score": 50, "analysis.certainty": 40}, ] @@ -308,7 +305,7 @@ def test_review_h1_authentication_failures_are_credential_access_whatever_their_ state = wazuh_state(event, severity="medium") assert never_close_class(state["alert"]) == "credential_access" assert model_close_class(state["alert"]) == "credential_access" - assert model_close_blockers(state, tier=AUTO, max_likelihood=8, min_certainty=85) == ["never_close_class"] + assert model_close_blockers(state, tier=AUTO) == ["never_close_class"] assert routing.decide(state, T, AUTO).route == "escalate" # The groups and ids only ever block, so they count whatever format built the alert. pushed = {**state["alert"], "normalizer": "canonical"} diff --git a/tests/unit/invariants/test_checks.py b/tests/unit/invariants/test_checks.py index 9422928c..547b5d72 100644 --- a/tests/unit/invariants/test_checks.py +++ b/tests/unit/invariants/test_checks.py @@ -217,7 +217,10 @@ def test_the_synthetic_routing_states_are_ones_the_policy_would_otherwise_close( t = RoutingThresholds() low = checks._benign_state("low") assert route_after_triage(low, t) == "close" and decide(low, t).route == "close" - assert decide(checks._benign_state("high"), t, "auto_full_audit").code == "false_positive_review" + # Policy v3: a high alert both tiers call benign closes on the models' word at the tier, never with it off. + assert decide(checks._benign_state("high"), t, "auto_full_audit").code == "model_closed" + assert decide(checks._benign_state("high"), t, "off").code == "false_positive_review" + assert decide(checks._benign_state("critical"), t, "auto_full_audit").code == "false_positive_review" assert decide(checks._benign_state("low", evidence=False), t, "auto_full_audit").code == "model_closed" credential = checks._benign_state("low", evidence=False, tactic="Credential Access") assert decide(credential, t, "auto_full_audit").code == "false_positive_review" diff --git a/tests/unit/observability/test_metrics.py b/tests/unit/observability/test_metrics.py index ea2aa7c1..853dc3e2 100644 --- a/tests/unit/observability/test_metrics.py +++ b/tests/unit/observability/test_metrics.py @@ -26,6 +26,8 @@ "lgsotf_llm_skipped_total": ("agent", "reason", "mode"), "lgsotf_enrichment_lookups_total": ("provider", "status", "mode"), "lgsotf_revocation_failures_total": ("path",), + "lgsotf_retro_reopen_failures_total": ("path",), + "lgsotf_retro_reopened_total": ("path",), # Integrations outbox and ticket sync (docs/integrations.md). "lgsotf_integration_deliveries_total": ("kind", "outcome"), "lgsotf_ticket_syncs_total": ("system", "outcome"),