From e03e3e7ab0fafb3f1a5d5a6b8a12b6344dba53b5 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Sat, 24 Oct 2026 21:13:37 -0700 Subject: [PATCH] fix(schemas): FSMAEvent TLC Digital Link hardening Replaces #3527 with the same net change against main (follow-up to #3499), as one conventional commit, including the .secrets.baseline line refresh. --- .secrets.baseline | 4 +- services/shared/schemas.py | 58 +++++++++++--- .../test_fsmaevent_lot_level_tlc.py | 77 +++++++++++++++++-- 3 files changed, 122 insertions(+), 17 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index 49f352545..feaf6d8f7 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -3117,7 +3117,7 @@ "filename": "services/shared/schemas.py", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", "is_verified": false, - "line_number": 164 + "line_number": 165 } ], "services/shared/secret_redaction.py": [ @@ -3754,5 +3754,5 @@ } ] }, - "generated_at": "2026-10-25T02:01:01Z" + "generated_at": "2026-10-25T03:46:47Z" } diff --git a/services/shared/schemas.py b/services/shared/schemas.py index e575b6ac6..517ec0ef0 100644 --- a/services/shared/schemas.py +++ b/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(?:/[^/\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(?:{_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(?:{_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 " diff --git a/tests/test_shared/test_fsmaevent_lot_level_tlc.py b/tests/test_shared/test_fsmaevent_lot_level_tlc.py index d6adc2eee..a31397c0c 100644 --- a/tests/test_shared/test_fsmaevent_lot_level_tlc.py +++ b/tests/test_shared/test_fsmaevent_lot_level_tlc.py @@ -1,8 +1,10 @@ """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 @@ -10,7 +12,17 @@ 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 = ( @@ -18,6 +30,7 @@ "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" 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("%")