mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-06-16 01:35:36 -04:00
Fahim
facc50cb0f
* fix(api): attribute bearer-token actions to the token owner on owner-scoped routes Owner-scoped chat, session, and upload routes called get_current_user(), which resolves a bearer ody_ API token to the sandboxed "api" pseudo-user. A paired API-token client (companion, CLI, IDE extension) therefore saw and created a separate "api"-owned silo instead of the owner's data. effective_user() already exists for exactly this: it attributes a token's actions to request.state.api_token_owner, is identical to get_current_user() for cookie sessions, and falls back safely when a token has no owner. session_routes.py was already migrated; this completes the migration for the remaining owner-scoped routes: - chat_helpers.py: chat-privilege enforcement, message attribution, prefs/context - chat_routes.py: orphaned-endpoint owner, session-auth owner, message search - upload_routes.py: upload owner attribution + access checks The /api/models swap is intentionally omitted: #4292 already migrated it to effective_user (plus the chat-scope gate and ownerless-token 403), so this PR keeps dev's version of routes/model_routes.py unchanged. chat_routes.py keeps importing get_current_user for the workspace owner gate; session_routes.py drops the now-unused import. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * test: target effective_user in auth monkeypatches and owner-scope assertion The owner-scoped routes now call effective_user() instead of get_current_user(), so the tests that stubbed get_current_user (or asserted on it) follow suit: - test_chat_helpers.py, test_review_regressions.py, test_kv_cache_invalidation_2927.py: monkeypatch effective_user - test_session_endpoint_owner_scope.py: assert the owner-scope guard uses effective_user(request) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
# Import the route helper during collection so sibling session tests that use
|
|
# partial import stubs do not become the first loader of core.session_manager.
|
|
from routes.session_routes import _reject_raw_endpoint_url_for_non_admin
|
|
|
|
|
|
def _request(user, *, admin=False):
|
|
auth_manager = SimpleNamespace(is_admin=lambda username: bool(admin))
|
|
return SimpleNamespace(
|
|
state=SimpleNamespace(current_user=user),
|
|
app=SimpleNamespace(state=SimpleNamespace(auth_manager=auth_manager)),
|
|
)
|
|
|
|
|
|
def test_non_admin_session_create_rejects_raw_endpoint_url_without_endpoint_id():
|
|
with pytest.raises(HTTPException) as exc:
|
|
_reject_raw_endpoint_url_for_non_admin(
|
|
_request("alice", admin=False),
|
|
"alice",
|
|
"",
|
|
"http://169.254.169.254/latest/meta-data",
|
|
)
|
|
|
|
assert exc.value.status_code == 403
|
|
|
|
|
|
def test_admin_and_registered_endpoint_can_use_endpoint_url():
|
|
_reject_raw_endpoint_url_for_non_admin(
|
|
_request("alice", admin=False),
|
|
"alice",
|
|
"endpoint-id",
|
|
"http://127.0.0.1:8000/v1/chat/completions",
|
|
)
|
|
_reject_raw_endpoint_url_for_non_admin(
|
|
_request("admin", admin=True),
|
|
"admin",
|
|
"",
|
|
"http://127.0.0.1:8000/v1/chat/completions",
|
|
)
|
|
|
|
|
|
def test_chat_endpoint_recovery_paths_are_owner_scoped():
|
|
root = Path(__file__).resolve().parents[1]
|
|
chat_routes = (root / "routes" / "chat_routes.py").read_text(encoding="utf-8")
|
|
chat_helpers = (root / "routes" / "chat_helpers.py").read_text(encoding="utf-8")
|
|
|
|
assert "def _clear_orphaned_session_endpoint(sess, owner:" in chat_routes
|
|
assert "def _recover_empty_session_model(sess, session_id: str, owner:" in chat_routes
|
|
assert "q = owner_filter(q, ModelEndpoint, owner)" in chat_routes
|
|
assert "resolve_session_auth(sess, session, owner=effective_user(request))" in chat_routes
|
|
assert "def resolve_session_auth(sess, session_id: str, owner:" in chat_helpers
|
|
assert "update_q = update_q.filter(DBSession.owner == owner)" in chat_helpers
|