Handle 403 and 404 issues in FileResponse class

CHANGES/8182.bugfix.rst

@@ -0,0 +1,10 @@
+Adjusted ``FileResponse`` to check file existence and access when preparing the response -- by :user:`steverep`.
+
+The :py:class:`~aiohttp.web.FileResponse` class was modified to respond with
+ 403 Forbidden or 404 Not Found as appropriate.  Previously, it would cause a
+ server error if the path did not exist or could not be accessed.  Checks for
+ existence, non-regular files, and permissions were expected to be done in the
+ route handler.  For static routes, this now permits a compressed file to exist
+ without its uncompressed variant and still be served.  In addition, this
+ changes the response status for files without read permission to 403, and for
+ non-regular files from 404 to 403 for consistency.

aiohttp/web_fileresponse.py

@@ -4,6 +4,7 @@
 import sys
 from contextlib import suppress
 from mimetypes import MimeTypes
+from stat import S_ISREG
 from types import MappingProxyType
 from typing import (
     IO,
@@ -22,6 +23,8 @@
 from .helpers import ETAG_ANY, ETag, must_be_empty_body
 from .typedefs import LooseHeaders, PathLike
 from .web_exceptions import (
+    HTTPForbidden,
+    HTTPNotFound,
     HTTPNotModified,
     HTTPPartialContent,
     HTTPPreconditionFailed,
@@ -177,13 +180,22 @@ def _get_file_path_stat_encoding(
         return file_path, file_path.stat(), None
 
     async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter]:
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         # Encoding comparisons should be case-insensitive
         # https://www.rfc-editor.org/rfc/rfc9110#section-8.4.1
         accept_encoding = request.headers.get(hdrs.ACCEPT_ENCODING, "").lower()
-        file_path, st, file_encoding = await loop.run_in_executor(
-            None, self._get_file_path_stat_encoding, accept_encoding
-        )
+        try:
+            file_path, st, file_encoding = await loop.run_in_executor(
+                None, self._get_file_path_stat_encoding, accept_encoding
+            )
+        except FileNotFoundError:
+            self.set_status(HTTPNotFound.status_code)
+            return await super().prepare(request)
+
+        # Forbid special files like sockets, pipes, devices, etc.
+        if not S_ISREG(st.st_mode):
+            self.set_status(HTTPForbidden.status_code)
+            return await super().prepare(request)
 
         etag_value = f"{st.st_mtime_ns:x}-{st.st_size:x}"
         last_modified = st.st_mtime
@@ -320,7 +332,12 @@ async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter
         if count == 0 or must_be_empty_body(request.method, self.status):
             return await super().prepare(request)
 
-        fobj = await loop.run_in_executor(None, file_path.open, "rb")
+        try:
+            fobj = await loop.run_in_executor(None, file_path.open, "rb")
+        except PermissionError:
+            self.set_status(HTTPForbidden.status_code)
+            return await super().prepare(request)
+
         if start:  # be aware that start could be None or int=0 here.
             offset = start
         else:

aiohttp/web_urldispatcher.py

@@ -690,10 +690,7 @@ def _resolve_path_to_response(self, unresolved_path: Path) -> StreamResponse:
         except PermissionError as error:
             raise HTTPForbidden() from error
 
-        # Not a regular file or does not exist.
-        if not file_path.is_file():
-            raise HTTPNotFound()
-
+        # Return the file response, which handles all other checks.
         return FileResponse(file_path, chunk_size=self._chunk_size)
 
     def _directory_as_html(self, dir_path: Path) -> str:

tests/test_web_sendfile.py

@@ -1,11 +1,14 @@
 from pathlib import Path
+from stat import S_IFREG, S_IRUSR, S_IWUSR
 from typing import Any
 from unittest import mock
 
 from aiohttp import hdrs
 from aiohttp.test_utils import make_mocked_coro, make_mocked_request
 from aiohttp.web_fileresponse import FileResponse
 
+MOCK_MODE = S_IFREG | S_IRUSR | S_IWUSR
+
 
 def test_using_gzip_if_header_present_and_file_available(loop: Any) -> None:
     request = make_mocked_request(
@@ -18,6 +21,7 @@ def test_using_gzip_if_header_present_and_file_available(loop: Any) -> None:
     gz_filepath = mock.create_autospec(Path, spec_set=True)
     gz_filepath.stat.return_value.st_size = 1024
     gz_filepath.stat.return_value.st_mtime_ns = 1603733507222449291
+    gz_filepath.stat.return_value.st_mode = MOCK_MODE
 
     filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
@@ -39,12 +43,14 @@ def test_gzip_if_header_not_present_and_file_available(loop: Any) -> None:
     gz_filepath = mock.create_autospec(Path, spec_set=True)
     gz_filepath.stat.return_value.st_size = 1024
     gz_filepath.stat.return_value.st_mtime_ns = 1603733507222449291
+    gz_filepath.stat.return_value.st_mode = MOCK_MODE
 
     filepath = mock.create_autospec(Path, spec_set=True)
     filepath.name = "logo.png"
     filepath.with_suffix.return_value = gz_filepath
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
+    filepath.stat.return_value.st_mode = MOCK_MODE
 
     file_sender = FileResponse(filepath)
     file_sender._path = filepath
@@ -67,6 +73,7 @@ def test_gzip_if_header_not_present_and_file_not_available(loop: Any) -> None:
     filepath.with_suffix.return_value = gz_filepath
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
+    filepath.stat.return_value.st_mode = MOCK_MODE
 
     file_sender = FileResponse(filepath)
     file_sender._path = filepath
@@ -91,6 +98,7 @@ def test_gzip_if_header_present_and_file_not_available(loop: Any) -> None:
     filepath.with_suffix.return_value = gz_filepath
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
+    filepath.stat.return_value.st_mode = MOCK_MODE
 
     file_sender = FileResponse(filepath)
     file_sender._path = filepath
@@ -109,6 +117,7 @@ def test_status_controlled_by_user(loop: Any) -> None:
     filepath.name = "logo.png"
     filepath.stat.return_value.st_size = 1024
     filepath.stat.return_value.st_mtime_ns = 1603733507222449291
+    filepath.stat.return_value.st_mode = MOCK_MODE
 
     file_sender = FileResponse(filepath, status=203)
     file_sender._path = filepath

tests/test_web_sendfile_functional.py

@@ -40,7 +40,7 @@ def hello_txt(request, tmp_path_factory) -> pathlib.Path:
         "br": txt.with_suffix(f"{txt.suffix}.br"),
         "bzip2": txt.with_suffix(f"{txt.suffix}.bz2"),
     }
-    hello[None].write_bytes(HELLO_AIOHTTP)
+    # Uncompressed file is not actually written to test it is not required.
     hello["gzip"].write_bytes(gzip.compress(HELLO_AIOHTTP))
     hello["br"].write_bytes(brotli.compress(HELLO_AIOHTTP))
     hello["bzip2"].write_bytes(bz2.compress(HELLO_AIOHTTP))

tests/test_web_urldispatcher.py

@@ -1,10 +1,11 @@
 import asyncio
 import functools
+import os
 import pathlib
+import socket
 import sys
-from typing import Generator, Optional
-from unittest import mock
-from unittest.mock import MagicMock
+from stat import S_IFIFO, S_IMODE
+from typing import Any, Generator, Optional
 
 import pytest
 import yarl
@@ -445,6 +446,56 @@ def mock_is_dir(self: pathlib.Path) -> bool:
     assert r.status == 403
 
 
+@pytest.mark.skipif(
+    sys.platform.startswith("win32"), reason="Cannot remove read access on Windows"
+)
+async def test_static_file_without_read_permission(
+    tmp_path: pathlib.Path, aiohttp_client: AiohttpClient
+) -> None:
+    """Test static file without read permission receives forbidden response."""
+    my_file = tmp_path / "my_file.txt"
+    my_file.write_text("secret")
+    my_file.chmod(0o000)
+
+    app = web.Application()
+    app.router.add_static("/", str(tmp_path))
+    client = await aiohttp_client(app)
+
+    r = await client.get(f"/{my_file.name}")
+    assert r.status == 403
+
+
+async def test_static_file_with_mock_permission_error(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: pathlib.Path,
+    aiohttp_client: AiohttpClient,
+) -> None:
+    """Test static file with mock permission errors receives forbidden response."""
+    my_file = tmp_path / "my_file.txt"
+    my_file.write_text("secret")
+    my_readable = tmp_path / "my_readable.txt"
+    my_readable.write_text("info")
+
+    real_open = pathlib.Path.open
+
+    def mock_open(self: pathlib.Path, *args: Any, **kwargs: Any) -> Any:
+        if my_file.samefile(self):
+            raise PermissionError()
+        return real_open(self, *args, **kwargs)
+
+    monkeypatch.setattr("pathlib.Path.open", mock_open)
+
+    app = web.Application()
+    app.router.add_static("/", str(tmp_path))
+    client = await aiohttp_client(app)
+
+    # Test the mock only applies to my_file, then test the permission error.
+    r = await client.get(f"/{my_readable.name}")
+    assert r.status == 200
+    r = await client.get(f"/{my_file.name}")
+    assert r.status == 403
+
+
 async def test_access_symlink_loop(
     tmp_path: pathlib.Path, aiohttp_client: AiohttpClient
 ) -> None:
@@ -464,32 +515,54 @@ async def test_access_symlink_loop(
 
 
 async def test_access_special_resource(
-    tmp_path: pathlib.Path, aiohttp_client: AiohttpClient
+    tmp_path_factory: pytest.TempPathFactory, aiohttp_client: AiohttpClient
 ) -> None:
-    # Tests the access to a resource that is neither a file nor a directory.
-    # Checks that if a special resource is accessed (f.e. named pipe or UNIX
-    # domain socket) then 404 HTTP status returned.
+    """Test access to non-regular files is forbidden using a UNIX domain socket."""
+    if not getattr(socket, "AF_UNIX", None):
+        pytest.skip("UNIX domain sockets not supported")
+
+    tmp_path = tmp_path_factory.mktemp("special")
+    my_special = tmp_path / "sock"
+    my_socket = socket.socket(socket.AF_UNIX)
+    my_socket.bind(str(my_special))
+    assert my_special.is_socket()
+
     app = web.Application()
+    app.router.add_static("/", str(tmp_path))
 
-    with mock.patch("pathlib.Path.__new__") as path_constructor:
-        special = MagicMock()
-        special.is_dir.return_value = False
-        special.is_file.return_value = False
+    client = await aiohttp_client(app)
+    r = await client.get(f"/{my_special.name}")
+    assert r.status == 403
+    my_socket.close()
 
-        path = MagicMock()
-        path.joinpath.side_effect = lambda p: (special if p == "special" else path)
-        path.resolve.return_value = path
-        special.resolve.return_value = special
 
-        path_constructor.return_value = path
+async def test_access_mock_special_resource(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: pathlib.Path,
+    aiohttp_client: AiohttpClient,
+) -> None:
+    """Test access to non-regular files is forbidden using a mock FIFO."""
+    my_special = tmp_path / "my_special"
+    my_special.touch()
+
+    real_result = my_special.stat()
+    real_stat = pathlib.Path.stat
+
+    def mock_stat(self: pathlib.Path) -> os.stat_result:
+        s = real_stat(self)
+        if os.path.samestat(s, real_result):
+            mock_mode = S_IFIFO | S_IMODE(s.st_mode)
+            s = os.stat_result([mock_mode] + list(s)[1:])
+        return s
 
-        # Register global static route:
-        app.router.add_static("/", str(tmp_path), show_index=True)
-        client = await aiohttp_client(app)
+    monkeypatch.setattr("pathlib.Path.stat", mock_stat)
 
-        # Request the root of the static directory.
-        r = await client.get("/special")
-        assert r.status == 403
+    app = web.Application()
+    app.router.add_static("/", str(tmp_path))
+    client = await aiohttp_client(app)
+
+    r = await client.get(f"/{my_special.name}")
+    assert r.status == 403
 
 
 async def test_partially_applied_handler(aiohttp_client: AiohttpClient) -> None: