Classify CZDS malformed record structure without disclosing zone data

README.md

@@ -312,12 +312,30 @@ 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;
+`reason=record_fields_space_delimited`. 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.
 
+Malformed records that do not have five tab-separated fields are **still
+rejected**. A shared classifier reports one of these fixed structural hints:
+
+- `record_blank`: empty or whitespace-only record.
+- `record_fields_comment`: comment-like line beginning with a semicolon.
+- `record_fields_directive`: directive-like line beginning with a dollar sign.
+- `record_fields_space_delimited`: no tabs but contains horizontal whitespace.
+- `record_fields_missing_tabs`: no tabs and no horizontal whitespace.
+- `record_fields_partial_tabs`: some tabs, but fewer than four.
+
+The classification is intentionally heuristic. In particular,
+`record_fields_space_delimited` **does not establish** that the entire zone
+uses valid DNS master-file syntax or that accepting whitespace separators
+would preserve correct nameserver counts. Both the first-record preflight
+check and the full stream parser use the same classification. No records
+are skipped, normalized differently, or published as a result of these
+diagnostic changes.
+
 **Troubleshooting licensed inputs:** Run the normal daily-release workflow
 only on the protected runner. If preflight fails, use its reason code to
 construct a synthetic zone reproducer before changing validation behavior.

scripts/build_nsdb.py

@@ -10,9 +10,26 @@
 
 
 # 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 code only; never return private source text or offsets.
+
+    Classification does not change validation: malformed records still fail.
+    """
+    if not line.strip():
+        return "record_blank"
+    if line.lstrip().startswith(";"):
+        return "record_fields_comment"
+    if line.lstrip().startswith("$"):
+        return "record_fields_directive"
+    if "\t" not in line:
+        if any(char in line for char in " \v\f"):
+            return "record_fields_space_delimited"
+        return "record_fields_missing_tabs"
+    return "record_fields_partial_tabs"
+
 class ZoneValidationError(ValueError):
     """Detailed errors remain private; a fixed code identifies the invariant."""
 
@@ -318,7 +335,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,15 @@ 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 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 +172,33 @@ 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 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)