From 61da9e405a0a156dce96d316ece4212289fe4815 Mon Sep 17 00:00:00 2001 From: Roberto Prevato Date: Wed, 25 Feb 2026 07:34:28 +0100 Subject: [PATCH 1/2] Fix #668 --- CHANGELOG.md | 7 ++----- blacksheep/server/application.py | 23 ++++++++++++++++------- 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b211be55..da706e5d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,12 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [2.6.2] - 2026-??-?? +- Fix regression that broke compatibility with `Starlette` mounts + [#668](https://github.com/Neoteroi/BlackSheep/issues/668). - Fix [#561](https://github.com/Neoteroi/BlackSheep/issues/561). -- Change `FromBodyBinder.binder_types` from a constructor-local variable to a - class attribute, defaulting to `[JSONBinder]` only. Users can configure - additional formats by setting `FromBodyBinder.binder_types` (e.g. - `FromBodyBinder.binder_types = [JSONBinder, FormBinder]`) before the - application starts. - Add support for baking OpenAPI Specification files to disk, to support running with `PYTHONOPTIMIZE=2` (or `-OO`) where docstrings are stripped and cannot be used to enrich OpenAPI Documentation automatically. diff --git a/blacksheep/server/application.py b/blacksheep/server/application.py index 6797ceb6..70d270b3 100644 --- a/blacksheep/server/application.py +++ b/blacksheep/server/application.py @@ -806,8 +806,13 @@ async def _handle_websocket(self, scope, receive, send) -> None: ws = WebSocket(scope, receive, send) # TODO: support filters + root_path = scope.get("root_path", "") + path = scope["path"] + if root_path and path.startswith(root_path): + path = path[len(root_path):] or "/" + route = self.router.get_match_by_method_and_path( - RouteMethod.GET_WS, scope["path"] + RouteMethod.GET_WS, path ) if route is None: @@ -847,6 +852,10 @@ def instantiate_request(self, scope, receive) -> Request: raw_path = scope["path"].encode("utf-8") scope["raw_path"] = raw_path + 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"/" + request = Request.incoming( scope["method"], raw_path, @@ -918,16 +927,16 @@ def handle_mount_path(self, scope, route_match): assert tail is not None tail = "/" + tail - # Update root_path per the ASGI spec: the child app must know its mount - # prefix so it can generate correct absolute URLs (analogous to WSGI - # SCRIPT_NAME). root_path = parent root_path + the stripped mount prefix. + # Compute the mount prefix (the part of path before the tail). mount_prefix = ( scope["path"][: -len(tail)] if tail != "/" else scope["path"].rstrip("/") ) - scope["root_path"] = scope.get("root_path", "") + mount_prefix - scope["path"] = tail - scope["raw_path"] = tail.encode("utf8") + # Set root_path per the ASGI spec so ASGI-compliant child apps + # can derive the application-relative path by stripping root_path from path. + # Do NOT modify scope["path"] or scope["raw_path"] — the child app handles that + # itself. + scope["root_path"] = scope.get("root_path", "") + mount_prefix def _is_browser_navigation(self, scope) -> bool: """ From 0c639efdfef033089674edfc6f5d92d551e0e3d4 Mon Sep 17 00:00:00 2001 From: Roberto Prevato Date: Wed, 25 Feb 2026 15:18:21 +0100 Subject: [PATCH 2/2] Add integration tests Add integration tests for Starlette and Piccolo-Admin --- .github/workflows/main.yml | 8 + CHANGELOG.md | 10 +- requirements.pypy.txt | 1 + requirements.txt | 1 + tests/test_piccolo_admin_compat.py | 247 +++++++++++++++++++++++++++++ 5 files changed, 263 insertions(+), 4 deletions(-) create mode 100644 tests/test_piccolo_admin_compat.py diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index f1145017..53dbdf19 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -148,6 +148,14 @@ jobs: APP_DEFAULT_ROUTER=false ASGI_SERVER=hypercorn pytest itests/test_server.py fi + - name: Test Piccolo Admin / Starlette mount compatibility + if: matrix.os == 'ubuntu-latest' && matrix.python-version == 3.14 + run: | + echo "Running Starlette mount compatibility tests (ref: piccolo-orm/piccolo_admin#472)..." + pip install piccolo-admin # include the dedicated test + + pytest tests/test_piccolo_admin_compat.py -v + - name: Install distribution dependencies run: pip install --upgrade twine setuptools wheel if: matrix.os == 'ubuntu-latest' && matrix.python-version == 3.12 diff --git a/CHANGELOG.md b/CHANGELOG.md index da706e5d..33e889d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fix regression that broke compatibility with `Starlette` mounts [#668](https://github.com/Neoteroi/BlackSheep/issues/668). -- Fix [#561](https://github.com/Neoteroi/BlackSheep/issues/561). + Add integration tests to verify support for `Starlette` and `Piccolo-Admin`. + Reported by @snow-born and @sinisaos. +- Fix [#561](https://github.com/Neoteroi/BlackSheep/issues/561): fix support for `PYTHONOPTIMIZE=2`. - Add support for baking OpenAPI Specification files to disk, to support running with `PYTHONOPTIMIZE=2` (or `-OO`) where docstrings are stripped and cannot be used to enrich OpenAPI Documentation automatically. @@ -25,7 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Issue a `UserWarning` when `PYTHONOPTIMIZE >= 2` and a request handler has no docstring, advising the user to bake the spec file. - **Typical workflow:** + **Proposed workflow:** ```python # 1. bake_spec.py — run once in CI, without -OO @@ -71,8 +73,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 app, OpenIDSettings( client_id="...", - authority="https://login.microsoftonline.com/", client_secret=Secret("..."), # confidential client + authority="https://", use_pkce=True, # adds code_challenge on top of the secret ), ) @@ -102,7 +104,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 async def create_item(data: FromBody[Item]) -> Item: ... ``` - - New **`FromXML[T]`** and **`XMLBinder`**: parse `application/xml` / `text/xml` + - Add new **`FromXML[T]`** and **`XMLBinder`**: parse `application/xml` / `text/xml` request bodies using [`defusedxml`](https://github.com/tiran/defusedxml), protecting against **XXE injection**, **entity expansion (billion laughs)**, and **DTD-based attacks**. Security exceptions propagate unmodified so the diff --git a/requirements.pypy.txt b/requirements.pypy.txt index 1331304d..d62fe927 100644 --- a/requirements.pypy.txt +++ b/requirements.pypy.txt @@ -20,3 +20,4 @@ uvicorn==0.34.2 pydantic==2.12.3 pydantic_core==2.41.4 defusedxml>=0.7.1 +starlette>=0.37.0 diff --git a/requirements.txt b/requirements.txt index e3927cbe..1cba17f9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -25,3 +25,4 @@ uvicorn==0.34.2 pydantic==2.12.3 pydantic_core==2.41.4 defusedxml>=0.7.1 +starlette>=0.37.0 diff --git a/tests/test_piccolo_admin_compat.py b/tests/test_piccolo_admin_compat.py new file mode 100644 index 00000000..29028edc --- /dev/null +++ b/tests/test_piccolo_admin_compat.py @@ -0,0 +1,247 @@ +""" +Tests to verify that BlackSheep correctly mounts ASGI applications that follow +the ASGI spec for root_path / path handling — most notably Starlette-based apps +like Piccolo Admin. + +Reference: https://github.com/piccolo-orm/piccolo_admin/issues/472 + https://github.com/Neoteroi/BlackSheep/issues/668 + +The ASGI spec says: + - root_path: The root path this application is mounted at (same as SCRIPT_NAME + in CGI). The parent (or reverse-proxy) sets this before calling the child. + - path: The HTTP request target, excluding query string. It MUST NOT be + modified by the parent when forwarding to a mounted child. + - raw_path: The original byte-string of the path. Same rule — left intact. + +The child app is responsible for deriving its own application-relative path by +stripping root_path from path. +""" +import re +import pytest + +from blacksheep.testing.helpers import get_example_scope +from blacksheep.testing.messages import MockReceive, MockSend +from tests.utils.application import FakeApplication + +from starlette.applications import Starlette +from starlette.responses import HTMLResponse, PlainTextResponse +from starlette.routing import Mount, Route +from starlette.staticfiles import StaticFiles + +# --------------------------------------------------------------------------- +# Tests: real Starlette (required dependency) +# +# Piccolo Admin is a Starlette/FastAPI app that internally mounts sub-apps: +# / → serves index.html (login page) +# /assets/* → StaticFiles serving index-XXX.js, index-XXX.css, etc. +# /api/* → REST endpoints (auth-protected) +# /public/* → public endpoints (login, translations, meta) +# +# The bug in issue #472 was: the login page HTML loaded fine, but the +# ' + '' + ) + + async def login_endpoint(request): + return PlainTextResponse("login page") + + async def public_meta(request): + return PlainTextResponse("meta ok") + + app = Starlette( + routes=[ + Route("/", admin_root), + Route("/login/", login_endpoint), + Mount("/assets", app=StaticFiles(directory=str(assets_dir))), + Mount( + "/public", + routes=[Route("/meta/", public_meta)], + ), + ], + ) + return app + + +@pytest.mark.asyncio +async def test_real_starlette_app_mounted_root(starlette_admin_like_app): + """GET /admin/ → Starlette root route returns 200.""" + parent = FakeApplication() + parent.mount("/admin", starlette_admin_like_app) + await parent.start() + + scope = get_example_scope("GET", "/admin/") + mock_send = MockSend() + await parent(scope, MockReceive(), mock_send) + + assert mock_send.messages[0]["status"] == 200 + body = mock_send.messages[1]["body"] + assert b"index-abc123.js" in body + + +@pytest.mark.asyncio +async def test_real_starlette_app_mounted_static_js(starlette_admin_like_app): + """ + GET /admin/assets/index-abc123.js → must return 200. + This is the exact failure mode from piccolo_admin#472: the login page + HTML loads, but browser requests for JS/CSS assets get 404. + """ + parent = FakeApplication() + parent.mount("/admin", starlette_admin_like_app) + await parent.start() + + scope = get_example_scope("GET", "/admin/assets/index-abc123.js") + mock_send = MockSend() + await parent(scope, MockReceive(), mock_send) + + start = mock_send.messages[0] + assert start["status"] == 200, ( + f"Expected 200 for /admin/assets/index-abc123.js, got {start['status']}. " + "This indicates a regression in ASGI scope handling for nested mounts " + "(ref: piccolo-orm/piccolo_admin#472)." + ) + + +@pytest.mark.asyncio +async def test_real_starlette_app_mounted_static_css(starlette_admin_like_app): + """GET /admin/assets/index-abc123.css → must return 200.""" + parent = FakeApplication() + parent.mount("/admin", starlette_admin_like_app) + await parent.start() + + scope = get_example_scope("GET", "/admin/assets/index-abc123.css") + mock_send = MockSend() + await parent(scope, MockReceive(), mock_send) + + assert mock_send.messages[0]["status"] == 200 + + +@pytest.mark.asyncio +async def test_real_starlette_app_mounted_sub_route(starlette_admin_like_app): + """GET /admin/login/ → Starlette sub-route returns 200.""" + parent = FakeApplication() + parent.mount("/admin", starlette_admin_like_app) + await parent.start() + + scope = get_example_scope("GET", "/admin/login/") + mock_send = MockSend() + await parent(scope, MockReceive(), mock_send) + + assert mock_send.messages[0]["status"] == 200 + assert mock_send.messages[1]["body"] == b"login page" + + +@pytest.mark.asyncio +async def test_real_starlette_app_mounted_nested_mount(starlette_admin_like_app): + """GET /admin/public/meta/ → nested Starlette Mount returns 200.""" + parent = FakeApplication() + parent.mount("/admin", starlette_admin_like_app) + await parent.start() + + scope = get_example_scope("GET", "/admin/public/meta/") + mock_send = MockSend() + await parent(scope, MockReceive(), mock_send) + + assert mock_send.messages[0]["status"] == 200 + assert mock_send.messages[1]["body"] == b"meta ok" + + +@pytest.mark.asyncio +async def test_real_starlette_app_mounted_nonexistent_asset( + starlette_admin_like_app, +): + """GET /admin/assets/does-not-exist.js → 404 from StaticFiles (not crash).""" + parent = FakeApplication() + parent.mount("/admin", starlette_admin_like_app) + await parent.start() + + scope = get_example_scope("GET", "/admin/assets/does-not-exist.js") + mock_send = MockSend() + await parent(scope, MockReceive(), mock_send) + + assert mock_send.messages[0]["status"] in (404, 405) + + +# --------------------------------------------------------------------------- +# Tests: optional Piccolo Admin (if installed) +# --------------------------------------------------------------------------- + +try: + from piccolo_admin.endpoints import create_admin + + _has_piccolo_admin = True +except ImportError: + _has_piccolo_admin = False + + +@pytest.mark.asyncio +@pytest.mark.skipif(not _has_piccolo_admin, reason="piccolo_admin not installed") +async def test_piccolo_admin_mounted_serves_login_page(): + """ + Mount the real Piccolo Admin ASGI app and verify that the login page + is served (returns 200) rather than a 404 or crash. + + This is the exact scenario reported in: + https://github.com/piccolo-orm/piccolo_admin/issues/472 + """ + admin_app = create_admin(tables=[]) + + parent = FakeApplication() + parent.mount("/admin", admin_app) + await parent.start() + + scope = get_example_scope("GET", "/admin/") + mock_send = MockSend() + await parent(scope, MockReceive(), mock_send) + + start = mock_send.messages[0] + assert start["status"] == 200, ( + f"Expected 200 from Piccolo Admin at /admin/, got {start['status']}. " + "This may indicate a regression in ASGI scope handling for mounted apps." + ) + + body = b"" + for msg in mock_send.messages: + if msg.get("type") == "http.response.body": + body += msg.get("body", b"") + html_text = body.decode("utf-8", errors="replace") + + asset_paths = re.findall(r'(?:src|href)="\.(/assets/[^"]+)"', html_text) + assert asset_paths, ( + "Could not find any asset references in Piccolo Admin HTML. " + "The HTML template may have changed." + ) + for asset_path in asset_paths: + full_path = f"/admin{asset_path}" + asset_scope = get_example_scope("GET", full_path) + asset_send = MockSend() + await parent(asset_scope, MockReceive(), asset_send) + asset_start = asset_send.messages[0] + assert asset_start["status"] == 200, ( + f"Expected 200 for {full_path}, got {asset_start['status']}. " + "Static assets under mounted Piccolo Admin are broken." + )