Compare commits
5 Commits
Author | SHA1 | Date | |
---|---|---|---|
f04ea7ac18 | |||
15ea650ddd | |||
5b255b6504 | |||
919b8d0dc3 | |||
604ed0be27 |
@ -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
|
||||
|
||||
|
@ -2,6 +2,14 @@ Change Log
|
||||
==========
|
||||
|
||||
|
||||
Version 0.6.1
|
||||
-------------
|
||||
|
||||
Released 2023/5/3
|
||||
|
||||
* Revised the code for the upcoming Werkzeug 2.4.
|
||||
|
||||
|
||||
Version 0.6.0
|
||||
-------------
|
||||
|
||||
|
@ -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.6.0"
|
||||
VERSION: str = "0.6.1"
|
||||
"""The package version."""
|
||||
|
@ -225,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:
|
||||
@ -344,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) -> Optional[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.
|
||||
|
@ -23,6 +23,7 @@ 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
|
||||
@ -118,7 +119,7 @@ class Client(WerkzeugClient):
|
||||
:return: The request authorization.
|
||||
"""
|
||||
qop: Optional[Literal["auth", "auth-int"]] = None
|
||||
if www_authenticate.qop is not None and "auth" in www_authenticate.qop:
|
||||
if "auth" in parse_set_header(www_authenticate.get("qop")):
|
||||
qop = "auth"
|
||||
|
||||
cnonce: Optional[str] = None
|
||||
|
@ -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(
|
||||
|
@ -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"):
|
||||
|
Reference in New Issue
Block a user