Certificate V2 generator

- Added individual certificate v2 element types with serialization logic: SgxQuote, SgxAttestationKey and X509
- Added CStruct abstraction to simplify representing C structs
- Added struct definitions as CStruct subtypes based on OpenEnclave's endorsement type definitions
- Updated SGX attestation gathering logic to parse the envelope and generate the actual certificate elements
- Added and updated unit tests
- Ignoring long line linting in envelope unit tests

middleware/admin/certificate.py

@@ -21,4 +21,6 @@
 # SOFTWARE.
 
 from .certificate_v1 import HSMCertificate, HSMCertificateElement
-from .certificate_v2 import HSMCertificateV2, HSMCertificateV2Element
+from .certificate_v2 import HSMCertificateV2, HSMCertificateV2ElementSGXQuote, \
+                            HSMCertificateV2ElementSGXAttestationKey, \
+                            HSMCertificateV2ElementX509

middleware/admin/certificate_v2.py

@@ -24,16 +24,62 @@
 
 
 class HSMCertificateV2Element:
-    # TODO: actual logic and subclasses
-    def __init__(self, element_map):
-        self.element_map = element_map
+    pass
 
-    # Stub
-    def name(self):
-        return "attestation"
+
+class HSMCertificateV2ElementSGXQuote(HSMCertificateV2Element):
+    def __init__(self, name, message, custom_data, signature, signed_by):
+        self.name = name
+        self.message = message
+        self.custom_data = custom_data
+        self.signature = signature
+        self.signed_by = signed_by
+
+    def to_dict(self):
+        return {
+            "name": self.name,
+            "type": "sgx_quote",
+            "message": self.message.hex(),
+            "custom_data": self.custom_data.hex(),
+            "signature": self.signature.hex(),
+            "signed_by": self.signed_by,
+        }
+
+
+class HSMCertificateV2ElementSGXAttestationKey(HSMCertificateV2Element):
+    def __init__(self, name, message, key, auth_data, signature, signed_by):
+        self.name = name
+        self.message = message
+        self.key = key
+        self.auth_data = auth_data
+        self.signature = signature
+        self.signed_by = signed_by
+
+    def to_dict(self):
+        return {
+            "name": self.name,
+            "type": "sgx_attestation_key",
+            "message": self.message.hex(),
+            "key": self.key.hex(),
+            "auth_data": self.auth_data.hex(),
+            "signature": self.signature.hex(),
+            "signed_by": self.signed_by,
+        }
+
+
+class HSMCertificateV2ElementX509(HSMCertificateV2Element):
+    def __init__(self, name, message, signed_by):
+        self.name = name
+        self.message = message
+        self.signed_by = signed_by
 
     def to_dict(self):
-        return self.element_map
+        return {
+            "name": self.name,
+            "type": "x509_pem",
+            "message": self.message.decode('ASCII'),
+            "signed_by": self.signed_by,
+        }
 
 
 class HSMCertificateV2(HSMCertificate):

middleware/admin/sgx_attestation.py

@@ -20,9 +20,13 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
+import ecdsa
 from .misc import info, head, get_hsm, AdminError, get_ud_value_for_attestation
 from .unlock import do_unlock
-from .certificate import HSMCertificateV2, HSMCertificateV2Element
+from sgx.envelope import SgxEnvelope
+from .certificate import HSMCertificateV2, HSMCertificateV2ElementSGXQuote, \
+                         HSMCertificateV2ElementSGXAttestationKey, \
+                         HSMCertificateV2ElementX509
 
 
 def do_attestation(options):
@@ -58,14 +62,72 @@ def do_attestation(options):
 
     hsm.disconnect()
 
+    # Parse envelope
+    info("Parsing the powHSM attestation envelope...")
+    try:
+        envelope = SgxEnvelope(
+            bytes.fromhex(powhsm_attestation["envelope"]),
+            bytes.fromhex(powhsm_attestation["message"]))
+    except Exception as e:
+        raise AdminError(f"SGX envelope parse error: {str(e)}")
+
+    # Conversions
+    quote_signature = ecdsa.util.sigdecode_string(
+        envelope.quote_auth_data.signature.r +
+        envelope.quote_auth_data.signature.s,
+        ecdsa.NIST256p.order)
+    quote_signature = ecdsa.util.sigencode_der(
+        quote_signature[0],
+        quote_signature[1],
+        ecdsa.NIST256p.order)
+    att_key = ecdsa.VerifyingKey.from_string(
+        envelope.quote_auth_data.attestation_key.x +
+        envelope.quote_auth_data.attestation_key.y,
+        ecdsa.NIST256p)
+    qe_rb_signature = ecdsa.util.sigdecode_string(
+        envelope.quote_auth_data.qe_report_body_signature.r +
+        envelope.quote_auth_data.qe_report_body_signature.s,
+        ecdsa.NIST256p.order)
+    qe_rb_signature = ecdsa.util.sigencode_der(
+        qe_rb_signature[0],
+        qe_rb_signature[1],
+        ecdsa.NIST256p.order)
+
     # Generate and save the attestation certificate
     info("Generating the attestation certificate... ", options.verbose)
-
     att_cert = HSMCertificateV2()
-    # TODO:
-    # 1. Parse envelope
-    # 2. Add actual elements of the certificate
-    att_cert.add_element(HSMCertificateV2Element(powhsm_attestation))
+
+    att_cert.add_element(
+        HSMCertificateV2ElementSGXQuote(
+            name="quote",
+            message=envelope.quote.get_raw_data(),
+            custom_data=envelope.custom_message,
+            signature=quote_signature,
+            signed_by="attestation",
+        ))
+    att_cert.add_element(
+        HSMCertificateV2ElementSGXAttestationKey(
+            name="attestation",
+            message=envelope.quote_auth_data.qe_report_body.get_raw_data(),
+            key=att_key.to_string("uncompressed"),
+            auth_data=envelope.qe_auth_data.data,
+            signature=qe_rb_signature,
+            signed_by="quoting_enclave",
+        ))
+    att_cert.add_element(
+        HSMCertificateV2ElementX509(
+            name="quoting_enclave",
+            message=envelope.qe_cert_data.certs[0],
+            signed_by="platform_ca",
+        ))
+    att_cert.add_element(
+        HSMCertificateV2ElementX509(
+            name="platform_ca",
+            message=envelope.qe_cert_data.certs[1],
+            signed_by="sgx_root",
+        ))
+
+    att_cert.add_target("quote")
     att_cert.save_to_jsonfile(options.output_file_path)
 
     info(f"Attestation certificate saved to {options.output_file_path}")

middleware/comm/cstruct.py

@@ -0,0 +1,163 @@
+# The MIT License (MIT)
+#
+# Copyright (c) 2021 RSK Labs Ltd
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy of
+# this software and associated documentation files (the "Software"), to deal in
+# the Software without restriction, including without limitation the rights to
+# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
+# of the Software, and to permit persons to whom the Software is furnished to do
+# so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+
+import struct
+import re
+
+
+class CStruct:
+    MAP = {
+        "uint8_t": ["B", "s"],
+        "uint16_t": "H",
+        "uint32_t": "I",
+        "uint64_t": "Q",
+    }
+    SPEC = None
+    TYPENAME = None
+
+    @classmethod
+    def _spec(cls, little=True):
+        if cls.SPEC is None or little not in cls.SPEC:
+            fmt = "<" if little else ">"
+            atrmap = {}
+            names = []
+            types = []
+            index = 0
+            typename = None
+            for line in cls.__doc__.split("\n")[1:]:
+                line = re.sub(r"\s+", " ", line.strip())
+                if line == "":
+                    continue
+                if typename is None:
+                    typename = line
+                    continue
+                tspec = line.split(" ")
+
+                length = "" if len(tspec) < 3 else str(int(tspec[2].strip(), 10))
+
+                typ = tspec[0].strip()
+                actual_type = None
+                derived_type = None
+                if typ not in cls.MAP.keys():
+                    for kls in cls.__base__.__subclasses__():
+                        if cls != kls:
+                            if typ == kls._typename():
+                                actual_type = kls
+                                derived_type = kls
+                    if derived_type is None:
+                        raise ValueError(f"Invalid type: {typ}")
+                else:
+                    actual_type = cls.MAP[typ]
+
+                if length != "" and not isinstance(actual_type, list):
+                    raise ValueError(f"Invalid type spec: {line}")
+
+                name = tspec[1].strip()
+
+                if isinstance(actual_type, list):
+                    actual_type = actual_type[0] if length == "" else actual_type[1]
+                elif not isinstance(actual_type, str) and \
+                        issubclass(actual_type, cls.__base__):
+                    actual_type = str(actual_type.get_bytelength(little)) + "s"
+
+                fmt += length + actual_type
+                names.append(name)
+                types.append(derived_type)
+                atrmap[name] = index
+                index += 1
+            if cls.SPEC is None:
+                cls.SPEC = {}
+            cls.SPEC[little] = (struct.Struct(fmt), atrmap, names, types, typename)
+
+        return cls.SPEC[little]
+
+    @classmethod
+    def _struct(cls, little=True):
+        return cls._spec(little)[0]
+
+    @classmethod
+    def _atrmap(cls, little=True):
+        return cls._spec(little)[1]
+
+    @classmethod
+    def _names(cls, little=True):
+        return cls._spec(little)[2]
+
+    @classmethod
+    def _types(cls, little=True):
+        return cls._spec(little)[3]
+
+    @classmethod
+    def _typename(cls):
+        if cls.TYPENAME is None:
+            for line in cls.__doc__.split("\n"):
+                line = re.sub(r"\s+", " ", line.strip())
+                if line == "":
+                    continue
+                cls.TYPENAME = line
+                break
+
+        return cls.TYPENAME
+
+    @classmethod
+    def get_bytelength(cls, little=True):
+        return cls._struct(little).size
+
+    def __init__(self, value, offset=0, little=True):
+        self._offset = offset
+        self._little = little
+        self._raw_value = value
+
+        try:
+            self._parsed = list(self._struct(little).unpack_from(value, offset))
+        except Exception as e:
+            raise ValueError(f"While parsing: {e}")
+
+        for index, derived_type in enumerate(self._types(little)):
+            if derived_type is not None:
+                self._parsed[index] = derived_type(self._parsed[index], little=little)
+
+    def _value(self, name):
+        amap = self._atrmap(self._little)
+        if name in amap:
+            return self._parsed[amap[name]]
+        raise NameError(f"Property {name} does not exist")
+
+    def __getattr__(self, name):
+        return self._value(name)
+
+    def get_raw_data(self):
+        return self._raw_value[
+            self._offset:self._offset+self.get_bytelength(self._little)]
+
+    def to_dict(self):
+        result = {}
+        for name in self._names(self._little):
+            value = self._value(name)
+            if isinstance(value, bytes):
+                value = value.hex()
+            result[name] = value.to_dict() if isinstance(value, CStruct) else value
+        return result
+
+    def __repr__(self):
+        return f"<{self.__class__.__name__}: {self.to_dict()}>"