Add tests for auto_error=False

Author: JonasKs

demo_project/api/dependencies.py

@@ -6,7 +6,7 @@
 from fastapi import Depends
 from fastapi.security.api_key import APIKeyHeader
 
-from fastapi_azure_auth import SingleTenantAzureAuthorizationCodeBearer
+from fastapi_azure_auth import MultiTenantAzureAuthorizationCodeBearer, SingleTenantAzureAuthorizationCodeBearer
 from fastapi_azure_auth.exceptions import InvalidAuth
 from fastapi_azure_auth.user import User
 
@@ -50,7 +50,7 @@ async def __call__(self, tid: str) -> str:
             # logic to find your allowed tenants and it's issuers here
             # (This example cache in memory for 1 hour)
             self.tid_to_iss = {
-                'intility_tenant': 'intility_tenant',
+                'intility_tenant_id': 'https://login.microsoftonline.com/intility_tenant/v2.0',
             }
         try:
             return self.tid_to_iss[tid]
@@ -59,12 +59,15 @@ async def __call__(self, tid: str) -> str:
             raise InvalidAuth('You must be an Intility customer to access this resource')
 
 
-azure_scheme_auto_error_false = SingleTenantAzureAuthorizationCodeBearer(
+issuer_fetcher = IssuerFetcher()
+
+azure_scheme_auto_error_false = MultiTenantAzureAuthorizationCodeBearer(
     app_client_id=settings.APP_CLIENT_ID,
     scopes={
-        f'api://{settings.APP_CLIENT_ID}/user_impersonation': '**No client secret needed, leave blank**',
+        f'api://{settings.APP_CLIENT_ID}/user_impersonation': 'User impersonation',
     },
-    tenant_id=settings.TENANT_ID,
+    validate_iss=True,
+    iss_callable=issuer_fetcher,
     auto_error=False,
 )
 

demo_project/core/config.py

@@ -1,6 +1,6 @@
 from typing import List, Optional, Union
 
-from pydantic import AnyHttpUrl, BaseSettings, Field, HttpUrl, validator
+from pydantic import AnyHttpUrl, BaseSettings, Field, HttpUrl
 
 
 class AzureActiveDirectory(BaseSettings):
@@ -18,29 +18,9 @@ class Settings(AzureActiveDirectory):
     # "http://localhost:8080", "http://local.dockertoolbox.tiangolo.com"]'
     BACKEND_CORS_ORIGINS: List[Union[str, AnyHttpUrl]] = ['http://localhost:8000']
 
-    @validator('BACKEND_CORS_ORIGINS', pre=True)
-    def assemble_cors_origins(cls, value: Union[str, List[str]]) -> Union[List[str], str]:  # pragma: no cover
-        """
-        Validate cors list
-        """
-        if isinstance(value, str) and not value.startswith('['):
-            return [i.strip() for i in value.split(',')]
-        elif isinstance(value, (list, str)):
-            return value
-        raise ValueError(value)
-
     PROJECT_NAME: str = 'My Project'
     SENTRY_DSN: Optional[HttpUrl] = None
 
-    @validator('SENTRY_DSN', pre=True)
-    def sentry_dsn_can_be_blank(cls, value: str) -> Optional[str]:  # pragma: no cover
-        """
-        Validate sentry DSN
-        """
-        if not value:
-            return None
-        return value
-
     class Config:  # noqa
         env_file = 'demo_project/.env'
         env_file_encoding = 'utf-8'

fastapi_azure_auth/auth.py

@@ -1,8 +1,7 @@
-import asyncio
 import inspect
 import logging
 from collections.abc import Callable
-from typing import Any, Literal, Optional
+from typing import Any, Awaitable, Literal, Optional
 
 from fastapi.exceptions import HTTPException
 from fastapi.security import OAuth2AuthorizationCodeBearer, SecurityScopes
@@ -27,7 +26,7 @@ def __init__(
         scopes: Optional[dict[str, str]] = None,
         multi_tenant: bool = False,
         validate_iss: bool = True,
-        iss_callable: Optional[Callable[..., Any]] = None,
+        iss_callable: Optional[Callable[[str], Awaitable[str]]] = None,
         token_version: Literal[1, 2] = 2,
         openid_config_use_app_id: bool = False,
         openapi_authorization_url: Optional[str] = None,
@@ -55,7 +54,7 @@ def __init__(
         :param validate_iss: bool
         **Only used for multi-tenant applications**
             Whether to validate the token `iss` (issuer) or not. This can be skipped to allow anyone to log in.
-        :param iss_callable: Callable
+        :param iss_callable: Async Callable
         **Only used for multi-tenant application**
             Async function that has to accept a `tid` (tenant ID) and return a `iss` (issuer) or
              raise an InvalidIssuer exception
@@ -81,8 +80,6 @@ def __init__(
         if multi_tenant:
             if validate_iss and not callable(iss_callable):
                 raise RuntimeError('`validate_iss` is enabled, so you must provide an `iss_callable`')
-            elif iss_callable and not asyncio.iscoroutinefunction(iss_callable):
-                raise RuntimeError('`iss_callable` must be a coroutine')
             elif iss_callable and 'tid' not in inspect.signature(iss_callable).parameters.keys():
                 raise RuntimeError('`iss_callable` must accept `tid` as an argument')
 
@@ -277,7 +274,7 @@ def __init__(
         auto_error: bool = True,
         scopes: Optional[dict[str, str]] = None,
         validate_iss: bool = True,
-        iss_callable: Optional[Callable[..., Any]] = None,
+        iss_callable: Optional[Callable[[str], Awaitable[str]]] = None,
         openid_config_use_app_id: bool = False,
         openapi_authorization_url: Optional[str] = None,
         openapi_token_url: Optional[str] = None,
@@ -299,7 +296,7 @@ def __init__(
 
         :param validate_iss: bool
             Whether to validate the token `iss` (issuer) or not. This can be skipped to allow anyone to log in.
-        :param iss_callable: Callable
+        :param iss_callable: Async Callable
             Async function that has to accept a `tid` (tenant ID) and return a `iss` (issuer) or
              raise an InvalidIssuer exception
             This is required when validate_iss is set to `True`.

tests/multi_tenant/conftest.py

@@ -23,7 +23,7 @@ async def issuer_fetcher(tid):
         iss_callable=issuer_fetcher,
     )
     app.dependency_overrides[azure_scheme] = azure_scheme_overrides
-    yield azure_scheme
+    yield
 
 
 @pytest.fixture

tests/multi_tenant/multi_auth/README.md

@@ -0,0 +1 @@
+This folder contains tests to ensure that multiple authentication schemes will work together.

tests/multi_auth/__init__.py tests/multi_tenant/multi_auth/__init__.pytests/multi_auth/__init__.py renamed to tests/multi_tenant/multi_auth/__init__.py

@@ -0,0 +1,34 @@
+import pytest
+from demo_project.main import app
+from httpx import AsyncClient
+from tests.utils import build_access_token, build_access_token_expired
+
+
+@pytest.mark.anyio
+async def test_normal_azure_user_valid_token(multi_tenant_app, mock_openid_and_keys, freezer):
+    access_token = build_access_token(version=2)
+    async with AsyncClient(app=app, base_url='http://test', headers={'Authorization': 'Bearer ' + access_token}) as ac:
+        response = await ac.get('api/v1/hello-multi-auth')
+        assert response.json() == {'api_key': False, 'azure_auth': True}
+
+
+@pytest.mark.anyio
+async def test_api_key_valid_key(multi_tenant_app, mock_openid_and_keys, freezer):
+    async with AsyncClient(app=app, base_url='http://test', headers={'TEST-API-KEY': 'JonasIsCool'}) as ac:
+        response = await ac.get('api/v1/hello-multi-auth')
+        assert response.json() == {'api_key': True, 'azure_auth': False}
+
+
+@pytest.mark.anyio
+async def test_normal_azure_user_but_invalid_token(multi_tenant_app, mock_openid_and_keys, freezer):
+    access_token = build_access_token_expired(version=2)
+    async with AsyncClient(app=app, base_url='http://test', headers={'Authorization': 'Bearer ' + access_token}) as ac:
+        response = await ac.get('api/v1/hello-multi-auth')
+        assert response.json() == {'detail': 'You must either provide a valid bearer token or API key'}
+
+
+@pytest.mark.anyio
+async def test_api_key_but_invalid_key(multi_tenant_app, mock_openid_and_keys, freezer):
+    async with AsyncClient(app=app, base_url='http://test', headers={'TEST-API-KEY': 'JonasIsNotCool'}) as ac:
+        response = await ac.get('api/v1/hello-multi-auth')
+        assert response.json() == {'detail': 'You must either provide a valid bearer token or API key'}

tests/multi_auth/test_auto_error.py tests/multi_tenant/multi_auth/conftest.pytests/multi_auth/test_auto_error.py renamed to tests/multi_tenant/multi_auth/conftest.py

@@ -7,9 +7,7 @@ async def iss_callable_do_not_accept_tid_argument(t):
     pass
 
 
-@pytest.mark.parametrize(
-    'iss_callable', [None, '', True, False, 1, lambda tid: True, iss_callable_do_not_accept_tid_argument]
-)
+@pytest.mark.parametrize('iss_callable', [None, '', True, False, 1, iss_callable_do_not_accept_tid_argument])
 def test_non_accepted_issue_fetcher_given(iss_callable):
     with pytest.raises(RuntimeError):
         MultiTenantAzureAuthorizationCodeBearer(app_client_id='some id', validate_iss=True, iss_callable=iss_callable)

tests/multi_tenant/multi_auth/test_auto_error.py

@@ -34,7 +34,6 @@ def single_tenant_app(token_version):
             token_version=1,
         )
         app.dependency_overrides[azure_scheme] = azure_scheme_overrides
-        yield azure_scheme
     elif token_version == 2:
         azure_scheme_overrides = SingleTenantAzureAuthorizationCodeBearer(
             app_client_id=settings.APP_CLIENT_ID,
@@ -45,7 +44,7 @@ def single_tenant_app(token_version):
             token_version=2,
         )
         app.dependency_overrides[azure_scheme] = azure_scheme_overrides
-        yield azure_scheme
+    yield
 
 
 @pytest.fixture