7 Commits

9 changed files with 125 additions and 100 deletions

View File

@ -26,7 +26,7 @@ version: 2
build:
os: ubuntu-22.04
tools:
python: "3.7"
python: "3.8"
# Build documentation in the docs/ directory with Sphinx

View File

@ -2,6 +2,25 @@ Change Log
==========
Version 0.6.1
-------------
Released 2023/5/3
* Revised the code for the upcoming Werkzeug 2.4.
Version 0.6.0
-------------
Released 2023/4/26
* Updated the minimal Python version to 3.8.
* Switched from ``setup.cfg`` to ``pyproject.toml``.
* Added the change log.
* Simplified ``README.rst``.
Version 0.5.0
-------------

View File

@ -22,5 +22,5 @@ from flask_digest_auth.algo import make_password_hash, calc_response
from flask_digest_auth.auth import DigestAuth
from flask_digest_auth.test import Client
VERSION: str = "0.5.0"
VERSION: str = "0.6.1"
"""The package version."""

View File

@ -20,8 +20,8 @@
"""
from __future__ import annotations
import typing as t
from hashlib import md5
from typing import Optional, Literal
def make_password_hash(realm: str, username: str, password: str) -> str:
@ -44,10 +44,10 @@ def make_password_hash(realm: str, username: str, password: str) -> str:
def calc_response(
method: str, uri: str, password_hash: str,
nonce: str, qop: t.Optional[t.Literal["auth", "auth-int"]] = None,
algorithm: t.Optional[t.Literal["MD5", "MD5-sess"]] = "MD5-sess",
cnonce: t.Optional[str] = None, nc: t.Optional[str] = None,
body: t.Optional[bytes] = None) -> str:
nonce: str, qop: Optional[Literal["auth", "auth-int"]] = None,
algorithm: Optional[Literal["MD5", "MD5-sess"]] = "MD5-sess",
cnonce: Optional[str] = None, nc: Optional[str] = None,
body: Optional[bytes] = None) -> str:
"""Calculates the response value of the HTTP digest authentication.
:param method: The request method.

View File

@ -23,9 +23,9 @@ See `RFC 2617`_ HTTP Authentication: Basic and Digest Access Authentication
from __future__ import annotations
import sys
import typing as t
from functools import wraps
from secrets import token_urlsafe, randbits
from typing import Any, Optional, Literal, Callable, List
from flask import g, request, Response, session, abort, Flask, Request, \
current_app
@ -38,7 +38,7 @@ from flask_digest_auth.algo import calc_response
class DigestAuth:
"""The HTTP digest authentication."""
def __init__(self, realm: t.Optional[str] = None):
def __init__(self, realm: Optional[str] = None):
"""Constructs the HTTP digest authentication.
:param realm: The realm.
@ -48,16 +48,15 @@ class DigestAuth:
"""The serializer to generate and validate the nonce and opaque."""
self.realm: str = "Login Required" if realm is None else realm
"""The realm. Default is "Login Required"."""
self.algorithm: t.Optional[t.Literal["MD5", "MD5-sess"]] = None
self.algorithm: Optional[Literal["MD5", "MD5-sess"]] = None
"""The algorithm, either None, ``MD5``, or ``MD5-sess``. Default is
None."""
self.use_opaque: bool = True
"""Whether to use an opaque. Default is True."""
self.__domain: t.List[str] = []
self.__domain: List[str] = []
"""A list of directories that this username and password applies to.
Default is empty."""
self.__qop: t.List[t.Literal["auth", "auth-int"]] \
= ["auth", "auth-int"]
self.__qop: List[Literal["auth", "auth-int"]] = ["auth", "auth-int"]
"""A list of supported quality of protection supported, either
``qop``, ``auth-int``, both, or empty. Default is both."""
self.__get_password_hash: BasePasswordHashGetter \
@ -68,7 +67,7 @@ class DigestAuth:
self.__on_login: BaseOnLogInCallback = BaseOnLogInCallback()
"""The callback to run when the user logs in."""
def login_required(self, view) -> t.Callable:
def login_required(self, view) -> Callable:
"""The view decorator for the HTTP digest authentication.
:Example:
@ -89,7 +88,7 @@ class DigestAuth:
class NoLogInException(Exception):
"""The exception thrown when the user is not authorized."""
def get_logged_in_user() -> t.Any:
def get_logged_in_user() -> Any:
"""Returns the currently logged-in user.
:return: The currently logged-in user.
@ -97,13 +96,13 @@ class DigestAuth:
"""
if "user" not in session:
raise NoLogInException
user: t.Optional[t.Any] = self.__get_user(session["user"])
user: Optional[Any] = self.__get_user(session["user"])
if user is None:
del session["user"]
raise NoLogInException
return user
def auth_user(state: AuthState) -> t.Any:
def auth_user(state: AuthState) -> Any:
"""Authenticates a user.
:param state: The authentication state.
@ -121,7 +120,7 @@ class DigestAuth:
return self.__get_user(authorization.username)
@wraps(view)
def login_required_view(*args, **kwargs) -> t.Any:
def login_required_view(*args, **kwargs) -> Any:
"""The login-protected view.
:param args: The positional arguments of the view.
@ -171,7 +170,7 @@ class DigestAuth:
except BadData:
raise UnauthorizedException("Invalid opaque")
state.opaque = authorization.opaque
password_hash: t.Optional[str] \
password_hash: Optional[str] \
= self.__get_password_hash(authorization.username)
if password_hash is None:
raise UnauthorizedException(
@ -202,7 +201,7 @@ class DigestAuth:
:return: The ``WWW-Authenticate`` response header.
"""
def get_opaque() -> t.Optional[str]:
def get_opaque() -> Optional[str]:
"""Returns the opaque value.
:return: The opaque value.
@ -213,7 +212,7 @@ class DigestAuth:
return state.opaque
return self.__serializer.dumps(randbits(32), salt="opaque")
opaque: t.Optional[str] = get_opaque()
opaque: Optional[str] = get_opaque()
nonce: str = self.__serializer.dumps(
randbits(32),
salt="nonce" if opaque is None else f"nonce-{opaque}")
@ -226,7 +225,7 @@ class DigestAuth:
if opaque is not None:
header += f", opaque=\"{opaque}\""
if state.stale is not None:
header += f", stale=TRUE" if state.stale else f", stale=FALSE"
header += ", stale=TRUE" if state.stale else ", stale=FALSE"
if self.algorithm is not None:
header += f", algorithm=\"{self.algorithm}\""
if len(self.__qop) > 0:
@ -234,7 +233,7 @@ class DigestAuth:
header += f", qop=\"{qop_list}\""
return header
def register_get_password(self, func: t.Callable[[str], t.Optional[str]])\
def register_get_password(self, func: Callable[[str], Optional[str]]) \
-> None:
"""The decorator to register the callback to obtain the password hash.
@ -256,7 +255,7 @@ class DigestAuth:
"""The base password hash getter."""
@staticmethod
def __call__(username: str) -> t.Optional[str]:
def __call__(username: str) -> Optional[str]:
"""Returns the password hash of a user.
:param username: The username.
@ -266,8 +265,7 @@ class DigestAuth:
self.__get_password_hash = PasswordHashGetter()
def register_get_user(self, func: t.Callable[[str], t.Optional[t.Any]])\
-> None:
def register_get_user(self, func: Callable[[str], Optional[Any]]) -> None:
"""The decorator to register the callback to obtain the user.
:Example:
@ -287,7 +285,7 @@ class DigestAuth:
"""The user getter."""
@staticmethod
def __call__(username: str) -> t.Optional[t.Any]:
def __call__(username: str) -> Optional[Any]:
"""Returns a user.
:param username: The username.
@ -297,7 +295,7 @@ class DigestAuth:
self.__get_user = UserGetter()
def register_on_login(self, func: t.Callable[[t.Any], None]) -> None:
def register_on_login(self, func: Callable[[Any], None]) -> None:
"""The decorator to register the callback to run when the user logs in.
:Example:
@ -316,7 +314,7 @@ class DigestAuth:
"""The callback when the user logs in."""
@staticmethod
def __call__(user: t.Any) -> None:
def __call__(user: Any) -> None:
"""Runs the callback when the user logs in.
:param user: The logged-in user.
@ -346,50 +344,57 @@ class DigestAuth:
self.realm = app.config["DIGEST_AUTH_REALM"]
if hasattr(app, "login_manager"):
from flask_login import LoginManager, login_user
self.__init_login_manager(app)
login_manager: LoginManager = getattr(app, "login_manager")
def __init_login_manager(self, app: Flask) -> None:
"""Initializes the Flask-Login login manager.
@login_manager.unauthorized_handler
def unauthorized() -> None:
"""Handles when the user is unauthorized.
:param app: The Flask application.
:return: None.
"""
from flask_login import LoginManager, login_user
login_manager: LoginManager = getattr(app, "login_manager")
:return: None.
"""
state: AuthState = getattr(request, "_digest_auth_state") \
if hasattr(request, "_digest_auth_state") \
else AuthState()
response: Response = Response()
response.status = 401
response.headers["WWW-Authenticate"] \
= self.__make_response_header(state)
abort(response)
@login_manager.unauthorized_handler
def unauthorized() -> None:
"""Handles when the user is unauthorized.
@login_manager.request_loader
def load_user_from_request(req: Request) -> t.Optional[t.Any]:
"""Loads the user from the request header.
:return: None.
"""
state: AuthState = getattr(request, "_digest_auth_state") \
if hasattr(request, "_digest_auth_state") \
else AuthState()
response: Response = Response()
response.status = 401
response.headers["WWW-Authenticate"] \
= self.__make_response_header(state)
abort(response)
:param req: The request.
:return: The authenticated user, or None if the
authentication fails
"""
request._digest_auth_state = AuthState()
authorization: Authorization = req.authorization
try:
if authorization is None:
raise UnauthorizedException
if authorization.type != "digest":
raise UnauthorizedException(
"Not an HTTP digest authorization")
self.__authenticate(request._digest_auth_state)
user = login_manager.user_callback(authorization.username)
login_user(user)
self.__on_login(user)
return user
except UnauthorizedException as e:
if str(e) != "":
app.logger.warning(str(e))
return None
@login_manager.request_loader
def load_user_from_request(req: Request) -> Optional[Any]:
"""Loads the user from the request header.
:param req: The request.
:return: The authenticated user, or None if the
authentication fails
"""
request._digest_auth_state = AuthState()
authorization: Authorization = req.authorization
try:
if authorization is None:
raise UnauthorizedException
if authorization.type != "digest":
raise UnauthorizedException(
"Not an HTTP digest authorization")
self.__authenticate(request._digest_auth_state)
user = login_manager.user_callback(authorization.username)
login_user(user)
self.__on_login(user)
return user
except UnauthorizedException as e:
if str(e) != "":
app.logger.warning(str(e))
return None
def logout(self) -> None:
"""Logs out the user.
@ -427,9 +432,9 @@ class AuthState:
def __init__(self):
"""Constructs the authorization state."""
self.opaque: t.Optional[str] = None
self.opaque: Optional[str] = None
"""The opaque value specified by the client, if valid."""
self.stale: t.Optional[bool] = None
self.stale: Optional[bool] = None
"""The stale value, if there is a previous log in attempt."""
@ -446,7 +451,7 @@ class BasePasswordHashGetter:
"""
@staticmethod
def __call__(username: str) -> t.Optional[str]:
def __call__(username: str) -> Optional[str]:
"""Returns the password hash of a user.
:param username: The username.
@ -467,7 +472,7 @@ class BaseUserGetter:
"""
@staticmethod
def __call__(username: str) -> t.Optional[t.Any]:
def __call__(username: str) -> Optional[Any]:
"""Returns a user.
:param username: The username.
@ -487,7 +492,7 @@ class BaseOnLogInCallback:
"""
@staticmethod
def __call__(user: t.Any) -> None:
def __call__(user: Any) -> None:
"""Runs the callback when the user logs in.
:param user: The logged-in user.

View File

@ -18,11 +18,12 @@
"""The test client with HTTP digest authentication enabled.
"""
import typing as t
from secrets import token_urlsafe
from typing import Optional, Literal, Tuple, Dict
from flask import g
from werkzeug.datastructures import Authorization, WWWAuthenticate
from werkzeug.http import parse_set_header
from werkzeug.test import TestResponse, Client as WerkzeugClient
from flask_digest_auth.algo import calc_response, make_password_hash
@ -83,7 +84,7 @@ class Client(WerkzeugClient):
.. _pytest: https://pytest.org
"""
def open(self, *args, digest_auth: t.Optional[t.Tuple[str, str]] = None,
def open(self, *args, digest_auth: Optional[Tuple[str, str]] = None,
**kwargs) -> TestResponse:
"""Opens a request.
@ -117,14 +118,14 @@ class Client(WerkzeugClient):
:param password: The password.
:return: The request authorization.
"""
qop: t.Optional[t.Literal["auth", "auth-int"]] = None
if www_authenticate.qop is not None and "auth" in www_authenticate.qop:
qop: Optional[Literal["auth", "auth-int"]] = None
if "auth" in parse_set_header(www_authenticate.get("qop")):
qop = "auth"
cnonce: t.Optional[str] = None
cnonce: Optional[str] = None
if qop is not None or www_authenticate.algorithm == "MD5-sess":
cnonce = token_urlsafe(8)
nc: t.Optional[str] = None
nc: Optional[str] = None
count: int = 1
if qop is not None:
nc: str = hex(count)[2:].zfill(8)
@ -137,7 +138,7 @@ class Client(WerkzeugClient):
algorithm=www_authenticate.algorithm, cnonce=cnonce, nc=nc,
body=None)
data: t.Dict[str, str] = {
data: Dict[str, str] = {
"username": username, "realm": www_authenticate.realm,
"nonce": www_authenticate.nonce, "uri": uri, "response": expected}
if www_authenticate.algorithm is not None:

View File

@ -18,8 +18,8 @@
"""The test case for the HTTP digest authentication algorithm.
"""
import typing as t
import unittest
from typing import Optional, Literal
from flask_digest_auth import make_password_hash, calc_response
@ -39,11 +39,11 @@ class AlgorithmTestCase(unittest.TestCase):
method: str = "GET"
uri: str = "/dir/index.html"
nonce: str = "dcd98b7102dd2f0e8b11d0f600bfb0c093"
qop: t.Optional[t.Literal["auth", "auth-int"]] = "auth"
algorithm: t.Optional[t.Literal["MD5", "MD5-sess"]] = None
cnonce: t.Optional[str] = "0a4f113b"
nc: t.Optional[str] = "00000001"
body: t.Optional[bytes] = None
qop: Optional[Literal["auth", "auth-int"]] = "auth"
algorithm: Optional[Literal["MD5", "MD5-sess"]] = None
cnonce: Optional[str] = "0a4f113b"
nc: Optional[str] = "00000001"
body: Optional[bytes] = None
password_hash: str = make_password_hash(realm, username, password)
response: str = calc_response(method, uri, password_hash, nonce, qop,

View File

@ -18,8 +18,8 @@
"""The test case for the HTTP digest authentication.
"""
import typing as t
from secrets import token_urlsafe
from typing import Any, Optional, Dict
from flask import Response, Flask, g, redirect, request
from flask_testing import TestCase
@ -66,10 +66,10 @@ class AuthenticationTestCase(TestCase):
auth: DigestAuth = DigestAuth()
auth.init_app(app)
self.user: User = User(_USERNAME, _PASSWORD)
user_db: t.Dict[str, User] = {_USERNAME: self.user}
user_db: Dict[str, User] = {_USERNAME: self.user}
@auth.register_get_password
def get_password_hash(username: str) -> t.Optional[str]:
def get_password_hash(username: str) -> Optional[str]:
"""Returns the password hash of a user.
:param username: The username.
@ -79,7 +79,7 @@ class AuthenticationTestCase(TestCase):
else None
@auth.register_get_user
def get_user(username: str) -> t.Optional[t.Any]:
def get_user(username: str) -> Optional[Any]:
"""Returns a user.
:param username: The username.
@ -158,7 +158,7 @@ class AuthenticationTestCase(TestCase):
self.assertEqual(response.status_code, 401)
www_authenticate = response.www_authenticate
self.assertEqual(www_authenticate.type, "digest")
self.assertEqual(www_authenticate.stale, None)
self.assertIsNone(www_authenticate.get("stale"))
opaque: str = www_authenticate.opaque
www_authenticate.nonce = "bad"
@ -167,7 +167,7 @@ class AuthenticationTestCase(TestCase):
response = super(Client, self.client).get(admin_uri, auth=auth_data)
self.assertEqual(response.status_code, 401)
www_authenticate = response.www_authenticate
self.assertEqual(www_authenticate.stale, True)
self.assertEqual(www_authenticate.get("stale"), "TRUE")
self.assertEqual(www_authenticate.opaque, opaque)
auth_data = Client.make_authorization(
@ -175,7 +175,7 @@ class AuthenticationTestCase(TestCase):
response = super(Client, self.client).get(admin_uri, auth=auth_data)
self.assertEqual(response.status_code, 401)
www_authenticate = response.www_authenticate
self.assertEqual(www_authenticate.stale, False)
self.assertEqual(www_authenticate.get("stale"), "FALSE")
self.assertEqual(www_authenticate.opaque, opaque)
auth_data = Client.make_authorization(

View File

@ -18,8 +18,8 @@
"""The test case for the Flask-Login integration.
"""
import typing as t
from secrets import token_urlsafe
from typing import Optional, Dict
from flask import Response, Flask, g, redirect, request
from flask_testing import TestCase
@ -97,10 +97,10 @@ class FlaskLoginTestCase(TestCase):
auth.init_app(app)
self.user: User = User(_USERNAME, _PASSWORD)
user_db: t.Dict[str, User] = {_USERNAME: self.user}
user_db: Dict[str, User] = {_USERNAME: self.user}
@auth.register_get_password
def get_password_hash(username: str) -> t.Optional[str]:
def get_password_hash(username: str) -> Optional[str]:
"""Returns the password hash of a user.
:param username: The username.
@ -119,7 +119,7 @@ class FlaskLoginTestCase(TestCase):
user.visits = user.visits + 1
@login_manager.user_loader
def load_user(user_id: str) -> t.Optional[User]:
def load_user(user_id: str) -> Optional[User]:
"""Loads a user.
:param user_id: The username.
@ -195,7 +195,7 @@ class FlaskLoginTestCase(TestCase):
self.assertEqual(response.status_code, 401)
www_authenticate = response.www_authenticate
self.assertEqual(www_authenticate.type, "digest")
self.assertEqual(www_authenticate.stale, None)
self.assertIsNone(www_authenticate.get("stale"))
opaque: str = www_authenticate.opaque
if hasattr(g, "_login_user"):
@ -206,7 +206,7 @@ class FlaskLoginTestCase(TestCase):
response = super(Client, self.client).get(admin_uri, auth=auth_data)
self.assertEqual(response.status_code, 401)
www_authenticate = response.www_authenticate
self.assertEqual(www_authenticate.stale, True)
self.assertEqual(www_authenticate.get("stale"), "TRUE")
self.assertEqual(www_authenticate.opaque, opaque)
if hasattr(g, "_login_user"):
@ -216,7 +216,7 @@ class FlaskLoginTestCase(TestCase):
response = super(Client, self.client).get(admin_uri, auth=auth_data)
self.assertEqual(response.status_code, 401)
www_authenticate = response.www_authenticate
self.assertEqual(www_authenticate.stale, False)
self.assertEqual(www_authenticate.get("stale"), "FALSE")
self.assertEqual(www_authenticate.opaque, opaque)
if hasattr(g, "_login_user"):