diff --git a/solstone/convey/__init__.py b/solstone/convey/__init__.py index d5c0a5bec..fbf65ab82 100644 --- a/solstone/convey/__init__.py +++ b/solstone/convey/__init__.py @@ -5,6 +5,7 @@ from __future__ import annotations +import logging import os from datetime import timedelta from typing import TYPE_CHECKING @@ -12,9 +13,12 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: from flask import Flask +logger = logging.getLogger(__name__) + __all__ = [ "create_app", "emit", + "install_api_error_handlers", ] @@ -51,6 +55,41 @@ def install_identity_stamper(app: Flask) -> None: ) +def install_api_error_handlers(app: Flask) -> None: + """Guarantee JSON error envelopes for every API path.""" + from flask import g, request + from werkzeug.exceptions import HTTPException, InternalServerError + + from solstone.think.utils import CorruptConfigError + + from .reasons import CORRUPT_CONFIG, HTTP_ERROR, INTERNAL_ERROR + from .utils import error_response + + def _is_api_request() -> bool: + return "api" in request.path.strip("/").split("/") + + @app.errorhandler(CorruptConfigError) + def _handle_corrupt_config(exc: CorruptConfigError): + return error_response(CORRUPT_CONFIG, detail=str(exc)) + + @app.errorhandler(HTTPException) + def _handle_http_exception(exc: HTTPException): + if not _is_api_request(): + return exc + + if isinstance(exc, InternalServerError) and exc.original_exception is not None: + original = exc.original_exception + logger.error( + "unhandled API exception request_id=%s path=%s", + getattr(g, "request_id", ""), + request.path, + exc_info=(type(original), original, original.__traceback__), + ) + return error_response(INTERNAL_ERROR) + + return error_response(HTTP_ERROR, status=exc.code or HTTP_ERROR.status) + + def create_app(journal: str = "") -> Flask: """Create and configure the Convey Flask application.""" from flask import Flask @@ -82,14 +121,7 @@ def create_app(journal: str = "") -> Flask: static_folder=os.path.join(os.path.dirname(__file__), "static"), ) - from solstone.think.utils import CorruptConfigError - - from .reasons import CORRUPT_CONFIG - from .utils import error_response - - @app.errorhandler(CorruptConfigError) - def _handle_corrupt_config(exc: CorruptConfigError): - return error_response(CORRUPT_CONFIG, detail=str(exc)) + install_api_error_handlers(app) # Add apps directory to template search path so apps can have their templates # in apps/{name}/workspace.html instead of needing a templates/ subfolder diff --git a/solstone/convey/reasons.py b/solstone/convey/reasons.py index 6bbbfdb1e..2f7d6210a 100644 --- a/solstone/convey/reasons.py +++ b/solstone/convey/reasons.py @@ -11,6 +11,18 @@ class Reason: status: int = 400 +# request boundary +HTTP_ERROR = Reason( + "http_error", + "I couldn't complete that request.", + 400, +) +INTERNAL_ERROR = Reason( + "internal_error", + "I couldn't complete that request.", + 500, +) + # auth AUTH_REQUIRED = Reason("auth_required", "I couldn't verify this request.", 401) AUTH_KEY_INVALID = Reason("auth_key_invalid", "I couldn't verify that key.", 401) diff --git a/tests/test_convey_api_error_boundary.py b/tests/test_convey_api_error_boundary.py new file mode 100644 index 000000000..e3c01a3c0 --- /dev/null +++ b/tests/test_convey_api_error_boundary.py @@ -0,0 +1,80 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import logging + +import pytest +from flask import Flask + +from solstone.convey import install_api_error_handlers +from solstone.convey.request_id import install_request_id_stamper + + +@pytest.fixture +def app() -> Flask: + application = Flask(__name__) + application.config["TESTING"] = False + install_request_id_stamper(application) + install_api_error_handlers(application) + + @application.get("/api/test/boom") + def api_boom() -> None: + raise RuntimeError("secret implementation detail") + + @application.get("/test/boom") + def html_boom() -> None: + raise RuntimeError("html implementation detail") + + return application + + +def test_unexpected_api_exception_returns_safe_json_and_logs_request_id( + app: Flask, + caplog: pytest.LogCaptureFixture, +) -> None: + with caplog.at_level(logging.ERROR, logger="solstone.convey"): + response = app.test_client().get("/api/test/boom") + + request_id = response.headers["X-Solstone-Request-Id"] + assert request_id + assert response.status_code == 500 + assert response.is_json + assert response.get_json() == { + "detail": "", + "error": "I couldn't complete that request.", + "reason_code": "internal_error", + } + assert "secret implementation detail" not in response.get_data(as_text=True) + assert any( + f"request_id={request_id}" in record.getMessage() for record in caplog.records + ) + + +def test_http_exception_on_api_path_preserves_status_in_json(app: Flask) -> None: + response = app.test_client().get("/api/not-found") + + assert response.status_code == 404 + assert response.is_json + assert response.get_json() == { + "detail": "", + "error": "I couldn't complete that request.", + "reason_code": "http_error", + } + + +def test_non_api_http_exception_keeps_default_html(app: Flask) -> None: + response = app.test_client().get("/not-found") + + assert response.status_code == 404 + assert response.content_type.startswith("text/html") + assert not response.is_json + + +def test_non_api_unexpected_exception_keeps_default_html(app: Flask) -> None: + response = app.test_client().get("/test/boom") + + assert response.status_code == 500 + assert response.content_type.startswith("text/html") + assert not response.is_json