aio-libs · Dreamsorcerer · Jul 13, 2024 · Jul 8, 2024 · Jul 9, 2024 · Jul 12, 2024
@@ -0,0 +1,7 @@
+Fixed server response headers for ``Content-Type`` and ``Content-Encoding`` for
+static compressed files -- by :user:`steverep`.
+
+Server will now respond with a ``Content-Type`` appropriate for the compressed
+file (e.g. ``"application/gzip"``), and omit the ``Content-Encoding`` header.
+Users should expect that most clients will no longer decompress such responses
+by default.
@@ -1,9 +1,9 @@
 import asyncio
-import mimetypes
 import os
 import pathlib
 import sys
 from contextlib import suppress
+from mimetypes import MimeTypes
 from types import MappingProxyType
 from typing import (
     IO,
@@ -40,14 +40,35 @@
 
 NOSENDFILE: Final[bool] = bool(os.environ.get("AIOHTTP_NOSENDFILE"))
 
+CONTENT_TYPES: Final[MimeTypes] = MimeTypes()
+
 if sys.version_info < (3, 9):
-    mimetypes.encodings_map[".br"] = "br"
+    CONTENT_TYPES.encodings_map[".br"] = "br"
 
 # File extension to IANA encodings map that will be checked in the order defined.
 ENCODING_EXTENSIONS = MappingProxyType(
-    {ext: mimetypes.encodings_map[ext] for ext in (".br", ".gz")}
+    {ext: CONTENT_TYPES.encodings_map[ext] for ext in (".br", ".gz")}
+)
+
+FALLBACK_CONTENT_TYPE = "application/octet-stream"
+
+# Provide additional MIME type/extension pairs to be recognized.
+# https://en.wikipedia.org/wiki/List_of_archive_formats#Compression_only
+ADDITIONAL_CONTENT_TYPES = MappingProxyType(
+    {
+        "application/gzip": ".gz",
+        "application/x-brotli": ".br",
+        "application/x-bzip2": ".bz2",
+        "application/x-compress": ".Z",
+        "application/x-xz": ".xz",
+    }
 )
 
+# Add custom pairs and clear the encodings map so guess_type ignores them.
+CONTENT_TYPES.encodings_map.clear()
+for content_type, extension in ADDITIONAL_CONTENT_TYPES.items():
+    CONTENT_TYPES.add_type(content_type, extension)  # type: ignore[attr-defined]
+
 
 class FileResponse(StreamResponse):
     """A response object can be used to send files."""
@@ -192,14 +213,6 @@ async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter
         ):
             return await self._not_modified(request, etag_value, last_modified)
 
-        ct = None
-        if hdrs.CONTENT_TYPE not in self.headers:
-            ct, encoding = mimetypes.guess_type(str(file_path))
-            if not ct:
-                ct = "application/octet-stream"
-        else:
-            encoding = file_encoding
-
         status = self._status
         file_size = st.st_size
         count = file_size
@@ -274,11 +287,16 @@ async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter
                 # return a HTTP 206 for a Range request.
                 self.set_status(status)
 
-        if ct:
-            self.content_type = ct
-        if encoding:
-            self.headers[hdrs.CONTENT_ENCODING] = encoding
+        # If the Content-Type header is not already set, guess it based on the
+        # extension of the request path. The encoding returned by guess_type
+        #  can be ignored since the map was cleared above.
+        if hdrs.CONTENT_TYPE not in self.headers:
+            self.content_type = (
+                CONTENT_TYPES.guess_type(self._path)[0] or FALLBACK_CONTENT_TYPE
+            )
+
         if file_encoding:
+            self.headers[hdrs.CONTENT_ENCODING] = file_encoding
             self.headers[hdrs.VARY] = hdrs.ACCEPT_ENCODING
             # Disable compression if we are already sending
             # a compressed file since we don't want to double

@@ -1,5 +1,6 @@
 # type: ignore
 import asyncio
+import bz2
 import gzip
 import pathlib
 import socket
@@ -37,10 +38,12 @@ def hello_txt(request, tmp_path_factory) -> pathlib.Path:
         None: txt,
         "gzip": txt.with_suffix(f"{txt.suffix}.gz"),
         "br": txt.with_suffix(f"{txt.suffix}.br"),
+        "bzip2": txt.with_suffix(f"{txt.suffix}.bz2"),
     }
     hello[None].write_bytes(HELLO_AIOHTTP)
     hello["gzip"].write_bytes(gzip.compress(HELLO_AIOHTTP))
     hello["br"].write_bytes(brotli.compress(HELLO_AIOHTTP))
+    hello["bzip2"].write_bytes(bz2.compress(HELLO_AIOHTTP))
     encoding = getattr(request, "param", None)
     return hello[encoding]
 
@@ -318,10 +321,16 @@ async def handler(request):
 
 
 @pytest.mark.parametrize(
-    ("hello_txt", "expect_encoding"), [["gzip"] * 2, ["br"] * 2], indirect=["hello_txt"]
+    ("hello_txt", "expect_type"),
+    [
+        ("gzip", "application/gzip"),
+        ("br", "application/x-brotli"),
+        ("bzip2", "application/x-bzip2"),
+    ],
+    indirect=["hello_txt"],
 )
 async def test_static_file_with_content_encoding(
-    hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any, expect_encoding: str
+    hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any, expect_type: str
 ) -> None:
     """Test requesting static compressed files returns the correct content type and encoding."""
 
@@ -334,9 +343,9 @@ async def handler(request):
 
     resp = await client.get("/")
     assert resp.status == 200
-    assert resp.headers.get("Content-Encoding") == expect_encoding
-    assert resp.headers["Content-Type"] == "text/plain"
-    assert await resp.read() == HELLO_AIOHTTP
+    assert resp.headers.get("Content-Encoding") is None
+    assert resp.headers["Content-Type"] == expect_type
+    assert await resp.read() == hello_txt.read_bytes()
     resp.close()
 
     await resp.release()