amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-07-16 05:24:56 +00:00
Code Issues Projects Releases Wiki Activity
Files
df9bf82e04fe949ac06480d61ef5ea8a8d1041b7
parsedmarc/tests/test_splunk.py
T
Sean WhalenGitHubClaude Fable 5
 df9bf82e04 Post-review follow-ups for Graph send (#825/#826) and requests-to-httpx migration (#827) 
Follow-ups from the review of PR #825 (whose implementation had already
landed on master via #826's stacked merge):

- Honor the documented [smtp] attachment and [smtp] message options.
  Both were parsed into opts but never passed to either summary-email
  transport (also broken in released 10.2.2), so a configured custom
  attachment filename or message body was silently ignored. Both the
  SMTP and Microsoft Graph transports now receive them, and the missing
  smtp_attachment Namespace default is added (also covers SIGHUP
  reload, which rebuilds opts from the CLI Namespace).
- Don't mislabel non-Graph mailbox errors as Microsoft Graph failures:
  the shared mailbox-fetch and watch handlers now log a generic
  "Mailbox Error" with traceback when the connection isn't Graph.
- Declare microsoft-kiota-abstractions as a direct dependency (imported
  directly in cli.py for Graph error handling; previously transitive).

Migrate all runtime HTTP from requests to httpx (webhook client, Splunk
HEC client, and the PSL-overrides / IP-database / reverse-DNS-map /
IPinfo-API fetches in utils.py):

- follow_redirects=True everywhere to preserve requests' default
  redirect-following; httpx does not follow redirects by default.
- The PSL-overrides and reverse-DNS-map fetches gain a 60s timeout
  (previously none), matching the IP-database fetch.
- response.ok -> response.is_success; requests.RequestException ->
  httpx.HTTPError; raw string bodies use content= (httpx's data= is
  form-encoding only); Splunk HEC verification moves to client
  construction (httpx has no per-request verify).
- requests drops out of [project] dependencies and moves to the [build]
  extra for the out-of-wheel maintainer script collect_domain_info.py,
  which deliberately stays on requests/urllib3 for its permissive-TLS
  adapter.
- Remove the requests-era module-level
  urllib3.disable_warnings(InsecureRequestWarning) in splunk.py; httpx
  doesn't route through urllib3, so its only remaining effect was
  globally silencing insecure-TLS warnings from other urllib3-based
  components as an import side effect. Nothing imports urllib3 directly
  anymore, so it also leaves [project] dependencies.

Tests: config-to-transport wiring for attachment/message on both
transports (including defaults), non-Graph errors keep the generic log
line, webhook/Splunk payload assertions moved to content=, and Splunk
verify asserted at httpx.Client construction. 736 passed; ruff and
pyright clean.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-14 16:11:47 -04:00

490 lines
19 KiB
Python
Raw Blame History

"""Tests for parsedmarc.splunk"""
import json
import time
import unittest
from unittest.mock import MagicMock, patch
from parsedmarc.splunk import HECClient, SplunkError
from tests.tzutil import force_tz
def _aggregate_report():
return {
"report_metadata": {
"org_name": "TestOrg",
"org_email": "dmarc@example.com",
"report_id": "agg-1",
"begin_date": "2024-01-01 00:00:00",
"end_date": "2024-01-02 00:00:00",
},
"policy_published": {"domain": "example.com", "p": "none"},
"records": [
{
"interval_begin": "2024-01-01 00:00:00",
"interval_end": "2024-01-02 00:00:00",
"normalized_timespan": False,
"source": {
"ip_address": "192.0.2.1",
"country": "US",
"reverse_dns": None,
"base_domain": None,
"name": None,
"type": None,
"asn": 64496,
"as_name": "Example AS",
"as_domain": "example.net",
},
"count": 4,
"alignment": {"spf": True, "dkim": True, "dmarc": True},
"policy_evaluated": {
"disposition": "none",
"dkim": "pass",
"spf": "pass",
"policy_override_reasons": [],
},
"identifiers": {
"header_from": "example.com",
"envelope_from": "example.com",
"envelope_to": None,
},
"auth_results": {
"dkim": [
{
"domain": "example.com",
"selector": "s",
"result": "pass",
"human_result": None,
}
],
"spf": [
{
"domain": "example.com",
"scope": "mfrom",
"result": "pass",
"human_result": None,
}
],
},
}
],
}
def _failure_report():
return {
"feedback_type": "auth-failure",
"user_agent": "test/1.0",
"version": "1",
"original_envelope_id": None,
"original_mail_from": "x@example.com",
"original_rcpt_to": None,
"arrival_date": "Thu, 1 Jan 2024 00:00:00 +0000",
"arrival_date_utc": "2024-01-01 00:00:00",
"authentication_results": None,
"delivery_result": "other",
"auth_failure": ["dmarc"],
"authentication_mechanisms": [],
"dkim_domain": None,
"reported_domain": "example.com",
"sample_headers_only": True,
"source": {
"ip_address": "192.0.2.5",
"country": "US",
"reverse_dns": None,
"base_domain": None,
"name": None,
"type": None,
"asn": 64496,
"as_name": "Example AS",
"as_domain": "example.net",
},
"sample": "...",
"parsed_sample": {"subject": "Test"},
}
def _smtp_tls_report():
return {
"organization_name": "example.com",
"begin_date": "2024-02-03T00:00:00Z",
"end_date": "2024-02-04T00:00:00Z",
"contact_info": "tls@example.com",
"report_id": "tls-1",
"policies": [
{
"policy_domain": "example.com",
"policy_type": "sts",
"successful_session_count": 100,
"failed_session_count": 0,
}
],
}
def _ok_response():
"""Splunk HEC success response shape: {"code": 0, ...}."""
r = MagicMock()
r.json.return_value = {"code": 0, "text": "Success"}
return r
def _client():
return HECClient(
url="https://splunk.example.com:8088",
access_token="abc-token-uuid",
index="dmarc",
)
class TestHECClientInit(unittest.TestCase):
"""The HEC URL is rebuilt from the user-supplied URL into the
/services/collector/event/1.0 endpoint, and the Authorization
header is set to `Splunk <token>`."""
def test_url_rewritten_to_collector_endpoint(self):
"""A user may supply any URL on the Splunk host; the client
rewrites to the documented HEC path."""
client = HECClient(
url="https://splunk.example.com:8088/some/random/path",
access_token="t",
index="dmarc",
)
self.assertEqual(
client.url, "https://splunk.example.com:8088/services/collector/event/1.0"
)
def test_authorization_header_uses_splunk_prefix(self):
client = HECClient(url="https://h:8088", access_token="my-token", index="dmarc")
self.assertEqual(client.session.headers["Authorization"], "Splunk my-token")
def test_user_agent_header_is_set(self):
client = HECClient(url="https://h:8088", access_token="my-token", index="dmarc")
self.assertIn("parsedmarc", client.session.headers["User-Agent"])
def test_token_with_splunk_prefix_is_normalized(self):
"""If a user pastes `Splunk <token>` from the Splunk UI into
config, the constructor strips the prefix so the resulting
Authorization header isn't `Splunk Splunk <token>`."""
client = HECClient(
url="https://h:8088",
access_token="Splunk abc-token-uuid",
index="dmarc",
)
self.assertEqual(client.access_token, "abc-token-uuid")
def test_token_without_prefix_is_unchanged(self):
"""The lstrip("Splunk ") implementation has character-set
semantics, not prefix semantics — it happens to work for the
UUID-shaped tokens HEC issues (none of S/p/l/u/n/k/space
appear in a UUID's hex character set). A token containing
only hex digits and dashes is unchanged."""
client = HECClient(
url="https://h:8088",
access_token="abc-token-uuid",
index="dmarc",
)
self.assertEqual(client.access_token, "abc-token-uuid")
def test_common_data_carries_host_source_and_index(self):
"""Splunk events inherit these three top-level fields. A
regression here would mis-route events to the wrong index."""
client = HECClient(
url="https://h:8088", access_token="t", index="dmarc", source="my-source"
)
self.assertEqual(client._common_data["index"], "dmarc")
self.assertEqual(client._common_data["source"], "my-source")
# host defaults to socket.getfqdn(); non-empty is enough.
self.assertTrue(client._common_data["host"])
class TestSaveAggregateReportsToSplunk(unittest.TestCase):
"""Each record is emitted as a separate Splunk event, with the
record's interval_begin as the event timestamp, the report's
metadata flattened onto the event, and sourcetype dmarc:aggregate."""
def test_sends_one_event_per_record(self):
"""Two-record report → two newline-separated events in the POST body."""
client = _client()
report = _aggregate_report()
report["records"].append(report["records"][0].copy())
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_aggregate_reports_to_splunk(report)
body = client.session.post.call_args.kwargs["content"]
events = [json.loads(line) for line in body.strip().split("\n")]
self.assertEqual(len(events), 2)
for event in events:
self.assertEqual(event["sourcetype"], "dmarc:aggregate")
self.assertEqual(event["index"], "dmarc")
def test_event_payload_carries_source_metadata(self):
"""The flattened event includes source attribution fields a
Splunk dashboard would filter on."""
client = _client()
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_aggregate_reports_to_splunk(_aggregate_report())
body = client.session.post.call_args.kwargs["content"]
event = json.loads(body.strip())["event"]
self.assertEqual(event["source_ip_address"], "192.0.2.1")
self.assertEqual(event["header_from"], "example.com")
self.assertEqual(event["message_count"], 4)
self.assertEqual(event["passed_dmarc"], True)
self.assertEqual(event["org_name"], "TestOrg")
def test_event_includes_published_policy(self):
client = _client()
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_aggregate_reports_to_splunk(_aggregate_report())
event = json.loads(client.session.post.call_args.kwargs["content"].strip())[
"event"
]
self.assertEqual(
event["published_policy"], {"domain": "example.com", "p": "none"}
)
def test_dict_input_normalized_to_list(self):
client = _client()
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_aggregate_reports_to_splunk(_aggregate_report())
client.session.post.assert_called_once()
def test_empty_list_is_a_noop(self):
client = _client()
client.session = MagicMock()
client.save_aggregate_reports_to_splunk([])
client.session.post.assert_not_called()
def test_post_uses_session_timeout(self):
"""httpx has no per-request verify= kwarg — verification is set
at client construction (see test_client_constructed_with_verify_
false below), so only the per-request timeout is asserted here."""
client = HECClient(
url="https://h:8088",
access_token="t",
index="dmarc",
verify=False,
timeout=15,
)
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_aggregate_reports_to_splunk(_aggregate_report())
kwargs = client.session.post.call_args.kwargs
self.assertNotIn("verify", kwargs)
self.assertEqual(kwargs["timeout"], 15)
def test_client_constructed_with_verify_false(self):
"""verify=False must disable TLS verification on the underlying
httpx.Client at construction time, since httpx.Client does not
accept a per-request verify= kwarg like requests.Session did.
Mocked at the httpx SDK boundary (httpx.Client itself)."""
with patch("parsedmarc.splunk.httpx.Client") as mock_client_cls:
HECClient(
url="https://h:8088",
access_token="t",
index="dmarc",
verify=False,
timeout=15,
)
_, kwargs = mock_client_cls.call_args
self.assertEqual(kwargs["verify"], False)
def test_non_zero_response_code_raises_splunk_error(self):
"""HEC returns code=0 on success and non-zero codes for
token/index/format errors. The error text from HEC carries
the diagnosis and is propagated."""
client = _client()
client.session = MagicMock()
bad = MagicMock()
bad.json.return_value = {"code": 4, "text": "Invalid token"}
client.session.post.return_value = bad
with self.assertRaises(SplunkError) as ctx:
client.save_aggregate_reports_to_splunk(_aggregate_report())
self.assertIn("Invalid token", str(ctx.exception))
def test_post_exception_translates_to_splunk_error(self):
client = _client()
client.session = MagicMock()
client.session.post.side_effect = OSError("network")
with self.assertRaises(SplunkError) as ctx:
client.save_aggregate_reports_to_splunk(_aggregate_report())
self.assertIn("network", str(ctx.exception))
@unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
def test_event_time_treats_interval_begin_as_utc(self):
"""interval_begin is a UTC wall-clock string; the HEC event
`time` must be its true UTC epoch regardless of the host
timezone. Regression test for
https://github.com/domainaware/parsedmarc/issues/819: the naive
parse used to shift the epoch by the host's UTC offset."""
force_tz(self)
client = _client()
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_aggregate_reports_to_splunk(_aggregate_report())
event_wrapper = json.loads(
client.session.post.call_args.kwargs["content"].strip()
)
self.assertEqual(event_wrapper["time"], 1704067200)
class TestSaveFailureReportsToSplunk(unittest.TestCase):
def test_sends_one_event_per_report(self):
client = _client()
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_failure_reports_to_splunk([_failure_report(), _failure_report()])
events = [
json.loads(line)
for line in client.session.post.call_args.kwargs["content"]
.strip()
.split("\n")
]
self.assertEqual(len(events), 2)
for event in events:
self.assertEqual(event["sourcetype"], "dmarc:failure")
def test_event_payload_is_the_report_dict(self):
client = _client()
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_failure_reports_to_splunk(_failure_report())
event = json.loads(client.session.post.call_args.kwargs["content"].strip())[
"event"
]
self.assertEqual(event["reported_domain"], "example.com")
def test_empty_list_is_a_noop(self):
client = _client()
client.session = MagicMock()
client.save_failure_reports_to_splunk([])
client.session.post.assert_not_called()
@unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
def test_event_time_treats_arrival_date_utc_as_utc(self):
"""arrival_date_utc is a UTC wall-clock string; the HEC event
`time` must be its true UTC epoch regardless of the host
timezone. Regression test for
https://github.com/domainaware/parsedmarc/issues/811 (bug 1):
the naive parse used to shift the epoch by the host's UTC
offset (-3600 s under Europe/Warsaw in January)."""
force_tz(self)
client = _client()
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_failure_reports_to_splunk(_failure_report())
event = json.loads(client.session.post.call_args.kwargs["content"].strip())
# Fixture arrival_date_utc is 2024-01-01 00:00:00 UTC.
self.assertEqual(event["time"], 1704067200)
def test_non_zero_response_code_raises_splunk_error(self):
client = _client()
client.session = MagicMock()
bad = MagicMock()
bad.json.return_value = {"code": 6, "text": "Invalid data format"}
client.session.post.return_value = bad
with self.assertRaises(SplunkError):
client.save_failure_reports_to_splunk(_failure_report())
def test_post_exception_translates_to_splunk_error(self):
client = _client()
client.session = MagicMock()
client.session.post.side_effect = RuntimeError("conn refused")
with self.assertRaises(SplunkError):
client.save_failure_reports_to_splunk(_failure_report())
def test_verify_false_logs_skip_message(self):
"""verify=False should leave a debug breadcrumb so operators
can spot misconfigured TLS in their logs."""
client = HECClient(
url="https://h:8088", access_token="t", index="dmarc", verify=False
)
client.session = MagicMock()
client.session.post.return_value = _ok_response()
with self.assertLogs("parsedmarc.log", level="DEBUG") as cm:
client.save_failure_reports_to_splunk(_failure_report())
self.assertTrue(
any("Skipping certificate verification" in m for m in cm.output)
)
class TestSaveSmtpTlsReportsToSplunk(unittest.TestCase):
def test_sends_one_event_per_report(self):
client = _client()
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_smtp_tls_reports_to_splunk([_smtp_tls_report()])
events = [
json.loads(line)
for line in client.session.post.call_args.kwargs["content"]
.strip()
.split("\n")
]
self.assertEqual(len(events), 1)
self.assertEqual(events[0]["sourcetype"], "smtp:tls")
def test_dict_input_normalized_to_list(self):
client = _client()
client.session = MagicMock()
client.session.post.return_value = _ok_response()
client.save_smtp_tls_reports_to_splunk(_smtp_tls_report())
client.session.post.assert_called_once()
def test_empty_list_is_a_noop(self):
client = _client()
client.session = MagicMock()
client.save_smtp_tls_reports_to_splunk([])
client.session.post.assert_not_called()
def test_non_zero_response_code_raises_splunk_error(self):
client = _client()
client.session = MagicMock()
bad = MagicMock()
bad.json.return_value = {"code": 7, "text": "Incorrect index"}
client.session.post.return_value = bad
with self.assertRaises(SplunkError):
client.save_smtp_tls_reports_to_splunk(_smtp_tls_report())
def test_post_exception_translates_to_splunk_error(self):
client = _client()
client.session = MagicMock()
client.session.post.side_effect = RuntimeError("conn refused")
with self.assertRaises(SplunkError):
client.save_smtp_tls_reports_to_splunk(_smtp_tls_report())
def test_verify_false_logs_skip_message(self):
client = HECClient(
url="https://h:8088", access_token="t", index="dmarc", verify=False
)
client.session = MagicMock()
client.session.post.return_value = _ok_response()
with self.assertLogs("parsedmarc.log", level="DEBUG") as cm:
client.save_smtp_tls_reports_to_splunk(_smtp_tls_report())
self.assertTrue(
any("Skipping certificate verification" in m for m in cm.output)
)
class TestHECClientClose(unittest.TestCase):
def test_close_closes_session(self):
client = _client()
client.session = MagicMock()
client.close()
client.session.close.assert_called_once()
class TestSplunkBackwardCompatAlias(unittest.TestCase):
def test_forensic_alias_points_to_failure_method(self):
self.assertIs(
HECClient.save_forensic_reports_to_splunk,
HECClient.save_failure_reports_to_splunk,
)
if __name__ == "__main__":
unittest.main(verbosity=2)
Reference in New Issue View Git Blame Copy Permalink