Validate key and ciphertext lengths before passing them to liboqs

oqs/oqs.py

@@ -319,6 +319,19 @@ def oqs_version() -> str:
         )
 
 
+def _check_length(name: str, value: bytes, length: int) -> None:
+    """
+    Raise ValueError unless value is exactly length bytes.
+
+    ctypes.create_string_buffer(value, length) silently zero-pads a short value,
+    and liboqs reads exactly length bytes, so a short key or ciphertext must be
+    rejected before it reaches either.
+    """
+    if len(value) != length:
+        msg = f"{name} length must be {length} bytes, got {len(value)} bytes"
+        raise ValueError(msg)
+
+
 class MechanismNotSupportedError(Exception):
     """Exception raised when an algorithm is not supported by OQS."""
 
@@ -429,7 +442,8 @@ def __init__(self, alg_name: str, secret_key: Union[int, bytes, None] = None) ->
             "length_keypair_seed": int(self.length_keypair_seed),
         }
 
-        if secret_key:
+        if secret_key is not None:
+            _check_length("Secret key", secret_key, self._kem.contents.length_secret_key)
             self.secret_key = ct.create_string_buffer(
                 secret_key,
                 self._kem.contents.length_secret_key,
@@ -507,6 +521,7 @@ def encap_secret(self, public_key: Union[int, bytes]) -> tuple[bytes, bytes]:
 
         :param public_key: the peer's public key.
         """
+        _check_length("Public key", public_key, self._kem.contents.length_public_key)
         c_public_key = ct.create_string_buffer(
             public_key,
             self._kem.contents.length_public_key,
@@ -534,6 +549,7 @@ def decap_secret(self, ciphertext: Union[int, bytes]) -> bytes:
 
         :param ciphertext: the ciphertext received from the peer.
         """
+        _check_length("Ciphertext", ciphertext, self._kem.contents.length_ciphertext)
         c_ciphertext = ct.create_string_buffer(
             ciphertext,
             self._kem.contents.length_ciphertext,
@@ -676,7 +692,8 @@ def __init__(self, alg_name: str, secret_key: Union[int, bytes, None] = None) ->
             "length_signature": int(self.length_signature),
         }
 
-        if secret_key:
+        if secret_key is not None:
+            _check_length("Secret key", secret_key, self._sig.contents.length_secret_key)
             self.secret_key = ct.create_string_buffer(
                 secret_key,
                 self._sig.contents.length_secret_key,
@@ -761,6 +778,7 @@ def verify(self, message: bytes, signature: bytes, public_key: bytes) -> bool:
         :param signature: the signature on the message.
         :param public_key: the signer's public key.
         """
+        _check_length("Public key", public_key, self._sig.contents.length_public_key)
         # Provide length to avoid extra null char
         c_message = ct.create_string_buffer(message, len(message))
         c_message_len = ct.c_size_t(len(c_message))
@@ -841,6 +859,7 @@ def verify_with_ctx_str(
         if context and not self.sig_with_ctx_support:
             msg = "Verifying with context string not supported"
             raise RuntimeError(msg)
+        _check_length("Public key", public_key, self._sig.contents.length_public_key)
 
         # Provide length to avoid extra null char
         c_message = ct.create_string_buffer(message, len(message))
@@ -1226,6 +1245,7 @@ def verify(self, message: bytes, signature: bytes, public_key: bytes) -> bool:
         :return: `True` if the signature is valid, `False` otherwise.
         """
         self._check_not_freed()
+        _check_length("Public key", public_key, self.length_public_key)
         msg = ct.create_string_buffer(message, len(message))
         sig = ct.create_string_buffer(signature, len(signature))
         pk = ct.create_string_buffer(public_key, len(public_key))

tests/test_kem.py

@@ -1,6 +1,7 @@
 import os
 import platform  # to learn the OS we're on
 import random
+from collections.abc import Callable
 from unittest import mock
 
 import oqs
@@ -142,17 +143,84 @@ def recording_free(ptr: object) -> None:
         real_free(ptr)
 
     for alg_name in oqs.get_enabled_kem_mechanisms():
+        with oqs.KeyEncapsulation(alg_name) as kem:
+            secret_key = b"\x01" * kem.length_secret_key
         freed.clear()
         with mock.patch.object(lib, "OQS_KEM_free", recording_free):
             # Passing a secret key exercises the cleanse path in free().
-            with oqs.KeyEncapsulation(alg_name, secret_key=b"\x01") as obj:
+            with oqs.KeyEncapsulation(alg_name, secret_key=secret_key) as obj:
                 pass
             obj.free()  # Explicit free after the context manager must be a no-op.
         if len(freed) != 1:
             msg = f"{alg_name}: OQS_KEM_free called {len(freed)} times"
             raise AssertionError(msg)
 
 
+def _wrong_lengths(length: int) -> set[int]:
+    # Empty, a bare 32-byte seed (e.g. ML-KEM rho with t-hat = 0), off by one either way
+    return {0, min(32, length - 1), length - 1, length + 1}
+
+
+def _assert_raises_value_error(
+    label: str, call: Callable[..., object], *args: object, **kwargs: object
+) -> None:
+    try:
+        call(*args, **kwargs)
+    except ValueError:
+        pass
+    else:
+        msg = f"{label} did not raise ValueError"
+        raise AssertionError(msg)
+
+
+def test_wrong_length_public_key() -> tuple[None, str]:
+    for alg_name in oqs.get_enabled_kem_mechanisms():
+        if any(item in alg_name for item in disabled_KEM_patterns):
+            continue
+        yield check_wrong_length_public_key, alg_name
+
+
+def check_wrong_length_public_key(alg_name: str) -> None:
+    with oqs.KeyEncapsulation(alg_name) as kem:
+        for n in _wrong_lengths(kem.length_public_key):
+            _assert_raises_value_error(
+                f"{alg_name}: encap_secret() with a {n}-byte public key",
+                kem.encap_secret,
+                bytes(n),
+            )
+
+
+def test_wrong_length_ciphertext() -> tuple[None, str]:
+    for alg_name in oqs.get_enabled_kem_mechanisms():
+        if any(item in alg_name for item in disabled_KEM_patterns):
+            continue
+        yield check_wrong_length_ciphertext, alg_name
+
+
+def check_wrong_length_ciphertext(alg_name: str) -> None:
+    with oqs.KeyEncapsulation(alg_name) as kem:
+        kem.generate_keypair()
+        for n in _wrong_lengths(kem.length_ciphertext):
+            _assert_raises_value_error(
+                f"{alg_name}: decap_secret() with a {n}-byte ciphertext",
+                kem.decap_secret,
+                bytes(n),
+            )
+
+
+def test_wrong_length_secret_key() -> None:
+    for alg_name in oqs.get_enabled_kem_mechanisms():
+        with oqs.KeyEncapsulation(alg_name) as kem:
+            length = kem.length_secret_key
+        for n in _wrong_lengths(length):
+            _assert_raises_value_error(
+                f"{alg_name}: KeyEncapsulation() with a {n}-byte secret key",
+                oqs.KeyEncapsulation,
+                alg_name,
+                secret_key=bytes(n),
+            )
+
+
 if __name__ == "__main__":
     try:
         import nose2

tests/test_sig.py

@@ -1,5 +1,6 @@
 import platform  # to learn the OS we're on
 import random
+from collections.abc import Callable
 from unittest import mock
 
 import oqs
@@ -200,17 +201,80 @@ def recording_free(ptr: object) -> None:
         real_free(ptr)
 
     for alg_name in oqs.get_enabled_sig_mechanisms():
+        with oqs.Signature(alg_name) as sig:
+            secret_key = b"\x01" * sig.length_secret_key
         freed.clear()
         with mock.patch.object(lib, "OQS_SIG_free", recording_free):
             # Passing a secret key exercises the cleanse path in free().
-            with oqs.Signature(alg_name, secret_key=b"\x01") as obj:
+            with oqs.Signature(alg_name, secret_key=secret_key) as obj:
                 pass
             obj.free()  # Explicit free after the context manager must be a no-op.
         if len(freed) != 1:
             msg = f"{alg_name}: OQS_SIG_free called {len(freed)} times"
             raise AssertionError(msg)
 
 
+def _wrong_lengths(length: int) -> set[int]:
+    # Empty, a bare 32-byte seed (e.g. ML-DSA rho with t1 = 0), off by one either way
+    return {0, min(32, length - 1), length - 1, length + 1}
+
+
+def _assert_raises_value_error(
+    label: str, call: Callable[..., object], *args: object, **kwargs: object
+) -> None:
+    try:
+        call(*args, **kwargs)
+    except ValueError:
+        pass
+    else:
+        msg = f"{label} did not raise ValueError"
+        raise AssertionError(msg)
+
+
+def test_wrong_length_public_key() -> tuple[None, str]:
+    for alg_name in oqs.get_enabled_sig_mechanisms():
+        if any(item in alg_name for item in disabled_sig_patterns):
+            continue
+        yield check_wrong_length_public_key, alg_name
+
+
+def check_wrong_length_public_key(alg_name: str) -> None:
+    with oqs.Signature(alg_name) as sig:
+        message = bytes(_message_len(alg_name))
+        signature = bytes(sig.length_signature)
+        for n in _wrong_lengths(sig.length_public_key):
+            public_key = bytes(n)
+            _assert_raises_value_error(
+                f"{alg_name}: verify() with a {n}-byte public key",
+                sig.verify,
+                message,
+                signature,
+                public_key,
+            )
+            if sig.sig_with_ctx_support:
+                _assert_raises_value_error(
+                    f"{alg_name}: verify_with_ctx_str() with a {n}-byte public key",
+                    sig.verify_with_ctx_str,
+                    message,
+                    signature,
+                    b"",
+                    public_key,
+                )
+
+
+def test_wrong_length_secret_key() -> None:
+    for alg_name in oqs.get_enabled_sig_mechanisms():
+        with oqs.Signature(alg_name) as sig:
+            length = sig.length_secret_key
+        for n in _wrong_lengths(length):
+            _assert_raises_value_error(
+                f"{alg_name}: Signature() with a {n}-byte secret key",
+                oqs.Signature,
+                alg_name,
+                secret_key=bytes(n),
+            )
+
+
 if __name__ == "__main__":
     try:
         import nose2

tests/test_stfl_sig.py

@@ -277,6 +277,40 @@ def test_use_after_free() -> None:
             _assert_raises_freed(f"{alg_name}.{name}", call)
 
 
+def _assert_raises_value_error(
+    label: str, call: Callable[..., object], *args: object, **kwargs: object
+) -> None:
+    try:
+        call(*args, **kwargs)
+    except ValueError:
+        pass
+    else:
+        msg = f"{label} did not raise ValueError"
+        raise AssertionError(msg)
+
+
+def test_wrong_length_public_key() -> tuple[None, str]:
+    for alg_name in oqs.get_enabled_stateful_sig_mechanisms():
+        if any(item in alg_name for item in disabled_sig_patterns):
+            continue
+        yield check_wrong_length_public_key, alg_name
+
+
+def check_wrong_length_public_key(alg_name: str) -> None:
+    # A short key must not reach OQS_SIG_STFL_verify, which reads length_public_key bytes.
+    with oqs.StatefulSignature(alg_name) as sig:
+        length = sig.length_public_key
+        signature = bytes(sig.length_signature)
+        for n in {0, min(32, length - 1), length - 1, length + 1}:
+            _assert_raises_value_error(
+                f"{alg_name}: verify() with a {n}-byte public key",
+                sig.verify,
+                b"message",
+                signature,
+                bytes(n),
+            )
+
+
 if __name__ == "__main__":
     try:
         import nose2