Update to EDHOC -12

edhoc/definitions.py

@@ -4,11 +4,26 @@
 
 import cbor2
 from cose.algorithms import AESCCM1664128, Sha256, EdDSA, AESCCM16128128, Es256, A128GCM, A256GCM, Sha384, Es384
-from cose.curves import X25519, Ed25519, P256, P384
+from cose.keys.curves import X25519, Ed25519, P256, P384
+from cose.keys import CoseKey
 
 from edhoc.exceptions import EdhocException
 
 
+def cborstream(items) -> bytes:
+    """Encode an iterable of items in CBOR into a stream"""
+    return b"".join(cbor2.dumps(i) for i in items)
+
+def compress_id_cred_x(id_cred_x):
+    if list(id_cred_x.keys()) == 4 and type(id_cred_x[4]) in (int, bytes):
+        return id_cred_x[4]
+    else:
+        return id_cred_x
+
+def bytewise_xor(a: bytes, b: bytes) -> bytes:
+    assert len(a) == len(b) # Python 3.10: zip(a, b, True) and remove this line
+    return bytes((_a ^ _b) for (_a, _b) in zip(a, b))
+
 class EdhocState(IntEnum):
     EDHOC_WAIT = 0
     MSG_1_SENT = 1
@@ -117,9 +132,9 @@ def check_identifiers(cls):
         return (
                 cls.aead.identifier,
                 cls.hash.identifier,
+                cls.edhoc_mac_length,
                 cls.dh_curve.identifier,
                 cls.sign_alg.identifier,
-                cls.sign_curve.identifier,
                 cls.app_aead.identifier,
                 cls.app_hash.identifier,
                 )
@@ -138,6 +153,9 @@ class CipherSuite0(CipherSuite):
     app_aead = AESCCM1664128
     app_hash = Sha256
 
+    edhoc_mac_length = 8
+assert CipherSuite0.check_identifiers() == (10, -16, 8, 4, -8, 10, -16)
+
 
 @CipherSuite.register_ciphersuite()
 class CipherSuite1(CipherSuite):
@@ -152,6 +170,9 @@ class CipherSuite1(CipherSuite):
     app_aead = AESCCM1664128
     app_hash = Sha256
 
+    edhoc_mac_length = 16
+assert CipherSuite1.check_identifiers() == (30, -16, 16, 4, -8, 10, -16)
+
 
 @CipherSuite.register_ciphersuite()
 class CipherSuite2(CipherSuite):
@@ -166,6 +187,9 @@ class CipherSuite2(CipherSuite):
     app_aead = AESCCM1664128
     app_hash = Sha256
 
+    edhoc_mac_length = 8
+assert CipherSuite2.check_identifiers() == (10, -16, 8, 1, -7, 10, -16)
+
 
 @CipherSuite.register_ciphersuite()
 class CipherSuite3(CipherSuite):
@@ -180,10 +204,15 @@ class CipherSuite3(CipherSuite):
     app_aead = AESCCM1664128
     app_hash = Sha256
 
+    edhoc_mac_length = 16
+assert CipherSuite3.check_identifiers() == (30, -16, 16, 1, -7, 10, -16)
+
+# ChaCha missing from pycose
+
 @CipherSuite.register_ciphersuite()
-class CipherSuite4(CipherSuite):
-    identifier = 4
-    fullname = "SUITE_4"
+class CipherSuite6(CipherSuite):
+    identifier = 6
+    fullname = "SUITE_6"
 
     aead = A128GCM
     hash = Sha256
@@ -192,12 +221,14 @@ class CipherSuite4(CipherSuite):
     sign_curve = P256
     app_aead = A128GCM
     app_hash = Sha256
-assert CipherSuite4.check_identifiers() == (1, -16, 4, -7, 1, 1, -16)
+
+    edhoc_mac_length = 16
+assert CipherSuite6.check_identifiers() == (1, -16, 16, 4, -7, 1, -16)
 
 @CipherSuite.register_ciphersuite()
-class CipherSuite5(CipherSuite):
-    identifier = 5
-    fullname = "SUITE_5"
+class CipherSuite24(CipherSuite):
+    identifier = 24
+    fullname = "SUITE_24"
 
     aead = A256GCM
     hash = Sha384
@@ -206,20 +237,44 @@ class CipherSuite5(CipherSuite):
     sign_curve = P384
     app_aead = A256GCM
     app_hash = Sha384
-assert CipherSuite5.check_identifiers() == (3, -43, 2, -35, 2, 3, -43)
+
+    edhoc_mac_length = 16
+assert CipherSuite24.check_identifiers() == (3, -43, 16, 2, -35, 3, -43)
 
 
 class EdhocKDFInfo(NamedTuple):
-    edhoc_aead_id: int
     transcript_hash: bytes
     label: str
+    context: bytes
     length: int
 
     def encode(self) -> bytes:
-        info = [self.edhoc_aead_id, self.transcript_hash, self.label, self.length]
-        info = cbor2.dumps(info)
-        return info
+        return cborstream(self)
+
+class CCS:
+    """A CWT Claims Set (containing an unencrypted COSE key in its CNF)"""
+    def __init__(self, encoded: bytes):
+        """Decode a CWT. Raises some ValueError if the set can not be
+        decoded."""
+        self.encoded = encoded
+        self._set_details_from(cbor2.loads(encoded))
 
+    @classmethod
+    def from_unencoded(cls, unencoded: dict):
+        """Load a CWT from a dictionary that all parties agree to encoded
+        canonically"""
+        # The optimization of going directly to _set_details is probably
+        # unwarranted.
+        return cls(cbor2.dumps(unencoded))
+
+    def _set_details_from(self, d: dict):
+        CWT_SUB = 2
+        CWT_CNF = 8
+        CNF_KEY = 1
+        self.sub = d.get(CWT_SUB, None)
+        cnf = d[CWT_CNF]
+        key = cnf[CNF_KEY]
+        self.key = CoseKey.from_dict(key)
 
 CS = TypeVar('CS', bound='CipherSuite')
 

edhoc/messages/base.py

@@ -1,6 +1,8 @@
 from abc import ABCMeta, abstractmethod
 from typing import Union
 from edhoc.definitions import Correlation
+import warnings
+from io import BytesIO
 
 import cbor2
 
@@ -17,30 +19,21 @@ def decode(cls, received: bytes) -> list:
         :return: a decode EDHOC message
         """
 
+        received = BytesIO(received)
+
         decoded = []
 
-        while len(received) > 0:
-            decoded += [cbor2.loads(received)]
-            received = received[received.startswith(cbor2.dumps(decoded[-1])) and len(cbor2.dumps(decoded[-1])):]
+        total_length = len(received.getvalue())
+        while received.tell() < total_length:
+            decoded.append(cbor2.load(received))
         return decoded
 
     @abstractmethod
-    def encode(self, corr: Correlation):
+    def encode(self):
         """ Encodes an EDHOC message as bytes, ready to be sent over reliable transport. """
 
         raise NotImplementedError
 
-    @staticmethod
-    def encode_bstr_id(conn_id: bytes) -> Union[int, bytes]:
-        if len(conn_id) == 1:
-            return int.from_bytes(conn_id, byteorder='big') - 24
-        else:
-            return conn_id
-
-    @staticmethod
-    def decode_bstr_id(conn_id: int) -> bytes:
-        return int(conn_id + 24).to_bytes(1, byteorder="big")
-
     @classmethod
     def _truncate(cls, payload: bytes):
         return f'{payload[:5]} ... ({len(payload)} bytes)'

edhoc/messages/message1.py

@@ -12,14 +12,13 @@
 
 
 class MessageOne(EdhocMessage):
-    METHOD_CORR = 0
-    CIPHERS = 1
-    G_X = 2
-    CONN_ID = 3
-    AAD1 = 4
+    ## Stored at decoding (or encoding, for sent messages) time to contain the
+    ## full byte string of the message, including any padding. (Unlike later
+    ## transcripts, the full message goes in).
+    encoded: bytes
 
     @classmethod
-    def decode(cls, received: bytes) -> 'MessageOne':
+    def decode(cls, received) -> 'MessageOne':
         """
         Tries to decode the bytes as an EDHOC MessageOne.
 
@@ -31,71 +30,58 @@ def decode(cls, received: bytes) -> 'MessageOne':
 
         decoded = super().decode(received)
 
-        method_corr = decoded[cls.METHOD_CORR]
+        (method, ciphers, g_x, c_i, *aad) = decoded
 
-        if isinstance(decoded[cls.CIPHERS], int):
-            selected_cipher = decoded[cls.CIPHERS]
-            supported_ciphers = [decoded[cls.CIPHERS]]
+        if isinstance(ciphers, int):
+            selected_cipher = ciphers
+            supported_ciphers = [ciphers]
         elif isinstance(decoded[cls.CIPHERS], list):
-            selected_cipher = decoded[cls.CIPHERS][0]
-            supported_ciphers = decoded[cls.CIPHERS][1:]
+            selected_cipher = ciphers[0]
+            supported_ciphers = ciphers[1:]
         else:
             raise EdhocInvalidMessage("Failed to decode bytes as MessageOne")
 
-        g_x = decoded[cls.G_X]
-
-        if decoded[cls.CONN_ID] != b'':
-            if isinstance(decoded[cls.CONN_ID], int):
-                conn_idi = EdhocMessage.decode_bstr_id(decoded[cls.CONN_ID])
-            else:
-                conn_idi = decoded[cls.CONN_ID]
-        else:
-            conn_idi = b''
-
         msg = cls(
-            method_corr=method_corr,
+            method=method,
             selected_cipher=CipherSuite.from_id(selected_cipher),
             cipher_suites=[CipherSuite.from_id(c) for c in supported_ciphers],
             g_x=g_x,
-            conn_idi=conn_idi)
+            c_i=c_i)
 
-        try:
-            msg.aad1 = decoded[cls.AAD1]
-        except IndexError:
-            pass
+        msg.encoded = received
+
+        if aad:
+            raise NotImplementedError("AAD changed")
 
         return msg
 
     def __init__(self,
-                 method_corr: int,
+                 method: int,
                  cipher_suites: List['CS'],
                  selected_cipher: Type['CS'],
                  g_x: bytes,
-                 conn_idi: Optional[bytes] = None,
+                 c_i: Optional[bytes] = None,
                  external_aad: bytes = b''):
 
         """
         Creates an EDHOC MessageOne object.
 
-        :param method_corr: Combination of the method parameter and correlation parameter (4 * method + correlation)
+        :param method: EDHOC method (indicating who signs / who does static derivation)
         :param cipher_suites: Supported cipher suites (ordered by decreasing preference).
         :param selected_cipher: The preferred cipher suite.
         :param g_x: The ephemeral public key of the Initiator.
-        :param conn_idi: A variable length connection identifier.
+        :param c_i: A variable length connection identifier.
         :param external_aad: Unprotected opaque auxiliary data (transferred together with EDHOC message 1).
         """
 
-        self.method_corr = method_corr
+        self.method = method
         self.cipher_suites = cipher_suites
         self.selected_cipher = selected_cipher
         self.g_x = g_x
-        self.conn_idi = conn_idi
+        self.c_i = c_i
         self.aad1 = external_aad
 
-        self.corr = self.method_corr % 4
-        self.method = (self.method_corr - self.corr) // 4
-
-    def encode(self, corr: Correlation) -> bytes:
+    def encode(self) -> bytes:
         """
         Encodes the first EDHOC message as a CBOR sequence.
 
@@ -109,16 +95,19 @@ def encode(self, corr: Correlation) -> bytes:
         else:
             raise ValueError('Cipher suite list must contain at least 1 item.')
 
-        msg = [self.method_corr, suites, self.g_x, self.encode_bstr_id(self.conn_idi)]
+        msg = [self.method, suites, self.g_x, self.c_i]
 
         if self.aad1 != b'':
+            raise NotImplementedError("AAD stuff changed")
             msg.append(cbor2.dumps(self.aad1))
 
-        return b"".join(cbor2.dumps(chunk) for chunk in msg)
+        # FIXME We might want to have "encode precisely once" semantics more generally
+        self.encoded = b"".join(cbor2.dumps(chunk) for chunk in msg)
+        return self.encoded
 
     def __repr__(self) -> str:
-        output = f'<MessageOne: [{self.method_corr}, {self.selected_cipher} | {self.cipher_suites}, ' \
-                 f'{EdhocMessage._truncate(self.g_x)}, {hexlify(self.conn_idi)}'
+        output = f'<MessageOne: [{self.method}, {self.selected_cipher} | {self.cipher_suites}, ' \
+                 f'{EdhocMessage._truncate(self.g_x)}, {hexlify(self.c_i)}'
         if self.aad1 != b'':
             output += f'{hexlify(self.aad1)}'
         output += ']>'