|
23 | 23 | from fastapi.testclient import TestClient |
24 | 24 |
|
25 | 25 | from ...core.error_utils import format_error_summary, format_error_traceback |
26 | | -from ..restful_api import DetailedHTTPException |
| 26 | +from ..restful_api import DetailedHTTPException, RESTfulAPI |
27 | 27 |
|
28 | 28 |
|
29 | 29 | def _build_app() -> FastAPI: |
@@ -113,3 +113,105 @@ def test_exception_subclasses_http_exception(): |
113 | 113 | assert exc.status_code == 503 |
114 | 114 | assert exc.detail == "boom" |
115 | 115 | assert exc.tb is None |
| 116 | + |
| 117 | + |
| 118 | +class _FakeDB: |
| 119 | + def __init__(self, permissions): |
| 120 | + self._permissions = permissions |
| 121 | + |
| 122 | + def get_user_by_id(self, user_id): |
| 123 | + if user_id is None: |
| 124 | + return None |
| 125 | + return {"id": user_id, "permissions": list(self._permissions)} |
| 126 | + |
| 127 | + |
| 128 | +class _FakeAuthService: |
| 129 | + """Minimal stand-in for AdvancedAuthService's diagnostics-relevant surface.""" |
| 130 | + |
| 131 | + def __init__(self, permissions, valid_jwt=True): |
| 132 | + self.db = _FakeDB(permissions) |
| 133 | + self._valid_jwt = valid_jwt |
| 134 | + |
| 135 | + def verify_access_token(self, token): |
| 136 | + # A None payload is how an API key (rather than a JWT) presents here. |
| 137 | + return {"user_id": 1} if self._valid_jwt else None |
| 138 | + |
| 139 | + |
| 140 | +def _api_with_auth(auth_service): |
| 141 | + api = RESTfulAPI.__new__(RESTfulAPI) |
| 142 | + api._advanced_auth_service = auth_service |
| 143 | + return api |
| 144 | + |
| 145 | + |
| 146 | +def _request_with_token(token="tok"): |
| 147 | + headers = [(b"authorization", f"Bearer {token}".encode())] if token else [] |
| 148 | + return Request({"type": "http", "headers": headers, "method": "POST", "path": "/"}) |
| 149 | + |
| 150 | + |
| 151 | +class TestDiagnosticsPermission: |
| 152 | + """Launching needs models:write; a traceback needs logs:list.""" |
| 153 | + |
| 154 | + def test_no_auth_configured_allows_diagnostics(self): |
| 155 | + api = _api_with_auth(None) |
| 156 | + assert api._caller_may_see_diagnostics(_request_with_token()) is True |
| 157 | + |
| 158 | + def test_logs_list_allows_diagnostics(self): |
| 159 | + api = _api_with_auth(_FakeAuthService(["models:write", "logs:list"])) |
| 160 | + assert api._caller_may_see_diagnostics(_request_with_token()) is True |
| 161 | + |
| 162 | + def test_admin_allows_diagnostics(self): |
| 163 | + api = _api_with_auth(_FakeAuthService(["admin"])) |
| 164 | + assert api._caller_may_see_diagnostics(_request_with_token()) is True |
| 165 | + |
| 166 | + def test_models_write_alone_is_denied(self): |
| 167 | + # The whole point of the gate: a model operator without log access |
| 168 | + # must not receive filesystem paths and runtime internals. |
| 169 | + api = _api_with_auth(_FakeAuthService(["models:write"])) |
| 170 | + assert api._caller_may_see_diagnostics(_request_with_token()) is False |
| 171 | + |
| 172 | + def test_legacy_scope_alias_is_honoured(self): |
| 173 | + # models:start normalizes to models:write -- still not logs:list. |
| 174 | + api = _api_with_auth(_FakeAuthService(["models:start"])) |
| 175 | + assert api._caller_may_see_diagnostics(_request_with_token()) is False |
| 176 | + |
| 177 | + def test_missing_token_is_denied(self): |
| 178 | + api = _api_with_auth(_FakeAuthService(["logs:list"])) |
| 179 | + assert api._caller_may_see_diagnostics(_request_with_token(None)) is False |
| 180 | + |
| 181 | + def test_api_key_is_denied(self): |
| 182 | + # API keys cannot hold logs:list, so they never qualify. |
| 183 | + api = _api_with_auth(_FakeAuthService(["logs:list"], valid_jwt=False)) |
| 184 | + assert api._caller_may_see_diagnostics(_request_with_token()) is False |
| 185 | + |
| 186 | + def test_unknown_user_is_denied(self): |
| 187 | + service = _FakeAuthService(["logs:list"]) |
| 188 | + service.db.get_user_by_id = lambda _uid: None |
| 189 | + api = _api_with_auth(service) |
| 190 | + assert api._caller_may_see_diagnostics(_request_with_token()) is False |
| 191 | + |
| 192 | + def test_auth_failure_fails_closed(self): |
| 193 | + service = _FakeAuthService(["logs:list"]) |
| 194 | + |
| 195 | + def _boom(_token): |
| 196 | + raise RuntimeError("auth backend down") |
| 197 | + |
| 198 | + service.verify_access_token = _boom |
| 199 | + api = _api_with_auth(service) |
| 200 | + # Must not propagate out of the error path, and must deny. |
| 201 | + assert api._caller_may_see_diagnostics(_request_with_token()) is False |
| 202 | + |
| 203 | + |
| 204 | +class TestLaunchErrorTraceback: |
| 205 | + def test_traceback_withheld_without_permission(self): |
| 206 | + api = _api_with_auth(_FakeAuthService(["models:write"])) |
| 207 | + assert ( |
| 208 | + api._launch_error_traceback(_request_with_token(), ValueError("x")) is None |
| 209 | + ) |
| 210 | + |
| 211 | + def test_traceback_returned_with_permission(self): |
| 212 | + api = _api_with_auth(_FakeAuthService(["logs:list"])) |
| 213 | + try: |
| 214 | + raise ValueError("boom") |
| 215 | + except ValueError as e: |
| 216 | + tb = api._launch_error_traceback(_request_with_token(), e) |
| 217 | + assert tb is not None and "ValueError: boom" in tb |
0 commit comments