fix(schemas): FSMAEvent TLC Digital Link hardening

services/shared/schemas.py

@@ -8,6 +8,7 @@
 from datetime import datetime, timezone
 from enum import Enum
 from typing import Any, Dict, List, Literal, Optional
+from urllib.parse import unquote
 from uuid import UUID, uuid4
 
 from pydantic import BaseModel, Field, field_validator
@@ -315,20 +316,53 @@ class FSMAEventType(str, Enum):
     RECEIVING = "RECEIVING"
 
 
-_TLC_PRODUCTION_RE = re.compile(r"^\d{14}[A-Za-z0-9\-\.]+$")
-
-# GS1 General Specifications Table 7-20: the 82-character AI set (no space).
-# AI (10) batch/lot is 1 to 20 of these characters.
-_GS1_AI_82_LOT = r"""[!"%&'()*+,\-./0-9:;<=>?A-Z_a-z]{1,20}"""
+_TLC_PRODUCTION_RE = re.compile(r"^[0-9]{14}[A-Za-z0-9\-\.]+\Z")
+
+# GS1 General Specifications Figure 7.11-1: 82 characters, space excluded.
+# AI (10) is 1 to 20 of those characters after percent-decoding. A well-formed
+# %HH escape is one character. The structural patterns reject a raw character
+# that the carrier requires to be percent-encoded, and a % that is not an escape.
+_GS1_AI_82_CHARS = frozenset("!\"%&'()*+,-./0123456789:;<=>?ABCDEFGHIJKLMNOPQRSTUVWXYZ_abcdefghijklmnopqrstuvwxyz")
+_PCT_ESCAPE = r"%[0-9A-Fa-f]{2}"
+# Digital Link XCHAR literals. Reserved symbols exist only as %HH escapes.
+_DL_LOT_RAW_CLASS = r'[0-9A-Za-z"._-]'
+# TDS GS3A3 URI form. Appendix A escapes that must not appear raw: " % & / < > ?
+_LGTIN_LOT_RAW_CLASS = r"[!'()*+,\-.0-9:;=A-Z_a-z]"
+# Zero or more custom path segments between the host and /01 (segment = *pchar).
+_DL_CUSTOM_STEM = r"(?P<stem>(?:/[^/\s?#]*)*?)"
 # Two non-empty digit runs whose lengths sum to 13 (company prefix + indicator/item ref).
-_LGTIN_DIGIT_PAIRS = "(?:" + "|".join(rf"\d{{{n}}}\.\d{{{13 - n}}}" for n in range(1, 13)) + ")"
-_TLC_LGTIN_RE = re.compile(rf"^urn:epc:class:lgtin:{_LGTIN_DIGIT_PAIRS}\.{_GS1_AI_82_LOT}$")
+_LGTIN_DIGIT_PAIRS = "(?:" + "|".join(rf"[0-9]{{{n}}}\.[0-9]{{{13 - n}}}" for n in range(1, 13)) + ")"
+_TLC_LGTIN_RE = re.compile(
+    rf"^urn:epc:class:lgtin:{_LGTIN_DIGIT_PAIRS}\."
+    rf"(?P<lot>(?:{_LGTIN_LOT_RAW_CLASS}|{_PCT_ESCAPE})+)\Z"
+)
 _TLC_DIGITAL_LINK_RE = re.compile(
-    rf"^https?://[^/\s?#]+/01/\d{{14}}/10/{_GS1_AI_82_LOT}"
-    rf"(?:/[^\s?#]*)?(?:\?[^\s#]*)?$"
+    rf"^https?://[^/\s?#]+{_DL_CUSTOM_STEM}/01/[0-9]{{14}}/10/"
+    rf"(?P<lot>(?:{_DL_LOT_RAW_CLASS}|{_PCT_ESCAPE})+)"
+    rf"(?:/[^\s?#]*)?(?:\?[^\s#]*)?\Z"
 )
 
 
+def _decoded_ai10_lot_within_limit(encoded_lot: str) -> bool:
+    """Return whether an AI (10) lot is 1-20 GS1 characters after unquote.
+
+    The caller has already matched a carrier pattern. This does not rewrite
+    the TLC; it only accepts or rejects the lot substring.
+    """
+    index = 0
+    size = len(encoded_lot)
+    while index < size:
+        if encoded_lot[index] != "%":
+            index += 1
+            continue
+        hex_pair = encoded_lot[index + 1 : index + 3]
+        if len(hex_pair) != 2 or any(char not in "0123456789ABCDEFabcdef" for char in hex_pair):
+            return False
+        index += 3
+    decoded = unquote(encoded_lot)
+    return 1 <= len(decoded) <= 20 and all(char in _GS1_AI_82_CHARS for char in decoded)
+
+
 class FSMAEvent(BaseModel):
     """FSMA event model for Critical Tracking Events.
 
@@ -391,8 +425,12 @@ class FSMAEvent(BaseModel):
     @classmethod
     def validate_tlc_format(cls, v: str) -> str:
         """Accept GTIN-14 + lot, an LGTIN class URN, or a GS1 Digital Link."""
-        if _TLC_PRODUCTION_RE.match(v) or _TLC_LGTIN_RE.match(v) or _TLC_DIGITAL_LINK_RE.match(v):
+        if _TLC_PRODUCTION_RE.fullmatch(v):
             return v
+        for pattern in (_TLC_LGTIN_RE, _TLC_DIGITAL_LINK_RE):
+            matched = pattern.fullmatch(v)
+            if matched is not None and _decoded_ai10_lot_within_limit(matched.group("lot")):
+                return v
         raise ValueError(
             f"TLC '{v}' does not match FSMA 204 production pattern "
             "(accepted formats: a fourteen-digit GTIN prefix followed by a lot "

tests/test_shared/test_fsmaevent_lot_level_tlc.py

@@ -1,23 +1,36 @@
 """Data-free checks for FSMAEvent lot-level TLC acceptance.
 
-No identifier values, GTINs, lots, or records. These tests only reject
-blank input and bare scheme prefixes, and check that the new patterns
-are anchored.
+No identifier values, GTINs, lots, or records. These tests reject blank
+input, bare scheme prefixes, a bare percent, and a trailing newline on
+inputs that are already rejected. They also check pattern shape: ASCII
+digits, an optional Digital Link stem before /01, and that the AI (10)
+length limit is applied to the percent-decoded lot.
 """
 
 from __future__ import annotations
 
 import re
 
 import pytest
-from shared.schemas import FSMAEvent, _TLC_DIGITAL_LINK_RE, _TLC_LGTIN_RE
+from shared.schemas import (
+    _DL_CUSTOM_STEM,
+    _DL_LOT_RAW_CLASS,
+    _GS1_AI_82_CHARS,
+    _LGTIN_LOT_RAW_CLASS,
+    _PCT_ESCAPE,
+    _TLC_DIGITAL_LINK_RE,
+    _TLC_LGTIN_RE,
+    FSMAEvent,
+    _decoded_ai10_lot_within_limit,
+)
 
 _BLANK = ("", " ", "\t", "\n", "\r\n", " \t ", " " * 15, "\t" * 15)
 _BARE_PREFIXES = (
     "urn:epc:class:lgtin:",
     "urn:epc:class:lgtin:..",
     "https:///01//10/",
 )
+_ALREADY_REJECTED = (*_BLANK, *_BARE_PREFIXES)
 
 
 @pytest.mark.parametrize("value", _BLANK)
@@ -32,9 +45,63 @@ def test_bare_lot_level_prefixes_rejected(value: str) -> None:
         FSMAEvent.validate_tlc_format(value)
 
 
+@pytest.mark.parametrize("value", _ALREADY_REJECTED)
+def test_trailing_newline_rejected_on_inputs_already_rejected(value: str) -> None:
+    with pytest.raises(ValueError, match="does not match FSMA 204 production pattern"):
+        FSMAEvent.validate_tlc_format(value + "\n")
+
+
+@pytest.mark.parametrize("prefix", _BARE_PREFIXES)
+def test_bare_percent_rejected_on_prefixes(prefix: str) -> None:
+    with pytest.raises(ValueError, match="does not match FSMA 204 production pattern"):
+        FSMAEvent.validate_tlc_format(prefix + "%")
+
+
 def test_lot_level_patterns_are_anchored() -> None:
     for pattern in (_TLC_LGTIN_RE, _TLC_DIGITAL_LINK_RE):
         assert isinstance(pattern, re.Pattern)
         assert pattern.pattern.startswith("^")
-        assert pattern.pattern.endswith("$")
+        assert pattern.pattern.endswith(r"\Z")
         assert pattern.flags & re.MULTILINE == 0
+
+
+def test_identifier_digits_are_ascii() -> None:
+    for pattern in (_TLC_LGTIN_RE, _TLC_DIGITAL_LINK_RE):
+        assert r"\d" not in pattern.pattern
+        assert "[0-9]" in pattern.pattern
+
+
+def test_digital_link_pattern_has_optional_stem_before_01() -> None:
+    pattern = _TLC_DIGITAL_LINK_RE.pattern
+    assert _DL_CUSTOM_STEM in pattern
+    assert pattern.index(_DL_CUSTOM_STEM) < pattern.index("/01/")
+    assert "/01/[0-9]{14}/10/" in pattern
+
+
+def test_lot_patterns_count_escapes_outside_the_length_quantifier() -> None:
+    for pattern in (_TLC_LGTIN_RE, _TLC_DIGITAL_LINK_RE):
+        assert "(?P<lot>" in pattern.pattern
+        assert _PCT_ESCAPE in pattern.pattern
+        assert "{1,20}" not in pattern.pattern
+    assert _DL_LOT_RAW_CLASS in _TLC_DIGITAL_LINK_RE.pattern
+    assert _LGTIN_LOT_RAW_CLASS in _TLC_LGTIN_RE.pattern
+    for char in "!%&'()*+,/:;<=>?#":
+        assert char not in _DL_LOT_RAW_CLASS
+    for char in '"%&/<>?':
+        assert char not in _LGTIN_LOT_RAW_CLASS
+
+
+def test_ai82_charset_is_the_gs1_graphic_set() -> None:
+    assert len(_GS1_AI_82_CHARS) == 82
+    assert " " not in _GS1_AI_82_CHARS
+    assert all(len(char) == 1 and char.isascii() and not char.isspace() for char in _GS1_AI_82_CHARS)
+
+
+def test_ai10_length_limit_is_applied_after_percent_decode() -> None:
+    assert _decoded_ai10_lot_within_limit("x" * 1)
+    assert _decoded_ai10_lot_within_limit("x" * 20)
+    assert not _decoded_ai10_lot_within_limit("x" * 21)
+    assert _decoded_ai10_lot_within_limit("%41" * 20)
+    assert not _decoded_ai10_lot_within_limit("%41" * 21)
+    assert not _decoded_ai10_lot_within_limit("")
+    assert not _decoded_ai10_lot_within_limit("%")