Version 0.1.0

Remove usage of all globals (except permissions) and add unit tests
jacobsvante · Feb 10, 2020 · 8ee6d23 · 8ee6d23
1 parent c74810d
commit 8ee6d23
Show file tree

Hide file tree

Showing 25 changed files with 697 additions and 225 deletions.
diff --git a/.gitignore b/.gitignore
@@ -1,3 +1,5 @@
+/.coverage
+/.pytest_cache
 /build
 /dist
 *.egg-info

diff --git a/.isort.cfg b/.isort.cfg
@@ -4,4 +4,4 @@ include_trailing_comma = True
 force_grid_wrap = 0
 use_parentheses = True
 line_length = 88
-known_third_party = 
+known_third_party = aiohttp,aioresponses,cryptography,fastapi,jwt,pydantic,pytest,setuptools,starlette
diff --git a/README.md b/README.md
@@ -4,15 +4,15 @@ Implements authentication and authorization as dependencies in FastAPI.
 
 ## Features
 
-- Authentication via JWT-based OAuth 2 access tokens and Basic Auth
+- Authentication via JWT-based OAuth 2 access tokens and via Basic Auth
 - Pydantic-based `User` model for authenticated and anonymous users
 - Sub-classable `UserPermission` dependency to check against the `permissions` attribute returned in OAuth 2 access tokens
+- Able to extract user info from access tokens via OpenID Connect
 
 ## Limitations
 
 - Only supports validating access tokens using public keys from a JSON Web Key Set (JWKS) endpoint. I.e. for use with external identity providers such as Auth0 and ORY Hydra.
-- Clients authenticating with Basic Auth and OAuth 2 *Client Credentials* are always granted all permissions (NOTE: A Client Credentials token is only detected if the `gty` attribute is set to `client-credentials`, which is a non-standardized attribute provided by Auth0. PRs are welcome for other identity providers)
-- All other clients authenticating - i.e. OAuth2 that is not a Client Credentials token will only have the permissions specified in the `permissions` attribute.
+- Permissions can only be picked up automatically from OAuth2 tokens, from the non-standard `permissions` list attribute (Auth0 provides this, maybe other identity providers as well). For all other use cases, `permission_overrides` must be used. For example if there's a basic auth user called `user1` you can set `permission_overrides={"user1": ["*"]}` to give the user access to all permissions, or `permission_overrides={"user1": ["products:create"]}` to only assign `user1` with the permission `products:create`.
 
 
 ## Installation
@@ -24,3 +24,7 @@ pip install fastapi-security
 ## Usage examples
 
 Examples on how to use [can be found here](/examples).
+
+## TODO
+
+- Write more tests
diff --git a/examples/app1/README.md b/examples/app1/README.md
@@ -7,5 +7,6 @@ pip install fastapi-security uvicorn
 export BASIC_AUTH_CREDENTIALS='[{"username": "user1", "password": "test"}]'
 export AUTH_JWKS_URL='https://my-auth0-tenant.eu.auth0.com/.well-known/jwks.json'
 export AUTH_AUDIENCES='["my-audience"]'
+export PERMISSION_OVERRIDES='{"user1": ["products:create"]}'
 uvicorn app1:app
 ```
diff --git a/examples/app1/app.py b/examples/app1/app.py
@@ -3,65 +3,48 @@
 
 from fastapi import Depends, FastAPI
 
-from fastapi_security import (
-    User,
-    UserPermission,
-    UserWithPermissions,
-    basic_auth,
-    get_authenticated_user_or_401,
-    get_user,
-    get_user_with_info,
-    oauth2_jwt,
-    oidc_discovery
-)
+from fastapi_security import FastAPISecurity, User, UserPermission
 
 from . import db
 from .models import Product
 from .settings import get_settings
 
 app = FastAPI()
 
-logger = logging.getLogger(__name__)
+settings = get_settings()
 
-create_product_perm = UserPermission("products:create")
+security = FastAPISecurity()
 
+security.init(
+    app,
+    basic_auth_credentials=settings.basic_auth_credentials,
+    jwks_url=settings.oauth2_jwks_url,
+    audiences=settings.oauth2_audiences,
+    oidc_discovery_url=settings.oidc_discovery_url,
+    permission_overrides=settings.permission_overrides,
+)
 
-@app.on_event("startup")
-def enable_security():
-    settings = get_settings()
+logger = logging.getLogger(__name__)
 
-    if settings.basic_auth_credentials:
-        # Initialize basic auth (superusers with all permissions)
-        basic_auth.init(settings.basic_auth_credentials)
-    else:
-        logger.warning("Basic Auth disabled - not configured in settings")
-    if settings.auth_jwks_url and settings.auth_audiences:
-        # # Initialize OAuth 2.0 - user permissions are required for all flows
-        # # except Client Credentials
-        oauth2_jwt.init(settings.auth_jwks_url, audiences=settings.auth_audiences)
-    else:
-        logger.warning("OAuth2 disabled - not configured in settings")
-    if settings.oidc_discovery_url and oauth2_jwt.is_configured():
-        oidc_discovery.init(settings.oidc_discovery_url)
-    else:
-        logger.info("OIDC Discovery disabled - not configured in settings")
+create_product_perm = UserPermission("products:create")
 
 
 @app.get("/users/me")
-async def get_user_details(user: User = Depends(get_user_with_info)):
+async def get_user_details(user: User = Depends(security.user_with_info)):
     """Return user details, regardless of whether user is authenticated or not"""
     return user.without_access_token()
 
 
 @app.get("/users/me/permissions", response_model=List[str])
-def get_user_permissions(user: User = Depends(get_authenticated_user_or_401),):
+def get_user_permissions(user: User = Depends(security.authenticated_user_or_401)):
     """Return user permissions or HTTP401 if not authenticated"""
     return user.permissions
 
 
 @app.post("/products", response_model=Product)
 async def create_product(
-    product: Product, user: User = Depends(UserWithPermissions(create_product_perm)),
+    product: Product,
+    user: User = Depends(security.user_with_permissions(create_product_perm)),
 ):
     """Create product
 

diff --git a/examples/app1/settings.py b/examples/app1/settings.py
@@ -1,5 +1,5 @@
 from functools import lru_cache
-from typing import List
+from typing import Dict, List
 
 from fastapi.security import HTTPBasicCredentials
 from pydantic import BaseSettings
@@ -8,10 +8,11 @@
 
 
 class _Settings(BaseSettings):
-    auth_jwks_url: str = None
-    auth_audiences: List[str] = None
+    oauth2_jwks_url: str = None  # TODO: This could be retrieved from OIDC discovery URL
+    oauth2_audiences: List[str] = None
     basic_auth_credentials: List[HTTPBasicCredentials] = None
     oidc_discovery_url: str = None
+    permission_overrides: Dict[str, List[str]] = None
 
 
 @lru_cache()

diff --git a/fastapi_security/__init__.py b/fastapi_security/__init__.py
@@ -1,7 +1,8 @@
+from .api import *  # noqa
 from .basic import *  # noqa
-from .dependencies import *  # noqa
 from .entities import *  # noqa
-from .jwt import *  # noqa
+from .oauth2 import *  # noqa
 from .oidc import *  # noqa
 from .permissions import *  # noqa
+from .registry import *  # noqa
 from .schemes import *  # noqa
diff --git a/fastapi_security/api.py b/fastapi_security/api.py
@@ -0,0 +1,216 @@
+import logging
+from typing import Callable, Dict, List, Optional, Tuple
+
+from fastapi import Depends, FastAPI, HTTPException
+from fastapi.security import HTTPBasicCredentials
+from fastapi.security.http import HTTPAuthorizationCredentials
+
+from . import registry
+from .basic import BasicAuthValidator
+from .entities import AuthMethod, User, UserAuth, UserInfo
+from .exceptions import AuthNotConfigured
+from .oauth2 import Oauth2JwtAccessTokenValidator
+from .oidc import OpenIdConnectDiscovery
+from .permissions import UserPermission
+from .schemes import http_basic_scheme, jwt_bearer_scheme
+
+logger = logging.getLogger(__name__)
+
+__all__ = ("FastAPISecurity",)
+
+
+class FastAPISecurity:
+    """FastAPI Security main class, to be instantiated by users of the package
+
+    Must be initialized after object creation via the `init()` method.
+    """
+
+    def __init__(self):
+        self.basic_auth = BasicAuthValidator()
+        self.oauth2_jwt = Oauth2JwtAccessTokenValidator()
+        self.oidc_discovery = OpenIdConnectDiscovery()
+        self._permission_overrides = None
+
+    def init(
+        self,
+        app: FastAPI,
+        basic_auth_credentials: List[HTTPBasicCredentials] = None,
+        permission_overrides: Dict[str, List[str]] = None,
+        jwks_url: str = None,
+        audiences: List[str] = None,
+        oidc_discovery_url: str = None,
+    ):
+        self._permission_overrides = permission_overrides
+
+        if basic_auth_credentials:
+            # Initialize basic auth (superusers with all permissions)
+            self.basic_auth.init(basic_auth_credentials)
+
+        if jwks_url:
+            # # Initialize OAuth 2.0 - user permissions are required for all flows
+            # # except Client Credentials
+            self.oauth2_jwt.init(jwks_url, audiences=audiences or [])
+
+        if oidc_discovery_url and self.oauth2_jwt.is_configured():
+            self.oidc_discovery.init(oidc_discovery_url)
+
+    @property
+    def user(self) -> Callable:
+        """Dependency that returns User object, authenticated or not"""
+
+        async def dependency(user_auth: UserAuth = Depends(self._user_auth)):
+            return User(auth=user_auth)
+
+        return dependency
+
+    @property
+    def authenticated_user_or_401(self) -> Callable:
+        """Dependency that returns User object if authenticated,
+        otherwise raises HTTP401
+        """
+
+        async def dependency(user_auth: UserAuth = Depends(self._user_auth_or_401)):
+            return User(auth=user_auth)
+
+        return dependency
+
+    @property
+    def user_with_info(self) -> Callable:
+        """Dependency that returns User object with user info, authenticated or not"""
+
+        async def dependency(user_auth: UserAuth = Depends(self._user_auth)):
+            if user_auth.is_oauth2():
+                info = await self.oidc_discovery.get_user_info(user_auth.access_token)
+            else:
+                info = UserInfo.make_dummy()
+            return User(auth=user_auth, info=info)
+
+        return dependency
+
+    @property
+    def authenticated_user_with_info_or_401(self) -> Callable:
+        """Dependency that returns User object along with user info if authenticated,
+        otherwise raises HTTP401
+        """
+
+        async def dependency(user_auth: UserAuth = Depends(self._user_auth_or_401)):
+            if user_auth.is_oauth2():
+                info = await self.oidc_discovery.get_user_info(user_auth.access_token)
+            else:
+                info = UserInfo.make_dummy()
+            return User(auth=user_auth, info=info)
+
+        return dependency
+
+    def has_permission(self, permission: UserPermission) -> Callable:
+        """Dependency that raises HTTP403 if the user is missing the given permission
+        """
+
+        async def dependency(
+            user: User = Depends(self.authenticated_user_or_401),
+        ) -> User:
+            self._has_permission_or_raise_forbidden(user, permission)
+            return user
+
+        return dependency
+
+    def user_with_permissions(
+        self, *permissions: Tuple[UserPermission, ...]
+    ) -> Callable:
+        """Dependency that returns the user if it has the given permissions, otherwise
+        raises HTTP403
+        """
+
+        async def dependency(
+            user: User = Depends(self.authenticated_user_or_401),
+        ) -> User:
+            for perm in permissions:
+                self._has_permission_or_raise_forbidden(user, perm)
+            return user
+
+        return dependency
+
+    @property
+    def _user_auth(self) -> Callable:
+        """Dependency that returns UserAuth object if authentication was successful"""
+
+        async def dependency(
+            bearer_credentials: HTTPAuthorizationCredentials = Depends(
+                jwt_bearer_scheme
+            ),
+            http_credentials: HTTPAuthorizationCredentials = Depends(http_basic_scheme),
+        ) -> Optional[UserAuth]:
+            if not any(
+                [self.oauth2_jwt.is_configured(), self.basic_auth.is_configured()]
+            ):
+
+                raise AuthNotConfigured()
+
+            if bearer_credentials is not None:
+                bearer_token = bearer_credentials.credentials
+                access_token = await self.oauth2_jwt.parse(bearer_token)
+                if access_token:
+                    return self._maybe_override_permissions(
+                        UserAuth.from_jwt_access_token(access_token)
+                    )
+            elif http_credentials is not None and self.basic_auth.is_configured():
+                if self.basic_auth.validate(http_credentials):
+                    return self._maybe_override_permissions(
+                        UserAuth(
+                            subject=http_credentials.username,
+                            auth_method=AuthMethod.basic_auth,
+                        )
+                    )
+
+            return UserAuth.make_anonymous()
+
+        return dependency
+
+    @property
+    def _user_auth_or_401(self) -> Callable:
+        """Dependency that returns UserAuth object on success, or raises HTTP401"""
+
+        async def dependency(
+            user_auth: UserAuth = Depends(self._user_auth),
+            http_credentials: HTTPAuthorizationCredentials = Depends(http_basic_scheme),
+        ):
+
+            if user_auth and user_auth.is_authenticated():
+                return user_auth
+
+            if self.basic_auth.is_configured() and http_credentials is not None:
+                www_authenticate_header_val = "Basic"
+            else:
+                www_authenticate_header_val = "Bearer"
+
+            raise HTTPException(
+                status_code=401,
+                detail="Could not validate credentials",
+                headers={"WWW-Authenticate": www_authenticate_header_val},
+            )
+
+        return dependency
+
+    def _has_permission_or_raise_forbidden(self, user: User, perm: UserPermission):
+        if not user.has_permission(perm.identifier):
+            self._raise_forbidden(perm.identifier)
+
+    def _raise_forbidden(self, required_permission: str):
+        raise HTTPException(
+            403, detail=f"Missing required permission {required_permission}",
+        )
+
+    def _maybe_override_permissions(self, user_auth: UserAuth) -> UserAuth:
+        overrides = (self._permission_overrides or {}).get(user_auth.subject)
+
+        if overrides is None:
+            return user_auth
+
+        all_permissions = registry.get_all_permissions()
+
+        if "*" in overrides:
+            return user_auth.with_permissions(all_permissions)
+        else:
+            return user_auth.with_permissions(
+                [p for p in overrides if p in all_permissions]
+            )