|
8 | 8 | from string import hexdigits |
9 | 9 | from typing import Any, Mapping, Protocol |
10 | 10 |
|
| 11 | +from govengine._json_boundary import bounded_json_copy |
11 | 12 | from govengine.api import GovApiError, require_mapping |
12 | 13 | from govengine.core import ArtifactDescriptor, GovernanceContext, ReasonCode, TransitionDecision |
13 | 14 |
|
@@ -146,8 +147,9 @@ def __post_init__(self) -> None: |
146 | 147 |
|
147 | 148 | @classmethod |
148 | 149 | def from_mapping(cls, value: Mapping[str, Any]) -> "KeyResolutionResult": |
149 | | - raw = require_mapping(value, reason_code="invalid_key_resolution_result") |
150 | | - _reject_forbidden_trust_material(raw) |
| 150 | + raw = _bounded_trust_mapping( |
| 151 | + require_mapping(value, reason_code="invalid_key_resolution_result") |
| 152 | + ) |
151 | 153 | metadata = raw.get("metadata") if isinstance(raw.get("metadata"), Mapping) else {} |
152 | 154 | return cls( |
153 | 155 | status=str(raw.get("status") or ""), |
@@ -195,8 +197,9 @@ def __post_init__(self) -> None: |
195 | 197 |
|
196 | 198 | @classmethod |
197 | 199 | def from_mapping(cls, value: Mapping[str, Any]) -> "TrustStoreDecision": |
198 | | - raw = require_mapping(value, reason_code="invalid_trust_store_decision") |
199 | | - _reject_forbidden_trust_material(raw) |
| 200 | + raw = _bounded_trust_mapping( |
| 201 | + require_mapping(value, reason_code="invalid_trust_store_decision") |
| 202 | + ) |
200 | 203 | metadata = raw.get("metadata") if isinstance(raw.get("metadata"), Mapping) else {} |
201 | 204 | return cls( |
202 | 205 | status=str(raw.get("status") or raw.get("trust_status") or ""), |
@@ -435,14 +438,27 @@ def _validate_govengine_record_digest(record_digest: str) -> str: |
435 | 438 |
|
436 | 439 | def _bounded_trust_metadata(value: Mapping[str, Any] | None) -> dict[str, Any]: |
437 | 440 | metadata = value if isinstance(value, Mapping) else {} |
438 | | - _reject_forbidden_trust_material(metadata) |
439 | | - return dict(metadata) |
| 441 | + return _bounded_trust_mapping(metadata) |
| 442 | + |
440 | 443 |
|
| 444 | +def _bounded_trust_mapping(value: Mapping[str, Any]) -> dict[str, Any]: |
| 445 | + copied = bounded_json_copy(value) |
| 446 | + if not isinstance(copied, dict): |
| 447 | + raise GovApiError("invalid_trust_metadata") |
| 448 | + _reject_forbidden_trust_material(copied) |
| 449 | + return copied |
441 | 450 |
|
442 | | -def _reject_forbidden_trust_material(value: Mapping[str, Any]) -> None: |
443 | | - for key in value: |
444 | | - if str(key).lower() in FORBIDDEN_TRUST_MATERIAL_KEYS: |
445 | | - raise GovApiError("forbidden_trust_material") |
| 451 | + |
| 452 | +def _reject_forbidden_trust_material(value: Any) -> None: |
| 453 | + if isinstance(value, Mapping): |
| 454 | + for key, nested in value.items(): |
| 455 | + if key.strip().lower() in FORBIDDEN_TRUST_MATERIAL_KEYS: |
| 456 | + raise GovApiError("forbidden_trust_material") |
| 457 | + _reject_forbidden_trust_material(nested) |
| 458 | + return |
| 459 | + if isinstance(value, (list, tuple)): |
| 460 | + for nested in value: |
| 461 | + _reject_forbidden_trust_material(nested) |
446 | 462 |
|
447 | 463 |
|
448 | 464 | def _canonical_record_value(value: Any) -> Any: |
|
0 commit comments