Compare commits
23 Commits
65c3322ecc
...
v0.2.1
Author | SHA1 | Date | |
---|---|---|---|
495e9a9785 | |||
cbbd2248f0 | |||
2028cb1362 | |||
7e71115844 | |||
491da61a79 | |||
bbaebbc80d | |||
0dfdf70c45 | |||
0432561b21 | |||
3709cb4d66 | |||
9d0d0b2686 | |||
8c98d35934 | |||
7db38c7eae | |||
9616fb3ddc | |||
f473db29a8 | |||
b39e9b1321 | |||
f3b525d715 | |||
0f3694ba05 | |||
2425d99492 | |||
be163d35fb | |||
2aaaa9f47f | |||
cb3e313e21 | |||
6f49a180e3 | |||
af8c3a484c |
9
.gitignore
vendored
9
.gitignore
vendored
@ -23,10 +23,13 @@ dist
|
||||
.pytest_cache
|
||||
venv
|
||||
|
||||
flask_session
|
||||
instance
|
||||
|
||||
.DS_Store
|
||||
.idea
|
||||
|
||||
instance
|
||||
flask_session
|
||||
|
||||
.scannerwork
|
||||
sonar-project.properties
|
||||
|
||||
excludes
|
||||
|
122
README.rst
122
README.rst
@ -12,6 +12,10 @@ views.
|
||||
|
||||
HTTP Digest Authentication is specified in `RFC 2617`_.
|
||||
|
||||
|
||||
Why HTTP Digest Authentication?
|
||||
-------------------------------
|
||||
|
||||
HTTP Digest Authentication has the advantage that it does not send the
|
||||
actual password to the server, which greatly enhances the security.
|
||||
It uses the challenge-response authentication scheme. The client
|
||||
@ -28,6 +32,46 @@ Flask-Digest-Auth works with Flask-Login_. Log in protection can be
|
||||
separated with the authentication mechanism. You can create protected
|
||||
Flask modules without knowing the actual authentication mechanisms.
|
||||
|
||||
|
||||
Features
|
||||
--------
|
||||
|
||||
There are a couple of Flask HTTP digest authentication
|
||||
implementations. Flask-Digest-Auth has the following features:
|
||||
|
||||
|
||||
Flask-Login Integration
|
||||
#######################
|
||||
|
||||
Flask-Digest-Auth features Flask-Login integration. The views
|
||||
can be totally independent with the actual authentication mechanism.
|
||||
You can write a Flask module that requires log in, without specify
|
||||
the actual authentication mechanism. The application can specify
|
||||
either HTTP Digest Authentication, or the log in forms, as needed.
|
||||
|
||||
|
||||
Session Integration
|
||||
###################
|
||||
|
||||
Flask-Digest-Auth features session integration. The user log in
|
||||
is remembered in the session. The authentication information is not
|
||||
requested again. This is different to the practice of the HTTP Digest
|
||||
Authentication, but is convenient for the log in accounting.
|
||||
|
||||
|
||||
Log Out Support
|
||||
###############
|
||||
|
||||
Flask-Digest-Auth supports log out. The user will be prompted for
|
||||
new username and password.
|
||||
|
||||
|
||||
Log In Bookkeeping
|
||||
##################
|
||||
|
||||
You can register a callback to run when the user logs in.
|
||||
|
||||
|
||||
.. _HTTP Digest Authentication: https://en.wikipedia.org/wiki/Digest_access_authentication
|
||||
.. _RFC 2617: https://www.rfc-editor.org/rfc/rfc2617
|
||||
.. _Flask: https://flask.palletsprojects.com
|
||||
@ -60,6 +104,8 @@ Flask-Digest-Auth Alone
|
||||
|
||||
Flask-Digest-Auth can authenticate the users alone.
|
||||
|
||||
The currently logged-in user can be retrieved at ``g.user``, if any.
|
||||
|
||||
|
||||
Example for Simple Applications with Flask-Digest-Auth Alone
|
||||
------------------------------------------------------------
|
||||
@ -75,6 +121,7 @@ In your ``my_app.py``:
|
||||
... (Configure the Flask application) ...
|
||||
|
||||
auth: DigestAuth = DigestAuth(realm="Admin")
|
||||
auth.init_app(app)
|
||||
|
||||
@auth.register_get_password
|
||||
def get_password_hash(username: str) -> t.Optional[str]:
|
||||
@ -87,7 +134,7 @@ In your ``my_app.py``:
|
||||
@app.get("/admin")
|
||||
@auth.login_required
|
||||
def admin():
|
||||
... (Process the view) ...
|
||||
return f"Hello, {g.user.username}!"
|
||||
|
||||
@app.post("/logout")
|
||||
@auth.login_required
|
||||
@ -113,6 +160,7 @@ In your ``my_app/__init__.py``:
|
||||
... (Configure the Flask application) ...
|
||||
|
||||
auth.realm = app.config["REALM"]
|
||||
auth.init_app(app)
|
||||
|
||||
@auth.register_get_password
|
||||
def get_password_hash(username: str) -> t.Optional[str]:
|
||||
@ -136,7 +184,7 @@ In your ``my_app/views.py``:
|
||||
@bp.get("/admin")
|
||||
@auth.login_required
|
||||
def admin():
|
||||
... (Process the view) ...
|
||||
return f"Hello, {g.user.username}!"
|
||||
|
||||
@app.post("/logout")
|
||||
@auth.login_required
|
||||
@ -156,6 +204,12 @@ module that requires log in, without specifying the authentication
|
||||
mechanism. The Flask application can specify the actual
|
||||
authentication mechanism as it sees fit.
|
||||
|
||||
``login_manager.init_app(app)`` must be called before
|
||||
``auth.init_app(app)``.
|
||||
|
||||
The currently logged-in user can be retrieved at
|
||||
``flask_login.current_user``, if any.
|
||||
|
||||
|
||||
Example for Simple Applications with Flask-Login Integration
|
||||
------------------------------------------------------------
|
||||
@ -164,14 +218,14 @@ In your ``my_app.py``:
|
||||
|
||||
::
|
||||
|
||||
import flask_login
|
||||
from flask import Flask, request, redirect
|
||||
from flask_digest_auth import DigestAuth
|
||||
from flask_login import LoginManager
|
||||
|
||||
app: flask = Flask(__name__)
|
||||
... (Configure the Flask application) ...
|
||||
|
||||
login_manager: LoginManager = LoginManager()
|
||||
login_manager: flask_login.LoginManager = flask_login.LoginManager()
|
||||
login_manager.init_app(app)
|
||||
|
||||
@login_manager.user_loader
|
||||
@ -186,9 +240,9 @@ In your ``my_app.py``:
|
||||
... (Load the password hash) ...
|
||||
|
||||
@app.get("/admin")
|
||||
@login_manager.login_required
|
||||
@flask_login.login_required
|
||||
def admin():
|
||||
... (Process the view) ...
|
||||
return f"Hello, {flask_login.current_user.get_id()}!"
|
||||
|
||||
@app.post("/logout")
|
||||
@flask_login.login_required
|
||||
@ -244,7 +298,7 @@ In your ``my_app/views.py``:
|
||||
@bp.get("/admin")
|
||||
@flask_login.login_required
|
||||
def admin():
|
||||
... (Process the view) ...
|
||||
return f"Hello, {flask_login.current_user.get_id()}!"
|
||||
|
||||
@app.post("/logout")
|
||||
@flask_login.login_required
|
||||
@ -288,14 +342,30 @@ the next browser automatic authentication to fail, forcing the browser
|
||||
to ask the user for the username and password again.
|
||||
|
||||
|
||||
Log In Bookkeeping
|
||||
==================
|
||||
|
||||
You can register a callback to run when the user logs in, for ex.,
|
||||
logging the log in event, adding the log in counter, etc.
|
||||
|
||||
::
|
||||
|
||||
@auth.register_on_login
|
||||
def on_login(user: User) -> None:
|
||||
user.visits = user.visits + 1
|
||||
|
||||
|
||||
Writing Tests
|
||||
=============
|
||||
|
||||
You can write tests with our test client that handles HTTP Digest
|
||||
Authentication. Example for a unittest testcase:
|
||||
Authentication.
|
||||
|
||||
Example for a unittest_ test case:
|
||||
|
||||
::
|
||||
|
||||
from flask import Flask
|
||||
from flask_digest_auth import Client
|
||||
from flask_testing import TestCase
|
||||
from my_app import create_app
|
||||
@ -318,6 +388,41 @@ Authentication. Example for a unittest testcase:
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
|
||||
|
||||
Example for a pytest_ test:
|
||||
|
||||
::
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
from flask_digest_auth import Client
|
||||
from my_app import create_app
|
||||
|
||||
@pytest.fixture()
|
||||
def app():
|
||||
app: Flask = create_app({
|
||||
"SECRET_KEY": token_urlsafe(32),
|
||||
"TESTING": True
|
||||
})
|
||||
app.test_client_class = Client
|
||||
yield app
|
||||
|
||||
@pytest.fixture()
|
||||
def client(app):
|
||||
return app.test_client()
|
||||
|
||||
def test_admin(app: Flask, client: Client):
|
||||
with app.app_context():
|
||||
response = self.client.get("/admin")
|
||||
assert response.status_code == 401
|
||||
response = self.client.get(
|
||||
"/admin", digest_auth=("my_name", "my_pass"))
|
||||
assert response.status_code == 200
|
||||
|
||||
.. _unittest: https://docs.python.org/3/library/unittest.html
|
||||
.. _pytest: https://pytest.org
|
||||
|
||||
|
||||
Copyright
|
||||
=========
|
||||
|
||||
@ -335,6 +440,7 @@ Copyright
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
|
||||
|
||||
Authors
|
||||
=======
|
||||
|
||||
|
@ -17,7 +17,7 @@
|
||||
|
||||
[metadata]
|
||||
name = flask-digest-auth
|
||||
version = 0.1.1
|
||||
version = 0.2.1
|
||||
author = imacat
|
||||
author_email = imacat@mail.imacat.idv.tw
|
||||
description = The Flask HTTP Digest Authentication project.
|
||||
|
@ -61,6 +61,16 @@ def calc_response(
|
||||
cnonce or nc is missing with the auth or auth-int qop.
|
||||
"""
|
||||
|
||||
def validate_required(field: t.Optional[str], error: str) -> None:
|
||||
"""Validates a required field.
|
||||
|
||||
:param field: The field that is required.
|
||||
:param error: The error message.
|
||||
:return: None.
|
||||
"""
|
||||
if field is None:
|
||||
raise UnauthorizedException(error)
|
||||
|
||||
def calc_ha1() -> str:
|
||||
"""Calculates and returns the first hash.
|
||||
|
||||
@ -68,16 +78,13 @@ def calc_response(
|
||||
:raise UnauthorizedException: When the cnonce is missing with the MD5-sess
|
||||
algorithm.
|
||||
"""
|
||||
if algorithm is None or algorithm == "MD5":
|
||||
return password_hash
|
||||
if algorithm == "MD5-sess":
|
||||
if cnonce is None:
|
||||
raise UnauthorizedException(
|
||||
f"Missing \"cnonce\" with algorithm=\"{algorithm}\"")
|
||||
validate_required(
|
||||
cnonce, f"Missing \"cnonce\" with algorithm=\"{algorithm}\"")
|
||||
return md5(f"{password_hash}:{nonce}:{cnonce}".encode("utf8")) \
|
||||
.hexdigest()
|
||||
raise UnauthorizedException(
|
||||
f"Unsupported algorithm=\"{algorithm}\"")
|
||||
# algorithm is None or algorithm == "MD5"
|
||||
return password_hash
|
||||
|
||||
def calc_ha2() -> str:
|
||||
"""Calculates the second hash.
|
||||
@ -86,30 +93,20 @@ def calc_response(
|
||||
:raise UnauthorizedException: When the body is missing with
|
||||
qop="auth-int".
|
||||
"""
|
||||
if qop is None or qop == "auth":
|
||||
return md5(f"{method}:{uri}".encode("utf8")).hexdigest()
|
||||
if qop == "auth-int":
|
||||
if body is None:
|
||||
raise UnauthorizedException(
|
||||
f"Missing \"body\" with qop=\"{qop}\"")
|
||||
validate_required(body, f"Missing \"body\" with qop=\"{qop}\"")
|
||||
return md5(
|
||||
f"{method}:{uri}:{md5(body).hexdigest()}".encode("utf8")) \
|
||||
.hexdigest()
|
||||
raise UnauthorizedException(f"Unsupported qop=\"{qop}\"")
|
||||
# qop is None or qop == "auth"
|
||||
return md5(f"{method}:{uri}".encode("utf8")).hexdigest()
|
||||
|
||||
ha1: str = calc_ha1()
|
||||
ha2: str = calc_ha2()
|
||||
if qop is None:
|
||||
return md5(f"{ha1}:{nonce}:{ha2}".encode("utf8")).hexdigest()
|
||||
if qop == "auth" or qop == "auth-int":
|
||||
if cnonce is None:
|
||||
raise UnauthorizedException(
|
||||
f"Missing \"cnonce\" with the qop=\"{qop}\"")
|
||||
if nc is None:
|
||||
raise UnauthorizedException(
|
||||
f"Missing \"nc\" with the qop=\"{qop}\"")
|
||||
validate_required(cnonce, f"Missing \"cnonce\" with the qop=\"{qop}\"")
|
||||
validate_required(nc, f"Missing \"nc\" with the qop=\"{qop}\"")
|
||||
return md5(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}".encode("utf8"))\
|
||||
.hexdigest()
|
||||
if cnonce is None:
|
||||
raise UnauthorizedException(
|
||||
f"Unsupported qop=\"{qop}\"")
|
||||
# qop is None
|
||||
return md5(f"{ha1}:{nonce}:{ha2}".encode("utf8")).hexdigest()
|
||||
|
@ -24,11 +24,9 @@ from __future__ import annotations
|
||||
import sys
|
||||
import typing as t
|
||||
from functools import wraps
|
||||
from random import random
|
||||
from secrets import token_urlsafe
|
||||
from secrets import token_urlsafe, randbits
|
||||
|
||||
from flask import g, request, Response, session, abort, Flask, Request, \
|
||||
current_app
|
||||
from flask import g, request, Response, session, abort, Flask, Request
|
||||
from itsdangerous import URLSafeTimedSerializer, BadData
|
||||
from werkzeug.datastructures import Authorization
|
||||
|
||||
@ -36,6 +34,50 @@ from flask_digest_auth.algo import calc_response
|
||||
from flask_digest_auth.exception import UnauthorizedException
|
||||
|
||||
|
||||
class BasePasswordHashGetter:
|
||||
"""The base password hash getter."""
|
||||
|
||||
@staticmethod
|
||||
def __call__(username: str) -> t.Optional[str]:
|
||||
"""Returns the password hash of a user.
|
||||
|
||||
:param username: The username.
|
||||
:return: The password hash, or None if the user does not exist.
|
||||
:raise UnboundLocalError: When the password hash getter function is
|
||||
not registered yet.
|
||||
"""
|
||||
raise UnboundLocalError("The function to return the password hash"
|
||||
" was not registered yet.")
|
||||
|
||||
|
||||
class BaseUserGetter:
|
||||
"""The base user getter."""
|
||||
|
||||
@staticmethod
|
||||
def __call__(username: str) -> t.Optional[t.Any]:
|
||||
"""Returns a user.
|
||||
|
||||
:param username: The username.
|
||||
:return: The user, or None if the user does not exist.
|
||||
:raise UnboundLocalError: When the user getter function is not
|
||||
registered yet.
|
||||
"""
|
||||
raise UnboundLocalError("The function to return the user"
|
||||
" was not registered yet.")
|
||||
|
||||
|
||||
class BaseOnLogInCallback:
|
||||
"""The base callback when the user logs in."""
|
||||
|
||||
@staticmethod
|
||||
def __call__(user: t.Any) -> None:
|
||||
"""Runs the callback when the user logs in.
|
||||
|
||||
:param user: The logged-in user.
|
||||
:return: None.
|
||||
"""
|
||||
|
||||
|
||||
class DigestAuth:
|
||||
"""The HTTP digest authentication."""
|
||||
|
||||
@ -52,9 +94,11 @@ class DigestAuth:
|
||||
self.use_opaque: bool = True
|
||||
self.domain: t.List[str] = []
|
||||
self.qop: t.List[str] = ["auth", "auth-int"]
|
||||
self.__get_password_hash: t.Callable[[str], t.Optional[str]] \
|
||||
= lambda x: None
|
||||
self.__get_user: t.Callable[[str], t.Optional] = lambda x: None
|
||||
self.app: t.Optional[Flask] = None
|
||||
self.__get_password_hash: BasePasswordHashGetter \
|
||||
= BasePasswordHashGetter()
|
||||
self.__get_user: BaseUserGetter = BaseUserGetter()
|
||||
self.__on_login: BaseOnLogInCallback = BaseOnLogInCallback()
|
||||
|
||||
def login_required(self, view) -> t.Callable:
|
||||
"""The view decorator for HTTP digest authentication.
|
||||
@ -66,6 +110,37 @@ class DigestAuth:
|
||||
class NoLogInException(Exception):
|
||||
"""The exception thrown when the user is not authorized."""
|
||||
|
||||
def get_logged_in_user() -> t.Any:
|
||||
"""Returns the currently logged-in user.
|
||||
|
||||
:return: The currently logged-in user.
|
||||
:raise NoLogInException: When the user is not logged in.
|
||||
"""
|
||||
if "user" not in session:
|
||||
raise NoLogInException
|
||||
user: t.Optional[t.Any] = self.__get_user(session["user"])
|
||||
if user is None:
|
||||
del session["user"]
|
||||
raise NoLogInException
|
||||
return user
|
||||
|
||||
def auth_user(state: AuthState) -> t.Any:
|
||||
"""Authenticates a user.
|
||||
|
||||
:param state: The authentication state.
|
||||
:return: The user.
|
||||
:raise UnauthorizedException: When the authentication fails.
|
||||
"""
|
||||
authorization: Authorization = request.authorization
|
||||
if authorization is None:
|
||||
raise UnauthorizedException
|
||||
if authorization.type != "digest":
|
||||
raise UnauthorizedException(
|
||||
"Not an HTTP digest authorization")
|
||||
self.authenticate(state)
|
||||
session["user"] = authorization.username
|
||||
return self.__get_user(authorization.username)
|
||||
|
||||
@wraps(view)
|
||||
def login_required_view(*args, **kwargs) -> t.Any:
|
||||
"""The login-protected view.
|
||||
@ -75,34 +150,24 @@ class DigestAuth:
|
||||
:return: The response.
|
||||
"""
|
||||
try:
|
||||
if "user" not in session:
|
||||
raise NoLogInException
|
||||
user: t.Optional[t.Any] = self.__get_user(session["user"])
|
||||
if user is None:
|
||||
raise NoLogInException
|
||||
g.user = user
|
||||
g.user = get_logged_in_user()
|
||||
return view(*args, **kwargs)
|
||||
except NoLogInException:
|
||||
state: AuthState = AuthState()
|
||||
authorization: Authorization = request.authorization
|
||||
try:
|
||||
if authorization is None:
|
||||
raise UnauthorizedException
|
||||
if authorization.type != "digest":
|
||||
raise UnauthorizedException(
|
||||
"Not an HTTP digest authorization")
|
||||
self.authenticate(state)
|
||||
session["user"] = authorization.username
|
||||
g.user = self.__get_user(authorization.username)
|
||||
return view(*args, **kwargs)
|
||||
except UnauthorizedException as e:
|
||||
if len(e.args) > 0:
|
||||
sys.stderr.write(e.args[0] + "\n")
|
||||
response: Response = Response()
|
||||
response.status = 401
|
||||
response.headers["WWW-Authenticate"] \
|
||||
= self.make_response_header(state)
|
||||
abort(response)
|
||||
pass
|
||||
|
||||
state: AuthState = AuthState()
|
||||
try:
|
||||
g.user = auth_user(state)
|
||||
self.__on_login(g.user)
|
||||
return view(*args, **kwargs)
|
||||
except UnauthorizedException as e:
|
||||
if len(e.args) > 0:
|
||||
sys.stderr.write(e.args[0] + "\n")
|
||||
response: Response = Response()
|
||||
response.status = 401
|
||||
response.headers["WWW-Authenticate"] \
|
||||
= self.make_response_header(state)
|
||||
abort(response)
|
||||
|
||||
return login_required_view
|
||||
|
||||
@ -127,8 +192,8 @@ class DigestAuth:
|
||||
except BadData:
|
||||
raise UnauthorizedException("Invalid opaque")
|
||||
state.opaque = authorization.opaque
|
||||
password_hash: t.Optional[str] = self.__get_password_hash(
|
||||
authorization.username)
|
||||
password_hash: t.Optional[str] \
|
||||
= self.__get_password_hash(authorization.username)
|
||||
if password_hash is None:
|
||||
raise UnauthorizedException(
|
||||
f"No such user \"{authorization.username}\"")
|
||||
@ -157,11 +222,22 @@ class DigestAuth:
|
||||
:param state: The authorization state.
|
||||
:return: The WWW-Authenticate response header.
|
||||
"""
|
||||
opaque: t.Optional[str] = None if not self.use_opaque else \
|
||||
(state.opaque if state.opaque is not None
|
||||
else self.serializer.dumps(random(), salt="opaque"))
|
||||
|
||||
def get_opaque() -> t.Optional[str]:
|
||||
"""Returns the opaque value.
|
||||
|
||||
:return: The opaque value.
|
||||
"""
|
||||
if not self.use_opaque:
|
||||
return None
|
||||
if state.opaque is not None:
|
||||
return state.opaque
|
||||
return self.serializer.dumps(randbits(32), salt="opaque")
|
||||
|
||||
opaque: t.Optional[str] = get_opaque()
|
||||
nonce: str = self.serializer.dumps(
|
||||
random(), salt="nonce" if opaque is None else f"nonce-{opaque}")
|
||||
randbits(32),
|
||||
salt="nonce" if opaque is None else f"nonce-{opaque}")
|
||||
|
||||
header: str = f"Digest realm=\"{self.realm}\""
|
||||
if len(self.domain) > 0:
|
||||
@ -187,7 +263,20 @@ class DigestAuth:
|
||||
hash, or None if the user does not exist.
|
||||
:return: None.
|
||||
"""
|
||||
self.__get_password_hash = func
|
||||
|
||||
class PasswordHashGetter(BasePasswordHashGetter):
|
||||
"""The base password hash getter."""
|
||||
|
||||
@staticmethod
|
||||
def __call__(username: str) -> t.Optional[str]:
|
||||
"""Returns the password hash of a user.
|
||||
|
||||
:param username: The username.
|
||||
:return: The password hash, or None if the user does not exist.
|
||||
"""
|
||||
return func(username)
|
||||
|
||||
self.__get_password_hash = PasswordHashGetter()
|
||||
|
||||
def register_get_user(self, func: t.Callable[[str], t.Optional[t.Any]])\
|
||||
-> None:
|
||||
@ -197,7 +286,41 @@ class DigestAuth:
|
||||
or None if the user does not exist.
|
||||
:return: None.
|
||||
"""
|
||||
self.__get_user = func
|
||||
|
||||
class UserGetter(BaseUserGetter):
|
||||
"""The user getter."""
|
||||
|
||||
@staticmethod
|
||||
def __call__(username: str) -> t.Optional[t.Any]:
|
||||
"""Returns a user.
|
||||
|
||||
:param username: The username.
|
||||
:return: The user, or None if the user does not exist.
|
||||
"""
|
||||
return func(username)
|
||||
|
||||
self.__get_user = UserGetter()
|
||||
|
||||
def register_on_login(self, func: t.Callable[[t.Any], None]) -> None:
|
||||
"""Registers the callback when the user logs in.
|
||||
|
||||
:param func: The callback given the logged-in user.
|
||||
:return: None.
|
||||
"""
|
||||
|
||||
class OnLogInCallback:
|
||||
"""The callback when the user logs in."""
|
||||
|
||||
@staticmethod
|
||||
def __call__(user: t.Any) -> None:
|
||||
"""Runs the callback when the user logs in.
|
||||
|
||||
:param user: The logged-in user.
|
||||
:return: None.
|
||||
"""
|
||||
func(user)
|
||||
|
||||
self.__on_login = OnLogInCallback()
|
||||
|
||||
def init_app(self, app: Flask) -> None:
|
||||
"""Initializes the Flask application.
|
||||
@ -205,13 +328,12 @@ class DigestAuth:
|
||||
:param app: The Flask application.
|
||||
:return: None.
|
||||
"""
|
||||
app.digest_auth = self
|
||||
self.app = app
|
||||
|
||||
try:
|
||||
if hasattr(app, "login_manager"):
|
||||
from flask_login import LoginManager, login_user
|
||||
|
||||
if not hasattr(app, "login_manager"):
|
||||
raise AttributeError(
|
||||
"Please run the Flask-Login init-app() first")
|
||||
login_manager: LoginManager = getattr(app, "login_manager")
|
||||
|
||||
@login_manager.unauthorized_handler
|
||||
@ -246,18 +368,14 @@ class DigestAuth:
|
||||
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
|
||||
|
||||
except ModuleNotFoundError:
|
||||
raise ModuleNotFoundError(
|
||||
"init_app() is only for Flask-Login integration")
|
||||
|
||||
@staticmethod
|
||||
def logout() -> None:
|
||||
def logout(self) -> None:
|
||||
"""Logs out the user.
|
||||
This actually causes the next authentication to fail, which forces
|
||||
the browser to ask the user for the username and password again.
|
||||
@ -267,7 +385,7 @@ class DigestAuth:
|
||||
if "user" in session:
|
||||
del session["user"]
|
||||
try:
|
||||
if hasattr(current_app, "login_manager"):
|
||||
if hasattr(self.app, "login_manager"):
|
||||
from flask_login import logout_user
|
||||
logout_user()
|
||||
except ModuleNotFoundError:
|
||||
|
@ -66,9 +66,8 @@ class Client(WerkzeugClient):
|
||||
:return: The request authorization.
|
||||
"""
|
||||
qop: t.Optional[t.Literal["auth", "auth-int"]] = None
|
||||
if www_authenticate.qop is not None:
|
||||
if "auth" in www_authenticate.qop:
|
||||
qop = "auth"
|
||||
if www_authenticate.qop is not None and "auth" in www_authenticate.qop:
|
||||
qop = "auth"
|
||||
|
||||
cnonce: t.Optional[str] = None
|
||||
if qop is not None or www_authenticate.algorithm == "MD5-sess":
|
||||
|
@ -20,7 +20,6 @@
|
||||
"""
|
||||
import typing as t
|
||||
from secrets import token_urlsafe
|
||||
from types import SimpleNamespace
|
||||
|
||||
from flask import Response, Flask, g, redirect, request
|
||||
from flask_testing import TestCase
|
||||
@ -33,6 +32,21 @@ _USERNAME: str = "Mufasa"
|
||||
_PASSWORD: str = "Circle Of Life"
|
||||
|
||||
|
||||
class User:
|
||||
"""A dummy user"""
|
||||
|
||||
def __init__(self, username: str, password: str):
|
||||
"""Constructs a dummy user.
|
||||
|
||||
:param username: The username.
|
||||
:param password: The clear-text password.
|
||||
"""
|
||||
self.username: str = username
|
||||
self.password_hash: str = make_password_hash(
|
||||
_REALM, username, password)
|
||||
self.visits: int = 0
|
||||
|
||||
|
||||
class AuthenticationTestCase(TestCase):
|
||||
"""The test case for the HTTP digest authentication."""
|
||||
|
||||
@ -49,8 +63,9 @@ class AuthenticationTestCase(TestCase):
|
||||
app.test_client_class = Client
|
||||
|
||||
auth: DigestAuth = DigestAuth(realm=_REALM)
|
||||
user_db: t.Dict[str, str] \
|
||||
= {_USERNAME: make_password_hash(_REALM, _USERNAME, _PASSWORD)}
|
||||
auth.init_app(app)
|
||||
self.user: User = User(_USERNAME, _PASSWORD)
|
||||
user_db: t.Dict[str, User] = {_USERNAME: self.user}
|
||||
|
||||
@auth.register_get_password
|
||||
def get_password_hash(username: str) -> t.Optional[str]:
|
||||
@ -59,7 +74,8 @@ class AuthenticationTestCase(TestCase):
|
||||
:param username: The username.
|
||||
:return: The password hash, or None if the user does not exist.
|
||||
"""
|
||||
return user_db[username] if username in user_db else None
|
||||
return user_db[username].password_hash if username in user_db \
|
||||
else None
|
||||
|
||||
@auth.register_get_user
|
||||
def get_user(username: str) -> t.Optional[t.Any]:
|
||||
@ -68,8 +84,16 @@ class AuthenticationTestCase(TestCase):
|
||||
:param username: The username.
|
||||
:return: The user, or None if the user does not exist.
|
||||
"""
|
||||
return SimpleNamespace(username=username) if username in user_db \
|
||||
else None
|
||||
return user_db[username] if username in user_db else None
|
||||
|
||||
@auth.register_on_login
|
||||
def on_login(user: User):
|
||||
"""The callback when the user logs in.
|
||||
|
||||
:param user: The logged-in user.
|
||||
:return: None.
|
||||
"""
|
||||
user.visits = user.visits + 1
|
||||
|
||||
@app.get("/admin-1/auth", endpoint="admin-1")
|
||||
@auth.login_required
|
||||
@ -117,6 +141,7 @@ class AuthenticationTestCase(TestCase):
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.data.decode("UTF-8"),
|
||||
f"Hello, {_USERNAME}! #2")
|
||||
self.assertEqual(self.user.visits, 1)
|
||||
|
||||
def test_stale_opaque(self) -> None:
|
||||
"""Tests the stale and opaque value.
|
||||
@ -193,3 +218,4 @@ class AuthenticationTestCase(TestCase):
|
||||
|
||||
response = self.client.get(admin_uri)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(self.user.visits, 2)
|
||||
|
@ -35,12 +35,16 @@ _PASSWORD: str = "Circle Of Life"
|
||||
class User:
|
||||
"""A dummy user."""
|
||||
|
||||
def __init__(self, username: str):
|
||||
def __init__(self, username: str, password: str):
|
||||
"""Constructs a dummy user.
|
||||
|
||||
:param username: The username.
|
||||
:param password: The clear-text password.
|
||||
"""
|
||||
self.username: str = username
|
||||
self.password_hash: str = make_password_hash(
|
||||
_REALM, username, password)
|
||||
self.visits: int = 0
|
||||
self.is_authenticated: bool = True
|
||||
self.is_active: bool = True
|
||||
self.is_anonymous: bool = False
|
||||
@ -82,8 +86,8 @@ class FlaskLoginTestCase(TestCase):
|
||||
auth: DigestAuth = DigestAuth(realm=_REALM)
|
||||
auth.init_app(app)
|
||||
|
||||
user_db: t.Dict[str, str] \
|
||||
= {_USERNAME: make_password_hash(_REALM, _USERNAME, _PASSWORD)}
|
||||
self.user: User = User(_USERNAME, _PASSWORD)
|
||||
user_db: t.Dict[str, User] = {_USERNAME: self.user}
|
||||
|
||||
@auth.register_get_password
|
||||
def get_password_hash(username: str) -> t.Optional[str]:
|
||||
@ -92,7 +96,17 @@ class FlaskLoginTestCase(TestCase):
|
||||
:param username: The username.
|
||||
:return: The password hash, or None if the user does not exist.
|
||||
"""
|
||||
return user_db[username] if username in user_db else None
|
||||
return user_db[username].password_hash if username in user_db \
|
||||
else None
|
||||
|
||||
@auth.register_on_login
|
||||
def on_login(user: User):
|
||||
"""The callback when the user logs in.
|
||||
|
||||
:param user: The logged-in user.
|
||||
:return: None.
|
||||
"""
|
||||
user.visits = user.visits + 1
|
||||
|
||||
@login_manager.user_loader
|
||||
def load_user(user_id: str) -> t.Optional[User]:
|
||||
@ -101,7 +115,7 @@ class FlaskLoginTestCase(TestCase):
|
||||
:param user_id: The username.
|
||||
:return: The user, or None if the user does not exist.
|
||||
"""
|
||||
return User(user_id) if user_id in user_db else None
|
||||
return user_db[user_id] if user_id in user_db else None
|
||||
|
||||
@app.get("/admin-1/auth", endpoint="admin-1")
|
||||
@flask_login.login_required
|
||||
@ -110,7 +124,7 @@ class FlaskLoginTestCase(TestCase):
|
||||
|
||||
:return: The response.
|
||||
"""
|
||||
return f"Hello, {flask_login.current_user.username}! #1"
|
||||
return f"Hello, {flask_login.current_user.get_id()}! #1"
|
||||
|
||||
@app.get("/admin-2/auth", endpoint="admin-2")
|
||||
@flask_login.login_required
|
||||
@ -119,7 +133,7 @@ class FlaskLoginTestCase(TestCase):
|
||||
|
||||
:return: The response.
|
||||
"""
|
||||
return f"Hello, {flask_login.current_user.username}! #2"
|
||||
return f"Hello, {flask_login.current_user.get_id()}! #2"
|
||||
|
||||
@app.post("/logout", endpoint="logout")
|
||||
@flask_login.login_required
|
||||
@ -139,7 +153,7 @@ class FlaskLoginTestCase(TestCase):
|
||||
:return: None.
|
||||
"""
|
||||
if not self.has_flask_login:
|
||||
self.skipTest("Skipped testing Flask-Login integration without it.")
|
||||
self.skipTest("Skipped without Flask-Login.")
|
||||
|
||||
response: Response = self.client.get(self.app.url_for("admin-1"))
|
||||
self.assertEqual(response.status_code, 401)
|
||||
@ -152,6 +166,7 @@ class FlaskLoginTestCase(TestCase):
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.data.decode("UTF-8"),
|
||||
f"Hello, {_USERNAME}! #2")
|
||||
self.assertEqual(self.user.visits, 1)
|
||||
|
||||
def test_stale_opaque(self) -> None:
|
||||
"""Tests the stale and opaque value.
|
||||
@ -159,7 +174,7 @@ class FlaskLoginTestCase(TestCase):
|
||||
:return: None.
|
||||
"""
|
||||
if not self.has_flask_login:
|
||||
self.skipTest("Skipped testing Flask-Login integration without it.")
|
||||
self.skipTest("Skipped without Flask-Login.")
|
||||
|
||||
admin_uri: str = self.app.url_for("admin-1")
|
||||
response: Response
|
||||
@ -206,6 +221,9 @@ class FlaskLoginTestCase(TestCase):
|
||||
|
||||
:return: None.
|
||||
"""
|
||||
if not self.has_flask_login:
|
||||
self.skipTest("Skipped without Flask-Login.")
|
||||
|
||||
admin_uri: str = self.app.url_for("admin-1")
|
||||
logout_uri: str = self.app.url_for("logout")
|
||||
response: Response
|
||||
@ -237,3 +255,4 @@ class FlaskLoginTestCase(TestCase):
|
||||
|
||||
response = self.client.get(admin_uri)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(self.user.visits, 2)
|
||||
|
Reference in New Issue
Block a user