Implement If-None-Match handling

1 files changed, 55 insertions, 1 deletions

diff --git a/paste/__init__.py b/paste/__init__.py
index 0800d93..f7ee3a8 100644
--- a/paste/__init__.py
+++ b/paste/__init__.py
@@ -1,7 +1,7 @@
 from base64 import b64decode, b64encode
 from functools import wraps
 from wsgiref.util import request_uri
-from typing import Optional, Any, Protocol
+from typing import Optional, Any, Protocol, runtime_checkable
 from collections.abc import Callable, Iterable
 import binascii
 import traceback
@@ -11,6 +11,12 @@ from . import db
 from . import store
 
 
+@runtime_checkable
+class Closable(Protocol):
+    def close(self):
+        ...
+
+
 class StartResponse(Protocol):
     def __call__(
         self,
@@ -95,6 +101,53 @@ def validate_method(app: App, environ: Env, start_response: StartResponse) -> Re
 
 
 @middleware
+def if_none_match(app: App, environ: Env, start_response: StartResponse) -> Response:
+    if "HTTP_IF_NONE_MATCH" not in environ:
+        return app(environ, start_response)
+    if_none_match = environ["HTTP_IF_NONE_MATCH"]
+    del environ["HTTP_IF_NONE_MATCH"]
+    if environ["REQUEST_METHOD"] not in {"GET", "HEAD"}:
+        return app(environ, start_response)
+    head_env = environ.copy()
+    head_env["REQUEST_METHOD"] = "HEAD"
+    etag = None
+
+    def head_start_response(
+        status: str,
+        headers: list[tuple[str, str]],
+        exc_info: Optional[tuple] = None,
+    ) -> Callable[[bytes], object]:
+        _, _ = status, exc_info
+        nonlocal etag
+        for key, value in headers:
+            if key == "ETag":
+                etag = value[1:-1]
+        return lambda _: None
+
+    resp = app(head_env, head_start_response)
+    if isinstance(resp, Closable):
+        resp.close()
+
+    if not isinstance(etag, str):
+        return app(environ, start_response)
+
+    if if_none_match == "*":
+        start_response("304 Not Modified", [("ETag", etag)])
+        return []
+
+    etags = if_none_match.split(",")
+    etags = {e.strip(" \t").removeprefix("W/") for e in etags}
+    for e in etags:
+        if e[0] != '"' or e[-1] != '"':
+            return simple_response(start_response, "400 Bad Request")
+    etags = {e[1:-1] for e in etags}
+    if isinstance(etag, str) and etag in etags:
+        start_response("304 Not Modified", [("ETag", etag)])
+        return []
+    return app(environ, start_response)
+
+
+@middleware
 def options(app: App, environ: Env, start_response: StartResponse) -> Response:
     if environ["REQUEST_METHOD"] != "OPTIONS":
         return app(environ, start_response)
@@ -140,6 +193,7 @@ def authenticate(app: App, environ: Env, start_response: StartResponse) -> Respo
 @catch_exceptions
 @validate_method
 @options
+@if_none_match
 @open_database
 @authenticate
 def application(environ: Env, start_response: StartResponse) -> Response: