Merge remote-tracking branch 'agent/benma-agent/validate-py-antiklepto-signatures'

### .github/workflows/ci-common.yml
@@ -59,6 +59,9 @@ jobs:
           make -C py
           ./.ci/check-pep8
 
+      - name: Run Python unit tests
+        run: PYTHONPATH=py/bitbox02 python3 -m unittest discover -s py/bitbox02/tests
+
   lint-src:
     runs-on: ubuntu-22.04
     container:

### py/bitbox02/CHANGELOG.md
@@ -1,6 +1,7 @@
 # Changelog
 
 ## [Unreleased]
+- Validate ECDSA signatures and recovery IDs in Anti-Klepto and direct signing flows
 - Require firmware v9.4.0 or newer for BTC transaction signing and v9.5.0 or newer for
   BTC/ETH message and ETH transaction signing
 - `device_info()`: add the installed bootloader version to the returned device info

### py/bitbox02/bitbox02/bitbox02/bitbox02.py
@@ -18,7 +18,12 @@
     ERR_DUPLICATE_ENTRY,
 )
 
-from .secp256k1 import antiklepto_host_commit, antiklepto_verify
+from .secp256k1 import (
+    antiklepto_host_commit,
+    antiklepto_verify,
+    antiklepto_verify_recoverable,
+    validate_recoverable_ecdsa_signature,
+)
 
 try:
     from bitbox02.communication.generated import hww_pb2 as hww
@@ -693,7 +698,7 @@ def btc_sign_msg(
         signature = self._btc_msg_query(
             request, expected_response="sign_message"
         ).sign_message.signature
-        antiklepto_verify(host_nonce, signer_commitment, signature[:64])
+        antiklepto_verify_recoverable(host_nonce, signer_commitment, signature)
 
         if self.debug:
             print("Antiklepto nonce verification PASSED")
@@ -932,7 +937,7 @@ def handle_antiklepto(request: eth.ETHRequest) -> bytes:
             )
 
             signature = self._eth_msg_query(request, expected_response="sign").sign.signature
-            antiklepto_verify(host_nonce, signer_commitment, signature[:64])
+            antiklepto_verify_recoverable(host_nonce, signer_commitment, signature)
 
             if self.debug:
                 print("Antiklepto nonce verification PASSED")
@@ -1046,7 +1051,7 @@ def format_as_uncompressed(sig: bytes) -> bytes:
         )
 
         signature = self._eth_msg_query(request, expected_response="sign").sign.signature
-        antiklepto_verify(host_nonce, signer_commitment, signature[:64])
+        antiklepto_verify_recoverable(host_nonce, signer_commitment, signature)
 
         if self.debug:
             print("Antiklepto nonce verification PASSED")
@@ -1249,13 +1254,14 @@ def get_value(
             )
 
             signature = self._eth_msg_query(request, expected_response="sign").sign.signature
-            antiklepto_verify(host_nonce, signer_commitment, signature[:64])
+            antiklepto_verify_recoverable(host_nonce, signer_commitment, signature)
 
             if self.debug:
                 print("Antiklepto nonce verification PASSED")
         else:
             assert response.WhichOneof("response") == "sign"
             signature = response.sign.signature
+            validate_recoverable_ecdsa_signature(signature)
 
         return format_as_uncompressed(signature)
 

### py/bitbox02/bitbox02/bitbox02/secp256k1.py
@@ -10,6 +10,10 @@ class ECDSANonceException(Exception):
     pass
 
 
+class ECDSASignatureException(Exception):
+    pass
+
+
 def tagged_sha256(tag: bytes, msg: bytes) -> bytes:
     tag_hash = hashlib.sha256(tag).digest()
     return hashlib.sha256(tag_hash + tag_hash + msg).digest()
@@ -19,13 +23,56 @@ def antiklepto_host_commit(host_nonce: bytes) -> bytes:
     return tagged_sha256(b"s2c/ecdsa/data", host_nonce)
 
 
+def validate_compact_ecdsa_signature(signature: bytes) -> None:
+    """Validates a compact ECDSA signature encoded as r || s.
+
+    Both scalars must be nonzero and in range, and s must use its low-S encoding.
+    """
+    if len(signature) != 64:
+        raise ECDSASignatureException("Compact ECDSA signature must be 64 bytes")
+
+    order = ecdsa.curves.SECP256k1.order
+    sig_r = int.from_bytes(signature[:32], "big")
+    sig_s = int.from_bytes(signature[32:], "big")
+    if not 0 < sig_r < order or not 0 < sig_s < order:
+        raise ECDSASignatureException("Invalid compact ECDSA signature scalar")
+    if sig_s > order // 2:
+        raise ECDSASignatureException("ECDSA signature has high S")
+
+
+def validate_recoverable_ecdsa_signature(signature: bytes) -> None:
+    """Validates a recoverable ECDSA signature encoded as r || s || recovery ID.
+
+    The compact signature must be valid and the recovery ID must be in the range 0..3.
+    """
+    if len(signature) != 65:
+        raise ECDSASignatureException("Recoverable ECDSA signature must be 65 bytes")
+    validate_compact_ecdsa_signature(signature[:64])
+    if signature[64] > 3:
+        raise ECDSASignatureException("Invalid ECDSA recovery ID")
+
+
 def antiklepto_verify(host_nonce: bytes, signer_commitment: bytes, signature: bytes) -> None:
     """
     Verifies that hostNonce was used to tweak the nonce during signature
     generation according to k' = k + H(signerCommitment, hostNonce) by checking that
     k'*G = signerCommitment + H(signerCommitment, hostNonce)*G.
-    Throws ECDSANonceException if the verification fails.
+    Throws ECDSASignatureException if the signature is invalid and ECDSANonceException if
+    the nonce verification fails.
     """
+    validate_compact_ecdsa_signature(signature)
+    _antiklepto_verify_nonce(host_nonce, signer_commitment, signature)
+
+
+def antiklepto_verify_recoverable(
+    host_nonce: bytes, signer_commitment: bytes, signature: bytes
+) -> None:
+    """Validates a recoverable ECDSA signature and verifies its Anti-Klepto nonce."""
+    validate_recoverable_ecdsa_signature(signature)
+    _antiklepto_verify_nonce(host_nonce, signer_commitment, signature[:64])
+
+
+def _antiklepto_verify_nonce(host_nonce: bytes, signer_commitment: bytes, signature: bytes) -> None:
     assert len(host_nonce) == 32
     assert len(signer_commitment) == 33, "expected compressed pubkey"
     assert len(signature) == 64

### py/bitbox02/tests/test_secp256k1.py
@@ -0,0 +1,89 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for secp256k1 signature validation and Anti-Klepto verification."""
+
+import unittest
+
+from bitbox02.bitbox02.secp256k1 import (
+    ECDSASignatureException,
+    antiklepto_verify,
+    antiklepto_verify_recoverable,
+    validate_compact_ecdsa_signature,
+    validate_recoverable_ecdsa_signature,
+)
+
+
+HOST_NONCE = bytes.fromhex("8b4c26aa2695a34bdbc34235f6c91be14b93037a063b13f7c814101359561092")
+SIGNER_COMMITMENT = bytes.fromhex(
+    "0236ff92fe02c08d0d04851e0ce1516104085215f05a178307de60ea53e207f971"
+)
+VALID_SIGNATURE = bytes.fromhex(
+    "7fd66b48ffea2fe048869880bbb3a1819e262af14980e8885df1e5765750cb8f"
+    "47e01eca356377870356d54853573a955076228e5044cd3dd3a049abe70d5585"
+)
+HIGH_S_SIGNATURE = bytes.fromhex(
+    "7fd66b48ffea2fe048869880bbb3a1819e262af14980e8885df1e5765750cb8f"
+    "b81fe135ca9c8878fca92ab7aca8c5696a38ba585f03d2fdec3214e0e928ebbc"
+)
+CURVE_ORDER = bytes.fromhex("fffffffffffffffffffffffffffffffebaaedce6af48a03bbfd25e8cd0364141")
+
+
+class TestECDSASignatureValidation(unittest.TestCase):
+    """Tests for compact and recoverable ECDSA signature validation."""
+
+    def test_validate_compact_ecdsa_signature(self) -> None:
+        """Accept low-S and reject malformed compact signatures."""
+        validate_compact_ecdsa_signature(VALID_SIGNATURE)
+
+        with self.assertRaisesRegex(ECDSASignatureException, "high S"):
+            validate_compact_ecdsa_signature(HIGH_S_SIGNATURE)
+
+        invalid_signatures = [
+            VALID_SIGNATURE[:-1],
+            VALID_SIGNATURE + b"\x00",
+        ]
+        for offset in (0, 32):
+            invalid_signatures.append(
+                VALID_SIGNATURE[:offset] + bytes(32) + VALID_SIGNATURE[offset + 32 :]
+            )
+            invalid_signatures.append(
+                VALID_SIGNATURE[:offset] + CURVE_ORDER + VALID_SIGNATURE[offset + 32 :]
+            )
+
+        for signature in invalid_signatures:
+            with self.subTest(signature=signature.hex()):
+                with self.assertRaises(ECDSASignatureException):
+                    validate_compact_ecdsa_signature(signature)
+
+    def test_validate_recoverable_ecdsa_signature(self) -> None:
+        """Accept valid and reject malformed recoverable signatures."""
+        validate_recoverable_ecdsa_signature(VALID_SIGNATURE + b"\x00")
+        validate_recoverable_ecdsa_signature(VALID_SIGNATURE + b"\x03")
+
+        for signature in (
+            VALID_SIGNATURE,
+            VALID_SIGNATURE + b"\x00\x00",
+            VALID_SIGNATURE + b"\x04",
+        ):
+            with self.subTest(signature=signature.hex()):
+                with self.assertRaises(ECDSASignatureException):
+                    validate_recoverable_ecdsa_signature(signature)
+
+    def test_antiklepto_verify_rejects_high_s(self) -> None:
+        """Reject a high-S signature even when its nonce contribution matches."""
+        antiklepto_verify(HOST_NONCE, SIGNER_COMMITMENT, VALID_SIGNATURE)
+
+        with self.assertRaisesRegex(ECDSASignatureException, "high S"):
+            antiklepto_verify(HOST_NONCE, SIGNER_COMMITMENT, HIGH_S_SIGNATURE)
+
+    def test_antiklepto_verify_recoverable(self) -> None:
+        """Validate recovery IDs while checking the nonce contribution."""
+        antiklepto_verify_recoverable(HOST_NONCE, SIGNER_COMMITMENT, VALID_SIGNATURE + b"\x00")
+        antiklepto_verify_recoverable(HOST_NONCE, SIGNER_COMMITMENT, VALID_SIGNATURE + b"\x03")
+
+        with self.assertRaisesRegex(ECDSASignatureException, "recovery ID"):
+            antiklepto_verify_recoverable(HOST_NONCE, SIGNER_COMMITMENT, VALID_SIGNATURE + b"\x04")
+
+
+if __name__ == "__main__":
+    unittest.main()