Handle 403 and 404 issues in FileResponse class (aio-libs#8538)

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: