feat: make jwt options configurable

cfpb · Sep 21, 2023 · 891112e · 891112e
1 parent 4742b53
commit 891112e
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 7 deletions.
diff --git a/src/.env.local b/src/.env.local
@@ -13,4 +13,5 @@ INST_DB_USER=fi
 INST_DB_PWD=fi
 INST_DB_HOST=localhost:5432
 INST_DB_SCHEMA=public
-INST_CONN=postgresql+asyncpg://${INST_DB_USER}:${INST_DB_PWD}@${INST_DB_HOST}/${INST_DB_NAME}
+INST_CONN=postgresql+asyncpg://${INST_DB_USER}:${INST_DB_PWD}@${INST_DB_HOST}/${INST_DB_NAME}
+JWT_OPTS=verify_at_hash:False,verify_aud:False,verify_iss:False
diff --git a/src/.env.template b/src/.env.template
@@ -13,4 +13,5 @@ INST_DB_USER=
 INST_DB_PWD=
 INST_DB_HOST=
 INST_DB_SCHEMA=
-INST_CONN=postgresql+asyncpg://${INST_DB_USER}:${INST_DB_PWD}@${INST_DB_HOST}/${INST_DB_NAME}
+INST_CONN=postgresql+asyncpg://${INST_DB_USER}:${INST_DB_PWD}@${INST_DB_HOST}/${INST_DB_NAME}
+JWT_OPTS=
diff --git a/src/oauth2/oauth2_admin.py b/src/oauth2/oauth2_admin.py
@@ -1,3 +1,4 @@
+import ast
 from http import HTTPStatus
 import logging
 import os
@@ -12,6 +13,28 @@
 log = logging.getLogger(__name__)
 
 
+def get_jwt_opts(opts_string: str) -> Dict[str, bool | int]:
+    """
+    Parses out the opts_string into JWT options dictionary.
+
+    Args:
+        opts_string (str): comma separated key value pairs in the form of "key1:value1,key2:value2", valid options can be found here:
+        https://github.com/mpdavis/python-jose/blob/4b0701b46a8d00988afcc5168c2b3a1fd60d15d8/jose/jwt.py#L81
+
+    Returns:
+        dict: dictionary of options supported by jwt, mentioned in link above
+    """
+    jwt_opts = {}
+    pairs = opts_string.split(",")
+    for pair in pairs:
+        [key, value] = pair.split(":", 1)
+        jwt_opts[key] = ast.literal_eval(value)
+    return jwt_opts
+
+
+JWT_OPTS = get_jwt_opts(os.getenv("JWT_OPTS", ""))
+
+
 class OAuth2Admin:
     def __init__(self) -> None:
         self._keys = None
@@ -30,11 +53,7 @@ def get_claims(self, token: str) -> Dict[str, str] | None:
                 key=self._get_keys(),
                 issuer=os.getenv("KC_REALM_URL"),
                 audience=os.getenv("AUTH_CLIENT"),
-                options={
-                    "verify_at_hash": False,
-                    "verify_aud": False,
-                    "verify_iss": False,
-                },
+                options=JWT_OPTS,
             )
         except jose.ExpiredSignatureError:
             pass