XFE Git
XFE Studio Git
Git 首页 全局搜索
XFE 主站 文档 NuGet
公开
关注 0 Fork 0 Star 0
返回提交历史

XFEstudio/gpt4free

feat: implement GitHub Copilot provider with OAuth authentication and token management

8db86ad7
hlohaus <983577+hlohaus@users.noreply.github.com>
提交于

代码差异

13 个文件 +920 -171
Added g4f.dev +1 -0
@@ -0,0 +1 @@
1 Subproject commit ad0e5f9fb3b41e80b275d7ce48fe650476a92d49
Added g4f.exe +1 -0
@@ -0,0 +1 @@
1 Subproject commit 5595da04099433411b1b81465b945adda35cc47e
Added g4f/Provider/github/GithubCopilot.py +321 -0
@@ -0,0 +1,321 @@
1 from __future__ import annotations
2
3 import sys
4 import json
5 import time
6 import asyncio
7 from pathlib import Path
8 from typing import Optional
9
10 from ...typing import Messages, AsyncResult
11 from ..template import OpenaiTemplate
12 from .githubOAuth2 import GithubOAuth2Client
13 from .copilotTokenProvider import CopilotTokenProvider, EDITOR_VERSION, EDITOR_PLUGIN_VERSION
14 from .sharedTokenManager import TokenManagerError, SharedTokenManager
15 from .oauthFlow import launch_browser_for_oauth
16
17
18 class GithubCopilot(OpenaiTemplate):
19 """
20 GitHub Copilot provider with OAuth authentication.
21
22 This provider uses GitHub OAuth device flow for authentication,
23 allowing users to authenticate via browser without sharing credentials.
24
25 Usage:
26 1. Run `g4f-github-copilot login` to authenticate
27 2. Use the provider normally after authentication
28
29 Example:
30 >>> from g4f.client import Client
31 >>> from g4f.Provider.github import GithubCopilot
32 >>> client = Client(provider=GithubCopilot)
33 >>> response = client.chat.completions.create(
34 ... model="gpt-4o",
35 ... messages=[{"role": "user", "content": "Hello!"}]
36 ... )
37 """
38
39 label = "GitHub Copilot (OAuth) 🔐"
40 url = "https://github.com/copilot"
41 login_url = "https://github.com/login"
42 working = True
43 needs_auth = True
44 active_by_default = True
45
46 default_model = "gpt-4.1"
47 base_url = "https://api.githubcopilot.com"
48
49 models = [
50 # GPT-5 Series
51 "gpt-5",
52 "gpt-5-mini",
53 "gpt-5.1",
54 "gpt-5.2",
55
56 # GPT-5 Codex (optimized for code)
57 "gpt-5-codex",
58 "gpt-5.1-codex",
59 "gpt-5.1-codex-mini",
60 "gpt-5.1-codex-max",
61 "gpt-5.2-codex",
62 "gpt-5.3-codex",
63
64 # GPT-4 Series
65 "gpt-4.1",
66 "gpt-4.1-2025-04-14",
67 "gpt-4o",
68 "gpt-4o-mini",
69 "gpt-4o-2024-11-20",
70 "gpt-4o-2024-08-06",
71 "gpt-4o-2024-05-13",
72 "gpt-4o-mini-2024-07-18",
73 "gpt-4",
74 "gpt-4-0613",
75 "gpt-4-0125-preview",
76 "gpt-4-o-preview",
77
78 # Claude 4 Series
79 "claude-opus-4.6",
80 "claude-opus-4.6-fast",
81 "claude-opus-4.5",
82 "claude-sonnet-4.5",
83 "claude-sonnet-4",
84 "claude-haiku-4.5",
85
86 # Gemini Series
87 "gemini-3-pro-preview",
88 "gemini-3-flash-preview",
89 "gemini-2.5-pro",
90
91 # Grok
92 "grok-code-fast-1",
93
94 # Legacy GPT-3.5
95 "gpt-3.5-turbo",
96 "gpt-3.5-turbo-0613",
97
98 # Embeddings
99 "text-embedding-3-small",
100 "text-embedding-ada-002",
101 ]
102
103 _token_provider: Optional[CopilotTokenProvider] = None
104
105 @classmethod
106 def _get_token_provider(cls) -> CopilotTokenProvider:
107 if cls._token_provider is None:
108 cls._token_provider = CopilotTokenProvider()
109 return cls._token_provider
110
111 @classmethod
112 async def create_async_generator(
113 cls,
114 model: str,
115 messages: Messages,
116 api_key: str = None,
117 base_url: str = None,
118 headers: dict = None,
119 **kwargs
120 ) -> AsyncResult:
121 """
122 Create an async generator for chat completions.
123
124 If api_key is provided, it will be used directly.
125 Otherwise, OAuth credentials will be used.
126 """
127 # If no API key provided, use OAuth token
128 if api_key is None:
129 try:
130 token_provider = cls._get_token_provider()
131 creds = await token_provider.get_valid_token()
132 api_key = creds.get("token")
133 if not base_url:
134 base_url = creds.get("endpoint", cls.base_url)
135 except TokenManagerError as e:
136 if "login" in str(e).lower() or "credentials" in str(e).lower():
137 raise RuntimeError(
138 "GitHub Copilot OAuth not configured. "
139 "Please run 'g4f-github-copilot login' to authenticate."
140 ) from e
141 raise
142
143 # Add required Copilot headers
144 copilot_headers = {
145 "Editor-Version": EDITOR_VERSION,
146 "Editor-Plugin-Version": EDITOR_PLUGIN_VERSION,
147 "Openai-Organization": "github-copilot",
148 "Copilot-Integration-Id": "vscode-chat",
149 "X-GitHub-Api-Version": "2024-12-15",
150 }
151 if headers:
152 copilot_headers.update(headers)
153
154 # Use parent class for actual API calls
155 async for chunk in super().create_async_generator(
156 model,
157 messages,
158 api_key=api_key,
159 base_url=base_url or cls.base_url,
160 headers=copilot_headers,
161 **kwargs
162 ):
163 yield chunk
164
165 @classmethod
166 async def login(cls, credentials_path: Optional[Path] = None) -> SharedTokenManager:
167 """
168 Perform interactive OAuth login and save credentials.
169
170 Args:
171 credentials_path: Path to save credentials (default: g4f cache)
172
173 Returns:
174 SharedTokenManager with active credentials
175
176 Example:
177 >>> import asyncio
178 >>> from g4f.Provider.github import GithubCopilot
179 >>> asyncio.run(GithubCopilot.login())
180 """
181 print("\n" + "=" * 60)
182 print("GitHub Copilot OAuth Login")
183 print("=" * 60)
184
185 await launch_browser_for_oauth()
186
187 shared_manager = SharedTokenManager.getInstance()
188 print("=" * 60 + "\n")
189
190 return shared_manager
191
192 @classmethod
193 def has_credentials(cls) -> bool:
194 """Check if valid credentials exist."""
195 shared_manager = SharedTokenManager.getInstance()
196 try:
197 path = shared_manager.getCredentialFilePath()
198 return path.exists()
199 except Exception:
200 return False
201
202 @classmethod
203 def get_credentials_path(cls) -> Optional[Path]:
204 """Get path to credentials file if it exists."""
205 shared_manager = SharedTokenManager.getInstance()
206 try:
207 path = shared_manager.getCredentialFilePath()
208 if path.exists():
209 return path
210 except Exception:
211 pass
212 return None
213
214
215 async def main():
216 """CLI entry point for GitHub Copilot OAuth authentication."""
217 import argparse
218
219 parser = argparse.ArgumentParser(
220 description="GitHub Copilot OAuth Authentication for gpt4free",
221 formatter_class=argparse.RawDescriptionHelpFormatter,
222 epilog="""
223 Examples:
224 %(prog)s login # Interactive device code login
225 %(prog)s status # Check authentication status
226 %(prog)s logout # Remove saved credentials
227 """
228 )
229
230 subparsers = parser.add_subparsers(dest="command", help="Commands")
231
232 # Login command
233 subparsers.add_parser("login", help="Authenticate with GitHub Copilot")
234
235 # Status command
236 subparsers.add_parser("status", help="Check authentication status")
237
238 # Logout command
239 subparsers.add_parser("logout", help="Remove saved credentials")
240
241 args = parser.parse_args()
242
243 if args.command == "login":
244 try:
245 await GithubCopilot.login()
246 except KeyboardInterrupt:
247 print("\n\nLogin cancelled.")
248 sys.exit(1)
249 except Exception as e:
250 print(f"\n❌ Login failed: {e}")
251 sys.exit(1)
252
253 elif args.command == "status":
254 print("\nGitHub Copilot OAuth Status")
255 print("=" * 40)
256
257 if GithubCopilot.has_credentials():
258 creds_path = GithubCopilot.get_credentials_path()
259 print(f"✓ Credentials found at: {creds_path}")
260
261 try:
262 with creds_path.open() as f:
263 creds = json.load(f)
264
265 expiry = creds.get("expiry_date")
266 if expiry:
267 expiry_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(expiry / 1000))
268 if expiry / 1000 > time.time():
269 print(f" Token expires: {expiry_time}")
270 else:
271 print(f" Token expired: {expiry_time}")
272
273 if creds.get("scope"):
274 print(f" Scope: {creds['scope']}")
275 except Exception as e:
276 print(f" (Could not read credential details: {e})")
277 else:
278 print("✗ No credentials found")
279 print(f"\nRun 'g4f-github-copilot login' to authenticate.")
280
281 print()
282
283 elif args.command == "logout":
284 print("\nGitHub Copilot OAuth Logout")
285 print("=" * 40)
286
287 removed = False
288
289 shared_manager = SharedTokenManager.getInstance()
290 path = shared_manager.getCredentialFilePath()
291
292 if path.exists():
293 path.unlink()
294 print(f"✓ Removed: {path}")
295 removed = True
296
297 # Also try the default location
298 default_path = Path.home() / ".github-copilot" / "oauth_creds.json"
299 if default_path.exists() and default_path != path:
300 default_path.unlink()
301 print(f"✓ Removed: {default_path}")
302 removed = True
303
304 if removed:
305 print("\n✓ Credentials removed successfully.")
306 else:
307 print("No credentials found to remove.")
308
309 print()
310
311 else:
312 parser.print_help()
313
314
315 def cli_main():
316 """Synchronous CLI entry point for setup.py console_scripts."""
317 asyncio.run(main())
318
319
320 if __name__ == "__main__":
321 cli_main()
Added g4f/Provider/github/__init__.py +1 -0
@@ -0,0 +1 @@
1 from .GithubCopilot import GithubCopilot
Added g4f/Provider/github/copilotTokenProvider.py +107 -0
@@ -0,0 +1,107 @@
1 """
2 GitHub Copilot Token Provider
3
4 This module handles the retrieval of Copilot API tokens using GitHub OAuth credentials.
5 """
6 from typing import Dict, Optional
7 import aiohttp
8 import time
9
10 from .githubOAuth2 import GithubOAuth2Client
11 from .sharedTokenManager import SharedTokenManager, TokenManagerError
12
13
14 # Editor/Plugin version headers required by Copilot API
15 EDITOR_VERSION = "vscode/1.95.0"
16 EDITOR_PLUGIN_VERSION = "copilot/1.250.0"
17
18
19 class CopilotTokenProvider:
20 """Provides Copilot API tokens from GitHub OAuth credentials."""
21
22 # Copilot token endpoint
23 COPILOT_TOKEN_URL = "https://api.github.com/copilot_internal/v2/token"
24
25 def __init__(self, github_client: GithubOAuth2Client = None):
26 self.github_client = github_client or GithubOAuth2Client()
27 self.shared_manager = SharedTokenManager.getInstance()
28 self._copilot_token = None
29 self._copilot_token_expires_at = 0
30
31 async def get_copilot_token(self) -> Optional[str]:
32 """
33 Get a valid Copilot API token.
34
35 This exchanges the GitHub OAuth token for a Copilot-specific token.
36
37 Returns:
38 The Copilot API token, or None if not available.
39 """
40 # Check if we have a valid cached token
41 if self._copilot_token and time.time() < self._copilot_token_expires_at - 60:
42 return self._copilot_token
43
44 # Get GitHub OAuth token
45 github_creds = await self.shared_manager.getValidCredentials(self.github_client)
46 if not github_creds or not github_creds.get("access_token"):
47 raise TokenManagerError("NO_TOKEN", "No GitHub OAuth token available. Please login first.")
48
49 github_token = github_creds["access_token"]
50
51 # Exchange for Copilot token
52 async with aiohttp.ClientSession() as session:
53 async with session.get(
54 self.COPILOT_TOKEN_URL,
55 headers={
56 "Authorization": f"token {github_token}",
57 "Accept": "application/json",
58 "User-Agent": "GithubCopilot/1.250.0",
59 "Editor-Version": EDITOR_VERSION,
60 "Editor-Plugin-Version": EDITOR_PLUGIN_VERSION,
61 "Openai-Organization": "github-copilot",
62 "X-GitHub-Api-Version": "2024-12-15",
63 }
64 ) as resp:
65 if resp.status == 401:
66 raise TokenManagerError(
67 "AUTH_FAILED",
68 "GitHub token is invalid or expired. Please login again."
69 )
70 if resp.status != 200:
71 text = await resp.text()
72 raise TokenManagerError(
73 "TOKEN_ERROR",
74 f"Failed to get Copilot token: {resp.status} - {text}"
75 )
76
77 data = await resp.json()
78 self._copilot_token = data.get("token")
79
80 # Parse expiration
81 expires_at = data.get("expires_at")
82 if expires_at:
83 try:
84 # Parse ISO format datetime
85 from datetime import datetime
86 dt = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
87 self._copilot_token_expires_at = dt.timestamp()
88 except Exception:
89 # Default to 30 minutes from now if parsing fails
90 self._copilot_token_expires_at = time.time() + 1800
91 else:
92 self._copilot_token_expires_at = time.time() + 1800
93
94 return self._copilot_token
95
96 async def get_valid_token(self) -> Dict[str, Optional[str]]:
97 """
98 Get valid credentials for the Copilot API.
99
100 Returns:
101 Dict with 'token' and optionally 'endpoint'
102 """
103 token = await self.get_copilot_token()
104 return {
105 "token": token,
106 "endpoint": "https://api.githubcopilot.com"
107 }
Added g4f/Provider/github/githubOAuth2.py +159 -0
@@ -0,0 +1,159 @@
1 import time
2 from typing import Dict, Optional, Union
3
4 import aiohttp
5
6 from .stubs import IGithubOAuth2Client, GithubCredentials, ErrorDataDict
7 from .sharedTokenManager import SharedTokenManager
8
9
10 # GitHub OAuth endpoints
11 GITHUB_DEVICE_CODE_ENDPOINT = "https://github.com/login/device/code"
12 GITHUB_TOKEN_ENDPOINT = "https://github.com/login/oauth/access_token"
13
14 # GitHub Copilot OAuth Client ID (VS Code Extension)
15 GITHUB_COPILOT_CLIENT_ID = "Iv1.b507a08c87ecfe98"
16
17 # Scopes needed for Copilot
18 GITHUB_COPILOT_SCOPE = "read:user"
19
20 TOKEN_REFRESH_BUFFER_MS = 30 * 1000 # 30 seconds
21
22
23 def object_to_urlencoded(data: Dict[str, str]) -> str:
24 return "&".join([f"{k}={v}" for k, v in data.items()])
25
26
27 def isDeviceAuthorizationSuccess(response: Union[Dict, ErrorDataDict]) -> bool:
28 return "device_code" in response
29
30
31 def isDeviceTokenSuccess(response: Union[Dict, ErrorDataDict]) -> bool:
32 return (
33 "access_token" in response
34 and response["access_token"]
35 and isinstance(response["access_token"], str)
36 and len(response["access_token"]) > 0
37 )
38
39
40 def isDeviceTokenPending(response: Union[Dict, ErrorDataDict]) -> bool:
41 return response.get("error") == "authorization_pending"
42
43
44 def isSlowDown(response: Union[Dict, ErrorDataDict]) -> bool:
45 return response.get("error") == "slow_down"
46
47
48 def isErrorResponse(response: Union[Dict, ErrorDataDict]) -> bool:
49 return "error" in response and response.get("error") not in ["authorization_pending", "slow_down"]
50
51
52 class GithubOAuth2Client(IGithubOAuth2Client):
53 def __init__(self, client_id: str = GITHUB_COPILOT_CLIENT_ID):
54 self.client_id = client_id
55 self.credentials: GithubCredentials = GithubCredentials()
56 self.sharedManager = SharedTokenManager.getInstance()
57
58 def setCredentials(self, credentials: GithubCredentials):
59 self.credentials = credentials
60
61 def getCredentials(self) -> GithubCredentials:
62 return self.credentials
63
64 async def getAccessToken(self) -> Dict[str, Optional[str]]:
65 try:
66 credentials = await self.sharedManager.getValidCredentials(self)
67 return {"token": credentials.get("access_token")}
68 except Exception:
69 # fallback to internal credentials if valid
70 if (
71 self.credentials.get("access_token")
72 and self.isTokenValid(self.credentials)
73 ):
74 return {"token": self.credentials["access_token"]}
75 return {"token": None}
76
77 async def requestDeviceAuthorization(self, options: dict) -> Union[Dict, ErrorDataDict]:
78 """
79 Request device authorization from GitHub.
80
81 Returns:
82 dict with device_code, user_code, verification_uri, expires_in, interval
83 """
84 body_data = {
85 "client_id": self.client_id,
86 "scope": options.get("scope", GITHUB_COPILOT_SCOPE),
87 }
88
89 async with aiohttp.ClientSession() as session:
90 async with session.post(
91 GITHUB_DEVICE_CODE_ENDPOINT,
92 headers={
93 "Content-Type": "application/x-www-form-urlencoded",
94 "Accept": "application/json",
95 },
96 data=object_to_urlencoded(body_data)
97 ) as resp:
98 resp_json = await resp.json()
99
100 if resp.status != 200:
101 raise Exception(f"Device authorization failed {resp.status}: {resp_json}")
102
103 if not isDeviceAuthorizationSuccess(resp_json):
104 raise Exception(
105 f"Device authorization error: {resp_json.get('error')} - {resp_json.get('error_description')}"
106 )
107
108 return resp_json
109
110 async def pollDeviceToken(self, options: dict) -> Union[Dict, ErrorDataDict]:
111 """
112 Poll for device token from GitHub.
113
114 Args:
115 options: dict with device_code
116
117 Returns:
118 dict with access_token, token_type, scope or status=pending
119 """
120 body_data = {
121 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
122 "client_id": self.client_id,
123 "device_code": options["device_code"],
124 }
125
126 async with aiohttp.ClientSession() as session:
127 async with session.post(
128 GITHUB_TOKEN_ENDPOINT,
129 headers={
130 "Content-Type": "application/x-www-form-urlencoded",
131 "Accept": "application/json",
132 },
133 data=object_to_urlencoded(body_data)
134 ) as resp:
135 resp_json = await resp.json()
136
137 # Check for OAuth RFC 8628 responses
138 if "error" in resp_json:
139 if resp_json["error"] == "authorization_pending":
140 return {"status": "pending"}
141 if resp_json["error"] == "slow_down":
142 return {"status": "pending", "slowDown": True}
143 if resp_json["error"] == "expired_token":
144 raise Exception("Device code expired. Please try again.")
145 if resp_json["error"] == "access_denied":
146 raise Exception("Authorization was denied by the user.")
147 raise Exception(f"Token poll failed: {resp_json.get('error')} - {resp_json.get('error_description')}")
148
149 return resp_json
150
151 def isTokenValid(self, credentials: GithubCredentials) -> bool:
152 """GitHub tokens don't expire by default, but we track expiry_date if set"""
153 if not credentials.get("access_token"):
154 return False
155 expiry_date = credentials.get("expiry_date")
156 if expiry_date is None:
157 # GitHub tokens don't expire unless explicitly set
158 return True
159 return time.time() * 1000 < expiry_date - TOKEN_REFRESH_BUFFER_MS
Added g4f/Provider/github/oauthFlow.py +109 -0
@@ -0,0 +1,109 @@
1 import asyncio
2 import webbrowser
3 import time
4
5 from .githubOAuth2 import GithubOAuth2Client, GITHUB_COPILOT_SCOPE
6
7
8 async def launch_browser_for_oauth(client_id: str = None):
9 """
10 Perform GitHub OAuth device flow for authentication.
11
12 This function:
13 1. Requests a device code from GitHub
14 2. Opens a browser for the user to authenticate
15 3. Polls for the access token
16 4. Saves the credentials
17
18 Args:
19 client_id: Optional custom client ID (defaults to Copilot VS Code extension)
20 """
21 # Initialize OAuth client
22 client = GithubOAuth2Client(client_id) if client_id else GithubOAuth2Client()
23
24 # Request device code
25 print("Requesting device authorization from GitHub...")
26 device_auth = await client.requestDeviceAuthorization({
27 "scope": GITHUB_COPILOT_SCOPE,
28 })
29
30 # Check device auth success
31 if not isinstance(device_auth, dict) or "device_code" not in device_auth:
32 print("Failed to receive device code")
33 return None
34
35 # Show user instructions
36 user_code = device_auth.get("user_code")
37 verification_uri = device_auth.get("verification_uri", "https://github.com/login/device")
38
39 print("\n" + "=" * 60)
40 print("GitHub Copilot Authorization")
41 print("=" * 60)
42 print(f"\nPlease visit: {verification_uri}")
43 print(f"Enter code: {user_code}")
44 print("=" * 60 + "\n")
45
46 # Attempt to automatically open the URL
47 try:
48 webbrowser.open(verification_uri)
49 print("Browser opened automatically.")
50 except Exception:
51 print(f"Please open the URL manually in your browser: {verification_uri}")
52
53 # Start polling for token
54 device_code = device_auth["device_code"]
55 expires_in = device_auth.get("expires_in", 900) # default 15 min
56 interval = device_auth.get("interval", 5) # default 5 seconds
57 start_time = time.time()
58
59 print("\nWaiting for authorization... Press Ctrl+C to cancel.")
60
61 while True:
62 if time.time() - start_time > expires_in:
63 print("\nAuthorization timed out. Please try again.")
64 return None
65
66 # Poll for token
67 token_response = await client.pollDeviceToken({
68 "device_code": device_code,
69 })
70
71 if isinstance(token_response, dict):
72 if token_response.get("status") == "pending":
73 if token_response.get("slowDown"):
74 interval += 5 # Increase interval as requested by GitHub
75 print(".", end="", flush=True)
76 await asyncio.sleep(interval)
77 continue
78 elif "access_token" in token_response:
79 # Success
80 print("\n\n✓ Authorization successful!")
81
82 # Save credentials
83 credentials = {
84 "access_token": token_response["access_token"],
85 "token_type": token_response.get("token_type", "bearer"),
86 "scope": token_response.get("scope", ""),
87 # GitHub tokens don't expire, but we can set a far future date
88 "expiry_date": int(time.time() * 1000) + (365 * 24 * 60 * 60 * 1000), # 1 year
89 }
90
91 await client.sharedManager.saveCredentialsToFile(credentials)
92 print(f"Credentials saved to: {client.sharedManager.getCredentialFilePath()}")
93
94 return credentials
95 else:
96 print(f"\nError during polling: {token_response}")
97 return None
98 else:
99 print(f"\nUnexpected response: {token_response}")
100 return None
101
102
103 async def main():
104 """Run the OAuth flow."""
105 await launch_browser_for_oauth()
106
107
108 if __name__ == "__main__":
109 asyncio.run(main())
Added g4f/Provider/github/sharedTokenManager.py +166 -0
@@ -0,0 +1,166 @@
1 import os
2 import json
3 import time
4 import asyncio
5 import threading
6 from typing import Optional, Dict
7 from pathlib import Path
8
9 from ..base_provider import AuthFileMixin
10 from ... import debug
11
12 GITHUB_DIR = ".github-copilot"
13 GITHUB_CREDENTIAL_FILENAME = "oauth_creds.json"
14 GITHUB_LOCK_FILENAME = "oauth_creds.lock"
15 TOKEN_REFRESH_BUFFER_MS = 30 * 1000
16 CACHE_CHECK_INTERVAL_MS = 1000
17
18
19 class TokenError:
20 REFRESH_FAILED = "REFRESH_FAILED"
21 NO_REFRESH_TOKEN = "NO_REFRESH_TOKEN"
22 LOCK_TIMEOUT = "LOCK_TIMEOUT"
23 FILE_ACCESS_ERROR = "FILE_ACCESS_ERROR"
24 NETWORK_ERROR = "NETWORK_ERROR"
25
26
27 class TokenManagerError(Exception):
28 def __init__(self, type_: str, message: str, original_error: Optional[Exception] = None):
29 super().__init__(message)
30 self.type = type_
31 self.original_error = original_error
32
33
34 class SharedTokenManager(AuthFileMixin):
35 parent = "GithubCopilotOAuth"
36 _instance: Optional["SharedTokenManager"] = None
37 _lock = threading.Lock()
38
39 def __init__(self):
40 self.memory_cache = {
41 "credentials": None,
42 "file_mod_time": 0,
43 "last_check": 0,
44 }
45 self.refresh_promise = None
46
47 @classmethod
48 def getInstance(cls):
49 with cls._lock:
50 if cls._instance is None:
51 cls._instance = cls()
52 return cls._instance
53
54 def getCredentialFilePath(self):
55 path = Path(os.path.expanduser(f"~/{GITHUB_DIR}/{GITHUB_CREDENTIAL_FILENAME}"))
56 if path.is_file():
57 return path
58 return SharedTokenManager.get_cache_file()
59
60 def getLockFilePath(self):
61 return Path(os.path.expanduser(f"~/{GITHUB_DIR}/{GITHUB_LOCK_FILENAME}"))
62
63 def getCurrentCredentials(self):
64 return self.memory_cache.get("credentials")
65
66 def checkAndReloadIfNeeded(self):
67 now = int(time.time() * 1000)
68 if now - self.memory_cache["last_check"] < CACHE_CHECK_INTERVAL_MS:
69 return
70 self.memory_cache["last_check"] = now
71
72 try:
73 file_path = self.getCredentialFilePath()
74 if not file_path.exists():
75 self.memory_cache["file_mod_time"] = 0
76 return
77 stat = file_path.stat()
78 file_mod_time = int(stat.st_mtime * 1000)
79 if file_mod_time > self.memory_cache["file_mod_time"]:
80 self.reloadCredentialsFromFile()
81 self.memory_cache["file_mod_time"] = file_mod_time
82 except FileNotFoundError:
83 self.memory_cache["file_mod_time"] = 0
84 except Exception as e:
85 self.memory_cache["credentials"] = None
86 raise TokenManagerError(TokenError.FILE_ACCESS_ERROR, str(e), e)
87
88 def reloadCredentialsFromFile(self):
89 file_path = self.getCredentialFilePath()
90 debug.log(f"Reloading credentials from {file_path}")
91 try:
92 with open(file_path, "r") as fs:
93 data = json.load(fs)
94 credentials = self.validateCredentials(data)
95 self.memory_cache["credentials"] = credentials
96 except FileNotFoundError as e:
97 self.memory_cache["credentials"] = None
98 raise TokenManagerError(TokenError.FILE_ACCESS_ERROR, "Credentials file not found", e) from e
99 except json.JSONDecodeError as e:
100 self.memory_cache["credentials"] = None
101 raise TokenManagerError(TokenError.FILE_ACCESS_ERROR, "Invalid JSON format", e) from e
102 except Exception as e:
103 self.memory_cache["credentials"] = None
104 raise TokenManagerError(TokenError.FILE_ACCESS_ERROR, str(e), e) from e
105
106 def validateCredentials(self, data):
107 if not data or not isinstance(data, dict):
108 raise ValueError("Invalid credentials format")
109 if "access_token" not in data or not isinstance(data["access_token"], str):
110 raise ValueError("Invalid credentials: missing access_token")
111 if "token_type" not in data or not isinstance(data["token_type"], str):
112 raise ValueError("Invalid credentials: missing token_type")
113 return data
114
115 def isTokenValid(self, credentials) -> bool:
116 """GitHub tokens don't expire by default"""
117 if not credentials or not credentials.get("access_token"):
118 return False
119 expiry_date = credentials.get("expiry_date")
120 if expiry_date is None:
121 return True
122 return time.time() * 1000 < expiry_date - TOKEN_REFRESH_BUFFER_MS
123
124 async def getValidCredentials(self, github_client, force_refresh: bool = False):
125 try:
126 self.checkAndReloadIfNeeded()
127
128 if (
129 self.memory_cache["credentials"]
130 and not force_refresh
131 and self.isTokenValid(self.memory_cache["credentials"])
132 ):
133 return self.memory_cache["credentials"]
134
135 if self.refresh_promise:
136 return await self.refresh_promise
137
138 # Try to reload credentials from file
139 try:
140 self.reloadCredentialsFromFile()
141 if self.memory_cache["credentials"] and self.isTokenValid(self.memory_cache["credentials"]):
142 return self.memory_cache["credentials"]
143 except TokenManagerError:
144 pass
145
146 raise TokenManagerError(
147 TokenError.FILE_ACCESS_ERROR,
148 "No valid credentials found. Please run login first."
149 )
150 except Exception as e:
151 if isinstance(e, TokenManagerError):
152 raise
153 raise TokenManagerError(TokenError.FILE_ACCESS_ERROR, str(e), e) from e
154
155 async def saveCredentialsToFile(self, credentials: dict):
156 """Save credentials to the credential file."""
157 file_path = self.getCredentialFilePath()
158 file_path.parent.mkdir(parents=True, exist_ok=True)
159
160 with open(file_path, "w") as f:
161 json.dump(credentials, f, indent=2)
162
163 self.memory_cache["credentials"] = credentials
164 self.memory_cache["file_mod_time"] = int(time.time() * 1000)
165
166 debug.log(f"Credentials saved to {file_path}")
Added g4f/Provider/github/stubs.py +26 -0
@@ -0,0 +1,26 @@
1 from typing import Dict, Optional, Union
2
3
4 class ErrorDataDict(Dict):
5 pass
6
7
8 class GithubCredentials(Dict):
9 pass
10
11
12 class IGithubOAuth2Client:
13 def setCredentials(self, credentials: GithubCredentials):
14 raise NotImplementedError
15
16 def getCredentials(self) -> GithubCredentials:
17 raise NotImplementedError
18
19 async def getAccessToken(self) -> Dict[str, Optional[str]]:
20 raise NotImplementedError
21
22 async def requestDeviceAuthorization(self, options: dict) -> Union[Dict, ErrorDataDict]:
23 raise NotImplementedError
24
25 async def pollDeviceToken(self, options: dict) -> Union[Dict, ErrorDataDict]:
26 raise NotImplementedError
Modified g4f/Provider/hf_space/BAAI_Ling.py +27 -3
@@ -13,10 +13,10 @@ from ... import debug
13 13
14 14 class BAAI_Ling(AsyncGeneratorProvider, ProviderModelMixin):
15 15 label = "Ling & Ring Playground"
16 url = "https://cafe3310-ling-playground.hf.space"
16 url = "https://cafe3310-ling-series-spaces.hf.space"
17 17 api_endpoint = f"{url}/gradio_api/queue/join"
18 18
19 working = True
19 working = False
20 20 supports_stream = True
21 21 supports_system_message = True
22 22 supports_message_history = False
@@ -67,9 +67,33 @@ class BAAI_Ling(AsyncGeneratorProvider, ProviderModelMixin):
67 67 ],
68 68 "event_data": None,
69 69 "fn_index": 11,
70 "trigger_id": 14,
70 "trigger_id": 33,
71 71 "session_hash": conversation.session_hash
72 72 }
73 payload = {
74 "data": [
75 "4aa9d0c6-81c2-4274-91c5-a0d96d827916",
76 [
77 {
78 "id": "4aa9d0c6-81c2-4274-91c5-a0d96d827917",
79 "title": "(New Conversation)",
80 "messages": [],
81 "timestamp": "2026-02-11T23:28:14.398499",
82 "system_prompt": "",
83 "model": "🦉 Ling-1T",
84 "temperature": 0.7
85 }
86 ],
87 "🦉 Ling-1T",
88 "hi",
89 [],
90 "",
91 0.7
92 ],
93 "fn_index": 11,
94 "trigger_id": 33,
95 "session_hash": "bis3t7jioto"
96 }
73 97
74 98 async with aiohttp.ClientSession() as session:
75 99 async with session.post(cls.api_endpoint, headers=headers, json=payload, proxy=proxy) as response:
Deleted g4f/Provider/needs_auth/GithubCopilot.py +0 -167
Modified g4f/Provider/needs_auth/__init__.py +1 -1
@@ -18,7 +18,7 @@ from .Gemini import Gemini
18 18 from .GeminiPro import GeminiPro
19 19 from .GeminiCLI import GeminiCLI
20 20 from .GigaChat import GigaChat
21 from .GithubCopilot import GithubCopilot
21 from ..github import GithubCopilot
22 22 from .GithubCopilotAPI import GithubCopilotAPI
23 23 from .GlhfChat import GlhfChat
24 24 from .Grok import Grok
Modified setup.py +1 -0
@@ -122,6 +122,7 @@ setup(
122 122 'g4f-antigravity=g4f.Provider.needs_auth.Antigravity:cli_main',
123 123 'g4f-geminicli=g4f.Provider.needs_auth.GeminiCLI:cli_main',
124 124 'g4f-qwencode=g4f.Provider.qwen.QwenCode:cli_main',
125 'g4f-github-copilot=g4f.Provider.github.GithubCopilot:cli_main',
125 126 ],
126 127 },
127 128 url='https://github.com/xtekky/gpt4free', # Link to your GitHub repository