feat(m365): add defender_domain_dmarc_records_published check#11936
Open
Rishi943 wants to merge 2 commits into
Open
feat(m365): add defender_domain_dmarc_records_published check#11936Rishi943 wants to merge 2 commits into
Rishi943 wants to merge 2 commits into
Conversation
Verify each accepted Exchange Online domain publishes a DMARC TXT record at _dmarc.<domain> with an enforcing policy (p=quarantine or p=reject); missing, malformed, and p=none records FAIL. Domains come from Microsoft Graph and the record from a dnspython TXT lookup (dnspython is already a pinned dependency). Fixes prowler-cloud#11800 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
github-actions
Bot
added
provider/m365
📝 WalkthroughWalkthroughAdds an M365 Defender check that retrieves DMARC records for verified Exchange Online domains, evaluates enforcement policies, reports per-domain results, and includes metadata plus unit tests for DNS, pagination, and policy scenarios. ChangesM365 DMARC enforcement check
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Defender
participant MicrosoftGraph
participant DNSResolver
participant DMARCCheck
participant CheckReportM365
Defender->>MicrosoftGraph: query verified domains
MicrosoftGraph-->>Defender: paginated domain data
Defender->>DNSResolver: resolve _dmarc.<domain> TXT
DNSResolver-->>Defender: DMARC TXT record or lookup failure
Defender-->>DMARCCheck: domain_dmarc_configurations
DMARCCheck->>DMARCCheck: parse DMARC enforcement policy
DMARCCheck->>CheckReportM365: create PASS or FAIL report
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
|
✅ No Conflicts No conflict markers, and the branch merges cleanly into its base. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@prowler/providers/m365/services/defender/defender_service.py`:
- Around line 669-696: Update _get_domain_dmarc_configurations and the DMARC
lookup helper to use dnspython’s asynchronous resolver instead of synchronous
dns.resolver.resolve. Make the helper async, create concurrent lookup tasks for
verified domains, and await them with asyncio.gather while preserving each
domain’s DomainDmarcConfiguration and existing error handling.
- Around line 683-684: The verified-domain retrieval in the Defender service
only processes the first Graph page. Update the logic around the domains fetch
and iteration to follow each page via odata_next_link (or use Graph’s
PageIterator), aggregating or processing every returned domain before completing
the DMARC check.
- Around line 88-115: Replace the manual event-loop setup and cleanup in the
Defender service initializer with a direct
asyncio.run(self._get_domain_dmarc_configurations()) call. Remove the
created_loop tracking, loop creation, closed-loop check, running-loop check, and
run_until_complete logic while preserving assignment to
self.domain_dmarc_configurations.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: a095c5a9-3213-4bbb-8fbe-f58a9a08c036
📒 Files selected for processing (7)
prowler/changelog.d/add-defender-domain-dmarc-records-published-check.added.mdprowler/providers/m365/services/defender/defender_domain_dmarc_records_published/__init__.pyprowler/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published.metadata.jsonprowler/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published.pyprowler/providers/m365/services/defender/defender_service.pytests/providers/m365/services/defender/defender_domain_dmarc_records_published/defender_domain_dmarc_records_published_test.pytests/providers/m365/services/defender/m365_defender_service_test.py
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tests/providers/m365/services/defender/m365_defender_service_test.py`:
- Around line 621-665: Add an assertion verifying that
domains_with_url_builder.get was awaited exactly once, alongside the existing
pagination assertions in
test_defender__get_domain_dmarc_configurations_handles_pagination, to confirm
the second page was fetched.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: edb2205f-ea70-4d9e-b790-e85201267e79
📒 Files selected for processing (2)
prowler/providers/m365/services/defender/defender_service.pytests/providers/m365/services/defender/m365_defender_service_test.py
Comment on lines
+621
to
+665
| def test_defender__get_domain_dmarc_configurations_handles_pagination(): | ||
| defender_service = Defender.__new__(Defender) | ||
|
|
||
| domains_page_one = [ | ||
| SimpleNamespace(id="domain1.com", is_verified=True), | ||
| SimpleNamespace(id="unverified.com", is_verified=False), | ||
| ] | ||
| domains_page_two = [ | ||
| SimpleNamespace(id="domain2.com", is_verified=True), | ||
| ] | ||
|
|
||
| domains_response_page_one = SimpleNamespace( | ||
| value=domains_page_one, | ||
| odata_next_link="next-link", | ||
| ) | ||
| domains_response_page_two = SimpleNamespace( | ||
| value=domains_page_two, odata_next_link=None | ||
| ) | ||
|
|
||
| domains_with_url_builder = SimpleNamespace( | ||
| get=AsyncMock(return_value=domains_response_page_two) | ||
| ) | ||
| with_url_mock = MagicMock(return_value=domains_with_url_builder) | ||
|
|
||
| domains_builder = SimpleNamespace( | ||
| get=AsyncMock(return_value=domains_response_page_one), | ||
| with_url=with_url_mock, | ||
| ) | ||
|
|
||
| defender_service.client = SimpleNamespace(domains=domains_builder) | ||
|
|
||
| with mock.patch( | ||
| "prowler.providers.m365.services.defender.defender_service.Defender._get_dmarc_txt_record", | ||
| new=AsyncMock(return_value="v=DMARC1; p=reject"), | ||
| ): | ||
| domain_dmarc_configurations = asyncio.run( | ||
| defender_service._get_domain_dmarc_configurations() | ||
| ) | ||
|
|
||
| assert set(domain_dmarc_configurations) == {"domain1.com", "domain2.com"} | ||
| assert ( | ||
| domain_dmarc_configurations["domain1.com"].dmarc_record == "v=DMARC1; p=reject" | ||
| ) | ||
| assert domains_builder.get.await_count == 1 | ||
| with_url_mock.assert_called_once_with("next-link") |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Add assertion for second-page get() call in pagination test.
The test verifies domains_builder.get.await_count == 1 and with_url_mock.assert_called_once_with("next-link"), but doesn't explicitly assert that the second-page fetch (domains_with_url_builder.get) was awaited. Adding assert domains_with_url_builder.get.await_count == 1 would close the loop on verifying both pages were fetched.
✨ Optional improvement
assert domains_builder.get.await_count == 1
with_url_mock.assert_called_once_with("next-link")
+ assert domains_with_url_builder.get.await_count == 1📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def test_defender__get_domain_dmarc_configurations_handles_pagination(): | |
| defender_service = Defender.__new__(Defender) | |
| domains_page_one = [ | |
| SimpleNamespace(id="domain1.com", is_verified=True), | |
| SimpleNamespace(id="unverified.com", is_verified=False), | |
| ] | |
| domains_page_two = [ | |
| SimpleNamespace(id="domain2.com", is_verified=True), | |
| ] | |
| domains_response_page_one = SimpleNamespace( | |
| value=domains_page_one, | |
| odata_next_link="next-link", | |
| ) | |
| domains_response_page_two = SimpleNamespace( | |
| value=domains_page_two, odata_next_link=None | |
| ) | |
| domains_with_url_builder = SimpleNamespace( | |
| get=AsyncMock(return_value=domains_response_page_two) | |
| ) | |
| with_url_mock = MagicMock(return_value=domains_with_url_builder) | |
| domains_builder = SimpleNamespace( | |
| get=AsyncMock(return_value=domains_response_page_one), | |
| with_url=with_url_mock, | |
| ) | |
| defender_service.client = SimpleNamespace(domains=domains_builder) | |
| with mock.patch( | |
| "prowler.providers.m365.services.defender.defender_service.Defender._get_dmarc_txt_record", | |
| new=AsyncMock(return_value="v=DMARC1; p=reject"), | |
| ): | |
| domain_dmarc_configurations = asyncio.run( | |
| defender_service._get_domain_dmarc_configurations() | |
| ) | |
| assert set(domain_dmarc_configurations) == {"domain1.com", "domain2.com"} | |
| assert ( | |
| domain_dmarc_configurations["domain1.com"].dmarc_record == "v=DMARC1; p=reject" | |
| ) | |
| assert domains_builder.get.await_count == 1 | |
| with_url_mock.assert_called_once_with("next-link") | |
| def test_defender__get_domain_dmarc_configurations_handles_pagination(): | |
| defender_service = Defender.__new__(Defender) | |
| domains_page_one = [ | |
| SimpleNamespace(id="domain1.com", is_verified=True), | |
| SimpleNamespace(id="unverified.com", is_verified=False), | |
| ] | |
| domains_page_two = [ | |
| SimpleNamespace(id="domain2.com", is_verified=True), | |
| ] | |
| domains_response_page_one = SimpleNamespace( | |
| value=domains_page_one, | |
| odata_next_link="next-link", | |
| ) | |
| domains_response_page_two = SimpleNamespace( | |
| value=domains_page_two, odata_next_link=None | |
| ) | |
| domains_with_url_builder = SimpleNamespace( | |
| get=AsyncMock(return_value=domains_response_page_two) | |
| ) | |
| with_url_mock = MagicMock(return_value=domains_with_url_builder) | |
| domains_builder = SimpleNamespace( | |
| get=AsyncMock(return_value=domains_response_page_one), | |
| with_url=with_url_mock, | |
| ) | |
| defender_service.client = SimpleNamespace(domains=domains_builder) | |
| with mock.patch( | |
| "prowler.providers.m365.services.defender.defender_service.Defender._get_dmarc_txt_record", | |
| new=AsyncMock(return_value="v=DMARC1; p=reject"), | |
| ): | |
| domain_dmarc_configurations = asyncio.run( | |
| defender_service._get_domain_dmarc_configurations() | |
| ) | |
| assert set(domain_dmarc_configurations) == {"domain1.com", "domain2.com"} | |
| assert ( | |
| domain_dmarc_configurations["domain1.com"].dmarc_record == "v=DMARC1; p=reject" | |
| ) | |
| assert domains_builder.get.await_count == 1 | |
| with_url_mock.assert_called_once_with("next-link") | |
| assert domains_with_url_builder.get.await_count == 1 |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/providers/m365/services/defender/m365_defender_service_test.py` around
lines 621 - 665, Add an assertion verifying that domains_with_url_builder.get
was awaited exactly once, alongside the existing pagination assertions in
test_defender__get_domain_dmarc_configurations_handles_pagination, to confirm
the second page was fetched.
- Follow odata_next_link when listing domains so tenants with multiple pages are fully covered - Resolve DMARC TXT records concurrently with dns.asyncresolver + asyncio.gather instead of sequential blocking lookups - Add pagination test mirroring admincenter service test style Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Context
Fix #11800
CIS Microsoft 365 Benchmark 2.1.10 / RFC 7489: every accepted Exchange Online domain should publish a DMARC TXT record at
_dmarc.<domain>with an enforcing policy. Without it (or withp=none) attackers can spoof the domain for phishing/BEC.Description
New check
defender_domain_dmarc_records_published(M365 provider,defenderservice, severitymedium):client.domains.get(), same endpoint used by AdminCenter's password-policy path) and performs a DNS TXT lookup at_dmarc.<domain>viadnspython(already a pinned dependency), storing results in a newDomainDmarcConfigurationmodel.p=tag (split on;, case-insensitive, whitespace-tolerant):p=quarantine/p=rejectPASS;p=none, missing record, or a malformed record (notv=DMARC1/ nop=tag) FAIL, each with a distinctstatus_extended.Design note: Graph's
serviceConfigurationRecords(used by the earlier SPF PR #7724) never contains DMARC records — Microsoft does not provision one — so a real DNS lookup is the only way to evaluate this, matching the issue's spec ("DNS TXT record at_dmarc.<domain>"). The event-loop bootstrapping follows the currentadmincenter_service.pypattern.Includes metadata, Google-style docstrings,
List[CheckReportM365]return annotation, service tests + 7 check tests, and a changelog fragment.Steps to review
defender_service.py:_get_domain_dmarc_configurations()(Graph domain list, verified domains only) and_get_dmarc_txt_record()(dnspython TXT lookup, returns the firstv=DMARC1record or None; NXDOMAIN/NoAnswer/Timeout treated as no record).defender_domain_dmarc_records_published.py—_get_dmarc_policyis a small tag parser, no regex.pytest tests/providers/m365/services/defender -p no:randomly— 110 passed (7 new check tests: no domains / reject / quarantine / none / missing / malformed / valid-without-p; plus service-level tests for the Graph fetch and DNS lookup, all mocked — no live DNS in tests).Checklist
Community Checklist
SDK/CLI
License
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
Summary by CodeRabbit
New Features
quarantineorreject, and fail whenp=none, DMARC is missing, or the policy is malformed/unsupported.Tests
p=tag).