Provide request input sanitization (#358)

* ➕ Provide request input sanitization

* 🔨 Check for json request body

* ➕ Add test  for input sanitization

backend/requirements.txt

@@ -16,6 +16,7 @@ icalendar==5.0.4
 itsdangerous==2.1.2
 markdown==3.6
 MarkupSafe==2.1.2
+nh3==0.2.17
 python-dotenv==1.0.0
 python-jose==3.3.0
 python-multipart==0.0.7

backend/src/appointment/main.py

@@ -10,6 +10,7 @@
 
 from .defines import APP_ENV_DEV, APP_ENV_TEST
 from .middleware.l10n import L10n
+from .middleware.SanitizeMiddleware import SanitizeMiddleware
 # Ignore "Module level import not at top of file"
 # ruff: noqa: E402
 from .secrets import normalize_secrets
@@ -117,6 +118,9 @@ def server():
         )
     )
 
+    # strip html tags from input requests
+    app.add_middleware(SanitizeMiddleware)
+
     app.add_middleware(
         SessionMiddleware,
         secret_key=os.getenv("SESSION_SECRET")

backend/src/appointment/middleware/SanitizeMiddleware.py

@@ -0,0 +1,56 @@
+import json
+import logging
+import nh3
+from starlette.types import ASGIApp, Scope, Receive, Send
+from ..utils import is_json
+
+
+class SanitizeMiddleware:
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+
+    @staticmethod
+    def sanitize_str(value: str) -> str:
+        return nh3.clean(value, tags={""}) if isinstance(value, str) else value
+
+
+    @staticmethod
+    def sanitize_dict(dict_value: str) -> str:
+        return {key: __class__.sanitize_str(value) for key, value in dict_value.items()}
+
+
+    @staticmethod
+    def sanitize_list(list_values: list) -> list:
+        for index, value in enumerate(list_values):
+            if isinstance(value, dict):
+                list_values[index] = {key: __class__.sanitize_str(value) for key, value in value.items()}
+            else:
+                list_values[index] = __class__.sanitize_str(value)
+        return list_values
+
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if "method" not in scope or scope["method"] in ("GET", "HEAD", "OPTIONS"):
+            return await self.app(scope, receive, send)
+
+        async def sanitize_request_body():
+            message = await receive()
+            body = message.get("body")
+            if not body or not isinstance(body, bytes):
+                return message
+            if is_json(body):
+                json_body = json.loads(body)
+                for key, value in json_body.items():
+                    if isinstance(value, dict):
+                        json_body[key] = __class__.sanitize_dict(value)
+                    elif isinstance(value, list):
+                        json_body[key] = __class__.sanitize_list(value)
+                    else:
+                        json_body[key] = __class__.sanitize_str(value)
+                message["body"] = bytes(json.dumps(json_body), encoding="utf-8")
+
+            return message
+
+        return await self.app(scope, sanitize_request_body, send)

backend/src/appointment/utils.py

@@ -1,3 +1,5 @@
+import json
+
 from functools import cache
 
 from argon2 import PasswordHasher
@@ -20,6 +22,16 @@ def list_first(items: list, default=None):
     """Returns the first item of a list or the default value."""
     return next(iter(items), default)
 
+
+def is_json(jsonstring: str):
+    """Return true if given string is valid JSON."""
+    try:
+        json.loads(jsonstring)
+    except ValueError as e:
+        return False
+    return True
+
+
 @cache
 def setup_encryption_engine():
     engine = AesEngine()

backend/test/integration/test_auth.py

@@ -89,4 +89,3 @@ def test_fxa_callback(self, with_db, with_client, monkeypatch):
             fxa = subscriber.get_external_connection(models.ExternalConnectionType.fxa)
             assert fxa
             assert fxa.type_id == FXA_CLIENT_PATCH.get('external_connection_type_id')
-

backend/test/integration/test_schedule.py

@@ -197,6 +197,23 @@ def test_update_existing_schedule(self, with_client, make_schedule):
         assert weekdays == [2, 4, 6]
         assert data["slot_duration"] == 60
 
+    def test_update_existing_schedule_with_html(self, with_client, make_schedule):
+        generated_schedule = make_schedule()
+
+        response = with_client.put(
+            f"/schedule/{generated_schedule.id}",
+            json={
+                "calendar_id": generated_schedule.calendar_id,
+                "name": "<i>Schedule</i>",
+                "details": "Lorem <p>test</p> Ipsum<br><script>run_evil();</script>",
+            },
+            headers=auth_headers,
+        )
+        assert response.status_code == 200, response.text
+        data = response.json()
+        assert data["name"] == "Schedule"
+        assert data["details"] == "Lorem test Ipsum"
+
     def test_update_missing_schedule(self, with_client, make_schedule):
         generated_schedule = make_schedule()