CVE-2022-23949: Authentication Bypass by Spoofing (CWE-290)
value: In Keylime before 6.3.0, unsanitized UUIDs can be passed by a rogue agent and can lead to log spoofing on the verifier and registrar.
CWE description: This attack-focused weakness is caused by incorrectly implemented authentication schemes that are subject to spoofing attacks.
CVSS v3 base score: 7.5
Published: 2022-09-21T19:15Z
Repository: https://github.com/keylime/keylime
Language: Python
Files changed: keylime/cloud_verifier_tornado.py, keylime/keylime_agent.py, keylime/registrar_common.py, keylime/tenant.py, keylime/tenant_webapp.py
Commit date: 2022-01-27 11:09:26 -0500
Commit message: Validate user ID in all public interfaces

The user ID is read from the config file, or from some public REST API.
We should validate that is composed with valid set of chars.

Signed-off-by: Alberto Planas <aplanas@suse.com>

Diff: diff --git a/keylime/cloud_verifier_tornado.py b/keylime/cloud_verifier_tornado.py
index 75d2bfe..40da93e 100644
--- a/keylime/cloud_verifier_tornado.py
+++ b/keylime/cloud_verifier_tornado.py
@@ -16,11 +16,11 @@ import tornado.web
 
 from keylime import config
 from keylime import json
 from keylime import registrar_client
 from keylime.agentstates import AgentAttestStates
-from keylime.common import states
+from keylime.common import states, validators
 from keylime.db.verifier_db import VerfierMain
 from keylime.db.verifier_db import VerifierAllowlist
 from keylime.db.keylime_db import DBEngineManager, SessionManager
 from keylime import keylime_logging
 from keylime import cloud_verifier_common
@@ -262,10 +262,17 @@ class AgentsHandler(BaseHandler):
             return
 
         agent_id = rest_params["agents"]
 
         if (agent_id is not None) and (agent_id != ''):
+            # If the agent ID is not valid (wrong set of characters),
+            # just do nothing.
+            if not validators.valid_agent_id(agent_id):
+                web_util.echo_json_response(self, 400, "agent_id not not valid")
+                logger.error("GET received an invalid agent ID: %s", agent_id)
+                return
+
             try:
                 agent = session.query(VerfierMain).filter_by(
                     agent_id=agent_id).one_or_none()
             except SQLAlchemyError as e:
                 logger.error('SQLAlchemy Error: %s', e)
@@ -328,10 +335,17 @@ class AgentsHandler(BaseHandler):
         if agent_id is None:
             web_util.echo_json_response(self, 400, "uri not supported")
             logger.warning('DELETE returning 400 response. uri not supported: %s', self.request.path)
             return
 
+        # If the agent ID is not valid (wrong set of characters), just
+        # do nothing.
+        if not validators.valid_agent_id(agent_id):
+            web_util.echo_json_response(self, 400, "agent_id not not valid")
+            logger.error("DELETE received an invalid agent ID: %s", agent_id)
+            return
+
         try:
             agent = session.query(VerfierMain).filter_by(
                 agent_id=agent_id).first()
         except SQLAlchemyError as e:
             logger.error('SQLAlchemy Error: %s', e)
@@ -394,10 +408,17 @@ class AgentsHandler(BaseHandler):
                 return
 
             agent_id = rest_params["agents"]
 
             if agent_id is not None:
+                # If the agent ID is not valid (wrong set of
+                # characters), just do nothing.
+                if not validators.valid_agent_id(agent_id):
+                    web_util.echo_json_response(self, 400, "agent_id not not valid")
+                    logger.error("POST received an invalid agent ID: %s", agent_id)
+                    return
+
                 content_length = len(self.request.body)
                 if content_length == 0:
                     web_util.echo_json_response(
                         self, 400, "Expected non zero content length")
                     logger.warning('POST returning 400 response. Expected non zero content length.')
@@ -532,10 +553,18 @@ class AgentsHandler(BaseHandler):
             agent_id = rest_params["agents"]
 
             if agent_id is None:
                 web_util.echo_json_response(self, 400, "uri not supported")
                 logger.warning("PUT returning 400 response. uri not supported")
+
+            # If the agent ID is not valid (wrong set of characters),
+            # just do nothing.
+            if not validators.valid_agent_id(agent_id):
+                web_util.echo_json_response(self, 400, "agent_id not not valid")
+                logger.error("PUT received an invalid agent ID: %s", agent_id)
+                return
+
             try:
                 verifier_id = config.get('cloud_verifier', 'cloudverifier_id', fallback=cloud_verifier_common.DEFAULT_VERIFIER_ID)
                 agent = session.query(VerfierMain).filter_by(
                     agent_id=agent_id, verifier_id=verifier_id).one()
             except SQLAlchemyError as e:
diff --git a/keylime/keylime_agent.py b/keylime/keylime_agent.py
index bf41328..47fcb24 100644
--- a/keylime/keylime_agent.py
+++ b/keylime/keylime_agent.py
@@ -674,10 +674,15 @@ def main():
         agent_uuid = socket.getfqdn()
     elif agent_uuid == 'environment':
         agent_uuid = os.getenv("KEYLIME_AGENT_UUID", None)
         if agent_uuid is None:
             raise RuntimeError("Env variable KEYLIME_AGENT_UUID is empty, but agent_uuid is set to 'environment'")
+    elif not validators.valid_uuid(agent_uuid):
+        raise RuntimeError("The UUID is not valid")
+
+    if not validators.valid_agent_id(agent_uuid):
+        raise RuntimeError("The agent ID set via agent uuid parameter use invalid characters")
 
     if config.STUB_VTPM and config.TPM_CANNED_VALUES is not None:
         # Use canned values for stubbing
         jsonIn = config.TPM_CANNED_VALUES
         if "add_vtpm_to_group" in jsonIn:
diff --git a/keylime/registrar_common.py b/keylime/registrar_common.py
index b9164b0..ca77ac0 100644
--- a/keylime/registrar_common.py
+++ b/keylime/registrar_common.py
@@ -15,10 +15,11 @@ from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm.exc import NoResultFound
 
 from cryptography.hazmat.backends import default_backend
 from cryptography.x509 import load_der_x509_certificate
 
+from keylime.common import validators
 from keylime.db.registrar_db import RegistrarMain
 from keylime.db.keylime_db import DBEngineManager, SessionManager
 from keylime import config
 from keylime import crypto
 from keylime import json
@@ -72,10 +73,17 @@ class ProtectedHandler(BaseHTTPRequestHandler, SessionManager):
             return
 
         agent_id = rest_params["agents"]
 
         if agent_id is not None:
+            # If the agent ID is not valid (wrong set of characters),
+            # just do nothing.
+            if not validators.valid_agent_id(agent_id):
+                web_util.echo_json_response(self, 400, "agent_id not not valid")
+                logger.error("GET received an invalid agent ID: %s", agent_id)
+                return
+
             try:
                 agent = session.query(RegistrarMain).filter_by(
                     agent_id=agent_id).first()
             except SQLAlchemyError as e:
                 logger.error('SQLAlchemy Error: %s', e)
@@ -148,10 +156,17 @@ class ProtectedHandler(BaseHTTPRequestHandler, SessionManager):
             return
 
         agent_id = rest_params["agents"]
 
         if agent_id is not None:
+            # If the agent ID is not valid (wrong set of characters),
+            # just do nothing.
+            if not validators.valid_agent_id(agent_id):
+                web_util.echo_json_response(self, 400, "agent_id not not valid")
+                logger.error("DELETE received an invalid agent ID: %s", agent_id)
+                return
+
             if session.query(RegistrarMain).filter_by(agent_id=agent_id).delete():
                 # send response
                 try:
                     session.commit()
                 except SQLAlchemyError as e:
@@ -231,10 +246,17 @@ class UnprotectedHandler(BaseHTTPRequestHandler, SessionManager):
         if agent_id is None:
             web_util.echo_json_response(self, 400, "agent id not found in uri")
             logger.warning('POST agent returning 400 response. agent id not found in uri %s', self.path)
             return
 
+        # If the agent ID is not valid (wrong set of characters), just
+        # do nothing.
+        if not validators.valid_agent_id(agent_id):
+            web_util.echo_json_response(self, 400, "agent id not valid")
+            logger.error("POST received an invalid agent ID: %s", agent_id)
+            return
+
         try:
             content_length = int(self.headers.get('Content-Length', 0))
             if content_length == 0:
                 web_util.echo_json_response(
                     self, 400, "Expected non zero content length")
@@ -408,10 +430,17 @@ class UnprotectedHandler(BaseHTTPRequestHandler, SessionManager):
         if agent_id is None:
             web_util.echo_json_response(self, 400, "agent id not found in uri")
             logger.warning('PUT agent returning 400 response. agent id not found in uri %s', self.path)
             return
 
+        # If the agent ID is not valid (wrong set of characters), just
+        # do nothing.
+        if not validators.valid_agent_id(agent_id):
+            web_util.echo_json_response(self, 400, "agent_id not not valid")
+            logger.error("PUT received an invalid agent ID: %s", agent_id)
+            return
+
         try:
             content_length = int(self.headers.get('Content-Length', 0))
             if content_length == 0:
                 web_util.echo_json_response(
                     self, 400, "Expected non zero content length")
diff --git a/keylime/tenant.py b/keylime/tenant.py
index 92d4a41..ff279e5 100755
--- a/keylime/tenant.py
+++ b/keylime/tenant.py
@@ -32,11 +32,11 @@ from keylime.tpm.tpm_main import tpm
 from keylime.tpm.tpm_abstract import TPM_Utilities
 from keylime import ima
 from keylime import crypto
 from keylime.cmd import user_data_encrypt
 from keylime import ca_util
-from keylime.common import algorithms
+from keylime.common import algorithms, validators
 from keylime import ima_file_signatures
 from keylime import measured_boot
 from keylime import gpg
 from keylime import api_version as keylime_api_version
 
@@ -1311,10 +1311,12 @@ def main(argv=sys.argv):
         mytenant.agent_uuid = args.agent_uuid
         # if the uuid is actually a public key, then hash it
         if mytenant.agent_uuid.startswith('-----BEGIN PUBLIC KEY-----'):
             mytenant.agent_uuid = hashlib.sha256(
                 mytenant.agent_uuid).hexdigest()
+        if not validators.valid_agent_id(mytenant.agent_uuid):
+            raise UserError("The agent ID set via agent uuid parameter use invalid characters")
     else:
         logger.warning("Using default UUID d432fbb3-d2f1-4a97-9ef7-75bd81c00000")
         mytenant.agent_uuid = "d432fbb3-d2f1-4a97-9ef7-75bd81c00000"
 
     if config.STUB_VTPM and config.TPM_CANNED_VALUES is not None:
diff --git a/keylime/tenant_webapp.py b/keylime/tenant_webapp.py
index fd8b60b..1fc12e9 100644
--- a/keylime/tenant_webapp.py
+++ b/keylime/tenant_webapp.py
@@ -14,11 +14,11 @@ import sys
 
 import tornado.ioloop
 import tornado.web
 
 from keylime.requests_client import RequestsClient
-from keylime.common import states
+from keylime.common import validators, states
 from keylime import config
 from keylime import json
 from keylime import keylime_logging
 from keylime import tenant
 from keylime import web_util
@@ -393,10 +393,17 @@ class AgentsHandler(BaseHandler):
             logger.warning('GET returning 400 response. uri not supported: %s', self.request.path)
             return
 
         agent_id = rest_params["agents"]
         if agent_id is not None:
+            # If the agent ID is not valid (wrong set of characters),
+            # just do nothing.
+            if not validators.valid_agent_id(agent_id):
+                web_util.echo_json_response(self, 400, "agent_id not not valid")
+                logger.error("GET received an invalid agent ID: %s", agent_id)
+                return
+
             # Handle request for specific agent data separately
             agents = await self.get_agent_state(agent_id)
             agents["id"] = agent_id
 
             web_util.echo_json_response(self, 200, "Success", agents)
@@ -453,10 +460,16 @@ class AgentsHandler(BaseHandler):
             web_util.echo_json_response(self, 400, "uri not supported")
             logger.warning('DELETE returning 400 response. uri not supported: %s', self.request.path)
             return
 
         agent_id = rest_params["agents"]
+        # If the agent ID is not valid (wrong set of characters), just
+        # do nothing.
+        if not validators.valid_agent_id(agent_id):
+            web_util.echo_json_response(self, 400, "agent_id not not valid")
+            logger.error("DELETE received an invalid agent ID: %s", agent_id)
+            return
 
         # let Tenant do dirty work of deleting agent
         mytenant = tenant.Tenant()
         mytenant.agent_uuid = agent_id
         mytenant.do_cvdelete()
@@ -480,10 +493,16 @@ class AgentsHandler(BaseHandler):
             web_util.echo_json_response(self, 400, "uri not supported")
             logger.warning('POST returning 400 response. uri not supported: %s', self.request.path)
             return
 
         agent_id = rest_params["agents"]
+        # If the agent ID is not valid (wrong set of characters), just
+        # do nothing.
+        if not validators.valid_agent_id(agent_id):
+            web_util.echo_json_response(self, 400, "agent_id not not valid")
+            logger.error("POST received an invalid agent ID: %s", agent_id)
+            return
 
         # Parse payload files (base64 data-uri)
         if self.get_argument("ptype", Agent_Init_Types.FILE, True) == Agent_Init_Types.FILE:
             keyfile = None
             payload = None
@@ -591,10 +610,16 @@ class AgentsHandler(BaseHandler):
             web_util.echo_json_response(self, 400, "uri not supported")
             logger.warning('PUT returning 400 response. uri not supported: %s', self.request.path)
             return
 
         agent_id = rest_params["agents"]
+        # If the agent ID is not valid (wrong set of characters), just
+        # do nothing.
+        if not validators.valid_agent_id(agent_id):
+            web_util.echo_json_response(self, 400, "agent_id not not valid")
+            logger.error("PUT received an invalid agent ID: %s", agent_id)
+            return
 
         # let Tenant do dirty work of reactivating agent
         mytenant = tenant.Tenant()
         mytenant.agent_uuid = agent_id
         mytenant.do_cvreactivate()

Vulnerable code: from keylime.common import states
from keylime.common import algorithms
from keylime.common import states
Fixed code: from keylime.common import states, validators
            # If the agent ID is not valid (wrong set of characters),
            # just do nothing.
            if not validators.valid_agent_id(agent_id):
                web_util.echo_json_response(self, 400, "agent_id not not valid")
                logger.error("GET received an invalid agent ID: %s", agent_id)
                return

        # If the agent ID is not valid (wrong set of characters), just
        # do nothing.
        if not validators.valid_agent_id(agent_id):
            web_util.echo_json_response(self, 400, "agent_id not not valid")
            logger.error("DELETE received an invalid agent ID: %s", agent_id)
            return

                # If the agent ID is not valid (wrong set of
                # characters), just do nothing.
                if not validators.valid_agent_id(agent_id):
                    web_util.echo_json_response(self, 400, "agent_id not not valid")
                    logger.error("POST received an invalid agent ID: %s", agent_id)
                    return


            # If the agent ID is not valid (wrong set of characters),
            # just do nothing.
            if not validators.valid_agent_id(agent_id):
                web_util.echo_json_response(self, 400, "agent_id not not valid")
                logger.error("PUT received an invalid agent ID: %s", agent_id)
                return

    elif not validators.valid_uuid(agent_uuid):
        raise RuntimeError("The UUID is not valid")

    if not validators.valid_agent_id(agent_uuid):
        raise RuntimeError("The agent ID set via agent uuid parameter use invalid characters")
from keylime.common import validators
            # If the agent ID is not valid (wrong set of characters),
            # just do nothing.
            if not validators.valid_agent_id(agent_id):
                web_util.echo_json_response(self, 400, "agent_id not not valid")
                logger.error("GET received an invalid agent ID: %s", agent_id)
                return

            # If the agent ID is not valid (wrong set of characters),
            # just do nothing.
            if not validators.valid_agent_id(agent_id):
                web_util.echo_json_response(self, 400, "agent_id not not valid")
                logger.error("DELETE received an invalid agent ID: %s", agent_id)
                return

        # If the agent ID is not valid (wrong set of characters), just
        # do nothing.
        if not validators.valid_agent_id(agent_id):
            web_util.echo_json_response(self, 400, "agent id not valid")
            logger.error("POST received an invalid agent ID: %s", agent_id)
            return

        # If the agent ID is not valid (wrong set of characters), just
        # do nothing.
        if not validators.valid_agent_id(agent_id):
            web_util.echo_json_response(self, 400, "agent_id not not valid")
            logger.error("PUT received an invalid agent ID: %s", agent_id)
            return

from keylime.common import algorithms, validators
        if not validators.valid_agent_id(mytenant.agent_uuid):
            raise UserError("The agent ID set via agent uuid parameter use invalid characters")
from keylime.common import validators, states
            # If the agent ID is not valid (wrong set of characters),
            # just do nothing.
            if not validators.valid_agent_id(agent_id):
                web_util.echo_json_response(self, 400, "agent_id not not valid")
                logger.error("GET received an invalid agent ID: %s", agent_id)
                return

        # If the agent ID is not valid (wrong set of characters), just
        # do nothing.
        if not validators.valid_agent_id(agent_id):
            web_util.echo_json_response(self, 400, "agent_id not not valid")
            logger.error("DELETE received an invalid agent ID: %s", agent_id)
            return
        # If the agent ID is not valid (wrong set of characters), just
        # do nothing.
        if not validators.valid_agent_id(agent_id):
            web_util.echo_json_response(self, 400, "agent_id not not valid")
            logger.error("POST received an invalid agent ID: %s", agent_id)
            return
        # If the agent ID is not valid (wrong set of characters), just
        # do nothing.
        if not validators.valid_agent_id(agent_id):
            web_util.echo_json_response(self, 400, "agent_id not not valid")
            logger.error("PUT received an invalid agent ID: %s", agent_id)
            return