Classify malformed CZDS records safely in preflight and aggregation

README.md

@@ -312,11 +312,24 @@ python scripts/preflight_czds.py \
 
 On validation failure, public logs now include a fixed `reason=<code>`, for example
 `reason=soa_serial_mismatch`, `reason=ns_owner_order`, or
-`reason=record_fields`. The code is selected by the production parser;
-neither the offending record, owner, NS target, SOA serial, record offset, nor
-the exception's original message is printed. Unexpected errors become
-`reason=unclassified_error`. Both preflight and the production CLI use
-the same error taxonomy, and all validation gates remain fail-closed.
+`reason=record_fields_partial_tabs`. A record that does not split into five tab-separated fields
+is now classified by its *structure*:
+
+- `record_fields_comment`: leading semicolon or hash (comment-like syntax; validity not inferred)
+- `record_fields_directive`: leading dollar sign (possible master-file directive)
+- `record_fields_space_delimited`: whitespace present but no tab (candidate only)
+- `record_fields_missing_tabs`: no tabs or whitespace
+- `record_fields_partial_tabs`: one to three tabs, without other whitespace
+- `record_fields_mixed_delimiters`: tabs mixed with spaces or other horizontal whitespace
+- `record_blank`: empty record
+
+These diagnostics apply to both the first record and later records. They
+describe the record's shape, **not** whether a different zone-file grammar
+should be accepted. Neither the offending record, owner, NS target, SOA serial,
+record offset, nor the exception's original message is printed. Unexpected
+errors become `reason=unclassified_error`. Both preflight and full
+aggregation use the same classification without changing parsing acceptance,
+delegation deduplication, DNS ordering, or SOA consistency.
 
 **Troubleshooting licensed inputs:** Run the normal daily-release workflow
 only on the protected runner. If preflight fails, use its reason code to

scripts/build_nsdb.py

@@ -10,7 +10,31 @@
 
 
 # These fixed codes may appear in public CI logs; exception messages may not.
-_PUBLIC_VALIDATION_CODES = frozenset(["dns_name_non_ascii","dns_escape_dangling","dns_escape_decimal_range","dns_label_empty","dns_label_too_long","dns_name_too_long","dns_label_non_ascii","ns_target_empty","ns_rdata_extra","ns_target_escape_dangling","ns_target_root","record_blank","record_fields","header_record_fields","apex_hint_missing","apex_hint_unavailable","after_final_soa","soa_owner_class","soa_serial_invalid","soa_serial_mismatch","soa_excess","ns_owner_outside_apex","ns_owner_order","zone_empty","soa_missing","soa_final_missing","archive_empty","archive_member_unreadable"])
+_PUBLIC_VALIDATION_CODES = frozenset(["dns_name_non_ascii","dns_escape_dangling","dns_escape_decimal_range","dns_label_empty","dns_label_too_long","dns_name_too_long","dns_label_non_ascii","ns_target_empty","ns_rdata_extra","ns_target_escape_dangling","ns_target_root","record_blank","record_fields","header_record_fields","apex_hint_missing","apex_hint_unavailable","after_final_soa","soa_owner_class","soa_serial_invalid","soa_serial_mismatch","soa_excess","ns_owner_outside_apex","ns_owner_order","zone_empty","soa_missing","soa_final_missing","archive_empty","archive_member_unreadable","record_fields_comment","record_fields_directive","record_fields_space_delimited","record_fields_missing_tabs","record_fields_partial_tabs","record_fields_mixed_delimiters"])
+
+
+def classify_record_fields(line: str) -> str:
+    """Fixed structural diagnostic only; never expose private record data.
+
+    This is deliberately not a record parser. Every malformed record is
+    still rejected; these heuristics identify the probable input shape.
+    """
+    if not line.strip():
+        return "record_blank"
+    stripped = line.lstrip()
+    if stripped.startswith((";", "#")):
+        return "record_fields_comment"
+    if stripped.startswith("$"):
+        return "record_fields_directive"
+    tabs = line.count("\t")
+    if tabs == 0:
+        if any(char in line for char in " \v\f"):
+            return "record_fields_space_delimited"
+        return "record_fields_missing_tabs"
+    if any(char in line for char in " \v\f"):
+        return "record_fields_mixed_delimiters"
+    return "record_fields_partial_tabs"
+
 
 
 class ZoneValidationError(ValueError):
@@ -318,7 +342,7 @@ def flush_owner():
             raise ZoneValidationError("record_blank", f"{label}: blank line violates CZDS zone format")
         fields = line.split("\t", 4)
         if len(fields) != 5:
-            raise ZoneValidationError("record_fields", f"{label}: expected five tab-separated fields at record {records + 1}")
+            raise ZoneValidationError(classify_record_fields(line), f"{label}: expected five tab-separated fields at record {records + 1}")
         owner_raw, ttl, rrclass, rrtype, rdata = fields
         owner = normalize_name(owner_raw)
         rrclass = rrclass.lower()

scripts/preflight_czds.py

@@ -11,6 +11,7 @@
 
 from build_nsdb import (
     ZoneValidationError,
+    classify_record_fields,
     public_validation_code,
     infer_apex_from_filename,
     load_special_zones,
@@ -35,7 +36,7 @@ def has_nonleading_soa(path: Path) -> bool:
         first = source.readline().rstrip("\r\n")
     fields = first.split("\t", 4)
     if len(fields) != 5:
-        raise ZoneValidationError("header_record_fields", "first zone record does not have five tab-separated fields")
+        raise ZoneValidationError(classify_record_fields(first), "first zone record does not have five tab-separated fields")
     return fields[3].lower() != "soa" or fields[2].lower() != "in"
 
 

tests/test_preflight.py

@@ -114,7 +114,18 @@ def test_failure_reasons_are_safe_and_match_full_aggregation(self):
         soa = "gov.\t3600\tIN\tSOA\tns.gov. hostmaster.gov. 123 2 3 4 5"
         cases = {
             "record_blank": [initial, "", soa],
-            "record_fields": [initial, "private-name.gov. malformed", soa],
+            "record_fields_comment": [initial, "  ; private-name.gov. hidden", soa],
+            "record_fields_directive": [initial, "  $ORIGIN private-name.gov.", soa],
+            "record_fields_space_delimited": [
+                initial, "private-name.gov. 3600 IN NS ns.hidden.example.", soa
+            ],
+            "record_fields_missing_tabs": [initial, "private-name.gov.", soa],
+            "record_fields_partial_tabs": [
+                initial, "private-name.gov.\t3600\tIN\tNS", soa
+            ],
+            "record_fields_mixed_delimiters": [
+                initial, "private-name.gov.\t3600 IN NS ns.hidden.example.", soa
+            ],
             "soa_missing": [initial, "private-name.gov.\t3600\tIN\tNS\tns.hidden.example."],
             "soa_final_missing": [
                 initial, soa, "private-name.gov.\t3600\tIN\tNS\tns.hidden.example."
@@ -164,6 +175,34 @@ def test_failure_reasons_are_safe_and_match_full_aggregation(self):
                     self.assertNotIn("hostmaster.gov", output)
                     self.assertNotIn("privaté.gov", output)
 
+    def test_first_record_format_reason_matches_full_aggregation(self):
+        # Both CLI entry points must use the same structural classifier,
+        # including the preflight header check before process_input().
+        cases = {
+            "record_blank": "",
+            "record_fields_comment": "; private-name.gov. confidential",
+            "record_fields_directive": "$TTL 3600",
+            "record_fields_space_delimited": "private-name.gov. 3600 IN NS ns.hidden.example.",
+            "record_fields_missing_tabs": "private-name.gov.",
+            "record_fields_partial_tabs": "private-name.gov.\t3600\tIN\tNS",
+            "record_fields_mixed_delimiters": "private-name.gov.\t3600 IN NS ns.hidden.example.",
+        }
+        final = "gov.\t3600\tIN\tSOA\tns.gov. hostmaster.gov. 123 2 3 4 5"
+        for reason, first in cases.items():
+            with self.subTest(reason=reason), tempfile.TemporaryDirectory() as d:
+                root = pathlib.Path(d)
+                self.setup_fixture(root, "gov.txt.gz", [first, final])
+                preflight = self.invoke(root)
+                build = self.invoke_build(root)
+                self.assertNotEqual(preflight.returncode, 0)
+                self.assertNotEqual(build.returncode, 0)
+                for result in (preflight, build):
+                    output = result.stdout + result.stderr
+                    self.assertIn(f"reason={reason}", output)
+                    self.assertNotIn("private-name.gov.", output)
+                    self.assertNotIn("ns.hidden.example", output)
+                    self.assertNotIn("hostmaster.gov", output)
+
     def test_nonleading_soa_and_glue_deduplicate_in_both_paths(self):
         with tempfile.TemporaryDirectory() as d:
             root = pathlib.Path(d)