+ | Your Web API Calls another web API on behalf of the user (OBO) |
In roadmap.
diff --git a/docs/app-vs-api.rst b/docs/app-vs-api.rst
new file mode 100644
index 0000000..07de9d2
--- /dev/null
+++ b/docs/app-vs-api.rst
@@ -0,0 +1,30 @@
+.. note::
+
+ Web Application (a.k.a. website) and Web API are different scenarios,
+ both supported by the same Identity Auth component with different decorators.
+ Make sure you are using the right decorator for your scenario.
+
+.. list-table::
+ :header-rows: 1
+ :widths: 20 40 40
+
+ * - Aspects
+ - Web Application (a.k.a. website)
+ - Web API
+ * - **Definition**
+ - A complete solution that users interact with directly through their browsers.
+ - A back-end system that provides data (typically in JSON format) to front-end or other system.
+ * - **Functionality**
+ - | - Users interact with views (HTML user interfaces) and data.
+ | - Users sign in and establish their sessions.
+ - | - Does not return views (in HTML); only provides data.
+ | - Other systems (clients) hit its endpoints.
+ | - Clients presents a token to access your API.
+ | - Each request has no session. They are stateless.
+ * - **Identity component**
+ - Same ``Auth`` class
+ - Same ``Auth`` class
+ * - **Decorator to use**
+ - ``@auth.login_required``
+ - ``@auth.authorization_required``
+
diff --git a/docs/django-webapi.rst b/docs/django-webapi.rst
new file mode 100644
index 0000000..d826846
--- /dev/null
+++ b/docs/django-webapi.rst
@@ -0,0 +1,77 @@
+Identity for a Django Web API
+=============================
+
+.. include:: app-vs-api.rst
+
+Prerequisite
+------------
+
+Create a hello world web project in Django.
+
+You can use
+`Django's own tutorial, part 1 `_
+as a reference. What we need are basically these steps:
+
+#. ``django-admin startproject mysite``
+#. ``python manage.py migrate`` (Optinoal if your project does not use a database)
+#. ``python manage.py runserver localhost:5000``
+
+#. Now, add a new `mysite/views.py` file with an `index` view to your project.
+ For now, it can simply return a "hello world" page to any visitor::
+
+ from django.http import JsonResponse
+ def index(request):
+ return JsonResponse({"message": "Hello, world!"})
+
+Configuration
+-------------
+
+#. Install dependency by ``pip install identity[django]``
+
+#. Create an instance of the :py:class:`identity.django.Auth` object,
+ and assign it to a global variable inside your ``settings.py``::
+
+ import os
+ from identity.django import Auth
+ AUTH = Auth(
+ client_id=os.getenv('CLIENT_ID'),
+ ...=..., # See below on how to feed in the authority url parameter
+ )
+
+ .. include:: auth.rst
+
+
+Django Web API protected by an access token
+-------------------------------------------
+
+#. In your web project's ``views.py``, decorate some views with the
+ :py:func:`identity.django.Auth.authorization_required` decorator::
+
+ from django.conf import settings
+
+ @settings.AUTH.authorization_required(expected_scopes={
+ "your_scope_1": "api://your_client_id/your_scope_1",
+ "your_scope_2": "api://your_client_id/your_scope_2",
+ })
+ def index(request, *, context):
+ claims = context['claims']
+ # The user is uniquely identified by claims['sub'] or claims["oid"],
+ # claims['tid'] and/or claims['iss'].
+ return JsonResponse(
+ {"message": f"Data for {claims['sub']}@{claims['tid']}"}
+ )
+
+
+All of the content above are demonstrated in
+`this django web app sample `_.
+
+
+API for Django web projects
+---------------------------
+
+.. autoclass:: identity.django.Auth
+ :members:
+ :inherited-members:
+
+ .. automethod:: __init__
+
diff --git a/docs/django.rst b/docs/django.rst
index 3e4dcb0..7c2870d 100644
--- a/docs/django.rst
+++ b/docs/django.rst
@@ -1,5 +1,7 @@
-Identity for Django
-===================
+Identity for a Django Web App
+=============================
+
+.. include:: app-vs-api.rst
Prerequisite
------------
@@ -15,7 +17,7 @@ as a reference. What we need are basically these steps:
#. ``python manage.py runserver localhost:5000``
You must use a port matching your redirect_uri that you registered.
-#. Now, add an `index` view to your project.
+#. Now, add a new `mysite/views.py` file with an `index` view to your project.
For now, it can simply return a "hello world" page to any visitor::
from django.http import HttpResponse
@@ -23,7 +25,7 @@ as a reference. What we need are basically these steps:
return HttpResponse("Hello, world. Everyone can read this line.")
Configuration
----------------------------------
+-------------
#. Install dependency by ``pip install identity[django]``
diff --git a/docs/flask-webapi.rst b/docs/flask-webapi.rst
new file mode 100644
index 0000000..b6d05b4
--- /dev/null
+++ b/docs/flask-webapi.rst
@@ -0,0 +1,67 @@
+Identity for a Flask Web API
+============================
+
+.. include:: app-vs-api.rst
+
+Prerequisite
+------------
+
+Create `a hello world web project in Flask `_.
+Here we assume the project's main file is named ``app.py``.
+
+
+Configuration
+-------------
+
+#. Install dependency by ``pip install identity[flask]``
+
+#. Create an instance of the :py:class:`identity.flask.Auth` object,
+ and assign it to a global variable inside your ``app.py``::
+
+ import os
+ from flask import Flask
+ from identity.flask import Auth
+
+ app = Flask(__name__)
+ auth = Auth(
+ client_id=os.getenv('CLIENT_ID'),
+ ...=..., # See below on how to feed in the authority url parameter
+ )
+
+ .. include:: auth.rst
+
+
+Flask Web API protected by an access token
+------------------------------------------
+
+#. In your web project's ``app.py``, decorate some views with the
+ :py:func:`identity.flask.Auth.authorization_required` decorator.
+ It will automatically put validated token claims into the ``context`` dictionary,
+ under the key ``claims``.
+ or emit an HTTP 401 or 403 response if the token is missing or invalid.
+
+ ::
+
+ @app.route("/")
+ @auth.authorization_required(expected_scopes={
+ "your_scope_1": "api://your_client_id/your_scope_1",
+ "your_scope_2": "api://your_client_id/your_scope_2",
+ })
+ def index(*, context):
+ claims = context['claims']
+ # The user is uniquely identified by claims['sub'] or claims["oid"],
+ # claims['tid'] and/or claims['iss'].
+ return {"message": f"Data for {claims['sub']}@{claims['tid']}"}
+
+All of the content above are demonstrated in
+`this Flask web API sample `_.
+
+API for Flask web API projects
+------------------------------
+
+.. autoclass:: identity.flask.Auth
+ :members:
+ :inherited-members:
+
+ .. automethod:: __init__
+
diff --git a/docs/flask.rst b/docs/flask.rst
index 8fc21c8..dbfc6e3 100644
--- a/docs/flask.rst
+++ b/docs/flask.rst
@@ -1,5 +1,7 @@
-Identity for Flask
-==================
+Identity for a Flask Web App
+============================
+
+.. include:: app-vs-api.rst
Prerequisite
------------
@@ -9,7 +11,7 @@ Here we assume the project's main file is named ``app.py``.
Configuration
---------------------------------
+-------------
#. Install dependency by ``pip install identity[flask]``
diff --git a/docs/index.rst b/docs/index.rst
index 4ad1835..a42ca4d 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -47,8 +47,11 @@ This Identity library is a Python authentication/authorization library that:
:hidden:
django
+ django-webapi
flask
+ flask-webapi
quart
+ quart-webapi
abc
generic
@@ -60,3 +63,9 @@ This Identity library is a Python authentication/authorization library that:
Other modules in the source code are all considered as internal helpers,
which could change at anytime in the future, without prior notice.
+This library is designed to be used in either a web app or a web API.
+Understand the difference between the two scenarios,
+before you choose the right component to build your project.
+
+.. include:: app-vs-api.rst
+
diff --git a/docs/quart-webapi.rst b/docs/quart-webapi.rst
new file mode 100644
index 0000000..2a713c0
--- /dev/null
+++ b/docs/quart-webapi.rst
@@ -0,0 +1,67 @@
+Identity for a Quart Web API
+==============================
+
+.. include:: app-vs-api.rst
+
+Prerequisite
+------------
+
+Create `a hello world web project in Quart `_.
+Here we assume the project's main file is named ``app.py``.
+
+
+Configuration
+-------------
+
+#. Install dependency by ``pip install identity[quart]``
+
+#. Create an instance of the :py:class:`identity.quart.Auth` object,
+ and assign it to a global variable inside your ``app.py``::
+
+ import os
+ from quart import Quart
+ from identity.quart import Auth
+
+ app = Quart(__name__)
+ auth = Auth(
+ client_id=os.getenv('CLIENT_ID'),
+ ...=..., # See below on how to feed in the authority url parameter
+ )
+
+ .. include:: auth.rst
+
+
+Quart Web API protected by an access token
+------------------------------------------
+
+#. In your web project's ``app.py``, decorate some views with the
+ :py:func:`identity.quart.Auth.authorization_required` decorator.
+ It will automatically put validated token claims into the ``context`` dictionary,
+ under the key ``claims``.
+ or emit an HTTP 401 or 403 response if the token is missing or invalid.
+
+ ::
+
+ @app.route("/")
+ @auth.authorization_required(expected_scopes={
+ "your_scope_1": "api://your_client_id/your_scope_1",
+ "your_scope_2": "api://your_client_id/your_scope_2",
+ })
+ async def index(*, context):
+ claims = context['claims']
+ # The user is uniquely identified by claims['sub'] or claims["oid"],
+ # claims['tid'] and/or claims['iss'].
+ return {"message": f"Data for {claims['sub']}@{claims['tid']}"}
+
+All of the content above follows the same pattern as
+`Flask web API sample `_
+but uses async/await syntax for Quart.
+
+API for Quart web API projects
+------------------------------
+
+.. autoclass:: identity.quart.Auth
+ :members:
+ :inherited-members:
+
+ .. automethod:: __init__
diff --git a/identity/django.py b/identity/django.py
index c490fc4..94c69c9 100644
--- a/identity/django.py
+++ b/identity/django.py
@@ -6,8 +6,9 @@
from django.shortcuts import redirect, render
from django.urls import include, path, reverse
+from django.http import HttpResponse
-from .web import WebFrameworkAuth
+from .web import WebFrameworkAuth, HttpError, ApiAuth as _ApiAuth
logger = logging.getLogger(__name__)
@@ -221,3 +222,17 @@ def wrapper(request, *args, **kwargs):
scopes=scopes,
)
return wrapper
+
+ def authorization_required(self, *, expected_scopes, **kwargs):
+ def decorator(function):
+ @wraps(function)
+ def wrapper(request, *args, **kwargs):
+ try:
+ context = self._validate(request, expected_scopes=expected_scopes)
+ except HttpError as e:
+ return HttpResponse(
+ e.description, status=e.status_code, headers=e.headers)
+ return function(request, *args, context=context, **kwargs)
+ return wrapper
+ return decorator
+
diff --git a/identity/flask.py b/identity/flask.py
index 848c65e..42877e0 100644
--- a/identity/flask.py
+++ b/identity/flask.py
@@ -1,6 +1,8 @@
+import functools
from typing import List, Optional # Needed in Python 3.7 & 3.8
from flask import (
Blueprint, Flask,
+ abort, make_response,
redirect, render_template, request, session, url_for,
)
from flask_session import Session
@@ -13,10 +15,12 @@ class Auth(PalletAuth):
_Session = Session
_redirect = redirect
_url_for = url_for
+ _abort = abort
+ _make_response = make_response
def __init__(
self,
- app: Optional[Flask],
+ app: Optional[Flask] = None,
*args,
post_logout_view: Optional[callable] = None,
**kwargs,
@@ -166,3 +170,10 @@ def call_an_api(*, context):
...
"""
return super(Auth, self).login_required(function, scopes=scopes)
+
+ def raise_http_error(self, status_code, *, headers=None, description=None):
+ """Flask-specific implementation using Flask's make_response and abort."""
+ response = self.__class__._make_response(description, status_code)
+ response.headers.extend(headers or {})
+ self.__class__._abort(response)
+
diff --git a/identity/pallet.py b/identity/pallet.py
index 0827bce..89886d2 100644
--- a/identity/pallet.py
+++ b/identity/pallet.py
@@ -129,3 +129,29 @@ def wrapper(*args, **kwargs):
# Save an http 302 by calling self.login(request) instead of redirect(self.login)
return self.login(next_link=self._request.url, scopes=scopes)
return wrapper
+
+ def authorization_required(self, *, expected_scopes, **kwargs):
+ def decorator(function):
+ if iscoroutinefunction(function): # For Quart
+ @wraps(function)
+ async def async_wrapper(*args, **kwargs):
+ try:
+ context = self._validate(self._request, expected_scopes=expected_scopes)
+ except Exception as e:
+ # Handle HttpError for Quart - convert to response
+ if hasattr(e, 'status_code'):
+ response = await self.__class__._make_response(e.description, e.status_code)
+ if e.headers:
+ response.headers.update(e.headers)
+ self.__class__._abort(response)
+ raise
+ return await function(*args, context=context, **kwargs)
+ return async_wrapper
+ else: # For Flask
+ @wraps(function)
+ def wrapper(*args, **kwargs):
+ context = self._validate(self._request, expected_scopes=expected_scopes)
+ return function(*args, context=context, **kwargs)
+ return wrapper
+ return decorator
+
diff --git a/identity/quart.py b/identity/quart.py
index 7edf420..5800532 100644
--- a/identity/quart.py
+++ b/identity/quart.py
@@ -1,6 +1,7 @@
from typing import List, Optional # Needed in Python 3.7 & 3.8
from quart import (
Blueprint, Quart,
+ abort, make_response,
redirect, render_template, request, session, url_for,
)
from quart_session import Session
@@ -13,10 +14,12 @@ class Auth(PalletAuth):
_Session = Session
_redirect = redirect
_url_for = url_for
+ _abort = abort
+ _make_response = make_response
def __init__(
self,
- app: Optional[Quart],
+ app: Optional[Quart] = None,
*args,
post_logout_view: Optional[callable] = None,
**kwargs,
@@ -165,3 +168,17 @@ async def call_api(*, context):
"""
return super(Auth, self).login_required(function, scopes=scopes)
+
+ def raise_http_error(self, status_code, *, headers=None, description=None):
+ """Override to use HttpError exception instead of direct response creation.
+
+ Unlike Flask's synchronous make_response and abort functions, Quart's
+ make_response is async and requires await. Since this method is called
+ from the synchronous _validate() method, we cannot await here.
+
+ Instead, we raise an HttpError exception which is caught and properly
+ converted to a Quart response in the async wrapper of PalletAuth's
+ authorization_required decorator.
+ """
+ from .web import HttpError
+ raise HttpError(status_code, headers=headers, description=description)
diff --git a/identity/web.py b/identity/web.py
index 48ffc9a..f2601bf 100644
--- a/identity/web.py
+++ b/identity/web.py
@@ -1,9 +1,13 @@
from abc import ABC, abstractmethod
import functools
+import json
import logging
import time
-from typing import List, Optional # Needed in Python 3.7 & 3.8
+from typing import (
+ Optional, Union, # Needed until Python 3.10+
+)
+import jwt # PyJWT
import requests
import msal
@@ -11,6 +15,24 @@
logger = logging.getLogger(__name__)
+def _get_http_client(): # Better reuse the result of this function to save resources
+ http_client = requests.Session()
+ a = requests.adapters.HTTPAdapter(
+ # Web app/API shall use minimal retry;
+ # let the calling client decide their own retry strategy
+ max_retries=1)
+ http_client.mount("http://", a)
+ http_client.mount("https://", a)
+ return http_client
+
+@functools.lru_cache(maxsize=128)
+def __http_get_json(http_client, url, timestamp):
+ return http_client.get(url).json()
+
+def _http_get_json(url, *, http_client): # Get JSON from a URL or a one-day cache
+ return __http_get_json(http_client, url, time.strftime("%x"))
+
+
class Auth(object): # This a low level helper which is web framework agnostic
# These key names are hopefully unique in session
_TOKEN_CACHE = "_token_cache"
@@ -66,6 +88,7 @@ def __init__(
self._client_id = client_id
self._client_credential = client_credential
self._http_cache = {} if http_cache is None else http_cache # All subsequent MSAL instances will share this
+ self._http_client = _get_http_client()
def _load_cache(self):
cache = msal.SerializableTokenCache()
@@ -99,7 +122,7 @@ def _save_user_into_session(self, id_token_claims):
def log_in(
self,
*,
- scopes: Optional[List[str]] = None,
+ scopes: Optional[list[str]] = None,
redirect_uri: Optional[str] = None,
state: Optional[str] = None,
prompt: Optional[str] = None,
@@ -287,17 +310,6 @@ def _get_token_for_user(self, scopes, force_refresh=None):
return result
return {"error": "interaction_required", "error_description": "Cache missed"}
- @functools.lru_cache(maxsize=1)
- def _get_oidc_config(self):
- # The self._authority is usually the V1 endpoint of Microsoft Entra ID,
- # which is still good enough for log_out()
- a = self._oidc_authority or self._authority
- conf = requests.get(f"{a}/.well-known/openid-configuration").json()
- if not conf.get(self._END_SESSION_ENDPOINT):
- logger.warning(
- "%s not found from OIDC config: %s", self._END_SESSION_ENDPOINT, conf)
- return conf
-
def log_out(self, post_logout_redirect_uri: str) -> str:
# The vocabulary is "log out" (rather than "sign out") in the specs
# https://openid.net/specs/openid-connect-frontchannel-1_0.html
@@ -314,9 +326,14 @@ def log_out(self, post_logout_redirect_uri: str) -> str:
self._session.pop(self._USER, None) # Must
self._session.pop(self._TOKEN_CACHE, None) # Optional
try:
- # Empirically, Microsoft Entra ID's /v2.0 endpoint shows an account picker
- # but its default (i.e. v1.0) endpoint will sign out the (only?) account
- endpoint = self._get_oidc_config().get(self._END_SESSION_ENDPOINT)
+ # The self._authority ends up w/ the V1 endpoint of Microsoft Entra ID,
+ # which is still good enough for log_out()
+ authority = self._oidc_authority or self._authority
+ endpoint = _http_get_json(
+ # Empirically, Microsoft Entra ID /v2 endpoint shows an account picker
+ # but its default (i.e. v1) endpoint will sign out the (only?) account
+ f"{authority}/.well-known/openid-configuration",
+ http_client=self._http_client).get("end_session_endpoint")
if endpoint:
return f"{endpoint}?post_logout_redirect_uri={post_logout_redirect_uri}"
except requests.exceptions.RequestException:
@@ -354,7 +371,236 @@ def _is_valid(id_token_claims, skew=None, seconds=None):
else id_token_claims["iat"] + seconds)
-class WebFrameworkAuth(ABC): # This is a mid-level helper to be subclassed
+class HttpError(Exception):
+ def __init__(self, status_code, *, headers, description=None):
+ self.status_code = status_code
+ self.headers = headers
+ self.description = description
+
+
+class ApiAuth(ABC):
+ _INVALID_REQUEST = "invalid_request"
+ _INVALID_TOKEN = "invalid_token"
+ _INSUFFICIENT_SCOPE = "insufficient_scope"
+ _ERROR_MISSING_AUTHORIZATION = "Authorization header is missing"
+
+ def __init__(
+ self,
+ client_id: str,
+ *,
+ oidc_authority: Optional[str] = None,
+ authority: Optional[str] = None,
+ # Unlike web application's auth, this does not use session
+ ):
+ """Create an ApiAuth instance for a web API.
+
+ This instance is expected to be long-lived with the web app.
+
+ :param str oidc_authority:
+ The authority which your app registers with your OpenID Connect provider.
+ For example, ``https://example.com/foo``.
+ This library will concatenate ``/.well-known/openid-configuration``
+ to form the metadata endpoint.
+
+ :param str authority:
+ The authority which your app registers with your Microsoft Entra ID.
+ For example, ``https://example.com/foo``.
+ Historically, the underlying library will *sometimes* automatically
+ append "/v2.0" to it.
+ If you do not want that behavior, you may use ``oidc_authority`` instead.
+
+ :param str client_id:
+ The client_id of your web app, issued by its authority.
+ """
+ self._client_id = client_id
+ self._oidc_authority = oidc_authority
+ self._authority = authority
+ self._realm = oidc_authority or authority
+ self._http_client = _get_http_client()
+
+ def raise_http_error(self, status_code, *, headers=None, description=None):
+ """A web api will use this to emit http error response.
+
+ This can be overridden by a subclass for each web framework.
+ If a web framework (Django) does not have a way to raise an HTTP error,
+ its subclass or app must catch HttpError and render a response accordingly.
+ """
+ raise HttpError(status_code, headers=headers, description=description)
+
+ def __raise_oauth2_error(
+ self,
+ error_code: str,
+ *,
+ error_description: str = None,
+ error_uri: str = None,
+ scopes: list[str] = None,
+ ):
+ # https://datatracker.ietf.org/doc/html/rfc6750#section-3
+ auth_params = ", ".join(
+ '{k}: "{v}"'.format(k=k, v=v.replace('"', "'")) for k, v in dict(
+ realm=self._realm,
+ error=error_code,
+ error_description=error_description,
+ error_uri=error_uri,
+ scope=' '.join(scopes) if scopes else None,
+ ).items() if v and isinstance(v, str))
+ self.raise_http_error(
+ { # https://datatracker.ietf.org/doc/html/rfc6750#section-3.1
+ self._INVALID_REQUEST: 400,
+ self._INVALID_TOKEN: 401,
+ self._INSUFFICIENT_SCOPE: 403,
+ }.get(error_code, 400),
+ headers={"WWW-Authenticate": f'Bearer {auth_params}'},
+ description=f"{error_code}: {error_description}",
+ )
+
+ def _oidc_discovery(self):
+ idp = self._oidc_authority or f"{self._authority}/v2.0" # Matching MSAL behavior
+ oidc_discovery = f"{idp}/.well-known/openid-configuration"
+ logger.debug("OIDC Discovery from %s (probably via cache)", oidc_discovery)
+ return _http_get_json(oidc_discovery, http_client=self._http_client)
+
+ @functools.lru_cache(maxsize=128)
+ def __get_keys(self, timestamp):
+ # Returns the keys from the authority's jwks_uri, remap to {kid: PyJWT's key}
+ if not (self._oidc_authority or self._authority):
+ self.raise_http_error( # So the API errors out gracefully
+ 500, description="No authority to fetch keys from")
+ idp = self._oidc_authority or f"{self._authority}/v2.0" # Matching MSAL behavior
+ try:
+ conf = self._oidc_discovery()
+ try:
+ return {
+ jwk["kid"]: dict( # Empirically, kid always exists in the wild
+ jwk, # https://www.rfc-editor.org/rfc/rfc7517.html#section-4.5
+ # Inspired from https://stackoverflow.com/a/68891371/728675
+ pyjwt_key=jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(jwk)),
+ ) for jwk in _http_get_json(
+ conf["jwks_uri"], http_client=self._http_client)["keys"]
+ }
+ except KeyError as e:
+ self.raise_http_error(500, description=f'Key should have "kid": {e}')
+ except requests.exceptions.RequestException as e:
+ self.raise_http_error(500, description=f"Failed to get keys: {e}")
+
+ def _get_keys(self):
+ return self.__get_keys(time.strftime("%x")) # Refresh keys daily
+
+ def _validate_bearer_token(
+ self,
+ token:str,
+ *,
+ scopes: Union[list[str], dict[str, str]],
+ ):
+ # Return claims of the JWT if valid, otherwise calls __raise_oauth2_error()
+ try:
+ kid = jwt.get_unverified_header(token)["kid"]
+ key = self._get_keys().get(kid)
+ if not key:
+ self.__raise_oauth2_error(
+ self._INVALID_TOKEN,
+ error_description=f"Key not found for kid {kid}")
+ claims = jwt.decode(
+ token,
+ key["pyjwt_key"],
+ algorithms=["RS256"], # Hardcode it to prevent downgrade attack
+ audience=self._client_id,
+ ) # https://pyjwt.readthedocs.io/en/stable/api.html#jwt.decode
+
+ expected_scopes = set(scopes or [])
+ authorized_scopes = set(claims.get("scp", "").split())
+ if expected_scopes - authorized_scopes:
+ # TODO: It could also be daemon app. Need to support actor validation
+ # https://learn.microsoft.com/en-us/entra/identity-platform/claims-validation#validate-the-actor
+ self.__raise_oauth2_error(
+ self._INSUFFICIENT_SCOPE,
+ error_description="Insufficient scope(s). "
+ f'''This API expects "{' '.join(expected_scopes)}", '''
+ f'''but got only "{' '.join(authorized_scopes)}".''',
+ # scopes can be a dict of {"scope_in_scp": "scope in request"}
+ scopes=scopes.values() if isinstance(scopes, dict) else scopes)
+
+ # https://learn.microsoft.com/en-us/entra/identity-platform/access-tokens#recap
+ expected_issuer = key.get("issuer") or self._oidc_discovery()["issuer"]
+ if self._authority and "tid" in claims: # Microsoft Entra ID code path
+ expected_issuer = expected_issuer.replace("{tenantid}", claims["tid"])
+ if claims.get("iss") != expected_issuer:
+ self.__raise_oauth2_error(
+ self._INVALID_TOKEN, error_description="Issuer mismatch. "
+ f"(Expected {expected_issuer}, got {claims.get('iss')})")
+ return claims
+
+ except (
+ jwt.DecodeError, jwt.InvalidSignatureError, jwt.InvalidAudienceError,
+ ) as e:
+ self.__raise_oauth2_error(self._INVALID_TOKEN, error_description=f"{e}")
+
+ def _validate(
+ self,
+ request, # We expect request.headers to be a dict-like object
+ *,
+ expected_scopes, # Defined in _validate_bearer_token(),
+ # documented in authorization_required()
+ ):
+ authz = request.headers.get("Authorization")
+ # https://datatracker.ietf.org/doc/html/rfc6750#section-3
+ if not authz:
+ self.raise_http_error(
+ 401, # https://stackoverflow.com/a/6937030/728675
+ headers={"WWW-Authenticate": f'Bearer realm="{self._realm}"'},
+ description=self._ERROR_MISSING_AUTHORIZATION,
+ )
+ authz_parts = authz.split(maxsplit=1)
+ scheme = authz_parts[0]
+ params = authz_parts[1:] # May be an empty list
+ if scheme == "Bearer" and params:
+ context = {
+ "claims": # TODO: Decide on the name
+ self._validate_bearer_token(
+ params[0], scopes=expected_scopes),
+ }
+ # TODO: Support more schemes
+ else:
+ self.raise_http_error(
+ 401,
+ headers={"WWW-Authenticate": f'Bearer realm="{self._realm}"'},
+ description=f"Authorization header is invalid. ({authz})",
+ )
+ return context
+
+ @abstractmethod
+ def authorization_required( # Lengthy but precise name
+ self,
+ *,
+ expected_scopes: list[str], # TODO: Accept a callable in RESTful API scenario?
+ #extra_scopes: list[str]=None, # TODO: For OBO. UPDATE: No, OBO shall be done by ApiCaller(context, scope)
+ ):
+ # Sub-classes inherit the docstring, so we only document the commen params.
+ """It returns a decorator that verifies the request's authorization header.
+
+ A request not meeting the requirement(s) will raise an HTTP 401 Unauthorized.
+ For a valid request, the view will be called with a keyword argument
+ named "context" which is a dict containing the user object.
+
+ Usage::
+
+ @settings.AUTH.authorization_required(expected_scopes=["foo", "bar"])
+ def resource(..., *, context): # The ... part is differnt per web framework
+ # The user is uniquely identified by claims['sub'] or claims["oid"],
+ # claims['tid'] and/or claims['iss'].
+ claims = context["claims"]
+ return {"content": f"content for {claims['sub']}@{claims['tid']}"}
+
+ :param expected_scopes:
+ These scopes are expected to be present in the token's "scp" claim.
+ If not, the view will emit an HTTP 401 Unauthorized or HTTP 403 Forbidden.
+ """
+ raise NotImplementedError("Subclass must implement this method")
+
+
+class WebFrameworkAuth( # This is a mid-level helper to be subclassed
+ ApiAuth, # After prototyping, we chose to have one class for web app and web API
+):
"""This is a mid-level helper to be subclassed. Do not use it directly."""
def __init__(
self,
@@ -421,18 +667,18 @@ def __init__(
Optional.
"""
- self._client_id = client_id
+ # self._client_id = client_id
self._client_credential = client_credential
self._redirect_uri = redirect_uri
self._http_cache: dict = {} # All subsequent Auth instances will share this
- self._authority: Optional[str] = None # It makes mypy happy
+ _authority: Optional[str] = None # It makes mypy happy
# Note: We do not use overload, because we want to allow the caller to
# have only one code path that relay in all the optional parameters.
if b2c_tenant_name and b2c_signup_signin_user_flow:
b2c_authority_template = ( # TODO: Support custom domain
"https://{tenant}.b2clogin.com/{tenant}.onmicrosoft.com/{user_flow}")
- self._authority = b2c_authority_template.format(
+ _authority = b2c_authority_template.format(
tenant=b2c_tenant_name,
user_flow=b2c_signup_signin_user_flow,
http_cache=self._http_cache,
@@ -456,10 +702,12 @@ def __init__(
http_cache=self._http_cache,
) if b2c_reset_password_user_flow else None
else:
- self._authority = authority
+ _authority = authority
self._edit_profile_auth = None
self._reset_password_auth = None
- self._oidc_authority = oidc_authority
+ # self._oidc_authority = oidc_authority
+ # self._authority = self._authority or _authority
+ super().__init__(client_id, oidc_authority=oidc_authority, authority=_authority)
def _get_configuration_error(self):
# Do not raise exception, because
@@ -483,7 +731,7 @@ def _build_auth(self, session) -> Auth:
http_cache=self._http_cache,
)
- def _login_required(self, auth: Auth, user: dict, scopes: List[str]):
+ def _login_required(self, auth: Auth, user: dict, scopes: list[str]):
# Returns the context. This logic is reused in the login_required decorators.
context = None
if user:
diff --git a/setup.cfg b/setup.cfg
index 6a0a437..9c77b7b 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -18,11 +18,11 @@ classifiers =
Programming Language :: Python
Programming Language :: Python :: 3
Programming Language :: Python :: 3 :: Only
- Programming Language :: Python :: 3.8
Programming Language :: Python :: 3.9
Programming Language :: Python :: 3.10
Programming Language :: Python :: 3.11
Programming Language :: Python :: 3.12
+ Programming Language :: Python :: 3.13
# NOTE: Settings of this section below this line should not need to be changed
@@ -30,10 +30,14 @@ long_description = file: README.md
long_description_content_type = text/markdown
[options]
-python_requires = >=3.8
+python_requires = >=3.9
install_requires =
msal>=1.28,<2
requests>=2.0.0,<3
+
+ # CVE-2022-29217 was fixed in PyJWT 2.4+
+ PyJWT[crypto]>=2.4,<3
+
# importlib; python_version == "2.6"
# See also https://setuptools.readthedocs.io/en/latest/userguide/quickstart.html#dependency-management
diff --git a/tests/test_django.py b/tests/test_django.py
index 0fcf348..6df5453 100644
--- a/tests/test_django.py
+++ b/tests/test_django.py
@@ -44,19 +44,18 @@ def test_logout():
request = mock.MagicMock(
build_absolute_uri=lambda relative_uri: f"http://localhost{relative_uri}"
)
- with mock.patch('identity.web.requests.get', new=mock.MagicMock(
- return_value=mock.MagicMock(
- json=mock.MagicMock(return_value={
- "end_session_endpoint": "https://example.com/end_session",
- }),
- status_code=200,
- )
- )):
- response = Auth("client_id").logout(request)
+ with mock.patch('identity.web._http_get_json', return_value={
+ "end_session_endpoint": "https://example.com/end_session",
+ }):
+ response = Auth("client_id", authority="https://example.com").logout(request)
assert response.status_code == 302
assert response.url == "https://example.com/end_session?post_logout_redirect_uri=http://localhost/"
- auth = Auth("client_id", post_logout_view=lambda r: "You have logged out")
+ auth = Auth(
+ "client_id",
+ authority="https://example.com",
+ post_logout_view=lambda r: "You have logged out",
+ )
with mock.patch('identity.django.reverse', return_value="/post_logout"):
response = auth.logout(request)
assert response.status_code == 302
diff --git a/tests/test_flask.py b/tests/test_flask.py
index 6761448..6c264db 100644
--- a/tests/test_flask.py
+++ b/tests/test_flask.py
@@ -16,7 +16,10 @@ def app(): # https://flask.palletsprojects.com/en/3.0.x/testing/
# see also https://stackoverflow.com/questions/26080872
})
yield app
- shutil.rmtree("flask_session") # clean up
+ try:
+ shutil.rmtree("flask_session") # clean up
+ except FileNotFoundError:
+ pass
def build_auth(app, post_logout_view=None):
return Auth(
@@ -41,7 +44,7 @@ def post_logout_view():
app,
post_logout_view=post_logout_view if customize_post_logout else None,
)
- with patch.object(auth._auth, "_get_oidc_config", new=Mock(return_value={
+ with patch("identity.web._http_get_json", new=Mock(return_value={
"end_session_endpoint": "https://example.com/end_session",
})):
with app.test_request_context("/", method="GET"):
@@ -73,3 +76,25 @@ def dummy_view():
"http://localhost/app_root/path?foo=bar" # The full url
), "Next path should honor APPLICATION_ROOT"
+def test_authorization(app):
+ auth = Auth(client_id="fake", oidc_authority="https://example.com/foo")
+
+ @app.route("/path")
+ @auth.authorization_required(expected_scopes=["foo"])
+ def dummy_view():
+ return "content visible when authorized"
+
+ with app.test_request_context("/path", method="GET"):
+ with app.test_client() as client:
+ result = client.get("/path")
+ assert result.status_code == 401
+ assert "WWW-Authenticate" in result.headers
+ assert "Bearer" in result.headers["WWW-Authenticate"]
+ assert result.text == auth._ERROR_MISSING_AUTHORIZATION
+
+ result = client.get("/path", headers={"Authorization": "Bearer h.b.s"})
+ assert result.status_code == 401
+ assert "WWW-Authenticate" in result.headers
+ assert "Bearer" in result.headers["WWW-Authenticate"]
+ assert result.text != auth._ERROR_MISSING_AUTHORIZATION
+
diff --git a/tests/test_quart.py b/tests/test_quart.py
index 0f16864..7a3e7c7 100644
--- a/tests/test_quart.py
+++ b/tests/test_quart.py
@@ -37,3 +37,29 @@ def test_logout():
In the future, we might remove the session dependency anyway and revisit this.
"""
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_authorization():
+ app = Quart(__name__)
+ auth = Auth(client_id="fake", oidc_authority="https://example.com/foo")
+
+ @app.route("/path")
+ @auth.authorization_required(expected_scopes=["foo"])
+ async def dummy_view(*, context):
+ return "content visible when authorized"
+
+ async with app.test_client() as client:
+ result = await client.get("/path")
+ assert result.status_code == 401
+ assert "WWW-Authenticate" in result.headers
+ assert "Bearer" in result.headers["WWW-Authenticate"]
+ response_data = await result.get_data()
+ assert response_data.decode() == auth._ERROR_MISSING_AUTHORIZATION
+
+ result = await client.get("/path", headers={"Authorization": "Bearer h.b.s"})
+ assert result.status_code == 401
+ assert "WWW-Authenticate" in result.headers
+ assert "Bearer" in result.headers["WWW-Authenticate"]
+ response_data = await result.get_data()
+ assert response_data.decode() != auth._ERROR_MISSING_AUTHORIZATION
+
|