Feature/25 add jwt config (#35)

closes #25

---------

Co-authored-by: Hans Keeler <[email protected]>

pyproject.toml

@@ -8,7 +8,7 @@ packages = [{ include = "regtech-user-fi-management" }]
 
 [tool.poetry.dependencies]
 python = "^3.11"
-fastapi = "^0.97.0"
+fastapi = "^0.103.1"
 uvicorn = "^0.22.0"
 python-dotenv = "^1.0.0"
 python-keycloak = "^3.0.0"
@@ -18,6 +18,7 @@ python-jose = "^3.3.0"
 requests = "^2.31.0"
 asyncpg = "^0.27.0"
 alembic = "^1.12.0"
+pydantic-settings = "^2.0.3"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.0.278"
@@ -28,6 +29,7 @@ pytest-asyncio = "^0.21.1"
 aiosqlite = "^0.19.0"
 pytest-cov = "^4.1.0"
 pytest-mock = "^3.11.1"
+pytest-env = "^1.0.1"
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
@@ -43,6 +45,18 @@ addopts = [
   "-rfE",
 ]
 testpaths = ["tests"]
+env = [
+  "INST_CONN=postgresql+asyncpg://localhost",
+  "KC_URL=http://localhost",
+  "KC_REALM=",
+  "KC_ADMIN_CLIENT_ID=",
+  "KC_ADMIN_CLIENT_SECRET=",
+  "KC_REALM_URL=http://localhost",
+  "AUTH_URL=http://localhost",
+  "TOKEN_URL=http://localhost",
+  "CERTS_URL=http://localhost",
+  "AUTH_CLIENT=",
+]
 
 [tool.black]
 line-length = 120

src/.env.local

@@ -13,4 +13,7 @@ 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"
+JWT_OPTS_VERIFY_AUD="false"
+JWT_OPTS_VERIFY_ISS="false"

src/config.py

@@ -0,0 +1,61 @@
+import os
+from typing import Dict, Any
+
+from pydantic import TypeAdapter
+from pydantic.networks import HttpUrl, PostgresDsn
+from pydantic.types import SecretStr
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+JWT_OPTS_PREFIX = "jwt_opts_"
+
+env_files_to_load = [".env"]
+if os.getenv("ENV", "LOCAL") == "LOCAL":
+    env_files_to_load.append(".env.local")
+
+
+class Settings(BaseSettings):
+    inst_conn: PostgresDsn
+    inst_db_schema: str = "public"
+    auth_client: str
+    auth_url: HttpUrl
+    token_url: HttpUrl
+    certs_url: HttpUrl
+    kc_url: HttpUrl
+    kc_realm: str
+    kc_admin_client_id: str
+    kc_admin_client_secret: SecretStr
+    kc_realm_url: HttpUrl
+    jwt_opts: Dict[str, bool | int] = {}
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        self.set_jwt_opts()
+
+    def set_jwt_opts(self) -> None:
+        """
+        Converts `jwt_opts_` prefixed settings, and env vars into JWT options dictionary.
+        all options are boolean, with exception of 'leeway' being int
+        valid options can be found here:
+        https://github.com/mpdavis/python-jose/blob/4b0701b46a8d00988afcc5168c2b3a1fd60d15d8/jose/jwt.py#L81
+
+        Because we're using model_extra to load in jwt_opts as a dynamic dictionary,
+        normal env overrides does not take place on top of dotenv files,
+        so we're merging settings.model_extra with environment variables.
+        """
+        jwt_opts_adapter = TypeAdapter(int | bool)
+        self.jwt_opts = {
+            **self.parse_jwt_vars(jwt_opts_adapter, self.model_extra.items()),
+            **self.parse_jwt_vars(jwt_opts_adapter, os.environ.items()),
+        }
+
+    def parse_jwt_vars(self, type_adapter: TypeAdapter, setting_variables: Dict[str, Any]) -> Dict[str, bool | int]:
+        return {
+            key.lower().replace(JWT_OPTS_PREFIX, ""): type_adapter.validate_python(value)
+            for (key, value) in setting_variables
+            if key.lower().startswith(JWT_OPTS_PREFIX)
+        }
+
+    model_config = SettingsConfigDict(env_file=env_files_to_load, extra="allow")
+
+
+settings = Settings()

src/entities/engine/engine.py

@@ -1,14 +1,14 @@
-import os
 from sqlalchemy.ext.asyncio import (
     create_async_engine,
     async_sessionmaker,
     async_scoped_session,
 )
 from asyncio import current_task
+from config import settings
 
-DB_URL = os.getenv("INST_CONN")
-DB_SCHEMA = os.getenv("INST_DB_SCHEMA", "public")
-engine = create_async_engine(DB_URL, echo=True).execution_options(schema_translate_map={None: DB_SCHEMA})
+engine = create_async_engine(settings.inst_conn.unicode_string(), echo=True).execution_options(
+    schema_translate_map={None: settings.inst_db_schema}
+)
 SessionLocal = async_scoped_session(async_sessionmaker(engine, expire_on_commit=False), current_task)
 
 

src/entities/models/dto.py

@@ -1,4 +1,4 @@
-from typing import List, Dict, Any, Set, Optional
+from typing import List, Dict, Any, Set
 from pydantic import BaseModel
 from starlette.authentication import BaseUser
 
@@ -15,7 +15,7 @@ class FinancialInsitutionDomainDto(FinancialInsitutionDomainBase):
     lei: str
 
     class Config:
-        orm_mode = True
+        from_attributes = True
 
 
 class FinancialInstitutionBase(BaseModel):
@@ -26,7 +26,7 @@ class FinancialInstitutionDto(FinancialInstitutionBase):
     lei: str
 
     class Config:
-        orm_mode = True
+        from_attributes = True
 
 
 class FinancialInstitutionWithDomainsDto(FinancialInstitutionDto):
@@ -37,13 +37,13 @@ class DeniedDomainDto(BaseModel):
     domain: str
 
     class Config:
-        orm_mode = True
+        from_attributes = True
 
 
 class UserProfile(BaseModel):
     first_name: str
     last_name: str
-    leis: Optional[Set[str]]
+    leis: Set[str] | None = None
 
     def to_keycloak_user(self):
         return {"firstName": self.first_name, "lastName": self.last_name}

src/main.py

@@ -1,6 +1,4 @@
-import os
 import logging
-import env  # noqa: F401
 from http import HTTPStatus
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import JSONResponse
@@ -12,6 +10,8 @@
 
 from oauth2 import BearerTokenAuthBackend
 
+from config import settings
+
 log = logging.getLogger()
 
 app = FastAPI()
@@ -32,7 +32,9 @@ async def general_exception_handler(request: Request, exception: Exception) -> J
     )
 
 
-oauth2_scheme = OAuth2AuthorizationCodeBearer(authorizationUrl=os.getenv("AUTH_URL"), tokenUrl=os.getenv("TOKEN_URL"))
+oauth2_scheme = OAuth2AuthorizationCodeBearer(
+    authorizationUrl=settings.auth_url.unicode_string(), tokenUrl=settings.token_url.unicode_string()
+)
 
 app.add_middleware(AuthenticationMiddleware, backend=BearerTokenAuthBackend(oauth2_scheme))
 app.add_middleware(

src/oauth2/oauth2_admin.py

@@ -1,6 +1,5 @@
 from http import HTTPStatus
 import logging
-import os
 from typing import Dict, Any, Set
 
 import jose.jwt
@@ -9,17 +8,19 @@
 
 from keycloak import KeycloakAdmin, KeycloakOpenIDConnection, exceptions as kce
 
+from config import settings
+
 log = logging.getLogger(__name__)
 
 
 class OAuth2Admin:
     def __init__(self) -> None:
         self._keys = None
         conn = KeycloakOpenIDConnection(
-            server_url=os.getenv("KC_URL"),
-            realm_name=os.getenv("KC_REALM"),
-            client_id=os.getenv("KC_ADMIN_CLIENT_ID"),
-            client_secret_key=os.getenv("KC_ADMIN_CLIENT_SECRET"),
+            server_url=settings.kc_url.unicode_string(),
+            realm_name=settings.kc_realm,
+            client_id=settings.kc_admin_client_id,
+            client_secret_key=settings.kc_admin_client_secret.get_secret_value(),
         )
         self._admin = KeycloakAdmin(connection=conn)
 
@@ -28,20 +29,16 @@ def get_claims(self, token: str) -> Dict[str, str] | None:
             return jose.jwt.decode(
                 token=token,
                 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,
-                },
+                issuer=settings.kc_realm_url.unicode_string(),
+                audience=settings.auth_client,
+                options=settings.jwt_opts,
             )
         except jose.ExpiredSignatureError:
             pass
 
     def _get_keys(self) -> Dict[str, Any]:
         if self._keys is None:
-            response = requests.get(os.getenv("CERTS_URL"))
+            response = requests.get(settings.certs_url)
             self._keys = response.json()
         return self._keys
 

tests/app/test_config.py

@@ -0,0 +1,23 @@
+import pytest
+from config import Settings
+
+
+def test_jwt_opts_valid_values():
+    mock_config = {
+        "jwt_opts_test1": "true",
+        "jwt_opts_test2": "true",
+        "jwt_opts_test3": "12",
+    }
+    settings = Settings(**mock_config)
+    assert settings.jwt_opts == {"test1": True, "test2": True, "test3": 12}
+
+
+def test_jwt_opts_invalid_values():
+    mock_config = {
+        "jwt_opts_test1": "not a bool or int",
+        "jwt_opts_test2": "true",
+        "jwt_opts_test3": "12",
+    }
+    with pytest.raises(Exception) as e:
+        Settings(**mock_config)
+    assert "validation error" in str(e.value)