feat: Apply nonce template for requests with specified file extensions

changes/3585.feature.md

@@ -0,0 +1 @@
+Apply nonce template for requests with specified file extensions

src/ai/backend/web/config.py

@@ -37,6 +37,8 @@
         tx.AliasedKey(["ssl_cert", "ssl-cert"], default=None): t.Null | tx.Path(type="file"),
         tx.AliasedKey(["ssl_privkey", "ssl-privkey"], default=None): t.Null | tx.Path(type="file"),
         t.Key("static_path", default=default_static_path): tx.Path(type="dir"),
+        t.Key("template_extension_list", default=[".html", ".js", ".css"]): t.Null
+        | tx.StringList(empty_str_as_empty_list=True),
         tx.AliasedKey(
             ["force_endpoint_protocol", "force-endpoint-protocol"],
             default=None,

src/ai/backend/web/security.py

@@ -1,3 +1,4 @@
+import secrets
 from typing import Callable, Iterable, Self
 
 from aiohttp import web
@@ -9,34 +10,42 @@ async def security_policy_middleware(request: web.Request, handler: Handler) ->
     security_policy: SecurityPolicy = request.app["security_policy"]
     security_policy.check_request_policies(request)
     response = await handler(request)
-    return security_policy.apply_response_policies(response)
+    return security_policy.apply_response_policies(request, response)
 
 
 class SecurityPolicy:
     _request_policies: Iterable[Callable[[web.Request], None]]
-    _response_policies: Iterable[Callable[[web.StreamResponse], web.StreamResponse]]
+    _response_policies: Iterable[Callable[[web.Request, web.StreamResponse], web.StreamResponse]]
 
     def __init__(
         self,
         request_policies: Iterable[Callable[[web.Request], None]],
-        response_policies: Iterable[Callable[[web.StreamResponse], web.StreamResponse]],
+        response_policies: Iterable[
+            Callable[[web.Request, web.StreamResponse], web.StreamResponse]
+        ],
     ) -> None:
         self._request_policies = request_policies
         self._response_policies = response_policies
 
     @classmethod
     def default_policy(cls) -> Self:
-        request_policies = [reject_metadata_local_link_policy, reject_access_for_unsafe_file_policy]
+        request_policies = [
+            reject_metadata_local_link_policy,
+            reject_access_for_unsafe_file_policy,
+            add_nonce_policy,
+        ]
         response_policies = [add_self_content_security_policy, set_content_type_nosniff_policy]
         return cls(request_policies, response_policies)
 
     def check_request_policies(self, request: web.Request) -> None:
         for policy in self._request_policies:
             policy(request)
 
-    def apply_response_policies(self, response: web.StreamResponse) -> web.StreamResponse:
+    def apply_response_policies(
+        self, request: web.Request, response: web.StreamResponse
+    ) -> web.StreamResponse:
         for policy in self._response_policies:
-            response = policy(response)
+            response = policy(request, response)
         return response
 
 
@@ -68,13 +77,29 @@ def reject_access_for_unsafe_file_policy(request: web.Request) -> None:
         raise web.HTTPForbidden()
 
 
-def add_self_content_security_policy(response: web.StreamResponse) -> web.StreamResponse:
-    response.headers["Content-Security-Policy"] = (
-        "default-src 'self'; style-src 'self' 'unsafe-inline'; frame-ancestors 'none'; form-action 'self';"
-    )
+def add_nonce_policy(request: web.Request) -> None:
+    nonce = secrets.token_urlsafe(16)
+    request["request_nonce"] = nonce
+
+
+def add_self_content_security_policy(
+    request: web.Request, response: web.StreamResponse
+) -> web.StreamResponse:
+    nonce = request.get("request_nonce")
+    if nonce:
+        response.headers["Content-Security-Policy"] = (
+            f"default-src 'self'; style-src 'self' 'nonce-{nonce}'; frame-ancestors 'none'; form-action 'self';"
+        )
+    else:
+        # Fallback to unsafe-inline for unexpected cases.
+        response.headers["Content-Security-Policy"] = (
+            "default-src 'self'; style-src 'self' 'unsafe-inline'; frame-ancestors 'none'; form-action 'self';"
+        )
     return response
 
 
-def set_content_type_nosniff_policy(response: web.StreamResponse) -> web.StreamResponse:
+def set_content_type_nosniff_policy(
+    _: web.Request, response: web.StreamResponse
+) -> web.StreamResponse:
     response.headers["X-Content-Type-Options"] = "nosniff"
     return response

src/ai/backend/web/server.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import logging.config
+import mimetypes
 import os
 import re
 import socket
@@ -78,11 +79,25 @@ def apply_cache_headers(response: web.StreamResponse, path: str) -> web.StreamRe
     return response
 
 
+def apply_nonce_template(file_path: Path, nonce: str) -> web.StreamResponse:
+    """
+    {{nonce}} is replaced with a random nonce in the response.
+    """
+    with file_path.open("r", encoding="utf-8") as f:
+        content = f.read()
+    content = content.replace("{{nonce}}", nonce)
+    content_type = mimetypes.guess_type(file_path.name)[0]
+    if content_type is None:
+        content_type = "application/octet-stream"
+    return web.Response(text=content, content_type=content_type)
+
+
 async def static_handler(request: web.Request) -> web.StreamResponse:
     stats: WebStats = request.app["stats"]
     stats.active_static_handlers.add(asyncio.current_task())  # type: ignore
     request_path = request.match_info["path"]
-    static_path = request.app["config"]["service"]["static_path"]
+    static_path: Path = request.app["config"]["service"]["static_path"]
+    template_extension_list: list[str] = request.app["config"]["service"]["template_extension_list"]
     file_path = (static_path / request_path).resolve()
     try:
         file_path.relative_to(static_path)
@@ -94,6 +109,9 @@ async def static_handler(request: web.Request) -> web.StreamResponse:
             }),
             content_type="application/problem+json",
         )
+    if file_path.suffix in template_extension_list:
+        nonce: str = request["request_nonce"]
+        return apply_nonce_template(file_path, nonce)
     if file_path.is_file():
         return apply_cache_headers(web.FileResponse(file_path), request_path)
     return web.HTTPNotFound(
@@ -142,7 +160,8 @@ async def console_handler(request: web.Request) -> web.StreamResponse:
     stats.active_webui_handlers.add(asyncio.current_task())  # type: ignore
     request_path = request.match_info["path"]
     config = request.app["config"]
-    static_path = config["service"]["static_path"]
+    static_path: Path = config["service"]["static_path"]
+    template_extension_list: list[str] = request.app["config"]["service"]["template_extension_list"]
     file_path = (static_path / request_path).resolve()
     # SECURITY: only allow reading files under static_path
     try:
@@ -155,6 +174,9 @@ async def console_handler(request: web.Request) -> web.StreamResponse:
             }),
             content_type="application/problem+json",
         )
+    if file_path.suffix in template_extension_list:
+        nonce: str = request["request_nonce"]
+        return apply_nonce_template(file_path, nonce)
     if file_path.is_file():
         return apply_cache_headers(web.FileResponse(file_path), request_path)
     # Fallback to index.html to support the URL routing for single-page application.

tests/webserver/test_security_policy.py

@@ -39,9 +39,10 @@ async def test_default_security_policy_reject_metadata_local_link(
 async def test_default_security_policy_response(default_app, async_handler) -> None:
     request = make_mocked_request("GET", "/", headers={"Host": "localhost"}, app=default_app)
     response = await security_policy_middleware(request, async_handler)
+    nonce = request.get("request_nonce")
     assert (
         response.headers["Content-Security-Policy"]
-        == "default-src 'self'; style-src 'self' 'unsafe-inline'; frame-ancestors 'none'; form-action 'self';"
+        == f"default-src 'self'; style-src 'self' 'nonce-{nonce}'; frame-ancestors 'none'; form-action 'self';"
     )
     assert response.headers["X-Content-Type-Options"] == "nosniff"