Yda 5546 harden static loader

setup.cfg

@@ -1,5 +1,5 @@
 [flake8]
-ignore=E501
+ignore=E221,E241,E402,W503,W605,F403,F405
 import-order-style=smarkets
 strictness=short
 docstring_style=sphinx

yoda_eus/app.py

@@ -7,7 +7,7 @@
 import urllib.parse
 from datetime import datetime
 from os import path
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 import bcrypt
 from flask import abort, Flask, jsonify, make_response, render_template, request, Response, send_from_directory
@@ -17,6 +17,7 @@
 from jinja2 import ChoiceLoader, FileSystemLoader
 from yoda_eus.mail import is_email_valid, send_email_template_if_needed
 from yoda_eus.password_complexity import check_password_complexity
+from yoda_eus.util import get_validated_static_path
 
 
 db = SQLAlchemy()
@@ -350,7 +351,10 @@ def process_forgot_password() -> Response:
             return response
 
         if (not is_email_valid(username) and app.config.get("MAIL_ONLY_TO_VALID_ADDRESS").lower() == "true"):
-            errors = {"errors": ["Unable to send password reset email, because your user name ('{}') is not a valid email address.".format(username)]}
+            errors = {
+                "errors": ["Unable to send password reset email, "
+                           "because your user name ('{}') is not a valid email address.".format(username)]
+            }
             response = make_response(render_template('forgot-password.html', **errors))
             response.status_code = 404
             return response
@@ -557,10 +561,14 @@ def refuse_api_requests() -> Response:
             The EUS presents two web interfaces (vhosts) on two different TCP ports. One of these
             does not have the API available, so that API access can be restricted on a TCP level (e.g. in firewalls).
 
-            If this instance does not have the API enabled, access to API functions is intercepted and blocked by this function.
+            If this instance does not have the API enabled, access to API functions is intercepted
+            and blocked by this function.
             """
             if request.path.startswith("/api/"):
-                abort(make_response(jsonify({'status': 'error', 'message': 'The EUS API has been disabled on this interface.'}), 403))
+                abort(make_response(jsonify({
+                    'status': 'error',
+                    'message': 'The EUS API has been disabled on this interface.'
+                }), 403))
 
     @app.before_request
     def check_api_secret() -> Response:
@@ -575,10 +583,13 @@ def check_api_secret() -> Response:
         elif secret_header in request.headers and request.headers[secret_header] == app.config.get("API_SECRET"):
             return
         else:
-            abort(make_response(jsonify({'status': 'error', 'message': 'EUS secret header not present or does not match.'}), 403))
+            abort(make_response(jsonify({
+                'status': 'error',
+                'message': 'EUS secret header not present or does not match.'
+            }), 403))
 
     @app.before_request
-    def static_loader() -> Response:
+    def static_loader() -> Optional[Response]:
         """
         Static files handling - recognisable through '/assets/'
         Override requested static file if present in user_static_area
@@ -589,17 +600,18 @@ def static_loader() -> Response:
 
         :returns: Static file
         """
-        if request.full_path.split('/')[1] == 'assets':
-            user_static_area = path.join(app.config.get('YODA_THEME_PATH'), app.config.get('YODA_THEME'))
-            asset_dir, asset_name = path.split(request.path)
-            static_dir = asset_dir.replace('/assets', user_static_area + '/static')
-            user_static_filename = path.join(static_dir, asset_name)
-
-            if not path.exists(user_static_filename):
-                static_dir = asset_dir.replace('/assets', '/var/www/yoda/static')
-
+        result = get_validated_static_path(
+            request.full_path,
+            request.path,
+            app.config.get('YODA_THEME_PATH'),
+            app.config.get('YODA_THEME')
+        )
+        if result is not None:
+            static_dir, asset_name = result
             return send_from_directory(static_dir, asset_name)
 
+        return None
+
     @ app.url_defaults
     def add_cache_buster(endpoint: str, values: Dict[str, str]) -> None:
         """Add cache buster to asset (static) URLs."""

yoda_eus/tests/test_unit.py

@@ -1,14 +1,15 @@
-__copyright__ = 'Copyright (c) 2023, Utrecht University'
-__license__ = 'GPLv3, see LICENSE'
+__copyright__ = "Copyright (c) 2023, Utrecht University"
+__license__   = "GPLv3, see LICENSE"
 
 import string
+from unittest.mock import patch
 
 from yoda_eus.mail import is_email_valid
 from yoda_eus.password_complexity import check_password_complexity
+from yoda_eus.util import get_validated_static_path
 
 
 class TestMain:
-
     def test_password_validation_ok(self):
         result = check_password_complexity("Test123456789!")
         assert len(result) == 0
@@ -50,10 +51,92 @@ def test_password_validation_multiple(self):
         assert len(result) == 3
         assert "Password is too short: it needs to be at least 10 characters." in result
         assert "Password needs to contain at least one digit." in result
-        assert "Password needs to contain at least one punctuation character ({})".format(string.punctuation) in result
+        assert (
+            "Password needs to contain at least one punctuation character ({})".format(
+                string.punctuation
+            )
+            in result
+        )
 
     def is_email_valid_yes(self):
         assert is_email_valid("[email protected]")
 
     def is_email_valid_no(self):
         assert not is_email_valid("this is not a valid email address")
+
+    def exists_return_value(self, pathname):
+        """ Mock path.exists function. True if path does not contain "theme" and "uu" """
+        return not ("theme" in pathname and "uu" in pathname)
+
+    @patch("os.path.exists")
+    def test_static_loader_valid_path(self, mock_exists):
+        mock_exists.side_effect = self.exists_return_value
+        # uu theme
+        static_dir, asset_name = get_validated_static_path(
+            "/assets/img/logo.svg?wekr",
+            "/assets/img/logo.svg",
+            "/var/www/yoda/themes",
+            "uu",
+        )
+        assert static_dir == "/var/www/yoda/static/img"
+        assert asset_name == "logo.svg"
+        # other theme
+        static_dir, asset_name = get_validated_static_path(
+            "/assets/img/logo.svg?wekr",
+            "/assets/img/logo.svg",
+            "/var/www/yoda/themes",
+            "wur",
+        )
+        assert static_dir == "/var/www/yoda/themes/wur/static/img"
+        assert asset_name == "logo.svg"
+
+    @patch("os.path.exists")
+    def test_static_loader_invalid_path(self, mock_exists):
+        mock_exists.side_effect = self.exists_return_value
+        # Too short
+        assert (
+            get_validated_static_path("/?sawerw", "/", "/var/www/yoda/themes", "uu")
+            is None
+        )
+        # Path traversal attack
+        assert (
+            get_validated_static_path(
+                "/assets/../../../../etc/passwd?werwrwr",
+                "/assets/../../../../etc/passwd",
+                "/var/www/yoda/themes",
+                "uu",
+            )
+            is None
+        )
+        # non-printable characters
+        full_path = "/assets/" + chr(13) + "img/logo.svg?werwer"
+        path = "/assets/" + chr(13) + "img/logo.svg"
+        assert (
+            get_validated_static_path(full_path, path, "/var/www/yoda/themes", "uu")
+            is None
+        )
+        assert (
+            get_validated_static_path(full_path, path, "/var/www/yoda/themes", "wur")
+            is None
+        )
+        # non-printable characters in asset name
+        full_path = "/assets/img/l" + chr(13) + "ogo.svg?werwer"
+        path = "/assets/img/l" + chr(13) + "ogo.svg"
+        assert (
+            get_validated_static_path(full_path, path, "/var/www/yoda/themes", "uu")
+            is None
+        )
+        assert (
+            get_validated_static_path(full_path, path, "/var/www/yoda/themes", "wur")
+            is None
+        )
+        # .. in file name
+        assert (
+            get_validated_static_path(
+                "/assets/img/lo..go.svg?sklaerw",
+                "/assets/img/lo..go.svg?sklaerw",
+                "/var/www/yoda/themes",
+                "uu",
+            )
+            is None
+        )

yoda_eus/util.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+
+__copyright__ = "Copyright (c) 2021-2023, Utrecht University"
+__license__ = "GPLv3, see LICENSE"
+
+from os import path
+from re import fullmatch
+from typing import Optional, Tuple
+
+from werkzeug.security import safe_join
+from werkzeug.utils import secure_filename
+
+
+def get_validated_static_path(
+    full_path: str, request_path: str, yoda_theme_path: str, yoda_theme: str
+) -> Optional[Tuple[str, str]]:
+    """
+    Static files handling - recognisable through '/assets/'
+    Confirms that input path is valid and return corresponding static path
+
+    :param full_path:       Full path of request
+    :param request_path:    Short path of request
+    :param yoda_theme_path: Path to the yoda themes
+    :param yoda_theme:      Name of the chosen theme
+
+    :returns: Tuple of static directory and filename for correct path, None for incorrect path
+    """
+    parts = full_path.split("/")
+
+    if (
+        len(parts) > 2
+        and fullmatch("[ -~]*", full_path) is not None
+        and parts[1] == "assets"
+    ):
+        parts = parts[2:-1]
+        user_static_area = path.join(yoda_theme_path, yoda_theme)
+        _, asset_name = path.split(request_path)
+        # Confirm that asset_name is safe
+        if asset_name != secure_filename(asset_name):
+            return None
+
+        static_dir = safe_join(user_static_area + "/static", *parts)
+        if not static_dir:
+            return None
+        user_static_filename = path.join(static_dir, asset_name)
+
+        if not path.exists(user_static_filename):
+            static_dir = safe_join("/var/www/yoda/static", *parts)
+
+        return static_dir, asset_name
+
+    return None