mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2026-10-16 19:11:57 +00:00
812aa3dda9
* Potential fix for code scanning alert no. 11: Information exposure through an exception traceback should not be in client side code Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> * fix: keep unexpected API error details private and test problem responses --------- Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>
96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
"""Problem responses keep server diagnostics private and preserve public errors."""
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.testclient import TestClient
|
|
from loguru import logger
|
|
|
|
from akkudoktoreos.server.rest.error import EOSProblem, register_problem_handlers
|
|
|
|
|
|
@pytest.mark.parametrize("api_level", ["TRACE", "DEBUG", "INFO", "WARNING", "ERROR"])
|
|
@pytest.mark.parametrize("unexpected", [False, True])
|
|
def test_problem_responses_keep_diagnostics_in_logs(config_eos, api_level, unexpected):
|
|
config_eos.logging.api_level = api_level
|
|
app = FastAPI()
|
|
register_problem_handlers(app)
|
|
private_detail = "Database failed at /private/server.db with password=secret"
|
|
|
|
@app.get("/failure")
|
|
def failure():
|
|
try:
|
|
raise RuntimeError(private_detail)
|
|
except RuntimeError as exc:
|
|
if unexpected:
|
|
raise
|
|
raise EOSProblem(
|
|
status=503,
|
|
title="Service Unavailable",
|
|
detail="The service is temporarily unavailable.",
|
|
type="/problems/service-unavailable",
|
|
cause=exc,
|
|
) from exc
|
|
|
|
logs: list[str] = []
|
|
sink_id = logger.add(logs.append, level="ERROR", backtrace=False, diagnose=False)
|
|
try:
|
|
with TestClient(app, raise_server_exceptions=False) as client:
|
|
response = client.get("/failure?token=private-query-token")
|
|
finally:
|
|
logger.remove(sink_id)
|
|
|
|
status = 500 if unexpected else 503
|
|
assert response.status_code == status
|
|
assert response.headers["content-type"] == "application/problem+json"
|
|
assert response.json() == {
|
|
"type": "/problems/internal-server-error" if unexpected else "/problems/service-unavailable",
|
|
"title": "Internal Server Error" if unexpected else "Service Unavailable",
|
|
"status": status,
|
|
"detail": (
|
|
"An unexpected error occurred."
|
|
if unexpected
|
|
else "The service is temporarily unavailable."
|
|
),
|
|
"instance": "/failure",
|
|
}
|
|
assert private_detail not in response.text
|
|
assert "private-query-token" not in response.text
|
|
assert any(private_detail in message and "Traceback" in message for message in logs)
|
|
|
|
|
|
def test_http_and_validation_errors_preserve_public_details():
|
|
app = FastAPI()
|
|
register_problem_handlers(app)
|
|
|
|
@app.get("/missing")
|
|
def missing():
|
|
raise HTTPException(status_code=404, detail="The requested item does not exist.")
|
|
|
|
@app.get("/validate")
|
|
def validate(value: int):
|
|
return {"value": value}
|
|
|
|
with TestClient(app) as client:
|
|
missing_response = client.get("/missing")
|
|
validation_response = client.get("/validate?value=private-invalid-input")
|
|
|
|
assert missing_response.status_code == 404
|
|
assert missing_response.headers["content-type"] == "application/problem+json"
|
|
assert missing_response.json() == {
|
|
"type": "about:blank",
|
|
"title": "HTTP Error",
|
|
"status": 404,
|
|
"detail": "The requested item does not exist.",
|
|
"instance": "/missing",
|
|
}
|
|
assert validation_response.status_code == 422
|
|
assert validation_response.headers["content-type"] == "application/problem+json"
|
|
assert validation_response.json() == {
|
|
"type": "/problems/validation",
|
|
"title": "Validation Error",
|
|
"status": 422,
|
|
"detail": "Request validation failed.",
|
|
"instance": "/validate",
|
|
}
|
|
assert "private-invalid-input" not in validation_response.text
|