Harden glyph planner manifest and marker preflight

glyph_hybrid_plan.py

@@ -6,10 +6,10 @@ request either receives one complete, reversible hybrid plan or remains on the
 unchanged legacy path.
 
 The profile scans *maximal printable-ASCII runs*.  A run is network-looking
-when it contains ``@`` or begins exactly with ``http://`` or ``https://``.
-Every such run must independently satisfy the strict ``glyph_ascii_v1``
-grammar.  There is no stripping, case folding, Unicode repair, substring
-borrowing, or partial activation.
+when a case-insensitive preflight finds a network marker anywhere within it.
+Every such complete run must independently satisfy the strict, native-
+lowercase ``glyph_ascii_v1`` grammar.  There is no stripping, case folding,
+Unicode repair, substring borrowing, or partial activation.
 
 Canonical boundary contract
 ---------------------------
@@ -28,6 +28,7 @@ from dataclasses import asdict, dataclass
 import hashlib
 import json
 import re
+from types import MappingProxyType
 import unicodedata
 
 from glyph_ascii_v1 import (
@@ -35,6 +36,7 @@ from glyph_ascii_v1 import (
     GlyphAsciiV1Error,
     GlyphAsciiV1Proof,
     glyph_ascii_v1_asset_manifest_sha256,
+    glyph_ascii_v1_identifier_kind,
     inverse_glyph_ascii_v1,
     render_glyph_ascii_v1,
 )
@@ -66,7 +68,7 @@ GLYPH_HYBRID_MAX_MODEL_GENERATED_TEXT_UNITS = 800
 
 _PRINTABLE_ASCII_RUN_RE = re.compile(r"[ -~]+", flags=re.ASCII)
 _SHA256_RE = re.compile(r"[0-9a-f]{64}\Z", flags=re.ASCII)
-_NETWORK_PREFIXES = ("http://", "https://")
+_NETWORK_MARKERS = ("http://", "https://", "ftp://", "www.", "@", "://")
 _ZH_CARRIER_COMPATIBILITY_PUNCTUATION = frozenset(
     "，。！？；：、（）【】「」『』《》〈〉“”‘’［］｛｝"
     "…⋯—–．｡､〔〕〖〗〘〙〚〛﹁﹂︰﹐﹑﹒﹔﹕﹖﹗"
@@ -193,7 +195,8 @@ def glyph_hybrid_plan_sha256(plan: HybridRenderPlan) -> str:
 
 
 def _network_looking(run: str) -> bool:
-    return "@" in run or run.startswith(_NETWORK_PREFIXES)
+    folded = run.lower()
+    return any(marker in folded for marker in _NETWORK_MARKERS)
 
 
 def _request_has_unsafe_unicode(raw_request: str) -> bool:
@@ -235,8 +238,17 @@ def _request_has_unsafe_unicode(raw_request: str) -> bool:
 def _raw_pieces(raw_request: str) -> tuple[_RawPiece, ...] | None:
     network_runs: list[re.Match[str]] = []
     for match in _PRINTABLE_ASCII_RUN_RE.finditer(raw_request):
-        if _network_looking(match.group(0)):
-            network_runs.append(match)
+        run = match.group(0)
+        if not _network_looking(run):
+            continue
+        try:
+            # A marker anywhere in a maximal printable-ASCII run makes that
+            # complete run security-relevant.  Only the exact, lowercase v1
+            # grammar may activate; never borrow a valid-looking substring.
+            glyph_ascii_v1_identifier_kind(run)
+        except GlyphAsciiV1Error:
+            return None
+        network_runs.append(match)
     if not network_runs:
         return None
     if len(network_runs) > GLYPH_HYBRID_MAX_IDENTIFIERS:
@@ -292,15 +304,26 @@ def _carrier_canonical(raw_fragment: str) -> str | None:
     return canonical
 
 
-def _manifest_digest(
+def _manifest_snapshot(
     asset_entry_sha256_by_asset_id: Mapping[str, str],
-) -> str:
+) -> tuple[Mapping[str, str], str]:
+    if not isinstance(asset_entry_sha256_by_asset_id, Mapping):
+        raise GlyphHybridPlanError("glyph asset manifest entries must be a mapping")
     try:
-        return glyph_ascii_v1_asset_manifest_sha256(
-            asset_entry_sha256_by_asset_id
-        )
+        # A caller-supplied Mapping may compute or cycle values on each
+        # access. Materialize it exactly once into an owned plain dict, then
+        # expose only an immutable view for every subsequent proof operation.
+        owned = dict(asset_entry_sha256_by_asset_id.items())
+    except Exception as error:
+        raise GlyphHybridPlanError(
+            "glyph asset manifest cannot be materialized"
+        ) from error
+    snapshot: Mapping[str, str] = MappingProxyType(owned)
+    try:
+        digest = glyph_ascii_v1_asset_manifest_sha256(snapshot)
     except GlyphAsciiV1Error as error:
         raise GlyphHybridPlanError("glyph asset manifest is invalid") from error
+    return snapshot, digest
 
 
 def _segment(
@@ -351,7 +374,24 @@ def build_glyph_hybrid_plan(
 
     if type(raw_request) is not str:
         raise GlyphHybridPlanError("raw request must be an exact string")
-    manifest_sha256 = _manifest_digest(asset_entry_sha256_by_asset_id)
+    manifest_snapshot, manifest_sha256 = _manifest_snapshot(
+        asset_entry_sha256_by_asset_id
+    )
+    return _build_glyph_hybrid_plan_from_snapshot(
+        raw_request,
+        manifest_snapshot=manifest_snapshot,
+        manifest_sha256=manifest_sha256,
+    )
+
+
+def _build_glyph_hybrid_plan_from_snapshot(
+    raw_request: str,
+    *,
+    manifest_snapshot: Mapping[str, str],
+    manifest_sha256: str,
+) -> HybridRenderPlan | None:
+    """Build using one already validated, immutable manifest snapshot."""
+
     if (
         not raw_request
         or len(raw_request) > GLYPH_HYBRID_MAX_RAW_CHARS
@@ -430,12 +470,16 @@ def build_glyph_hybrid_plan(
                 append_separator(piece.start)
             canonical, proof = render_glyph_ascii_v1(
                 piece.text,
-                asset_entry_sha256_by_asset_id=asset_entry_sha256_by_asset_id,
+                asset_entry_sha256_by_asset_id=manifest_snapshot,
                 absolute_raw_start=piece.start,
                 canonical_start=canonical_cursor,
             )
         except GlyphAsciiV1Error:
             return None
+        if proof.asset_manifest_sha256 != manifest_sha256:
+            raise GlyphHybridPlanError(
+                "identifier proof manifest does not match plan header"
+            )
 
         identifier_ordinal = len(identifier_proofs)
         identifier_proofs.append(proof)
@@ -548,7 +592,9 @@ def inverse_glyph_hybrid_plan(
         raise GlyphHybridPlanError("hybrid plan has an invalid type")
     if type(expected_raw_request) is not str:
         raise GlyphHybridPlanError("expected raw request must be an exact string")
-    manifest_sha256 = _manifest_digest(asset_entry_sha256_by_asset_id)
+    manifest_snapshot, manifest_sha256 = _manifest_snapshot(
+        asset_entry_sha256_by_asset_id
+    )
     if (
         type(plan.identifier_proofs) is not tuple
         or type(plan.segments) is not tuple
@@ -738,10 +784,20 @@ def inverse_glyph_hybrid_plan(
     for identifier_ordinal, proof in enumerate(plan.identifier_proofs):
         if type(proof) is not GlyphAsciiV1Proof:
             raise GlyphHybridPlanError("identifier proof has an invalid type")
+        if (
+            _require_sha256(
+                proof.asset_manifest_sha256,
+                field="proof.asset_manifest_sha256",
+            )
+            != plan.asset_manifest_logical_sha256
+        ):
+            raise GlyphHybridPlanError(
+                "identifier proof manifest does not match plan header"
+            )
         try:
             raw_identifier = inverse_glyph_ascii_v1(
                 proof,
-                asset_entry_sha256_by_asset_id=asset_entry_sha256_by_asset_id,
+                asset_entry_sha256_by_asset_id=manifest_snapshot,
             )
         except GlyphAsciiV1Error as error:
             raise GlyphHybridPlanError("identifier proof is invalid") from error
@@ -780,9 +836,10 @@ def inverse_glyph_hybrid_plan(
     if set(asset_segments) != expected_asset_keys:
         raise GlyphHybridPlanError("asset segment set does not match identifier proofs")
 
-    expected_plan = build_glyph_hybrid_plan(
+    expected_plan = _build_glyph_hybrid_plan_from_snapshot(
         reconstructed,
-        asset_entry_sha256_by_asset_id=asset_entry_sha256_by_asset_id,
+        manifest_snapshot=manifest_snapshot,
+        manifest_sha256=manifest_sha256,
     )
     if (
         expected_plan is None

tests/test_glyph_hybrid_plan.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from dataclasses import asdict, replace
 import hashlib
 import json
@@ -206,6 +207,21 @@ def test_carrier_uses_existing_zh_tw_normalization_without_network_content():
     )
 
 
+def test_non_network_ascii_code_and_semver_remain_ordinary_carrier_content():
+    raw = "版本v1.2.3與程式C++保持原讀法，聯絡mead@forest.org。"
+    plan = _require_plan(raw)
+    carrier_raw = "".join(
+        segment.raw_text
+        for segment in plan.segments
+        if segment.kind in {"carrier", "control"}
+    )
+
+    assert "v1.2.3" in carrier_raw
+    assert "C++" in carrier_raw
+    assert len(plan.identifier_proofs) == 1
+    assert _inverse(plan, raw) == raw
+
+
 def _alpha(rng: random.Random, minimum: int = 1, maximum: int = 7) -> str:
     return "".join(
         rng.choice("abcdefghijklmnopqrstuvwxyz")
@@ -316,6 +332,27 @@ def test_one_invalid_network_run_rejects_the_entire_profile_without_partial_plan
     assert _build(raw) is None
 
 
+@pytest.mark.parametrize(
+    "malformed_run",
+    (
+        "foohttps://",
+        "9http://",
+        "fooHTTPS://",
+        "fooftp://",
+        "foowww.",
+        "foo://",
+        "@",
+        "https://",
+    ),
+)
+def test_marker_anywhere_in_a_maximal_ascii_run_blocks_partial_activation(
+    malformed_run,
+):
+    raw = f"聯絡mead@forest.org，異常值{malformed_run}，結束。"
+
+    assert _build(raw) is None
+
+
 @pytest.mark.parametrize(
     "raw",
     (
@@ -560,6 +597,82 @@ def test_manifest_substitution_and_invalid_manifest_configuration_are_rejected()
         )
 
 
+class _CyclingManifest(Mapping[str, str]):
+    """Return the frozen value once, then a changed value on later reads."""
+
+    def __init__(self) -> None:
+        self.accesses = {asset_id: 0 for asset_id in ASSET_MANIFEST}
+
+    def __iter__(self):
+        return iter(ASSET_MANIFEST)
+
+    def __len__(self) -> int:
+        return len(ASSET_MANIFEST)
+
+    def __getitem__(self, asset_id: str) -> str:
+        count = self.accesses[asset_id]
+        self.accesses[asset_id] = count + 1
+        if count == 0:
+            return ASSET_MANIFEST[asset_id]
+        return hashlib.sha256(f"changed:{asset_id}".encode("ascii")).hexdigest()
+
+
+def test_build_materializes_a_changing_manifest_exactly_once():
+    raw = (
+        "先寄mead@forest.org，再看https://audiocoast.org/tour，"
+        "確認完成。"
+    )
+    cycling = _CyclingManifest()
+
+    plan = build_glyph_hybrid_plan(
+        raw,
+        asset_entry_sha256_by_asset_id=cycling,
+    )
+
+    assert plan == _require_plan(raw)
+    assert set(cycling.accesses.values()) == {1}
+    assert all(
+        proof.asset_manifest_sha256 == plan.asset_manifest_logical_sha256
+        for proof in plan.identifier_proofs
+    )
+
+
+def test_inverse_materializes_a_changing_manifest_exactly_once():
+    raw = (
+        "先寄mead@forest.org，再看https://audiocoast.org/tour，"
+        "確認完成。"
+    )
+    plan = _require_plan(raw)
+    cycling = _CyclingManifest()
+
+    reconstructed = inverse_glyph_hybrid_plan(
+        plan,
+        asset_entry_sha256_by_asset_id=cycling,
+        expected_raw_request=raw,
+    )
+
+    assert reconstructed == raw
+    assert set(cycling.accesses.values()) == {1}
+
+
+def test_identifier_proof_manifest_digest_must_equal_plan_header():
+    raw = "請寄mead@forest.org。"
+    plan = _require_plan(raw)
+    proof = replace(plan.identifier_proofs[0], asset_manifest_sha256="0" * 64)
+    forged = replace(
+        plan,
+        identifier_proofs=(proof,),
+        plan_sha256="",
+    )
+    forged = replace(forged, plan_sha256=glyph_hybrid_plan_sha256(forged))
+
+    with pytest.raises(
+        GlyphHybridPlanError,
+        match="identifier proof manifest does not match plan header",
+    ):
+        _inverse(forged, raw)
+
+
 def test_non_string_inputs_and_plan_subclasses_are_rejected():
     class ForgedPlan(HybridRenderPlan):
         pass