From d6fe9b44834bb9f2fb15875df394db5fb18ee94c Mon Sep 17 00:00:00 2001 From: Vishal Katyal Date: Sat, 8 Aug 2026 17:46:48 -0400 Subject: [PATCH] fix: enforce propertyNames on extra-allow generated models signals.json declares propertyNames (reverse-domain keys) alongside named properties and additionalProperties:true. datamodel-code-generator emits this as class Signals(BaseModel) with model_config=ConfigDict(extra="allow") and the two named fields, so unknown (extra) keys are never checked against the key pattern. Observed: Signals(**{"dev.ucp.buyer_ip": "1.2.3.4", "bogus KEY!": "x"}) is accepted and "bogus KEY!" is kept in model_extra. Expected: the malformed key is rejected, because signals.json requires every property name to match the reverse-domain pattern ^[a-z][a-z0-9]*(?:\.[a-z][a-z0-9_]*)+$ (signals.json propertyNames.pattern, the same pattern as reverse_domain_name.json). Well-formed reverse-domain extras (e.g. com.example.device_id) must still be preserved under extra="allow". The already-enforced sibling case is the dict-keyed maps (e.g. ucp.json capabilities/services/payment_handlers/supported_versions), emitted as dict[ReverseDomainName, V] where pydantic validates the keys. The gap is only the extra-allow BaseModel shape: propertyNames declared on an object that also has named properties. This extends the post-generation model_validator approach added for minProperties (#49-class): postprocess_models.py scans the preprocessed schemas for objects that declare propertyNames AND carry named properties, reads the key pattern from the source schema (inline pattern or a $ref to e.g. reverse_domain_name.json, never duplicated in code), and injects a model_validator(mode="after") that matches every model_extra key against the pattern with re.fullmatch. fullmatch (not re.match) is used so a $-anchored pattern does not admit a trailing newline (re.match lets $ match before a final \n); this agrees with pydantic-core / ECMA-262 (JSON Schema's regex dialect) key semantics, the same behavior the dict-keyed map path already applies. The check reaches the base model and its generated request variants (Signals, SignalsCreateRequest, SignalsUpdateRequest, SignalsCompleteRequest). Out of scope: identity_linking.json scopes map degrades to Any due to an allOf/$ref resolution problem, so no propertyNames-bearing model is emitted for it (a separate defect); and the dict-keyed maps above already enforce their key pattern. Regenerated with ./generate_models.sh 2026-04-08; the diff is limited to the propertyNames enforcement on the four Signals models and regeneration is byte-identical. --- postprocess_models.py | 219 +++++++++++++++++- .../models/schemas/shopping/types/signals.py | 17 +- .../types/signals_complete_request.py | 17 +- .../shopping/types/signals_create_request.py | 17 +- .../shopping/types/signals_update_request.py | 17 +- tests/test_codegen_pipeline.py | 174 ++++++++++++++ 6 files changed, 454 insertions(+), 7 deletions(-) diff --git a/postprocess_models.py b/postprocess_models.py index b1f43b9..a2c6cd2 100644 --- a/postprocess_models.py +++ b/postprocess_models.py @@ -14,7 +14,7 @@ """Post-generation fixes for constraints datamodel-code-generator ignores. -Three constraint families are handled: +Four constraint families are handled: * ``minProperties`` on an object schema WITH declared properties is dropped by the generator (issue #49): every field is optional, so an empty instance @@ -49,6 +49,18 @@ applied to the base model and to its generated request variants (linked by file stem), and travels wherever the alias is reused as a field type. +* ``propertyNames`` on an object WITH named ``properties`` is not enforced. Such + a schema is emitted as a ``BaseModel(extra="allow")`` with the named fields, so + unknown (extra) keys are accepted without being checked against the declared + key pattern (``signals.json`` requires reverse-domain keys, yet a malformed + extra key validates). The script scans for objects that declare + ``propertyNames`` AND carry named ``properties`` and injects a + ``model_validator(mode="after")`` that matches every ``model_extra`` key against + the pattern. The pattern is read from the source schema (inline or via ``$ref`` + to e.g. ``reverse_domain_name.json``), never duplicated here. An object with + ``propertyNames`` but *no* named properties is emitted as a ``dict[KeyType, V]`` + whose key type already carries the pattern, so it is out of scope. + * ``uniqueItems`` on an array is dropped entirely by the generator, so a list field accepts duplicate entries in violation of the schema. The script collects the names of array properties declared with ``uniqueItems`` and @@ -86,6 +98,23 @@ def {marker}(self): return self ''' +_PROPNAMES_MARKER = "_enforce_property_names" + +_PROPNAMES_VALIDATOR_TEMPLATE = ''' + @model_validator(mode="after") + def {marker}(self): + """JSON Schema propertyNames: every extra key must match the + declared reverse-domain pattern (schema propertyNames).""" + pattern = {pattern!r} + for key in self.model_extra or {{}}: + if re.fullmatch(pattern, key) is None: + raise ValueError( + f"Property name {{key!r}} does not match the schema " + f"propertyNames pattern {{pattern}}" + ) + return self +''' + _UNIQUE_MARKER = "_enforce_unique_items" _UNIQUE_VALIDATOR_TEMPLATE = ''' @@ -144,6 +173,153 @@ def _ensure_pydantic_import(source, symbol): ) +def _ensure_stdlib_import(source, statement): + """Add a top-level ``import`` statement if absent. + + Inserted right after ``from __future__ import annotations`` so ruff's + isort pass (run later in the pipeline) settles it into the stdlib group. + """ + if re.search(rf"^{re.escape(statement)}$", source, re.M): + return source + return re.sub( + r"^(from __future__ import annotations\n)", + lambda m: f"{m.group(1)}\n{statement}\n", + source, + count=1, + flags=re.M, + ) + + +def _resolve_property_names_pattern(prop_names, schema_path): + """Return the key pattern a ``propertyNames`` node enforces, or ``None``. + + Reads an inline ``pattern`` directly, or follows a ``$ref`` to an external + schema file's root ``pattern`` (e.g. ``reverse_domain_name.json``) so the + pattern is never duplicated here — it always comes from the source schema. + Local ``#/...`` pointer refs are not resolved and are skipped with a + warning rather than guessed. + """ + if not isinstance(prop_names, dict): + return None + inline = prop_names.get("pattern") + if isinstance(inline, str): + return inline + ref = prop_names.get("$ref") + if not isinstance(ref, str): + return None + if ref.startswith("#"): + sys.stderr.write( + f" ! {schema_path}: propertyNames $ref '{ref}' is a local " + "pointer; pattern not resolved\n" + ) + return None + file_part = ref.split("#", 1)[0] + target = (Path(schema_path).parent / file_part).resolve() + try: + referenced = json.loads(target.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + sys.stderr.write( + f" ! {schema_path}: propertyNames $ref '{ref}' could not be " + "loaded; pattern not resolved\n" + ) + return None + pattern = ( + referenced.get("pattern") if isinstance(referenced, dict) else None + ) + if not isinstance(pattern, str): + sys.stderr.write( + f" ! {schema_path}: propertyNames $ref '{ref}' target has no " + "root pattern; not resolved\n" + ) + return None + return pattern + + +def find_property_names_patterns(schema_dir): + """Map generated class name -> propertyNames pattern for extra-allow models. + + The gap this targets: an object schema that declares ``propertyNames`` AND + carries named ``properties`` is emitted by the generator as a + ``BaseModel(extra="allow")`` with those named fields, so unknown (extra) + keys are never pattern-checked. An object with ``propertyNames`` but *no* + named ``properties`` is emitted as a ``dict[KeyType, V]`` map whose key type + already carries the pattern (pydantic validates the keys), so it is out of + scope. The class is defined mechanically: has ``propertyNames`` (resolvable + to a pattern) AND non-empty ``properties`` AND a ``title`` to map to a class. + Nested titled objects are walked too, so the rule is general, not per-file. + """ + found = {} + + def walk(node, path_str): + if not isinstance(node, dict): + if isinstance(node, list): + for item in node: + walk(item, path_str) + return + props = node.get("properties") + if "propertyNames" in node and isinstance(props, dict) and props: + pattern = _resolve_property_names_pattern( + node["propertyNames"], path_str + ) + title = node.get("title") + if pattern is None: + pass + elif not title: + sys.stderr.write( + f" ! {path_str}: propertyNames on an extra-allow object " + "but no title; cannot map to a class\n" + ) + else: + # The injected validator uses re.fullmatch to mirror + # pydantic-core / ECMA-262 (JSON Schema's regex dialect) key + # semantics, which the sibling dict-map path already applies. + # That is exact for the ^...$-anchored patterns UCP uses. An + # unanchored pattern means JSON Schema unanchored-search + # semantics, where fullmatch would over-restrict; warn so a + # future schema does not silently get a stricter check. + if not (pattern.startswith("^") and pattern.endswith("$")): + sys.stderr.write( + f" ! {path_str}: propertyNames pattern {pattern!r} is " + "not ^/$-anchored; fullmatch enforcement may be " + "stricter than JSON Schema search semantics\n" + ) + found[_alias_name(title)] = pattern + for value in node.values(): + walk(value, path_str) + + for path in sorted(Path(schema_dir).rglob("*.json")): + try: + schema = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + continue + walk(schema, str(path)) + return found + + +def inject_property_names(source, class_name, pattern): + """Inject the propertyNames key validator at the end of ``class_name``.""" + class_re = re.compile(rf"^class {re.escape(class_name)}\(", re.M) + match = class_re.search(source) + if not match: + return source + # The class body ends at the next top-level statement or EOF. + tail = re.compile(r"^\S", re.M) + end_match = tail.search(source, match.end()) + end = end_match.start() if end_match else len(source) + # Scope the idempotency guard to this class's own body, so a second + # target class in the same module is still patched. + if f"def {_PROPNAMES_MARKER}(" in source[match.start() : end]: + return source + method = _PROPNAMES_VALIDATOR_TEMPLATE.format( + marker=_PROPNAMES_MARKER, pattern=pattern + ) + body = source[:end].rstrip("\n") + rest = source[end:] + out = body + "\n" + method + ("\n" + rest if rest else "") + out = _ensure_pydantic_import(out, "model_validator") + return _ensure_stdlib_import(out, "import re") + + def inject_min_properties(source, class_name, minimum): """Inject the minProperties validator at the end of ``class_name``.""" if f"def {_MARKER}(" in source: @@ -540,6 +716,42 @@ def _array_contains_targets(): return targets +def _patch_property_names(): + """Inject propertyNames validators; return (patched_count, exit_code).""" + patterns = find_property_names_patterns(SCHEMA_DIR) + if not patterns: + sys.stdout.write( + "postprocess: no propertyNames constraints on extra-allow " + "models found\n" + ) + return 0, 0 + patched = 0 + for class_name, pattern in sorted(patterns.items()): + hits = [] + for path in sorted(OUTPUT_DIR.rglob("*.py")): + source = path.read_text(encoding="utf-8") + if not re.search( + rf"^class {re.escape(class_name)}\(", source, re.M + ): + continue + updated = inject_property_names(source, class_name, pattern) + if updated != source: + path.write_text(updated, encoding="utf-8") + patched += 1 + hits.append(path) + label = ", ".join(str(h) for h in hits) or "NO GENERATED CLASS FOUND" + sys.stdout.write( + f" propertyNames {pattern!r} on '{class_name}' -> {label}\n" + ) + if not hits: + sys.stderr.write( + f" ! '{class_name}' has no generated class; " + "constraint not enforced\n" + ) + return patched, 1 + return patched, 0 + + def _patch_array_contains(): """Inject array-contains validators; return (patched_count, exit_code).""" targets = _array_contains_targets() @@ -606,11 +818,12 @@ def _patch_unique_items(): def main(): """Main entry point to scan schemas and patch generated models.""" patched_mp, rc_mp = _patch_min_properties() + patched_pn, rc_pn = _patch_property_names() patched_ac, rc_ac = _patch_array_contains() patched_ui, rc_ui = _patch_unique_items() - total = patched_mp + patched_ac + patched_ui + total = patched_mp + patched_pn + patched_ac + patched_ui sys.stdout.write(f"postprocess: {total} module(s) patched\n") - return rc_mp or rc_ac or rc_ui + return rc_mp or rc_pn or rc_ac or rc_ui if __name__ == "__main__": diff --git a/src/ucp_sdk/models/schemas/shopping/types/signals.py b/src/ucp_sdk/models/schemas/shopping/types/signals.py index d80f329..2e0ad9c 100644 --- a/src/ucp_sdk/models/schemas/shopping/types/signals.py +++ b/src/ucp_sdk/models/schemas/shopping/types/signals.py @@ -18,7 +18,9 @@ from __future__ import annotations -from pydantic import BaseModel, ConfigDict, Field +import re + +from pydantic import BaseModel, ConfigDict, Field, model_validator class Signals(BaseModel): @@ -37,3 +39,16 @@ class Signals(BaseModel): """ Client's HTTP User-Agent header or equivalent. """ + + @model_validator(mode="after") + def _enforce_property_names(self): + """JSON Schema propertyNames: every extra key must match the + declared reverse-domain pattern (schema propertyNames).""" + pattern = "^[a-z][a-z0-9]*(?:\\.[a-z][a-z0-9_]*)+$" + for key in self.model_extra or {}: + if re.fullmatch(pattern, key) is None: + raise ValueError( + f"Property name {key!r} does not match the schema " + f"propertyNames pattern {pattern}" + ) + return self diff --git a/src/ucp_sdk/models/schemas/shopping/types/signals_complete_request.py b/src/ucp_sdk/models/schemas/shopping/types/signals_complete_request.py index 436b901..fa2ecbf 100644 --- a/src/ucp_sdk/models/schemas/shopping/types/signals_complete_request.py +++ b/src/ucp_sdk/models/schemas/shopping/types/signals_complete_request.py @@ -18,7 +18,9 @@ from __future__ import annotations -from pydantic import BaseModel, ConfigDict, Field +import re + +from pydantic import BaseModel, ConfigDict, Field, model_validator class SignalsCompleteRequest(BaseModel): @@ -37,3 +39,16 @@ class SignalsCompleteRequest(BaseModel): """ Client's HTTP User-Agent header or equivalent. """ + + @model_validator(mode="after") + def _enforce_property_names(self): + """JSON Schema propertyNames: every extra key must match the + declared reverse-domain pattern (schema propertyNames).""" + pattern = "^[a-z][a-z0-9]*(?:\\.[a-z][a-z0-9_]*)+$" + for key in self.model_extra or {}: + if re.fullmatch(pattern, key) is None: + raise ValueError( + f"Property name {key!r} does not match the schema " + f"propertyNames pattern {pattern}" + ) + return self diff --git a/src/ucp_sdk/models/schemas/shopping/types/signals_create_request.py b/src/ucp_sdk/models/schemas/shopping/types/signals_create_request.py index fd1630d..df0b091 100644 --- a/src/ucp_sdk/models/schemas/shopping/types/signals_create_request.py +++ b/src/ucp_sdk/models/schemas/shopping/types/signals_create_request.py @@ -18,7 +18,9 @@ from __future__ import annotations -from pydantic import BaseModel, ConfigDict, Field +import re + +from pydantic import BaseModel, ConfigDict, Field, model_validator class SignalsCreateRequest(BaseModel): @@ -37,3 +39,16 @@ class SignalsCreateRequest(BaseModel): """ Client's HTTP User-Agent header or equivalent. """ + + @model_validator(mode="after") + def _enforce_property_names(self): + """JSON Schema propertyNames: every extra key must match the + declared reverse-domain pattern (schema propertyNames).""" + pattern = "^[a-z][a-z0-9]*(?:\\.[a-z][a-z0-9_]*)+$" + for key in self.model_extra or {}: + if re.fullmatch(pattern, key) is None: + raise ValueError( + f"Property name {key!r} does not match the schema " + f"propertyNames pattern {pattern}" + ) + return self diff --git a/src/ucp_sdk/models/schemas/shopping/types/signals_update_request.py b/src/ucp_sdk/models/schemas/shopping/types/signals_update_request.py index 09ea148..37d37a8 100644 --- a/src/ucp_sdk/models/schemas/shopping/types/signals_update_request.py +++ b/src/ucp_sdk/models/schemas/shopping/types/signals_update_request.py @@ -18,7 +18,9 @@ from __future__ import annotations -from pydantic import BaseModel, ConfigDict, Field +import re + +from pydantic import BaseModel, ConfigDict, Field, model_validator class SignalsUpdateRequest(BaseModel): @@ -37,3 +39,16 @@ class SignalsUpdateRequest(BaseModel): """ Client's HTTP User-Agent header or equivalent. """ + + @model_validator(mode="after") + def _enforce_property_names(self): + """JSON Schema propertyNames: every extra key must match the + declared reverse-domain pattern (schema propertyNames).""" + pattern = "^[a-z][a-z0-9]*(?:\\.[a-z][a-z0-9_]*)+$" + for key in self.model_extra or {}: + if re.fullmatch(pattern, key) is None: + raise ValueError( + f"Property name {key!r} does not match the schema " + f"propertyNames pattern {pattern}" + ) + return self diff --git a/tests/test_codegen_pipeline.py b/tests/test_codegen_pipeline.py index d601a64..b2aabf9 100644 --- a/tests/test_codegen_pipeline.py +++ b/tests/test_codegen_pipeline.py @@ -658,6 +658,180 @@ def test_all_fields_accepted(self): Description(plain="p", html="

p

", markdown="p") +@unittest.skipUnless( + HAVE_SDK, "requires the installed package (pip install -e .)" +) +class SignalsPropertyNamesTest(unittest.TestCase): + """signals.json declares propertyNames (reverse-domain keys). + + Signals has named ``properties`` AND ``additionalProperties: true``, so the + generator emits ``class Signals(BaseModel)`` with ``extra="allow"`` and named + fields; extra keys bypass the ``propertyNames`` pattern. The post-generation + injector restores the check on every ``model_extra`` key while preserving + well-formed reverse-domain extras (extra="allow" keeps them). + """ + + def _signals(self): + from ucp_sdk.models.schemas.shopping.types.signals import Signals + + return Signals + + def test_malformed_extra_key_rejected(self): + with self.assertRaisesRegex(ValidationError, "propertyNames"): + self._signals().model_validate( + {"dev.ucp.buyer_ip": "1.2.3.4", "bogus KEY!": "x"} + ) + + def test_trailing_newline_key_rejected(self): + # A $-anchored pattern with re.match would let a trailing newline + # slip through; the enforcement uses re.fullmatch to agree with + # pydantic-core's key validation on the sibling dict-map path. + with self.assertRaisesRegex(ValidationError, "propertyNames"): + self._signals().model_validate({"com.example.k\n": "x"}) + + def test_valid_reverse_domain_extra_accepted_and_preserved(self): + signals = self._signals().model_validate( + {"com.example.device_id": "abc123"} + ) + # extra="allow" must still keep a well-formed extra key. + self.assertEqual( + signals.model_extra, {"com.example.device_id": "abc123"} + ) + + def test_known_named_fields_still_work(self): + signals = self._signals().model_validate( + { + "dev.ucp.buyer_ip": "1.2.3.4", + "dev.ucp.user_agent": "curl/8", + } + ) + self.assertEqual(signals.dev_ucp_buyer_ip, "1.2.3.4") + self.assertEqual(signals.dev_ucp_user_agent, "curl/8") + self.assertEqual(signals.model_extra, {}) + + def test_request_variants_enforce_property_names(self): + # The gap and its fix travel to the generated request variants too. + from ucp_sdk.models.schemas.shopping.types.signals_complete_request import ( + SignalsCompleteRequest, + ) + from ucp_sdk.models.schemas.shopping.types.signals_create_request import ( + SignalsCreateRequest, + ) + from ucp_sdk.models.schemas.shopping.types.signals_update_request import ( + SignalsUpdateRequest, + ) + + for cls in ( + SignalsCreateRequest, + SignalsUpdateRequest, + SignalsCompleteRequest, + ): + with self.subTest(model=cls.__name__): + with self.assertRaisesRegex(ValidationError, "propertyNames"): + cls.model_validate({"bogus KEY!": "x"}) + self.assertEqual( + cls.model_validate({"com.example.k": "v"}).model_extra, + {"com.example.k": "v"}, + ) + + +class PropertyNamesInjectorTest(unittest.TestCase): + """The propertyNames post-generation injector's own behavior.""" + + PATTERN = "^[a-z][a-z0-9]*(?:\\.[a-z][a-z0-9_]*)+$" + + SCHEMA = { + "title": "Signals", + "type": "object", + "propertyNames": {"pattern": PATTERN}, + "properties": {"dev.ucp.buyer_ip": {"type": "string"}}, + "additionalProperties": True, + } + + # An object with propertyNames but no named properties is a dict-map + # (key type already carries the pattern) — out of scope. + DICT_MAP_SCHEMA = { + "title": "Requires", + "type": "object", + "propertyNames": {"pattern": PATTERN}, + "additionalProperties": {"type": "string"}, + } + + MODULE = ( + "from __future__ import annotations\n" + "\n" + "from pydantic import BaseModel, ConfigDict, Field\n" + "\n" + "\n" + "class Signals(BaseModel):\n" + ' """Signals."""\n' + "\n" + " model_config = ConfigDict(\n" + ' extra="allow",\n' + " )\n" + ' dev_ucp_buyer_ip: str | None = Field(None, alias="dev.ucp.buyer_ip")\n' + ) + + def test_scan_finds_only_extra_allow_object(self): + with tempfile.TemporaryDirectory() as tmp: + (Path(tmp) / "signals.json").write_text(json.dumps(self.SCHEMA)) + (Path(tmp) / "requires.json").write_text( + json.dumps(self.DICT_MAP_SCHEMA) + ) + found = postprocess_models.find_property_names_patterns(Path(tmp)) + self.assertEqual(found, {"Signals": self.PATTERN}) + + def test_scan_resolves_ref_pattern(self): + with tempfile.TemporaryDirectory() as tmp: + (Path(tmp) / "reverse_domain_name.json").write_text( + json.dumps({"type": "string", "pattern": self.PATTERN}) + ) + (Path(tmp) / "thing.json").write_text( + json.dumps( + { + "title": "Thing", + "type": "object", + "propertyNames": {"$ref": "reverse_domain_name.json"}, + "properties": {"a": {"type": "string"}}, + } + ) + ) + found = postprocess_models.find_property_names_patterns(Path(tmp)) + self.assertEqual(found, {"Thing": self.PATTERN}) + + def test_injects_validator_and_imports(self): + out = postprocess_models.inject_property_names( + self.MODULE, "Signals", self.PATTERN + ) + self.assertIn("model_validator", out) + self.assertIn("import re", out) + self.assertIn("propertyNames", out) + + def test_injection_is_idempotent(self): + once = postprocess_models.inject_property_names( + self.MODULE, "Signals", self.PATTERN + ) + twice = postprocess_models.inject_property_names( + once, "Signals", self.PATTERN + ) + self.assertEqual(once, twice) + + @unittest.skipUnless(HAVE_SDK, "executing the module needs pydantic") + def test_injected_validator_enforces_pattern(self): + out = postprocess_models.inject_property_names( + self.MODULE, "Signals", self.PATTERN + ) + namespace: dict = {} + exec(compile(out, "", "exec"), namespace) # noqa: S102 + signals_cls = namespace["Signals"] + with self.assertRaises(ValidationError): + signals_cls.model_validate({"bogus KEY!": "x"}) + # fullmatch (not match) — a trailing newline must not slip through. + with self.assertRaises(ValidationError): + signals_cls.model_validate({"com.example.ok\n": "v"}) + signals_cls.model_validate({"com.example.ok": "v"}) + + class InjectorTest(unittest.TestCase): """The post-generation injector's own behavior."""