fix(github-copilot): send the GitHub token directly instead of exchanging it (#50126)

* fix(github-copilot): use GitHub token directly

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(github-copilot): expire API base cache

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(github-copilot): restore authenticator coverage

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(github-copilot): update direct token auth tests

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(github-copilot): cover direct-token error paths

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(github-copilot): satisfy type-discipline gate

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Mateo Wang 2026-11-05 09:45:17 -08:00 • committed by GitHub
parent 78f1b09429
commit 88ea40e69c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 763 additions and 1013 deletions

View file

@ -28,7 +28,7 @@ GITHUB_COPILOT_PER_USER_AUTH_TYPE: Final = "per_user_oauth"
GITHUB_COPILOT_AUTH_TYPE_KEY: Final = "github_copilot_auth_type"
GITHUB_COPILOT_HOST_KEY: Final = "github_copilot_host"
GITHUB_COPILOT_DEFAULT_HOST: Final = "github.com"
GITHUB_COPILOT_USER_TOKEN_SAFETY_MARGIN_SECONDS: Final = 60
GITHUB_COPILOT_USER_SESSION_TTL_SECONDS: Final = 1800
GITHUB_COPILOT_USER_CREDENTIAL_CACHE_TTL_SECONDS: Final = 60
GITHUB_COPILOT_DEVICE_FLOW_CACHE_PREFIX: Final = "github_copilot_device_flow"
USER_PROVIDER_CREDENTIAL_CACHE_PREFIX: Final = "user_provider_credential"

View file

@ -1,62 +1,30 @@
import hashlib
import json
import os
import time
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Final, TypeVar
from collections.abc import Mapping
from typing import Final, Protocol, cast # noqa: TID251 # pins untyped clients and caches to local Protocols
import httpx
from pydantic import BaseModel, ConfigDict, TypeAdapter, with_config
from pydantic import ConfigDict, TypeAdapter, with_config
from typing_extensions import NotRequired, ReadOnly, TypedDict
from litellm._logging import verbose_logger
from litellm.caching.in_memory_cache import InMemoryCache
from litellm.constants import GITHUB_COPILOT_USER_SESSION_TTL_SECONDS
from litellm.litellm_core_utils.asyncify import can_block_current_thread
from litellm.llms.custom_httpx.http_handler import get_httpx_client
from .common_utils import (
GetAccessTokenError,
GetAPIKeyError,
GetDeviceCodeError,
RefreshAPIKeyError,
)
# Constants (default values — overridable via environment variables at call time)
DEFAULT_GITHUB_CLIENT_ID: Final = "Iv1.b507a08c87ecfe98"
DEFAULT_GITHUB_DEVICE_CODE_URL: Final = "https://github.com/login/device/code"
DEFAULT_GITHUB_ACCESS_TOKEN_URL: Final = "https://github.com/login/oauth/access_token"
DEFAULT_GITHUB_API_KEY_URL: Final = "https://api.github.com/copilot_internal/v2/token"
class CopilotEndpoints(BaseModel):
model_config = ConfigDict(frozen=True, strict=True, extra="ignore", hide_input_in_errors=True)
api: str | None = None
class CopilotCachedKey(BaseModel):
model_config = ConfigDict(frozen=True, strict=True, extra="ignore", hide_input_in_errors=True)
token: str
expires_at: float
endpoints: CopilotEndpoints | None = None
class CopilotCachedEndpoints(BaseModel):
model_config = ConfigDict(frozen=True, strict=True, extra="ignore", hide_input_in_errors=True)
endpoints: CopilotEndpoints | None = None
ApiKeyFileView = TypeVar("ApiKeyFileView", CopilotCachedKey, CopilotCachedEndpoints)
@dataclass(frozen=True, slots=True)
class _UnsavedApiKey:
access_token: str
key: CopilotCachedKey
_UNSAVED_API_KEYS: Final[dict[str, _UnsavedApiKey]] = {} # mutable-ok: refreshed keys api-key.json could not hold
DEFAULT_GITHUB_COPILOT_USER_URL: Final = "https://api.github.com/copilot_internal/user"
def _read_token_file(path: str) -> str:
@ -78,11 +46,66 @@ class _AccessTokenPoll(TypedDict):
error: ReadOnly[NotRequired[object]]
_JSON_OBJECT: Final = TypeAdapter(dict[str, object], config=ConfigDict(strict=True))
_JSON_OBJECT: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(
Mapping[str, object], config=ConfigDict(strict=True)
)
_DEVICE_CODE: Final = TypeAdapter(_DeviceCode)
_ACCESS_TOKEN_POLL: Final = TypeAdapter(_AccessTokenPoll)
class _CopilotAPIBaseCache(Protocol):
def get_cache(self, key: str) -> object: ...
def set_cache(self, key: str, value: str, ttl: float | None = None) -> object: ...
class _SyncGetClient(Protocol):
def get(self, url: str, *, headers: Mapping[str, str]) -> httpx.Response: ...
_COPILOT_API_BASE_CACHE: Final[_CopilotAPIBaseCache] = cast( # cast-ok: [LIT006] InMemoryCache methods lack type stubs
_CopilotAPIBaseCache,
InMemoryCache(
max_size_in_memory=1000,
default_ttl=GITHUB_COPILOT_USER_SESSION_TTL_SECONDS,
),
)
def _copilot_api_base_for_token(access_token: str) -> str | None:
cache_key: Final = hashlib.sha256(access_token.encode()).hexdigest()
cached_api_base: Final[object] = _COPILOT_API_BASE_CACHE.get_cache(cache_key)
if isinstance(cached_api_base, str):
return cached_api_base
try:
response: Final = cast( # cast-ok: [LIT006] HTTPHandler.get has no return annotation
_SyncGetClient,
get_httpx_client(),
).get(
DEFAULT_GITHUB_COPILOT_USER_URL,
headers=github_api_headers(access_token),
)
if not response.is_success:
return None
payload: Final[Mapping[str, object]] = _JSON_OBJECT.validate_python(response.json())
endpoints: Final[Mapping[str, object]] = _JSON_OBJECT.validate_python(payload.get("endpoints"))
api_base: Final[object] = endpoints.get("api")
if not isinstance(api_base, str):
return None
from .github_host import github_host_endpoints
from .per_user_auth import validated_copilot_api_base
validated: Final = validated_copilot_api_base(api_base, github_host_endpoints("github.com"))
if validated != api_base.rstrip("/"):
return None
_COPILOT_API_BASE_CACHE.set_cache(cache_key, validated)
except Exception:
return None
else:
return validated
def github_api_headers(
access_token: str | None = None,
) -> dict[str, str]: # mutable-ok: returned straight into httpx handlers whose headers params require dict
@ -111,7 +134,6 @@ class Authenticator:
self.token_dir,
os.getenv("GITHUB_COPILOT_ACCESS_TOKEN_FILE", "access-token"),
)
self.api_key_file = os.path.join(self.token_dir, os.getenv("GITHUB_COPILOT_API_KEY_FILE", "api-key.json"))
self._ensure_token_dir()
def get_access_token(self) -> str:
@ -155,7 +177,7 @@ class Authenticator:
except OSError:
verbose_logger.error("Error saving access token to file")
return access_token
except (GetDeviceCodeError, GetAccessTokenError, RefreshAPIKeyError) as e:
except (GetDeviceCodeError, GetAccessTokenError) as e:
verbose_logger.warning("Failed attempt %s: %s", attempt + 1, e)
continue
@ -165,69 +187,15 @@ class Authenticator:
)
def get_api_key(self) -> str:
"""
Get the API key, refreshing if necessary.
Returns:
str: The GitHub Copilot API key.
Raises:
GetAPIKeyError: If unable to obtain an API key.
"""
unsaved: Final = self._unsaved_api_key()
if unsaved is not None:
return unsaved.token
cached: Final = self._read_api_key_file(CopilotCachedKey)
if cached is not None and cached.expires_at > datetime.now().timestamp():
return cached.token
if cached is not None:
verbose_logger.warning("API key expired, refreshing")
try:
api_key_info: Final = self._refresh_api_key()
except RefreshAPIKeyError as e:
raise GetAPIKeyError(
message=f"Failed to refresh API key: {e}",
status_code=401,
) from e
serialized: Final = json.dumps(api_key_info)
try:
with open(self.api_key_file, "w") as f:
f.write(serialized)
except OSError as e:
verbose_logger.warning(
"Keeping the refreshed API key in memory, it could not be saved to %s: %s", self.api_key_file, e
)
self._keep_unsaved_api_key(serialized)
token: Final = api_key_info.get("token")
if not token:
raise GetAPIKeyError(
message="API key response missing token",
status_code=401,
)
return token
"""Return the stored GitHub access token, loading it with device flow when needed."""
return self.get_access_token()
def get_api_base(self) -> str | None:
"""
Get the API endpoint of the current API key.
Returns:
Optional[str]: The GitHub Copilot API endpoint, or None if not found.
"""
cached: Final = self._unsaved_api_key() or self._read_api_key_file(CopilotCachedEndpoints)
if cached is None or cached.endpoints is None:
return None
return cached.endpoints.api
def _read_api_key_file(self, view: type[ApiKeyFileView]) -> ApiKeyFileView | None:
try:
return view.model_validate_json(_read_token_file(self.api_key_file))
except OSError:
verbose_logger.warning("No API key file found or error opening file")
return None
except ValueError as e:
verbose_logger.warning("Ignoring %s, not a Copilot API key file: %s", self.api_key_file, e)
"""Return the Copilot API endpoint for a stored token without starting device flow."""
access_token: Final = self._stored_access_token()
if access_token is None:
return None
return _copilot_api_base_for_token(access_token)
def _stored_access_token(self) -> str | None:
try:
@ -236,59 +204,6 @@ class Authenticator:
return None
return access_token or None
def _unsaved_api_key(self) -> CopilotCachedKey | None:
unsaved: Final = _UNSAVED_API_KEYS.get(self.api_key_file)
if unsaved is None or unsaved.key.expires_at <= time.time():
return None
return unsaved.key if unsaved.access_token == self._stored_access_token() else None
def _keep_unsaved_api_key(self, serialized: str) -> None:
access_token: Final = self._stored_access_token()
if access_token is None:
return
try:
key: Final = CopilotCachedKey.model_validate_json(serialized)
except ValueError:
return
_UNSAVED_API_KEYS[self.api_key_file] = _UnsavedApiKey(access_token=access_token, key=key)
def _refresh_api_key(self) -> dict[str, Any]:
"""
Refresh the API key using the access token.
Returns:
Dict[str, Any]: The API key information including token and expiration.
Raises:
RefreshAPIKeyError: If unable to refresh the API key.
"""
access_token: Final = self.get_access_token()
headers: Final = self._get_github_headers(access_token)
api_key_url: Final = os.getenv("GITHUB_COPILOT_API_KEY_URL", DEFAULT_GITHUB_API_KEY_URL)
max_retries: Final = 3
for attempt in range(max_retries):
try:
sync_client = get_httpx_client()
response = sync_client.get(api_key_url, headers=headers)
response.raise_for_status()
response_json = response.json()
if "token" in response_json:
return response_json
else:
verbose_logger.warning("API key response missing token: %s", response_json)
except httpx.HTTPStatusError as e:
verbose_logger.error("HTTP error refreshing API key (attempt %s/%s): %s", attempt + 1, max_retries, e)
except Exception as e:
verbose_logger.error("Unexpected error refreshing API key: %s", e)
raise RefreshAPIKeyError(
message="Failed to refresh API key after maximum retries",
status_code=401,
)
def _ensure_token_dir(self) -> None:
"""Ensure the token directory exists."""
if not os.path.exists(self.token_dir):

View file

@ -16,7 +16,7 @@ from litellm.types.utils import ModelResponse
from ..authenticator import Authenticator
from ..common_utils import (
DEFAULT_GITHUB_COPILOT_API_BASE,
GetAPIKeyError,
GetAccessTokenError,
get_copilot_default_headers,
)
from ..per_user_auth import require_github_copilot_user_session
@ -56,7 +56,7 @@ class GithubCopilotConfig(OpenAIConfig):
dynamic_api_base: Final = self.api_base_without_login(api_base)
try:
dynamic_api_key: Final = self.authenticator.get_api_key()
except GetAPIKeyError as e:
except GetAccessTokenError as e:
raise AuthenticationError(
model=model,
llm_provider=custom_llm_provider,
@ -104,7 +104,7 @@ class GithubCopilotConfig(OpenAIConfig):
return get_copilot_default_headers(session_token)
try:
return get_copilot_default_headers(self.authenticator.get_api_key())
except GetAPIKeyError:
except GetAccessTokenError:
return {}
def validate_environment(

View file

@ -46,18 +46,6 @@ class GetAccessTokenError(GithubCopilotError):
pass
class APIKeyExpiredError(GithubCopilotError):
pass
class RefreshAPIKeyError(GithubCopilotError):
pass
class GetAPIKeyError(GithubCopilotError):
pass
def get_copilot_default_headers(
api_key: str,
) -> dict[str, str]: # mutable-ok: callers merge caller headers into the returned dict

View file

@ -30,7 +30,7 @@ from litellm.utils import convert_to_model_response_object
from ..authenticator import Authenticator
from ..common_utils import (
DEFAULT_GITHUB_COPILOT_API_BASE,
GetAPIKeyError,
GetAccessTokenError,
get_copilot_default_headers,
pin_session_authorization,
)
@ -100,7 +100,7 @@ class GithubCopilotEmbeddingConfig(BaseEmbeddingConfig):
return merged_headers
except GetAPIKeyError as e:
except GetAccessTokenError as e:
raise AuthenticationError(
model=model,
llm_provider="github_copilot",

View file

@ -7,8 +7,8 @@ from litellm.constants import GITHUB_COPILOT_DEFAULT_HOST
from .authenticator import (
DEFAULT_GITHUB_ACCESS_TOKEN_URL,
DEFAULT_GITHUB_API_KEY_URL,
DEFAULT_GITHUB_CLIENT_ID,
DEFAULT_GITHUB_COPILOT_USER_URL,
DEFAULT_GITHUB_DEVICE_CODE_URL,
)
@ -20,11 +20,9 @@ class GithubHostEndpoints:
host: str
device_code_url: str
access_token_url: str
api_key_url: str
copilot_user_url: str
user_url: str
client_id: str
exchanges_copilot_token: bool
copilot_api_hosts: frozenset[str]
copilot_api_host_suffix: str | None
default_copilot_api_base: str
@ -51,17 +49,14 @@ def github_host_endpoints(host: str) -> GithubHostEndpoints:
if normalized is None:
raise ValueError("host must be github.com or a single-subdomain GHE.com hostname")
client_id: Final = os.getenv("GITHUB_COPILOT_CLIENT_ID", DEFAULT_GITHUB_CLIENT_ID)
exchanges_copilot_token: Final = client_id == DEFAULT_GITHUB_CLIENT_ID
if normalized == GITHUB_COPILOT_DEFAULT_HOST:
return GithubHostEndpoints(
host=normalized,
device_code_url=os.getenv("GITHUB_COPILOT_DEVICE_CODE_URL", DEFAULT_GITHUB_DEVICE_CODE_URL),
access_token_url=os.getenv("GITHUB_COPILOT_ACCESS_TOKEN_URL", DEFAULT_GITHUB_ACCESS_TOKEN_URL),
api_key_url=os.getenv("GITHUB_COPILOT_API_KEY_URL", DEFAULT_GITHUB_API_KEY_URL),
copilot_user_url="https://api.github.com/copilot_internal/user",
copilot_user_url=DEFAULT_GITHUB_COPILOT_USER_URL,
user_url="https://api.github.com/user",
client_id=client_id,
exchanges_copilot_token=exchanges_copilot_token,
copilot_api_hosts=frozenset({"githubcopilot.com"}),
copilot_api_host_suffix=".githubcopilot.com",
default_copilot_api_base="https://api.githubcopilot.com",
@ -74,11 +69,9 @@ def github_host_endpoints(host: str) -> GithubHostEndpoints:
host=normalized,
device_code_url=f"https://{normalized}/login/device/code",
access_token_url=f"https://{normalized}/login/oauth/access_token",
api_key_url=f"https://{api_host}/copilot_internal/v2/token",
copilot_user_url=f"https://{api_host}/copilot_internal/user",
user_url=f"https://{api_host}/user",
client_id=client_id,
exchanges_copilot_token=exchanges_copilot_token,
copilot_api_hosts=frozenset({copilot_api_host}),
copilot_api_host_suffix=None,
default_copilot_api_base=f"https://{copilot_api_host}",

View file

@ -10,7 +10,7 @@ from litellm.llms.anthropic.messages.native.transformation import (
from ..authenticator import Authenticator
from ..common_utils import (
DEFAULT_GITHUB_COPILOT_API_BASE,
GetAPIKeyError,
GetAccessTokenError,
get_copilot_default_headers,
pin_session_authorization,
)
@ -84,7 +84,7 @@ class GithubCopilotAnthropicMessagesConfig(AnthropicMessagesConfig):
dynamic_api_key: Final = (
user_session.token if user_session is not None else self.authenticator.get_api_key()
)
except GetAPIKeyError as e:
except GetAccessTokenError as e:
raise AuthenticationError(
model=model,
llm_provider="github_copilot",

View file

@ -1,7 +1,6 @@
"""Per-user GitHub Copilot OAuth: exchange a caller's stored GitHub token for a
short-lived Copilot token, plus the device-flow helpers the proxy endpoints use
to establish that connection. The shared device login (Authenticator) is never
touched on the per-user path.
"""Per-user GitHub Copilot OAuth sessions and the device-flow helpers the proxy
endpoints use to establish those connections. The shared device login
(Authenticator) is never touched on the per-user path.
"""
import asyncio
@ -28,7 +27,7 @@ from litellm.constants import (
GITHUB_COPILOT_AUTH_TYPE_KEY,
GITHUB_COPILOT_HOST_KEY,
GITHUB_COPILOT_PER_USER_AUTH_TYPE,
GITHUB_COPILOT_USER_TOKEN_SAFETY_MARGIN_SECONDS,
GITHUB_COPILOT_USER_SESSION_TTL_SECONDS,
)
from litellm.exceptions import (
APIConnectionError,
@ -39,6 +38,7 @@ from litellm.exceptions import (
)
from .authenticator import github_api_headers
from .common_utils import get_copilot_default_headers
from .github_host import GithubHostEndpoints, github_host_endpoints, normalized_github_host
GITHUB_COPILOT_USER_SESSION_KWARG_KEY: Final = "github_copilot_user_session"
@ -46,7 +46,6 @@ _LLM_PROVIDER: Final = "github_copilot"
_DEVICE_FLOW_SCOPE: Final = "read:user"
_SESSION_CACHE_MAX_SIZE: Final = 1000
_SYNC_LOCK_COUNT: Final = 64
_ACCOUNT_LOOKUP_TTL_SECONDS: Final = 1800
class _SessionCache(Protocol):
@ -88,19 +87,19 @@ _SESSION_CACHE: Final = cast( # cast-ok: pins the untyped module cache to _Sess
_SessionCache,
InMemoryCache(
max_size_in_memory=_SESSION_CACHE_MAX_SIZE,
default_ttl=GITHUB_COPILOT_USER_TOKEN_SAFETY_MARGIN_SECONDS,
default_ttl=GITHUB_COPILOT_USER_SESSION_TTL_SECONDS,
),
)
_EXCHANGE_LOCKS: Final = tuple(threading.Lock() for _ in range(_SYNC_LOCK_COUNT))
_RESOLVE_LOCKS: Final = tuple(threading.Lock() for _ in range(_SYNC_LOCK_COUNT))
_IN_FLIGHT: Final[ # mutable-ok: single-flight dedup registry mutated under its own locking
dict[str, tuple[str, "asyncio.Future[GithubCopilotUserSession]"]]
] = {}
@with_config(ConfigDict(extra="allow", strict=True))
class _CopilotTokenPayload(TypedDict):
token: ReadOnly[str]
expires_at: ReadOnly[NotRequired[object]]
class _CopilotUserPayload(TypedDict):
access_type_sku: ReadOnly[NotRequired[object]]
chat_enabled: ReadOnly[NotRequired[object]]
endpoints: ReadOnly[NotRequired[object]]
@ -126,20 +125,12 @@ class _CopilotEndpointsPayload(TypedDict):
api: ReadOnly[NotRequired[object]]
@with_config(ConfigDict(extra="allow", strict=True))
class _CopilotAccountPayload(TypedDict):
access_type_sku: ReadOnly[NotRequired[object]]
chat_enabled: ReadOnly[NotRequired[object]]
endpoints: ReadOnly[NotRequired[object]]
@with_config(ConfigDict(extra="allow", strict=True))
class _GithubUserPayload(TypedDict):
login: ReadOnly[str]
_COPILOT_TOKEN: Final = TypeAdapter(_CopilotTokenPayload)
_COPILOT_ACCOUNT: Final = TypeAdapter(_CopilotAccountPayload)
_COPILOT_USER: Final = TypeAdapter(_CopilotUserPayload)
_DEVICE_CODE: Final = TypeAdapter(_DeviceCodePayload)
_ACCESS_TOKEN_POLL: Final = TypeAdapter(_AccessTokenPollPayload)
_GITHUB_USER: Final = TypeAdapter(_GithubUserPayload)
@ -237,13 +228,13 @@ def _cached_session(cache_key: str, github_token: str) -> GithubCopilotUserSessi
return None
if not hmac.compare_digest(cached_github_token, github_token):
return None
if expires_at - GITHUB_COPILOT_USER_TOKEN_SAFETY_MARGIN_SECONDS <= time.time():
if expires_at <= time.time():
return None
return GithubCopilotUserSession(token=token, api_base=api_base)
def _cache_session(cache_key: str, github_token: str, session: GithubCopilotUserSession, expires_at: float) -> None:
ttl: Final = expires_at - GITHUB_COPILOT_USER_TOKEN_SAFETY_MARGIN_SECONDS - time.time()
ttl: Final = expires_at - time.time()
if ttl <= 0:
return
_SESSION_CACHE.set_cache(cache_key, (github_token, session.token, session.api_base, expires_at), ttl=ttl)
@ -272,57 +263,32 @@ def _session_from_response(
raise _reject_connection(credential_name)
if response.status_code == 429:
raise CallerCredentialRateLimitError(
message="GitHub rate limited the GitHub Copilot token exchange",
message="GitHub rate limited the GitHub Copilot user lookup",
llm_provider=_LLM_PROVIDER,
model="",
)
if not response.is_success:
raise ServiceUnavailableError(
message=f"GitHub Copilot token exchange failed with status {response.status_code}",
message=f"GitHub Copilot user lookup failed with status {response.status_code}",
llm_provider=_LLM_PROVIDER,
model="",
)
if not endpoints.exchanges_copilot_token:
return _session_from_account(response, github_token, credential_name, cache_key, endpoints)
try:
payload: Final = _COPILOT_TOKEN.validate_python(response.json())
payload: Final = _COPILOT_USER.validate_python(response.json())
except Exception as e:
raise ServiceUnavailableError(
message="GitHub Copilot token exchange returned an invalid response",
message="GitHub Copilot user lookup returned an invalid response",
llm_provider=_LLM_PROVIDER,
model="",
) from e
api_base: Final = validated_copilot_api_base(_endpoints_api(payload.get("endpoints")), endpoints)
session: Final = GithubCopilotUserSession(token=payload["token"], api_base=api_base)
expires_at: Final = _expires_at_seconds(payload.get("expires_at"))
if expires_at > 0:
_cache_session(cache_key, github_token, session, expires_at)
return session
def _session_from_account(
response: httpx.Response,
github_token: str,
credential_name: str,
cache_key: str,
endpoints: GithubHostEndpoints,
) -> GithubCopilotUserSession:
try:
payload: Final = _COPILOT_ACCOUNT.validate_python(response.json())
except Exception as e:
raise ServiceUnavailableError(
message="GitHub Copilot account lookup returned an invalid response",
llm_provider=_LLM_PROVIDER,
model="",
) from e
if payload.get("chat_enabled") is False or payload.get("access_type_sku") == "no_access":
_SESSION_CACHE.delete_cache(cache_key)
raise _reject_connection(credential_name)
api_base: Final = validated_copilot_api_base(_endpoints_api(payload.get("endpoints")), endpoints)
session: Final = GithubCopilotUserSession(token=github_token, api_base=api_base)
_cache_session(cache_key, github_token, session, time.time() + _ACCOUNT_LOOKUP_TTL_SECONDS)
_cache_session(cache_key, github_token, session, time.time() + GITHUB_COPILOT_USER_SESSION_TTL_SECONDS)
return session
@ -332,10 +298,6 @@ def _endpoints_api(payload_endpoints: object) -> object:
return _COPILOT_ENDPOINTS.validate_python(payload_endpoints).get("api")
def _session_lookup_url(endpoints: GithubHostEndpoints) -> str:
return endpoints.api_key_url if endpoints.exchanges_copilot_token else endpoints.copilot_user_url
def _to_int(raw: object) -> int:
if isinstance(raw, bool):
return 0
@ -351,20 +313,7 @@ def _to_int(raw: object) -> int:
return 0
def _expires_at_seconds(raw: object) -> float:
if isinstance(raw, bool):
return 0.0
if isinstance(raw, (int, float)):
return float(raw)
if isinstance(raw, str):
try:
return float(raw)
except ValueError:
return 0.0
return 0.0
def _do_exchange(
def _do_resolve(
github_token: str,
credential_name: str,
cache_key: str,
@ -377,12 +326,12 @@ def _do_exchange(
_SyncGetClient, litellm.module_level_client
)
response: Final = client.get(
_session_lookup_url(endpoints),
endpoints.copilot_user_url,
headers=github_api_headers(github_token),
)
except Exception as e:
raise APIConnectionError(
message="GitHub Copilot token exchange request failed",
message="GitHub Copilot user lookup request failed",
llm_provider=_LLM_PROVIDER,
model="",
) from e
@ -398,7 +347,7 @@ def _new_async_github_client() -> _AsyncGitHubClient:
)(_LLM_PROVIDER)
async def _do_aexchange(
async def _do_aresolve(
github_token: str,
credential_name: str,
cache_key: str,
@ -407,19 +356,19 @@ async def _do_aexchange(
client: Final = _new_async_github_client()
try:
response: Final = await client.get(
_session_lookup_url(endpoints),
endpoints.copilot_user_url,
headers=github_api_headers(github_token),
)
except Exception as e:
raise APIConnectionError(
message="GitHub Copilot token exchange request failed",
message="GitHub Copilot user lookup request failed",
llm_provider=_LLM_PROVIDER,
model="",
) from e
return _session_from_response(response, github_token, credential_name, cache_key, endpoints)
def exchange_github_token(
def resolve_copilot_user_session(
user_id: str,
github_token: str,
credential_name: str,
@ -429,15 +378,15 @@ def exchange_github_token(
cached: Final = _cached_session(cache_key, github_token)
if cached is not None:
return cached
lock: Final = _EXCHANGE_LOCKS[hash(cache_key) % _SYNC_LOCK_COUNT]
lock: Final = _RESOLVE_LOCKS[hash(cache_key) % _SYNC_LOCK_COUNT]
with lock:
rechecked: Final = _cached_session(cache_key, github_token)
if rechecked is not None:
return rechecked
return _do_exchange(github_token, credential_name, cache_key, endpoints)
return _do_resolve(github_token, credential_name, cache_key, endpoints)
async def aexchange_github_token(
async def aresolve_copilot_user_session(
user_id: str,
github_token: str,
credential_name: str,
@ -452,14 +401,14 @@ async def aexchange_github_token(
if in_flight_entry is not None:
in_flight_token, in_flight = in_flight_entry
if not hmac.compare_digest(in_flight_token, github_token):
return await _do_aexchange(github_token, credential_name, cache_key, endpoints)
return await _do_aresolve(github_token, credential_name, cache_key, endpoints)
if not in_flight.done() and in_flight.get_loop() is loop:
return await asyncio.shield(in_flight)
future: Final = loop.create_future()
entry: Final = (github_token, future)
_IN_FLIGHT[cache_key] = entry
try:
session: Final = await _do_aexchange(github_token, credential_name, cache_key, endpoints)
session: Final = await _do_aresolve(github_token, credential_name, cache_key, endpoints)
except BaseException as e:
if not future.done():
future.set_exception(e)
@ -610,12 +559,41 @@ async def acheck_copilot_seat(
credential_name: str,
endpoints: GithubHostEndpoints,
) -> GithubCopilotUserSession:
return await aexchange_github_token(
session: Final = await aresolve_copilot_user_session(
user_id=user_id,
github_token=github_token,
credential_name=credential_name,
endpoints=endpoints,
)
client: Final = _new_async_github_client()
models_url: Final = f"{session.api_base}/models"
try:
response: Final = await client.get(
models_url,
headers=get_copilot_default_headers(session.token),
)
except Exception as e:
raise APIConnectionError(
message="GitHub Copilot models check request failed",
llm_provider=_LLM_PROVIDER,
model="",
) from e
if response.status_code in (401, 403, 404):
_SESSION_CACHE.delete_cache(_session_cache_key(user_id, credential_name, endpoints.host))
raise _reject_connection(credential_name)
if response.status_code == 429:
raise CallerCredentialRateLimitError(
message="GitHub rate limited the GitHub Copilot models check",
llm_provider=_LLM_PROVIDER,
model="",
)
if not response.is_success:
raise ServiceUnavailableError(
message=f"GitHub Copilot models check failed with status {response.status_code}",
llm_provider=_LLM_PROVIDER,
model="",
)
return session
def github_copilot_auth_mode(litellm_credential_name: object, auth_type_value: object) -> bool:
@ -804,7 +782,7 @@ def attach_github_copilot_user_session(
credential_name: Final = credential_name_obj if isinstance(credential_name_obj, str) else ""
endpoints: Final = github_copilot_host_endpoints(credential_name)
user_id, github_token = _caller_connection(kwargs, credential_name, endpoints)
session: Final = exchange_github_token(
session: Final = resolve_copilot_user_session(
user_id=user_id,
github_token=github_token,
credential_name=credential_name,
@ -826,7 +804,7 @@ async def aattach_github_copilot_user_session(
credential_name: Final = credential_name_obj if isinstance(credential_name_obj, str) else ""
endpoints: Final = github_copilot_host_endpoints(credential_name)
user_id, github_token = _caller_connection(kwargs, credential_name, endpoints)
session: Final = await aexchange_github_token(
session: Final = await aresolve_copilot_user_session(
user_id=user_id,
github_token=github_token,
credential_name=credential_name,

View file

@ -38,7 +38,7 @@ from litellm.utils import cached_get_model_info_helper
from ..authenticator import Authenticator
from ..common_utils import (
DEFAULT_GITHUB_COPILOT_API_BASE,
GetAPIKeyError,
GetAccessTokenError,
get_copilot_default_headers,
pin_session_authorization,
)
@ -256,7 +256,7 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
return merged_headers
except GetAPIKeyError as e:
except GetAccessTokenError as e:
raise AuthenticationError(
model=model,
llm_provider="github_copilot",

View file

@ -1,6 +1,5 @@
import json
import pytest
import respx
import litellm
from litellm.litellm_core_utils.llm_response_utils import get_api_base as get_api_base_module
@ -53,17 +52,20 @@ class TestDeclaredAuthenticatingProvider:
assert resolution_lookups == []
assert api_base == expected
def test_copilot_keeps_the_enterprise_endpoint_from_disk(self, isolated_token_dirs, resolution_lookups):
def test_copilot_uses_the_user_endpoint_for_an_enterprise_api_base(self, isolated_token_dirs, resolution_lookups):
token_dir = isolated_token_dirs / "github_copilot"
token_dir.mkdir()
(token_dir / "api-key.json").write_text(
json.dumps({"endpoints": {"api": "https://api.enterprise.githubcopilot.com"}})
)
api_base = litellm.get_api_base(model="github_copilot/gpt-4o", optional_params={})
token = f"gho_{isolated_token_dirs.name}"
(token_dir / "access-token").write_text(token)
with respx.mock(assert_all_called=True) as router:
route = router.get("https://api.github.com/copilot_internal/user").respond(
json={"endpoints": {"api": "https://api.enterprise.githubcopilot.com"}}
)
api_base = litellm.get_api_base(model="github_copilot/gpt-4o", optional_params={})
assert resolution_lookups == []
assert api_base == "https://api.enterprise.githubcopilot.com"
assert route.calls.last.request.headers["authorization"] == f"token {token}"
def test_explicit_api_base_still_wins(self, isolated_token_dirs, resolution_lookups):
api_base = litellm.get_api_base(

View file

@ -1,15 +1,14 @@
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock
import httpx
import pytest
from pydantic import ValidationError
from litellm.exceptions import AuthenticationError
from litellm.llms.github_copilot.common_utils import GetAccessTokenError
from litellm.llms.github_copilot.embedding.transformation import (
GithubCopilotEmbeddingConfig,
)
from litellm.llms.github_copilot.common_utils import GetAPIKeyError
from litellm.types.utils import EmbeddingResponse
@ -41,7 +40,7 @@ def test_github_copilot_embedding_config_validate_environment():
assert "x-request-id" in validated_headers
# Test with authentication failure
config.authenticator.get_api_key.side_effect = GetAPIKeyError(
config.authenticator.get_api_key.side_effect = GetAccessTokenError(
message="Failed to get API key",
status_code=401,
)

View file

@ -4,9 +4,8 @@ from unittest.mock import MagicMock
import httpx
import pytest
from litellm.exceptions import AuthenticationError
from litellm.llms.github_copilot.common_utils import GetAPIKeyError
from litellm.llms.github_copilot.common_utils import GetAccessTokenError
from litellm.llms.github_copilot.messages.transformation import (
GithubCopilotAnthropicMessagesConfig,
)
@ -216,7 +215,9 @@ def test_github_copilot_anthropic_messages_validate_environment_auth_error():
# Mock the authenticator to raise an error
config.authenticator = MagicMock()
config.authenticator.get_api_key.side_effect = GetAPIKeyError(status_code=401, message="No valid API key found")
config.authenticator.get_api_key.side_effect = GetAccessTokenError(
status_code=401, message="No valid API key found"
)
with pytest.raises(AuthenticationError):
config.validate_anthropic_messages_environment(

View file

@ -1,559 +1,384 @@
import json
import logging
import os
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
from pathlib import Path
from types import MappingProxyType
from typing import Final
from unittest.mock import MagicMock, mock_open, patch
from unittest.mock import MagicMock, patch
import pytest
import respx
from litellm.llms.github_copilot.authenticator import Authenticator
from litellm.llms.github_copilot.common_utils import (
APIKeyExpiredError,
GetAccessTokenError,
GetAPIKeyError,
GetDeviceCodeError,
RefreshAPIKeyError,
)
from litellm.llms.github_copilot.authenticator import Authenticator, github_api_headers
from litellm.llms.github_copilot.common_utils import GetAccessTokenError, GetDeviceCodeError
API_KEY_URL: Final = "https://copilot-key.test/copilot_internal/v2/token"
REFRESHED_KEY: Final = MappingProxyType({"token": "new-api-key", "expires_at": 4102444800})
COPILOT_USER_URL: Final = "https://api.github.com/copilot_internal/user"
DEVICE_CODE_URL: Final = "https://github.com/login/device/code"
TENANT_API: Final = "https://api.tenant.githubcopilot.com"
REPAIRED_API: Final = "https://api.repaired.githubcopilot.com"
TRUNCATED_UTF_16: Final = b"\xff\xfe\x00\xd8"
@pytest.fixture
def logged_in_token_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
def logged_in_token_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> tuple[Path, str]:
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(tmp_path))
monkeypatch.setenv("GITHUB_COPILOT_API_KEY_URL", API_KEY_URL)
(tmp_path / "access-token").write_text("gho_logged_in")
return tmp_path
token: Final = f"gho_{tmp_path.name}"
(tmp_path / "access-token").write_text(token)
return tmp_path, token
class TestGitHubCopilotAuthenticator:
@pytest.fixture
def authenticator(self):
with (
patch("os.path.exists", return_value=False),
patch("os.makedirs") as mock_makedirs,
):
auth = Authenticator()
mock_makedirs.assert_called_once()
return auth
def test_init_uses_access_token_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(tmp_path))
auth: Final = Authenticator()
assert auth.token_dir == str(tmp_path)
assert auth.access_token_file == str(tmp_path / "access-token")
@pytest.fixture
def mock_http_client(self):
mock_client = MagicMock()
mock_response = MagicMock()
mock_client.get.return_value = mock_response
mock_client.post.return_value = mock_response
mock_response.raise_for_status.return_value = None
return mock_client, mock_response
def test_init(self):
"""Test the initialization of the authenticator."""
with (
patch("os.path.exists", return_value=False),
patch("os.makedirs") as mock_makedirs,
):
auth = Authenticator()
assert auth.token_dir.endswith("/github_copilot")
assert auth.access_token_file.endswith("/access-token")
assert auth.api_key_file.endswith("/api-key.json")
mock_makedirs.assert_called_once()
def test_init_uses_default_access_token_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.delenv("GITHUB_COPILOT_TOKEN_DIR", raising=False)
def test_ensure_token_dir(self):
"""Test that the token directory is created if it doesn't exist."""
with (
patch("os.path.exists", return_value=False),
patch("os.makedirs") as mock_makedirs,
):
auth = Authenticator()
mock_makedirs.assert_called_once_with(auth.token_dir, exist_ok=True)
auth: Final = Authenticator()
def test_get_github_headers(self, authenticator):
"""Test that GitHub headers are correctly generated."""
headers = authenticator._get_github_headers()
assert "accept" in headers
assert "editor-version" in headers
assert "user-agent" in headers
assert "content-type" in headers
assert auth.token_dir == str(tmp_path / ".config/litellm/github_copilot")
assert auth.access_token_file == str(tmp_path / ".config/litellm/github_copilot/access-token")
headers_with_token = authenticator._get_github_headers("test-token")
assert headers_with_token["authorization"] == "token test-token"
def test_get_access_token_from_file(self, logged_in_token_dir: Path) -> None:
"""Test retrieving an access token from a file."""
assert Authenticator().get_access_token() == "gho_logged_in"
def test_ensure_token_dir_creates_missing_directory(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
token_dir: Final = str(tmp_path / "github_copilot")
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", token_dir)
def test_get_access_token_login(self, authenticator):
"""Test logging in to get an access token."""
mock_token = "mock-access-token"
auth: Final = Authenticator()
with (
patch.object(authenticator, "_login", return_value=mock_token),
patch("builtins.open", mock_open()),
patch("builtins.open", side_effect=IOError) as mock_read,
):
token = authenticator.get_access_token()
assert token == mock_token
authenticator._login.assert_called_once()
assert Path(auth.token_dir).is_dir()
@pytest.mark.asyncio
async def test_get_access_token_refuses_device_code_login_in_event_loop(self, authenticator):
with (
patch("builtins.open", side_effect=FileNotFoundError),
patch.object(authenticator, "_login") as mock_login,
):
with pytest.raises(GetAccessTokenError) as exc:
authenticator.get_access_token()
assert exc.value.status_code == 401
assert "event loop" in str(exc.value)
assert authenticator.access_token_file not in str(exc.value)
mock_login.assert_not_called()
def test_get_github_headers_include_access_token() -> None:
headers: Final = github_api_headers("github-token")
def test_get_access_token_refuses_device_code_login_in_worker_thread(self, authenticator):
with (
patch("builtins.open", side_effect=FileNotFoundError),
patch.object(authenticator, "_login") as mock_login,
):
with ThreadPoolExecutor(max_workers=1) as pool:
with pytest.raises(GetAccessTokenError) as exc:
pool.submit(authenticator.get_access_token).result()
assert headers == {
"accept": "application/json",
"editor-version": "vscode/1.85.1",
"editor-plugin-version": "copilot/1.155.0",
"user-agent": "GithubCopilot/1.155.0",
"accept-encoding": "gzip,deflate,br",
"content-type": "application/json",
"authorization": "token github-token",
}
assert exc.value.status_code == 401
assert "worker thread" in str(exc.value)
assert authenticator.access_token_file not in str(exc.value)
mock_login.assert_not_called()
@pytest.mark.parametrize("encoding", ["utf-16", "utf-16-le", "utf-8-sig"])
def test_saved_access_token_in_another_encoding_is_sent_on_refresh(
self, logged_in_token_dir: Path, encoding: str
) -> None:
"""An access-token file a Windows editor saved as UTF-16 or with a BOM still refreshes the key."""
(logged_in_token_dir / "access-token").write_bytes("gho_logged_in\n".encode(encoding))
def test_get_access_token_reads_saved_token(logged_in_token_dir: tuple[Path, str]) -> None:
_, token = logged_in_token_dir
assert Authenticator().get_access_token() == token
with respx.mock(assert_all_called=True) as router:
route: Final = router.get(API_KEY_URL).respond(json=dict(REFRESHED_KEY))
assert Authenticator().get_api_key() == "new-api-key"
assert route.calls.last.request.headers["authorization"] == "token gho_logged_in"
def test_get_access_token_logs_in_and_saves_token(
logged_in_token_dir: tuple[Path, str],
) -> None:
token_dir, _ = logged_in_token_dir
access_token_file: Final = token_dir / "access-token"
access_token_file.unlink()
auth: Final = Authenticator()
def test_undecodable_access_token_is_refused_like_a_missing_one(self, logged_in_token_dir: Path) -> None:
"""A proxy worker gets the login hint for an access-token file it cannot decode, and the file is kept."""
access_token_file: Final = logged_in_token_dir / "access-token"
access_token_file.write_bytes(TRUNCATED_UTF_16)
with patch.object(auth, "_login", return_value="github-access-token") as login:
assert auth.get_access_token() == "github-access-token"
with ThreadPoolExecutor(max_workers=1) as pool, pytest.raises(GetAccessTokenError) as exc:
pool.submit(Authenticator().get_access_token).result()
login.assert_called_once()
assert access_token_file.read_text() == "github-access-token"
assert "worker thread" in str(exc.value)
assert access_token_file.read_bytes() == TRUNCATED_UTF_16
def test_get_access_token_failure(self, authenticator):
"""Test that an exception is raised after multiple login failures."""
with (
patch.object(
authenticator,
"_login",
side_effect=GetDeviceCodeError(message="Test error", status_code=400),
),
patch("builtins.open", side_effect=IOError),
):
with pytest.raises(GetAccessTokenError):
authenticator.get_access_token()
assert authenticator._login.call_count == 3
def test_get_api_key_returns_github_token_without_http(
logged_in_token_dir: tuple[Path, str],
) -> None:
_, token = logged_in_token_dir
with patch("litellm.llms.github_copilot.authenticator.get_httpx_client") as get_client:
assert Authenticator().get_api_key() == token
get_client.assert_not_called()
def test_get_api_key_from_file(self, authenticator):
"""Test retrieving an API key from a file."""
future_time = (datetime.now() + timedelta(hours=1)).timestamp()
mock_api_key_data = json.dumps(
{"token": "mock-api-key", "expires_at": future_time}
).encode()
with patch("builtins.open", mock_open(read_data=mock_api_key_data)):
api_key = authenticator.get_api_key()
assert api_key == "mock-api-key"
def test_get_api_key_expired(self, authenticator):
"""Test refreshing an expired API key."""
past_time = (datetime.now() - timedelta(hours=1)).timestamp()
mock_expired_data = json.dumps(
{"token": "expired-api-key", "expires_at": past_time}
).encode()
mock_new_data = {
"token": "new-api-key",
"expires_at": (datetime.now() + timedelta(hours=1)).timestamp(),
}
with (
patch("builtins.open", mock_open(read_data=mock_expired_data)),
patch.object(authenticator, "_refresh_api_key", return_value=mock_new_data),
patch("json.dump") as mock_json_dump,
):
api_key = authenticator.get_api_key()
assert api_key == "new-api-key"
authenticator._refresh_api_key.assert_called_once()
@pytest.mark.parametrize(
"malformed",
[
b"[]",
json.dumps({"token": "tid=abc", "expires_at": "soon"}).encode(),
TRUNCATED_UTF_16,
json.dumps({"token": "tid=abc", "expires_at": 4102444800, "endpoints": []}).encode(),
],
ids=["json-list", "text-expires-at", "truncated-utf-16", "list-endpoints-beside-live-key"],
)
def test_get_api_key_malformed_file_refreshes_like_missing(
self, logged_in_token_dir: Path, malformed: bytes
) -> None:
"""A file that is not a Copilot key object is ignored, refreshed over, never answered with a parse error."""
api_key_file: Final = logged_in_token_dir / "api-key.json"
api_key_file.write_bytes(malformed)
with respx.mock(assert_all_called=True) as router:
route: Final = router.get(API_KEY_URL).respond(json=dict(REFRESHED_KEY))
assert Authenticator().get_api_key() == "new-api-key"
assert route.call_count == 1
assert route.calls.last.request.headers["authorization"] == "token gho_logged_in"
assert json.loads(api_key_file.read_text()) == REFRESHED_KEY
@pytest.mark.parametrize(
"malformed",
[
b"[]",
json.dumps({"token": "tid=abc", "expires_at": "soon"}).encode(),
TRUNCATED_UTF_16,
json.dumps({"token": "tid=abc", "expires_at": 4102444800, "endpoints": []}).encode(),
],
ids=["json-list", "text-expires-at", "truncated-utf-16", "list-endpoints-beside-live-key"],
)
def test_get_api_key_malformed_file_is_kept_when_refresh_fails(
self, logged_in_token_dir: Path, malformed: bytes
) -> None:
"""Ignoring the file never deletes it, so a failed refresh leaves the operator's file for inspection."""
api_key_file: Final = logged_in_token_dir / "api-key.json"
api_key_file.write_bytes(malformed)
with respx.mock(assert_all_called=True) as router:
router.get(API_KEY_URL).respond(401)
with pytest.raises(GetAPIKeyError):
Authenticator().get_api_key()
assert api_key_file.read_bytes() == malformed
def test_refreshed_key_that_cannot_be_saved_is_reused_with_its_endpoint(self, logged_in_token_dir: Path) -> None:
unwritable_key_file: Final = logged_in_token_dir / "api-key.json"
unwritable_key_file.mkdir()
first: Final = Authenticator()
second: Final = Authenticator()
with respx.mock(assert_all_called=True) as router:
route: Final = router.get(API_KEY_URL).respond(json={**REFRESHED_KEY, "endpoints": {"api": TENANT_API}})
served: Final = (first.get_api_key(), second.get_api_key(), first.get_api_key())
assert served == ("new-api-key",) * 3
assert (first.get_api_base(), second.get_api_base()) == (TENANT_API, TENANT_API)
assert route.call_count == 1
assert unwritable_key_file.is_dir()
def test_refreshed_key_that_cannot_be_saved_and_has_expired_is_refreshed_again(
self, logged_in_token_dir: Path
) -> None:
(logged_in_token_dir / "api-key.json").mkdir()
def test_get_api_base_uses_user_endpoint_and_memoizes_per_access_token(
logged_in_token_dir: tuple[Path, str],
) -> None:
_, token = logged_in_token_dir
with respx.mock(assert_all_called=True) as router:
route: Final = router.get(COPILOT_USER_URL).respond(json={"endpoints": {"api": TENANT_API}})
auth: Final = Authenticator()
assert auth.get_api_base() == TENANT_API
assert Authenticator().get_api_base() == TENANT_API
with respx.mock(assert_all_called=True) as router:
route: Final = router.get(API_KEY_URL).respond(json={"token": "expired-api-key", "expires_at": 1})
served: Final = (auth.get_api_key(), auth.get_api_key())
assert route.call_count == 1
assert route.calls.last.request.headers["authorization"] == f"token {token}"
assert served == ("expired-api-key",) * 2
assert route.call_count == 2
def test_kept_key_and_its_endpoint_stay_paired_after_the_key_file_is_repaired(
self, logged_in_token_dir: Path
) -> None:
key_file: Final = logged_in_token_dir / "api-key.json"
key_file.mkdir()
auth: Final = Authenticator()
with respx.mock(assert_all_called=True) as router:
router.get(API_KEY_URL).respond(json={**REFRESHED_KEY, "endpoints": {"api": TENANT_API}})
auth.get_api_key()
key_file.rmdir()
key_file.write_text(
json.dumps({"token": "repaired-api-key", "expires_at": 9999999999, "endpoints": {"api": REPAIRED_API}})
)
assert (auth.get_api_key(), auth.get_api_base()) in (
("new-api-key", TENANT_API),
("repaired-api-key", REPAIRED_API),
)
def test_unsaved_key_is_not_reused_after_the_access_token_changes(self, logged_in_token_dir: Path) -> None:
(logged_in_token_dir / "api-key.json").mkdir()
with respx.mock(assert_all_called=True) as router:
route: Final = router.get(API_KEY_URL).respond(json=dict(REFRESHED_KEY))
Authenticator().get_api_key()
(logged_in_token_dir / "access-token").write_text("gho_logged_in_again")
Authenticator().get_api_key()
assert [call.request.headers["authorization"] for call in route.calls] == [
"token gho_logged_in",
"token gho_logged_in_again",
]
@pytest.mark.parametrize(
"malformed",
[
b"[]",
json.dumps({"endpoints": {"api": 5}}).encode(),
json.dumps({"token": "t", "expires_at": 1, "endpoints": []}).encode(),
TRUNCATED_UTF_16,
],
ids=["json-list", "number-api", "list-endpoints", "truncated-utf-16"],
)
def test_get_api_base_malformed_file_returns_none(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, malformed: bytes
) -> None:
"""get_api_base runs before get_api_key on every request, so it must survive malformed shapes too."""
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(tmp_path))
(tmp_path / "api-key.json").write_bytes(malformed)
@pytest.mark.parametrize("status_code", [401, 403, 500])
def test_get_api_base_returns_none_when_user_lookup_fails(
logged_in_token_dir: tuple[Path, str], status_code: int
) -> None:
with respx.mock(assert_all_called=True) as router:
router.get(COPILOT_USER_URL).respond(status_code=status_code)
assert Authenticator().get_api_base() is None
def test_get_api_base_keeps_endpoints_when_the_key_is_malformed(self, logged_in_token_dir: Path) -> None:
"""The endpoint is read on its own, so a bad key never sends the refreshed key to the default host."""
(logged_in_token_dir / "api-key.json").write_text(
json.dumps({"token": "tid=abc", "expires_at": "soon", "endpoints": {"api": TENANT_API}})
)
auth: Final = Authenticator()
def test_get_api_base_retries_after_failed_user_lookup(logged_in_token_dir: tuple[Path, str]) -> None:
auth: Final = Authenticator()
with respx.mock(assert_all_called=True) as router:
failed_lookup: Final = router.get(COPILOT_USER_URL).respond(status_code=503)
assert auth.get_api_base() is None
assert failed_lookup.call_count == 1
with respx.mock(assert_all_called=True) as router:
successful_lookup: Final = router.get(COPILOT_USER_URL).respond(json={"endpoints": {"api": TENANT_API}})
assert auth.get_api_base() == TENANT_API
with respx.mock(assert_all_called=True) as router:
router.get(API_KEY_URL).respond(json=dict(REFRESHED_KEY))
assert auth.get_api_key() == "new-api-key"
assert successful_lookup.call_count == 1
@pytest.mark.parametrize("encoding", ["utf-16", "utf-16-le", "utf-8-sig"])
def test_live_key_file_in_another_json_encoding_is_used_without_a_refresh(
self, logged_in_token_dir: Path, encoding: str
) -> None:
"""A key file a Windows editor saved as UTF-16 or with a BOM still holds a live key and tenant."""
live_key: Final = json.dumps({"token": "tid=abc", "expires_at": 4102444800, "endpoints": {"api": TENANT_API}})
api_key_file: Final = logged_in_token_dir / "api-key.json"
api_key_file.write_bytes(live_key.encode(encoding))
auth: Final = Authenticator()
with respx.mock:
assert auth.get_api_base() == TENANT_API
assert auth.get_api_key() == "tid=abc"
def test_get_api_base_returns_none_when_user_lookup_response_has_no_endpoint(
logged_in_token_dir: tuple[Path, str],
) -> None:
with respx.mock(assert_all_called=True) as router:
router.get(COPILOT_USER_URL).respond(json={"endpoints": {}})
assert Authenticator().get_api_base() is None
assert api_key_file.read_bytes() == live_key.encode(encoding)
def test_ignored_key_file_is_logged_without_its_contents(
self, logged_in_token_dir: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""The warning names the file but never echoes a token saved in the wrong shape."""
(logged_in_token_dir / "api-key.json").write_text(
json.dumps({"token": ["tid=secret-value"], "expires_at": 4102444800})
)
def test_get_api_base_returns_none_when_user_lookup_endpoints_are_not_a_mapping(
logged_in_token_dir: tuple[Path, str],
) -> None:
with respx.mock(assert_all_called=True) as router:
router.get(COPILOT_USER_URL).respond(json={"endpoints": []})
assert Authenticator().get_api_base() is None
with caplog.at_level(logging.WARNING, logger="LiteLLM"), respx.mock(assert_all_called=True) as router:
router.get(API_KEY_URL).respond(json=dict(REFRESHED_KEY))
assert Authenticator().get_api_key() == "new-api-key"
assert "not a Copilot API key file" in caplog.text
assert "tid=secret-value" not in caplog.text
def test_get_api_base_rejects_untrusted_user_endpoint(
logged_in_token_dir: tuple[Path, str],
) -> None:
with respx.mock(assert_all_called=True) as router:
router.get(COPILOT_USER_URL).respond(json={"endpoints": {"api": "https://attacker.example"}})
assert Authenticator().get_api_base() is None
def test_refresh_api_key(self, authenticator, mock_http_client):
"""Test refreshing an API key."""
mock_client, mock_response = mock_http_client
mock_token = "mock-access-token"
mock_api_key_data = {"token": "new-api-key", "expires_at": 12345}
with (
patch.object(authenticator, "get_access_token", return_value=mock_token),
patch(
"litellm.llms.github_copilot.authenticator.get_httpx_client",
return_value=mock_client,
),
patch.object(mock_response, "json", return_value=mock_api_key_data),
):
result = authenticator._refresh_api_key()
assert result == mock_api_key_data
mock_client.get.assert_called_once()
authenticator.get_access_token.assert_called_once()
def test_get_api_base_does_not_trigger_device_login_without_saved_token(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(tmp_path))
with patch("litellm.llms.github_copilot.authenticator.get_httpx_client") as get_client:
assert Authenticator().get_api_base() is None
get_client.assert_not_called()
def test_refresh_api_key_failure(self, authenticator, mock_http_client):
"""Test failure to refresh an API key."""
mock_client, mock_response = mock_http_client
mock_token = "mock-access-token"
with (
patch.object(authenticator, "get_access_token", return_value=mock_token),
patch(
"litellm.llms.github_copilot.authenticator.get_httpx_client",
return_value=mock_client,
),
patch.object(mock_response, "json", return_value={}),
):
with pytest.raises(RefreshAPIKeyError):
authenticator._refresh_api_key()
assert mock_client.get.call_count == 3
@pytest.mark.asyncio
async def test_get_access_token_refuses_device_code_login_in_event_loop() -> None:
auth: Final = Authenticator()
with (
patch("builtins.open", side_effect=FileNotFoundError),
patch.object(auth, "_login") as login,
):
with pytest.raises(GetAccessTokenError) as exc:
auth.get_access_token()
def test_get_device_code(self, authenticator, mock_http_client):
"""Test getting a device code."""
mock_client, mock_response = mock_http_client
mock_device_code_data = {
"device_code": "mock-device-code",
"user_code": "ABCD-EFGH",
"verification_uri": "https://github.com/login/device",
}
assert exc.value.status_code == 401
assert "event loop" in str(exc.value)
login.assert_not_called()
with (
patch(
"litellm.llms.github_copilot.authenticator.get_httpx_client",
return_value=mock_client,
),
patch.object(mock_response, "json", return_value=mock_device_code_data),
):
result = authenticator._get_device_code()
assert result == mock_device_code_data
mock_client.post.assert_called_once()
def test_poll_for_access_token(self, authenticator, mock_http_client):
"""Test polling for an access token."""
mock_client, mock_response = mock_http_client
mock_token_data = {"access_token": "mock-access-token"}
def test_get_access_token_refuses_device_code_login_in_worker_thread() -> None:
auth: Final = Authenticator()
with (
patch("builtins.open", side_effect=FileNotFoundError),
patch.object(auth, "_login") as login,
):
with ThreadPoolExecutor(max_workers=1) as pool:
with pytest.raises(GetAccessTokenError) as exc:
pool.submit(auth.get_access_token).result()
with (
patch(
"litellm.llms.github_copilot.authenticator.get_httpx_client",
return_value=mock_client,
),
patch.object(mock_response, "json", return_value=mock_token_data),
patch("time.sleep"),
):
result = authenticator._poll_for_access_token("mock-device-code")
assert result == "mock-access-token"
mock_client.post.assert_called_once()
assert exc.value.status_code == 401
assert "worker thread" in str(exc.value)
login.assert_not_called()
def test_login(self, authenticator):
"""Test the login process."""
mock_device_code_data = {
"device_code": "mock-device-code",
"user_code": "ABCD-EFGH",
"verification_uri": "https://github.com/login/device",
}
mock_token = "mock-access-token"
with (
patch.object(
authenticator, "_get_device_code", return_value=mock_device_code_data
),
patch.object(
authenticator, "_poll_for_access_token", return_value=mock_token
),
patch("builtins.print") as mock_print,
):
result = authenticator._login()
assert result == mock_token
authenticator._get_device_code.assert_called_once()
authenticator._poll_for_access_token.assert_called_once_with(
"mock-device-code"
)
mock_print.assert_called_once()
@pytest.mark.parametrize("encoding", ["utf-16", "utf-16-le", "utf-8-sig"])
def test_get_access_token_reads_saved_token_with_supported_encoding(
logged_in_token_dir: tuple[Path, str], encoding: str
) -> None:
token_dir, token = logged_in_token_dir
(token_dir / "access-token").write_bytes(f"{token}\n".encode(encoding))
def test_get_api_base_from_file(self, authenticator):
"""Test retrieving the API base endpoint from a file."""
mock_api_key_data = json.dumps(
assert Authenticator().get_access_token() == token
def test_get_access_token_refuses_undecodable_saved_token(
logged_in_token_dir: tuple[Path, str],
) -> None:
token_dir, _ = logged_in_token_dir
access_token_file: Final = token_dir / "access-token"
access_token_file.write_bytes(TRUNCATED_UTF_16)
with ThreadPoolExecutor(max_workers=1) as pool, pytest.raises(GetAccessTokenError) as exc:
pool.submit(Authenticator().get_access_token).result()
assert exc.value.status_code == 401
assert "worker thread" in str(exc.value)
assert access_token_file.read_bytes() == TRUNCATED_UTF_16
def test_get_access_token_retries_device_flow_three_times(
logged_in_token_dir: tuple[Path, str],
) -> None:
token_dir, _ = logged_in_token_dir
(token_dir / "access-token").unlink()
auth: Final = Authenticator()
with (
patch.object(
auth,
"_login",
side_effect=GetDeviceCodeError(message="device flow failed", status_code=400),
) as login,
pytest.raises(GetAccessTokenError, match="after 3 attempts"),
):
auth.get_access_token()
assert login.call_count == 3
def test_get_access_token_returns_login_token_when_file_write_fails(
logged_in_token_dir: tuple[Path, str],
) -> None:
token_dir, _ = logged_in_token_dir
(token_dir / "access-token").unlink()
auth: Final = Authenticator()
with (
patch.object(auth, "_login", return_value="github-access-token"),
patch("builtins.open", side_effect=(FileNotFoundError, PermissionError)),
):
assert auth.get_access_token() == "github-access-token"
@pytest.mark.parametrize(
("environment", "url", "body"),
[
(
{"GITHUB_COPILOT_DEVICE_CODE_URL": "https://custom.example/device"},
"https://custom.example/device",
{"client_id": "Iv1.b507a08c87ecfe98", "scope": "read:user"},
),
(
{"GITHUB_COPILOT_CLIENT_ID": "custom-client"},
DEVICE_CODE_URL,
{"client_id": "custom-client", "scope": "read:user"},
),
],
)
def test_get_device_code_uses_runtime_url_and_client_id(
environment: dict[str, str], url: str, body: dict[str, str]
) -> None:
client: Final = MagicMock()
response: Final = MagicMock()
response.json.return_value = {
"device_code": "dc",
"user_code": "UC",
"verification_uri": "https://github.com/login/device",
}
response.raise_for_status.return_value = None
client.post.return_value = response
with (
patch.dict(os.environ, environment),
patch("litellm.llms.github_copilot.authenticator.get_httpx_client", return_value=client),
):
assert Authenticator()._get_device_code()["device_code"] == "dc"
client.post.assert_called_once()
assert client.post.call_args.args[0] == url
assert client.post.call_args.kwargs["json"] == body
def test_get_device_code_sends_device_flow_request(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("GITHUB_COPILOT_DEVICE_CODE_URL", raising=False)
monkeypatch.delenv("GITHUB_COPILOT_CLIENT_ID", raising=False)
client: Final = MagicMock()
response: Final = MagicMock()
device_code: Final = {
"device_code": "device-code",
"user_code": "ABCD-EFGH",
"verification_uri": "https://github.com/login/device",
}
response.json.return_value = device_code
response.raise_for_status.return_value = None
client.post.return_value = response
with patch("litellm.llms.github_copilot.authenticator.get_httpx_client", return_value=client):
assert Authenticator()._get_device_code() == device_code
client.post.assert_called_once_with(
DEVICE_CODE_URL,
headers=github_api_headers(),
json={"client_id": "Iv1.b507a08c87ecfe98", "scope": "read:user"},
)
def test_poll_for_access_token_sends_device_code_and_grant_type(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("GITHUB_COPILOT_ACCESS_TOKEN_URL", raising=False)
monkeypatch.delenv("GITHUB_COPILOT_CLIENT_ID", raising=False)
client: Final = MagicMock()
response: Final = MagicMock()
response.json.return_value = {"access_token": "github-access-token"}
response.raise_for_status.return_value = None
client.post.return_value = response
with patch("litellm.llms.github_copilot.authenticator.get_httpx_client", return_value=client):
assert Authenticator()._poll_for_access_token("device-code") == "github-access-token"
client.post.assert_called_once_with(
"https://github.com/login/oauth/access_token",
headers=github_api_headers(),
json={
"client_id": "Iv1.b507a08c87ecfe98",
"device_code": "device-code",
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
},
)
def test_poll_for_access_token_uses_runtime_url_and_client_id() -> None:
client: Final = MagicMock()
response: Final = MagicMock()
response.json.return_value = {"access_token": "github-token"}
response.raise_for_status.return_value = None
client.post.return_value = response
with (
patch.dict(
os.environ,
{
"token": "mock-api-key",
"expires_at": (datetime.now() + timedelta(hours=1)).timestamp(),
"endpoints": {"api": "https://api.enterprise.githubcopilot.com"},
}
).encode()
with patch("builtins.open", mock_open(read_data=mock_api_key_data)):
api_base = authenticator.get_api_base()
assert api_base == "https://api.enterprise.githubcopilot.com"
"GITHUB_COPILOT_ACCESS_TOKEN_URL": "https://custom.example/token",
"GITHUB_COPILOT_CLIENT_ID": "custom-client",
},
),
patch("litellm.llms.github_copilot.authenticator.get_httpx_client", return_value=client),
patch("time.sleep"),
):
assert Authenticator()._poll_for_access_token("dc") == "github-token"
assert client.post.call_args.args[0] == "https://custom.example/token"
assert client.post.call_args.kwargs["json"]["client_id"] == "custom-client"
def test_get_device_code_with_custom_url(self, authenticator, mock_http_client):
"""GITHUB_COPILOT_DEVICE_CODE_URL env var must be used by _get_device_code at call time."""
mock_client, mock_response = mock_http_client
custom_url = "https://custom.example.com/device"
mock_response.json.return_value = {
"device_code": "dc",
"user_code": "UC",
"verification_uri": "https://example.com",
}
with (
patch.dict(os.environ, {"GITHUB_COPILOT_DEVICE_CODE_URL": custom_url}),
patch("litellm.llms.github_copilot.authenticator.get_httpx_client", return_value=mock_client),
):
authenticator._get_device_code()
assert mock_client.post.call_args[0][0] == custom_url
def test_get_device_code_with_custom_client_id(self, authenticator, mock_http_client):
"""GITHUB_COPILOT_CLIENT_ID env var must appear as client_id in the device-code request body."""
mock_client, mock_response = mock_http_client
custom_id = "custom_client_id"
mock_response.json.return_value = {
"device_code": "dc",
"user_code": "UC",
"verification_uri": "https://example.com",
}
with (
patch.dict(os.environ, {"GITHUB_COPILOT_CLIENT_ID": custom_id}),
patch("litellm.llms.github_copilot.authenticator.get_httpx_client", return_value=mock_client),
):
authenticator._get_device_code()
assert mock_client.post.call_args[1]["json"]["client_id"] == custom_id
def test_login_requests_device_code_and_polls_for_access_token() -> None:
auth: Final = Authenticator()
device_code: Final = {
"device_code": "device-code",
"user_code": "ABCD-EFGH",
"verification_uri": "https://github.com/login/device",
}
def test_poll_for_access_token_with_custom_url(self, authenticator, mock_http_client):
"""GITHUB_COPILOT_ACCESS_TOKEN_URL env var must be used by _poll_for_access_token at call time."""
mock_client, mock_response = mock_http_client
custom_url = "https://custom.example.com/token"
mock_response.json.return_value = {"access_token": "tok"}
with (
patch.dict(os.environ, {"GITHUB_COPILOT_ACCESS_TOKEN_URL": custom_url}),
patch("litellm.llms.github_copilot.authenticator.get_httpx_client", return_value=mock_client),
patch("time.sleep"),
):
authenticator._poll_for_access_token("dc")
assert mock_client.post.call_args[0][0] == custom_url
with (
patch.object(auth, "_get_device_code", return_value=device_code) as get_device_code,
patch.object(auth, "_poll_for_access_token", return_value="github-access-token") as poll,
patch("builtins.print") as print_message,
):
assert auth._login() == "github-access-token"
def test_poll_for_access_token_with_custom_client_id(self, authenticator, mock_http_client):
"""GITHUB_COPILOT_CLIENT_ID env var must appear as client_id in the polling request body."""
mock_client, mock_response = mock_http_client
custom_id = "custom_client_id"
mock_response.json.return_value = {"access_token": "tok"}
with (
patch.dict(os.environ, {"GITHUB_COPILOT_CLIENT_ID": custom_id}),
patch("litellm.llms.github_copilot.authenticator.get_httpx_client", return_value=mock_client),
patch("time.sleep"),
):
authenticator._poll_for_access_token("dc")
assert mock_client.post.call_args[1]["json"]["client_id"] == custom_id
def test_refresh_api_key_with_custom_url(self, authenticator, mock_http_client):
"""GITHUB_COPILOT_API_KEY_URL env var must be used by _refresh_api_key at call time."""
mock_client, mock_response = mock_http_client
custom_url = "https://custom.example.com/api-key"
mock_response.json.return_value = {"token": "api-tok", "expires_at": 9999999999}
with (
patch.dict(os.environ, {"GITHUB_COPILOT_API_KEY_URL": custom_url}),
patch("litellm.llms.github_copilot.authenticator.get_httpx_client", return_value=mock_client),
patch.object(authenticator, "get_access_token", return_value="access-tok"),
):
authenticator._refresh_api_key()
assert mock_client.get.call_args[0][0] == custom_url
get_device_code.assert_called_once()
poll.assert_called_once_with("device-code")
print_message.assert_called_once_with(
"Please visit https://github.com/login/device and enter code ABCD-EFGH to authenticate.",
flush=True,
)

View file

@ -1,29 +1,17 @@
import asyncio
import json
from datetime import datetime, timedelta
from typing import AsyncGenerator
from unittest.mock import AsyncMock, MagicMock, mock_open, patch
import pytest
from unittest.mock import MagicMock, patch
import httpx
from respx import MockRouter
import pytest
import litellm
# Import at the top to make the patch work correctly
import litellm.llms.github_copilot.chat.transformation
from litellm import Choices, Message, ModelResponse, Usage, acompletion, completion
from litellm import ModelResponse, completion
from litellm.exceptions import AuthenticationError
from litellm.llms.github_copilot.authenticator import Authenticator
from litellm.llms.github_copilot.chat.transformation import GithubCopilotConfig
from litellm.llms.github_copilot.common_utils import (
APIKeyExpiredError,
GetAccessTokenError,
GetAPIKeyError,
GetDeviceCodeError,
RefreshAPIKeyError,
)
@ -37,9 +25,7 @@ def test_github_copilot_config_get_openai_compatible_provider_info():
config.authenticator = MagicMock()
config.authenticator.get_api_key.return_value = mock_api_key
# Test with dynamic endpoint
config.authenticator.get_api_base.return_value = (
"https://api.enterprise.githubcopilot.com"
)
config.authenticator.get_api_base.return_value = "https://api.enterprise.githubcopilot.com"
# Test with default values
model = "github_copilot/gpt-4"
@ -73,7 +59,7 @@ def test_github_copilot_config_get_openai_compatible_provider_info():
assert api_base == "https://api.githubcopilot.com"
# Test with authentication failure
config.authenticator.get_api_key.side_effect = GetAPIKeyError(
config.authenticator.get_api_key.side_effect = GetAccessTokenError(
message="Failed to get API key",
status_code=401,
)
@ -145,7 +131,6 @@ def test_completion_github_copilot_mock_response(
def test_transform_messages_disable_copilot_system_to_assistant(monkeypatch):
"""Test that system messages are converted to assistant unless disable_copilot_system_to_assistant is True."""
import litellm
from litellm.llms.github_copilot.chat.transformation import GithubCopilotConfig
# Save original value
@ -365,10 +350,6 @@ def test_x_initiator_header_system_only_messages():
assert headers["X-Initiator"] == "user"
def test_copilot_vision_request_header_with_image():
"""Test that Copilot-Vision-Request header is added when messages contain images"""
config = GithubCopilotConfig()
@ -693,13 +674,8 @@ class TestGithubCopilotTransformResponse:
assert result.choices[0].message.tool_calls is not None
assert len(result.choices[0].message.tool_calls) == 1
assert result.choices[0].message.tool_calls[0]["id"] == "toolu_01ABC"
assert (
result.choices[0].message.tool_calls[0]["function"]["name"] == "get_weather"
)
assert (
'"Boston, MA"'
in result.choices[0].message.tool_calls[0]["function"]["arguments"]
)
assert result.choices[0].message.tool_calls[0]["function"]["name"] == "get_weather"
assert '"Boston, MA"' in result.choices[0].message.tool_calls[0]["function"]["arguments"]
def test_transform_response_anthropic_native_multiple_text_blocks(self):
"""All text blocks must be concatenated, not only the first."""
@ -867,12 +843,8 @@ class TestGithubCopilotTransformParsedResponseDict:
@patch("litellm.llms.openai.openai.OpenAIChatCompletion._get_openai_client")
@patch(
"litellm.llms.openai.openai.OpenAIChatCompletion.make_sync_openai_chat_completion_request"
)
def test_openai_handler_repairs_github_copilot_empty_choices(
mock_request, mock_get_client
):
@patch("litellm.llms.openai.openai.OpenAIChatCompletion.make_sync_openai_chat_completion_request")
def test_openai_handler_repairs_github_copilot_empty_choices(mock_request, mock_get_client):
"""
The OpenAI SDK handler calls convert_to_model_response_object directly on the
SDK's parsed output, bypassing transform_response. convert raises APIError on
@ -929,7 +901,10 @@ def test_validate_environment_uses_per_user_session_and_skips_authenticator():
session = GithubCopilotUserSession(token="user-copilot-token", api_base="https://api.githubcopilot.com")
headers = config.validate_environment(
headers={}, model="github_copilot/gpt-4o", messages=[], optional_params={},
headers={},
model="github_copilot/gpt-4o",
messages=[],
optional_params={},
litellm_params={"github_copilot_user_session": session},
)
assert headers["Authorization"] == "Bearer user-copilot-token"
@ -946,9 +921,7 @@ def test_transform_response_carries_upstream_usage():
"object": "chat.completion",
"created": 1,
"model": "github_copilot/gpt-4o",
"choices": [
{"index": 0, "message": {"role": "assistant", "content": "hi"}, "finish_reason": "stop"}
],
"choices": [{"index": 0, "message": {"role": "assistant", "content": "hi"}, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 11, "completion_tokens": 7, "total_tokens": 18},
},
)

View file

@ -32,28 +32,25 @@ def test_normalized_github_host_accepts_only_bare_supported_hosts(raw: object, e
def test_github_com_endpoints_read_url_overrides_at_call_time(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITHUB_COPILOT_DEVICE_CODE_URL", "https://device.example/code")
monkeypatch.setenv("GITHUB_COPILOT_ACCESS_TOKEN_URL", "https://device.example/token")
monkeypatch.setenv("GITHUB_COPILOT_API_KEY_URL", "https://api.example/copilot")
monkeypatch.setenv("GITHUB_COPILOT_CLIENT_ID", "custom-client")
endpoints: Final = github_host_endpoints("github.com")
assert endpoints.device_code_url == "https://device.example/code"
assert endpoints.access_token_url == "https://device.example/token"
assert endpoints.api_key_url == "https://api.example/copilot"
assert endpoints.copilot_user_url == "https://api.github.com/copilot_internal/user"
assert endpoints.client_id == "custom-client"
def test_ghe_endpoints_ignore_github_com_url_overrides(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITHUB_COPILOT_DEVICE_CODE_URL", "https://device.example/code")
monkeypatch.setenv("GITHUB_COPILOT_ACCESS_TOKEN_URL", "https://device.example/token")
monkeypatch.setenv("GITHUB_COPILOT_API_KEY_URL", "https://api.example/copilot")
monkeypatch.setenv("GITHUB_COPILOT_CLIENT_ID", "custom-client")
endpoints: Final = github_host_endpoints("octo.ghe.com")
assert endpoints.device_code_url == "https://octo.ghe.com/login/device/code"
assert endpoints.access_token_url == "https://octo.ghe.com/login/oauth/access_token"
assert endpoints.api_key_url == "https://api.octo.ghe.com/copilot_internal/v2/token"
assert endpoints.copilot_user_url == "https://api.octo.ghe.com/copilot_internal/user"
assert endpoints.user_url == "https://api.octo.ghe.com/user"
assert endpoints.default_copilot_api_base == "https://copilot-api.octo.ghe.com"

View file

@ -12,8 +12,9 @@ import httpx
import pytest
import litellm
from litellm.constants import GITHUB_COPILOT_DEFAULT_HOST
from litellm.constants import GITHUB_COPILOT_DEFAULT_HOST, GITHUB_COPILOT_USER_SESSION_TTL_SECONDS
from litellm.exceptions import (
APIConnectionError,
AuthenticationError,
BadRequestError,
CallerCredentialAuthenticationError,
@ -21,27 +22,27 @@ from litellm.exceptions import (
ServiceUnavailableError,
)
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.llms.github_copilot.github_host import github_host_endpoints
from litellm.llms.github_copilot.per_user_auth import (
_SESSION_CACHE,
GITHUB_COPILOT_USER_SESSION_KWARG_KEY,
GithubCopilotUserSession,
_session_cache_key,
aattach_github_copilot_user_session,
aexchange_github_token,
acheck_copilot_seat,
apoll_device_flow,
aresolve_copilot_user_session,
astart_device_flow,
attach_github_copilot_user_session,
evict_copilot_user_session,
exchange_github_token,
github_copilot_host_endpoints,
github_copilot_auth_mode,
github_copilot_user_session_from,
resolve_copilot_user_session,
validated_copilot_api_base,
)
from litellm.llms.github_copilot.github_host import github_host_endpoints
from litellm.types.utils import CredentialItem, ModelResponse
EXCHANGE_URL = "https://api.github.com/copilot_internal/v2/token"
COPILOT_USER_URL = "https://api.github.com/copilot_internal/user"
DEVICE_CODE_URL = "https://github.com/login/device/code"
ACCESS_TOKEN_URL = "https://github.com/login/oauth/access_token"
_GITHUB_ENDPOINTS: Final = github_host_endpoints("github.com")
@ -62,10 +63,8 @@ def _credential(
)
def _exchange_payload(token="copilot-token", expires_at=None, api_base="https://api.githubcopilot.com"):
def _user_payload(api_base="https://api.githubcopilot.com"):
return {
"token": token,
"expires_at": expires_at if expires_at is not None else int(time.time()) + 3600,
"endpoints": {"api": api_base},
}
@ -92,10 +91,10 @@ def github_http(respond: Callable[[httpx.Request], httpx.Response]):
yield requests
def _exchange_responder(payload=None, status=200):
def _user_responder(payload=None, status=200):
def respond(request: httpx.Request) -> httpx.Response:
if str(request.url) == EXCHANGE_URL:
return httpx.Response(status, json=payload or _exchange_payload(), request=request)
if str(request.url) == COPILOT_USER_URL:
return httpx.Response(status, json=payload or _user_payload(), request=request)
pytest.fail(f"unexpected request to {request.url}")
return respond
@ -170,17 +169,17 @@ def test_validated_copilot_api_base(raw, expected):
"https://copilot-api.octo.ghe.com:8443",
],
)
def test_ghe_exchange_response_must_stay_on_its_tenant_host(raw: object) -> None:
def test_ghe_user_response_must_stay_on_its_tenant_host(raw: object) -> None:
endpoints: Final = github_host_endpoints("octo.ghe.com")
assert validated_copilot_api_base(raw, endpoints) == "https://copilot-api.octo.ghe.com"
def test_github_com_exchange_rejects_a_ghe_copilot_host() -> None:
def test_github_com_user_response_rejects_a_ghe_copilot_host() -> None:
endpoints: Final = github_host_endpoints("github.com")
assert validated_copilot_api_base("https://copilot-api.octo.ghe.com", endpoints) == "https://api.githubcopilot.com"
def test_ghe_exchange_uses_stored_host_not_request_kwargs():
def test_ghe_user_lookup_uses_stored_host_not_request_kwargs():
credential: Final = CredentialItem(
credential_name="copilot-cred",
credential_values={"github_copilot_auth_type": "per_user_oauth", "github_copilot_host": "octo.ghe.com"},
@ -188,8 +187,8 @@ def test_ghe_exchange_uses_stored_host_not_request_kwargs():
)
def respond(request: httpx.Request) -> httpx.Response:
assert str(request.url) == "https://api.octo.ghe.com/copilot_internal/v2/token"
return httpx.Response(200, json=_exchange_payload(api_base="https://copilot-api.octo.ghe.com"), request=request)
assert str(request.url) == "https://api.octo.ghe.com/copilot_internal/user"
return httpx.Response(200, json=_user_payload(api_base="https://copilot-api.octo.ghe.com"), request=request)
with patch.object(litellm, "credential_list", [credential]), github_http(respond) as requests:
kwargs: Final = _kwargs_with_connection(github_host="octo.ghe.com")
@ -201,7 +200,7 @@ def test_ghe_exchange_uses_stored_host_not_request_kwargs():
assert len(requests) == 1
def test_invalid_stored_host_fails_before_exchange():
def test_invalid_stored_host_fails_before_user_lookup():
credential: Final = CredentialItem(
credential_name="copilot-cred",
credential_values={"github_copilot_auth_type": "per_user_oauth", "github_copilot_host": "https://octo.ghe.com"},
@ -225,35 +224,36 @@ def test_session_repr_and_str_hide_the_token(caplog):
assert "secret-copilot-token" not in caplog.text
def test_exchange_returns_session_and_caches_per_user():
with github_http(_exchange_responder()) as requests:
s1 = exchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
def test_user_lookup_returns_github_token_and_caches_per_user():
api_base: Final = "https://tenant.githubcopilot.com"
with github_http(_user_responder(payload=_user_payload(api_base=api_base))) as requests:
s1 = resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert len(requests) == 1
assert str(requests[0].url) == COPILOT_USER_URL
assert requests[0].headers["Authorization"] == "token gh-token"
s2 = exchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
s2 = resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert len(requests) == 1
assert s2.token == s1.token == "copilot-token"
# different user -> separate exchange
exchange_github_token("user-b", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert s2.token == s1.token == "gh-token"
assert s1.api_base == api_base
resolve_copilot_user_session("user-b", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert len(requests) == 2
# different github token -> separate exchange
exchange_github_token("user-a", "gh-token-2", "copilot-cred", _GITHUB_ENDPOINTS)
resolve_copilot_user_session("user-a", "gh-token-2", "copilot-cred", _GITHUB_ENDPOINTS)
assert len(requests) == 3
def test_cached_session_is_not_reused_after_github_token_changes():
with github_http(_exchange_responder()) as requests:
exchange_github_token("user-a", "gh-token-a", "copilot-cred", _GITHUB_ENDPOINTS)
exchange_github_token("user-a", "gh-token-b", "copilot-cred", _GITHUB_ENDPOINTS)
with github_http(_user_responder()) as requests:
resolve_copilot_user_session("user-a", "gh-token-a", "copilot-cred", _GITHUB_ENDPOINTS)
resolve_copilot_user_session("user-a", "gh-token-b", "copilot-cred", _GITHUB_ENDPOINTS)
assert [request.headers["Authorization"] for request in requests] == ["token gh-token-a", "token gh-token-b"]
@pytest.mark.parametrize("status", [401, 403, 404])
def test_exchange_auth_failures_raise_caller_auth_error_and_evict(status):
with github_http(_exchange_responder(status=status)):
def test_user_lookup_auth_failures_raise_caller_auth_error_and_evict(status):
with github_http(_user_responder(status=status)):
with pytest.raises(CallerCredentialAuthenticationError) as exc:
exchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert "Reconnect GitHub Copilot for credential 'copilot-cred' in the LiteLLM UI (LLM Credentials)" in str(
exc.value
)
@ -262,52 +262,135 @@ def test_exchange_auth_failures_raise_caller_auth_error_and_evict(status):
)
def test_exchange_429_raises_caller_rate_limit():
with github_http(_exchange_responder(status=429)):
def test_user_lookup_429_raises_caller_rate_limit():
with github_http(_user_responder(status=429)):
with pytest.raises(CallerCredentialRateLimitError):
exchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
def test_exchange_other_failure_raises_service_unavailable_without_body():
with github_http(_exchange_responder(status=500, payload={"secret": "body"})):
def test_user_lookup_other_failure_raises_service_unavailable_without_body():
with github_http(_user_responder(status=500, payload={"secret": "body"})):
with pytest.raises(ServiceUnavailableError) as exc:
exchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert "body" not in str(exc.value)
def test_exchange_expiry_margin_refetches_near_expiry():
with github_http(_exchange_responder(_exchange_payload(expires_at=int(time.time()) + 30))) as requests:
exchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
# ttl under the 60s safety margin -> not cached, second call refetches
exchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
def test_session_cache_uses_fixed_ttl():
with (
patch("litellm.llms.github_copilot.per_user_auth.time.time", return_value=1000) as clock,
github_http(_user_responder()) as requests,
):
resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
clock.return_value = 1000 + GITHUB_COPILOT_USER_SESSION_TTL_SECONDS - 1
resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert len(requests) == 1
clock.return_value = 1000 + GITHUB_COPILOT_USER_SESSION_TTL_SECONDS + 1
resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert len(requests) == 2
@pytest.mark.asyncio
async def test_aexchange_single_flight_one_http_call():
async def test_acheck_copilot_seat_rejects_models_403(per_user_credential):
def respond(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=_exchange_payload(), request=request)
if str(request.url) == COPILOT_USER_URL:
return httpx.Response(200, json=_user_payload(), request=request)
if str(request.url) == "https://api.githubcopilot.com/models":
return httpx.Response(403, request=request)
pytest.fail(f"unexpected request to {request.url}")
with github_http(respond) as requests:
s1, s2 = await asyncio.gather(
aexchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS),
aexchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS),
)
assert len(requests) == 1
assert s1.token == s2.token == "copilot-token"
with pytest.raises(CallerCredentialAuthenticationError):
await acheck_copilot_seat("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert [str(request.url) for request in requests] == [
COPILOT_USER_URL,
"https://api.githubcopilot.com/models",
]
assert requests[1].headers["Authorization"] == "Bearer gh-token"
@pytest.mark.asyncio
async def test_aexchange_401_raises_and_evicts():
with github_http(_exchange_responder(status=401)):
@pytest.mark.parametrize(
("status", "error_type"),
[(429, CallerCredentialRateLimitError), (500, ServiceUnavailableError)],
)
async def test_acheck_copilot_seat_maps_models_failures(
per_user_credential,
status: int,
error_type: type[Exception],
) -> None:
def respond(request: httpx.Request) -> httpx.Response:
if str(request.url) == COPILOT_USER_URL:
return httpx.Response(200, json=_user_payload(), request=request)
if str(request.url) == "https://api.githubcopilot.com/models":
return httpx.Response(status, request=request)
pytest.fail(f"unexpected request to {request.url}")
with github_http(respond):
with pytest.raises(error_type):
await acheck_copilot_seat("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
@pytest.mark.asyncio
async def test_acheck_copilot_seat_maps_models_request_errors(per_user_credential) -> None:
def respond(request: httpx.Request) -> httpx.Response:
if str(request.url) == COPILOT_USER_URL:
return httpx.Response(200, json=_user_payload(), request=request)
if str(request.url) == "https://api.githubcopilot.com/models":
raise httpx.ConnectError("connection failed", request=request)
pytest.fail(f"unexpected request to {request.url}")
with github_http(respond):
with pytest.raises(APIConnectionError, match="models check request failed"):
await acheck_copilot_seat("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
@pytest.mark.asyncio
async def test_acheck_copilot_seat_passes_when_models_are_usable(per_user_credential):
def respond(request: httpx.Request) -> httpx.Response:
if str(request.url) == COPILOT_USER_URL:
return httpx.Response(200, json=_user_payload(), request=request)
if str(request.url) == "https://api.githubcopilot.com/models":
return httpx.Response(200, json={"data": []}, request=request)
pytest.fail(f"unexpected request to {request.url}")
with github_http(respond) as requests:
session = await acheck_copilot_seat("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert session.token == "gh-token"
assert session.api_base == "https://api.githubcopilot.com"
assert [str(request.url) for request in requests] == [
COPILOT_USER_URL,
"https://api.githubcopilot.com/models",
]
@pytest.mark.asyncio
async def test_aresolve_single_flight_one_http_call():
def respond(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=_user_payload(), request=request)
with github_http(respond) as requests:
s1, s2 = await asyncio.gather(
aresolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS),
aresolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS),
)
assert len(requests) == 1
assert s1.token == s2.token == "gh-token"
@pytest.mark.asyncio
async def test_aresolve_401_raises_and_evicts():
with github_http(_user_responder(status=401)):
with pytest.raises(CallerCredentialAuthenticationError):
await aexchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
await aresolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert (
_SESSION_CACHE.get_cache(_session_cache_key("user-a", "copilot-cred", GITHUB_COPILOT_DEFAULT_HOST)) is None
)
ACCOUNT_URL = "https://api.github.com/copilot_internal/user"
EXCHANGE_URL: Final = "https://api.github.com/copilot_internal/v2/token"
_CUSTOM_CLIENT_ID: Final = "Ov23liCustomOrgApp"
_ACCOUNT_API_BASE: Final = "https://api.business.githubcopilot.com"
@ -327,17 +410,20 @@ def _account_responder(payload=None, status=200, model_payload=None):
return respond
@pytest.fixture
def custom_client_id(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("GITHUB_COPILOT_CLIENT_ID", _CUSTOM_CLIENT_ID)
@pytest.fixture(params=[None, _CUSTOM_CLIENT_ID], ids=["default-client-id", "custom-client-id"])
def copilot_client_id(request, monkeypatch: pytest.MonkeyPatch):
if request.param is None:
monkeypatch.delenv("GITHUB_COPILOT_CLIENT_ID", raising=False)
return
monkeypatch.setenv("GITHUB_COPILOT_CLIENT_ID", request.param)
@pytest.mark.asyncio
async def test_custom_client_id_uses_the_github_token_without_the_copilot_token_swap(custom_client_id):
async def test_user_lookup_uses_the_github_token_for_any_client_id(copilot_client_id):
endpoints: Final = github_host_endpoints("github.com")
with github_http(_account_responder()) as requests:
first = await aexchange_github_token("user-a", "gho_org_app", "copilot-cred", endpoints)
second = await aexchange_github_token("user-a", "gho_org_app", "copilot-cred", endpoints)
first = await aresolve_copilot_user_session("user-a", "gho_org_app", "copilot-cred", endpoints)
second = await aresolve_copilot_user_session("user-a", "gho_org_app", "copilot-cred", endpoints)
assert [str(r.url) for r in requests] == [ACCOUNT_URL]
assert requests[0].headers["Authorization"] == "token gho_org_app"
@ -353,22 +439,24 @@ async def test_custom_client_id_uses_the_github_token_without_the_copilot_token_
_account_payload(access_type_sku="no_access"),
],
)
def test_custom_client_id_without_a_copilot_seat_raises_reconnect(custom_client_id, payload):
def test_user_lookup_without_a_copilot_seat_raises_reconnect(copilot_client_id, payload):
with github_http(_account_responder(payload)):
with pytest.raises(CallerCredentialAuthenticationError):
exchange_github_token("user-a", "gho_org_app", "copilot-cred", github_host_endpoints("github.com"))
resolve_copilot_user_session("user-a", "gho_org_app", "copilot-cred", github_host_endpoints("github.com"))
@pytest.mark.parametrize("status", [401, 403, 404])
def test_custom_client_id_rejected_account_lookup_raises_reconnect(custom_client_id, status):
def test_rejected_user_lookup_raises_reconnect(copilot_client_id, status):
with github_http(_account_responder(status=status)):
with pytest.raises(CallerCredentialAuthenticationError):
exchange_github_token("user-a", "gho_org_app", "copilot-cred", github_host_endpoints("github.com"))
resolve_copilot_user_session("user-a", "gho_org_app", "copilot-cred", github_host_endpoints("github.com"))
def test_custom_client_id_account_api_base_must_stay_on_copilot_hosts(custom_client_id):
def test_user_lookup_api_base_must_stay_on_copilot_hosts(copilot_client_id):
with github_http(_account_responder(_account_payload(api_base="https://evil.example"))):
session = exchange_github_token("user-a", "gho_org_app", "copilot-cred", github_host_endpoints("github.com"))
session = resolve_copilot_user_session(
"user-a", "gho_org_app", "copilot-cred", github_host_endpoints("github.com")
)
assert session.api_base == "https://api.githubcopilot.com"
@ -392,16 +480,16 @@ def _kwargs_with_connection(
@pytest.mark.asyncio
async def test_aattach_injects_session_into_kwargs(per_user_credential):
with github_http(_exchange_responder()):
with github_http(_user_responder()):
kwargs = _kwargs_with_connection()
await aattach_github_copilot_user_session(kwargs)
session = github_copilot_user_session_from(kwargs)
assert session is not None
assert session.token == "copilot-token"
assert session.token == "gho_user_token"
def test_attach_sync_injects_session(per_user_credential):
with github_http(_exchange_responder()):
with github_http(_user_responder()):
kwargs = _kwargs_with_connection()
attach_github_copilot_user_session(kwargs)
assert isinstance(kwargs[GITHUB_COPILOT_USER_SESSION_KWARG_KEY], GithubCopilotUserSession)
@ -424,7 +512,7 @@ def test_attach_rejects_connection_from_a_different_github_host(
):
with patch.object(litellm, "credential_list", [_credential(github_host=credential_host)]):
kwargs: Final = _kwargs_with_connection(github_host=connection_host)
with github_http(lambda request: pytest.fail(f"unexpected token exchange at {request.url}")) as requests:
with github_http(lambda request: pytest.fail(f"unexpected user lookup at {request.url}")) as requests:
with pytest.raises(CallerCredentialAuthenticationError) as exc:
attach_github_copilot_user_session(kwargs)
assert "Connect GitHub Copilot for credential 'copilot-cred'" in str(exc.value)
@ -432,11 +520,11 @@ def test_attach_rejects_connection_from_a_different_github_host(
def test_legacy_connection_without_host_uses_github_com(per_user_credential):
with github_http(_exchange_responder()) as requests:
with github_http(_user_responder()) as requests:
kwargs: Final = _kwargs_with_connection(github_host=None)
attach_github_copilot_user_session(kwargs)
assert len(requests) == 1
assert str(requests[0].url) == EXCHANGE_URL
assert str(requests[0].url) == COPILOT_USER_URL
def test_attach_shared_mode_does_nothing_without_connection():
@ -447,10 +535,10 @@ def test_attach_shared_mode_does_nothing_without_connection():
def test_evict_drops_cached_session():
with github_http(_exchange_responder()) as requests:
exchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
with github_http(_user_responder()) as requests:
resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
evict_copilot_user_session("user-a", "copilot-cred", _GITHUB_ENDPOINTS.host)
exchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
assert len(requests) == 2
@ -559,7 +647,7 @@ def test_two_per_user_calls_with_identical_bodies_both_reach_upstream( # test-q
monkeypatch.setattr(litellm, "cache", Cache())
with (
github_http(_exchange_responder()),
github_http(_user_responder()),
patch(
"litellm.main.openai_chat_completions.completion",
return_value=_chat_completion_response(),
@ -584,7 +672,7 @@ async def test_two_per_user_async_calls_with_identical_bodies_both_reach_upstrea
monkeypatch.setattr(litellm, "cache", Cache())
with (
github_http(_exchange_responder()),
github_http(_user_responder()),
patch(
"litellm.main.openai_chat_completions.acompletion",
new=AsyncMock(return_value=_chat_completion_response()),
@ -605,27 +693,19 @@ def test_shared_mode_second_identical_call_hits_response_cache( # test-quality-
):
"""Control: shared device-login github_copilot keeps caching exactly as before, so a second
identical call does not touch the upstream."""
import json
from litellm.caching import Cache
# a mounted shared login: api-key.json with a far-future token, no device flow needed
(tmp_path / "api-key.json").write_text(
json.dumps(
{
"token": "shared-copilot-token",
"expires_at": 4102444800,
"endpoints": {"api": "https://api.githubcopilot.com"},
}
)
)
(tmp_path / "access-token").write_text("shared-copilot-token")
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(tmp_path))
monkeypatch.setattr(litellm, "cache", Cache())
with patch(
"litellm.main.openai_chat_completions.completion",
return_value=_chat_completion_response(),
) as mock_completion:
with (
github_http(_user_responder()),
patch(
"litellm.main.openai_chat_completions.completion",
return_value=_chat_completion_response(),
) as mock_completion,
):
call = {
"model": "github_copilot/gpt-4o",
"messages": [{"role": "user", "content": "hi"}],
@ -679,8 +759,8 @@ def _anthropic_messages_payload():
def _copilot_surface_responder(model_payload):
def respond(request: httpx.Request) -> httpx.Response:
if str(request.url) == EXCHANGE_URL:
return httpx.Response(200, json=_exchange_payload(), request=request)
if str(request.url) == COPILOT_USER_URL:
return httpx.Response(200, json=_user_payload(), request=request)
if request.url.host.endswith("githubcopilot.com"):
return httpx.Response(200, json=model_payload, request=request)
pytest.fail(f"unexpected request to {request.url}")
@ -696,19 +776,19 @@ def _no_authenticator():
)
def _assert_exchange_then_model_call(requests, github_token, model_host="https://api.githubcopilot.com"):
exchange = [r for r in requests if str(r.url) == EXCHANGE_URL]
def _assert_user_lookup_then_model_call(requests, github_token, model_host="https://api.githubcopilot.com"):
lookup = [r for r in requests if str(r.url) == COPILOT_USER_URL]
upstream = [r for r in requests if r.url.host.endswith("githubcopilot.com")]
assert exchange, "expected the GitHub token exchange request"
assert lookup, "expected the GitHub user lookup request"
assert upstream, "expected the model request to the Copilot host"
assert exchange[0].headers["Authorization"] == f"token {github_token}"
assert upstream[0].headers["Authorization"].startswith("Bearer ")
assert lookup[0].headers["Authorization"] == f"token {github_token}"
assert upstream[0].headers["Authorization"] == f"Bearer {github_token}"
assert upstream[0].url.host == httpx.URL(model_host).host
def test_e2e_completion_per_user_credential(per_user_credential):
with (
github_http(_exchange_responder()) as requests,
github_http(_user_responder()) as requests,
_no_authenticator(),
patch(
"litellm.main.openai_chat_completions.completion",
@ -721,13 +801,13 @@ def test_e2e_completion_per_user_credential(per_user_credential):
litellm_credential_name="copilot-cred",
secret_fields=_secret_fields("user-a", "gho_a"),
)
assert len(requests) == 1 and str(requests[0].url) == EXCHANGE_URL
assert len(requests) == 1 and str(requests[0].url) == COPILOT_USER_URL
assert requests[0].headers["Authorization"] == "token gho_a"
call_kwargs = mock_completion.call_args.kwargs
assert call_kwargs["api_key"] == "copilot-token"
assert call_kwargs["api_key"] == "gho_a"
assert call_kwargs["api_base"] == "https://api.githubcopilot.com"
extra_headers = call_kwargs["optional_params"]["extra_headers"]
assert extra_headers["Authorization"] == "Bearer copilot-token"
assert extra_headers["Authorization"] == "Bearer gho_a"
assert response.usage.total_tokens > 0
@ -735,7 +815,7 @@ def test_e2e_completion_per_user_session_token_wins_over_caller_authorization(pe
"""The session token reaches the wire even when the caller passes their own
Authorization in extra_headers (chat dispatches through _complete_custom_openai)."""
with (
github_http(_exchange_responder()),
github_http(_user_responder()),
_no_authenticator(),
patch(
"litellm.main.openai_chat_completions.completion",
@ -750,17 +830,17 @@ def test_e2e_completion_per_user_session_token_wins_over_caller_authorization(pe
extra_headers={"Authorization": "Bearer caller-token", "x-custom": "keep"},
)
extra_headers = mock_completion.call_args.kwargs["optional_params"]["extra_headers"]
assert extra_headers["Authorization"] == "Bearer copilot-token"
assert extra_headers["Authorization"] == "Bearer gho_a"
assert extra_headers["x-custom"] == "keep"
_EVIL_AUTH_HEADERS = {"authorization": "Bearer caller-evil", "x-keep": "1"}
def _assert_session_token_on_the_wire(requests):
def _assert_session_token_on_the_wire(requests, github_token="gho_a"):
upstream = [r for r in requests if r.url.host.endswith("githubcopilot.com")]
assert upstream, "expected the model request to the Copilot host"
assert upstream[0].headers["authorization"] == "Bearer copilot-token"
assert upstream[0].headers["authorization"] == f"Bearer {github_token}"
def test_e2e_completion_per_user_caller_authorization_cannot_override_session(per_user_credential):
@ -769,7 +849,7 @@ def test_e2e_completion_per_user_caller_authorization_cannot_override_session(pe
handed to openai_chat_completions.completion."""
caller_headers: dict[str, str] = dict(_EVIL_AUTH_HEADERS)
with (
github_http(_exchange_responder()),
github_http(_user_responder()),
_no_authenticator(),
patch(
"litellm.main.openai_chat_completions.completion",
@ -784,7 +864,7 @@ def test_e2e_completion_per_user_caller_authorization_cannot_override_session(pe
extra_headers=caller_headers,
)
extra_headers = mock_completion.call_args.kwargs["optional_params"]["extra_headers"]
assert extra_headers["Authorization"] == "Bearer copilot-token"
assert extra_headers["Authorization"] == "Bearer gho_a"
assert "authorization" not in extra_headers
assert extra_headers["x-keep"] == "1"
assert caller_headers == dict(_EVIL_AUTH_HEADERS), "the caller's dict must not be mutated"
@ -805,9 +885,7 @@ def test_e2e_embedding_per_user_caller_authorization_cannot_override_session(per
_assert_session_token_on_the_wire(requests)
def test_e2e_custom_client_id_sends_the_github_token_to_the_accounts_copilot_host(
per_user_credential, custom_client_id
):
def test_e2e_per_user_sends_the_github_token_to_the_accounts_copilot_host(per_user_credential, copilot_client_id):
with (
github_http(_account_responder(model_payload=_embedding_response_payload())) as requests,
_no_authenticator(),
@ -869,7 +947,7 @@ def test_e2e_embedding_per_user_credential(per_user_credential):
litellm_credential_name="copilot-cred",
secret_fields=_secret_fields("user-a", "gho_a"),
)
_assert_exchange_then_model_call(requests, "gho_a")
_assert_user_lookup_then_model_call(requests, "gho_a")
assert requests[-1].url.path == "/embeddings"
assert response.usage.prompt_tokens == 4
@ -885,7 +963,7 @@ def test_e2e_responses_per_user_credential(per_user_credential):
litellm_credential_name="copilot-cred",
secret_fields=_secret_fields("user-a", "gho_a"),
)
_assert_exchange_then_model_call(requests, "gho_a")
_assert_user_lookup_then_model_call(requests, "gho_a")
assert requests[-1].url.path == "/responses"
assert response.usage.total_tokens == 6
@ -903,7 +981,7 @@ async def test_e2e_anthropic_messages_per_user_credential(per_user_credential):
litellm_credential_name="copilot-cred",
secret_fields=_secret_fields("user-a", "gho_a"),
)
_assert_exchange_then_model_call(requests, "gho_a")
_assert_user_lookup_then_model_call(requests, "gho_a")
assert requests[-1].url.path == "/v1/messages"
assert response["usage"]["input_tokens"] == 4
@ -954,26 +1032,19 @@ async def test_e2e_anthropic_messages_not_connected_raises_connect_401(per_user_
def _shared_login(tmp_path, monkeypatch):
import json
(tmp_path / "api-key.json").write_text(
json.dumps(
{
"token": "shared-copilot-token",
"expires_at": 4102444800,
"endpoints": {"api": "https://api.githubcopilot.com"},
}
)
)
(tmp_path / "access-token").write_text("shared-copilot-token")
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(tmp_path))
def test_e2e_completion_shared_mode_still_uses_authenticator(tmp_path, monkeypatch):
_shared_login(tmp_path, monkeypatch)
with patch(
"litellm.main.openai_chat_completions.completion",
return_value=_chat_completion_response(),
) as mock_completion:
with (
github_http(_user_responder()),
patch(
"litellm.main.openai_chat_completions.completion",
return_value=_chat_completion_response(),
) as mock_completion,
):
response = litellm.completion(
model="github_copilot/gpt-4o",
messages=[{"role": "user", "content": "hi"}],
@ -1007,45 +1078,37 @@ def test_e2e_responses_shared_mode_still_uses_authenticator(tmp_path, monkeypatc
@pytest.mark.asyncio
async def test_e2e_anthropic_messages_shared_mode_still_uses_authenticator(tmp_path, monkeypatch):
# The shared Authenticator's access-token refresh refuses to run inside an event
# loop, so on this async surface the mounted-login path can't reach it; patch the
# shared-login edge to emulate a mounted api-key.json.
async def test_e2e_anthropic_messages_shared_mode_still_uses_authenticator():
from litellm.llms.github_copilot.authenticator import Authenticator
with (
github_http(_copilot_surface_responder(_anthropic_messages_payload())) as requests,
patch.object( # test-quality-ok: patching the shared-login edge is the fixture under test
Authenticator, "get_api_key", return_value="shared-copilot-token"
) as mock_api_key,
patch.object( # test-quality-ok: same edge; api_key read is mocked so base must be too
Authenticator, "get_api_base", return_value="https://api.githubcopilot.com"
),
patch.object(Authenticator, "get_api_key", return_value="shared-copilot-token"),
patch.object(Authenticator, "get_api_base", return_value="https://api.githubcopilot.com"),
):
response = await litellm.anthropic.messages.acreate(
model="github_copilot/claude-haiku-4.5",
max_tokens=16,
messages=[{"role": "user", "content": "hi"}],
)
mock_api_key.assert_called()
upstream = [r for r in requests if r.url.host.endswith("githubcopilot.com")]
assert upstream and upstream[0].headers["Authorization"] == "Bearer shared-copilot-token"
assert response["usage"]["input_tokens"] == 4
def test_sync_exchange_single_flight_one_http_call():
def test_sync_resolve_single_flight_one_http_call():
import threading
def slow_respond(request: httpx.Request) -> httpx.Response:
time.sleep(0.05)
return httpx.Response(200, json=_exchange_payload(), request=request)
return httpx.Response(200, json=_user_payload(), request=request)
sessions: list[GithubCopilotUserSession] = []
with github_http(slow_respond) as requests:
threads = [
threading.Thread(
target=lambda: sessions.append(
exchange_github_token("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
resolve_copilot_user_session("user-a", "gh-token", "copilot-cred", _GITHUB_ENDPOINTS)
)
)
for _ in range(6)
@ -1055,13 +1118,13 @@ def test_sync_exchange_single_flight_one_http_call():
for t in threads:
t.join()
assert len(requests) == 1
assert len(sessions) == 6 and all(s.token == "copilot-token" for s in sessions)
assert len(sessions) == 6 and all(s.token == "gh-token" for s in sessions)
def test_two_users_same_github_token_do_not_share_session():
with github_http(_exchange_responder()):
s1: Final = exchange_github_token("user-a", "gh-same", "copilot-cred", _GITHUB_ENDPOINTS)
s2: Final = exchange_github_token("user-b", "gh-same", "copilot-cred", _GITHUB_ENDPOINTS)
with github_http(_user_responder()):
s1: Final = resolve_copilot_user_session("user-a", "gh-same", "copilot-cred", _GITHUB_ENDPOINTS)
s2: Final = resolve_copilot_user_session("user-b", "gh-same", "copilot-cred", _GITHUB_ENDPOINTS)
user_a_key: Final = _session_cache_key("user-a", "copilot-cred", _GITHUB_ENDPOINTS.host)
user_b_key: Final = _session_cache_key("user-b", "copilot-cred", _GITHUB_ENDPOINTS.host)
other_credential_key: Final = _session_cache_key("user-a", "another-credential", _GITHUB_ENDPOINTS.host)
@ -1069,7 +1132,7 @@ def test_two_users_same_github_token_do_not_share_session():
assert user_a_key != user_b_key
assert user_a_key != other_credential_key
assert user_a_key != other_host_key
assert s1.token == s2.token == "copilot-token"
assert s1.token == s2.token == "gh-same"
def test_same_user_and_token_do_not_share_sessions_across_hosts():
@ -1078,11 +1141,11 @@ def test_same_user_and_token_do_not_share_sessions_across_hosts():
ghe_endpoints: Final = github_host_endpoints("octo.ghe.com")
def respond(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=_exchange_payload(), request=request)
return httpx.Response(200, json=_user_payload(), request=request)
with github_http(respond) as requests:
github_session: Final = exchange_github_token("user-a", "gh-same", "copilot-cred", _GITHUB_ENDPOINTS)
ghe_session: Final = exchange_github_token("user-a", "gh-same", "copilot-cred", ghe_endpoints)
github_session: Final = resolve_copilot_user_session("user-a", "gh-same", "copilot-cred", _GITHUB_ENDPOINTS)
ghe_session: Final = resolve_copilot_user_session("user-a", "gh-same", "copilot-cred", ghe_endpoints)
assert len(requests) == 2
assert github_session.api_base == "https://api.githubcopilot.com"
assert ghe_session.api_base == "https://copilot-api.octo.ghe.com"
@ -1141,33 +1204,37 @@ def test_provider_info_shared_github_copilot_still_uses_authenticator():
@pytest.mark.asyncio
async def test_aexchange_cancelled_waiter_does_not_cancel_the_shared_exchange(per_user_credential):
"""Two waiters share one in-flight exchange; cancelling one must not cancel the
async def test_aresolve_cancelled_waiter_does_not_cancel_the_shared_session(per_user_credential):
"""Two waiters share one in-flight resolution; cancelling one must not cancel the
underlying future the other is waiting on."""
gate = asyncio.Event()
class _GatedClient:
async def get(self, url, headers=None):
await gate.wait()
return httpx.Response(200, json=_exchange_payload(), request=httpx.Request("GET", url))
return httpx.Response(200, json=_user_payload(), request=httpx.Request("GET", url))
with patch( # test-quality-ok: the gate holds the HTTP edge open so two waiters overlap
"litellm.llms.custom_httpx.http_handler.get_async_httpx_client", return_value=_GatedClient()
):
first = asyncio.ensure_future(aexchange_github_token("user-a", "gh-shared", "copilot-cred", _GITHUB_ENDPOINTS))
first = asyncio.ensure_future(
aresolve_copilot_user_session("user-a", "gh-shared", "copilot-cred", _GITHUB_ENDPOINTS)
)
await asyncio.sleep(0)
second = asyncio.ensure_future(aexchange_github_token("user-a", "gh-shared", "copilot-cred", _GITHUB_ENDPOINTS))
second = asyncio.ensure_future(
aresolve_copilot_user_session("user-a", "gh-shared", "copilot-cred", _GITHUB_ENDPOINTS)
)
await asyncio.sleep(0)
second.cancel()
gate.set()
session = await first
assert session.token == "copilot-token" and session.api_base == "https://api.githubcopilot.com"
assert session.token == "gh-shared" and session.api_base == "https://api.githubcopilot.com"
with pytest.raises(asyncio.CancelledError):
await second
@pytest.mark.asyncio
async def test_aexchange_does_not_share_an_in_flight_session_after_token_changes(per_user_credential):
async def test_aresolve_does_not_share_an_in_flight_session_after_token_changes(per_user_credential):
gate = asyncio.Event()
auth_headers: asyncio.Queue[str] = asyncio.Queue()
@ -1175,26 +1242,29 @@ async def test_aexchange_does_not_share_an_in_flight_session_after_token_changes
async def get(self, url, headers=None):
await auth_headers.put(headers["authorization"])
await gate.wait()
token = "copilot-token-a" if headers["authorization"] == "token gh-a" else "copilot-token-b"
return httpx.Response(
200,
json=_exchange_payload(token=token),
json=_user_payload(),
request=httpx.Request("GET", url),
)
with patch("litellm.llms.custom_httpx.http_handler.get_async_httpx_client", return_value=_GatedClient()):
first = asyncio.ensure_future(aexchange_github_token("user-a", "gh-a", "copilot-cred", _GITHUB_ENDPOINTS))
first = asyncio.ensure_future(
aresolve_copilot_user_session("user-a", "gh-a", "copilot-cred", _GITHUB_ENDPOINTS)
)
assert await asyncio.wait_for(auth_headers.get(), timeout=1) == "token gh-a"
second = asyncio.ensure_future(aexchange_github_token("user-a", "gh-b", "copilot-cred", _GITHUB_ENDPOINTS))
second = asyncio.ensure_future(
aresolve_copilot_user_session("user-a", "gh-b", "copilot-cred", _GITHUB_ENDPOINTS)
)
assert await asyncio.wait_for(auth_headers.get(), timeout=1) == "token gh-b"
gate.set()
sessions = await asyncio.gather(first, second)
assert [session.token for session in sessions] == ["copilot-token-a", "copilot-token-b"]
assert [session.token for session in sessions] == ["gh-a", "gh-b"]
@pytest.mark.asyncio
async def test_aexchange_failure_with_a_single_caller_emits_no_unretrieved_warning(per_user_credential):
async def test_aresolve_failure_with_a_single_caller_emits_no_unretrieved_warning(per_user_credential):
"""The leader removes the failed future before anyone awaits it; the stored
exception must be consumed so the loop does not log 'never retrieved'."""
import litellm.llms.github_copilot.per_user_auth as pua
@ -1205,10 +1275,10 @@ async def test_aexchange_failure_with_a_single_caller_emits_no_unretrieved_warni
loop.set_exception_handler(lambda _loop, context: captured.append((_loop, context)))
try:
with (
github_http(_exchange_responder(status=401)),
github_http(_user_responder(status=401)),
pytest.raises(CallerCredentialAuthenticationError),
):
await aexchange_github_token("user-a", "gh-fail", "copilot-cred", _GITHUB_ENDPOINTS)
await aresolve_copilot_user_session("user-a", "gh-fail", "copilot-cred", _GITHUB_ENDPOINTS)
_IN_FLIGHT_KEYS = list(pua._IN_FLIGHT)
import gc
@ -1240,7 +1310,7 @@ async def test_per_user_streaming_call_writes_nothing_to_the_response_cache(per_
"model": "github_copilot/gpt-4o",
"messages": [{"role": "user", "content": "hi"}],
GITHUB_COPILOT_USER_SESSION_KWARG_KEY: GithubCopilotUserSession(
token="copilot-token", api_base="https://api.githubcopilot.com"
token="gh-token", api_base="https://api.githubcopilot.com"
),
}
await handler.async_get_cache(

View file

@ -2069,7 +2069,10 @@ from litellm.proxy.credential_endpoints import user_provider_credentials as upc
_DEVICE_CODE_URL = "https://github.com/login/device/code"
_ACCESS_TOKEN_URL = "https://github.com/login/oauth/access_token"
_COPILOT_TOKEN_URL = "https://api.github.com/copilot_internal/v2/token"
_COPILOT_USER_URL = "https://api.github.com/copilot_internal/user"
_COPILOT_API_BASE = "https://api.githubcopilot.com"
_COPILOT_MODELS_URL = f"{_COPILOT_API_BASE}/models"
_COPILOT_USER_RESPONSE = {"endpoints": {"api": _COPILOT_API_BASE}}
_GITHUB_USER_URL = "https://api.github.com/user"
@ -2245,7 +2248,8 @@ def test_user_connection_poll_connected_persists_and_reports_login(monkeypatch):
{
_DEVICE_CODE_URL: _DEVICE_FLOW_START,
_ACCESS_TOKEN_URL: {"access_token": "gho_1"},
_COPILOT_TOKEN_URL: {"token": "copilot-tok", "expires_at": 4102444800},
_COPILOT_USER_URL: _COPILOT_USER_RESPONSE,
_COPILOT_MODELS_URL: {},
_GITHUB_USER_URL: {"login": "octo"},
}
):
@ -2276,19 +2280,18 @@ def test_poll_host_change_race_requires_reconnect_on_the_next_request(monkeypatc
table: Final = _patch_user_connection_env(monkeypatch, [_per_user_credential(host="octo.ghe.com")])
def change_host_after_token_exchange(request: httpx.Request) -> dict[str, str]:
def change_host_after_access_token(request: httpx.Request) -> dict[str, str]:
monkeypatch.setattr(litellm, "credential_list", [_per_user_credential()])
return {"access_token": "gho_1"}
with _github_http(
{
"https://octo.ghe.com/login/device/code": _DEVICE_FLOW_START,
"https://octo.ghe.com/login/oauth/access_token": change_host_after_token_exchange,
"https://api.octo.ghe.com/copilot_internal/v2/token": {
"token": "copilot-tok",
"expires_at": 4102444800,
"endpoints": {"api": "https://copilot-api.octo.ghe.com"},
"https://octo.ghe.com/login/oauth/access_token": change_host_after_access_token,
"https://api.octo.ghe.com/copilot_internal/user": {
"endpoints": {"api": "https://copilot-api.octo.ghe.com"}
},
"https://copilot-api.octo.ghe.com/models": {},
"https://api.octo.ghe.com/user": {"login": "octo"},
}
):
@ -2318,7 +2321,7 @@ def test_poll_host_change_race_requires_reconnect_on_the_next_request(monkeypatc
},
}
module_level_client: Final = MagicMock()
module_level_client.get.side_effect = AssertionError("unexpected token exchange")
module_level_client.get.side_effect = AssertionError("unexpected GitHub user lookup")
with _github_http({}), patch.object(litellm, "module_level_client", module_level_client):
with pytest.raises(litellm.CallerCredentialAuthenticationError) as exc:
attach_github_copilot_user_session(kwargs)
@ -2326,17 +2329,16 @@ def test_poll_host_change_race_requires_reconnect_on_the_next_request(monkeypatc
assert stored_connection.github_host == "octo.ghe.com"
def test_ghe_user_connection_uses_tenant_urls_for_device_flow_exchange_and_login(monkeypatch):
def test_ghe_user_connection_uses_tenant_urls_for_device_flow_and_user_lookup(monkeypatch):
table: Final = _patch_user_connection_env(monkeypatch, [_per_user_credential(host="octo.ghe.com")])
with _github_http(
{
"https://octo.ghe.com/login/device/code": _DEVICE_FLOW_START,
"https://octo.ghe.com/login/oauth/access_token": {"access_token": "gho_1"},
"https://api.octo.ghe.com/copilot_internal/v2/token": {
"token": "copilot-tok",
"expires_at": 4102444800,
"endpoints": {"api": "https://copilot-api.octo.ghe.com"},
"https://api.octo.ghe.com/copilot_internal/user": {
"endpoints": {"api": "https://copilot-api.octo.ghe.com"}
},
"https://copilot-api.octo.ghe.com/models": {},
"https://api.octo.ghe.com/user": {"login": "octo"},
}
):
@ -2471,7 +2473,9 @@ def test_user_connection_poll_reports_no_copilot_seat(monkeypatch, seat_status):
return httpx.Response(200, json=_DEVICE_FLOW_START, request=request)
if url == _ACCESS_TOKEN_URL:
return httpx.Response(200, json={"access_token": "gho_seat"}, request=request)
if url == _COPILOT_TOKEN_URL:
if url == _COPILOT_USER_URL:
return httpx.Response(200, json=_COPILOT_USER_RESPONSE, request=request)
if url == _COPILOT_MODELS_URL:
return httpx.Response(seat_status, json={}, request=request)
pytest.fail(f"unexpected request to {url}")
@ -2499,7 +2503,7 @@ def test_user_connection_poll_rate_limit_returns_429(monkeypatch):
return httpx.Response(200, json=_DEVICE_FLOW_START, request=request)
if url == _ACCESS_TOKEN_URL:
return httpx.Response(200, json={"access_token": "gho_limited"}, request=request)
if url == _COPILOT_TOKEN_URL:
if url == _COPILOT_USER_URL:
return httpx.Response(429, json={}, request=request)
pytest.fail(f"unexpected request to {url}")
@ -2875,7 +2879,8 @@ def test_user_connection_poll_valid_handle_works_with_a_fresh_cache(monkeypatch)
{
_DEVICE_CODE_URL: _DEVICE_FLOW_START,
_ACCESS_TOKEN_URL: {"access_token": "gho_1"},
_COPILOT_TOKEN_URL: {"token": "copilot-tok", "expires_at": 4102444800},
_COPILOT_USER_URL: _COPILOT_USER_RESPONSE,
_COPILOT_MODELS_URL: {},
_GITHUB_USER_URL: {"login": "octo"},
}
):
@ -2959,7 +2964,8 @@ def test_user_connection_poll_reports_503_when_the_cache_overwrite_cannot_land(m
{
_DEVICE_CODE_URL: _DEVICE_FLOW_START,
_ACCESS_TOKEN_URL: {"access_token": "gho_1"},
_COPILOT_TOKEN_URL: {"token": "copilot-tok", "expires_at": 4102444800},
_COPILOT_USER_URL: _COPILOT_USER_RESPONSE,
_COPILOT_MODELS_URL: {},
_GITHUB_USER_URL: {"login": "octo"},
}
):

View file

@ -57,6 +57,7 @@ _IN_MEMORY_ONLY_CALLERS: Final = frozenset(
"litellm/llms/databricks/common_utils.py",
"litellm/llms/gigachat/authenticator.py",
"litellm/llms/github_copilot/per_user_auth.py",
"litellm/llms/github_copilot/authenticator.py",
"litellm/llms/litellm_proxy/skills/handler.py",
"litellm/llms/openai/common_utils.py",
"litellm/llms/openai_like/model_info.py",

View file

@ -2533,9 +2533,7 @@ def non_openai_responses_bridge(
json.dumps({"access_token": "token", "account_id": "acct", "expires_at": 4102444800})
)
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(tmp_path))
(tmp_path / "api-key.json").write_text(
json.dumps({"token": "token", "expires_at": 4102444800, "endpoints": {"api": "https://api.githubcopilot.com"}})
)
(tmp_path / "access-token").write_text("gho_test")
litellm.register_model(
{
**{
@ -2579,7 +2577,11 @@ def non_openai_responses_bridge(
{"api_base": _FOUNDRY_API_BASE, "api_key": "fake-key"},
),
"chatgpt": ("chatgpt/gpt-6.1-bridge-test", "https://chatgpt.com/backend-api/codex/responses", {}),
"github_copilot": ("github_copilot/gpt-6.1-bridge-test", "https://api.githubcopilot.com/responses", {}),
"github_copilot": (
"github_copilot/gpt-6.1-bridge-test",
"https://api.githubcopilot.com/responses",
{"api_base": "https://api.githubcopilot.com"},
),
"databricks": (
"databricks/responses/databricks-gpt-bridge-test",
"https://dbx.example/serving-endpoints/responses",