Validate File When Requested

sanic/response.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-from datetime import datetime
-from email.utils import formatdate
+from datetime import datetime, timezone
+from email.utils import formatdate, parsedate_to_datetime
 from functools import partial
 from mimetypes import guess_type
 from os import path
@@ -319,9 +319,26 @@ def html(
     )
 
 
+async def validate_file(
+    request_headers: Header, last_modified: Union[datetime, float, int]
+):
+    if_modified_since = request_headers.get("If-Modified-Since")
+    if not if_modified_since:
+        return
+    if_modified_since = parsedate_to_datetime(if_modified_since)
+    if not isinstance(last_modified, datetime):
+        last_modified = datetime.fromtimestamp(
+            float(last_modified), tz=timezone.utc
+        ).replace(microsecond=0)
+    if last_modified <= if_modified_since:
+        return HTTPResponse(status=304)
+
+
 async def file(
     location: Union[str, PurePath],
     status: int = 200,
+    request_headers: Optional[Header] = None,
+    validate_when_requested: bool = False,
     mime_type: Optional[str] = None,
     headers: Optional[Dict[str, str]] = None,
     filename: Optional[str] = None,
@@ -331,7 +348,12 @@ async def file(
     _range: Optional[Range] = None,
 ) -> HTTPResponse:
     """Return a response object with file data.
-
+    :param status: HTTP response code. Won't enforce the passed in
+        status if only a part of the content will be sent (206)
+        or file is being validated (304).
+    :param request_headers: The request headers.
+    :param validate_when_requested: If True, will validate the
+        file when requested.
     :param location: Location of file on system.
     :param mime_type: Specific mime_type.
     :param headers: Custom Headers.
@@ -341,21 +363,31 @@ async def file(
     :param no_store: Any cache should not store this response.
     :param _range:
     """
-    headers = headers or {}
-    if filename:
-        headers.setdefault(
-            "Content-Disposition", f'attachment; filename="{filename}"'
-        )
 
     if isinstance(last_modified, datetime):
         last_modified = last_modified.replace(microsecond=0).timestamp()
     elif isinstance(last_modified, Default):
         stat = await stat_async(location)
         last_modified = stat.st_mtime
 
+    if (
+        validate_when_requested
+        and request_headers is not None
+        and last_modified
+    ):
+        response = await validate_file(request_headers, last_modified)
+        if response:
+            return response
+
+    headers = headers or {}
     if last_modified:
         headers.setdefault(
-            "last-modified", formatdate(last_modified, usegmt=True)
+            "Last-Modified", formatdate(last_modified, usegmt=True)
+        )
+
+    if filename:
+        headers.setdefault(
+            "Content-Disposition", f'attachment; filename="{filename}"'
         )
 
     if no_store:

tests/test_response.py

@@ -1,6 +1,7 @@
 import asyncio
 import inspect
 import os
+import time
 
 from collections import namedtuple
 from datetime import datetime
@@ -767,9 +768,9 @@ def file_route_no_store(request, filename):
         == formatdate(test_expires, usegmt=True)[:-6]
         # [:-6] to allow at most 1 min difference
         # It's minimal for cases like:
-        # Thu, 26 May 2022 05:36:49 GMT
+        # Thu, 26 May 2022 05:36:59 GMT
         # AND
-        # Thu, 26 May 2022 05:36:50 GMT
+        # Thu, 26 May 2022 05:37:00 GMT
     )
 
     assert "last-modified" in headers and headers.get(
@@ -798,3 +799,74 @@ def file_route_no_store(request, filename):
     assert "cache-control" in headers and f"no-store" == headers.get(
         "cache-control"
     )
+
+
+def test_file_validate(app: Sanic, static_file_directory: str):
+    file_name = "test_validate.txt"
+    static_file_directory = Path(static_file_directory)
+    file_path = static_file_directory / file_name
+    file_path = file_path.absolute()
+    test_max_age = 10
+
+    with open(file_path, "w+") as f:
+        f.write("foo\n")
+
+    @app.route("/validate", methods=["GET"])
+    def file_route_cache(request: Request):
+        return file(
+            file_path,
+            request_headers=request.headers,
+            max_age=test_max_age,
+            validate_when_requested=True,
+        )
+
+    _, response = app.test_client.get("/validate")
+
+    assert response.status == 200
+    assert response.body == b"foo\n"
+    last_modified = response.headers["Last-Modified"]
+
+    _, response = app.test_client.get("/validate")
+    assert response.status == 200
+    assert response.body == b"foo\n"
+
+    _, response = app.test_client.get(
+        "/validate", headers={"If-Modified-Since": last_modified}
+    )
+    assert response.status == 304
+    assert response.body == b""
+
+    _, response = app.test_client.get(
+        "/validate", headers={"if-modified-since": last_modified}
+    )
+    assert response.status == 304
+    assert response.body == b""
+
+    time.sleep(1)
+    with open(file_path, "a") as f:
+        f.write("bar\n")
+
+    _, response = app.test_client.get(
+        "/validate", headers={"If-Modified-Since": last_modified}
+    )
+    assert response.status == 200
+    assert response.body == b"foo\nbar\n"
+
+    _, response = app.test_client.get(
+        "/validate", headers={"if-modified-since": ""}
+    )
+    assert response.status == 200
+    assert response.body == b"foo\nbar\n"
+
+    _, response = app.test_client.get("/validate")
+    assert response.status == 200
+    assert response.body == b"foo\nbar\n"
+
+    last_modified = response.headers["Last-Modified"]
+    _, response = app.test_client.get(
+        "/validate", headers={"if-modified-since": last_modified}
+    )
+    assert response.status == 304
+    assert response.body == b""
+
+    file_path.unlink()