Allow for more ui customizations

Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com>

Author: timkpaine

Makefile

@@ -97,7 +97,7 @@ test-py:  ## run python tests
 tests-py: test-py
 
 coverage-py:  ## run python tests and collect test coverage
-	python -m pytest -v csp_gateway/tests --cov=csp_gateway --cov-report term-missing --cov-report xml
+	python -m pytest -v csp_gateway/tests --junitxml=junit.xml --cov=csp_gateway --cov-report term-missing --cov-report xml
 
 .PHONY: test-js tests-js coverage-js
 test-js:  ## run js tests

csp_gateway/server/middleware/api_key.py

@@ -66,8 +66,8 @@ def rest(self, app: GatewayWebApp) -> None:
         check = self.get_check_dependency()
 
         @auth_router.get("/login")
-        async def route_login_and_add_cookie(api_key: str = Depends(check)):
-            response = RedirectResponse(url="/")
+        async def route_login_and_add_cookie(request: Request, api_key: str = Depends(check)):
+            response = RedirectResponse(url=app.root_path_url(request, "/"))
             response.set_cookie(
                 self.api_key_name,
                 value=api_key,
@@ -79,8 +79,8 @@ async def route_login_and_add_cookie(api_key: str = Depends(check)):
             return response
 
         @auth_router.get("/logout")
-        async def route_logout_and_remove_cookie():
-            response = RedirectResponse(url="/login")
+        async def route_logout_and_remove_cookie(request: Request):
+            response = RedirectResponse(url=app.root_path_url(request, "/login"))
             response.delete_cookie(self.api_key_name, domain=self.domain)
             return response
 
@@ -94,7 +94,7 @@ def _setup_public_routes(self, app: GatewayWebApp) -> None:
         async def get_login_page(token: str = "", request: Request = None):
             if token:
                 if token != "":
-                    return RedirectResponse(url=f"{app.settings.API_STR}/auth/login?token={token}")
+                    return RedirectResponse(url=app.root_path_url(request, f"{app.settings.API_STR}/auth/login?token={token}"))
             return app.templates.TemplateResponse(
                 request,
                 "login.html.j2",

csp_gateway/server/middleware/oauth.py

@@ -311,7 +311,7 @@ async def oauth_callback(request: Request, code: str = None, error: str = None):
                     "id_token": tokens.get("id_token"),
                 }
 
-                response = RedirectResponse(url="/")
+                response = RedirectResponse(url=app.root_path_url(request, "/"))
                 response.set_cookie(
                     self.cookie_name,
                     value=session_uuid,
@@ -332,7 +332,7 @@ async def logout(request: Request):
             if session_uuid and session_uuid in self._identity_store:
                 self._identity_store.pop(session_uuid, None)
 
-            response = RedirectResponse(url="/login")
+            response = RedirectResponse(url=app.root_path_url(request, "/login"))
             response.delete_cookie(self.cookie_name, domain=self.domain)
             return response
 
@@ -355,4 +355,4 @@ async def auth_error_handler(request: Request, exc):
                     {"detail": self.unauthorized_status_message, "status_code": exc.status_code},
                     status_code=exc.status_code,
                 )
-            return RedirectResponse(url="/login")
+            return RedirectResponse(url=app.root_path_url(request, "/login"))

csp_gateway/server/middleware/simple.py

@@ -383,13 +383,13 @@ async def post_login(request: Request):
                 password = form.get("password", "")
 
                 if not username or not password:
-                    return RedirectResponse(url="/login?error=missing_credentials", status_code=303)
+                    return RedirectResponse(url=app.root_path_url(request, "/login?error=missing_credentials"), status_code=303)
 
                 identity = self._validate_credentials(str(username), str(password))
 
                 if identity and isinstance(identity, dict):
                     session_uuid = self._create_session(identity)
-                    response = RedirectResponse(url="/", status_code=303)
+                    response = RedirectResponse(url=app.root_path_url(request, "/"), status_code=303)
                     response.set_cookie(
                         self.cookie_name,
                         value=session_uuid,
@@ -400,12 +400,12 @@ async def post_login(request: Request):
                     )
                     return response
 
-                return RedirectResponse(url="/login?error=invalid_credentials", status_code=303)
+                return RedirectResponse(url=app.root_path_url(request, "/login?error=invalid_credentials"), status_code=303)
 
         @auth_router.get("/login")
-        async def api_login(session_uuid: str = Depends(check)):
+        async def api_login(request: Request, session_uuid: str = Depends(check)):
             """API login endpoint - validates Basic Auth and returns session cookie."""
-            response = RedirectResponse(url="/")
+            response = RedirectResponse(url=app.root_path_url(request, "/"))
             response.set_cookie(
                 self.cookie_name,
                 value=session_uuid,
@@ -423,7 +423,7 @@ async def logout(request: Request):
             if session_uuid and session_uuid in self._identity_store:
                 self._identity_store.pop(session_uuid, None)
 
-            response = RedirectResponse(url="/login")
+            response = RedirectResponse(url=app.root_path_url(request, "/login"))
             response.delete_cookie(self.cookie_name, domain=self.domain)
             return response
 
@@ -450,7 +450,7 @@ async def auth_error_handler(request: Request, exc):
                     status_code=exc.status_code,
                 )
             if self.enable_form_login:
-                return RedirectResponse(url="/login")
+                return RedirectResponse(url=app.root_path_url(request, "/login"))
             # Return 401 with Basic Auth challenge
             return JSONResponse(
                 {"detail": self.unauthorized_status_message},

csp_gateway/server/settings.py

@@ -33,9 +33,47 @@ class Settings(BaseSettings):
     BIND: str = "0.0.0.0"
     PORT: int = 8000
 
+    ROOT_PATH: str = Field(
+        default="",
+        description="URL path prefix the app is served under when behind a reverse proxy that "
+        "strips the prefix (e.g. '/watchtower'). Passed to the ASGI server as root_path and used "
+        "to prefix all server-rendered asset/API URLs so the UI works under a sub-path. Leave "
+        "empty when served at a domain root.",
+    )
+
     UI: bool = Field(False, description="Enables ui in the web application")
 
-    # --- DEPRECATED auth settings ---
+    # UI customization fields that let downstream applications white-label the
+    # default UI from server-side config alone, without a custom Javascript bundle.
+    HEADER_LOGO: Optional[str] = Field(
+        default=None,
+        description="Header logo image, given as an http(s) URL, a data URI, an absolute "
+        "URL path, or a local file path (local files are served automatically).",
+    )
+    FOOTER_LOGO: Optional[str] = Field(
+        default=None,
+        description="Footer logo image, given as an http(s) URL, a data URI, an absolute "
+        "URL path, or a local file path (local files are served automatically).",
+    )
+    CUSTOM_JS: List[str] = Field(
+        default_factory=list,
+        description="Custom Javascript files to inject into the UI, given as URLs or local "
+        "file paths (local files are served automatically). Loaded after the main bundle.",
+    )
+    CUSTOM_CSS: List[str] = Field(
+        default_factory=list,
+        description="Custom CSS files to inject into the UI, given as URLs or local file "
+        "paths (local files are served automatically). Loaded after the main stylesheet.",
+    )
+    CUSTOM_STATIC_DIR: Optional[str] = Field(
+        default=None,
+        description="Local directory served at <root_path>/custom. The entire directory is exposed "
+        "as public static content (useful for assets like logos, e.g. /custom/logo.svg); its top-level "
+        "*.js and *.css files are additionally auto-injected into the UI in sorted filename order. Do "
+        "not point this at a directory containing private files.",
+    )
+
+    # DEPRECATED auth settings
     # Historically (csp-gateway <2.5), auth was configured via these two fields
     # on Settings. In 2.5+, auth moved onto `MountAPIKeyMiddleware` as module
     # fields. Keeping these here (default-None sentinels) lets existing YAML

csp_gateway/server/web/app.py

@@ -8,7 +8,7 @@
 from typing import Any, Callable, Dict, List, Optional, Set
 
 from csp.impl.types.tstype import isTsType
-from fastapi import APIRouter, FastAPI
+from fastapi import APIRouter, FastAPI, HTTPException, Request
 from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
 from fastapi.openapi.utils import get_openapi
 from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
@@ -77,11 +77,13 @@ def __init__(
         _in_test: bool = False,
     ):
         # Instantiate a new FastAPI instance
+        root_path = self._normalize_root_path(settings.ROOT_PATH)
         self.app = FastAPI(
             title=settings.TITLE,
             description=settings.DESCRIPTION,
             version=settings.VERSION,
             contact={"name": settings.AUTHOR, "email": settings.EMAIL},
+            root_path=root_path,
             lifespan=self._lifespan,
         )
         self.templates = Jinja2Templates(
@@ -98,8 +100,14 @@ def __init__(
         # setup controls
         self._controls = {}
 
+        # local files (logos / custom js / css) served by url, keyed by url path
+        self._custom_asset_routes: Dict[str, str] = {}
+
+        # raw UI customization config (root-relative URLs); populated in add_static_files
+        self._ui_config_raw: Dict[str, Any] = {}
+
         # update ui in settings
-        self.settings = settings.model_copy()
+        self.settings = settings.model_copy(update={"ROOT_PATH": root_path})
         if ui:
             self.settings.UI = True
 
@@ -189,24 +197,131 @@ def add_docs(self) -> None:
 
         # Mount openapi
         @app_router.get("/openapi.json", include_in_schema=False)
-        def getOpenapi() -> Dict[str, Any]:
+        def getOpenapi(request: Request) -> Dict[str, Any]:
+            root_path = request.scope.get("root_path", "")
             return get_openapi(
                 title=self.settings.TITLE,
                 version=self.settings.VERSION,
                 routes=self.app.routes,
+                servers=[{"url": root_path}] if root_path else None,
             )
 
         @app_router.get("/docs/", include_in_schema=False, response_class=HTMLResponse)
-        def getDocs():
-            return get_swagger_ui_html(openapi_url="/openapi.json", title=self.settings.TITLE)
+        def getDocs(request: Request):
+            root_path = request.scope.get("root_path", "")
+            return get_swagger_ui_html(openapi_url=f"{root_path}/openapi.json", title=self.settings.TITLE)
 
         @app_router.get("/redoc/", include_in_schema=False, response_class=HTMLResponse)
-        def getRedoc():
-            return get_redoc_html(openapi_url="/openapi.json", title=self.settings.TITLE)
+        def getRedoc(request: Request):
+            root_path = request.scope.get("root_path", "")
+            return get_redoc_html(openapi_url=f"{root_path}/openapi.json", title=self.settings.TITLE)
+
+    @staticmethod
+    def _is_asset_url(value: str) -> bool:
+        """Whether an asset reference is already a servable URL (vs a local file path)."""
+        return value.startswith(("http://", "https://", "data:"))
+
+    @staticmethod
+    def _normalize_root_path(value: Optional[str]) -> str:
+        """Normalize a configured ROOT_PATH to '' or a leading-slash, no-trailing-slash path.
+
+        '' / '/' -> '', 'watchtower' -> '/watchtower', '/watchtower/' -> '/watchtower'.
+        """
+        if not value:
+            return ""
+        value = value.strip().rstrip("/")
+        if not value:
+            return ""
+        if not value.startswith("/"):
+            value = "/" + value
+        return value
+
+    @staticmethod
+    def _join_root_path(root_path: str, url: Optional[str]) -> Optional[str]:
+        """Prefix a root-relative URL with the proxy root_path, leaving absolute URLs alone."""
+        if not url or not root_path:
+            return url
+        if url.startswith(("http://", "https://", "data:")):
+            return url
+        if url.startswith("/"):
+            return f"{root_path.rstrip('/')}{url}"
+        return url
+
+    @staticmethod
+    def root_path_url(request: Optional[Request], url: str) -> str:
+        """Prefix a root-relative URL with the current request's proxy root_path.
+
+        Use for redirect targets and template links so auth and navigation flows
+        work when the app is served under a sub-path behind a reverse proxy.
+        """
+        root_path = request.scope.get("root_path", "") if request is not None else ""
+        return GatewayWebApp._join_root_path(root_path, url) or url
+
+    def _prefixed_ui_config(self, root_path: str) -> Dict[str, Any]:
+        """Return the UI config with all local asset URLs prefixed for the current root_path."""
+        raw = self._ui_config_raw
+        return {
+            **raw,
+            "basePath": root_path or "",
+            "headerLogo": self._join_root_path(root_path, raw["headerLogo"]),
+            "footerLogo": self._join_root_path(root_path, raw["footerLogo"]),
+            "customCss": [self._join_root_path(root_path, css) for css in raw["customCss"]],
+            "customJs": [self._join_root_path(root_path, js) for js in raw["customJs"]],
+        }
+
+    def _resolve_asset(self, value: Optional[str], kind: str) -> Optional[str]:
+        """Resolve an asset reference to a URL, serving local files automatically."""
+        if not value:
+            return None
+        # http(s) URLs and data URIs are used as-is. Anything that exists on disk
+        # is served automatically; everything else is treated as a URL path
+        # (e.g. an already-mounted "/static/..." or "/img/..." asset).
+        if self._is_asset_url(value) or not path.isfile(value):
+            return value
+        abspath = path.abspath(value)
+        url = f"/custom-assets/{kind}-{len(self._custom_asset_routes)}-{path.basename(abspath)}"
+        self._custom_asset_routes[url] = abspath
+        return url
+
+    def _resolve_ui_assets(self):
+        """Resolve all configured UI assets into URLs, mounting/serving local files."""
+        header_logo = self._resolve_asset(self.settings.HEADER_LOGO, "logo")
+        footer_logo = self._resolve_asset(self.settings.FOOTER_LOGO, "logo")
+        custom_css = [self._resolve_asset(css, "css") for css in self.settings.CUSTOM_CSS]
+        custom_js = [self._resolve_asset(js, "js") for js in self.settings.CUSTOM_JS]
+
+        # Auto-discover any *.js / *.css in the configured custom static directory
+        if self.settings.CUSTOM_STATIC_DIR:
+            custom_dir = path.abspath(self.settings.CUSTOM_STATIC_DIR)
+            if path.isdir(custom_dir):
+                self.app.mount(
+                    "/custom",
+                    CacheControlledStaticFiles(directory=custom_dir, check_dir=False),
+                    name="custom",
+                )
+                for fname in sorted(os.listdir(custom_dir)):
+                    if fname.endswith(".css"):
+                        custom_css.append(f"/custom/{fname}")
+                    elif fname.endswith(".js"):
+                        custom_js.append(f"/custom/{fname}")
+            else:
+                self.logger.warning("CUSTOM_STATIC_DIR %s is not a directory", custom_dir)
+
+        # Raw config with root-relative URLs; prefixed per-request via _prefixed_ui_config.
+        self._ui_config_raw = {
+            "title": self.settings.TITLE,
+            "description": "",
+            "headerLogo": header_logo,
+            "footerLogo": footer_logo,
+            "customCss": custom_css,
+            "customJs": custom_js,
+        }
+        return self._ui_config_raw
 
     def add_static_files(self) -> None:
         """Add static file handlers to FastAPI app"""
         app_router: APIRouter = self.get_router("app")
+        public_router: APIRouter = self.get_router("public")
 
         # Mount static files
         self.app.mount(
@@ -222,6 +337,25 @@ def add_static_files(self) -> None:
             name="img",
         )
 
+        # Resolve UI customization assets (logos, custom js/css), serving local files
+        self._resolve_ui_assets()
+
+        # Serve any local files referenced by the UI customization settings
+        if self._custom_asset_routes:
+
+            @app_router.get("/custom-assets/{name:path}", include_in_schema=False, response_class=FileResponse)
+            async def serve_custom_asset(name: str):
+                target = self._custom_asset_routes.get(f"/custom-assets/{name}")
+                if target is None:
+                    raise HTTPException(status_code=404, detail="Not found")
+                return FileResponse(target)
+
+        # Expose the UI customization config (title, logos, custom assets) for the frontend.
+        # Public (no auth) so the UI shell can render before authentication.
+        @public_router.get("/ui-config", include_in_schema=False)
+        async def get_ui_config(request: Request) -> Dict[str, Any]:
+            return self._prefixed_ui_config(request.scope.get("root_path", ""))
+
         # Mount top level routes
         @self.app.get("/favicon.ico", include_in_schema=False, response_class=FileResponse)
         async def readFavicon():
@@ -230,15 +364,29 @@ async def readFavicon():
         # Add UI if present, otherwise redirect to docs
         if self.settings.UI:
 
-            @app_router.get("/", include_in_schema=False, response_class=FileResponse)
-            async def serve_react_app():
-                return FileResponse(path.join(build_files_dir, "index.html"))
+            @app_router.get("/", include_in_schema=False, response_class=HTMLResponse)
+            async def serve_react_app(request: Request):
+                root_path = request.scope.get("root_path", "")
+                ui_config = self._prefixed_ui_config(root_path)
+                return self.templates.TemplateResponse(
+                    request,
+                    "index.html.j2",
+                    {
+                        "title": ui_config["title"],
+                        "description": ui_config["description"],
+                        "base_path": root_path,
+                        "ui_config": ui_config,
+                        "custom_css": ui_config["customCss"],
+                        "custom_js": ui_config["customJs"],
+                    },
+                )
 
         else:
 
             @self.app.get("/", include_in_schema=False, response_class=RedirectResponse)
-            async def serve_react_app():
-                return RedirectResponse("/redoc")
+            async def serve_react_app(request: Request):
+                root_path = request.scope.get("root_path", "")
+                return RedirectResponse(f"{root_path}/redoc")
 
     def add_api(self) -> None:
         """Add API handlers to FastAPI app"""