Path protection with pathlib

sanic/__version__.py

@@ -1 +1 @@
-__version__ = "20.12.5"
+__version__ = "20.12.7"

sanic/app.py

@@ -2,6 +2,7 @@ import logging
 import logging.config
 import os
 import re
+import sys
 
 from asyncio import CancelledError, Protocol, ensure_future, get_event_loop
 from collections import defaultdict, deque
@@ -65,6 +66,18 @@ class Sanic:
         if configure_logging:
             logging.config.dictConfig(log_config or LOGGING_CONFIG_DEFAULTS)
 
+        if sys.version_info >= (3, 10):
+            error_logger.error(
+                "Unsupported version of Python has been detected.\n\nPython "
+                f"version {sys.version} is not supported by this version of "
+                "Sanic. There is a security advisory that has been issued for "
+                "Sanic v20.12 while running Python 3.10+. You should either "
+                "use a supported version of Python (v3.6 - v3.9) or upgrade "
+                "Sanic to v21+.\n\nPlease see https://github.com/sanic-org/"
+                "sanic/security/advisories/GHSA-7p79-6x2v-5h88 for "
+                "more information.\n"
+            )
+
         self.name = name
         self.asgi = False
         self.router = router or Router(self)

sanic/server.py

@@ -169,7 +169,11 @@ class HttpProtocol(asyncio.Protocol):
         self.request_class = self.app.request_class or Request
         self.is_request_stream = self.app.is_request_stream
         self._is_stream_handler = False
-        self._not_paused = asyncio.Event(loop=deprecated_loop)
+        self._not_paused = (
+            asyncio.Event()
+            if sys.version_info >= (3, 10)
+            else asyncio.Event(loop=deprecated_loop)
+        )
         self._total_request_size = 0
         self._request_timeout_handler = None
         self._response_timeout_handler = None

sanic/static.py

@@ -1,7 +1,7 @@
 from functools import partial, wraps
 from mimetypes import guess_type
-from os import path
-from re import sub
+from os import path, sep
+from pathlib import Path
 from time import gmtime, strftime
 from urllib.parse import unquote
 
@@ -26,28 +26,41 @@ async def _static_request_handler(
     content_type=None,
     file_uri=None,
 ):
-    # Using this to determine if the URL is trying to break out of the path
-    # served.  os.path.realpath seems to be very slow
-    if file_uri and "../" in file_uri:
-        raise InvalidUsage("Invalid URL")
     # Merge served directory and requested file if provided
-    # Strip all / that in the beginning of the URL to help prevent python
-    # from herping a derp and treating the uri as an absolute path
-    root_path = file_path = file_or_directory
-    if file_uri:
-        file_path = path.join(file_or_directory, sub("^[/]*", "", file_uri))
+    file_path_raw = Path(unquote(file_or_directory))
+    root_path = file_path = file_path_raw.resolve()
+    not_found = FileNotFound(
+        "File not found",
+        path=file_or_directory,
+        relative_url=file_uri,
+    )
+
+    if file_uri:
+        # Strip all / that in the beginning of the URL to help prevent
+        # python from herping a derp and treating the uri as an
+        # absolute path
+        unquoted_file_uri = unquote(file_uri).lstrip("/")
+        file_path_raw = Path(file_or_directory, unquoted_file_uri)
+        file_path = file_path_raw.resolve()
+        if (
+            file_path < root_path and not file_path_raw.is_symlink()
+        ) or ".." in file_path_raw.parts:
+            error_logger.exception(
+                f"File not found: path={file_or_directory}, "
+                f"relative_url={file_uri}"
+            )
+            raise not_found
+
+    try:
+        file_path.relative_to(root_path)
+    except ValueError:
+        if not file_path_raw.is_symlink():
+            error_logger.exception(
+                f"File not found: path={file_or_directory}, "
+                f"relative_url={file_uri}"
+            )
+            raise not_found
 
-    # URL decode the path sent by the browser otherwise we won't be able to
-    # match filenames which got encoded (filenames with spaces etc)
-    file_path = path.abspath(unquote(file_path))
-    if not file_path.startswith(path.abspath(unquote(root_path))):
-        error_logger.exception(
-            f"File not found: path={file_or_directory}, "
-            f"relative_url={file_uri}"
-        )
-        raise FileNotFound(
-            "File not found", path=file_or_directory, relative_url=file_uri
-        )
     try:
         headers = {}
         # Check if the client has been sent this file before

tests/test_static.py

@@ -1,10 +1,14 @@
 import inspect
 import os
+import sys
 
+from pathlib import Path
 from time import gmtime, strftime
 
 import pytest
 
+from sanic.app import Sanic
+
 
 @pytest.fixture(scope="module")
 def static_file_directory():
@@ -15,6 +19,22 @@ def static_file_directory():
     return static_directory
 
 
+@pytest.fixture(scope="module")
+def double_dotted_directory_file(static_file_directory: str):
+    """Generate double dotted directory and its files"""
+    if sys.platform == "win32":
+        raise Exception("Windows doesn't support double dotted directories")
+
+    file_path = Path(static_file_directory) / "dotted.." / "dot.txt"
+    double_dotted_dir = file_path.parent
+    Path.mkdir(double_dotted_dir, exist_ok=True)
+    with open(file_path, "w") as f:
+        f.write("DOT\n")
+    yield file_path
+    Path.unlink(file_path)
+    Path.rmdir(double_dotted_dir)
+
+
 def get_file_path(static_file_directory, file_name):
     return os.path.join(static_file_directory, file_name)
 
@@ -374,3 +394,43 @@ def test_static_name(app, static_file_directory, static_name, file_name):
     request, response = app.test_client.get(f"/static/{file_name}")
 
     assert response.status == 200
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32",
+    reason="Windows does not support double dotted directories",
+)
+def test_dotted_dir_ok(
+    app: Sanic, static_file_directory: str, double_dotted_directory_file: Path
+):
+    app.static("/foo", static_file_directory)
+    dot_relative_path = str(
+        double_dotted_directory_file.relative_to(static_file_directory)
+    )
+    _, response = app.test_client.get("/foo/" + dot_relative_path)
+    assert response.status == 200
+    assert response.body == b"DOT\n"
+
+
+def test_breakout(app: Sanic, static_file_directory: str):
+    app.static("/foo", static_file_directory)
+
+    _, response = app.test_client.get("/foo/..%2Ffake/server.py")
+    assert response.status == 404
+
+    _, response = app.test_client.get("/foo/..%2Fstatic/test.file")
+    assert response.status == 404
+
+
+@pytest.mark.skipif(
+    sys.platform != "win32", reason="Block backslash on Windows only"
+)
+def test_double_backslash_prohibited_on_win32(
+    app: Sanic, static_file_directory: str
+):
+    app.static("/foo", static_file_directory)
+
+    _, response = app.test_client.get("/foo/static/..\\static/test.file")
+    assert response.status == 404
+    _, response = app.test_client.get("/foo/static\\../static/test.file")
+    assert response.status == 404