Enable CSRF protection globally by default

h/accounts/schemas.py

@@ -16,7 +16,6 @@
     USERNAME_PATTERN,
 )
 from h.schemas import validators
-from h.schemas.base import CSRFSchema
 from h.schemas.forms.accounts.util import PASSWORD_MIN_LENGTH
 from h.util.user import format_userid
 
@@ -153,7 +152,7 @@ def _privacy_accepted_message():
     return privacy_msg
 
 
-class RegisterSchema(CSRFSchema):
+class RegisterSchema(colander.Schema):
     username = colander.SchemaNode(
         colander.String(),
         validator=colander.All(
@@ -197,13 +196,12 @@ class RegisterSchema(CSRFSchema):
     )
 
 
-class EmailChangeSchema(CSRFSchema):
+class EmailChangeSchema(colander.Schema):
     email = email_node(title=_("Email address"))
     # No validators: all validation is done on the email field
     password = password_node(title=_("Confirm password"), hide_until_form_active=True)
 
     def validator(self, node, value):
-        super().validator(node, value)
         exc = colander.Invalid(node)
         request = node.bindings["request"]
         svc = request.find_service(name="user_password")
@@ -216,7 +214,7 @@ def validator(self, node, value):
             raise exc
 
 
-class PasswordChangeSchema(CSRFSchema):
+class PasswordChangeSchema(colander.Schema):
     password = password_node(title=_("Current password"), inactive_label=_("Password"))
     new_password = new_password_node(
         title=_("New password"), hide_until_form_active=True
@@ -231,7 +229,6 @@ class PasswordChangeSchema(CSRFSchema):
     )
 
     def validator(self, node, value):  # pragma: no cover
-        super().validator(node, value)
         exc = colander.Invalid(node)
         request = node.bindings["request"]
         svc = request.find_service(name="user_password")
@@ -247,12 +244,10 @@ def validator(self, node, value):  # pragma: no cover
             raise exc
 
 
-class DeleteAccountSchema(CSRFSchema):
+class DeleteAccountSchema(colander.Schema):
     password = password_node(title=_("Confirm password"))
 
     def validator(self, node, value):
-        super().validator(node, value)
-
         request = node.bindings["request"]
         svc = request.find_service(name="user_password")
 
@@ -262,7 +257,7 @@ def validator(self, node, value):
             raise exc
 
 
-class NotificationsSchema(CSRFSchema):
+class NotificationsSchema(colander.Schema):
     types = (("reply", _("Email me when someone replies to one of my annotations.")),)
 
     notifications = colander.SchemaNode(

h/app.py

@@ -30,6 +30,8 @@ def create_app(_global_config, **settings):  # pragma: no cover
 
 
 def includeme(config):  # pragma: no cover
+    config.set_default_csrf_options(require_csrf=True)
+
     config.scan("h.subscribers")
 
     config.add_tween("h.tweens.conditional_http_tween_factory", under=EXCVIEW)

h/form.py

@@ -5,10 +5,13 @@
 form templates in preference to the defaults.
 """
 
+from functools import partial
+
 import deform
 import pyramid_jinja2
 from markupsafe import Markup
 from pyramid import httpexceptions
+from pyramid.csrf import get_csrf_token
 from pyramid.path import AssetResolver
 
 from h import i18n
@@ -46,6 +49,10 @@ def __call__(self, template_name, **kwargs):
         context = self._system.copy()
         context.update(kwargs)
 
+        context.setdefault(
+            "get_csrf_token", partial(get_csrf_token, context["request"])
+        )
+
         return Markup(template.render(context))
 
 
@@ -75,7 +82,7 @@ def create_form(request, *args, **kwargs):
     default) will use the renderer configured in the :py:mod:`h.form` module.
     """
     env = request.registry[ENVIRONMENT_KEY]
-    renderer = Jinja2Renderer(env, {"feature": request.feature})
+    renderer = Jinja2Renderer(env, {"feature": request.feature, "request": request})
     kwargs.setdefault("renderer", renderer)
 
     return deform.Form(*args, **kwargs)

h/schemas/auth_client.py

@@ -3,13 +3,13 @@
 
 from h import i18n
 from h.models.auth_client import GrantType
-from h.schemas.base import CSRFSchema, enum_type
+from h.schemas.base import enum_type
 
 _ = i18n.TranslationString
 GrantTypeSchemaType = enum_type(GrantType)
 
 
-class CreateAuthClientSchema(CSRFSchema):
+class CreateAuthClientSchema(colander.Schema):
     name = colander.SchemaNode(
         colander.String(),
         title=_("Name"),
@@ -57,8 +57,6 @@ class CreateAuthClientSchema(CSRFSchema):
     )
 
     def validator(self, node, value):
-        super().validator(node, value)
-
         grant_type = value.get("grant_type")
         redirect_url = value.get("redirect_url")
 

h/schemas/base.py

@@ -3,42 +3,14 @@
 import copy
 
 import colander
-import deform
 import jsonschema
 from pyramid import httpexceptions
-from pyramid.csrf import check_csrf_token, get_csrf_token
-
-
-@colander.deferred
-def deferred_csrf_token(_node, kwargs):
-    request = kwargs.get("request")
-    return get_csrf_token(request)
 
 
 class ValidationError(httpexceptions.HTTPBadRequest):
     pass
 
 
-class CSRFSchema(colander.Schema):
-    """
-    A CSRFSchema backward-compatible with the one from the hem module.
-
-    Unlike hem, this doesn't require that the csrf_token appear in the
-    serialized appstruct.
-    """
-
-    csrf_token = colander.SchemaNode(
-        colander.String(),
-        widget=deform.widget.HiddenWidget(),
-        default=deferred_csrf_token,
-        missing=None,
-    )
-
-    def validator(self, node, _value):
-        request = node.bindings["request"]
-        check_csrf_token(request)
-
-
 class JSONSchema:
     """
     Validate data according to a JSON Schema.

h/schemas/forms/accounts/edit_profile.py

@@ -5,7 +5,6 @@
 from h.accounts import util
 from h.models.user import DISPLAY_NAME_MAX_LENGTH
 from h.schemas import validators
-from h.schemas.base import CSRFSchema
 
 _ = i18n.TranslationString
 
@@ -24,7 +23,7 @@ def validate_orcid(node, cstruct):
         raise colander.Invalid(node, str(exc))  # noqa: B904
 
 
-class EditProfileSchema(CSRFSchema):
+class EditProfileSchema(colander.Schema):
     display_name = colander.SchemaNode(
         colander.String(),
         missing=None,

h/schemas/forms/accounts/forgot_password.py

@@ -3,12 +3,11 @@
 
 from h import i18n, models
 from h.schemas import validators
-from h.schemas.base import CSRFSchema
 
 _ = i18n.TranslationString
 
 
-class ForgotPasswordSchema(CSRFSchema):
+class ForgotPasswordSchema(colander.Schema):
     email = colander.SchemaNode(
         colander.String(),
         validator=colander.All(validators.Email()),
@@ -17,8 +16,6 @@ class ForgotPasswordSchema(CSRFSchema):
     )
 
     def validator(self, node, value):
-        super().validator(node, value)
-
         request = node.bindings["request"]
         email = value.get("email")
         user = models.User.get_by_email(request.db, email, request.default_authority)

h/schemas/forms/accounts/login.py

@@ -2,7 +2,6 @@
 import deform
 
 from h import i18n
-from h.schemas.base import CSRFSchema
 from h.services.user import UserNotActivated
 
 _ = i18n.TranslationString
@@ -25,7 +24,7 @@ def _deferred_password_widget(_node, kwargs):
     )
 
 
-class LoginSchema(CSRFSchema):
+class LoginSchema(colander.Schema):
     username = colander.SchemaNode(
         colander.String(),
         title=_("Username / email"),
@@ -36,8 +35,6 @@ class LoginSchema(CSRFSchema):
     )
 
     def validator(self, node, value):
-        super().validator(node, value)
-
         request = node.bindings["request"]
         username = value.get("username")
         password = value.get("password")

h/schemas/forms/accounts/reset_password.py

@@ -4,7 +4,6 @@
 from itsdangerous import BadData, SignatureExpired
 
 from h import i18n, models
-from h.schemas.base import CSRFSchema
 from h.schemas.forms.accounts import util
 
 _ = i18n.TranslationString
@@ -60,7 +59,7 @@ def deserialize(self, node, cstruct):
         return user
 
 
-class ResetPasswordSchema(CSRFSchema):
+class ResetPasswordSchema(colander.Schema):
     # N.B. this is the field into which the user puts their reset code, but we
     # call it `user` because when validated, it will return a `User` object.
     user = colander.SchemaNode(

h/schemas/forms/admin/group.py

@@ -14,7 +14,6 @@
     GROUP_NAME_MIN_LENGTH,
 )
 from h.schemas import validators
-from h.schemas.base import CSRFSchema
 from h.util import group_scope
 
 _ = i18n.TranslationString
@@ -137,7 +136,7 @@ def group_organization_select_widget(_node, kwargs):
     return SelectWidget(values=list(zip(org_pubids, org_labels, strict=False)))
 
 
-class AdminGroupSchema(CSRFSchema):
+class AdminGroupSchema(colander.Schema):
     def __init__(self, *args):
         super().__init__(*args)
 
@@ -219,5 +218,4 @@ def __init__(self, *args):
     )
 
     def validator(self, node, value):
-        super().validator(node, value)
         username_validator(node, value)

h/schemas/forms/admin/organization.py

@@ -6,7 +6,6 @@
 import h.i18n
 from h.models.organization import Organization
 from h.schemas import validators
-from h.schemas.base import CSRFSchema
 
 _ = h.i18n.TranslationString
 
@@ -36,7 +35,7 @@ def validate_logo(node, value):
         raise colander.Invalid(node, _("Logo does not start with <svg> tag"))
 
 
-class OrganizationSchema(CSRFSchema):
+class OrganizationSchema(colander.Schema):
     authority = colander.SchemaNode(colander.String(), title=_("Authority"))
 
     name = colander.SchemaNode(

h/templates/deform/form.jinja2

@@ -8,6 +8,7 @@
       class="form {{ field.css_class or '' }}
              {%- if field.use_inline_editing %} js-form {% endif %}">
   <input type="hidden" name="__formid__" value="{{ field.formid }}" />
+  <input type="hidden" name="csrf_token" value="{{ get_csrf_token() }}">
 
   <div class="form__backdrop" data-ref="formBackdrop"></div>
 

h/views/api/auth.py

@@ -104,6 +104,7 @@ def post(self):
         request_param="response_mode=web_message",
         is_authenticated=True,
         renderer="h:templates/oauth/authorize_web_message.html.jinja2",
+        require_csrf=False,
     )
     def post_web_message(self):
         """

h/views/api/config.py

@@ -65,6 +65,7 @@ def add_api_view(  # noqa: PLR0913
                                   `route_name` must be specified.
     :param dict **settings: Arguments to pass on to ``config.add_view``
     """
+    settings.setdefault("require_csrf", False)
     settings.setdefault("renderer", "json")
     settings.setdefault("decorator", (cors_policy, version_media_type_header(subtype)))
 

tests/functional/h/views/admin/permissions_test.py

@@ -35,10 +35,7 @@ def test_accessible_by_staff(self, app, url, accessible):
 
         assert res.status_code == 200 if accessible else 404
 
-    GROUP_PAGES = (
-        ("POST", "/admin/groups/delete/{pubid}", 302),
-        ("GET", "/admin/groups/{pubid}", 200),
-    )
+    GROUP_PAGES = (("GET", "/admin/groups/{pubid}", 200),)
 
     @pytest.mark.usefixtures("with_logged_in_admin")
     @pytest.mark.parametrize("method,url_template,success_code", GROUP_PAGES)

tests/unit/h/accounts/schemas_test.py

@@ -3,7 +3,6 @@
 
 import colander
 import pytest
-from pyramid.exceptions import BadCSRFToken
 
 from h.accounts import schemas
 from h.services.user_password import UserPasswordService
@@ -156,9 +155,7 @@ def test_it_validates_with_valid_payload(
 
         result = schema.deserialize(valid_params)
 
-        assert result == dict(
-            valid_params, privacy_accepted=True, comms_opt_in=None, csrf_token=None
-        )
+        assert result == dict(valid_params, privacy_accepted=True, comms_opt_in=None)
 
     @pytest.fixture
     def valid_params(self):
@@ -194,18 +191,6 @@ def test_it_is_valid_if_email_same_as_users_existing_email(
 
         schema.deserialize({"email": user.email, "password": "flibble"})
 
-    def test_it_is_invalid_if_csrf_token_missing(self, pyramid_request, schema):
-        del pyramid_request.headers["X-CSRF-Token"]
-
-        with pytest.raises(BadCSRFToken):
-            schema.deserialize({"email": "[email protected]", "password": "flibble"})
-
-    def test_it_is_invalid_if_csrf_token_wrong(self, pyramid_request, schema):
-        pyramid_request.headers["X-CSRF-Token"] = "WRONG"
-
-        with pytest.raises(BadCSRFToken):
-            schema.deserialize({"email": "[email protected]", "password": "flibble"})
-
     def test_it_is_invalid_if_password_wrong(self, schema, user_password_service):
         user_password_service.check_password.return_value = False