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>

keylime/cloud_verifier_tornado.py

@@ -18,7 +18,7 @@
 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
@@ -264,6 +264,13 @@ def get(self):
         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()
@@ -330,6 +337,13 @@ def delete(self):
             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()
@@ -396,6 +410,13 @@ def post(self):
             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(
@@ -534,6 +555,14 @@ def put(self):
             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(

keylime/keylime_agent.py

@@ -676,6 +676,11 @@ def main():
         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

keylime/registrar_common.py

@@ -17,6 +17,7 @@
 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
@@ -74,6 +75,13 @@ def do_GET(self):
         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()
@@ -150,6 +158,13 @@ def do_DELETE(self):
         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:
@@ -233,6 +248,13 @@ def do_POST(self):
             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:
@@ -410,6 +432,13 @@ def do_PUT(self):
             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:

keylime/tenant.py

@@ -34,7 +34,7 @@
 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
@@ -1313,6 +1313,8 @@ def main(argv=sys.argv):
         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"

keylime/tenant_webapp.py

@@ -16,7 +16,7 @@
 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
@@ -395,6 +395,13 @@ async def get(self):
 
         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
@@ -455,6 +462,12 @@ def delete(self):
             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()
@@ -482,6 +495,12 @@ def post(self):
             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:
@@ -593,6 +612,12 @@ def put(self):
             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()