DMARCbis support: new tags, removed tag warnings, DNS tree walk, bug fixes (#236)

* Initial plan

* Add DMARCbis support: new tags (np, psd, t), removed tag warnings, optional p tag, DNS tree walk

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

* Add comprehensive unit tests for DMARCbis support

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

* Address code review feedback: fix np default logic, simplify conditionals

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

* Fix pre-existing bugs: populate allowed_values in get_dmarc_tag_description, fix f-string/format mix-up

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

* Fix missing raise keywords in mta_sts.py and smtp_tls_reporting.py, add comprehensive tests

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

* Address code review: fix fragile string parsing, move import to top of file

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

* Replace RFCPLACEHOLDER DMARCbis

* Remove unused import of MagicMock from unittest.mock in tests.py

* Fix tests

* Update checkdmarc/dmarc.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update tests.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Apply optional review fixes: fix test docstrings, deterministic warning order, remove network dependency from tree walk test

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

* Normalize domain at start of query_dmarc_record() to handle unicode, zero-width chars, and trailing dots

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: 3 people

checkdmarc/dmarc.py

@@ -162,13 +162,16 @@ class DMARCTagMap(TypedDict):
     adkim: DMARCTagMapItemWithDefault
     aspf: DMARCTagMapItemWithDefault
     fo: DMARCTagMapItemWithDefaultAndValues
+    np: DMARCTagMapItemWithValues
     p: DMARCTagMapItemWithValues
     pct: DMARCTagMapItemWithDefault
+    psd: DMARCTagMapItemWithDefaultAndValues
     rf: DMARCTagMapItemWithDefaultAndValues
     ri: DMARCTagMapItemWithDefault
     rua: DMARCTagMapItem
     ruf: DMARCTagMapItem
     sp: DMARCTagMapItemWithValues
+    t: DMARCTagMapItemWithDefaultAndValues
     v: DMARCTagMapItem
 
 
@@ -384,9 +387,54 @@ class DMARCErrorData(TypedDict, total=False):
             ),
         },
     },
+    "np": {
+        "name": "Non-existent Subdomain Policy",
+        "required": False,
+        "description": (
+            "Indicates the policy to "
+            "be enacted by the "
+            "Receiver at the request "
+            "of the Domain Owner for "
+            "non-existent subdomains "
+            "of the domain queried. "
+            "Its syntax is identical "
+            'to that of the "p" tag. '
+            "If absent, the policy "
+            'specified by the "sp" tag '
+            "(if present) or the "
+            '"p" tag MUST be applied '
+            "for non-existent subdomains."
+        ),
+        "values": {
+            "none": (
+                "The Domain Owner requests "
+                "no specific action be "
+                "taken regarding delivery "
+                "of messages."
+            ),
+            "quarantine": (
+                "The Domain Owner "
+                "wishes to have "
+                "email that fails "
+                "the DMARC mechanism "
+                "check be treated by "
+                "Mail Receivers as "
+                "suspicious."
+            ),
+            "reject": (
+                "The Domain Owner wishes "
+                "for Mail Receivers to "
+                "reject email that fails "
+                "the DMARC mechanism check. "
+                "Rejection SHOULD occur "
+                "during the SMTP "
+                "transaction."
+            ),
+        },
+    },
     "p": {
         "name": "Requested Mail Receiver Policy",
-        "required": True,
+        "required": False,
         "description": (
             "Specifies the policy to "
             "be enacted by the "
@@ -455,8 +503,46 @@ class DMARCErrorData(TypedDict, total=False):
             "Domain Owners to enact "
             "a slow rollout of "
             "enforcement of the "
-            "DMARC mechanism."
+            "DMARC mechanism. "
+            "Removed in DMARCbis."
+        ),
+    },
+    "psd": {
+        "name": "PSD Flag",
+        "required": False,
+        "default": "u",
+        "description": (
+            "A flag indicating "
+            "whether the domain is a "
+            "Public Suffix Domain (PSD)."
         ),
+        "values": {
+            "y": (
+                "The domain is a PSD. "
+                "This tag is included "
+                "by PSOs to indicate "
+                "that the domain is a "
+                "Public Suffix Domain."
+            ),
+            "n": (
+                "The DMARC Policy Record "
+                "is published for a domain "
+                "that is not a PSD, but it "
+                "is the Organizational "
+                "Domain for itself and "
+                "its subdomains."
+            ),
+            "u": (
+                "The default indicates "
+                "that the DMARC Policy "
+                "Record is published for "
+                "a domain that is not a "
+                "PSD, and may or may not "
+                "be an Organizational "
+                "Domain for itself and "
+                "its subdomains."
+            ),
+        },
     },
     "rf": {
         "name": "Report Format",
@@ -476,7 +562,8 @@ class DMARCErrorData(TypedDict, total=False):
             "(the auth-failure report "
             "type) is currently "
             "supported in the "
-            "DMARC standard."
+            "DMARC standard. "
+            "Removed in DMARCbis."
         ),
         "values": {
             "afrf": (
@@ -509,7 +596,8 @@ class DMARCErrorData(TypedDict, total=False):
             "than a daily report is "
             "understood to be "
             "accommodated on a "
-            "best-effort basis."
+            "best-effort basis. "
+            "Removed in DMARCbis."
         ),
     },
     "rua": {
@@ -585,6 +673,46 @@ class DMARCErrorData(TypedDict, total=False):
             ),
         },
     },
+    "t": {
+        "name": "DMARC Policy Test Mode",
+        "required": False,
+        "default": "n",
+        "description": (
+            "Signals whether or not "
+            "the Domain Owner wishes "
+            "the Domain Owner Assessment "
+            "Policy declared in the "
+            '"p", "sp", and/or "np" tags '
+            "to actually be applied. "
+            "This tag does not affect "
+            "the generation of DMARC "
+            "reports, and it has no "
+            "effect on any policy that "
+            'is "none".'
+        ),
+        "values": {
+            "y": (
+                "A request that the actor "
+                "performing the DMARC "
+                "validation check not "
+                "apply the policy, but "
+                "instead apply any special "
+                "handling rules it might "
+                "have in place. The Domain "
+                "Owner is currently testing "
+                "its specified DMARC "
+                "assessment policy."
+            ),
+            "n": (
+                "The default is a request "
+                "to apply the Domain Owner "
+                "Assessment Policy as "
+                "specified to any message "
+                "that produces a DMARC "
+                '"fail" result.'
+            ),
+        },
+    },
     "v": {
         "name": "Version",
         "required": True,
@@ -692,7 +820,7 @@ def _query_dmarc_record(
         except dns.resolver.NoAnswer:
             pass
         except dns.resolver.NXDOMAIN:
-            raise DMARCRecordNotFound(f"The domain {0} does not exist.".format(domain))
+            raise DMARCRecordNotFound(f"The domain {domain} does not exist.")
         except Exception as error:
             raise DMARCRecordNotFound(error)
 
@@ -744,10 +872,10 @@ def query_dmarc_record(
         :exc:`checkdmarc.dmarc.SPFRecordFoundWhereDMARCRecordShouldBe`
 
     """
+    domain = normalize_domain(domain).rstrip(".")
     logging.debug(f"Checking for a DMARC record on {domain}")
     warnings = []
-    base_domain = get_base_domain(domain)
-    location = domain.lower()
+    location = domain
 
     try:
         record = _query_dmarc_record(
@@ -759,7 +887,7 @@ def query_dmarc_record(
             ignore_unrelated_records=ignore_unrelated_records,
         )
     except DMARCRecordNotFound:
-        # Skip this exception as we want to query the base domain. If we fail
+        # Skip this exception as we want to perform a tree walk. If we fail
         # at that, at the end of this function we will raise another
         # DMARCRecordNotFound.
         record = None
@@ -781,22 +909,51 @@ def query_dmarc_record(
     except dns.exception.DNSException:
         pass
 
-    if record is None and domain != base_domain:
-        record = _query_dmarc_record(
-            base_domain,
-            nameservers=nameservers,
-            resolver=resolver,
-            timeout=timeout,
-            timeout_retries=timeout_retries,
-            ignore_unrelated_records=ignore_unrelated_records,
-        )
-        location = base_domain
+    # DMARCbis DNS tree walk for DMARC policy discovery
+    if record is None:
+        labels = domain.split(".")
+        num_labels = len(labels)
+        if num_labels > 1:
+            # Determine starting point for tree walk
+            if num_labels <= 8:
+                # Start from the parent domain (remove leftmost label)
+                start = 1
+            else:
+                # Skip to 7 labels
+                start = num_labels - 7
+            # Walk up the tree
+            for i in range(start, num_labels):
+                parent = ".".join(labels[i:])
+                if "." not in parent:
+                    # Don't query TLDs
+                    break
+                try:
+                    record = _query_dmarc_record(
+                        parent,
+                        nameservers=nameservers,
+                        resolver=resolver,
+                        timeout=timeout,
+                        timeout_retries=timeout_retries,
+                        ignore_unrelated_records=ignore_unrelated_records,
+                    )
+                    if record is not None:
+                        location = parent
+                        break
+                except DMARCRecordNotFound:
+                    # No DMARC record at this parent; continue walking up the tree
+                    continue
+                except DMARCError:
+                    # A DMARC record exists but is invalid or otherwise problematic;
+                    # re-raise so the caller can surface the actual configuration error.
+                    raise
+
     if record is None:
         error_str = "A DMARC record does not exist"
-        if domain == base_domain:
+        labels = domain.split(".")
+        if len(labels) <= 2:
             error_str += "."
         else:
-            error_str += " for this subdomain or its base domain."
+            error_str += " for this domain or its parent domains."
         raise DMARCRecordNotFound(error_str)
 
     return {"record": record, "location": location, "warnings": warnings}
@@ -825,6 +982,8 @@ def get_dmarc_tag_description(
     allowed_values = {}
     if "default" in dmarc_tags[tag]:
         default = dmarc_tags[tag]["default"]
+    if "values" in dmarc_tags[tag]:
+        allowed_values = dmarc_tags[tag]["values"]
     if type(value) is str and value in allowed_values:
         description = allowed_values[value]
     elif type(value) is list and len(allowed_values):
@@ -1185,15 +1344,36 @@ def parse_dmarc_record(
         if tag not in tags and "default" in dmarc_tags[tag]:
             tags[tag] = {"value": dmarc_tags[tag]["default"], "explicit": False}
     if "p" not in tags:
-        raise DMARCSyntaxError('The record is missing the required policy ("p") tag.')
+        tags["p"] = {"value": "none", "explicit": False}
+        warnings.append(
+            "The p tag is optional in DMARCbis, but is required "
+            "in older versions of DMARC."
+        )
     tags["p"]["value"] = tags["p"]["value"].lower()
     if "sp" not in tags:
         tags["sp"] = {"value": tags["p"]["value"], "explicit": False}
     # Normalize sp value for validation consistency (mirrors p behavior)
     tags["sp"]["value"] = tags["sp"]["value"].lower()
-    if list(tags.keys())[1] != "p":
-        raise DMARCSyntaxError("the p tag must immediately follow the v tag.")
+    if "np" not in tags:
+        if tags["sp"]["explicit"]:
+            tags["np"] = {"value": tags["sp"]["value"], "explicit": False}
+        else:
+            tags["np"] = {"value": tags["p"]["value"], "explicit": False}
+    tags["np"]["value"] = tags["np"]["value"].lower()
+    # The p tag must immediately follow v if it is explicit
+    if tags["p"]["explicit"]:
+        if list(tags.keys())[1] != "p":
+            raise DMARCSyntaxError("the p tag must immediately follow the v tag.")
     tags["v"]["value"] = tags["v"]["value"].upper()
+
+    # Warn about tags removed in DMARCbis
+    removed_tags = ("pct", "rf", "ri")
+    for tag in removed_tags:
+        if tag in tags and tags[tag]["explicit"]:
+            warnings.append(
+                f"Support for the {tag} tag was removed in DMARCbis."
+            )
+
     # Validate tag values
     for tag in tags:
         tag_value = tags[tag]["value"]

checkdmarc/mta_sts.py

@@ -472,6 +472,7 @@ def parse_mta_sts_policy(policy: str) -> MTASTSPolicyParsingResults:
     required_keys = ["version", "mode", "max_age"]
     acceptable_keys = required_keys.copy()
     acceptable_keys.append("mx")
+    seen_keys: set[str] = set()
     if "\n" in policy and "\r\n" not in policy:
         policy = policy.replace("\n", "\r\n")
     lines = policy.split("\r\n")
@@ -486,12 +487,13 @@ def parse_mta_sts_policy(policy: str) -> MTASTSPolicyParsingResults:
         value = key_value[1].strip()
         if key not in acceptable_keys:
             raise MTASTSPolicySyntaxError(f"Line {line}: Unexpected key: {key}")
-        if key in parsed_policy and key != "mx":
-            MTASTSPolicySyntaxError(f"Line {line}: Duplicate key: {key}")
-        elif key == "version" and value not in versions:
-            MTASTSPolicySyntaxError(f"Line {line}: Invalid version: {value}")
+        if key in seen_keys and key != "mx":
+            raise MTASTSPolicySyntaxError(f"Line {line}: Duplicate key: {key}")
+        seen_keys.add(key)
+        if key == "version" and value not in versions:
+            raise MTASTSPolicySyntaxError(f"Line {line}: Invalid version: {value}")
         elif key == "mode" and value not in modes:
-            MTASTSPolicySyntaxError(f"Line {line}: Invalid mode: {value}")
+            raise MTASTSPolicySyntaxError(f"Line {line}: Invalid mode: {value}")
         elif key == "max_age":
             error_msg = "max_age must be an integer value between 0 and 31557600."
             if "." in value:
@@ -501,7 +503,7 @@ def parse_mta_sts_policy(policy: str) -> MTASTSPolicyParsingResults:
                 if value < 0 or value > 31557600:
                     raise MTASTSPolicySyntaxError(error_msg)
             except ValueError:
-                MTASTSPolicySyntaxError(error_msg)
+                raise MTASTSPolicySyntaxError(error_msg)
         if key != "mx":
             parsed_policy[key] = value
         else:

checkdmarc/smtp_tls_reporting.py

@@ -372,7 +372,7 @@ def parse_smtp_tls_reporting_record(
         if include_tag_descriptions:
             tags[tag]["description"] = smtp_rpt_tags[tag]["description"]
     if "rua" not in tags:
-        SMTPTLSReportingSyntaxError("The record is missing the required rua tag.")
+        raise SMTPTLSReportingSyntaxError("The record is missing the required rua tag.")
     tags["rua"]["value"] = tags["rua"]["value"].split(",")
     for uri in tags["rua"]["value"]:
         if len(SMTPTLSREPORTING_URI_REGEX.findall(uri)) != 1: