feat: add SIEM integration templates
This commit is contained in:
@@ -0,0 +1,77 @@
|
||||
import json
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.services.siem_templates import (
|
||||
SIEM_EVENT_TYPES,
|
||||
SIEM_SCHEMA_VERSION,
|
||||
get_siem_templates,
|
||||
validate_siem_event,
|
||||
)
|
||||
|
||||
|
||||
def _string_values(value):
|
||||
if isinstance(value, dict):
|
||||
for item in value.values():
|
||||
yield from _string_values(item)
|
||||
elif isinstance(value, list):
|
||||
for item in value:
|
||||
yield from _string_values(item)
|
||||
elif isinstance(value, str):
|
||||
yield value
|
||||
|
||||
|
||||
def test_siem_template_bundle_is_versioned_and_examples_match_schema():
|
||||
"""Bundled SIEM examples stay aligned with the stable event envelope."""
|
||||
templates = get_siem_templates()
|
||||
|
||||
assert templates["schema_version"] == SIEM_SCHEMA_VERSION
|
||||
assert templates["event_schema"]["properties"]["schema_version"]["const"] == SIEM_SCHEMA_VERSION
|
||||
assert templates["event_schema"]["properties"]["event_type"]["enum"] == SIEM_EVENT_TYPES
|
||||
|
||||
examples = templates["event_examples"]
|
||||
assert {"sender_new", "compliance_drop", "alert_created"}.issubset(examples)
|
||||
|
||||
for name, example in examples.items():
|
||||
assert validate_siem_event(example) == [], name
|
||||
encoded = json.dumps(list(_string_values(example))).lower()
|
||||
assert "secret" not in encoded
|
||||
assert "token" not in encoded
|
||||
assert "raw_report_xml" not in encoded
|
||||
|
||||
|
||||
def test_siem_ingestion_examples_wrap_valid_dmarq_events():
|
||||
"""SIEM-specific examples keep the normalized event intact."""
|
||||
templates = get_siem_templates()
|
||||
ingestion_examples = templates["ingestion_examples"]
|
||||
|
||||
splunk_event = ingestion_examples["splunk_hec"]["event"]
|
||||
elastic_event = ingestion_examples["elastic_ecs"]["dmarq"]
|
||||
sentinel_event = ingestion_examples["microsoft_sentinel_custom_log"][0]["DmarqEvent"]
|
||||
|
||||
for wrapped_event in [splunk_event, elastic_event, sentinel_event]:
|
||||
assert validate_siem_event(wrapped_event) == []
|
||||
assert wrapped_event["schema_version"] == SIEM_SCHEMA_VERSION
|
||||
|
||||
|
||||
def test_siem_templates_endpoint_returns_common_configs(authed_client: TestClient):
|
||||
"""Administrators can fetch schemas, examples, and SIEM config hints."""
|
||||
response = authed_client.get("/api/v1/integrations/siem/templates")
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["schema_version"] == SIEM_SCHEMA_VERSION
|
||||
assert set(body["config_templates"]) == {
|
||||
"splunk_hec",
|
||||
"elastic_logstash_http",
|
||||
"microsoft_sentinel",
|
||||
}
|
||||
assert body["event_examples"]["compliance_drop"]["redaction"]["pii_redacted"] is True
|
||||
assert "raw report" in " ".join(body["redaction_guidance"]).lower()
|
||||
|
||||
|
||||
def test_siem_templates_endpoint_requires_admin_auth(client: TestClient):
|
||||
"""Template endpoint follows the same admin boundary as other integrations."""
|
||||
response = client.get("/api/v1/integrations/siem/templates")
|
||||
|
||||
assert response.status_code == 401
|
||||
Reference in New Issue
Block a user