fix(api): align claim bounds across runtime snapshots
This commit is contained in:
@@ -15,6 +15,7 @@ from .validation import (
|
||||
require_goods_id,
|
||||
require_lower_hex_64,
|
||||
require_money,
|
||||
require_persisted_text,
|
||||
require_positive_int,
|
||||
require_rfc3339_z,
|
||||
rfc3339_z_nanoseconds,
|
||||
@@ -85,18 +86,12 @@ class PurchaseTask:
|
||||
def __post_init__(self) -> None:
|
||||
require_uuid4(self.id, "invalid_task_id")
|
||||
require_positive_int(self.version, "invalid_task_version")
|
||||
require_string(self.title, "invalid_task_title", maximum=MAX_TITLE_CODE_POINTS)
|
||||
if not self.title.strip() or self.title.strip() != self.title:
|
||||
raise ValidationError("invalid_task_title")
|
||||
require_persisted_text(self.title, "invalid_task_title", maximum=MAX_TITLE_CODE_POINTS)
|
||||
require_goods_id(self.goods_id)
|
||||
if self.product_url != canonical_product_url(self.goods_id):
|
||||
raise ValidationError("invalid_product_url")
|
||||
require_string(self.sku_color, "invalid_sku_color", maximum=MAX_SKU_TEXT_CODE_POINTS)
|
||||
require_string(self.sku_size, "invalid_sku_size", maximum=MAX_SKU_TEXT_CODE_POINTS)
|
||||
if self.sku_color.strip() != self.sku_color:
|
||||
raise ValidationError("invalid_sku_color")
|
||||
if self.sku_size.strip() != self.sku_size:
|
||||
raise ValidationError("invalid_sku_size")
|
||||
require_persisted_text(self.sku_color, "invalid_sku_color", maximum=MAX_SKU_TEXT_CODE_POINTS)
|
||||
require_persisted_text(self.sku_size, "invalid_sku_size", maximum=MAX_SKU_TEXT_CODE_POINTS)
|
||||
require_positive_int(self.quantity, "invalid_quantity")
|
||||
require_money(self.max_total_price, "invalid_max_total_price")
|
||||
|
||||
|
||||
@@ -25,6 +25,10 @@ MAX_TITLE_CODE_POINTS = 120
|
||||
MAX_SKU_TEXT_CODE_POINTS = 80
|
||||
MAX_GOODS_ID_ASCII_CHARACTERS = 32
|
||||
MAX_MONEY_ASCII_CHARACTERS = 32
|
||||
# Go strings.TrimSpace uses Unicode White_Space plus the six ASCII space
|
||||
# characters below, but unlike Python str.strip it does not include U+001C--
|
||||
# U+001F. Keep the wire contract independent of either runtime's defaults.
|
||||
GO_UNICODE_WHITE_SPACE = "\t\n\v\f\r \u0085\u00a0\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a\u2028\u2029\u202f\u205f\u3000"
|
||||
|
||||
|
||||
def require_string(value: object, reason: str, *, maximum: int = 4096) -> str:
|
||||
@@ -35,6 +39,20 @@ def require_string(value: object, reason: str, *, maximum: int = 4096) -> str:
|
||||
return value
|
||||
|
||||
|
||||
def require_persisted_text(value: object, reason: str, *, maximum: int) -> str:
|
||||
"""Validate text stored by Go after TrimSpace, without Python trim drift."""
|
||||
|
||||
text = require_string(value, reason, maximum=maximum)
|
||||
if text.strip(GO_UNICODE_WHITE_SPACE) != text:
|
||||
raise ValidationError(reason)
|
||||
# Python str.strip treats these C0 separators as whitespace while Go does
|
||||
# not. Reject them anywhere on both ends instead of assigning them two
|
||||
# runtime-dependent meanings.
|
||||
if any(0x1C <= ord(character) <= 0x1F for character in text):
|
||||
raise ValidationError(reason)
|
||||
return text
|
||||
|
||||
|
||||
def require_uuid4(value: object, reason: str = "invalid_uuid") -> str:
|
||||
text = require_string(value, reason, maximum=36)
|
||||
if UUID4_RE.fullmatch(text) is None:
|
||||
|
||||
@@ -118,6 +118,7 @@ class CoreModelsTests(unittest.TestCase):
|
||||
legal = claim_wire()
|
||||
legal_goods = "1" * 32
|
||||
legal["task"].update(
|
||||
version=9_223_372_036_854_775_807,
|
||||
title="😀" * 120,
|
||||
goods_id=legal_goods,
|
||||
product_url="https://mobile.yangkeduo.com/goods.html?goods_id=" + legal_goods,
|
||||
@@ -125,6 +126,7 @@ class CoreModelsTests(unittest.TestCase):
|
||||
sku_size="码" * 80,
|
||||
max_total_price="1" * 29 + ".00",
|
||||
)
|
||||
legal["authorization"]["task_version"] = 9_223_372_036_854_775_806
|
||||
claimed = ClaimedTask.from_wire(legal)
|
||||
self.assertEqual(len(claimed.task.title), 120)
|
||||
# Python's default ensure_ascii=True expands astral characters to surrogate
|
||||
@@ -166,8 +168,20 @@ class CoreModelsTests(unittest.TestCase):
|
||||
self.assertEqual(ClaimedTask.from_wire(value).task.title, "😀")
|
||||
|
||||
def test_title_rejects_ascii_and_unicode_whitespace_only(self) -> None:
|
||||
for title in ("", " \t\r\n", "\u3000", " \u3000\t"):
|
||||
for title in ("", " \t\r\n", "\u3000", " \u3000\t", "\u00a0title", "title\u00a0"):
|
||||
value = claim_wire()
|
||||
value["task"]["title"] = title
|
||||
with self.subTest(title=repr(title)), self.assertRaises(ValidationError):
|
||||
ClaimedTask.from_wire(value)
|
||||
|
||||
def test_persisted_text_has_runtime_independent_c0_and_nbsp_domain(self) -> None:
|
||||
for field in ("title", "sku_color", "sku_size"):
|
||||
for invalid in ("\u001cvalue", "value\u001f", "value\u001dinside", "\u00a0value", "value\u00a0"):
|
||||
value = claim_wire()
|
||||
value["task"][field] = invalid
|
||||
with self.subTest(field=field, invalid=repr(invalid)), self.assertRaises(ValidationError):
|
||||
ClaimedTask.from_wire(value)
|
||||
|
||||
value = claim_wire()
|
||||
value["task"][field] = "left\u00a0right"
|
||||
self.assertEqual(getattr(ClaimedTask.from_wire(value).task, field), "left\u00a0right")
|
||||
|
||||
Reference in New Issue
Block a user