1  """GitHub Copilot authentication utilities.
  2  
  3  Implements the OAuth device code flow used by the Copilot CLI and handles
  4  token validation/exchange for the Copilot API.
  5  
  6  Token type support (per GitHub docs):
  7    gho_          OAuth token           ✓  (default via copilot login)
  8    github_pat_   Fine-grained PAT      ✓  (needs Copilot Requests permission)
  9    ghu_          GitHub App token      ✓  (via environment variable)
 10    ghp_          Classic PAT           ✗  NOT SUPPORTED
 11  
 12  Credential search order (matching Copilot CLI behaviour):
 13    1. COPILOT_GITHUB_TOKEN env var
 14    2. GH_TOKEN env var
 15    3. GITHUB_TOKEN env var
 16    4. gh auth token  CLI fallback
 17  """
 18  
 19  from __future__ import annotations
 20  
 21  import json
 22  import logging
 23  import os
 24  import shutil
 25  import subprocess
 26  import time
 27  from pathlib import Path
 28  from typing import Optional
 29  
 30  logger = logging.getLogger(__name__)
 31  
 32  # OAuth device code flow constants (same client ID as opencode/Copilot CLI)
 33  COPILOT_OAUTH_CLIENT_ID = "Ov23li8tweQw6odWQebz"
 34  # Token type prefixes
 35  _CLASSIC_PAT_PREFIX = "ghp_"
 36  _SUPPORTED_PREFIXES = ("gho_", "github_pat_", "ghu_")
 37  
 38  # Env var search order (matches Copilot CLI)
 39  COPILOT_ENV_VARS = ("COPILOT_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN")
 40  
 41  # Polling constants
 42  _DEVICE_CODE_POLL_INTERVAL = 5  # seconds
 43  _DEVICE_CODE_POLL_SAFETY_MARGIN = 3  # seconds
 44  
 45  
 46  def validate_copilot_token(token: str) -> tuple[bool, str]:
 47      """Validate that a token is usable with the Copilot API.
 48  
 49      Returns (valid, message).
 50      """
 51      token = token.strip()
 52      if not token:
 53          return False, "Empty token"
 54  
 55      if token.startswith(_CLASSIC_PAT_PREFIX):
 56          return False, (
 57              "Classic Personal Access Tokens (ghp_*) are not supported by the "
 58              "Copilot API. Use one of:\n"
 59              "  → `copilot login` or `hermes model` to authenticate via OAuth\n"
 60              "  → A fine-grained PAT (github_pat_*) with Copilot Requests permission\n"
 61              "  → `gh auth login` with the default device code flow (produces gho_* tokens)"
 62          )
 63  
 64      return True, "OK"
 65  
 66  
 67  def resolve_copilot_token() -> tuple[str, str]:
 68      """Resolve a GitHub token suitable for Copilot API use.
 69  
 70      Returns (token, source) where source describes where the token came from.
 71      Raises ValueError if only a classic PAT is available.
 72      """
 73      # 1. Check env vars in priority order
 74      for env_var in COPILOT_ENV_VARS:
 75          val = os.getenv(env_var, "").strip()
 76          if val:
 77              valid, msg = validate_copilot_token(val)
 78              if not valid:
 79                  logger.warning(
 80                      "Token from %s is not supported: %s", env_var, msg
 81                  )
 82                  continue
 83              return val, env_var
 84  
 85      # 2. Fall back to gh auth token
 86      token = _try_gh_cli_token()
 87      if token:
 88          valid, msg = validate_copilot_token(token)
 89          if not valid:
 90              raise ValueError(
 91                  f"Token from `gh auth token` is a classic PAT (ghp_*). {msg}"
 92              )
 93          return token, "gh auth token"
 94  
 95      return "", ""
 96  
 97  
 98  def _gh_cli_candidates() -> list[str]:
 99      """Return candidate ``gh`` binary paths, including common Homebrew installs."""
100      candidates: list[str] = []
101  
102      resolved = shutil.which("gh")
103      if resolved:
104          candidates.append(resolved)
105  
106      for candidate in (
107          "/opt/homebrew/bin/gh",
108          "/usr/local/bin/gh",
109          str(Path.home() / ".local" / "bin" / "gh"),
110      ):
111          if candidate in candidates:
112              continue
113          if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
114              candidates.append(candidate)
115  
116      return candidates
117  
118  
119  def _try_gh_cli_token() -> Optional[str]:
120      """Return a token from ``gh auth token`` when the GitHub CLI is available.
121  
122      When COPILOT_GH_HOST is set, passes ``--hostname`` so gh returns the
123      correct host's token.  Also strips GITHUB_TOKEN / GH_TOKEN from the
124      subprocess environment so ``gh`` reads from its own credential store
125      (hosts.yml) instead of just echoing the env var back.
126      """
127      hostname = os.getenv("COPILOT_GH_HOST", "").strip()
128  
129      # Build a clean env so gh doesn't short-circuit on GITHUB_TOKEN / GH_TOKEN
130      clean_env = {k: v for k, v in os.environ.items()
131                   if k not in ("GITHUB_TOKEN", "GH_TOKEN")}
132  
133      for gh_path in _gh_cli_candidates():
134          cmd = [gh_path, "auth", "token"]
135          if hostname:
136              cmd += ["--hostname", hostname]
137          try:
138              result = subprocess.run(
139                  cmd,
140                  capture_output=True,
141                  text=True,
142                  timeout=5,
143                  env=clean_env,
144              )
145          except (FileNotFoundError, subprocess.TimeoutExpired) as exc:
146              logger.debug("gh CLI token lookup failed (%s): %s", gh_path, exc)
147              continue
148          if result.returncode == 0 and result.stdout.strip():
149              return result.stdout.strip()
150      return None
151  
152  
153  # ─── OAuth Device Code Flow ────────────────────────────────────────────────
154  
155  def copilot_device_code_login(
156      *,
157      host: str = "github.com",
158      timeout_seconds: float = 300,
159  ) -> Optional[str]:
160      """Run the GitHub OAuth device code flow for Copilot.
161  
162      Prints instructions for the user, polls for completion, and returns
163      the OAuth access token on success, or None on failure/cancellation.
164  
165      This replicates the flow used by opencode and the Copilot CLI.
166      """
167      import urllib.request
168      import urllib.parse
169  
170      domain = host.rstrip("/")
171      device_code_url = f"https://{domain}/login/device/code"
172      access_token_url = f"https://{domain}/login/oauth/access_token"
173  
174      # Step 1: Request device code
175      data = urllib.parse.urlencode({
176          "client_id": COPILOT_OAUTH_CLIENT_ID,
177          "scope": "read:user",
178      }).encode()
179  
180      req = urllib.request.Request(
181          device_code_url,
182          data=data,
183          headers={
184              "Accept": "application/json",
185              "Content-Type": "application/x-www-form-urlencoded",
186              "User-Agent": "HermesAgent/1.0",
187          },
188      )
189  
190      try:
191          with urllib.request.urlopen(req, timeout=15) as resp:
192              device_data = json.loads(resp.read().decode())
193      except Exception as exc:
194          logger.error("Failed to initiate device authorization: %s", exc)
195          print(f"  ✗ Failed to start device authorization: {exc}")
196          return None
197  
198      verification_uri = device_data.get("verification_uri", "https://github.com/login/device")
199      user_code = device_data.get("user_code", "")
200      device_code = device_data.get("device_code", "")
201      interval = max(device_data.get("interval", _DEVICE_CODE_POLL_INTERVAL), 1)
202  
203      if not device_code or not user_code:
204          print("  ✗ GitHub did not return a device code.")
205          return None
206  
207      # Step 2: Show instructions
208      print()
209      print(f"  Open this URL in your browser: {verification_uri}")
210      print(f"  Enter this code: {user_code}")
211      print()
212      print("  Waiting for authorization...", end="", flush=True)
213  
214      # Step 3: Poll for completion
215      deadline = time.time() + timeout_seconds
216  
217      while time.time() < deadline:
218          time.sleep(interval + _DEVICE_CODE_POLL_SAFETY_MARGIN)
219  
220          poll_data = urllib.parse.urlencode({
221              "client_id": COPILOT_OAUTH_CLIENT_ID,
222              "device_code": device_code,
223              "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
224          }).encode()
225  
226          poll_req = urllib.request.Request(
227              access_token_url,
228              data=poll_data,
229              headers={
230                  "Accept": "application/json",
231                  "Content-Type": "application/x-www-form-urlencoded",
232                  "User-Agent": "HermesAgent/1.0",
233              },
234          )
235  
236          try:
237              with urllib.request.urlopen(poll_req, timeout=10) as resp:
238                  result = json.loads(resp.read().decode())
239          except Exception:
240              print(".", end="", flush=True)
241              continue
242  
243          if result.get("access_token"):
244              print(" ✓")
245              return result["access_token"]
246  
247          error = result.get("error", "")
248          if error == "authorization_pending":
249              print(".", end="", flush=True)
250              continue
251          elif error == "slow_down":
252              # RFC 8628: add 5 seconds to polling interval
253              server_interval = result.get("interval")
254              if isinstance(server_interval, (int, float)) and server_interval > 0:
255                  interval = int(server_interval)
256              else:
257                  interval += 5
258              print(".", end="", flush=True)
259              continue
260          elif error == "expired_token":
261              print()
262              print("  ✗ Device code expired. Please try again.")
263              return None
264          elif error == "access_denied":
265              print()
266              print("  ✗ Authorization was denied.")
267              return None
268          elif error:
269              print()
270              print(f"  ✗ Authorization failed: {error}")
271              return None
272  
273      print()
274      print("  ✗ Timed out waiting for authorization.")
275      return None
276  
277  
278  # ─── Copilot Token Exchange ────────────────────────────────────────────────
279  
280  # Module-level cache for exchanged Copilot API tokens.
281  # Maps raw_token_fingerprint -> (api_token, expires_at_epoch).
282  _jwt_cache: dict[str, tuple[str, float]] = {}
283  _JWT_REFRESH_MARGIN_SECONDS = 120  # refresh 2 min before expiry
284  
285  # Token exchange endpoint and headers (matching VS Code / Copilot CLI)
286  _TOKEN_EXCHANGE_URL = "https://api.github.com/copilot_internal/v2/token"
287  _EDITOR_VERSION = "vscode/1.104.1"
288  _EXCHANGE_USER_AGENT = "GitHubCopilotChat/0.26.7"
289  
290  
291  def _token_fingerprint(raw_token: str) -> str:
292      """Short fingerprint of a raw token for cache keying (avoids storing full token)."""
293      import hashlib
294      return hashlib.sha256(raw_token.encode()).hexdigest()[:16]
295  
296  
297  def exchange_copilot_token(raw_token: str, *, timeout: float = 10.0) -> tuple[str, float]:
298      """Exchange a raw GitHub token for a short-lived Copilot API token.
299  
300      Calls ``GET https://api.github.com/copilot_internal/v2/token`` with
301      the raw GitHub token and returns ``(api_token, expires_at)``.
302  
303      The returned token is a semicolon-separated string (not a standard JWT)
304      used as ``Authorization: Bearer <token>`` for Copilot API requests.
305  
306      Results are cached in-process and reused until close to expiry.
307      Raises ``ValueError`` on failure.
308      """
309      import urllib.request
310  
311      fp = _token_fingerprint(raw_token)
312  
313      # Check cache first
314      cached = _jwt_cache.get(fp)
315      if cached:
316          api_token, expires_at = cached
317          if time.time() < expires_at - _JWT_REFRESH_MARGIN_SECONDS:
318              return api_token, expires_at
319  
320      req = urllib.request.Request(
321          _TOKEN_EXCHANGE_URL,
322          method="GET",
323          headers={
324              "Authorization": f"token {raw_token}",
325              "User-Agent": _EXCHANGE_USER_AGENT,
326              "Accept": "application/json",
327              "Editor-Version": _EDITOR_VERSION,
328          },
329      )
330  
331      try:
332          with urllib.request.urlopen(req, timeout=timeout) as resp:
333              data = json.loads(resp.read().decode())
334      except Exception as exc:
335          raise ValueError(f"Copilot token exchange failed: {exc}") from exc
336  
337      api_token = data.get("token", "")
338      expires_at = data.get("expires_at", 0)
339      if not api_token:
340          raise ValueError("Copilot token exchange returned empty token")
341  
342      # Convert expires_at to float if needed
343      expires_at = float(expires_at) if expires_at else time.time() + 1800
344  
345      _jwt_cache[fp] = (api_token, expires_at)
346      logger.debug(
347          "Copilot token exchanged, expires_at=%s",
348          expires_at,
349      )
350      return api_token, expires_at
351  
352  
353  def get_copilot_api_token(raw_token: str) -> str:
354      """Exchange a raw GitHub token for a Copilot API token, with fallback.
355  
356      Convenience wrapper: returns the exchanged token on success, or the
357      raw token unchanged if the exchange fails (e.g. network error, unsupported
358      account type). This preserves existing behaviour for accounts that don't
359      need exchange while enabling access to internal-only models for those that do.
360      """
361      if not raw_token:
362          return raw_token
363      try:
364          api_token, _ = exchange_copilot_token(raw_token)
365          return api_token
366      except Exception as exc:
367          logger.debug("Copilot token exchange failed, using raw token: %s", exc)
368          return raw_token
369  
370  
371  # ─── Copilot API Headers ───────────────────────────────────────────────────
372  
373  def copilot_request_headers(
374      *,
375      is_agent_turn: bool = True,
376      is_vision: bool = False,
377  ) -> dict[str, str]:
378      """Build the standard headers for Copilot API requests.
379  
380      Replicates the header set used by opencode and the Copilot CLI.
381      """
382      headers: dict[str, str] = {
383          "Editor-Version": "vscode/1.104.1",
384          "User-Agent": "HermesAgent/1.0",
385          "Copilot-Integration-Id": "vscode-chat",
386          "Openai-Intent": "conversation-edits",
387          "x-initiator": "agent" if is_agent_turn else "user",
388      }
389      if is_vision:
390          headers["Copilot-Vision-Request"] = "true"
391  
392      return headers