Validate static paths

CHANGES/8079.bugfix.rst

@@ -0,0 +1 @@
+Improved validation of paths for static resources -- by :user:`bdraco`.

aiohttp/web_urldispatcher.py

@@ -573,9 +573,14 @@ def url_for(  # type: ignore[override]
             url = url / filename
 
         if append_version:
+            unresolved_path = self._directory.joinpath(filename)
             try:
-                filepath = self._directory.joinpath(filename).resolve()
-                if not self._follow_symlinks:
+                if self._follow_symlinks:
+                    normalized_path = Path(os.path.normpath(unresolved_path))
+                    normalized_path.relative_to(self._directory)
+                    filepath = normalized_path.resolve()
+                else:
+                    filepath = unresolved_path.resolve()
                     filepath.relative_to(self._directory)
             except (ValueError, FileNotFoundError):
                 # ValueError for case when path point to symlink
@@ -640,8 +645,13 @@ async def _handle(self, request: Request) -> StreamResponse:
                 # /static/\\machine_name\c$ or /static/D:\path
                 # where the static dir is totally different
                 raise HTTPForbidden()
-            filepath = self._directory.joinpath(filename).resolve()
-            if not self._follow_symlinks:
+            unresolved_path = self._directory.joinpath(filename)
+            if self._follow_symlinks:
+                normalized_path = Path(os.path.normpath(unresolved_path))
+                normalized_path.relative_to(self._directory)
+                filepath = normalized_path.resolve()
+            else:
+                filepath = unresolved_path.resolve()
                 filepath.relative_to(self._directory)
         except (ValueError, FileNotFoundError) as error:
             # relatively safe

docs/web_advanced.rst

@@ -262,12 +262,22 @@ instead could be enabled with ``show_index`` parameter set to ``True``::
 
    web.static('/prefix', path_to_static_folder, show_index=True)
 
-When a symlink from the static directory is accessed, the server responses to
-client with ``HTTP/404 Not Found`` by default. To allow the server to follow
-symlinks, parameter ``follow_symlinks`` should be set to ``True``::
+When a symlink that leads outside the static directory is accessed, the server
+responds to the client with ``HTTP/404 Not Found`` by default. To allow the server to
+follow symlinks that lead outside the static root, the parameter ``follow_symlinks``
+should be set to ``True``::
 
    web.static('/prefix', path_to_static_folder, follow_symlinks=True)
 
+.. caution::
+
+   Enabling ``follow_symlinks`` can be a security risk, and may lead to
+   a directory transversal attack. You do NOT need this option to follow symlinks
+   which point to somewhere else within the static directory, this option is only
+   used to break out of the security sandbox. Enabling this option is highly
+   discouraged, and only expected to be used for edge cases in a local
+   development setting where remote users do not have access to the server.
+
 When you want to enable cache busting,
 parameter ``append_version`` can be set to ``True``
 

docs/web_reference.rst

@@ -1784,9 +1784,15 @@ Application and Router
                               by default it's not allowed and HTTP/403 will
                               be returned on directory access.
 
-      :param bool follow_symlinks: flag for allowing to follow symlinks from
-                              a directory, by default it's not allowed and
-                              HTTP/404 will be returned on access.
+      :param bool follow_symlinks: flag for allowing to follow symlinks that lead
+                              outside the static root directory, by default it's not allowed and
+                              HTTP/404 will be returned on access.  Enabling ``follow_symlinks``
+                              can be a security risk, and may lead to a directory transversal attack.
+                              You do NOT need this option to follow symlinks which point to somewhere
+                              else within the static directory, this option is only used to break out
+                              of the security sandbox. Enabling this option is highly discouraged,
+                              and only expected to be used for edge cases in a local development
+                              setting where remote users do not have access to the server.
 
       :param bool append_version: flag for adding file version (hash)
                               to the url query string, this value will

tests/test_web_urldispatcher.py

@@ -108,6 +108,97 @@ async def test_follow_symlink(
     assert (await r.text()) == data
 
 
+async def test_follow_symlink_directory_traversal(
+    tmp_path: pathlib.Path, aiohttp_client: AiohttpClient
+) -> None:
+    # Tests that follow_symlinks does not allow directory transversal
+    data = "private"
+
+    private_file = tmp_path / "private_file"
+    private_file.write_text(data)
+
+    safe_path = tmp_path / "safe_dir"
+    safe_path.mkdir()
+
+    app = web.Application()
+
+    # Register global static route:
+    app.router.add_static("/", str(safe_path), follow_symlinks=True)
+    client = await aiohttp_client(app)
+
+    await client.start_server()
+    # We need to use a raw socket to test this, as the client will normalize
+    # the path before sending it to the server.
+    reader, writer = await asyncio.open_connection(client.host, client.port)
+    writer.write(b"GET /../private_file HTTP/1.1\r\n\r\n")
+    response = await reader.readuntil(b"\r\n\r\n")
+    assert b"404 Not Found" in response
+    writer.close()
+    await writer.wait_closed()
+    await client.close()
+
+
+async def test_follow_symlink_directory_traversal_after_normalization(
+    tmp_path: pathlib.Path, aiohttp_client: AiohttpClient
+) -> None:
+    # Tests that follow_symlinks does not allow directory transversal
+    # after normalization
+    #
+    # Directory structure
+    # |-- secret_dir
+    # |   |-- private_file (should never be accessible)
+    # |   |-- symlink_target_dir
+    # |       |-- symlink_target_file (should be accessible via the my_symlink symlink)
+    # |       |-- sandbox_dir
+    # |           |-- my_symlink -> symlink_target_dir
+    #
+    secret_path = tmp_path / "secret_dir"
+    secret_path.mkdir()
+
+    # This file is below the symlink target and should not be reachable
+    private_file = secret_path / "private_file"
+    private_file.write_text("private")
+
+    symlink_target_path = secret_path / "symlink_target_dir"
+    symlink_target_path.mkdir()
+
+    sandbox_path = symlink_target_path / "sandbox_dir"
+    sandbox_path.mkdir()
+
+    # This file should be reachable via the symlink
+    symlink_target_file = symlink_target_path / "symlink_target_file"
+    symlink_target_file.write_text("readable")
+
+    my_symlink_path = sandbox_path / "my_symlink"
+    pathlib.Path(str(my_symlink_path)).symlink_to(str(symlink_target_path), True)
+
+    app = web.Application()
+
+    # Register global static route:
+    app.router.add_static("/", str(sandbox_path), follow_symlinks=True)
+    client = await aiohttp_client(app)
+
+    await client.start_server()
+    # We need to use a raw socket to test this, as the client will normalize
+    # the path before sending it to the server.
+    reader, writer = await asyncio.open_connection(client.host, client.port)
+    writer.write(b"GET /my_symlink/../private_file HTTP/1.1\r\n\r\n")
+    response = await reader.readuntil(b"\r\n\r\n")
+    assert b"404 Not Found" in response
+    writer.close()
+    await writer.wait_closed()
+
+    reader, writer = await asyncio.open_connection(client.host, client.port)
+    writer.write(b"GET /my_symlink/symlink_target_file HTTP/1.1\r\n\r\n")
+    response = await reader.readuntil(b"\r\n\r\n")
+    assert b"200 OK" in response
+    response = await reader.readuntil(b"readable")
+    assert response == b"readable"
+    writer.close()
+    await writer.wait_closed()
+    await client.close()
+
+
 @pytest.mark.parametrize(
     "dir_name,filename,data",
     [