mirror of
https://github.com/domainaware/parsedmarc.git
synced 2026-07-16 13:34:56 +00:00
MS Graph: case-insensitive auth_method + ClientAssertion support (#809)
* draft fix MS Graph * Add ClientAssertion auth method support for MS Graph in cli.py MSGraphConnection/AuthMethod (via mailsuite) already supported ClientAssertion, but cli.py never parsed a client_assertion config value or passed it through, so it was unusable from the CLI. Wire config_msgraph.client_assertion through _parse_config and the MSGraphConnection call, and document the auth method (including its short-lived-JWT caveat vs. Certificate/ClientSecret for watch mode). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> --------- Co-authored-by: MISAPOR LAB <misapor@lab.misapor.pl> Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
MISAPOR LAB
Claude Sonnet 5
parent
a6241d537e
commit
fa8faa6b39
@@ -1431,6 +1431,272 @@ certificate_path = /tmp/msgraph-cert.pem
|
||||
mock_graph_connection.assert_not_called()
|
||||
mock_get_mailbox_reports.assert_not_called()
|
||||
|
||||
@patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
|
||||
@patch("parsedmarc.cli.MSGraphConnection")
|
||||
def testCliPassesMsGraphClientAssertionAuthSettings(
|
||||
self, mock_graph_connection, mock_get_mailbox_reports
|
||||
):
|
||||
mock_graph_connection.return_value = object()
|
||||
mock_get_mailbox_reports.return_value = {
|
||||
"aggregate_reports": [],
|
||||
"failure_reports": [],
|
||||
"smtp_tls_reports": [],
|
||||
}
|
||||
|
||||
config_text = """[general]
|
||||
silent = true
|
||||
|
||||
[msgraph]
|
||||
auth_method = ClientAssertion
|
||||
client_id = client-id
|
||||
client_assertion = signed-jwt-assertion
|
||||
tenant_id = tenant-id
|
||||
mailbox = shared@example.com
|
||||
"""
|
||||
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
|
||||
cfg.write(config_text)
|
||||
cfg_path = cfg.name
|
||||
self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
|
||||
|
||||
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]):
|
||||
parsedmarc.cli._main()
|
||||
|
||||
self.assertEqual(
|
||||
mock_graph_connection.call_args.kwargs.get("auth_method"),
|
||||
"ClientAssertion",
|
||||
)
|
||||
self.assertEqual(
|
||||
mock_graph_connection.call_args.kwargs.get("client_assertion"),
|
||||
"signed-jwt-assertion",
|
||||
)
|
||||
self.assertEqual(
|
||||
mock_graph_connection.call_args.kwargs.get("tenant_id"), "tenant-id"
|
||||
)
|
||||
self.assertEqual(
|
||||
mock_graph_connection.call_args.kwargs.get("mailbox"),
|
||||
"shared@example.com",
|
||||
)
|
||||
|
||||
@patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
|
||||
@patch("parsedmarc.cli.MSGraphConnection")
|
||||
@patch("parsedmarc.cli.logger")
|
||||
def testCliRequiresMsGraphClientAssertionForClientAssertionAuth(
|
||||
self, mock_logger, mock_graph_connection, mock_get_mailbox_reports
|
||||
):
|
||||
config_text = """[general]
|
||||
silent = true
|
||||
|
||||
[msgraph]
|
||||
auth_method = ClientAssertion
|
||||
client_id = client-id
|
||||
tenant_id = tenant-id
|
||||
mailbox = shared@example.com
|
||||
"""
|
||||
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
|
||||
cfg.write(config_text)
|
||||
cfg_path = cfg.name
|
||||
self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
|
||||
|
||||
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]):
|
||||
with self.assertRaises(SystemExit) as system_exit:
|
||||
parsedmarc.cli._main()
|
||||
|
||||
self.assertEqual(system_exit.exception.code, -1)
|
||||
mock_logger.critical.assert_called_once_with(
|
||||
"client_assertion setting missing from the msgraph config section"
|
||||
)
|
||||
mock_graph_connection.assert_not_called()
|
||||
mock_get_mailbox_reports.assert_not_called()
|
||||
|
||||
@patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
|
||||
@patch("parsedmarc.cli.MSGraphConnection")
|
||||
@patch("parsedmarc.cli.logger")
|
||||
def testCliRequiresMsGraphTenantIdForClientAssertionAuth(
|
||||
self, mock_logger, mock_graph_connection, mock_get_mailbox_reports
|
||||
):
|
||||
config_text = """[general]
|
||||
silent = true
|
||||
|
||||
[msgraph]
|
||||
auth_method = ClientAssertion
|
||||
client_id = client-id
|
||||
client_assertion = signed-jwt-assertion
|
||||
mailbox = shared@example.com
|
||||
"""
|
||||
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
|
||||
cfg.write(config_text)
|
||||
cfg_path = cfg.name
|
||||
self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
|
||||
|
||||
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]):
|
||||
with self.assertRaises(SystemExit) as system_exit:
|
||||
parsedmarc.cli._main()
|
||||
|
||||
self.assertEqual(system_exit.exception.code, -1)
|
||||
mock_logger.critical.assert_called_once_with(
|
||||
"tenant_id setting missing from the msgraph config section"
|
||||
)
|
||||
mock_graph_connection.assert_not_called()
|
||||
mock_get_mailbox_reports.assert_not_called()
|
||||
|
||||
@patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
|
||||
@patch("parsedmarc.cli.MSGraphConnection")
|
||||
@patch("parsedmarc.cli.logger")
|
||||
def testCliRequiresMsGraphMailboxForClientAssertionAuth(
|
||||
self, mock_logger, mock_graph_connection, mock_get_mailbox_reports
|
||||
):
|
||||
config_text = """[general]
|
||||
silent = true
|
||||
|
||||
[msgraph]
|
||||
auth_method = ClientAssertion
|
||||
client_id = client-id
|
||||
client_assertion = signed-jwt-assertion
|
||||
tenant_id = tenant-id
|
||||
"""
|
||||
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
|
||||
cfg.write(config_text)
|
||||
cfg_path = cfg.name
|
||||
self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
|
||||
|
||||
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]):
|
||||
with self.assertRaises(SystemExit) as system_exit:
|
||||
parsedmarc.cli._main()
|
||||
|
||||
self.assertEqual(system_exit.exception.code, -1)
|
||||
mock_logger.critical.assert_called_once_with(
|
||||
"mailbox setting missing from the msgraph config section"
|
||||
)
|
||||
mock_graph_connection.assert_not_called()
|
||||
mock_get_mailbox_reports.assert_not_called()
|
||||
|
||||
@patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
|
||||
@patch("parsedmarc.cli.MSGraphConnection")
|
||||
def testCliAcceptsLowercaseMsGraphCertificateAuthMethod(
|
||||
self, mock_graph_connection, mock_get_mailbox_reports
|
||||
):
|
||||
"""auth_method values are case-insensitive (e.g. ``certificate``)."""
|
||||
mock_graph_connection.return_value = object()
|
||||
mock_get_mailbox_reports.return_value = {
|
||||
"aggregate_reports": [],
|
||||
"failure_reports": [],
|
||||
"smtp_tls_reports": [],
|
||||
}
|
||||
|
||||
config_text = """[general]
|
||||
silent = true
|
||||
|
||||
[msgraph]
|
||||
auth_method = certificate
|
||||
client_id = client-id
|
||||
tenant_id = tenant-id
|
||||
mailbox = shared@example.com
|
||||
certificate_path = /tmp/msgraph-cert.pem
|
||||
certificate_password = cert-pass
|
||||
"""
|
||||
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
|
||||
cfg.write(config_text)
|
||||
cfg_path = cfg.name
|
||||
self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
|
||||
|
||||
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]):
|
||||
parsedmarc.cli._main()
|
||||
|
||||
self.assertEqual(
|
||||
mock_graph_connection.call_args.kwargs.get("auth_method"), "Certificate"
|
||||
)
|
||||
self.assertEqual(
|
||||
mock_graph_connection.call_args.kwargs.get("mailbox"),
|
||||
"shared@example.com",
|
||||
)
|
||||
self.assertEqual(
|
||||
mock_graph_connection.call_args.kwargs.get("certificate_path"),
|
||||
"/tmp/msgraph-cert.pem",
|
||||
)
|
||||
self.assertEqual(
|
||||
mock_graph_connection.call_args.kwargs.get("certificate_password"),
|
||||
"cert-pass",
|
||||
)
|
||||
|
||||
@patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
|
||||
@patch("parsedmarc.cli.MSGraphConnection")
|
||||
@patch("parsedmarc.cli.logger")
|
||||
def testCliRejectsInvalidMsGraphAuthMethod(
|
||||
self, mock_logger, mock_graph_connection, mock_get_mailbox_reports
|
||||
):
|
||||
config_text = """[general]
|
||||
silent = true
|
||||
|
||||
[msgraph]
|
||||
auth_method = NotARealMethod
|
||||
client_id = client-id
|
||||
tenant_id = tenant-id
|
||||
mailbox = shared@example.com
|
||||
certificate_path = /tmp/msgraph-cert.pem
|
||||
"""
|
||||
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
|
||||
cfg.write(config_text)
|
||||
cfg_path = cfg.name
|
||||
self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
|
||||
|
||||
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]):
|
||||
with self.assertRaises(SystemExit) as system_exit:
|
||||
parsedmarc.cli._main()
|
||||
|
||||
self.assertEqual(system_exit.exception.code, -1)
|
||||
mock_logger.critical.assert_called_once()
|
||||
critical_message = mock_logger.critical.call_args.args[0]
|
||||
self.assertIn("Invalid msgraph auth_method", critical_message)
|
||||
self.assertIn("NotARealMethod", critical_message)
|
||||
mock_graph_connection.assert_not_called()
|
||||
mock_get_mailbox_reports.assert_not_called()
|
||||
|
||||
@patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
|
||||
@patch("parsedmarc.cli.MSGraphConnection")
|
||||
def testCliLogsMsGraphConnectionAttempt(
|
||||
self, mock_graph_connection, mock_get_mailbox_reports
|
||||
):
|
||||
"""An INFO log is emitted when parsedmarc starts a Graph connection."""
|
||||
mock_graph_connection.return_value = object()
|
||||
mock_get_mailbox_reports.return_value = {
|
||||
"aggregate_reports": [],
|
||||
"failure_reports": [],
|
||||
"smtp_tls_reports": [],
|
||||
}
|
||||
|
||||
config_text = """[general]
|
||||
silent = true
|
||||
|
||||
[msgraph]
|
||||
auth_method = Certificate
|
||||
client_id = client-id
|
||||
tenant_id = tenant-id
|
||||
mailbox = shared@example.com
|
||||
certificate_path = /tmp/msgraph-cert.pem
|
||||
"""
|
||||
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
|
||||
cfg.write(config_text)
|
||||
cfg_path = cfg.name
|
||||
self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
|
||||
|
||||
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, "--verbose"]):
|
||||
with self.assertLogs("parsedmarc.log", level="INFO") as cm:
|
||||
parsedmarc.cli._main()
|
||||
|
||||
self.assertTrue(
|
||||
any(
|
||||
"Connecting to Microsoft Graph mailbox shared@example.com" in message
|
||||
for message in cm.output
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class TestSighupReload(unittest.TestCase):
|
||||
"""Tests for SIGHUP-driven configuration reload in watch mode."""
|
||||
|
||||
Reference in New Issue
Block a user