CVE-2022-23951: Insufficient Information (NVD-CWE-noinfo)
value: In Keylime before 6.3.0, quote responses from the agent can contain possibly untrusted ZIP data which can lead to zip bombs.
CWE description: There is insufficient information about the issue to classify it; details are unkown or unspecified.
CVSS v3 base score: 5.5
Published: 2022-09-21T19:15Z
Repository: https://github.com/keylime/keylime
Language: Python
Files changed: keylime/cloud_verifier_common.py, keylime/tenant.py, keylime/tpm/tpm_abstract.py, keylime/tpm/tpm_main.py
Commit date: 2022-01-27 11:09:26 -0500
Commit message: tpm: do not compress quote with zlib by default

Pythons zlib decompression has no mitigations against zip bombs or similar
attacks, so we remove the compression for the quote data.

The data itself is rather small compared to the IMA or UEFI event log.

Signed-off-by: Thore Sommer <mail@thson.de>

Diff: diff --git a/keylime/cloud_verifier_common.py b/keylime/cloud_verifier_common.py
index 74e5281..6c004b3 100644
--- a/keylime/cloud_verifier_common.py
+++ b/keylime/cloud_verifier_common.py
@@ -150,11 +150,12 @@ def process_quote_response(agent, json_response, agentAttestState) -> Failure:
         ima_measurement_list,
         agent['allowlist'],
         algorithms.Hash(hash_alg),
         ima_keyrings,
         mb_measurement_list,
-        agent['mb_refstate'])
+        agent['mb_refstate'],
+        compressed=(agent['supported_version'] == "1.0"))  # TODO: change this to always False after initial update
     failure.merge(quote_validation_failure)
 
     if not failure:
         # set a flag so that we know that the agent was verified once.
         # we only issue notifications for agents that were at some point good
diff --git a/keylime/tenant.py b/keylime/tenant.py
index d5b4913..92d4a41 100755
--- a/keylime/tenant.py
+++ b/keylime/tenant.py
@@ -543,11 +543,13 @@ class Tenant():
         registrar_client.init_client_tls('tenant')
         if self.registrar_data is None:
             logger.warning("AIK not found in registrar, quote not validated")
             return False
 
-        failure = self.tpm_instance.check_quote(AgentAttestState(self.agent_uuid), self.nonce, public_key, quote, self.registrar_data['aik_tpm'], hash_alg=hash_alg)
+        failure = self.tpm_instance.check_quote(AgentAttestState(self.agent_uuid), self.nonce, public_key, quote,
+                                                self.registrar_data['aik_tpm'], hash_alg=hash_alg,
+                                                compressed=(self.supported_version == "1.0"))
         if failure:
             if self.registrar_data['regcount'] > 1:
                 logger.error("WARNING: This UUID had more than one ek-ekcert registered to it! This might indicate that your system is misconfigured or a malicious host is present. Run 'regdelete' for this agent and restart")
                 sys.exit()
             return False
diff --git a/keylime/tpm/tpm_abstract.py b/keylime/tpm/tpm_abstract.py
index 9f4e134..76f4079 100644
--- a/keylime/tpm/tpm_abstract.py
+++ b/keylime/tpm/tpm_abstract.py
@@ -156,15 +156,15 @@ class AbstractTPM(metaclass=ABCMeta):
     def tpm_init(self, self_activate=False, config_pw=None):
         pass
 
     # tpm_quote
     @abstractmethod
-    def create_quote(self, nonce, data=None, pcrmask=EMPTYMASK, hash_alg=None):
+    def create_quote(self, nonce, data=None, pcrmask=EMPTYMASK, hash_alg=None, compress=False):
         pass
 
     @abstractmethod
-    def check_quote(self, agentAttestState, nonce, data, quote, aikTpmFromRegistrar, tpm_policy={}, ima_measurement_list=None, allowlist={}, hash_alg=None, ima_keyrings=None, mb_measurement_list=None, mb_refstate=None):
+    def check_quote(self, agentAttestState, nonce, data, quote, aikTpmFromRegistrar, tpm_policy={}, ima_measurement_list=None, allowlist={}, hash_alg=None, ima_keyrings=None, mb_measurement_list=None, mb_refstate=None, compressed=False):
         pass
 
     def START_HASH(self, algorithm=None):
         if algorithm is None:
             algorithm = self.defaults['hash']
diff --git a/keylime/tpm/tpm_main.py b/keylime/tpm/tpm_main.py
index eb66186..85d8c5d 100644
--- a/keylime/tpm/tpm_main.py
+++ b/keylime/tpm/tpm_main.py
@@ -957,11 +957,11 @@ class tpm(tpm_abstract.AbstractTPM):
         for pcr in range(24):
             if tpm_abstract.TPM_Utilities.check_mask(mask, pcr):
                 pcr_list.append(str(pcr))
         return ",".join(pcr_list)
 
-    def create_quote(self, nonce, data=None, pcrmask=tpm_abstract.AbstractTPM.EMPTYMASK, hash_alg=None):
+    def create_quote(self, nonce, data=None, pcrmask=tpm_abstract.AbstractTPM.EMPTYMASK, hash_alg=None, compress=False):
         if hash_alg is None:
             hash_alg = self.defaults['hash']
 
         quote = ""
 
@@ -990,15 +990,19 @@ class tpm(tpm_abstract.AbstractTPM):
                     command = ["tpm2_quote", "-k", hex(keyhandle), "-L", "%s:%s" % (hash_alg, pcrlist), "-q", nonce, "-m", quotepath.name, "-s", sigpath.name, "-p", pcrpath.name, "-G", hash_alg, "-P", aik_pw]
                 elif self.tools_version in ["4.0", "4.2"]:
                     command = ["tpm2_quote", "-c", keyhandle, "-l", "%s:%s" % (hash_alg, pcrlist), "-q", nonce, "-m", quotepath.name, "-s", sigpath.name, "-o", pcrpath.name, "-g", hash_alg, "-p", aik_pw]
                 retDict = self.__run(command, lock=False, outputpaths=[quotepath.name, sigpath.name, pcrpath.name])
                 quoteraw = retDict['fileouts'][quotepath.name]
-                quote_b64encode = base64.b64encode(zlib.compress(quoteraw))
                 sigraw = retDict['fileouts'][sigpath.name]
-                sigraw_b64encode = base64.b64encode(zlib.compress(sigraw))
                 pcrraw = retDict['fileouts'][pcrpath.name]
-                pcrraw_b64encode = base64.b64encode(zlib.compress(pcrraw))
+                if compress:
+                    quoteraw = zlib.compress(quoteraw)
+                    sigraw = zlib.compress(sigraw)
+                    pcrraw = zlib.compress(pcrraw)
+                quote_b64encode = base64.b64encode(quoteraw)
+                sigraw_b64encode = base64.b64encode(sigraw)
+                pcrraw_b64encode = base64.b64encode(pcrraw)
                 quote = quote_b64encode.decode('utf-8') + ":" + sigraw_b64encode.decode('utf-8') + ":" + pcrraw_b64encode.decode('utf-8')
 
         return 'r' + quote
 
     def __tpm2_checkquote(self, pubaik, nonce, quoteFile, sigFile, pcrFile, hash_alg):
@@ -1016,16 +1020,17 @@ class tpm(tpm_abstract.AbstractTPM):
         elif self.tools_version in ["4.0", "4.2"]:
             command = ["tpm2_checkquote", "-u", pubaik, "-m", quoteFile, "-s", sigFile, "-f", pcrFile, "-g", hash_alg, "-q", nonce]
         retDict = self.__run(command, lock=False)
         return retDict
 
-    def _tpm2_checkquote(self, aikTpmFromRegistrar, quote, nonce, hash_alg):
+    def _tpm2_checkquote(self, aikTpmFromRegistrar, quote, nonce, hash_alg, compressed):
         """Write the files from data returned from tpm2_quote for running tpm2_checkquote
         :param aikTpmFromRegistrar: AIK used to generate the quote and is needed for verifying it now.
         :param quote: quote data in the format 'r<b64-compressed-quoteblob>:<b64-compressed-sigblob>:<b64-compressed-pcrblob>
         :param nonce: nonce that was used to create the quote
         :param hash_alg: the hash algorithm that was used
+        :param compressed: if the quote data is compressed with zlib or not
         :returns: Returns the 'retout' from running tpm2_checkquote and True in case of success, None and False in case of error.
         This function throws an Exception on bad input.
         """
         aikFromRegistrar = tpm2_objects.pubkey_from_tpm2b_public(
             base64.b64decode(aikTpmFromRegistrar),
@@ -1040,13 +1045,20 @@ class tpm(tpm_abstract.AbstractTPM):
 
         quote_tokens = quote.split(":")
         if len(quote_tokens) < 3:
             raise Exception("Quote is not compound! %s" % quote)
 
-        quoteblob = zlib.decompress(base64.b64decode(quote_tokens[0]))
-        sigblob = zlib.decompress(base64.b64decode(quote_tokens[1]))
-        pcrblob = zlib.decompress(base64.b64decode(quote_tokens[2]))
+        quoteblob = base64.b64decode(quote_tokens[0])
+        sigblob = base64.b64decode(quote_tokens[1])
+        pcrblob = base64.b64decode(quote_tokens[2])
+
+        if compressed:
+            logger.warning("Decompressing quote data which is unsafe!")
+            quoteblob = zlib.decompress(quoteblob)
+            sigblob = zlib.decompress(sigblob)
+            pcrblob = zlib.decompress(pcrblob)
+
 
         qfd = sfd = pfd = afd = -1
         quoteFile = None
         aikFile = None
         sigFile = None
@@ -1098,16 +1110,16 @@ class tpm(tpm_abstract.AbstractTPM):
 
         return retout, True
 
     def check_quote(self, agentAttestState, nonce, data, quote, aikTpmFromRegistrar, tpm_policy={},
                     ima_measurement_list=None, allowlist={}, hash_alg=None, ima_keyrings=None,
-                    mb_measurement_list=None, mb_refstate=None) -> Failure:
+                    mb_measurement_list=None, mb_refstate=None, compressed=False) -> Failure:
         failure = Failure(Component.QUOTE_VALIDATION)
         if hash_alg is None:
             hash_alg = self.defaults['hash']
 
-        retout, success = self._tpm2_checkquote(aikTpmFromRegistrar, quote, nonce, hash_alg)
+        retout, success = self._tpm2_checkquote(aikTpmFromRegistrar, quote, nonce, hash_alg, compressed)
         if not success:
             # If the quote validation fails we will skip all other steps therefore this failure is irrecoverable.
             failure.add_event("quote_validation", {"message": "Quote validation using tpm2-tools", "data": retout}, False)
             return failure
 

Vulnerable code:         agent['mb_refstate'])
        failure = self.tpm_instance.check_quote(AgentAttestState(self.agent_uuid), self.nonce, public_key, quote, self.registrar_data['aik_tpm'], hash_alg=hash_alg)
    def create_quote(self, nonce, data=None, pcrmask=EMPTYMASK, hash_alg=None):
    def check_quote(self, agentAttestState, nonce, data, quote, aikTpmFromRegistrar, tpm_policy={}, ima_measurement_list=None, allowlist={}, hash_alg=None, ima_keyrings=None, mb_measurement_list=None, mb_refstate=None):
    def create_quote(self, nonce, data=None, pcrmask=tpm_abstract.AbstractTPM.EMPTYMASK, hash_alg=None):
                quote_b64encode = base64.b64encode(zlib.compress(quoteraw))
                sigraw_b64encode = base64.b64encode(zlib.compress(sigraw))
                pcrraw_b64encode = base64.b64encode(zlib.compress(pcrraw))
    def _tpm2_checkquote(self, aikTpmFromRegistrar, quote, nonce, hash_alg):
        quoteblob = zlib.decompress(base64.b64decode(quote_tokens[0]))
        sigblob = zlib.decompress(base64.b64decode(quote_tokens[1]))
        pcrblob = zlib.decompress(base64.b64decode(quote_tokens[2]))
                    mb_measurement_list=None, mb_refstate=None) -> Failure:
        retout, success = self._tpm2_checkquote(aikTpmFromRegistrar, quote, nonce, hash_alg)
Fixed code:         agent['mb_refstate'],
        compressed=(agent['supported_version'] == "1.0"))  # TODO: change this to always False after initial update
        failure = self.tpm_instance.check_quote(AgentAttestState(self.agent_uuid), self.nonce, public_key, quote,
                                                self.registrar_data['aik_tpm'], hash_alg=hash_alg,
                                                compressed=(self.supported_version == "1.0"))
    def create_quote(self, nonce, data=None, pcrmask=EMPTYMASK, hash_alg=None, compress=False):
    def check_quote(self, agentAttestState, nonce, data, quote, aikTpmFromRegistrar, tpm_policy={}, ima_measurement_list=None, allowlist={}, hash_alg=None, ima_keyrings=None, mb_measurement_list=None, mb_refstate=None, compressed=False):
    def create_quote(self, nonce, data=None, pcrmask=tpm_abstract.AbstractTPM.EMPTYMASK, hash_alg=None, compress=False):
                if compress:
                    quoteraw = zlib.compress(quoteraw)
                    sigraw = zlib.compress(sigraw)
                    pcrraw = zlib.compress(pcrraw)
                quote_b64encode = base64.b64encode(quoteraw)
                sigraw_b64encode = base64.b64encode(sigraw)
                pcrraw_b64encode = base64.b64encode(pcrraw)
    def _tpm2_checkquote(self, aikTpmFromRegistrar, quote, nonce, hash_alg, compressed):
        :param compressed: if the quote data is compressed with zlib or not
        quoteblob = base64.b64decode(quote_tokens[0])
        sigblob = base64.b64decode(quote_tokens[1])
        pcrblob = base64.b64decode(quote_tokens[2])

        if compressed:
            logger.warning("Decompressing quote data which is unsafe!")
            quoteblob = zlib.decompress(quoteblob)
            sigblob = zlib.decompress(sigblob)
            pcrblob = zlib.decompress(pcrblob)

                    mb_measurement_list=None, mb_refstate=None, compressed=False) -> Failure:
        retout, success = self._tpm2_checkquote(aikTpmFromRegistrar, quote, nonce, hash_alg, compressed)