Merge pull request #27 from djpugh/feature/get-access-token

docs/source/usage.rst

@@ -210,3 +210,16 @@ Tools like Postman allow you to configure authentication via oauth - this shows
    :alt: Overview of authenticating for postman
 
    An example of how to configure client credentials (using another app registration) for postman - replace the {tenant} and {appid} info, along with the client id and client secret
+
+
+Accessing User Tokens/View
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+There are two routes that are automatically added to this, the ``/me`` and ``/me/getToken`` routes. The ``/me`` route provides a summary of the current user, and enables them to get a bearer token from Azure AD.
+The ``/me/token`` endpoint provides that same token (for the logged in user) in a JSON object
+
+.. warning::
+
+    To get the token, this is primarily an interactive method, as it requires caching the token through the UI session based login approach, so it can fail intermittently depending on if the user has logged in recently.
+
+This can be disabled by setting the ``config.routing.user_path`` to ``None`` or ``''``. 

requirements.txt

@@ -3,7 +3,7 @@ aiofiles
 jinja2
 fastapi
 authlib
-cryptography
+cryptography>3.2
 python-multipart
 requests
 pydantic[dotenv]

setup.cfg

@@ -38,6 +38,11 @@ python_requires = >=3
      *.whl
      *.html
      *.css
+     *.woff
+     *.otf
+     *.svg
+     *.eot
+     *.ttf
      *.png
      *.css.map
 

src/fastapi_aad_auth/auth.py

@@ -4,19 +4,19 @@
 from pathlib import Path
 from typing import Any, Dict, List
 
-from fastapi import FastAPI
+from fastapi import Depends, FastAPI
 from starlette.authentication import requires
 from starlette.middleware.authentication import AuthenticationError, AuthenticationMiddleware
 from starlette.middleware.sessions import SessionMiddleware
 from starlette.requests import Request
-from starlette.responses import RedirectResponse, Response
+from starlette.responses import JSONResponse, RedirectResponse, Response
 from starlette.routing import Mount, request_response, Route
 from starlette.staticfiles import StaticFiles
 from starlette.templating import Jinja2Templates
 
 from fastapi_aad_auth.config import Config
 from fastapi_aad_auth.errors import base_error_handler, ConfigurationError
-from fastapi_aad_auth.oauth import AADOAuthBackend
+from fastapi_aad_auth.oauth import AADOAuthBackend, AuthenticationState
 
 
 logger = logging.getLogger(__name__)
@@ -198,24 +198,68 @@ def build_auth_ui(self, context: Dict[str, Any] = None):
         Keyword Args:
             contex: a dicitionary of predefined parameters to pass to the Jinja2 Login UI template
         """
+        login_template_path = Path(self.config.login_ui.template_file)
+        user_template_path = Path(self.config.login_ui.user_template_file)
+        login_templates = Jinja2Templates(directory=str(login_template_path.parent))
+        user_templates = Jinja2Templates(directory=str(user_template_path.parent))
         if context is None:
             context = {}
-        template_path = Path(self.config.login_ui.template_file)
-        templates = Jinja2Templates(directory=str(template_path.parent))
 
         async def login(request: Request, *args, **kwargs):
+            nonlocal context
+            view_context = context.copy()  # type: ignore
             if not self.oauth_backend.enabled or request.user.is_authenticated:
                 # This is authenticated so go straight to the homepage
                 return RedirectResponse(self.config.routing.home_path)
-            context['request'] = request  # type: ignore
-            if 'login' not in context or context['login'] is None:  # type: ignore
+            view_context['request'] = request  # type: ignore
+            if 'login' not in view_context or view_context['login'] is None:  # type: ignore
                 post_redirect = self.oauth_backend.authenticator.pop_post_auth_redirect(request)
-                context['login'] = self.oauth_backend.authenticator.get_login_button(self.config.routing.login_path, post_redirect)  # type: ignore
-            return templates.TemplateResponse(template_path.name, context)  # type: ignore
+                view_context['login'] = self.oauth_backend.authenticator.get_login_button(self.config.routing.login_path, post_redirect)  # type: ignore
+            return login_templates.TemplateResponse(login_template_path.name, view_context)  # type: ignore
 
         routes = [Route(self.config.routing.landing_path, endpoint=login, methods=['GET'], name='login'),
                   Mount(self.config.login_ui.static_path, StaticFiles(directory=str(self.config.login_ui.static_directory)), name='static-login')]
 
+        if self.config.routing.user_path:
+
+            @self.auth_required()
+            async def get_user(request: Request):
+                nonlocal context
+                view_context = context.copy()  # type: ignore
+                logger.debug(f'Getting token for {request.user}')
+                view_context['request'] = request
+                if self.oauth_backend.enabled:
+                    logger.debug(f'Auth {request.auth}')
+                    try:
+                        view_context['user'] = self.oauth_backend.authenticator.get_user_from_request(request)
+                        view_context['token'] = self.oauth_backend.authenticator.get_access_token(view_context['user'])
+                    except ValueError:
+                        return self.oauth_backend.authenticator.process_login_request(request, force=True, redirect=request.url.path)
+                else:
+                    logger.debug('Auth not enabled')
+                    view_context['token'] = None
+                return user_templates.TemplateResponse(user_template_path.name, view_context)
+
+            async def get_token(request: Request, auth_state: AuthenticationState = Depends(self.api_auth_scheme)):
+                if not isinstance(auth_state, AuthenticationState):
+                    if hasattr(request.user, 'username'):
+                        user = request.user
+                    else:
+                        auth_state = await self.api_auth_scheme(request)
+                        user = auth_state.user
+                if hasattr(user, 'username'):  # type: ignore
+                    try:
+                        return JSONResponse(self.oauth_backend.authenticator.get_access_token(user))   # type: ignore
+                    except ValueError:
+                        if any([u in request.headers['user-agent'] for u in ['Mozilla', 'Gecko', 'Trident', 'WebKit', 'Presto', 'Edge', 'Blink']]):
+                            return self.oauth_backend.authenticator.process_login_request(request, force=True, redirect=request.url.path)
+                        else:
+                            return JSONResponse('Unable to access token as user has not authenticated via session')
+                return RedirectResponse(f'{self.config.routing.landing_path}?redirect=/me/token')
+
+            routes += [Route(self.config.routing.user_path, endpoint=get_user, methods=['GET'], name='user'),
+                       Route(f'{self.config.routing.user_path}/token', endpoint=get_token, methods=['GET'], name='get-token')]
+
         return routes
 
     @property

src/fastapi_aad_auth/config.py

@@ -62,11 +62,11 @@ class RoutingConfig(BaseSettings):
     login_redirect_path: str = Field('/login/oauth/redirect', description="Path for handling the AAD redirect call", env='FASTAPI_AUTH_LOGIN_REDIRECT_ROUTE')
     logout_path: str = Field('/logout', description="Path for processing a logout request", env='FASTAPI_AUTH_LOGOUT_ROUTE')
     landing_path: str = Field('/login', description="Path for the login UI page", env='FASTAPI_AUTH_LOGIN_UI_ROUTE')
+    user_path: Optional[str] = Field('/me', description="Path for getting the user view", env='FASTAPI_AUTH_USER_ROUTE')
     home_path: str = Field('/', description="Path for the application home page (default redirect if none provided)",
                            env='APP_HOME_ROUTE')
     post_logout_path: str = Field(None, description="Path for the redirect post logout - defaults to the landing path if not provided",
                                   env='FASTAPI_AUTH_POST_LOGOUT_ROUTE')
-    # TODO: Add an API Token Route to get a bearer token interactively.
 
     class Config:  # noqa D106
         env_file = '.env'
@@ -92,6 +92,9 @@ class LoginUIConfig(BaseSettings):
     error_template_file: FilePath = Field(resource_filename('fastapi_aad_auth.ui', 'error.html'),
                                           description="The jinja2 template to use",
                                           env='FASTAPI_AUTH_LOGIN_ERROR_TEMPLATE_FILE')
+    user_template_file: FilePath = Field(resource_filename('fastapi_aad_auth.ui', 'user.html'),
+                                         description="The jinja2 template to use",
+                                         env='FASTAPI_AUTH_USER_TEMPLATE_FILE')
     static_directory: DirectoryPath = Field(resource_filename('fastapi_aad_auth.ui', 'static'),
                                             description="Static path for the Login UI",
                                             env='FASTAPI_AUTH_LOGIN_STATIC_DIR')

src/fastapi_aad_auth/oauth/authenticators/session.py

@@ -4,6 +4,7 @@
 
 import msal
 from pkg_resources import resource_string
+from starlette.requests import Request
 from starlette.responses import RedirectResponse
 
 from fastapi_aad_auth.errors import ConfigurationError
@@ -38,16 +39,16 @@ def redirect_to_provider_login(self, auth_state, request):
     def _get_authorization_url(self, request, session_state):
         raise NotImplementedError('Implement in specific subclass')
 
-    def process_login_request(self, request):
+    def process_login_request(self, request, force=False, redirect='/'):
         """Process the provider login request."""
         logger.debug(f'Logging in - request url {request.url}')
         auth_state = self._session_validator.get_state_from_session(request)
-        if auth_state.is_authenticated():
+        if auth_state.is_authenticated() and not force:
             logger.debug(f'Authenticated - redirecting {auth_state}')
             response = self.redirect_if_authenticated(auth_state)
         else:
             # Set the redirect parameter here
-            self._session_validator.set_post_auth_redirect(request, request.query_params.get('redirect', '/'))
+            self._session_validator.set_post_auth_redirect(request, request.query_params.get('redirect', redirect))
             logger.debug(f'No Auth state - redirecting to provider login {auth_state}')
             response = self.redirect_to_provider_login(auth_state, request)
         return response
@@ -69,6 +70,22 @@ def process_login_callback(self, request):
     def _process_code(self, request, auth_state, code):
         raise NotImplementedError('Implement in subclass')
 
+    def get_access_token(self, user):
+        """Get the access token for the user."""
+        raise NotImplementedError('Implement in subclass')
+
+    def get_access_token_from_request(self, request: Request):
+        """Get the access token from a request object."""
+        auth_state = self._session_validator.get_state_from_session(request)
+        if auth_state.is_authenticated():
+            return self.get_access_token(auth_state.user)['access_token']
+        return None
+
+    def get_user_from_request(self, request: Request):
+        """Get the user from a request object."""
+        auth_state = self._session_validator.get_state_from_session(request)
+        return auth_state.user
+
     def _get_user_from_token(self, token, options=None):
         validated_claims = self._token_validator.validate_token(token, options=options)
         return self._token_validator._get_user_from_claims(validated_claims)
@@ -117,6 +134,7 @@ def __init__(
         self._redirect_uri = redirect_uri
         self._domain_hint = domain_hint
         self._prompt = prompt
+        self.client_id = client_id
         if scopes is None:
             scopes = [f'api://{self.client_id}']
         elif isinstance(scopes, str):
@@ -176,3 +194,22 @@ def get_login_button(self, url, post_redirect='/'):
         url = self._add_redirect_to_url(url, post_redirect)
         logo = base64.b64encode(resource_string('fastapi_aad_auth.oauth', 'ms-logo.png')).decode()
         return f'<a class="btn btn-lg btn-light btn-ms" href="{url}"><div class="row align-items-center justify-center login-ms"><img alt="Microsoft Logo" class="rounded splash-ms" src="data:image/png;base64,{logo}" />Sign in with Microsoft Work Account</div></a>'
+
+    def get_access_token(self, user):
+        """Get the access token for the user."""
+        result = None
+        account = None
+        if user.username:
+            account = self.msal_application.get_accounts(user.username)
+        if account:
+            account = account[0]
+            logger.info(account)
+            # This needs you to register the openid api
+            result = self.msal_application.acquire_token_silent_with_error(scopes=[f'api://{self.client_id}/openid'], account=account)
+            logger.info(result)
+        if result is None:
+            raise ValueError('Token not found')
+        else:
+            return {'token_type': result['token_type'],
+                    'expires_in': result['expires_in'],
+                    'access_token': result['access_token']}

src/fastapi_aad_auth/oauth/validators/session.py

@@ -40,7 +40,7 @@ def check(self, request):
 
     def pop_post_auth_redirect(self, request):
         """Clear post-authentication redirects."""
-        return request.session.pop(REDIRECT_KEY, '/')
+        return request.session.pop(REDIRECT_KEY, request.query_params.get('redirect', '/'))
 
     def set_post_auth_redirect(self, request, redirect='/'):
         """Set post-authentication redirects."""

src/fastapi_aad_auth/ui/static/css/cover.css

@@ -139,4 +139,11 @@ code{
   color: #fff;
   padding-left: 12px;
   padding-right: 12px;
-}
+}
+
+.table {
+  color: #fff;
+}
+#expires{
+  text-shadow: none;
+}