proxy openai request with copilot

copilot-openai.py

#!/usr/bin/env python3

## Copilot OpenAI Proxy
## A http server that proxies openai requests to copilot by adding required headers
## Copilot cli must be installed and logged in
## Inspired by https://github.com/aaamoon/copilot-gpt4-service
##
## only external dependency is requests, cuz http.client is shit, so as http.server

import json
import uuid
import requests
import time
import secrets

from pathlib import Path
from http.server import BaseHTTPRequestHandler, HTTPServer

def get_access_token():
    """Get the access token for the copilot cli"""
    try:
        with open(Path.home() / '.copilot-cli-access-token', 'r') as f:
            return f.read().strip()
    except FileNotFoundError:
        raise Exception('Failed to get copilot access token, try to login with copilot cli')

def get_copilot_token():
    """Get a new token with copilot cli access token, unused for now"""

    # read from cache
    try:
        with open(Path.home() / '.copilot-openai-token', 'r') as f:
            cache = json.loads(f.read())
            if cache['expires_at'] > time.time():
                print("found valid token in cache")
                return cache['token']
    except (FileNotFoundError, json.JSONDecodeError):
        pass

    print("token not found or expired, get new token from server")
    headers = {
        'Authorization': f'token {get_access_token()}',
    }

    r = requests.get('https://api.github.com/copilot_internal/v2/token', headers=headers)
    if r.status_code != requests.codes.ok:
        raise Exception(f'Failed to get copilot token: {r.status_code} {r.text}')

    with open(Path.home() / '.copilot-openai-token', 'wb') as f:
        f.write(r.content)

    return r.json()['token']

def gen_hex_str(n):
    return secrets.token_hex(n)

class CopilotOpenAIProxy(BaseHTTPRequestHandler):
    DEFAULT_HEADERS = {
        'Editor-Version': "vscode/1.83.1",
        'Editor-Plugin-Version': "copilot-chat/0.8.0",
        'Openai-Organization': "github-copilot",
        'Openai-Intent': "conversation-panel",
        'Content-Type': "text/event-stream; charset=utf-8",
        'User-Agent': "GitHubCopilotChat/0.8.0",
        'Accept': "*/*",
        'Accept-Encoding': "gzip,deflate,br",
    }

    def __init__(self, token):
        self.copilot_token = token

    def __call__(self, *args, **kwargs):
        """Handle a request."""
        super().__init__(*args, **kwargs)

    def request_id(self):
        return gen_hex_str(4) + "-" + gen_hex_str(2) + "-" + gen_hex_str(2) + "-" + gen_hex_str(2) + "-" + gen_hex_str(6)

    def session_id(self):
        return gen_hex_str(4) + "-" + gen_hex_str(2) + "-" + gen_hex_str(2) + "-" + gen_hex_str(2) + "-" + gen_hex_str(12)

    def machine_id(self):
        return gen_hex_str(32)

    def do_POST(self):
        if self.path != '/v1/chat/completions':
            self.send_response(requests.codes.method_not_allowed, "Method Not Allowed")
            self.end_headers()

            return

        content_length = int(self.headers['Content-Length'])
        body = self.rfile.read(content_length)
        copilot_resp = self.copilot_request(body)

        if copilot_resp.status_code != requests.codes.ok:
            self.send_response(copilot_resp.status_code)
            self.end_headers()

            return

        self.send_response(requests.codes.ok)
        self.end_headers()
        self.wfile.write(copilot_resp.content)
        print(copilot_resp.content)

    def copilot_request(self, body):
        url = "https://api.githubcopilot.com/chat/completions"

        headers = self.DEFAULT_HEADERS.copy()
        headers['Authorization'] = f'Bearer {self.copilot_token}'
        headers['X-Request-ID'] = self.request_id()
        headers['Vscode-Sessionid'] = self.session_id()
        headers['Vscode-Machineid'] = self.machine_id()

        return requests.post(url, headers=headers, data=body)
    pass

if __name__ == '__main__':
    token = get_copilot_token()
    api_proxy = CopilotOpenAIProxy(token)

    server_address = ('127.0.0.1', 8080)
    httpd = HTTPServer(server_address, api_proxy)
    httpd.serve_forever()