Hide upstream exception details from fallback context responses #112

Merged
rockachopa merged 1 commits from timmy/111-hide-context-exception-details into main 2026-08-06 14:18:13 +00:00
2 changed files with 38 additions and 8 deletions

View File

@ -18,6 +18,10 @@ EVENT_STREAM_TIMEOUT_SECONDS = 5.0
READINESS_TIMEOUT_SECONDS = 5.0
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
class ContextPayloadError(ValueError):
"""Raised when Gitea returns a structurally invalid context payload."""
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
@ -86,20 +90,25 @@ async def context() -> JSONResponse:
timeout=CONTEXT_TIMEOUT_SECONDS,
)
if not isinstance(user_data, dict):
raise ValueError("Gitea current-user response was not an object")
raise ContextPayloadError("Gitea current-user response was not an object")
if not all(field in user_data for field in ("id", "login")):
raise ValueError("Gitea current-user response did not include id and login")
raise ContextPayloadError(
"Gitea current-user response did not include id and login"
)
if any(
data is not None and not isinstance(data, list)
for data in (repo_data, issues_data, prs_data)
):
raise ValueError("Gitea collection response was not a list")
raise ContextPayloadError("Gitea collection response was not a list")
except Exception as e:
error_message = (
f"Gitea context request timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"
if isinstance(e, TimeoutError)
else str(e)
)
if isinstance(e, TimeoutError):
error_message = (
f"Gitea context request timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"
)
elif isinstance(e, ContextPayloadError):
error_message = str(e)
else:
error_message = "Gitea context is temporarily unavailable"
return JSONResponse({
"user": {"id": None, "login": "timmy", "full_name": "Timmy Jr", "email": ""},
"repos": [],

View File

@ -34,6 +34,27 @@ async def test_context_returns_fallback_when_gitea_exceeds_deadline(monkeypatch)
assert cancelled.is_set()
@pytest.mark.anyio
async def test_context_hides_upstream_exception_details(monkeypatch):
async def failing_user():
raise RuntimeError("connection refused by secret.internal.example")
async def empty_collection():
return []
monkeypatch.setattr(main, "current_user", failing_user)
monkeypatch.setattr(main, "repos", empty_collection)
monkeypatch.setattr(main, "issues", empty_collection)
monkeypatch.setattr(main, "pull_requests", empty_collection)
response = await main.context()
body = bytes(response.body)
context = json.loads(body)
assert context["error"] == "Gitea context is temporarily unavailable"
assert b"secret.internal.example" not in body
@pytest.mark.anyio
async def test_context_normalizes_nullable_issue_collections(monkeypatch):
async def user():