"""GitHub Copilot device sign-in and account-scoped inference, offline. The Copilot extension is a device-code OAuth client, an account-owned live roster, and a rotation pool. Only an end-to-end run against a fake GitHub catches a token leaked into a public response, a roster or request served under the wrong account, a silent cross-account entitlement fallback, or a retry under a denied identity. Every request stays on the loopback fake. """ import json import time import unittest from github_copilot_support import CopilotProvider from harness import Albedo, error, exclusive, operation_id, python, text CHAT = "fixture-copilot-chat" RESPONSES = "fixture-copilot-responses" MESSAGES = "fixture-copilot-claude" ROUTES = {CHAT: "chat.completions", RESPONSES: "/responses", MESSAGES: "/v1/messages"} INFERENCE_PATHS = ("/chat/completions", "/responses", "/v1/messages") def roster(*ids): """Roster rows the parser keeps: chat capabilities, tools, one route each.""" return [ { "id": id, "supported_endpoints": [ROUTES[id]], "capabilities": { "type": "chat", "limits": { "max_context_window_tokens": 128000, "max_output_tokens": 1000, }, "supports": {"tool_calls": True, "vision": False}, }, } for id in ids ] def account(url, token, user, *, selected=False): """A stored Copilot credential: the canonical `endpoint`, never a refresh.""" return { "type": "oauth", "access": token, "accountId": f"{url}#{user}", "login": f"fixture-copilot-{user}", "endpoint": url, "selected": selected, "_albedo_account_id": operation_id(), } def inference(provider): return [entry for entry in provider.seen if entry["path"] in INFERENCE_PATHS] # exclusive: the fixture points daemon environment and stored credentials at # its own fake GitHub, which no shared daemon may keep. @exclusive class GitHubCopilotTest(unittest.TestCase): def fixture(self, rosters): provider = CopilotProvider(rosters) self.addCleanup(provider.close) return provider def enter(self, provider, *, model=CHAT, profile="fixture"): def ready(app): app.write_extensions( {"models": {"url": provider.url + "/unavailable-models-dev"}} ) for name in ( "GITHUB_COPILOT_OAUTH_SERVER", "GITHUB_COPILOT_GITHUB_API", "GITHUB_COPILOT_ENDPOINT", ): app.env[name] = provider.url app = Albedo( provider, prepare=ready, providers={ profile: { "extension": "github-copilot", "baseUrl": provider.url, "model": model, "protocol": "chat_completions", } }, ).__enter__() self.addCleanup(app.__exit__, None, None, None) return app def app_for(self, rosters, **kwargs): return self.enter(self.fixture(rosters), **kwargs) def seed(self, app, *accounts): app.store_secrets("accounts", {"github-copilot": list(accounts)}) def seed_pair(self, app, first): """Two accounts on one fake GitHub, the first one selected.""" accounts = [ account(app.provider.url, first, "first", selected=True), account(app.provider.url, f"{first}-second", "second"), ] self.seed(app, *accounts) return accounts def turn(self, app, prompt, session=None): session = session or app.session() app.prompt(session, prompt).close() return session, app.idle(session) def blocked(self, app, session): deadline = time.monotonic() + 10 while True: with app.api(f"/sessions/{session}?tail=0") as response: snapshot = json.load(response) if snapshot["status"]["blocking_reason"] is not None: return snapshot self.assertLess(time.monotonic(), deadline, snapshot) time.sleep(0.02) def requests_view(self, app, session): with app.api(f"/sessions/{session}/context?view=requests") as response: return json.load(response)["items"] def errors(self, snapshot): return [ line["text"] for line in snapshot["activity"]["lines"] if line["kind"] == "error" ] def patch_profile(self, app, **fields): route = "/settings?group=providers" with app.api(route) as response: response.read() etag = response.headers["ETag"] with app.api( route, {"profiles": {app.profile: fields}}, method="PATCH", headers={"If-Match": etag}, ) as response: return json.load(response)["resource"]["value"]["profiles"][app.profile] def test_device_sign_in_pends_completes_and_projects_a_public_account(self): app = self.app_for({}) login = operation_id() with app.api( f"/auth/logins/{login}", {"provider": "github-copilot", "flow": "browser", "values": {}}, method="PUT", ) as response: started = json.load(response) # The device page and its code reach the user before any token exists. self.assertIn("/login/device", started["url"]) self.assertIn("FIXT-1234", started["instructions"]) code = next( entry for entry in app.provider.seen if entry["path"] == "/login/device/code" ) self.assertEqual(code["body"]["client_id"], "Ov23li8tweQw6odWQebz") self.assertEqual(code["body"]["scope"], "read:user") # The daemon polls while authorization is pending, then completes. self.assertTrue( app.provider.polled.wait(20), "the daemon never polled the token endpoint" ) app.provider.authorised.set() deadline = time.monotonic() + 30 while True: with app.api(f"/auth/logins/{login}") as response: state = json.load(response) if state["state"] not in ("waiting", "exchanging"): break self.assertLess(time.monotonic(), deadline, state) time.sleep(0.05) self.assertEqual(state["state"], "complete", state) self.assertGreaterEqual( sum( entry["path"] == "/login/oauth/access_token" for entry in app.provider.seen ), 2, ) # The stored account is public: its endpoint, never its token. with app.api("/auth") as response: page = response.read().decode() saved = [ item for item in json.loads(page)["accounts"] if item["provider"] == "github-copilot" ] self.assertEqual(len(saved), 1, saved) self.assertEqual(saved[0]["endpoint"], app.provider.url) self.assertEqual(saved[0]["source"], "saved") self.assertNotIn(app.provider.oauth_token, page) paths = [entry["path"] for entry in app.provider.seen] self.assertIn("/user", paths) self.assertIn("/copilot_internal/user", paths) def test_each_advertised_route_serves_inference_with_copilot_params(self): token = "fixture-copilot-route-token" rosters = {token: roster(CHAT, RESPONSES, MESSAGES)} chat = self.fixture(rosters) chat.script = lambda _: [python("print(21 * 2)"), text("chat done")] chat_app = self.enter(chat, model=CHAT, profile="chat") self.seed(chat_app, account(chat.url, token, "first", selected=True)) session, snapshot = self.turn(chat_app, "use the chat route") self.assertTrue(inference(chat), self.errors(snapshot)) self.turn(chat_app, "and once more", session=session) headers = [entry["headers"] for entry in inference(chat)] self.assertEqual( [header["x-initiator"] for header in headers], ["user", "agent", "user"] ) self.assertEqual(headers[-1]["x-interaction-type"], "conversation-user") self.assertEqual( [entry["path"] for entry in inference(chat)], ["/chat/completions"] * 3 ) body = chat.requests[0]["request"] self.assertNotIn("service_tier", body) self.assertNotIn("reasoning_effort", body) for tool in body["tools"]: self.assertNotIn("strict", tool["function"]) # The python cell ran and its answer travelled back to the model. self.assertIn("42", json.dumps(chat.requests[1]["request"])) self.assertIn("chat done", json.dumps(chat_app.history(session)["items"][-1])) for model, path in ((RESPONSES, "/responses"), (MESSAGES, "/v1/messages")): provider = self.fixture(rosters) provider.script = lambda _: [ python("print(6 * 7)"), text("portable tool done"), ] app = self.enter(provider, model=model, profile=model) self.seed(app, account(provider.url, token, "first", selected=True)) session, _ = self.turn(app, f"use the {model} route") sent = inference(provider) self.assertEqual([entry["path"] for entry in sent], [path, path]) self.assertEqual( [entry["headers"]["x-initiator"] for entry in sent], ["user", "agent"] ) self.assertIn("42", json.dumps(provider.requests[1]["request"])) self.assertEqual( [entry["headers"]["authorization"] for entry in sent], [f"Bearer {token}", f"Bearer {token}"], ) if path == "/v1/messages": self.assertEqual(sent[0]["headers"]["anthropic-version"], "2023-06-01") self.assertIn( "portable tool done", json.dumps(app.history(session)["items"][-1]) ) def test_auto_negotiates_hidden_models_with_conversation_scoped_grants(self): access = "fixture-auto-access" rows = roster(RESPONSES) rows[0]["model_picker_enabled"] = False provider = self.fixture({access: rows}) def grant(): number = sum(e["path"] == "/models/session" for e in provider.seen) return 200, { "available_models": [RESPONSES], "selected_model": RESPONSES, "session_token": f"fixture-auto-grant-{number}", "expires_at": time.time() + 3600, } provider.auto_sessions[access] = grant provider.script = lambda _: [python("print(6 * 7)"), text("auto done")] app = self.enter(provider, model="auto") self.seed(app, account(provider.url, access, "first", selected=True)) with app.api( f"/models?provider=github-copilot&endpoint={provider.url}" ) as response: listed = json.load(response)["items"] self.assertEqual([model["id"] for model in listed], ["auto"]) self.assertEqual(listed[0]["default_context_tokens"], 128000) self.assertEqual(listed[0]["max_output_tokens"], 1000) self.assertEqual(listed[0]["efforts"], []) session, snapshot = self.turn(app, "use auto") self.assertFalse(self.errors(snapshot)) sent = inference(provider) self.assertEqual([e["path"] for e in sent], ["/responses", "/responses"]) self.assertEqual( [e["headers"]["copilot-session-token"] for e in sent], ["fixture-auto-grant-2"] * 2, ) self.assertEqual([e["headers"]["x-initiator"] for e in sent], ["user", "agent"]) self.assertTrue( all(r["request"]["model"] == RESPONSES for r in provider.requests) ) self.assertIn("42", json.dumps(provider.requests[1]["request"])) self.assertIn("auto done", json.dumps(app.history(session)["items"][-1])) def reload(): with app.api( f"/sessions/{session}/reload", {"target": "models"} ) as response: return next( item for item in json.load(response)["models"] if item["provider"] == "github-copilot" ) self.assertEqual(reload()["state"], "refreshed") self.turn(app, "same conversation", session=session) self.assertEqual( inference(provider)[-1]["headers"]["copilot-session-token"], "fixture-auto-grant-2", ) self.turn(app, "a new conversation") self.assertEqual( inference(provider)[-1]["headers"]["copilot-session-token"], "fixture-auto-grant-4", ) grants = [e for e in provider.seen if e["path"] == "/models/session"] self.assertEqual(len(grants), 4) for request in grants: self.assertEqual(request["body"], {"auto_mode": {"model_hints": ["auto"]}}) self.assertEqual( request["headers"]["copilot-integration-id"], "copilot-chat" ) self.assertNotIn("copilot-session-token", request["headers"]) self.assertTrue( all( "fixture-auto-grant-" not in path.read_text() for path in (app.home / "model-lists").glob("*.json") ) ) del provider.auto_sessions[access] self.assertEqual(reload()["state"], "failed") with app.api( f"/models?provider=github-copilot&endpoint={provider.url}" ) as response: self.assertEqual( [item["id"] for item in json.load(response)["items"]], ["auto"] ) self.turn(app, "still the same conversation", session=session) self.assertEqual( inference(provider)[-1]["headers"]["copilot-session-token"], "fixture-auto-grant-2", ) def test_auto_does_not_claim_limits_for_an_unknown_backing_pool_member(self): access = "fixture-auto-unknown-pool-access" provider = self.fixture({access: roster(RESPONSES)}) provider.auto_sessions[access] = ( 200, { "available_models": [RESPONSES, "unpublished-backing-model"], "selected_model": RESPONSES, "session_token": "fixture-auto-unknown-pool-grant", "expires_at": time.time() + 3600, }, ) app = self.enter(provider, model="auto") self.seed(app, account(provider.url, access, "first", selected=True)) with app.api( f"/models?provider=github-copilot&endpoint={provider.url}" ) as response: listed = json.load(response)["items"] automatic = next(model for model in listed if model["id"] == "auto") self.assertIsNone(automatic["default_context_tokens"]) self.assertIsNone(automatic["max_output_tokens"]) def test_auto_refuses_invalid_grants_without_borrowing_another_account(self): access = "fixture-auto-invalid-access" base = { "available_models": [RESPONSES], "session_token": "fixture-private-grant", "expires_at": time.time() + 3600, } cases = [ (200, {**base, "selected_model": "outside-the-pool"}), (200, {**base, "selected_model": 123}), (200, {**base, "expires_at": time.time() - 1}), (403, {"error": {"message": access + " fixture-private-grant"}}), ] for status, payload in cases: with self.subTest(status=status, payload=payload): provider = self.fixture( {access: roster(RESPONSES), access + "-second": roster(RESPONSES)} ) provider.auto_sessions[access] = (status, payload) app = self.enter(provider, model="auto") accounts = self.seed_pair(app, access) self.patch_profile( app, account_ids=[a["_albedo_account_id"] for a in accounts] ) session = app.session() app.prompt(session, "use auto").close() snapshot = self.blocked(app, session) self.assertEqual( snapshot["status"]["blocking_reason"]["code"], "inputs_blocked" ) self.assertEqual(inference(provider), []) grants = [e for e in provider.seen if e["path"] == "/models/session"] self.assertEqual(len(grants), 1) self.assertEqual( grants[0]["headers"]["authorization"], f"Bearer {access}" ) public = json.dumps(snapshot) self.assertNotIn(access, public) self.assertNotIn("fixture-private-grant", public) with app.api("/auth") as response: self.assertEqual(len(json.load(response)["accounts"]), 2) def test_hidden_models_are_not_manually_selectable(self): access = "fixture-hidden-access" rows = roster(RESPONSES) rows[0]["model_picker_enabled"] = False app = self.app_for({access: rows}, model=RESPONSES) self.seed(app, account(app.provider.url, access, "first", selected=True)) session = app.session() app.prompt(session, "select the hidden backend directly").close() snapshot = self.blocked(app, session) self.assertEqual(inference(app.provider), []) self.assertTrue( any("choose auto" in line for line in self.errors(snapshot)), self.errors(snapshot), ) def test_auto_inference_denials_redact_both_credentials_without_identity_retry( self, ): access, token = "fixture-auto-denied-access", "fixture-auto-denied-grant" provider = self.fixture({access: roster(RESPONSES)}) provider.auto_sessions[access] = ( 200, { "available_models": [RESPONSES], "session_token": token, "expires_at": time.time() + 3600, }, ) provider.script = lambda _: error(403, f"denied {access} {token}") app = self.enter(provider, model="auto") self.seed(app, account(provider.url, access, "first", selected=True)) session, snapshot = self.turn(app, "use auto") sent = inference(provider) self.assertEqual(len(sent), 1) self.assertEqual(sent[0]["headers"]["copilot-session-token"], token) self.assertEqual(sent[0]["headers"]["copilot-integration-id"], "copilot-chat") explained = json.dumps(self.errors(snapshot)) self.assertIn("[redacted]", explained) self.assertNotIn(access, explained) self.assertNotIn(token, explained) self.assertEqual(self.requests_view(app, session)[-1]["outcome"], "failed") def test_a_profile_pin_never_borrows_another_accounts_roster(self): first = "fixture-first-token" second = f"{first}-second" app = self.app_for( {first: roster(CHAT), second: roster(RESPONSES)}, ) accounts = self.seed_pair(app, first) session, _ = self.turn(app, "use the chat route") self.assertEqual( [entry["path"] for entry in inference(app.provider)], ["/chat/completions"] ) self.patch_profile(app, account_id=accounts[1]["_albedo_account_id"]) before = len(inference(app.provider)) blocked = app.session() app.prompt(blocked, "use the chat route again").close() snapshot = self.blocked(app, blocked) self.assertEqual(len(inference(app.provider)), before) self.assertEqual( snapshot["status"]["blocking_reason"]["code"], "inputs_blocked" ) self.assertEqual(len(snapshot["pending_inputs"]), 1) asked = [ entry for entry in app.provider.seen if entry["path"] == "/models" and entry["headers"]["authorization"] == f"Bearer {second}" ] self.assertTrue(asked, "the roster was never asked under the new account") self.assertTrue( any( "does not advertise this model" in line for line in self.errors(snapshot) ), self.errors(snapshot), ) self.patch_profile(app, account_id=accounts[0]["_albedo_account_id"]) app.idle(blocked) self.assertEqual(len(inference(app.provider)), before + 1) self.assertEqual( inference(app.provider)[-1]["headers"]["authorization"], f"Bearer {first}" ) self.assertIn("copilot ready", json.dumps(app.history(blocked)["items"][-1])) def test_an_empty_roster_lists_no_models_and_none_fall_back(self): token = "fixture-empty-roster-token" app = self.app_for({token: []}) self.seed(app, account(app.provider.url, token, "first", selected=True)) (app.home / "models.json").write_text( json.dumps( { "github-copilot": { "id": "github-copilot", "models": {"unavailable-model": {"id": "unavailable-model"}}, } } ) ) with app.api("/models?provider=github-copilot") as response: self.assertEqual(json.load(response)["items"], []) def test_remote_lists_reload_and_survive_restart_and_fetch_failure(self): access = "fixture-persistent-list-access" provider = self.fixture({access: roster(CHAT)}) app = self.enter(provider) self.seed(app, account(provider.url, access, "first", selected=True)) session = app.session() def listed(): with app.api( f"/models?provider=github-copilot&endpoint={provider.url}" ) as response: return [item["id"] for item in json.load(response)["items"]] def reload(): with app.api( f"/sessions/{session}/reload", {"target": "models"} ) as response: return next( item for item in json.load(response)["models"] if item["provider"] == "github-copilot" ) self.assertEqual(listed(), [CHAT]) provider.rosters[access] = roster(RESPONSES) self.assertEqual(listed(), [CHAT]) self.assertEqual(reload()["state"], "refreshed") self.assertEqual(listed(), [RESPONSES]) saved = list((app.home / "model-lists").glob("*.json")) self.assertTrue(saved) self.assertTrue(all(access not in path.read_text() for path in saved)) del provider.rosters[access] app.restart() before = len([e for e in provider.seen if e["path"] == "/models"]) self.assertEqual(listed(), [RESPONSES]) self.assertEqual( len([e for e in provider.seen if e["path"] == "/models"]), before ) outcome = reload() self.assertEqual(outcome["state"], "failed") self.assertTrue(outcome["failure"]) self.assertEqual(listed(), [RESPONSES]) def test_an_explicit_rate_limit_rotates_to_the_next_permitted_account(self): first = "fixture-first-token" app = self.app_for( {first: roster(CHAT), f"{first}-second": roster(CHAT)}, ) accounts = self.seed_pair(app, first) self.patch_profile( app, account_ids=[ accounts[0]["_albedo_account_id"], accounts[1]["_albedo_account_id"], ], ) app.provider.script = lambda _: [error(429, "rate limited"), text("rotated")] session, _ = self.turn(app, "which account") self.assertEqual( [entry["headers"]["authorization"] for entry in inference(app.provider)], [f"Bearer {first}", f"Bearer {first}-second"], ) rows = self.requests_view(app, session) self.assertEqual(rows[-1]["outcome"], "completed") self.assertEqual(rows[-1]["account_id"], accounts[1]["login"]) with app.api("/auth") as response: saved = {item["id"]: item for item in json.load(response)["accounts"]} self.assertIsNotNone(saved[accounts[0]["_albedo_account_id"]]["limited_until"]) def test_a_denied_request_fails_without_retrying_another_identity(self): first = "fixture-first-token" app = self.app_for( {first: roster(CHAT), f"{first}-second": roster(CHAT)}, ) accounts = self.seed_pair(app, first) self.patch_profile( app, account_ids=[ accounts[0]["_albedo_account_id"], accounts[1]["_albedo_account_id"], ], ) # The backend echoes the account token back in its denial; the # explanation the user sees must carry the redaction, never the token. app.provider.script = lambda _: error(403, f"model access denied for {first}") session, snapshot = self.turn(app, "which account") self.assertEqual(len(app.provider.requests), 1, self.errors(snapshot)) self.assertEqual( [entry["headers"]["authorization"] for entry in inference(app.provider)], [f"Bearer {first}"], ) self.assertTrue( any("denied the request" in line for line in self.errors(snapshot)), self.errors(snapshot), ) self.assertEqual(self.requests_view(app, session)[-1]["outcome"], "failed") explained = json.dumps(self.errors(snapshot)) self.assertIn("[redacted]", explained) self.assertNotIn(first, explained)