Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145"""GitHub's side of a Copilot sign-in, for github_copilot_test.py.
Serves the device flow the daemon starts (POST /login/device/code), the token itpolls for (POST /login/oauth/access_token), and the Copilot entitlement andsession-token lookups it makes (/copilot_internal/user and/copilot_internal/v2/token). ``authorize()`` is the person approving the code ina browser; every request is kept in ``seen`` so a test can assert what GitHubwas told."""
import jsonimport threadingimport timefrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerfrom typing import castfrom urllib.parse import parse_qs, urlparse
ACCESS_TOKEN = "github-fixture-access"COPILOT_TOKEN = "copilot-fixture-token"
class Github(ThreadingHTTPServer): """A fake github.com and api.github.com for one test class."""
daemon_threads = True
def __init__(self, api: str = "https://api.githubcopilot.com", interval=1): super().__init__(("127.0.0.1", 0), Handler) # The Copilot API base the account's credentials name. self.api = api self.interval = interval self.login = "octocat" self.authorized = False self.seen = [] self.lock = threading.Lock()
@property def url(self): return f"http://127.0.0.1:{self.server_address[1]}"
def start(self): threading.Thread(target=self.serve_forever, daemon=True).start() return self
def close(self) -> None: self.shutdown() self.server_close()
def authorize(self): with self.lock: self.authorized = True
def requests(self, path): with self.lock: return [entry for entry in self.seen if entry["path"] == path]
class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1"
@property def github(self) -> Github: return cast(Github, self.server)
def log_message(self, format, *args): pass
def record(self, body=None): with self.github.lock: self.github.seen.append( { "path": urlparse(self.path).path, "headers": { key.lower(): value for key, value in self.headers.items() }, "body": body or {}, } )
def do_GET(self): url = urlparse(self.path) self.record() if url.path == "/copilot_internal/user": return self.json( 200, { "login": self.github.login, "chat_enabled": True, "can_signup_for_limited": False, "endpoints": {"api": self.github.api}, }, ) if url.path == "/copilot_internal/v2/token": return self.json( 200, { "token": COPILOT_TOKEN, "expires_at": int((time.time() + 1800) * 1000), "refresh_in": 1500, "endpoints": {"api": self.github.api}, }, ) self.json(404, {"error": "not_found"})
def do_POST(self): url = urlparse(self.path) body = self.rfile.read(int(self.headers.get("content-length") or 0)) form = {key: values[0] for key, values in parse_qs(body.decode()).items()} self.record(form) if url.path == "/login/device/code": if form.get("client_id") == "": return self.json(400, {"error": "invalid_client"}) return self.json( 200, { "device_code": "device-fixture-code", "user_code": "ABCD-1234", "verification_uri": f"{self.github.url}/login/device", "expires_in": 900, "interval": self.github.interval, }, ) if url.path == "/login/oauth/access_token": with self.github.lock: authorized = self.github.authorized if not authorized: return self.json(200, {"error": "authorization_pending"}) return self.json( 200, { "access_token": ACCESS_TOKEN, "token_type": "bearer", "scope": "read:user", }, ) self.json(404, {"error": "not_found"})
def json(self, status, value): body = json.dumps(value).encode() self.send_response(status) self.send_header("content-type", "application/json") self.send_header("content-length", str(len(body))) self.end_headers() self.wfile.write(body)