Handle 403 and 404 issues in FileResponse class

CHANGES/8182.bugfix.rst

@@ -0,0 +1,9 @@
+Move file existence and type checks to ``FileResponse``,
+and fix it to handle permission errors -- 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.

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,23 @@ 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()
         # 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
-        )
+
+        loop = asyncio.get_running_loop()
+        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 +333,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_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:
@@ -466,30 +517,51 @@ async def test_access_symlink_loop(
 async def test_access_special_resource(
     tmp_path: pathlib.Path, 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 socket.AF_UNIX:
+        pytest.skip("UNIX domain sockets not supported")
+
+    my_special = tmp_path / "my_special"
+    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: