smdpp: proper checks, proper check order, test mode

Change-Id: Ie4db65594b2eaf6c95ffc5e73ff9ae61c4d9d3a3

.gitignore

@@ -7,9 +7,10 @@
 /.local
 /build
 /pySim.egg-info
-/smdpp-data/sm-dp-sessions
+/smdpp-data/sm-dp-sessions*
 dist
 tags
 smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_NIST.pem
+smdpp-data/certs/DPtls/CERT_S_SM_DP_TLS_BRP.pem
 smdpp-data/generated
 smdpp-data/certs/dhparam2048.pem

pySim/esim/bsp.py

@@ -73,7 +73,7 @@ class BspAlgoCrypt(BspAlgo, abc.ABC):
         block_nr = self.block_nr
         ciphertext = self._encrypt(padded_data)
         logger.debug("encrypt(block_nr=%u, s_enc=%s, plaintext=%s, padded=%s) -> %s",
-                     block_nr, b2h(self.s_enc), b2h(data), b2h(padded_data), b2h(ciphertext))
+                     block_nr, b2h(self.s_enc)[:20], b2h(data)[:20], b2h(padded_data)[:20], b2h(ciphertext)[:20])
         return ciphertext
 
     def decrypt(self, data:bytes) -> bytes:
@@ -149,10 +149,20 @@ class BspAlgoMac(BspAlgo, abc.ABC):
         temp_data = self.mac_chain + tag_and_length + data
         old_mcv = self.mac_chain
         c_mac = self._auth(temp_data)
+
+        # DEBUG: Show MAC computation details
+        logger.debug(f"MAC_DEBUG: tag=0x{tag:02x}, lcc={lcc}")
+        logger.debug(f"MAC_DEBUG: tag_and_length: {tag_and_length.hex()}")
+        logger.debug(f"MAC_DEBUG: mac_chain[:20]: {old_mcv[:20].hex()}")
+        logger.debug(f"MAC_DEBUG: temp_data[:20]: {temp_data[:20].hex()}")
+        logger.debug(f"MAC_DEBUG: c_mac: {c_mac.hex()}")
+
         # The output data is computed by concatenating the following data: the tag, the final length, the result of step 2 and the C-MAC value.
         ret = tag_and_length + data + c_mac
+        logger.debug(f"MAC_DEBUG: final_output[:20]: {ret[:20].hex()}")
+
         logger.debug("auth(tag=0x%x, mcv=%s, s_mac=%s, plaintext=%s, temp=%s) -> %s",
-                     tag, b2h(old_mcv), b2h(self.s_mac), b2h(data), b2h(temp_data), b2h(ret))
+                     tag, b2h(old_mcv)[:20], b2h(self.s_mac)[:20], b2h(data)[:20], b2h(temp_data)[:20], b2h(ret)[:20])
         return ret
 
     def verify(self, ciphertext: bytes) -> bool:
@@ -204,6 +214,11 @@ def bsp_key_derivation(shared_secret: bytes, key_type: int, key_length: int, hos
     s_enc = out[l:2*l]
     s_mac = out[l*2:3*l]
 
+    logger.debug(f"BSP_KDF_DEBUG: kdf_out = {b2h(out)}")
+    logger.debug(f"BSP_KDF_DEBUG: initial_mcv = {b2h(initial_mac_chaining_value)}")
+    logger.debug(f"BSP_KDF_DEBUG: s_enc = {b2h(s_enc)}")
+    logger.debug(f"BSP_KDF_DEBUG: s_mac = {b2h(s_mac)}")
+
     return s_enc, s_mac, initial_mac_chaining_value
 
 
@@ -231,9 +246,21 @@ class BspInstance:
         """Encrypt + MAC a single plaintext TLV. Returns the protected ciphertext."""
         assert tag <= 255
         assert len(plaintext) <= self.max_payload_size
-        logger.debug("encrypt_and_mac_one(tag=0x%x, plaintext=%s)", tag, b2h(plaintext))
+
+        # DEBUG: Show what we're processing
+        logger.debug(f"BSP_DEBUG: encrypt_and_mac_one(tag=0x{tag:02x}, plaintext_len={len(plaintext)})")
+        logger.debug(f"BSP_DEBUG: plaintext[:20]: {plaintext[:20].hex()}")
+        logger.debug(f"BSP_DEBUG: s_enc[:20]: {self.c_algo.s_enc[:20].hex()}")
+        logger.debug(f"BSP_DEBUG: s_mac[:20]: {self.m_algo.s_mac[:20].hex()}")
+
+        logger.debug("encrypt_and_mac_one(tag=0x%x, plaintext=%s)", tag, b2h(plaintext)[:20])
         ciphered = self.c_algo.encrypt(plaintext)
+        logger.debug(f"BSP_DEBUG: ciphered[:20]: {ciphered[:20].hex()}")
+
         maced = self.m_algo.auth(tag, ciphered)
+        logger.debug(f"BSP_DEBUG: final_result[:20]: {maced[:20].hex()}")
+        logger.debug(f"BSP_DEBUG: final_result_len: {len(maced)}")
+
         return maced
 
     def encrypt_and_mac(self, tag: int, plaintext:bytes) -> List[bytes]:

pySim/esim/es8p.py

@@ -25,6 +25,9 @@ import pySim.esim.rsp as rsp
 from pySim.esim.bsp import BspInstance
 from pySim.esim import PMO
 
+import logging
+logger = logging.getLogger(__name__)
+
 # Given that GSMA RSP uses ASN.1 in a very weird way, we actually cannot encode the full data type before
 # signing, but we have to build parts of it separately first, then sign that, so we can put the signature
 # into the same sequence as the signed data.  We use the existing pySim TLV code for this.
@@ -196,8 +199,12 @@ class BoundProfilePackage(ProfilePackage):
         # 'initialiseSecureChannelRequest'
         bpp_seq = rsp.asn1.encode('InitialiseSecureChannelRequest', iscr)
         # firstSequenceOf87
+        logger.debug(f"BPP_ENCODE_DEBUG: Encrypting ConfigureISDP with BSP keys")
+        logger.debug(f"BPP_ENCODE_DEBUG: BSP S-ENC: {bsp.c_algo.s_enc.hex()}")
+        logger.debug(f"BPP_ENCODE_DEBUG: BSP S-MAC: {bsp.m_algo.s_mac.hex()}")
         bpp_seq += encode_seq(0xa0, bsp.encrypt_and_mac(0x87, conf_idsp_bin))
         # sequenceOF88
+        logger.debug(f"BPP_ENCODE_DEBUG: MAC-only StoreMetadata with BSP keys")
         bpp_seq += encode_seq(0xa1, bsp.mac_only(0x88, smr_bin))
 
         if self.ppp: # we have to use session keys

pySim/esim/rsp.py

@@ -94,9 +94,57 @@ class RspSessionState:
         self.__dict__.update(state)
 
 
-class RspSessionStore(shelve.DbfilenameShelf):
-    """A derived class as wrapper around the database-backed non-volatile storage 'shelve', in case we might
-    need to extend it in the future. We use it to store RspSessionState objects indexed by transactionId."""
+class RspSessionStore:
+    """A wrapper around the database-backed storage 'shelve' for storing RspSessionState objects.
+    Can be configured to use either file-based storage or in-memory storage.
+    We use it to store RspSessionState objects indexed by transactionId."""
+
+    def __init__(self, filename: Optional[str] = None, in_memory: bool = False):
+        self._in_memory = in_memory
+
+        if in_memory:
+            self._shelf = shelve.Shelf(dict())
+        else:
+            if filename is None:
+                raise ValueError("filename is required for file-based session store")
+            self._shelf = shelve.open(filename)
+
+    # dunder magic
+    def __getitem__(self, key):
+        return self._shelf[key]
+
+    def __setitem__(self, key, value):
+        self._shelf[key] = value
+
+    def __delitem__(self, key):
+        del self._shelf[key]
+
+    def __contains__(self, key):
+        return key in self._shelf
+
+    def __iter__(self):
+        return iter(self._shelf)
+
+    def __len__(self):
+        return len(self._shelf)
+
+    # everything else
+    def __getattr__(self, name):
+        """Delegate attribute access to the underlying shelf object."""
+        return getattr(self._shelf, name)
+
+    def close(self):
+        """Close the session store."""
+        if hasattr(self._shelf, 'close'):
+            self._shelf.close()
+        if self._in_memory:
+            # For in-memory store, clear the reference
+            self._shelf = None
+
+    def sync(self):
+        """Synchronize the cache with the underlying storage."""
+        if hasattr(self._shelf, 'sync'):
+            self._shelf.sync()
 
 def extract_euiccSigned1(authenticateServerResponse: bytes) -> bytes:
     """Extract the raw, DER-encoded binary euiccSigned1 field from the given AuthenticateServerResponse. This

pySim/esim/x509_cert.py

@@ -25,6 +25,7 @@ from cryptography.hazmat.primitives.serialization import load_pem_private_key, E
 from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
 
 from pySim.utils import b2h
+from . import x509_err
 
 def check_signed(signed: x509.Certificate, signer: x509.Certificate) -> bool:
     """Verify if 'signed' certificate was signed using 'signer'."""
@@ -64,9 +65,6 @@ class oid:
     id_rspRole_ds_tls_v2 = x509.ObjectIdentifier(ID_RSP_ROLE + '.6')
     id_rspRole_ds_auth_v2 = x509.ObjectIdentifier(ID_RSP_ROLE + '.7')
 
-class VerifyError(Exception):
-    """An error during certificate verification,"""
-
 class CertificateSet:
     """A set of certificates consisting of a trusted [self-signed] CA root certificate,
     and an optional number of intermediate certificates.  Can be used to verify the certificate chain
@@ -135,7 +133,7 @@ class CertificateSet:
             # we cannot check if there's no CRL
             return
         if self.crl.get_revoked_certificate_by_serial_number(cert.serial_nr):
-            raise VerifyError('Certificate is present in CRL, verification failed')
+            raise x509_err.CertificateRevoked()
 
     def verify_cert_chain(self, cert: x509.Certificate, max_depth: int = 100):
         """Verify if a given certificate's signature chain can be traced back to the root CA of this
@@ -149,14 +147,14 @@ class CertificateSet:
                 check_signed(c, self.root_cert)
                 return
             parent_cert = self.intermediate_certs.get(aki, None)
-            if not aki:
-                raise VerifyError('Could not find intermediate certificate for AuthKeyId %s' % b2h(aki))
+            if not parent_cert:
+                raise x509_err.MissingIntermediateCert(b2h(aki))
             check_signed(c, parent_cert)
             # if we reach here, we passed (no exception raised)
             c = parent_cert
             depth += 1
             if depth > max_depth:
-                raise VerifyError('Maximum depth %u exceeded while verifying certificate chain' % max_depth)
+                raise x509_err.MaxDepthExceeded(max_depth, depth)
 
 
 def ecdsa_dss_to_tr03111(sig: bytes) -> bytes:

pySim/esim/x509_err.py

@@ -0,0 +1,58 @@
+"""X.509 certificate verification exceptions for GSMA eSIM."""
+
+class VerifyError(Exception):
+    """Base class for certificate verification errors."""
+    pass
+
+
+class MissingIntermediateCert(VerifyError):
+    """Raised when an intermediate certificate in the chain cannot be found."""
+    def __init__(self, auth_key_id: str):
+        self.auth_key_id = auth_key_id
+        super().__init__(f'Could not find intermediate certificate for AuthKeyId {auth_key_id}')
+
+
+class CertificateRevoked(VerifyError):
+    """Raised when a certificate is found in the CRL."""
+    def __init__(self, cert_serial: str = None):
+        self.cert_serial = cert_serial
+        msg = 'Certificate is present in CRL, verification failed'
+        if cert_serial:
+            msg += f' (serial: {cert_serial})'
+        super().__init__(msg)
+
+
+class MaxDepthExceeded(VerifyError):
+    """Raised when certificate chain depth exceeds the maximum allowed."""
+    def __init__(self, max_depth: int, actual_depth: int):
+        self.max_depth = max_depth
+        self.actual_depth = actual_depth
+        super().__init__(f'Maximum depth {max_depth} exceeded while verifying certificate chain (actual: {actual_depth})')
+
+
+class SignatureVerification(VerifyError):
+    """Raised when certificate signature verification fails."""
+    def __init__(self, cert_subject: str = None, signer_subject: str = None):
+        self.cert_subject = cert_subject
+        self.signer_subject = signer_subject
+        msg = 'Certificate signature verification failed'
+        if cert_subject and signer_subject:
+            msg += f': {cert_subject} not signed by {signer_subject}'
+        super().__init__(msg)
+
+
+class InvalidCertificate(VerifyError):
+    """Raised when a certificate is invalid (missing required fields, wrong type, etc)."""
+    def __init__(self, reason: str):
+        self.reason = reason
+        super().__init__(f'Invalid certificate: {reason}')
+
+
+class CertificateExpired(VerifyError):
+    """Raised when a certificate has expired."""
+    def __init__(self, cert_subject: str = None):
+        self.cert_subject = cert_subject
+        msg = 'Certificate has expired'
+        if cert_subject:
+            msg += f': {cert_subject}'
+        super().__init__(msg)