Fix #679

CHANGELOG.md

@@ -20,6 +20,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   crashed with `AttributeError` when a handler returned `AsyncIterable[ServerSentEvent]`
   (or any `collections.abc` async-iterable generic). `_try_get_schema_for_generic` now
   skips types whose origin has no `__parameters__`.
+- Feature [#679](https://github.com/Neoteroi/BlackSheep/issues/679): add `on_response`
+  async callback to `serve_files()` and `serve_files_dynamic()`. The callback receives
+  `(request, response)` and can mutate the response in-place (e.g. inject a dynamic
+  `Content-Security-Policy` header). It is called for every response produced by the
+  static file handler.
 
 ## [2.6.2] - 2026-02-25 :gift:
 

blacksheep/scribe.pyx

@@ -6,7 +6,6 @@ from .cookies cimport Cookie, write_cookie_for_response
 from .messages cimport Request, Response
 from .url cimport URL
 
-
 MAX_RESPONSE_CHUNK_SIZE = 61440  # 64kb — Python-accessible
 cdef int _MAX_RESPONSE_CHUNK_SIZE = MAX_RESPONSE_CHUNK_SIZE
 

blacksheep/server/application.py

@@ -49,7 +49,7 @@
 from blacksheep.server.env import EnvironmentSettings
 from blacksheep.server.errors import ServerErrorDetailsHandler
 from blacksheep.server.files import DefaultFileOptions
-from blacksheep.server.files.dynamic import serve_files_dynamic
+from blacksheep.server.files.dynamic import ResponseCallback, serve_files_dynamic
 from blacksheep.server.normalization import normalize_handler, normalize_middleware
 from blacksheep.server.process import use_shutdown_handler
 from blacksheep.server.remotes.scheme import configure_scheme_middleware
@@ -598,6 +598,7 @@ def serve_files(
         fallback_document: str | None = None,
         allow_anonymous: bool = True,
         default_file_options: DefaultFileOptions | None = None,
+        on_response: ResponseCallback | None = None,
     ):
         """
         Configures dynamic file serving from a given folder, relative to the server cwd.
@@ -620,6 +621,9 @@ def serve_files(
             use HTML5 History API for client side routing.
             default_file_options: Optional options to serve the default file
             (index.html)
+            on_response: Optional async callback called for every response produced by
+            the static file handler, receiving (request, response). Can be used to
+            inject dynamic response headers such as Content-Security-Policy.
         """
         serve_files_dynamic(
             self.router,
@@ -633,6 +637,7 @@ def serve_files(
             fallback_document=fallback_document,
             anonymous_access=allow_anonymous,
             default_file_options=default_file_options,
+            on_response=on_response,
         )
 
     def _apply_middlewares_in_routes(self):
@@ -809,11 +814,9 @@ async def _handle_websocket(self, scope, receive, send) -> None:
         root_path = scope.get("root_path", "")
         path = scope["path"]
         if root_path and path.startswith(root_path):
-            path = path[len(root_path):] or "/"
+            path = path[len(root_path) :] or "/"
 
-        route = self.router.get_match_by_method_and_path(
-            RouteMethod.GET_WS, path
-        )
+        route = self.router.get_match_by_method_and_path(RouteMethod.GET_WS, path)
 
         if route is None:
             await ws.close()
@@ -854,7 +857,7 @@ def instantiate_request(self, scope, receive) -> Request:
 
         root_path = scope.get("root_path", "")
         if root_path and raw_path.startswith(root_path.encode("utf8")):
-            raw_path = raw_path[len(root_path.encode("utf8")):] or b"/"
+            raw_path = raw_path[len(root_path.encode("utf8")) :] or b"/"
 
         request = Request.incoming(
             scope["method"],

blacksheep/server/authentication/oidc.py

@@ -230,9 +230,7 @@ def build_signin_parameters(self, request: Request):
         if self._settings.use_pkce:
             code_verifier = generate_pkce_code_verifier()
             state["code_verifier"] = code_verifier
-            parameters["code_challenge"] = generate_pkce_code_challenge(
-                code_verifier
-            )
+            parameters["code_challenge"] = generate_pkce_code_challenge(code_verifier)
             parameters["code_challenge_method"] = "S256"
 
         if self._settings.audience:

blacksheep/server/files/dynamic.py

@@ -175,6 +175,9 @@ def get_response_for_resource_path(
     return get_response_for_file(files_handler, request, resource_path, cache_time)
 
 
+ResponseCallback = Callable[[Request, Response], Awaitable[None]]
+
+
 def get_files_route_handler(
     files_handler: FilesHandler,
     source_folder_name: str,
@@ -185,6 +188,7 @@ def get_files_route_handler(
     index_document: str | None,
     fallback_document: str | None,
     default_file_options: DefaultFileOptions | None = None,
+    on_response: ResponseCallback | None = None,
 ) -> Callable[[Request], Awaitable[Response]]:
     files_list_html = get_resource_file_content("fileslist.html")
     source_folder_full_path = os.path.abspath(str(source_folder_name))
@@ -194,7 +198,7 @@ async def static_files_handler(request: Request) -> Response:
         tail = unquote(request.route_values.get("tail", "")).lstrip("/")
 
         try:
-            return get_response_for_resource_path(
+            response = get_response_for_resource_path(
                 request,
                 tail,
                 files_list_html,
@@ -230,7 +234,10 @@ async def static_files_handler(request: Request) -> Response:
             if default_file_options and index_document == fallback_document:
                 default_file_options.handle(request, response)
 
-            return response
+        if on_response is not None:
+            await on_response(request, response)
+
+        return response
 
     return static_files_handler
 
@@ -258,6 +265,7 @@ def serve_files_dynamic(
     fallback_document: str | None,
     anonymous_access: bool = True,
     default_file_options: DefaultFileOptions | None = None,
+    on_response: ResponseCallback | None = None,
 ) -> None:
     """
     Configures a route to serve files dynamically, using the given files handler and
@@ -284,6 +292,7 @@ def serve_files_dynamic(
         index_document,
         fallback_document,
         default_file_options,
+        on_response,
     )
 
     if anonymous_access:

blacksheep/server/openapi/docstrings.py

@@ -168,7 +168,7 @@ def get_parameters_info(self, docstring: str) -> dict[str, ParameterInfo]:
                 # @param int foo:
                 # @param int or None foo:
 
-                (*type_parts, name) = param_name.split(" ")
+                *type_parts, name = param_name.split(" ")
                 types[name] = " ".join(type_parts)
                 param_name = name
 

tests/test_auth.py

@@ -304,9 +304,7 @@ async def home():
 
     assert app.response.status == 200
     content = await app.response.text()
-    assert (
-        content
-        == """<!DOCTYPE html>
+    assert content == """<!DOCTYPE html>
 <html>
   <head>
     <title>Example.</title>
@@ -319,7 +317,6 @@ async def home():
   </body>
 </html>
 """
-    )
 
 
 async def test_authorization_supports_allow_anonymous(app):
@@ -1082,11 +1079,12 @@ async def test_jwt_openid_tokens_handler_authenticate_with_refresh_token(
     Verifies that JWTOpenIDTokensHandler.authenticate restores the refresh token
     on the returned identity when the refresh token header is present.
     """
+    from itsdangerous import URLSafeSerializer
+
     from blacksheep.server.authentication.oidc import (
-        JWTOpenIDTokensHandler,
         HTMLStorageType,
+        JWTOpenIDTokensHandler,
     )
-    from itsdangerous import URLSafeSerializer
 
     jwt_auth = JWTBearerAuthentication(
         valid_audiences=["test-audience"],

tests/test_bindings.py

@@ -1265,7 +1265,10 @@ class MultiItem:
 
 async def test_multi_format_binder_dispatches_to_json():
     binder = MultiFormatBodyBinder(
-        [JSONBinder(MultiItem, "body", False, True), FormBinder(MultiItem, "body", False, True)],
+        [
+            JSONBinder(MultiItem, "body", False, True),
+            FormBinder(MultiItem, "body", False, True),
+        ],
         MultiItem,
         "body",
         required=True,
@@ -1282,7 +1285,10 @@ async def test_multi_format_binder_dispatches_to_json():
 
 async def test_multi_format_binder_dispatches_to_form():
     binder = MultiFormatBodyBinder(
-        [JSONBinder(MultiItem, "body", False, True), FormBinder(MultiItem, "body", False, True)],
+        [
+            JSONBinder(MultiItem, "body", False, True),
+            FormBinder(MultiItem, "body", False, True),
+        ],
         MultiItem,
         "body",
         required=True,
@@ -1298,7 +1304,10 @@ async def test_multi_format_binder_dispatches_to_form():
 
 async def test_multi_format_binder_raises_415_for_unsupported_content_type():
     binder = MultiFormatBodyBinder(
-        [JSONBinder(MultiItem, "body", False, True), FormBinder(MultiItem, "body", False, True)],
+        [
+            JSONBinder(MultiItem, "body", False, True),
+            FormBinder(MultiItem, "body", False, True),
+        ],
         MultiItem,
         "body",
         required=True,
@@ -1315,7 +1324,10 @@ async def test_multi_format_binder_raises_415_for_unsupported_content_type():
 
 async def test_multi_format_binder_returns_none_when_optional_and_no_match():
     binder = MultiFormatBodyBinder(
-        [JSONBinder(MultiItem, "body", False, False), FormBinder(MultiItem, "body", False, False)],
+        [
+            JSONBinder(MultiItem, "body", False, False),
+            FormBinder(MultiItem, "body", False, False),
+        ],
         MultiItem,
         "body",
         required=False,
@@ -1365,7 +1377,6 @@ def test_multi_format_binder_content_type_combines_inner():
 from blacksheep.contents import Content
 from blacksheep.server.bindings import FromXML, XMLBinder
 
-
 XML_ITEM = b"<Item><name>hello</name><value>7</value></Item>"
 XML_NESTED = b"<Root><inner><x>1</x></inner></Root>"
 XML_ATTR = b'<Item id="99"><name>attr</name></Item>'
@@ -1386,9 +1397,9 @@ async def test_xml_binder_parses_simple_fields():
 
 async def test_xml_binder_accepts_text_xml_content_type():
     binder = XMLBinder(MultiItem, "body", required=True)
-    request = Request(
-        "POST", b"/", [(b"content-type", b"text/xml")]
-    ).with_content(Content(b"text/xml", XML_ITEM))
+    request = Request("POST", b"/", [(b"content-type", b"text/xml")]).with_content(
+        Content(b"text/xml", XML_ITEM)
+    )
 
     result = await binder.get_value(request)
     assert isinstance(result, MultiItem)
@@ -1433,7 +1444,11 @@ def test_xml_binder_rejects_xxe_attack():
         b"<Item><name>&xxe;</name><value>1</value></Item>"
     )
     with pytest.raises(
-        (defusedxml.DTDForbidden, defusedxml.EntitiesForbidden, defusedxml.ExternalReferenceForbidden)
+        (
+            defusedxml.DTDForbidden,
+            defusedxml.EntitiesForbidden,
+            defusedxml.ExternalReferenceForbidden,
+        )
     ):
         XMLBinder._parse_xml(xxe_payload)
 
@@ -1455,29 +1470,32 @@ def test_xml_binder_rejects_billion_laughs():
 
 
 def test_element_to_dict_handles_attributes():
-    from blacksheep.server.bindings import _element_to_dict
     import xml.etree.ElementTree as ET
 
+    from blacksheep.server.bindings import _element_to_dict
+
     root = ET.fromstring(XML_ATTR)
     d = _element_to_dict(root)
     assert d["id"] == "99"
     assert d["name"] == "attr"
 
 
 def test_element_to_dict_handles_nested():
-    from blacksheep.server.bindings import _element_to_dict
     import xml.etree.ElementTree as ET
 
+    from blacksheep.server.bindings import _element_to_dict
+
     root = ET.fromstring(XML_NESTED)
     d = _element_to_dict(root)
     assert isinstance(d["inner"], dict)
     assert d["inner"]["x"] == "1"
 
 
 def test_element_to_dict_collects_repeated_tags_as_list():
-    from blacksheep.server.bindings import _element_to_dict
     import xml.etree.ElementTree as ET
 
+    from blacksheep.server.bindings import _element_to_dict
+
     root = ET.fromstring(XML_LIST)
     d = _element_to_dict(root)
     assert d["tag"] == ["a", "b"]

tests/test_controllers.py

@@ -284,12 +284,10 @@ class Home(Controller):
         @get("/")
         async def index(self):
             assert isinstance(self, Home)
-            return self.html(
-                """
+            return self.html("""
                 <h1>Title</h1>
                 <p>Lorem ipsum</p>
-                """
-            )
+                """)
 
     await app(get_example_scope("GET", "/"), MockReceive(), MockSend())
 

tests/test_files_serving.py

@@ -9,6 +9,7 @@
 from blacksheep import Application, Request
 from blacksheep.common.files.asyncfs import FileContext, FilesHandler
 from blacksheep.exceptions import BadRequest, InvalidArgument
+from blacksheep.messages import Response
 from blacksheep.ranges import Range, RangePart
 from blacksheep.server.files import (
     DefaultFileOptions,
@@ -1193,3 +1194,52 @@ async def test_mounted_app_without_child_prefix_returns_404_for_unknown_file():
     await parent_app(scope, MockReceive(), send)
 
     assert send.messages[0]["status"] == 404
+
+
+async def test_serve_files_on_response_callback_sets_csp_dynamically(
+    app: Application,
+):
+    """
+    Demonstrates using the async on_response callback on serve_files() to inject a
+    dynamic Content-Security-Policy header derived from the incoming request.
+    """
+
+    async def add_csp(request: Request, response: Response) -> None:
+        site = request.query.get("site", ["none"])[0]
+        response.add_header(
+            b"content-security-policy",
+            f"frame-ancestors https://{site}.example.com".encode(),
+        )
+
+    app.serve_files(
+        get_folder_path("files2"),
+        fallback_document="index.html",
+        on_response=add_csp,
+    )
+
+    await app.start()
+
+    # Direct file hit
+    scope = get_example_scope("GET", "/index.html", query={"site": "acme"})
+    await app(scope, MockReceive(), MockSend())
+    response = app.response
+    assert response.status == 200
+    assert response.headers[b"content-security-policy"] == (
+        b"frame-ancestors https://acme.example.com",
+    )
+
+    # Fallback path (SPA route not matching any file)
+    scope = get_example_scope("GET", "/some-spa-route", query={"site": "beta"})
+    await app(scope, MockReceive(), MockSend())
+    response = app.response
+    assert response.status == 200
+    assert response.headers[b"content-security-policy"] == (
+        b"frame-ancestors https://beta.example.com",
+    )
+
+    # Static asset also receives the callback
+    scope = get_example_scope("GET", "/scripts/main.js")
+    await app(scope, MockReceive(), MockSend())
+    response = app.response
+    assert response.status == 200
+    assert b"content-security-policy" in response.headers

tests/test_normalization.py

@@ -680,7 +680,13 @@ async def example_2() -> AsyncIterable[str]:
 from dataclasses import dataclass
 
 from blacksheep import FormContent, JSONContent, Request
-from blacksheep.server.bindings import FormBinder, FromBody, FromBodyBinder, FromForm, MultiFormatBodyBinder
+from blacksheep.server.bindings import (
+    FormBinder,
+    FromBody,
+    FromBodyBinder,
+    FromForm,
+    MultiFormatBodyBinder,
+)
 
 
 @dataclass