diff --git a/src/chatgpt_subscription.py b/src/chatgpt_subscription.py index d8e7f6201..d7e7afdae 100644 --- a/src/chatgpt_subscription.py +++ b/src/chatgpt_subscription.py @@ -258,6 +258,16 @@ def resolve_runtime_credentials( force_refresh: bool = False, allow_live_probes: bool = True, ) -> Dict[str, Any]: + normalized_owner = None + if not allow_live_probes: + from src.owner_identity import is_request_sentinel_owner, normalize_owner + + normalized_owner = normalize_owner(owner) + if normalized_owner is None or is_request_sentinel_owner(normalized_owner): + raise ChatGPTSubscriptionAuthNotFound( + "ChatGPT Subscription credentials require an authenticated owner." + ) + ProviderAuthSession, SessionLocal, utcnow_naive = _database_handles() db = SessionLocal() try: @@ -265,7 +275,9 @@ def resolve_runtime_credentials( ProviderAuthSession.id == auth_id, ProviderAuthSession.provider == CHATGPT_SUBSCRIPTION_PROVIDER, ) - if owner: + if not allow_live_probes: + q = q.filter(ProviderAuthSession.owner == normalized_owner) + elif owner: q = q.filter(ProviderAuthSession.owner == owner) row = q.first() if row is None: diff --git a/tests/test_api_token_repair_cycle8.py b/tests/test_api_token_repair_cycle8.py index 9801fab5c..194d2e9ef 100644 --- a/tests/test_api_token_repair_cycle8.py +++ b/tests/test_api_token_repair_cycle8.py @@ -1,6 +1,8 @@ -"""Cycle-8 regressions for bearer provider-auth session repair.""" +"""Cycle-8/9 regressions for bearer provider-auth session repair.""" +import base64 import json +import time from types import SimpleNamespace import pytest @@ -14,7 +16,16 @@ import core.database as cdb _CODEX_BASE = "https://chatgpt.com/backend-api/codex" -def _provider_db(monkeypatch): +def _future_access_token(): + def encode(payload): + return base64.urlsafe_b64encode( + json.dumps(payload, separators=(",", ":")).encode() + ).rstrip(b"=").decode() + + return f"{encode({'alg': 'none'})}.{encode({'exp': int(time.time()) + 3600})}.signature" + + +def _provider_db(monkeypatch, *, access_token="cached-access-token"): from routes import chat_helpers from src import chatgpt_subscription @@ -34,7 +45,7 @@ def _provider_db(monkeypatch): provider="chatgpt-subscription", owner="alice", base_url=_CODEX_BASE, - access_token="cached-access-token", + access_token=access_token, refresh_token="refresh-token", auth_mode="chatgpt", )) @@ -92,3 +103,58 @@ def test_bearer_validator_rejects_provider_auth_when_cache_is_unusable(monkeypat with pytest.raises(HTTPException) as exc: _validate_bearer_session_model(_registered_session(), owner="alice") assert exc.value.status_code == 401 + + +@pytest.mark.parametrize( + "owner", + [None, "", " ", "api", "demo", "system", "internal-tool", " API "], +) +def test_cache_only_provider_auth_rejects_missing_and_sentinel_owners_before_query(monkeypatch, owner): + from src import chatgpt_subscription + + def forbidden_database_handles(): + pytest.fail("cache-only provider auth queried without a real owner") + + monkeypatch.setattr(chatgpt_subscription, "_database_handles", forbidden_database_handles) + monkeypatch.setattr( + chatgpt_subscription, + "refresh_oauth_tokens", + lambda *args, **kwargs: pytest.fail("cache-only provider auth refreshed credentials"), + ) + + with pytest.raises(chatgpt_subscription.ChatGPTSubscriptionAuthNotFound): + chatgpt_subscription.resolve_runtime_credentials( + "auth-1", owner=owner, allow_live_probes=False + ) + + +def test_cache_only_provider_auth_normalizes_exact_owner_and_blocks_live_io(monkeypatch): + from src import chatgpt_subscription + + access_token = _future_access_token() + _provider_db(monkeypatch, access_token=access_token) + monkeypatch.setattr( + chatgpt_subscription, + "refresh_oauth_tokens", + lambda *args, **kwargs: pytest.fail("cache-only provider auth refreshed credentials"), + ) + monkeypatch.setattr( + chatgpt_subscription.httpx, + "get", + lambda *args, **kwargs: pytest.fail("cache-only provider auth probed the provider"), + ) + monkeypatch.setattr( + chatgpt_subscription.httpx, + "post", + lambda *args, **kwargs: pytest.fail("cache-only provider auth probed the provider"), + ) + + result = chatgpt_subscription.resolve_runtime_credentials( + "auth-1", owner=" alice ", allow_live_probes=False + ) + assert result["api_key"] == access_token + + with pytest.raises(chatgpt_subscription.ChatGPTSubscriptionAuthNotFound): + chatgpt_subscription.resolve_runtime_credentials( + "auth-1", owner="bob", allow_live_probes=False + )