Fix session-only authentication. (pallets-eco#813)

* Update test_common.py

Added testcase for failing toke-authentication on session-only endpoint

* Update conftest.py

Added session-only authenticated route to test-fixture

* Update decorators.py

Added the `_check_session` function to specifically check session data to be used as authentication_method in the `auth_required`

* Update decorators.py

* Update decorators.py

* fixed decorator and added tests

* Fix session-only authentication.

If an endpoint was decorated with "session" only - a properly submitted token would also be accepted.
Fix that by checking as part of the auth_required() decorator and the user is authenticated AND was authenticated using the _user_loader (which is what flask-login calls for session based authenticated).

close pallets-eco#791

---------

Co-authored-by: N247S <[email protected]>

CHANGES.rst

@@ -16,6 +16,8 @@ Fixes
   Improve MongoDB quickstart.
 - (:issue:`801`) Fix Quickstart for SQLAlchemy with scoped session.
 - (:issue:`806`) Login no longer, by default, check for email deliverability.
+- (:issue:`791`) Token authentication is accepted on endpoints which only allow
+  'session' as authentication-method. (N247S)
 
 Backwards Compatibility Concerns
 +++++++++++++++++++++++++++++++++

flask_security/decorators.py

@@ -32,6 +32,7 @@
     check_and_update_authn_fresh,
     json_error_response,
     set_request_attr,
+    get_request_attr,
     url_for_security,
 )
 
@@ -150,6 +151,25 @@ def _check_token():
     return False
 
 
+def _check_session():
+    """
+    Note that flask_login will have already run _load_user (due to someone referencing
+    current_user proxy). This will have called our _user_loader or _request_loader
+    already.
+    This method needs to determine whether the authenticated user was authenticated
+    due to _user_loader or _request_loader and return True for the former.
+    This routine makes sure that if an endpoint is decorated with just allowing
+    'session' that token authentication won't return True (even though the user
+    is in fact correctly authenticated). There are certainly endpoints that need to
+    be web browser/session only (often those that in fact return tokens!).
+    """
+    if not current_user.is_authenticated:
+        return False
+    if get_request_attr("fs_authn_via") != "session":
+        return False
+    return True
+
+
 def _check_http_auth():
     auth = request.authorization or BasicAuth(username=None, password=None)
     if not auth.username:
@@ -332,7 +352,7 @@ def dashboard():
 
     login_mechanisms = {
         "token": lambda: _check_token(),
-        "session": lambda: current_user.is_authenticated,
+        "session": lambda: _check_session(),
         "basic": lambda: _check_http_auth(),
     }
     mechanisms_order = ["token", "session", "basic"]

mypy.ini

@@ -19,3 +19,7 @@ warn_unreachable = True
 # This false-positives with non-annotated methods (like imported packages)
 warn_return_any = False
 warn_unused_configs = True
+
+[mypy-flask_security.cli]
+# Due to click 8.1.4
+ignore_errors = True

tests/conftest.py

@@ -204,6 +204,11 @@ def http_custom_realm():
         assert get_request_attr("fs_authn_via") == "basic"
         return render_template("index.html", content="HTTP Authentication")
 
+    @app.route("/session")
+    @auth_required("session")
+    def session():
+        return "Session Authentication"
+
     @app.route("/token", methods=["GET", "POST"])
     @auth_token_required
     def token():

tests/test_common.py

@@ -13,9 +13,11 @@
 import re
 import pytest
 
-from flask import Blueprint
+from flask import Blueprint, g
 
 from flask_security import uia_email_mapper
+from flask_security.decorators import auth_required
+from flask_principal import identity_loaded
 
 from tests.test_utils import (
     authenticate,
@@ -527,6 +529,17 @@ def test_token_auth_via_header_invalid_token(client):
     assert response.status_code == 401
 
 
+def test_token_auth_invalid_for_session_auth(client):
+    # when user is loaded from token data, session authentication should fail.
+    response = json_authenticate(client)
+    token = response.json["response"]["user"]["authentication_token"]
+    # logout so session doesn't contain valid user details
+    logout(client)
+    headers = {"Authentication-Token": token, "Accept": "application/json"}
+    response = client.get("/session", headers=headers)
+    assert response.status_code == 401
+
+
 def test_http_auth(client):
     # browsers expect 401 response with WWW-Authenticate header - which will prompt
     # them to pop up a login form.
@@ -741,6 +754,28 @@ def test_user_deleted_during_session_reverts_to_anonymous_user(app, client):
     assert b"Hello [email protected]" not in response.data
 
 
+def test_session_loads_identity(app, client):
+    @app.route("/identity_check")
+    @auth_required("session")
+    def id_check():
+        if hasattr(g, "identity"):
+            identity = g.identity
+            assert hasattr(identity, "loader_called")
+            assert identity.loader_called
+        return "Success"
+
+    json_authenticate(client)
+
+    # add identity loader after authentication to only fire it for
+    # session-authentication next `get` call
+    @identity_loaded.connect_via(app)
+    def identity_loaded_check(sender, identity):
+        identity.loader_called = True
+
+    response = client.get("/identity_check")
+    assert b"Success" == response.data
+
+
 def test_remember_token(client):
     response = authenticate(client, follow_redirects=False)
     client.delete_cookie("session")

tests/test_webauthn.py

@@ -1212,15 +1212,15 @@ def test_user_handle(app, client, get_message):
             .decode("utf-8")
             .replace("=", "")
         )
-        upd_signin_data = copy.deepcopy(SIGNIN_DATA_UH)
-        upd_signin_data["response"]["userHandle"] = b64_user_handle
-        signin_options, response_url, _ = _signin_start_json(client, "[email protected]")
-        response = client.post(
-            response_url, json=dict(credential=json.dumps(upd_signin_data))
-        )
-        # verify actually logged in
-        response = client.get("/profile", headers={"accept": "application/json"})
-        assert response.status_code == 200
+    upd_signin_data = copy.deepcopy(SIGNIN_DATA_UH)
+    upd_signin_data["response"]["userHandle"] = b64_user_handle
+    signin_options, response_url, _ = _signin_start_json(client, "[email protected]")
+    response = client.post(
+        response_url, json=dict(credential=json.dumps(upd_signin_data))
+    )
+    # verify actually logged in
+    response = client.get("/profile", headers={"accept": "application/json"})
+    assert response.status_code == 200
 
 
 @pytest.mark.settings(webauthn_util_cls=HackWebauthnUtil)