Compare commits
37 Commits
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
|
.pytest_cache
|
||||||
venv
|
venv
|
||||||
|
|
||||||
flask_session
|
|
||||||
instance
|
|
||||||
|
|
||||||
.DS_Store
|
.DS_Store
|
||||||
.idea
|
.idea
|
||||||
|
|
||||||
|
instance
|
||||||
|
flask_session
|
||||||
|
|
||||||
|
.scannerwork
|
||||||
|
sonar-project.properties
|
||||||
|
|
||||||
excludes
|
excludes
|
||||||
|
240
README.rst
240
README.rst
@ -6,12 +6,74 @@ Flask HTTP Digest Authentication
|
|||||||
Description
|
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
|
for Flask_ applications. It authenticates the user for the protected
|
||||||
views. It works with Flask-Login_, so that log in protection can be
|
views.
|
||||||
separated with the authentication mechanism. You can write Flask
|
|
||||||
modules that work with different authentication mechanisms.
|
|
||||||
|
|
||||||
|
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: https://flask.palletsprojects.com
|
||||||
.. _Flask-Login: https://flask-login.readthedocs.io
|
.. _Flask-Login: https://flask-login.readthedocs.io
|
||||||
|
|
||||||
@ -19,14 +81,14 @@ modules that work with different authentication mechanisms.
|
|||||||
Installation
|
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
|
You may also install the latest source from the
|
||||||
`flask-digest-auth Github repository`_.
|
`Flask-Digest-Auth GitHub repository`_.
|
||||||
|
|
||||||
::
|
::
|
||||||
|
|
||||||
@ -34,7 +96,7 @@ You may also install from the latest source from the
|
|||||||
cd flask-digest-auth
|
cd flask-digest-auth
|
||||||
pip install .
|
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
|
Flask-Digest-Auth Alone
|
||||||
@ -42,6 +104,8 @@ Flask-Digest-Auth Alone
|
|||||||
|
|
||||||
Flask-Digest-Auth can authenticate the users 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
|
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
|
from flask_digest_auth import DigestAuth
|
||||||
|
|
||||||
app: flask = Flask(__name__)
|
app: flask = Flask(__name__)
|
||||||
... (Configure the Flask application) ...
|
... (Configure the Flask application) ...
|
||||||
|
|
||||||
auth: DigestAuth = DigestAuth(realm="Admin")
|
auth: DigestAuth = DigestAuth(realm="Admin")
|
||||||
|
auth.init_app(app)
|
||||||
|
|
||||||
@auth.register_get_password
|
@auth.register_get_password
|
||||||
def get_password_hash(username: str) -> t.Optional[str]:
|
def get_password_hash(username: str) -> t.Optional[str]:
|
||||||
@ -69,7 +134,13 @@ In your ``my_app.py``:
|
|||||||
@app.get("/admin")
|
@app.get("/admin")
|
||||||
@auth.login_required
|
@auth.login_required
|
||||||
def admin():
|
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
|
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``:
|
In your ``my_app/__init__.py``:
|
||||||
|
|
||||||
:::
|
::
|
||||||
|
|
||||||
from flask import Flask
|
from flask import Flask
|
||||||
from flask_digest_auth import DigestAuth
|
from flask_digest_auth import DigestAuth
|
||||||
@ -89,6 +160,7 @@ In your ``my_app/__init__.py``:
|
|||||||
... (Configure the Flask application) ...
|
... (Configure the Flask application) ...
|
||||||
|
|
||||||
auth.realm = app.config["REALM"]
|
auth.realm = app.config["REALM"]
|
||||||
|
auth.init_app(app)
|
||||||
|
|
||||||
@auth.register_get_password
|
@auth.register_get_password
|
||||||
def get_password_hash(username: str) -> t.Optional[str]:
|
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 my_app import auth
|
||||||
from flask import Flask, Blueprint
|
from flask import Flask, Blueprint, request, redirect
|
||||||
|
|
||||||
bp = Blueprint("admin", __name__, url_prefix="/admin")
|
bp = Blueprint("admin", __name__, url_prefix="/admin")
|
||||||
|
|
||||||
@bp.get("/")
|
@bp.get("/admin")
|
||||||
@auth.login_required
|
@auth.login_required
|
||||||
def admin():
|
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:
|
def init_app(app: Flask) -> None:
|
||||||
app.register_blueprint(bp)
|
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
|
mechanism. The Flask application can specify the actual
|
||||||
authentication mechanism as it sees fit.
|
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
|
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_digest_auth import DigestAuth
|
||||||
from flask_login import LoginManager
|
|
||||||
|
|
||||||
app: flask = Flask(__name__)
|
app: flask = Flask(__name__)
|
||||||
... (Configure the Flask application) ...
|
... (Configure the Flask application) ...
|
||||||
|
|
||||||
login_manager: LoginManager = LoginManager()
|
login_manager: flask_login.LoginManager = flask_login.LoginManager()
|
||||||
login_manager.init_app(app)
|
login_manager.init_app(app)
|
||||||
|
|
||||||
@login_manager.user_loader
|
@login_manager.user_loader
|
||||||
@ -156,9 +240,16 @@ In your ``my_app.py``:
|
|||||||
... (Load the password hash) ...
|
... (Load the password hash) ...
|
||||||
|
|
||||||
@app.get("/admin")
|
@app.get("/admin")
|
||||||
@login_manager.login_required
|
@flask_login.login_required
|
||||||
def admin():
|
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
|
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``:
|
In your ``my_app/__init__.py``:
|
||||||
|
|
||||||
:::
|
::
|
||||||
|
|
||||||
from flask import Flask
|
from flask import Flask
|
||||||
from flask_digest_auth import DigestAuth
|
from flask_digest_auth import DigestAuth
|
||||||
from flask_login import LoginManager
|
from flask_login import LoginManager
|
||||||
|
|
||||||
|
auth: DigestAuth = DigestAuth()
|
||||||
|
|
||||||
def create_app(test_config = None) -> Flask:
|
def create_app(test_config = None) -> Flask:
|
||||||
app: flask = Flask(__name__)
|
app: flask = Flask(__name__)
|
||||||
... (Configure the Flask application) ...
|
... (Configure the Flask application) ...
|
||||||
@ -183,7 +276,7 @@ In your ``my_app/__init__.py``:
|
|||||||
def load_user(user_id: str) -> t.Optional[User]:
|
def load_user(user_id: str) -> t.Optional[User]:
|
||||||
... (Load the user with the username) ...
|
... (Load the user with the username) ...
|
||||||
|
|
||||||
auth: DigestAuth = DigestAuth(realm=app.config["REALM"])
|
auth.realm = app.config["REALM"]
|
||||||
auth.init_app(app)
|
auth.init_app(app)
|
||||||
|
|
||||||
@auth.register_get_password
|
@auth.register_get_password
|
||||||
@ -197,33 +290,82 @@ In your ``my_app/views.py``:
|
|||||||
::
|
::
|
||||||
|
|
||||||
import flask_login
|
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 = Blueprint("admin", __name__, url_prefix="/admin")
|
||||||
|
|
||||||
@bp.get("/")
|
@bp.get("/admin")
|
||||||
@flask_login.login_required
|
@flask_login.login_required
|
||||||
def admin():
|
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:
|
def init_app(app: Flask) -> None:
|
||||||
app.register_blueprint(bp)
|
app.register_blueprint(bp)
|
||||||
|
|
||||||
The views only depend on Flask-Login, but not its underlying
|
The views only depend on Flask-Login, but not the actual
|
||||||
authentication mechanism. You can always change the
|
authentication mechanism. You can change the actual authentication
|
||||||
authentication mechanism without changing the views, or release a
|
mechanism without changing the views.
|
||||||
protected Flask module without specifying the authentication
|
|
||||||
mechanism.
|
|
||||||
|
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
|
Writing Tests
|
||||||
=============
|
=============
|
||||||
|
|
||||||
You can write tests with our test client that handles HTTP Digest
|
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_digest_auth import Client
|
||||||
from flask_testing import TestCase
|
from flask_testing import TestCase
|
||||||
from my_app import create_app
|
from my_app import create_app
|
||||||
@ -246,6 +388,41 @@ Authentication. Example for a unittest testcase:
|
|||||||
self.assertEqual(response.status_code, 200)
|
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
|
Copyright
|
||||||
=========
|
=========
|
||||||
|
|
||||||
@ -263,6 +440,7 @@ Copyright
|
|||||||
See the License for the specific language governing permissions and
|
See the License for the specific language governing permissions and
|
||||||
limitations under the License.
|
limitations under the License.
|
||||||
|
|
||||||
|
|
||||||
Authors
|
Authors
|
||||||
=======
|
=======
|
||||||
|
|
||||||
|
@ -17,7 +17,7 @@
|
|||||||
|
|
||||||
[metadata]
|
[metadata]
|
||||||
name = flask-digest-auth
|
name = flask-digest-auth
|
||||||
version = 0.1.0
|
version = 0.2.1
|
||||||
author = imacat
|
author = imacat
|
||||||
author_email = imacat@mail.imacat.idv.tw
|
author_email = imacat@mail.imacat.idv.tw
|
||||||
description = The Flask HTTP Digest Authentication project.
|
description = The Flask HTTP Digest Authentication project.
|
||||||
@ -38,7 +38,7 @@ classifiers =
|
|||||||
package_dir =
|
package_dir =
|
||||||
= src
|
= src
|
||||||
packages = find:
|
packages = find:
|
||||||
python_requires = >=3.10
|
python_requires = >=3.7
|
||||||
install_requires =
|
install_requires =
|
||||||
flask
|
flask
|
||||||
tests_require =
|
tests_require =
|
||||||
|
@ -61,6 +61,16 @@ def calc_response(
|
|||||||
cnonce or nc is missing with the auth or auth-int qop.
|
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:
|
def calc_ha1() -> str:
|
||||||
"""Calculates and returns the first hash.
|
"""Calculates and returns the first hash.
|
||||||
|
|
||||||
@ -68,16 +78,13 @@ def calc_response(
|
|||||||
:raise UnauthorizedException: When the cnonce is missing with the MD5-sess
|
:raise UnauthorizedException: When the cnonce is missing with the MD5-sess
|
||||||
algorithm.
|
algorithm.
|
||||||
"""
|
"""
|
||||||
if algorithm is None or algorithm == "MD5":
|
|
||||||
return password_hash
|
|
||||||
if algorithm == "MD5-sess":
|
if algorithm == "MD5-sess":
|
||||||
if cnonce is None:
|
validate_required(
|
||||||
raise UnauthorizedException(
|
cnonce, f"Missing \"cnonce\" with algorithm=\"{algorithm}\"")
|
||||||
f"Missing \"cnonce\" with algorithm=\"{algorithm}\"")
|
|
||||||
return md5(f"{password_hash}:{nonce}:{cnonce}".encode("utf8")) \
|
return md5(f"{password_hash}:{nonce}:{cnonce}".encode("utf8")) \
|
||||||
.hexdigest()
|
.hexdigest()
|
||||||
raise UnauthorizedException(
|
# algorithm is None or algorithm == "MD5"
|
||||||
f"Unsupported algorithm=\"{algorithm}\"")
|
return password_hash
|
||||||
|
|
||||||
def calc_ha2() -> str:
|
def calc_ha2() -> str:
|
||||||
"""Calculates the second hash.
|
"""Calculates the second hash.
|
||||||
@ -86,30 +93,20 @@ def calc_response(
|
|||||||
:raise UnauthorizedException: When the body is missing with
|
:raise UnauthorizedException: When the body is missing with
|
||||||
qop="auth-int".
|
qop="auth-int".
|
||||||
"""
|
"""
|
||||||
if qop is None or qop == "auth":
|
|
||||||
return md5(f"{method}:{uri}".encode("utf8")).hexdigest()
|
|
||||||
if qop == "auth-int":
|
if qop == "auth-int":
|
||||||
if body is None:
|
validate_required(body, f"Missing \"body\" with qop=\"{qop}\"")
|
||||||
raise UnauthorizedException(
|
|
||||||
f"Missing \"body\" with qop=\"{qop}\"")
|
|
||||||
return md5(
|
return md5(
|
||||||
f"{method}:{uri}:{md5(body).hexdigest()}".encode("utf8")) \
|
f"{method}:{uri}:{md5(body).hexdigest()}".encode("utf8")) \
|
||||||
.hexdigest()
|
.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()
|
ha1: str = calc_ha1()
|
||||||
ha2: str = calc_ha2()
|
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 qop == "auth" or qop == "auth-int":
|
||||||
if cnonce is None:
|
validate_required(cnonce, f"Missing \"cnonce\" with the qop=\"{qop}\"")
|
||||||
raise UnauthorizedException(
|
validate_required(nc, f"Missing \"nc\" with the qop=\"{qop}\"")
|
||||||
f"Missing \"cnonce\" with the qop=\"{qop}\"")
|
|
||||||
if nc is None:
|
|
||||||
raise UnauthorizedException(
|
|
||||||
f"Missing \"nc\" with the qop=\"{qop}\"")
|
|
||||||
return md5(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}".encode("utf8"))\
|
return md5(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}".encode("utf8"))\
|
||||||
.hexdigest()
|
.hexdigest()
|
||||||
if cnonce is None:
|
# qop is None
|
||||||
raise UnauthorizedException(
|
return md5(f"{ha1}:{nonce}:{ha2}".encode("utf8")).hexdigest()
|
||||||
f"Unsupported qop=\"{qop}\"")
|
|
||||||
|
@ -24,8 +24,7 @@ from __future__ import annotations
|
|||||||
import sys
|
import sys
|
||||||
import typing as t
|
import typing as t
|
||||||
from functools import wraps
|
from functools import wraps
|
||||||
from random import random
|
from secrets import token_urlsafe, randbits
|
||||||
from secrets import token_urlsafe
|
|
||||||
|
|
||||||
from flask import g, request, Response, session, abort, Flask, Request
|
from flask import g, request, Response, session, abort, Flask, Request
|
||||||
from itsdangerous import URLSafeTimedSerializer, BadData
|
from itsdangerous import URLSafeTimedSerializer, BadData
|
||||||
@ -35,6 +34,50 @@ from flask_digest_auth.algo import calc_response
|
|||||||
from flask_digest_auth.exception import UnauthorizedException
|
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:
|
class DigestAuth:
|
||||||
"""The HTTP digest authentication."""
|
"""The HTTP digest authentication."""
|
||||||
|
|
||||||
@ -51,9 +94,11 @@ class DigestAuth:
|
|||||||
self.use_opaque: bool = True
|
self.use_opaque: bool = True
|
||||||
self.domain: t.List[str] = []
|
self.domain: t.List[str] = []
|
||||||
self.qop: t.List[str] = ["auth", "auth-int"]
|
self.qop: t.List[str] = ["auth", "auth-int"]
|
||||||
self.__get_password_hash: t.Callable[[str], t.Optional[str]] \
|
self.app: t.Optional[Flask] = None
|
||||||
= lambda x: None
|
self.__get_password_hash: BasePasswordHashGetter \
|
||||||
self.__get_user: t.Callable[[str], t.Optional] = lambda x: None
|
= BasePasswordHashGetter()
|
||||||
|
self.__get_user: BaseUserGetter = BaseUserGetter()
|
||||||
|
self.__on_login: BaseOnLogInCallback = BaseOnLogInCallback()
|
||||||
|
|
||||||
def login_required(self, view) -> t.Callable:
|
def login_required(self, view) -> t.Callable:
|
||||||
"""The view decorator for HTTP digest authentication.
|
"""The view decorator for HTTP digest authentication.
|
||||||
@ -65,6 +110,37 @@ class DigestAuth:
|
|||||||
class NoLogInException(Exception):
|
class NoLogInException(Exception):
|
||||||
"""The exception thrown when the user is not authorized."""
|
"""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)
|
@wraps(view)
|
||||||
def login_required_view(*args, **kwargs) -> t.Any:
|
def login_required_view(*args, **kwargs) -> t.Any:
|
||||||
"""The login-protected view.
|
"""The login-protected view.
|
||||||
@ -74,34 +150,24 @@ class DigestAuth:
|
|||||||
:return: The response.
|
:return: The response.
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
if "user" not in session:
|
g.user = get_logged_in_user()
|
||||||
raise NoLogInException
|
|
||||||
user: t.Optional[t.Any] = self.__get_user(session["user"])
|
|
||||||
if user is None:
|
|
||||||
raise NoLogInException
|
|
||||||
g.user = user
|
|
||||||
return view(*args, **kwargs)
|
return view(*args, **kwargs)
|
||||||
except NoLogInException:
|
except NoLogInException:
|
||||||
state: AuthState = AuthState()
|
pass
|
||||||
authorization: Authorization = request.authorization
|
|
||||||
try:
|
state: AuthState = AuthState()
|
||||||
if authorization is None:
|
try:
|
||||||
raise UnauthorizedException
|
g.user = auth_user(state)
|
||||||
if authorization.type != "digest":
|
self.__on_login(g.user)
|
||||||
raise UnauthorizedException(
|
return view(*args, **kwargs)
|
||||||
"Not an HTTP digest authorization")
|
except UnauthorizedException as e:
|
||||||
self.authenticate(state)
|
if len(e.args) > 0:
|
||||||
session["user"] = authorization.username
|
sys.stderr.write(e.args[0] + "\n")
|
||||||
g.user = self.__get_user(authorization.username)
|
response: Response = Response()
|
||||||
return view(*args, **kwargs)
|
response.status = 401
|
||||||
except UnauthorizedException as e:
|
response.headers["WWW-Authenticate"] \
|
||||||
if len(e.args) > 0:
|
= self.make_response_header(state)
|
||||||
sys.stderr.write(e.args[0] + "\n")
|
abort(response)
|
||||||
response: Response = Response()
|
|
||||||
response.status = 401
|
|
||||||
response.headers["WWW-Authenticate"] \
|
|
||||||
= self.make_response_header(state)
|
|
||||||
abort(response)
|
|
||||||
|
|
||||||
return login_required_view
|
return login_required_view
|
||||||
|
|
||||||
@ -112,6 +178,9 @@ class DigestAuth:
|
|||||||
:return: None.
|
:return: None.
|
||||||
:raise UnauthorizedException: When the authentication failed.
|
: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
|
authorization: Authorization = request.authorization
|
||||||
if self.use_opaque:
|
if self.use_opaque:
|
||||||
if authorization.opaque is None:
|
if authorization.opaque is None:
|
||||||
@ -123,8 +192,8 @@ class DigestAuth:
|
|||||||
except BadData:
|
except BadData:
|
||||||
raise UnauthorizedException("Invalid opaque")
|
raise UnauthorizedException("Invalid opaque")
|
||||||
state.opaque = authorization.opaque
|
state.opaque = authorization.opaque
|
||||||
password_hash: t.Optional[str] = self.__get_password_hash(
|
password_hash: t.Optional[str] \
|
||||||
authorization.username)
|
= self.__get_password_hash(authorization.username)
|
||||||
if password_hash is None:
|
if password_hash is None:
|
||||||
raise UnauthorizedException(
|
raise UnauthorizedException(
|
||||||
f"No such user \"{authorization.username}\"")
|
f"No such user \"{authorization.username}\"")
|
||||||
@ -153,11 +222,22 @@ class DigestAuth:
|
|||||||
:param state: The authorization state.
|
:param state: The authorization state.
|
||||||
:return: The WWW-Authenticate response header.
|
: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
|
def get_opaque() -> t.Optional[str]:
|
||||||
else self.serializer.dumps(random(), salt="opaque"))
|
"""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(
|
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}\""
|
header: str = f"Digest realm=\"{self.realm}\""
|
||||||
if len(self.domain) > 0:
|
if len(self.domain) > 0:
|
||||||
@ -183,7 +263,20 @@ class DigestAuth:
|
|||||||
hash, or None if the user does not exist.
|
hash, or None if the user does not exist.
|
||||||
:return: None.
|
: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]])\
|
def register_get_user(self, func: t.Callable[[str], t.Optional[t.Any]])\
|
||||||
-> None:
|
-> None:
|
||||||
@ -193,7 +286,41 @@ class DigestAuth:
|
|||||||
or None if the user does not exist.
|
or None if the user does not exist.
|
||||||
:return: None.
|
: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:
|
def init_app(self, app: Flask) -> None:
|
||||||
"""Initializes the Flask application.
|
"""Initializes the Flask application.
|
||||||
@ -201,13 +328,12 @@ class DigestAuth:
|
|||||||
:param app: The Flask application.
|
:param app: The Flask application.
|
||||||
:return: None.
|
:return: None.
|
||||||
"""
|
"""
|
||||||
|
app.digest_auth = self
|
||||||
|
self.app = app
|
||||||
|
|
||||||
try:
|
if hasattr(app, "login_manager"):
|
||||||
from flask_login import LoginManager, login_user
|
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: LoginManager = getattr(app, "login_manager")
|
||||||
|
|
||||||
@login_manager.unauthorized_handler
|
@login_manager.unauthorized_handler
|
||||||
@ -242,15 +368,29 @@ class DigestAuth:
|
|||||||
user = login_manager.user_callback(
|
user = login_manager.user_callback(
|
||||||
authorization.username)
|
authorization.username)
|
||||||
login_user(user)
|
login_user(user)
|
||||||
|
self.__on_login(user)
|
||||||
return user
|
return user
|
||||||
except UnauthorizedException as e:
|
except UnauthorizedException as e:
|
||||||
if str(e) != "":
|
if str(e) != "":
|
||||||
app.logger.warning(str(e))
|
app.logger.warning(str(e))
|
||||||
return None
|
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:
|
except ModuleNotFoundError:
|
||||||
raise ModuleNotFoundError(
|
pass
|
||||||
"init_app() is only for Flask-Login integration")
|
session["digest_auth_logout"] = True
|
||||||
|
|
||||||
|
|
||||||
class AuthState:
|
class AuthState:
|
||||||
|
@ -49,14 +49,14 @@ class Client(WerkzeugClient):
|
|||||||
return response
|
return response
|
||||||
if hasattr(g, "_login_user"):
|
if hasattr(g, "_login_user"):
|
||||||
delattr(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])
|
www_authenticate, args[0], digest_auth[0], digest_auth[1])
|
||||||
response = super(Client, self).open(*args, auth=auth_data, **kwargs)
|
response = super(Client, self).open(*args, auth=auth_data, **kwargs)
|
||||||
return response
|
return response
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def __make_authorization(www_authenticate: WWWAuthenticate, uri: str,
|
def make_authorization(www_authenticate: WWWAuthenticate, uri: str,
|
||||||
username: str, password: str) -> Authorization:
|
username: str, password: str) -> Authorization:
|
||||||
"""Composes and returns the request authorization.
|
"""Composes and returns the request authorization.
|
||||||
|
|
||||||
:param www_authenticate: The WWW-Authenticate response.
|
:param www_authenticate: The WWW-Authenticate response.
|
||||||
@ -66,9 +66,8 @@ class Client(WerkzeugClient):
|
|||||||
:return: The request authorization.
|
:return: The request authorization.
|
||||||
"""
|
"""
|
||||||
qop: t.Optional[t.Literal["auth", "auth-int"]] = None
|
qop: t.Optional[t.Literal["auth", "auth-int"]] = None
|
||||||
if www_authenticate.qop is not None:
|
if www_authenticate.qop is not None and "auth" in www_authenticate.qop:
|
||||||
if "auth" in www_authenticate.qop:
|
qop = "auth"
|
||||||
qop = "auth"
|
|
||||||
|
|
||||||
cnonce: t.Optional[str] = None
|
cnonce: t.Optional[str] = None
|
||||||
if qop is not None or www_authenticate.algorithm == "MD5-sess":
|
if qop is not None or www_authenticate.algorithm == "MD5-sess":
|
||||||
|
@ -20,10 +20,10 @@
|
|||||||
"""
|
"""
|
||||||
import typing as t
|
import typing as t
|
||||||
from secrets import token_urlsafe
|
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 flask_testing import TestCase
|
||||||
|
from werkzeug.datastructures import WWWAuthenticate, Authorization
|
||||||
|
|
||||||
from flask_digest_auth import DigestAuth, make_password_hash, Client
|
from flask_digest_auth import DigestAuth, make_password_hash, Client
|
||||||
|
|
||||||
@ -32,6 +32,21 @@ _USERNAME: str = "Mufasa"
|
|||||||
_PASSWORD: str = "Circle Of Life"
|
_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):
|
class AuthenticationTestCase(TestCase):
|
||||||
"""The test case for the HTTP digest authentication."""
|
"""The test case for the HTTP digest authentication."""
|
||||||
|
|
||||||
@ -48,8 +63,9 @@ class AuthenticationTestCase(TestCase):
|
|||||||
app.test_client_class = Client
|
app.test_client_class = Client
|
||||||
|
|
||||||
auth: DigestAuth = DigestAuth(realm=_REALM)
|
auth: DigestAuth = DigestAuth(realm=_REALM)
|
||||||
user_db: t.Dict[str, str] \
|
auth.init_app(app)
|
||||||
= {_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
|
@auth.register_get_password
|
||||||
def get_password_hash(username: str) -> t.Optional[str]:
|
def get_password_hash(username: str) -> t.Optional[str]:
|
||||||
@ -58,7 +74,8 @@ class AuthenticationTestCase(TestCase):
|
|||||||
:param username: The username.
|
:param username: The username.
|
||||||
:return: The password hash, or None if the user does not exist.
|
: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
|
@auth.register_get_user
|
||||||
def get_user(username: str) -> t.Optional[t.Any]:
|
def get_user(username: str) -> t.Optional[t.Any]:
|
||||||
@ -67,27 +84,45 @@ class AuthenticationTestCase(TestCase):
|
|||||||
:param username: The username.
|
:param username: The username.
|
||||||
:return: The user, or None if the user does not exist.
|
:return: The user, or None if the user does not exist.
|
||||||
"""
|
"""
|
||||||
return SimpleNamespace(username=username) if username in user_db \
|
return user_db[username] if username in user_db else None
|
||||||
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
|
@auth.login_required
|
||||||
def login_required_1() -> str:
|
def admin_1() -> str:
|
||||||
"""The first dummy view.
|
"""The first administration section.
|
||||||
|
|
||||||
:return: The response.
|
:return: The response.
|
||||||
"""
|
"""
|
||||||
return f"Hello, {g.user.username}! #1"
|
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
|
@auth.login_required
|
||||||
def login_required_2() -> str:
|
def admin_2() -> str:
|
||||||
"""The second dummy view.
|
"""The second administration section.
|
||||||
|
|
||||||
:return: The response.
|
:return: The response.
|
||||||
"""
|
"""
|
||||||
return f"Hello, {g.user.username}! #2"
|
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
|
return app
|
||||||
|
|
||||||
def test_auth(self) -> None:
|
def test_auth(self) -> None:
|
||||||
@ -95,14 +130,92 @@ class AuthenticationTestCase(TestCase):
|
|||||||
|
|
||||||
:return: None.
|
: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)
|
self.assertEqual(response.status_code, 401)
|
||||||
response = self.client.get(
|
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.status_code, 200)
|
||||||
self.assertEqual(response.data.decode("UTF-8"),
|
self.assertEqual(response.data.decode("UTF-8"),
|
||||||
f"Hello, {_USERNAME}! #1")
|
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.status_code, 200)
|
||||||
self.assertEqual(response.data.decode("UTF-8"),
|
self.assertEqual(response.data.decode("UTF-8"),
|
||||||
f"Hello, {_USERNAME}! #2")
|
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
|
import typing as t
|
||||||
from secrets import token_urlsafe
|
from secrets import token_urlsafe
|
||||||
|
|
||||||
from flask import Response, Flask
|
from flask import Response, Flask, g, redirect, request
|
||||||
from flask_testing import TestCase
|
from flask_testing import TestCase
|
||||||
|
from werkzeug.datastructures import WWWAuthenticate, Authorization
|
||||||
|
|
||||||
from flask_digest_auth import DigestAuth, make_password_hash, Client
|
from flask_digest_auth import DigestAuth, make_password_hash, Client
|
||||||
|
|
||||||
@ -32,8 +33,18 @@ _PASSWORD: str = "Circle Of Life"
|
|||||||
|
|
||||||
|
|
||||||
class User:
|
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.username: str = username
|
||||||
|
self.password_hash: str = make_password_hash(
|
||||||
|
_REALM, username, password)
|
||||||
|
self.visits: int = 0
|
||||||
self.is_authenticated: bool = True
|
self.is_authenticated: bool = True
|
||||||
self.is_active: bool = True
|
self.is_active: bool = True
|
||||||
self.is_anonymous: bool = False
|
self.is_anonymous: bool = False
|
||||||
@ -75,8 +86,8 @@ class FlaskLoginTestCase(TestCase):
|
|||||||
auth: DigestAuth = DigestAuth(realm=_REALM)
|
auth: DigestAuth = DigestAuth(realm=_REALM)
|
||||||
auth.init_app(app)
|
auth.init_app(app)
|
||||||
|
|
||||||
user_db: t.Dict[str, str] \
|
self.user: User = User(_USERNAME, _PASSWORD)
|
||||||
= {_USERNAME: make_password_hash(_REALM, _USERNAME, _PASSWORD)}
|
user_db: t.Dict[str, User] = {_USERNAME: self.user}
|
||||||
|
|
||||||
@auth.register_get_password
|
@auth.register_get_password
|
||||||
def get_password_hash(username: str) -> t.Optional[str]:
|
def get_password_hash(username: str) -> t.Optional[str]:
|
||||||
@ -85,7 +96,17 @@ class FlaskLoginTestCase(TestCase):
|
|||||||
:param username: The username.
|
:param username: The username.
|
||||||
:return: The password hash, or None if the user does not exist.
|
: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
|
@login_manager.user_loader
|
||||||
def load_user(user_id: str) -> t.Optional[User]:
|
def load_user(user_id: str) -> t.Optional[User]:
|
||||||
@ -94,25 +115,35 @@ class FlaskLoginTestCase(TestCase):
|
|||||||
:param user_id: The username.
|
:param user_id: The username.
|
||||||
:return: The user, or None if the user does not exist.
|
: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
|
@flask_login.login_required
|
||||||
def login_required_1() -> str:
|
def admin_1() -> str:
|
||||||
"""The first dummy view.
|
"""The first administration section.
|
||||||
|
|
||||||
:return: The response.
|
: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
|
@flask_login.login_required
|
||||||
def login_required_2() -> str:
|
def admin_2() -> str:
|
||||||
"""The second dummy view.
|
"""The second administration section.
|
||||||
|
|
||||||
:return: The response.
|
: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
|
return app
|
||||||
|
|
||||||
@ -122,16 +153,106 @@ class FlaskLoginTestCase(TestCase):
|
|||||||
:return: None.
|
:return: None.
|
||||||
"""
|
"""
|
||||||
if not self.has_flask_login:
|
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)
|
self.assertEqual(response.status_code, 401)
|
||||||
response = self.client.get(
|
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.status_code, 200)
|
||||||
self.assertEqual(response.data.decode("UTF-8"),
|
self.assertEqual(response.data.decode("UTF-8"),
|
||||||
f"Hello, {_USERNAME}! #1")
|
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.status_code, 200)
|
||||||
self.assertEqual(response.data.decode("UTF-8"),
|
self.assertEqual(response.data.decode("UTF-8"),
|
||||||
f"Hello, {_USERNAME}! #2")
|
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