feat: add x-algokit-byte-length vendor extension support for fixed-length byte validation

- Add byte_length and list_inner_byte_length fields to TypeInfo in builder.py
- Update _build_metadata() to generate fixed-length byte serde helpers
- Add encode_fixed_bytes_base64/decode_fixed_bytes_base64 helpers for 32/64 byte validation
- Add encode_fixed_bytes_sequence/decode_fixed_bytes_sequence for array validation
- Regenerate API clients with fixed-length byte fields (group, lease, signature, etc.)
- Update OAS_BRANCH to fix/byte-len-validation for spec generation

Author: aorumbayev

api/oas-generator/src/oas_generator/builder.py

@@ -33,6 +33,8 @@ class TypeInfo:
     is_locals_reference: bool = False
     is_holding_reference: bool = False
     needs_datetime: bool = False
+    byte_length: int | None = None  # Fixed byte length from x-algokit-byte-length
+    list_inner_byte_length: int | None = None  # Fixed byte length for list items
     imports: set[str] = field(default_factory=set)
 
 
@@ -170,7 +172,10 @@ def resolve(self, schema: ctx.RawSchema, *, hint: str = "Inline") -> TypeInfo:
         if schema_type == "string":
             fmt = schema.get("format")
             if fmt in {"byte", "binary"} or schema.get("x-algokit-bytes-base64"):
-                info = TypeInfo(annotation="bytes", is_bytes=True)
+                # Extract fixed byte length if present
+                byte_length_val = schema.get("x-algokit-byte-length")
+                byte_length = int(byte_length_val) if byte_length_val is not None else None
+                info = TypeInfo(annotation="bytes", is_bytes=True, byte_length=byte_length)
             elif fmt == "date-time":
                 info = TypeInfo(
                     annotation="datetime",
@@ -219,6 +224,7 @@ def _resolve_array(self, schema: ctx.RawSchema, *, hint: str) -> TypeInfo:
             list_inner_model=inner.model,
             list_inner_enum=inner.enum,
             list_inner_is_bytes=inner.is_bytes,
+            list_inner_byte_length=inner.byte_length,
             is_signed_transaction=inner.is_signed_transaction,
             is_box_reference=inner.is_box_reference,
             is_locals_reference=inner.is_locals_reference,
@@ -462,6 +468,10 @@ def _build_model(self, entry: SchemaEntry) -> ctx.ModelDescriptor:  # noqa: C901
                 imports.add("from ._serde_helpers import decode_bytes_base64, encode_bytes_base64")
             if "encode_bytes_sequence" in field.metadata or "decode_bytes_sequence" in field.metadata:
                 imports.add("from ._serde_helpers import decode_bytes_sequence, encode_bytes_sequence")
+            if "encode_fixed_bytes_base64" in field.metadata or "decode_fixed_bytes_base64" in field.metadata:
+                imports.add("from ._serde_helpers import decode_fixed_bytes_base64, encode_fixed_bytes_base64")
+            if "encode_fixed_bytes_sequence" in field.metadata or "decode_fixed_bytes_sequence" in field.metadata:
+                imports.add("from ._serde_helpers import decode_fixed_bytes_sequence, encode_fixed_bytes_sequence")
             if "nested(" in field.metadata:
                 uses_nested = True
             if "flatten(" in field.metadata:
@@ -585,6 +595,15 @@ def _build_metadata(self, wire_name: str, type_info: TypeInfo, *, required: bool
                 "        )"
             )
         if type_info.is_list and type_info.list_inner_is_bytes:
+            # Handle fixed-length bytes in sequences
+            if type_info.list_inner_byte_length is not None:
+                return (
+                    "wire(\n"
+                    f'            "{alias}",\n'
+                    f"            encode=lambda v: encode_fixed_bytes_sequence(v, {type_info.list_inner_byte_length}),\n"
+                    f"            decode=lambda raw: decode_fixed_bytes_sequence(raw, {type_info.list_inner_byte_length}),\n"
+                    "        )"
+                )
             return (
                 "wire(\n"
                 f'            "{alias}",\n'
@@ -593,6 +612,15 @@ def _build_metadata(self, wire_name: str, type_info: TypeInfo, *, required: bool
                 "        )"
             )
         if type_info.is_bytes:
+            # Handle fixed-length bytes
+            if type_info.byte_length is not None:
+                return (
+                    "wire(\n"
+                    f'            "{alias}",\n'
+                    f"            encode=lambda v: encode_fixed_bytes_base64(v, {type_info.byte_length}),\n"
+                    f"            decode=lambda raw: decode_fixed_bytes_base64(raw, {type_info.byte_length}),\n"
+                    "        )"
+                )
             return (
                 "wire(\n"
                 f'            "{alias}",\n'

api/oas-generator/src/oas_generator/renderer/templates/models/_serde_helpers.py.j2

@@ -37,6 +37,22 @@ def decode_bytes_base64(raw: object) -> bytes:
     raise TypeError(f"Unsupported value for bytes field: {type(raw)!r}")
 
 
+def encode_fixed_bytes_base64(value: BytesLike, expected_length: int) -> str:
+    """Encode fixed-length bytes to base64, validating the length."""
+    coerced = _coerce_bytes(value)
+    if len(coerced) != expected_length:
+        raise ValueError(f"Expected {expected_length} bytes, got {len(coerced)}")
+    return base64.b64encode(coerced).decode("ascii")
+
+
+def decode_fixed_bytes_base64(raw: object, expected_length: int) -> bytes:
+    """Decode base64 to fixed-length bytes, validating the length."""
+    decoded = decode_bytes_base64(raw)
+    if len(decoded) != expected_length:
+        raise ValueError(f"Expected {expected_length} bytes, got {len(decoded)}")
+    return decoded
+
+
 def decode_bytes_map_key(raw: object) -> bytes:
     if isinstance(raw, bytes | bytearray | memoryview):
         return bytes(raw)
@@ -77,6 +93,36 @@ def decode_bytes_sequence(raw: object) -> list[bytes | None] | None:
     return decoded or None
 
 
+def encode_fixed_bytes_sequence(
+    values: Iterable[BytesLike | None] | None, expected_length: int
+) -> list[str | None] | None:
+    """Encode a sequence of fixed-length bytes to base64, validating each element's length."""
+    if values is None:
+        return None
+    encoded: list[str | None] = []
+    for value in values:
+        if value is None:
+            encoded.append(None)
+            continue
+        if not isinstance(value, bytes | bytearray | memoryview):
+            raise TypeError(f"Unsupported value for bytes field sequence: {type(value)!r}")
+        encoded.append(encode_fixed_bytes_base64(value, expected_length))
+    return encoded or None
+
+
+def decode_fixed_bytes_sequence(raw: object, expected_length: int) -> list[bytes | None] | None:
+    """Decode a sequence of base64 strings to fixed-length bytes, validating each element's length."""
+    if not isinstance(raw, list):
+        return None
+    decoded: list[bytes | None] = []
+    for item in raw:
+        if item is None:
+            decoded.append(None)
+            continue
+        decoded.append(decode_fixed_bytes_base64(item, expected_length))
+    return decoded or None
+
+
 def encode_model_sequence(values: Iterable[object] | None) -> list[dict[str, object]] | None:
     if values is None:
         return None

pyproject.toml

@@ -239,7 +239,7 @@ sequence = [
   { cmd = "ruff check --fix src/algokit_${SPEC}_client" },
   { cmd = "ruff format src/algokit_${SPEC}_client" },
 ]
-env.OAS_BRANCH = "main"
+env.OAS_BRANCH = "fix/byte-len-validation"
 
 [tool.poe.tasks.generate-algod-client]
 ref = "generate-client"

src/algokit_algod_client/models/_account_participation.py

@@ -5,7 +5,7 @@
 
 from algokit_common.serde import wire
 
-from ._serde_helpers import decode_bytes_base64, encode_bytes_base64
+from ._serde_helpers import decode_fixed_bytes_base64, encode_fixed_bytes_base64
 
 
 @dataclass(slots=True)
@@ -19,8 +19,8 @@ class AccountParticipation:
         default=b"",
         metadata=wire(
             "selection-participation-key",
-            encode=encode_bytes_base64,
-            decode=decode_bytes_base64,
+            encode=lambda v: encode_fixed_bytes_base64(v, 32),
+            decode=lambda raw: decode_fixed_bytes_base64(raw, 32),
         ),
     )
     vote_first_valid: int = field(
@@ -39,15 +39,15 @@ class AccountParticipation:
         default=b"",
         metadata=wire(
             "vote-participation-key",
-            encode=encode_bytes_base64,
-            decode=decode_bytes_base64,
+            encode=lambda v: encode_fixed_bytes_base64(v, 32),
+            decode=lambda raw: decode_fixed_bytes_base64(raw, 32),
         ),
     )
     state_proof_key: bytes | None = field(
         default=None,
         metadata=wire(
             "state-proof-key",
-            encode=encode_bytes_base64,
-            decode=decode_bytes_base64,
+            encode=lambda v: encode_fixed_bytes_base64(v, 64),
+            decode=lambda raw: decode_fixed_bytes_base64(raw, 64),
         ),
     )

src/algokit_algod_client/models/_asset_params.py

@@ -5,7 +5,12 @@
 
 from algokit_common.serde import wire
 
-from ._serde_helpers import decode_bytes_base64, encode_bytes_base64
+from ._serde_helpers import (
+    decode_bytes_base64,
+    decode_fixed_bytes_base64,
+    encode_bytes_base64,
+    encode_fixed_bytes_base64,
+)
 
 
 @dataclass(slots=True)
@@ -51,8 +56,8 @@ class AssetParams:
         default=None,
         metadata=wire(
             "metadata-hash",
-            encode=encode_bytes_base64,
-            decode=decode_bytes_base64,
+            encode=lambda v: encode_fixed_bytes_base64(v, 32),
+            decode=lambda raw: decode_fixed_bytes_base64(raw, 32),
         ),
     )
     name: str | None = field(

src/algokit_algod_client/models/_serde_helpers.py

@@ -37,6 +37,22 @@ def decode_bytes_base64(raw: object) -> bytes:
     raise TypeError(f"Unsupported value for bytes field: {type(raw)!r}")
 
 
+def encode_fixed_bytes_base64(value: BytesLike, expected_length: int) -> str:
+    """Encode fixed-length bytes to base64, validating the length."""
+    coerced = _coerce_bytes(value)
+    if len(coerced) != expected_length:
+        raise ValueError(f"Expected {expected_length} bytes, got {len(coerced)}")
+    return base64.b64encode(coerced).decode("ascii")
+
+
+def decode_fixed_bytes_base64(raw: object, expected_length: int) -> bytes:
+    """Decode base64 to fixed-length bytes, validating the length."""
+    decoded = decode_bytes_base64(raw)
+    if len(decoded) != expected_length:
+        raise ValueError(f"Expected {expected_length} bytes, got {len(decoded)}")
+    return decoded
+
+
 def decode_bytes_map_key(raw: object) -> bytes:
     if isinstance(raw, bytes | bytearray | memoryview):
         return bytes(raw)
@@ -77,6 +93,36 @@ def decode_bytes_sequence(raw: object) -> list[bytes | None] | None:
     return decoded or None
 
 
+def encode_fixed_bytes_sequence(
+    values: Iterable[BytesLike | None] | None, expected_length: int
+) -> list[str | None] | None:
+    """Encode a sequence of fixed-length bytes to base64, validating each element's length."""
+    if values is None:
+        return None
+    encoded: list[str | None] = []
+    for value in values:
+        if value is None:
+            encoded.append(None)
+            continue
+        if not isinstance(value, bytes | bytearray | memoryview):
+            raise TypeError(f"Unsupported value for bytes field sequence: {type(value)!r}")
+        encoded.append(encode_fixed_bytes_base64(value, expected_length))
+    return encoded or None
+
+
+def decode_fixed_bytes_sequence(raw: object, expected_length: int) -> list[bytes | None] | None:
+    """Decode a sequence of base64 strings to fixed-length bytes, validating each element's length."""
+    if not isinstance(raw, list):
+        return None
+    decoded: list[bytes | None] = []
+    for item in raw:
+        if item is None:
+            decoded.append(None)
+            continue
+        decoded.append(decode_fixed_bytes_base64(item, expected_length))
+    return decoded or None
+
+
 def encode_model_sequence(values: Iterable[object] | None) -> list[dict[str, object]] | None:
     if values is None:
         return None

src/algokit_algod_client/models/_transaction_parameters_response.py

@@ -5,7 +5,7 @@
 
 from algokit_common.serde import wire
 
-from ._serde_helpers import decode_bytes_base64, encode_bytes_base64
+from ._serde_helpers import decode_fixed_bytes_base64, encode_fixed_bytes_base64
 
 
 @dataclass(slots=True)
@@ -27,8 +27,8 @@ class TransactionParametersResponse:
         default=b"",
         metadata=wire(
             "genesis-hash",
-            encode=encode_bytes_base64,
-            decode=decode_bytes_base64,
+            encode=lambda v: encode_fixed_bytes_base64(v, 32),
+            decode=lambda raw: decode_fixed_bytes_base64(raw, 32),
         ),
     )
     genesis_id: str = field(

src/algokit_indexer_client/models/_account_participation.py

@@ -5,7 +5,7 @@
 
 from algokit_common.serde import wire
 
-from ._serde_helpers import decode_bytes_base64, encode_bytes_base64
+from ._serde_helpers import decode_fixed_bytes_base64, encode_fixed_bytes_base64
 
 
 @dataclass(slots=True)
@@ -19,8 +19,8 @@ class AccountParticipation:
         default=b"",
         metadata=wire(
             "selection-participation-key",
-            encode=encode_bytes_base64,
-            decode=decode_bytes_base64,
+            encode=lambda v: encode_fixed_bytes_base64(v, 32),
+            decode=lambda raw: decode_fixed_bytes_base64(raw, 32),
         ),
     )
     vote_first_valid: int = field(
@@ -39,15 +39,15 @@ class AccountParticipation:
         default=b"",
         metadata=wire(
             "vote-participation-key",
-            encode=encode_bytes_base64,
-            decode=decode_bytes_base64,
+            encode=lambda v: encode_fixed_bytes_base64(v, 32),
+            decode=lambda raw: decode_fixed_bytes_base64(raw, 32),
         ),
     )
     state_proof_key: bytes | None = field(
         default=None,
         metadata=wire(
             "state-proof-key",
-            encode=encode_bytes_base64,
-            decode=decode_bytes_base64,
+            encode=lambda v: encode_fixed_bytes_base64(v, 64),
+            decode=lambda raw: decode_fixed_bytes_base64(raw, 64),
         ),
     )

src/algokit_indexer_client/models/_asset_params.py

@@ -5,7 +5,12 @@
 
 from algokit_common.serde import wire
 
-from ._serde_helpers import decode_bytes_base64, encode_bytes_base64
+from ._serde_helpers import (
+    decode_bytes_base64,
+    decode_fixed_bytes_base64,
+    encode_bytes_base64,
+    encode_fixed_bytes_base64,
+)
 
 
 @dataclass(slots=True)
@@ -51,8 +56,8 @@ class AssetParams:
         default=None,
         metadata=wire(
             "metadata-hash",
-            encode=encode_bytes_base64,
-            decode=decode_bytes_base64,
+            encode=lambda v: encode_fixed_bytes_base64(v, 32),
+            decode=lambda raw: decode_fixed_bytes_base64(raw, 32),
         ),
     )
     name: str | None = field(