Compare commits
	
		
			37 Commits
		
	
	
		
			v0.1.0
			...
			495e9a9785
		
	
	| 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 | |||
| 65c3322ecc | |||
| cb5cfaf7d4 | |||
| dda8472a76 | |||
| 177f549786 | |||
| ff8ada129d | |||
| ccec1365bf | |||
| 78514a8f17 | |||
| 3dcc409bef | |||
| 2156aa710f | |||
| a43c6aea89 | |||
| 8e29c91f92 | |||
| 8ec1e6fd02 | |||
| b37861eccc | |||
| f2e51b46cb | 
							
								
								
									
										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 | ||||
|   | ||||
							
								
								
									
										240
									
								
								README.rst
									
									
									
									
									
								
							
							
						
						
									
										240
									
								
								README.rst
									
									
									
									
									
								
							| @@ -6,12 +6,74 @@ Flask HTTP Digest Authentication | ||||
| Description | ||||
| =========== | ||||
|  | ||||
| *Flask-Digest-Auth* is an HTTP Digest Authentication implementation | ||||
| *Flask-Digest-Auth* is an `HTTP Digest Authentication`_ implementation | ||||
| for Flask_ applications.  It authenticates the user for the protected | ||||
| views.  It works with Flask-Login_, so that log in protection can be | ||||
| separated with the authentication mechanism.  You can write Flask | ||||
| modules that work with different authentication mechanisms. | ||||
| 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 | ||||
| returns the response calculated from the challenge and the password, | ||||
| but not the original password. | ||||
|  | ||||
| Log in forms has the advantage of freedom, in the senses of both the | ||||
| visual design and the actual implementation.  You may implement your | ||||
| own challenge-response log in form, but then you are reinventing the | ||||
| wheels.  If a pretty log in form is not critical to your project, HTTP | ||||
| Digest Authentication should be a good choice. | ||||
|  | ||||
| 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 | ||||
| .. _Flask-Login: https://flask-login.readthedocs.io | ||||
|  | ||||
| @@ -19,14 +81,14 @@ modules that work with different authentication mechanisms. | ||||
| Installation | ||||
| ============ | ||||
|  | ||||
| It's suggested that you install with ``pip``: | ||||
| You can install Flask-Digest-Auth with ``pip``: | ||||
|  | ||||
| :: | ||||
|  | ||||
|     pip install flask-digest-auth | ||||
|     pip install Flask-Digest-Auth | ||||
|  | ||||
| You may also install from the latest source from the | ||||
| `flask-digest-auth Github repository`_. | ||||
| You may also install the latest source from the | ||||
| `Flask-Digest-Auth GitHub repository`_. | ||||
|  | ||||
| :: | ||||
|  | ||||
| @@ -34,7 +96,7 @@ You may also install from the latest source from the | ||||
|     cd flask-digest-auth | ||||
|     pip install . | ||||
|  | ||||
| .. _flask-digest-auth Github repository: https://github.com/imacat/flask-digest-auth | ||||
| .. _Flask-Digest-Auth GitHub repository: https://github.com/imacat/flask-digest-auth | ||||
|  | ||||
|  | ||||
| Flask-Digest-Auth Alone | ||||
| @@ -42,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 | ||||
| ------------------------------------------------------------ | ||||
| @@ -50,13 +114,14 @@ In your ``my_app.py``: | ||||
|  | ||||
| :: | ||||
|  | ||||
|     from flask import Flask | ||||
|     from flask import Flask, request, redirect | ||||
|     from flask_digest_auth import DigestAuth | ||||
|  | ||||
|     app: flask = Flask(__name__) | ||||
|     ... (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]: | ||||
| @@ -69,7 +134,13 @@ 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 | ||||
|     def logout(): | ||||
|         auth.logout() | ||||
|         return redirect(request.form.get("next")) | ||||
|  | ||||
|  | ||||
| Example for Larger Applications with ``create_app()`` with Flask-Digest-Auth Alone | ||||
| @@ -77,7 +148,7 @@ Example for Larger Applications with ``create_app()`` with Flask-Digest-Auth Alo | ||||
|  | ||||
| In your ``my_app/__init__.py``: | ||||
|  | ||||
| ::: | ||||
| :: | ||||
|  | ||||
|     from flask import Flask | ||||
|     from flask_digest_auth import DigestAuth | ||||
| @@ -89,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]: | ||||
| @@ -105,14 +177,20 @@ In your ``my_app/views.py``: | ||||
| :: | ||||
|  | ||||
|     from my_app import auth | ||||
|     from flask import Flask, Blueprint | ||||
|     from flask import Flask, Blueprint, request, redirect | ||||
|  | ||||
|     bp = Blueprint("admin", __name__, url_prefix="/admin") | ||||
|  | ||||
|     @bp.get("/") | ||||
|     @bp.get("/admin") | ||||
|     @auth.login_required | ||||
|     def admin(): | ||||
|         ... (Process the view) ... | ||||
|         return f"Hello, {g.user.username}!" | ||||
|  | ||||
|     @app.post("/logout") | ||||
|     @auth.login_required | ||||
|     def logout(): | ||||
|         auth.logout() | ||||
|         return redirect(request.form.get("next")) | ||||
|  | ||||
|     def init_app(app: Flask) -> None: | ||||
|         app.register_blueprint(bp) | ||||
| @@ -126,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 | ||||
| ------------------------------------------------------------ | ||||
| @@ -134,14 +218,14 @@ In your ``my_app.py``: | ||||
|  | ||||
| :: | ||||
|  | ||||
|     from flask import Flask | ||||
|     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 | ||||
| @@ -156,9 +240,16 @@ 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 | ||||
|     def logout(): | ||||
|         auth.logout() | ||||
|         # Do not call flask_login.logout_user() | ||||
|         return redirect(request.form.get("next")) | ||||
|  | ||||
|  | ||||
| Example for Larger Applications with ``create_app()`` with Flask-Login Integration | ||||
| @@ -166,12 +257,14 @@ Example for Larger Applications with ``create_app()`` with Flask-Login Integrati | ||||
|  | ||||
| In your ``my_app/__init__.py``: | ||||
|  | ||||
| ::: | ||||
| :: | ||||
|  | ||||
|     from flask import Flask | ||||
|     from flask_digest_auth import DigestAuth | ||||
|     from flask_login import LoginManager | ||||
|  | ||||
|     auth: DigestAuth = DigestAuth() | ||||
|  | ||||
|     def create_app(test_config = None) -> Flask: | ||||
|         app: flask = Flask(__name__) | ||||
|         ... (Configure the Flask application) ... | ||||
| @@ -183,7 +276,7 @@ In your ``my_app/__init__.py``: | ||||
|         def load_user(user_id: str) -> t.Optional[User]: | ||||
|             ... (Load the user with the username) ... | ||||
|  | ||||
|         auth: DigestAuth = DigestAuth(realm=app.config["REALM"]) | ||||
|         auth.realm = app.config["REALM"] | ||||
|         auth.init_app(app) | ||||
|  | ||||
|         @auth.register_get_password | ||||
| @@ -197,33 +290,82 @@ In your ``my_app/views.py``: | ||||
| :: | ||||
|  | ||||
|     import flask_login | ||||
|     from flask import Flask, Blueprint | ||||
|     from flask import Flask, Blueprint, request, redirect | ||||
|     from my_app import auth | ||||
|  | ||||
|     bp = Blueprint("admin", __name__, url_prefix="/admin") | ||||
|  | ||||
|     @bp.get("/") | ||||
|     @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 | ||||
|     def logout(): | ||||
|         auth.logout() | ||||
|         # Do not call flask_login.logout_user() | ||||
|         return redirect(request.form.get("next")) | ||||
|  | ||||
|     def init_app(app: Flask) -> None: | ||||
|         app.register_blueprint(bp) | ||||
|  | ||||
| The views only depend on Flask-Login, but not its underlying | ||||
| authentication mechanism.  You can always change the | ||||
| authentication mechanism without changing the views, or release a | ||||
| protected Flask module without specifying the authentication | ||||
| mechanism. | ||||
| The views only depend on Flask-Login, but not the actual | ||||
| authentication mechanism.  You can change the actual authentication | ||||
| mechanism without changing the views. | ||||
|  | ||||
|  | ||||
| Setting the Password Hash | ||||
| ========================= | ||||
|  | ||||
| The password hash of the HTTP Digest Authentication is composed of the | ||||
| realm, the username, and the password.  Example for setting the | ||||
| password: | ||||
|  | ||||
| :: | ||||
|  | ||||
|     from flask_digest_auth import make_password_hash | ||||
|  | ||||
|     user.password = make_password_hash(realm, username, password) | ||||
|  | ||||
| The username is part of the hash.  If the user changes their username, | ||||
| you need to ask their password, to generate and store the new password | ||||
| hash. | ||||
|  | ||||
|  | ||||
| Log Out | ||||
| ======= | ||||
|  | ||||
| Call ``auth.logout()`` when the user wants to log out. | ||||
| Besides the usual log out routine, ``auth.logout()`` actually causes | ||||
| 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 | ||||
| @@ -246,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 | ||||
| ========= | ||||
|  | ||||
| @@ -263,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.0 | ||||
| version = 0.2.1 | ||||
| author = imacat | ||||
| author_email = imacat@mail.imacat.idv.tw | ||||
| description = The Flask HTTP Digest Authentication project. | ||||
| @@ -38,7 +38,7 @@ classifiers = | ||||
| package_dir = | ||||
|     = src | ||||
| packages = find: | ||||
| python_requires = >=3.10 | ||||
| python_requires = >=3.7 | ||||
| install_requires = | ||||
|   flask | ||||
| tests_require = | ||||
|   | ||||
| @@ -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,8 +24,7 @@ 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 | ||||
| from itsdangerous import URLSafeTimedSerializer, BadData | ||||
| @@ -35,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.""" | ||||
|  | ||||
| @@ -51,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. | ||||
| @@ -65,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. | ||||
| @@ -74,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 | ||||
|  | ||||
| @@ -112,6 +178,9 @@ class DigestAuth: | ||||
|         :return: None. | ||||
|         :raise UnauthorizedException: When the authentication failed. | ||||
|         """ | ||||
|         if "digest_auth_logout" in session: | ||||
|             del session["digest_auth_logout"] | ||||
|             raise UnauthorizedException("Logging out") | ||||
|         authorization: Authorization = request.authorization | ||||
|         if self.use_opaque: | ||||
|             if authorization.opaque is None: | ||||
| @@ -123,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}\"") | ||||
| @@ -153,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: | ||||
| @@ -183,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: | ||||
| @@ -193,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. | ||||
| @@ -201,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 | ||||
| @@ -242,15 +368,29 @@ 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 | ||||
|  | ||||
|     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. | ||||
|  | ||||
|         :return: None. | ||||
|         """ | ||||
|         if "user" in session: | ||||
|             del session["user"] | ||||
|         try: | ||||
|             if hasattr(self.app, "login_manager"): | ||||
|                 from flask_login import logout_user | ||||
|                 logout_user() | ||||
|         except ModuleNotFoundError: | ||||
|             raise ModuleNotFoundError( | ||||
|                 "init_app() is only for Flask-Login integration") | ||||
|             pass | ||||
|         session["digest_auth_logout"] = True | ||||
|  | ||||
|  | ||||
| class AuthState: | ||||
|   | ||||
| @@ -49,14 +49,14 @@ class Client(WerkzeugClient): | ||||
|             return response | ||||
|         if hasattr(g, "_login_user"): | ||||
|             delattr(g, "_login_user") | ||||
|         auth_data: Authorization = self.__class__.__make_authorization( | ||||
|         auth_data: Authorization = self.__class__.make_authorization( | ||||
|             www_authenticate, args[0], digest_auth[0], digest_auth[1]) | ||||
|         response = super(Client, self).open(*args, auth=auth_data, **kwargs) | ||||
|         return response | ||||
|  | ||||
|     @staticmethod | ||||
|     def __make_authorization(www_authenticate: WWWAuthenticate, uri: str, | ||||
|                              username: str, password: str) -> Authorization: | ||||
|     def make_authorization(www_authenticate: WWWAuthenticate, uri: str, | ||||
|                            username: str, password: str) -> Authorization: | ||||
|         """Composes and returns the request authorization. | ||||
|  | ||||
|         :param www_authenticate: The WWW-Authenticate response. | ||||
| @@ -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,10 +20,10 @@ | ||||
| """ | ||||
| import typing as t | ||||
| from secrets import token_urlsafe | ||||
| from types import SimpleNamespace | ||||
|  | ||||
| from flask import Response, Flask, g | ||||
| from flask import Response, Flask, g, redirect, request | ||||
| from flask_testing import TestCase | ||||
| from werkzeug.datastructures import WWWAuthenticate, Authorization | ||||
|  | ||||
| from flask_digest_auth import DigestAuth, make_password_hash, Client | ||||
|  | ||||
| @@ -32,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.""" | ||||
|  | ||||
| @@ -48,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]: | ||||
| @@ -58,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]: | ||||
| @@ -67,27 +84,45 @@ 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 | ||||
|  | ||||
|         @app.get("/login-required-1/auth", endpoint="auth-1") | ||||
|         @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 | ||||
|         def login_required_1() -> str: | ||||
|             """The first dummy view. | ||||
|         def admin_1() -> str: | ||||
|             """The first administration section. | ||||
|  | ||||
|             :return: The response. | ||||
|             """ | ||||
|             return f"Hello, {g.user.username}! #1" | ||||
|  | ||||
|         @app.get("/login-required-2/auth", endpoint="auth-2") | ||||
|         @app.get("/admin-2/auth", endpoint="admin-2") | ||||
|         @auth.login_required | ||||
|         def login_required_2() -> str: | ||||
|             """The second dummy view. | ||||
|         def admin_2() -> str: | ||||
|             """The second administration section. | ||||
|  | ||||
|             :return: The response. | ||||
|             """ | ||||
|             return f"Hello, {g.user.username}! #2" | ||||
|  | ||||
|         @app.post("/logout", endpoint="logout") | ||||
|         @auth.login_required | ||||
|         def logout() -> redirect: | ||||
|             """Logs out the user. | ||||
|  | ||||
|             :return: The response. | ||||
|             """ | ||||
|             auth.logout() | ||||
|             return redirect(request.form.get("next")) | ||||
|  | ||||
|         return app | ||||
|  | ||||
|     def test_auth(self) -> None: | ||||
| @@ -95,14 +130,92 @@ class AuthenticationTestCase(TestCase): | ||||
|  | ||||
|         :return: None. | ||||
|         """ | ||||
|         response: Response = self.client.get(self.app.url_for("auth-1")) | ||||
|         response: Response = self.client.get(self.app.url_for("admin-1")) | ||||
|         self.assertEqual(response.status_code, 401) | ||||
|         response = self.client.get( | ||||
|             self.app.url_for("auth-1"), digest_auth=(_USERNAME, _PASSWORD)) | ||||
|             self.app.url_for("admin-1"), digest_auth=(_USERNAME, _PASSWORD)) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|         self.assertEqual(response.data.decode("UTF-8"), | ||||
|                          f"Hello, {_USERNAME}! #1") | ||||
|         response: Response = self.client.get(self.app.url_for("auth-2")) | ||||
|         response: Response = self.client.get(self.app.url_for("admin-2")) | ||||
|         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. | ||||
|  | ||||
|         :return: None. | ||||
|         """ | ||||
|         admin_uri: str = self.app.url_for("admin-1") | ||||
|         response: Response | ||||
|         www_authenticate: WWWAuthenticate | ||||
|         auth_data: Authorization | ||||
|  | ||||
|         response = super(Client, self.client).get(admin_uri) | ||||
|         self.assertEqual(response.status_code, 401) | ||||
|         www_authenticate = response.www_authenticate | ||||
|         self.assertEqual(www_authenticate.type, "digest") | ||||
|         self.assertEqual(www_authenticate.stale, None) | ||||
|         opaque: str = www_authenticate.opaque | ||||
|  | ||||
|         www_authenticate.nonce = "bad" | ||||
|         auth_data = Client.make_authorization( | ||||
|             www_authenticate, admin_uri, _USERNAME, _PASSWORD) | ||||
|         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.opaque, opaque) | ||||
|  | ||||
|         auth_data = Client.make_authorization( | ||||
|             www_authenticate, admin_uri, _USERNAME, _PASSWORD + "2") | ||||
|         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.opaque, opaque) | ||||
|  | ||||
|         auth_data = Client.make_authorization( | ||||
|             www_authenticate, admin_uri, _USERNAME, _PASSWORD) | ||||
|         response = super(Client, self.client).get(admin_uri, auth=auth_data) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|  | ||||
|     def test_logout(self) -> None: | ||||
|         """Tests the logging out. | ||||
|  | ||||
|         :return: None. | ||||
|         """ | ||||
|         admin_uri: str = self.app.url_for("admin-1") | ||||
|         logout_uri: str = self.app.url_for("logout") | ||||
|         response: Response | ||||
|  | ||||
|         response = self.client.get(admin_uri) | ||||
|         self.assertEqual(response.status_code, 401) | ||||
|  | ||||
|         response = self.client.get(admin_uri, | ||||
|                                    digest_auth=(_USERNAME, _PASSWORD)) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|  | ||||
|         response = self.client.get(admin_uri) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|  | ||||
|         response = self.client.post(logout_uri, data={"next": admin_uri}) | ||||
|         self.assertEqual(response.status_code, 302) | ||||
|         self.assertEqual(response.location, admin_uri) | ||||
|  | ||||
|         response = self.client.get(admin_uri) | ||||
|         self.assertEqual(response.status_code, 401) | ||||
|  | ||||
|         response = self.client.get(admin_uri, | ||||
|                                    digest_auth=(_USERNAME, _PASSWORD)) | ||||
|         self.assertEqual(response.status_code, 401) | ||||
|  | ||||
|         response = self.client.get(admin_uri, | ||||
|                                    digest_auth=(_USERNAME, _PASSWORD)) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|  | ||||
|         response = self.client.get(admin_uri) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|         self.assertEqual(self.user.visits, 2) | ||||
|   | ||||
| @@ -21,8 +21,9 @@ | ||||
| import typing as t | ||||
| from secrets import token_urlsafe | ||||
|  | ||||
| from flask import Response, Flask | ||||
| from flask import Response, Flask, g, redirect, request | ||||
| from flask_testing import TestCase | ||||
| from werkzeug.datastructures import WWWAuthenticate, Authorization | ||||
|  | ||||
| from flask_digest_auth import DigestAuth, make_password_hash, Client | ||||
|  | ||||
| @@ -32,8 +33,18 @@ _PASSWORD: str = "Circle Of Life" | ||||
|  | ||||
|  | ||||
| class User: | ||||
|     def __init__(self, username: str): | ||||
|     """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 | ||||
|         self.is_authenticated: bool = True | ||||
|         self.is_active: bool = True | ||||
|         self.is_anonymous: bool = False | ||||
| @@ -75,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]: | ||||
| @@ -85,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]: | ||||
| @@ -94,25 +115,35 @@ 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("/login-required-1/auth", endpoint="auth-1") | ||||
|         @app.get("/admin-1/auth", endpoint="admin-1") | ||||
|         @flask_login.login_required | ||||
|         def login_required_1() -> str: | ||||
|             """The first dummy view. | ||||
|         def admin_1() -> str: | ||||
|             """The first administration section. | ||||
|  | ||||
|             :return: The response. | ||||
|             """ | ||||
|             return f"Hello, {flask_login.current_user.username}! #1" | ||||
|             return f"Hello, {flask_login.current_user.get_id()}! #1" | ||||
|  | ||||
|         @app.get("/login-required-2/auth", endpoint="auth-2") | ||||
|         @app.get("/admin-2/auth", endpoint="admin-2") | ||||
|         @flask_login.login_required | ||||
|         def login_required_2() -> str: | ||||
|             """The second dummy view. | ||||
|         def admin_2() -> str: | ||||
|             """The second administration section. | ||||
|  | ||||
|             :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 | ||||
|         def logout() -> redirect: | ||||
|             """Logs out the user. | ||||
|  | ||||
|             :return: The response. | ||||
|             """ | ||||
|             auth.logout() | ||||
|             return redirect(request.form.get("next")) | ||||
|  | ||||
|         return app | ||||
|  | ||||
| @@ -122,16 +153,106 @@ 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("auth-1")) | ||||
|         response: Response = self.client.get(self.app.url_for("admin-1")) | ||||
|         self.assertEqual(response.status_code, 401) | ||||
|         response = self.client.get( | ||||
|             self.app.url_for("auth-1"), digest_auth=(_USERNAME, _PASSWORD)) | ||||
|             self.app.url_for("admin-1"), digest_auth=(_USERNAME, _PASSWORD)) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|         self.assertEqual(response.data.decode("UTF-8"), | ||||
|                          f"Hello, {_USERNAME}! #1") | ||||
|         response: Response = self.client.get(self.app.url_for("auth-2")) | ||||
|         response: Response = self.client.get(self.app.url_for("admin-2")) | ||||
|         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. | ||||
|  | ||||
|         :return: None. | ||||
|         """ | ||||
|         if not self.has_flask_login: | ||||
|             self.skipTest("Skipped without Flask-Login.") | ||||
|  | ||||
|         admin_uri: str = self.app.url_for("admin-1") | ||||
|         response: Response | ||||
|         www_authenticate: WWWAuthenticate | ||||
|         auth_data: Authorization | ||||
|  | ||||
|         response = super(Client, self.client).get(admin_uri) | ||||
|         self.assertEqual(response.status_code, 401) | ||||
|         www_authenticate = response.www_authenticate | ||||
|         self.assertEqual(www_authenticate.type, "digest") | ||||
|         self.assertEqual(www_authenticate.stale, None) | ||||
|         opaque: str = www_authenticate.opaque | ||||
|  | ||||
|         if hasattr(g, "_login_user"): | ||||
|             delattr(g, "_login_user") | ||||
|         www_authenticate.nonce = "bad" | ||||
|         auth_data = Client.make_authorization( | ||||
|             www_authenticate, admin_uri, _USERNAME, _PASSWORD) | ||||
|         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.opaque, opaque) | ||||
|  | ||||
|         if hasattr(g, "_login_user"): | ||||
|             delattr(g, "_login_user") | ||||
|         auth_data = Client.make_authorization( | ||||
|             www_authenticate, admin_uri, _USERNAME, _PASSWORD + "2") | ||||
|         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.opaque, opaque) | ||||
|  | ||||
|         if hasattr(g, "_login_user"): | ||||
|             delattr(g, "_login_user") | ||||
|         auth_data = Client.make_authorization( | ||||
|             www_authenticate, admin_uri, _USERNAME, _PASSWORD) | ||||
|         response = super(Client, self.client).get(admin_uri, auth=auth_data) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|  | ||||
|     def test_logout(self) -> None: | ||||
|         """Tests the logging out. | ||||
|  | ||||
|         :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 | ||||
|  | ||||
|         response = self.client.get(admin_uri) | ||||
|         self.assertEqual(response.status_code, 401) | ||||
|  | ||||
|         response = self.client.get(admin_uri, | ||||
|                                    digest_auth=(_USERNAME, _PASSWORD)) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|  | ||||
|         response = self.client.get(admin_uri) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|  | ||||
|         response = self.client.post(logout_uri, data={"next": admin_uri}) | ||||
|         self.assertEqual(response.status_code, 302) | ||||
|         self.assertEqual(response.location, admin_uri) | ||||
|  | ||||
|         response = self.client.get(admin_uri) | ||||
|         self.assertEqual(response.status_code, 401) | ||||
|  | ||||
|         response = self.client.get(admin_uri, | ||||
|                                    digest_auth=(_USERNAME, _PASSWORD)) | ||||
|         self.assertEqual(response.status_code, 401) | ||||
|  | ||||
|         response = self.client.get(admin_uri, | ||||
|                                    digest_auth=(_USERNAME, _PASSWORD)) | ||||
|         self.assertEqual(response.status_code, 200) | ||||
|  | ||||
|         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