Fix login and unified signin templates to send CSRF tokens. (#825)

Add additional tests.

The oauthstart view needs to be decorated with @unauth_csrf.

Turn on complete CSRF support in view_scaffold since we really don't test CSRF enough.

CHANGES.rst

@@ -25,8 +25,9 @@ Fixes
 - (:issue:`281`) Reset password can be exploited and other OWASP improvements.
 - (:pr:`817`) Confirmation can be exploited and other OWASP improvements.
 - (:pr:`819`) Convert to pyproject.toml, build, remove setup.
-- (:pr:`xxx`) the tf_validity feature now ONLY sets a cookie - and the token is no longer
+- (:pr:`823`) the tf_validity feature now ONLY sets a cookie - and the token is no longer
   returned as part of a JSON response.
+- (:pr:`xxx`) Fix login/unified signin templates to properly send CSRF token. Add more tests.
 
 Backwards Compatibility Concerns
 +++++++++++++++++++++++++++++++++

docs/patterns.rst

@@ -251,7 +251,7 @@ Note that we use the header name ``X-CSRF-Token`` as that is one of the default
 headers configured in Flask-WTF (*WTF_CSRF_HEADERS*)
 
 To protect your application's endpoints (that presumably are not using Flask forms),
-you need to enable CSRF as described in the FlaskWTF `documentation <https://flask-wtf.readthedocs.io/en/1.0.x/csrf/>`_: ::
+you need to enable CSRF as described in the FlaskWTF `documentation <https://flask-wtf.readthedocs.io/en/1.1.x/csrf/>`_: ::
 
     flask_wtf.CSRFProtect(app)
 

flask_security/oauth_glue.py

@@ -4,7 +4,7 @@
 
     Class and methods to glue our login path with authlib for to support 'social' auth.
 
-    :copyright: (c) 2022-2022 by J. Christopher Wagner (jwag).
+    :copyright: (c) 2022-2023 by J. Christopher Wagner (jwag).
     :license: MIT, see LICENSE for more details.
 
 """
@@ -23,6 +23,7 @@
 
 from flask import abort, after_this_request, redirect, request
 
+from .decorators import unauth_csrf
 from .proxies import _security
 from .utils import (
     config_value as cv,
@@ -73,6 +74,7 @@ def google_fetch_identity(
     return "email", profile["email"]
 
 
+@unauth_csrf()
 def oauthstart(name: str) -> "ResponseValue":
     """View to start an oauth authentication.
     Name is a pre-registered oauth provider.

flask_security/templates/security/login_user.html

@@ -21,7 +21,7 @@ <h1>{{ _fsdomain('Login') }}</h1>
     <hr class="fs-gap">
     <h2>{{ _fsdomain("Use WebAuthn to Sign In") }}</h2>
     <div>
-      <form method="get" id="wan-signin-form" name="wan_signin_form">
+      <form method="get" id="wan_signin_form" name="wan_signin_form">
         <input id="wan_signin" name="wan_signin" type="submit" value="{{ _fsdomain('Sign in with WebAuthn') }}" formaction="{{ url_for_security('wan_signin') }}{{ prop_next() }}">
       </form>
     </div>
@@ -31,8 +31,11 @@ <h2>{{ _fsdomain("Use WebAuthn to Sign In") }}</h2>
     <h2>{{ _fsdomain("Use Social Oauth to Sign In") }}</h2>
     {% for provider in security.oauthglue.provider_names %}
       <div class="fs-gap">
-        <form method="post" id="{{ provider }}"-form name="{{ provider }}"_form>
+        <form method="post" id="{{ provider }}_form" name="{{ provider }}_form">
           <input id="{{ provider }}" name="{{ provider }}" type="submit" value="{{ _fsdomain('Sign in with ')~provider }}" formaction="{{ url_for_security('oauthstart', name=provider) }}{{ prop_next() }}">
+          {% if csrf_token is defined %}
+            <input id="{{ provider }}_csrf_token" name="{{ provider }}_csrf_token" type="hidden" value="{{ csrf_token() }}">
+          {% endif %}
         </form>
       </div>
     {% endfor %}

flask_security/templates/security/us_signin.html

@@ -34,8 +34,11 @@ <h2>{{ _fsdomain("Use WebAuthn to Sign In") }}</h2>
     <h2>{{ _fsdomain("Use Social Oauth to Sign In") }}</h2>
     {% for provider in security.oauthglue.provider_names %}
       <div class="fs-gap">
-        <form method="post" id="{{ provider }}"-form name="{{ provider }}_form">
+        <form method="post" id="{{ provider }}_form" name="{{ provider }}_form">
           <input id="{{ provider }}" name="{{ provider }}" type="submit" value="{{ _fsdomain('Sign in with ')~provider }}" formaction="{{ url_for_security('oauthstart', name=provider) }}{{ prop_next() }}">
+          {% if csrf_token is defined %}
+            <input id="{{ provider }}_csrf_token" name="{{ provider }}_csrf_token" type="hidden" value="{{ csrf_token() }}">
+          {% endif %}
         </form>
       </div>
     {% endfor %}

tests/test_oauthglue.py

@@ -4,7 +4,7 @@
 
     Oauth glue tests - oauthglue is a very thin shim between FS and authlib
 
-    :copyright: (c) 2022-2022 by J. Christopher Wagner (jwag).
+    :copyright: (c) 2022-2023 by J. Christopher Wagner (jwag).
     :license: MIT, see LICENSE for more details.
 """
 
@@ -13,10 +13,13 @@
 from urllib.parse import parse_qsl, urlsplit
 
 from flask import redirect
+from flask_wtf import CSRFProtect
 
 from tests.test_utils import (
     authenticate,
+    get_csrf_token,
     get_form_action,
+    get_form_input,
     init_app_with_options,
     logout,
     setup_tf_sms,
@@ -69,15 +72,24 @@ def register(self, name, **kwargs):
 
 
 @pytest.mark.settings(oauth_enable=True, post_login_view="/post_login")
+@pytest.mark.app_settings(wtf_csrf_enabled=True)
 def test_github(app, sqlalchemy_datastore, get_message):
+    CSRFProtect(app)
     init_app_with_options(
         app, sqlalchemy_datastore, **{"security_args": {"oauth": MockOAuth()}}
     )
     client = app.test_client()
     response = client.get("/login")
     github_url = get_form_action(response, 1)
+    csrf_token = get_form_input(response, field_id="github_csrf_token")
 
+    # make sure required CSRF
     response = client.post(github_url, follow_redirects=False)
+    assert "The CSRF token is missing"
+
+    response = client.post(
+        github_url, data=dict(csrf_token=csrf_token), follow_redirects=False
+    )
     assert "/whatever" in response.location
 
     response = client.get("/login/oauthresponse/github", follow_redirects=False)
@@ -88,6 +100,23 @@ def test_github(app, sqlalchemy_datastore, get_message):
     assert response.status_code == 200
 
 
+@pytest.mark.settings(
+    oauth_enable=True, post_login_view="/post_login", csrf_ignore_unauth_endpoints=True
+)
+@pytest.mark.app_settings(wtf_csrf_enabled=True, wtf_csrf_check_default=False)
+def test_github_nocsrf(app, sqlalchemy_datastore, get_message):
+    # Test if ignore_unauth_endpoints is true - doesn't require CSRF
+    CSRFProtect(app)
+    init_app_with_options(
+        app, sqlalchemy_datastore, **{"security_args": {"oauth": MockOAuth()}}
+    )
+    client = app.test_client()
+    response = client.get("/login")
+    github_url = get_form_action(response, 1)
+    response = client.post(github_url, follow_redirects=False)
+    assert "/whatever" in response.location
+
+
 @pytest.mark.settings(oauth_enable=True, post_login_view="/post_login")
 def test_outside_register(app, sqlalchemy_datastore, get_message):
     def myoauth_fetch_identity(oauth, token):
@@ -188,14 +217,20 @@ def test_tf(app, sqlalchemy_datastore, get_message):
     redirect_behavior="spa",
     login_error_view="/login-error",
     post_login_view="/post-login",
+    csrf_ignore_unauth_endpoints=False,
 )
+@pytest.mark.app_settings(wtf_csrf_enabled=True)
 def test_spa(app, sqlalchemy_datastore, get_message):
+    CSRFProtect(app)
     headers = {"Accept": "application/json", "Content-Type": "application/json"}
 
     init_app_with_options(
         app, sqlalchemy_datastore, **{"security_args": {"oauth": MockOAuth()}}
     )
     client = app.test_client()
+    csrf_token = get_csrf_token(client)
+    headers["X-CSRF-Token"] = csrf_token
+
     response = client.post("/login/oauthstart/github", headers=headers)
     assert "/whatever" in response.location
     redirect_url = urllib.parse.urlsplit(urllib.parse.unquote(response.location))

tests/test_utils.py

@@ -1,10 +1,10 @@
 """
-    utils
-    ~~~~~
+    test_utils
+    ~~~~~~~~~~
 
     Test utils
 
-    :copyright: (c) 2019-2022 by J. Christopher Wagner (jwag).
+    :copyright: (c) 2019-2023 by J. Christopher Wagner (jwag).
     :license: MIT, see LICENSE for more details.
 """
 from contextlib import contextmanager
@@ -42,11 +42,11 @@ def authenticate(
 
 
 def json_authenticate(client, email="[email protected]", password="password", endpoint=None):
-    data = f'{{"email": "{email}", "password": "{password}"}}'
+    data = dict(email=email, password=password)
 
     # Get auth token always
     ep = endpoint or "/login?include_auth_token"
-    return client.post(ep, content_type="application/json", data=data)
+    return client.post(ep, content_type="application/json", json=data)
 
 
 def is_authenticated(client, get_message):
@@ -165,6 +165,19 @@ def get_form_action(response, ordinal=0):
     return matcher[ordinal]
 
 
+def get_form_input(response, field_id):
+    # return value of field with the id == field_id or None if not found
+    rex = f'<input id="{field_id}"[^>]*value="([^"]*)">'
+    matcher = re.findall(
+        rex,
+        response.data.decode("utf-8"),
+        re.IGNORECASE | re.DOTALL,
+    )
+    if matcher:
+        return matcher[0]
+    return None
+
+
 def check_xlation(app, locale):
     """Return True if locale is loaded"""
     with app.test_request_context():
@@ -251,12 +264,7 @@ def get_num_queries(datastore):
     return None if datastore doesn't support this.
     """
     if is_sqlalchemy(datastore):
-        try:
-            # Flask-SQLAlachemy >= 3.0.0
-            from flask_sqlalchemy.record_queries import get_recorded_queries
-        except ImportError:
-            # Flask-SQLAlchemy < 3.0.0
-            from flask_sqlalchemy import get_debug_queries as get_recorded_queries
+        from flask_sqlalchemy.record_queries import get_recorded_queries
 
         return len(get_recorded_queries())
     return None

tests/view_scaffold.py

@@ -1,4 +1,4 @@
-# :copyright: (c) 2019-2022 by J. Christopher Wagner (jwag).
+# :copyright: (c) 2019-2023 by J. Christopher Wagner (jwag).
 # :license: MIT, see LICENSE for more details.
 
 """
@@ -27,6 +27,7 @@
 
 from flask import Flask, flash, render_template_string, request, session
 from flask_sqlalchemy import SQLAlchemy
+from flask_wtf import CSRFProtect
 
 from flask_security import (
     MailUtil,
@@ -152,6 +153,7 @@ def origin(self) -> str:
         ):
             app.config[ev] = _find_bool(os.environ.get(ev))
 
+    CSRFProtect(app)
     # Create database models and hook up.
     app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
     db = SQLAlchemy(app)