config, web_util: move get_restful_params(..) to web_util

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

demo/agent_monitor/tenant_agent_monitor.py

@@ -95,7 +95,7 @@ def post(self):
         """
         logger.info('Agent Monitor POST')
         try:
-            rest_params = common.get_restful_params(self.request.path)
+            rest_params = keylime.web_util.get_restful_params(self.request.path)
 
             if "agents" not in rest_params:
                 keylime.web_util.echo_json_response(self, 400, "uri not supported")

keylime/cloud_verifier_tornado.py

@@ -187,7 +187,7 @@ def head(self):
             self, 405, "Not Implemented: Use GET interface instead")
 
     def get(self):
-        rest_params = config.get_restful_params(self.request.uri)
+        rest_params = web_util.get_restful_params(self.request.uri)
         if rest_params is None:
             web_util.echo_json_response(self, 405, "Not Implemented")
             return
@@ -235,7 +235,7 @@ def get(self):
         to contact the Cloud Agent.
         """
         session = get_session()
-        rest_params = config.get_restful_params(self.request.uri)
+        rest_params = web_util.get_restful_params(self.request.uri)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /agents/ interface")
@@ -298,7 +298,7 @@ def delete(self):
         agents requests require a single agent_id parameter which identifies the agent to be deleted.
         """
         session = get_session()
-        rest_params = config.get_restful_params(self.request.uri)
+        rest_params = web_util.get_restful_params(self.request.uri)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /agents/ interface")
@@ -367,7 +367,7 @@ def post(self):
         """
         session = get_session()
         try:
-            rest_params = config.get_restful_params(self.request.uri)
+            rest_params = web_util.get_restful_params(self.request.uri)
             if rest_params is None:
                 web_util.echo_json_response(
                     self, 405, "Not Implemented: Use /agents/ interface")
@@ -471,7 +471,7 @@ def put(self):
         """
         session = get_session()
         try:
-            rest_params = config.get_restful_params(self.request.uri)
+            rest_params = web_util.get_restful_params(self.request.uri)
             if rest_params is None:
                 web_util.echo_json_response(
                     self, 405, "Not Implemented: Use /agents/ interface")
@@ -546,7 +546,7 @@ def get(self):
         GET /allowlists/{name}
         """
 
-        rest_params = config.get_restful_params(self.request.uri)
+        rest_params = web_util.get_restful_params(self.request.uri)
         if rest_params is None or 'allowlists' not in rest_params:
             web_util.echo_json_response(self, 400, "Invalid URL")
             return
@@ -585,7 +585,7 @@ def delete(self):
         DELETE /allowlists/{name}
         """
 
-        rest_params = config.get_restful_params(self.request.uri)
+        rest_params = web_util.get_restful_params(self.request.uri)
         if rest_params is None or 'allowlists' not in rest_params:
             web_util.echo_json_response(self, 400, "Invalid URL")
             return
@@ -637,7 +637,7 @@ def post(self):
         body: {"tpm_policy": {..}, "vtpm_policy": {..}
         """
 
-        rest_params = config.get_restful_params(self.request.uri)
+        rest_params = web_util.get_restful_params(self.request.uri)
         if rest_params is None or 'allowlists' not in rest_params:
             web_util.echo_json_response(self, 400, "Invalid URL")
             return

keylime/config.py

@@ -5,7 +5,6 @@
 import os
 import os.path
 import configparser
-import urllib.parse
 import re
 from typing import Optional
 
@@ -16,7 +15,6 @@
     from yaml import SafeLoader
 from yaml.reader import ReaderError
 
-from keylime import api_version as keylime_api_version
 from keylime import json
 
 
@@ -170,16 +168,6 @@ def ch_dir(path, logger):
     os.chdir(path)
 
 
-def list_to_dict(alist):
-    """Convert list into dictionary via grouping [k0,v0,k1,v1,...]"""
-    params = {}
-    i = 0
-    while i < len(alist):
-        params[alist[i]] = alist[i + 1] if (i + 1) < len(alist) else None
-        i = i + 2
-    return params
-
-
 def yaml_to_dict(arry, add_newlines=True, logger=None) -> Optional[dict]:
     arry = convert(arry)
     sep = "\n" if add_newlines else ""
@@ -191,28 +179,6 @@ def yaml_to_dict(arry, add_newlines=True, logger=None) -> Optional[dict]:
     return None
 
 
-def get_restful_params(urlstring):
-    """Returns a dictionary of paired RESTful URI parameters"""
-    parsed_path = urllib.parse.urlsplit(urlstring.strip("/"))
-    query_params = urllib.parse.parse_qsl(parsed_path.query)
-    path_tokens = parsed_path.path.split('/')
-
-    # If first token looks like an API version, validate it and make sure it's supported
-    api_version = 0
-    if path_tokens[0] and len(path_tokens[0]) >= 0 and re.match(r"^v?[0-9]+(\.[0-9]+)?", path_tokens[0]):
-        version = keylime_api_version.normalize_version(path_tokens[0])
-
-        if keylime_api_version.is_supported_version(version):
-            api_version = version
-
-        path_tokens.pop(0)
-
-    path_params = list_to_dict(path_tokens)
-    path_params["api_version"] = api_version
-    path_params.update(query_params)
-    return path_params
-
-
 def valid_exclude_list(exclude_list):
     if not exclude_list:
         return True, None, None

keylime/keylime_agent.py

@@ -71,7 +71,7 @@ def do_GET(self):
         """
 
         logger.info('GET invoked from %s with uri: %s', self.client_address, self.path)
-        rest_params = config.get_restful_params(self.path)
+        rest_params = web_util.get_restful_params(self.path)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /keys/ or /quotes/ interfaces")
@@ -203,7 +203,7 @@ def do_POST(self):
         Only tenant and cloudverifier uri's are supported. Both requests require a nonce parameter.
         The Cloud verifier requires an additional mask parameter.  If the uri or parameters are incorrect, a 400 response is returned.
         """
-        rest_params = config.get_restful_params(self.path)
+        rest_params = web_util.get_restful_params(self.path)
 
         if rest_params is None:
             web_util.echo_json_response(

keylime/registrar_common.py

@@ -56,7 +56,7 @@ def do_GET(self):
         agent to be returned. If the agent_id is not found, a 404 response is returned.
         """
         session = SessionManager().make_session(engine)
-        rest_params = config.get_restful_params(self.path)
+        rest_params = web_util.get_restful_params(self.path)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /agents/ interface")
@@ -131,7 +131,7 @@ def do_DELETE(self):
         agents requests require a single agent_id parameter which identifies the agent to be deleted.
         """
         session = SessionManager().make_session(engine)
-        rest_params = config.get_restful_params(self.path)
+        rest_params = web_util.get_restful_params(self.path)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /agents/ interface")
@@ -184,7 +184,7 @@ def do_GET(self):
 
         Currently the only supported path is /versions which shows the supported API versions
         """
-        rest_params = config.get_restful_params(self.path)
+        rest_params = web_util.get_restful_params(self.path)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /version/ interface")
@@ -210,7 +210,7 @@ def do_POST(self):
         block sent in the body with 2 entries: ek and aik.
         """
         session = SessionManager().make_session(engine)
-        rest_params = config.get_restful_params(self.path)
+        rest_params = web_util.get_restful_params(self.path)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /agents/ interface")
@@ -381,7 +381,7 @@ def do_PUT(self):
         will return errors.
         """
         session = SessionManager().make_session(engine)
-        rest_params = config.get_restful_params(self.path)
+        rest_params = web_util.get_restful_params(self.path)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /agents/ interface")

keylime/tenant_webapp.py

@@ -371,7 +371,7 @@ async def get(self):
         will return errors.
         """
 
-        rest_params = config.get_restful_params(self.request.uri)
+        rest_params = web_util.get_restful_params(self.request.uri)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /agents/ or /logs/ interface")
@@ -443,7 +443,7 @@ def delete(self):
         agents requests require a single agent_id parameter which identifies the agent to be deleted.
         """
 
-        rest_params = config.get_restful_params(self.request.uri)
+        rest_params = web_util.get_restful_params(self.request.uri)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /agents/ interface")
@@ -470,7 +470,7 @@ def post(self):
         agents requests require a yaml block sent in the body
         """
 
-        rest_params = config.get_restful_params(self.request.uri)
+        rest_params = web_util.get_restful_params(self.request.uri)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /agents/ interface")
@@ -581,7 +581,7 @@ def put(self):
         Currently, only agents resources are available for PUTing, i.e. /agents. All other PUT uri's will return errors.
         """
 
-        rest_params = config.get_restful_params(self.request.uri)
+        rest_params = web_util.get_restful_params(self.request.uri)
         if rest_params is None:
             web_util.echo_json_response(
                 self, 405, "Not Implemented: Use /agents/ interface")

keylime/web_util.py

@@ -1,13 +1,15 @@
 import http.client
 import os
+import re
 import socket
 import ssl
 import sys
+import urllib.parse
 from http.server import BaseHTTPRequestHandler
 
 import tornado.web
 
-from keylime import config, ca_util, json
+from keylime import config, ca_util, json, api_version as keylime_api_version
 from keylime.cloud_verifier_common import logger
 
 
@@ -143,3 +145,35 @@ def echo_json_response(handler, code, status=None, results=None):
         return True
 
     return False
+
+
+def get_restful_params(urlstring):
+    """Returns a dictionary of paired RESTful URI parameters"""
+    parsed_path = urllib.parse.urlsplit(urlstring.strip("/"))
+    query_params = urllib.parse.parse_qsl(parsed_path.query)
+    path_tokens = parsed_path.path.split('/')
+
+    # If first token looks like an API version, validate it and make sure it's supported
+    api_version = 0
+    if path_tokens[0] and len(path_tokens[0]) >= 0 and re.match(r"^v?[0-9]+(\.[0-9]+)?", path_tokens[0]):
+        version = keylime_api_version.normalize_version(path_tokens[0])
+
+        if keylime_api_version.is_supported_version(version):
+            api_version = version
+
+        path_tokens.pop(0)
+
+    path_params = _list_to_dict(path_tokens)
+    path_params["api_version"] = api_version
+    path_params.update(query_params)
+    return path_params
+
+
+def _list_to_dict(alist):
+    """Convert list into dictionary via grouping [k0,v0,k1,v1,...]"""
+    params = {}
+    i = 0
+    while i < len(alist):
+        params[alist[i]] = alist[i + 1] if (i + 1) < len(alist) else None
+        i = i + 2
+    return params