cnap: add event log replay and verify RTMR value with measurement (#185)

Signed-off-by: Longyin Hu <[email protected]>

Author: Hulongyin

cnap/core/eventlog.py

@@ -0,0 +1,112 @@
+"""A EventLog module.
+
+This module provides functions related to event log processing. The event log can be fetched by
+Confidential Cloud-Native Primitives (CCNP), and is defined according to several tcg supported event
+log formats defined in TCG_PCClient Spec, Canonical Eventlog Spec, etc.
+
+CCNP: https://github.com/intel/confidential-cloud-native-primitives
+TCG_PCClient Spec:
+  https://trustedcomputinggroup.org/wp-content/uploads/TCG_PCClientSpecPlat_TPM_2p0_1p04_pub.pdf
+Canonical Eventlog Spec:
+  https://trustedcomputinggroup.org/wp-content/uploads/TCG_IWG_CEL_v1_r0p41_pub.pdf
+
+Functions:
+    replay_event_log: Replay event logs by IMR index.
+    verify_event_log: Verify event log by comparing IMR value from CCNP fetching and replayed from
+      event log.
+"""
+import logging
+import base64
+from hashlib import sha1, sha256, sha384, sha512
+
+from ccnp.eventlog.eventlog_sdk import CCEventLogEntry, CCAlgorithms
+from ccnp import Measurement, MeasurementType
+
+LOG = logging.getLogger(__name__)
+
+IMR_VERIFY_COUNT = 3
+
+def replay_event_log(event_logs: list[CCEventLogEntry]) -> dict:
+    """Replay event logs by Integrated Measurement Register (IMR) index.
+
+    Args:
+        event_logs (list[CCEventLogEntry]): Event logs fetched by CCNP.
+
+    Returns:
+        dict: A dictionary containing the replay result displayed by IMR index and hash algorithm. 
+        Layer 1 key of the dict is the IMR index, the value is another dict which using the hash 
+        algorithm as the key and the replayed measurement as value.
+        Sample value:
+            { 0: { 12: <measurement_replayed>}}
+    """
+    measurement_dict = {}
+    for event_log in event_logs:
+        # pylint: disable-next=consider-iterating-dictionary
+        if event_log.reg_idx not in measurement_dict.keys():
+            measurement_dict[event_log.reg_idx] = {}
+
+        alg_id = event_log.alg_id.algo_id
+        # Check algorithm type and prepare for replay
+        if alg_id == CCAlgorithms.ALG_SHA1:
+            algo = sha1()
+        elif alg_id == CCAlgorithms.ALG_SHA384:
+            algo = sha384()
+        elif alg_id == CCAlgorithms.ALG_SHA256:
+            algo = sha256()
+        elif alg_id == CCAlgorithms.ALG_SHA512:
+            algo = sha512()
+        else:
+            LOG.error("Unsupported hash algorithm %d", alg_id)
+            continue
+
+        # Initialize value if alg_id not found in dict
+        if alg_id not in measurement_dict[event_log.reg_idx].keys():
+            measurement_dict[event_log.reg_idx][alg_id] = bytearray(
+                event_log.alg_id.digest_size)
+
+        # Do replay and update the result into dict
+        digest = list(map(int, event_log.digest.strip('[]').split(' ')))
+        # pylint: disable-next=consider-using-f-string
+        digest_hex = ''.join('{:02x}'.format(i) for i in digest)
+        algo.update(measurement_dict[event_log.reg_idx][alg_id] + bytes.fromhex(digest_hex))
+        measurement_dict[event_log.reg_idx][alg_id] = algo.digest()
+
+    return measurement_dict
+
+def verify_event_log(measurement_dict: dict) -> bool:
+    """Verify event log by comparing IMR value from CCNP fetching and replayed via event log.
+
+    IMR: Integrated Measurement Register.
+
+    Args:
+        measurement_dict (dict): The event logs replay result displayed by IMR index and hash
+          algorithm.
+
+    Returns:
+        bool: True if event log verify success, False if event log verify failed.
+    """
+    LOG.info("Verify IMR measurement value and replayed value from event logs")
+    for index in range(IMR_VERIFY_COUNT):
+        # Fectch IMR measurement
+        LOG.info("Fetch measurements in IMR[%d]", index)
+        imr_measurement = base64.b64decode(Measurement.get_platform_measurement(
+            MeasurementType.TYPE_TDX_RTMR, None, index))
+        LOG.info("IMR[%d](measurement): %s", index, imr_measurement.hex())
+
+        # Get IMR value from replayed event log
+        if index not in measurement_dict or measurement_dict[index] == {}:
+            LOG.error("IMR[%d] verify failed, the replayed value from event logs doesn't exist",
+                index)
+            return False
+        for value in measurement_dict[index].values():
+            imr_replayed = value
+            break
+
+        LOG.info("IMR[%d](replayed): %s", index, imr_replayed.hex())
+        if imr_measurement == imr_replayed:
+            LOG.info("IMR[%d] passed the verification.", index)
+        else:
+            LOG.error("IMR[%d] did not pass the verification.", index)
+            return False
+
+    return True

cnap/core/keybroker.py

@@ -15,13 +15,15 @@
 import struct
 import requests
 
-from ccnp import Quote
+from ccnp import Eventlog, Quote
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.hazmat.primitives.asymmetric import padding
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 
+from core.eventlog import replay_event_log, verify_event_log
+
 LOG = logging.getLogger(__name__)
 
 # Set the connection timeout to 10s
@@ -68,17 +70,19 @@ class SimpleKeyBrokerClient(KeyBrokerClientBase):
           key (SWK) to encrypt the user key (wrapped_key).
         - Encrypt the SWK by the public key from client (wrapped_swk).
     For a key broker client, here is an example flow to get a key from KBS:
+        - Get and replay all event logs, and verify by the measurement register.
         - Generate 2048 bit RSA key pair (a public key and a private key).
         - Encode the public key to base64 for transferring (user_data).
         - Get quote in the TEE with the hash of the public key for measurement (quote).
         - Request wrapped_key and wrapped_swk from KBS with quote and user_data.
         - Decrypt the user key by the SWK.
     """
-    def get_key(self, server_url: str, key_id: str) -> bytes:
+    def get_key(self, server_url: str, key_id: str) -> bytes: # pylint: disable=too-many-locals
         """Get model key by key ID from the KBS.
   
-        This method construct the request headers and body to request the wrapped_key and
-        wrapped_swk from KBS, decrypt the user key by SWK and return the key.
+        This method get and replay all event logs, and verify by the measurement register, then
+        construct the request headers and body to request the wrapped_key and wrapped_swk from KBS,
+        decrypt the user key by SWK and return the key.
 
         A example requests and response:
             - request headers:
@@ -105,14 +109,27 @@ def get_key(self, server_url: str, key_id: str) -> bytes:
     
         Raises:
             ValueError: If the server_url or key_id is None.
-            RuntimeError: If get quote or get key failed.
+            RuntimeError: If get or verify event log failed, and if get quote or get key failed.
             NotImplementedError: If the subclasses don't implement the method.
         """
         if server_url is None:
             raise ValueError("KBS server url can not be None")
         if key_id is None:
             raise ValueError("KBS key id can not be None")
 
+        # Get and verify event logs before get quote.
+        # The exectuion environment judgment will be implemented by ccnp in the future.
+        LOG.debug("Getting event log by CCNP")
+        event_logs = Eventlog.get_platform_eventlog()
+        if event_logs is None:
+            raise RuntimeError("Get event log failed")
+        measurement_dict = replay_event_log(event_logs)
+        if verify_event_log(measurement_dict):
+            LOG.info("Event log verify successfully.\n")
+        else:
+            LOG.error("Event log verify failed.\n")
+            raise RuntimeError("Event log verify failed.")
+
         private_key = rsa.generate_private_key(public_exponent=65537, key_size=3072)
         pubkey = private_key.public_key()
         pubkey_der = pubkey.public_bytes(encoding=serialization.Encoding.DER,