Something went wrong. Try again.
opencode v2 mirror
Something went wrong. Try again.
11 kB · 327 lines
TypeScript
at working
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328export * as BrowserProxy from "./proxy.js"
import { randomBytes, timingSafeEqual } from "node:crypto"import { Agent, createServer, request, type IncomingHttpHeaders, type IncomingMessage, type ServerResponse,} from "node:http"import { Duplex } from "node:stream"import { Schema } from "effect"import { Browser } from "./rpc.js"
export type Transport = { open(target: Browser.TunnelTarget, signal: AbortSignal): Promise<string> read(id: string, signal: AbortSignal): Promise<Browser.TunnelRead> write(id: string, data: Uint8Array, end: boolean, signal: AbortSignal): Promise<void> close(id: string): Promise<void>}export type Proxy = Awaited<ReturnType<typeof make>>
// Desktop-only leaf. This listener is never loaded by the server plugin.export async function make(transport: Transport) { const username = randomBytes(16).toString("hex") const password = randomBytes(32).toString("hex") const expected = Buffer.from(`Basic ${Buffer.from(`${username}:${password}`).toString("base64")}`) const clients = new Set<Duplex>() const tunnels = new Set<Duplex>() const pending = new Set<AbortController>() let closed = false const authorized = (value: string | undefined) => { if (!value) return false const actual = Buffer.from(value) return actual.length === expected.length && timingSafeEqual(actual, expected) } const connect = async (target: Browser.TunnelTarget, signal: AbortSignal) => { if (closed) throw new Error("Browser proxy is closed") const abort = new AbortController() const cancel = () => abort.abort() signal.addEventListener("abort", cancel, { once: true }) if (signal.aborted) cancel() pending.add(abort) try { const id = await transport.open(target, abort.signal) const socket = new TunnelSocket(transport, id) if (closed || abort.signal.aborted) { socket.destroy() throw new Error("Browser proxy connection was cancelled") } tunnels.add(socket) socket.once("close", () => tunnels.delete(socket)) return socket } finally { pending.delete(abort) signal.removeEventListener("abort", cancel) } } const server = createServer({ maxHeaderSize: 64 * 1024 }, (incoming, response) => { void forward(incoming, response, connect, authorized).catch(() => { if (!response.headersSent) { response.writeHead(502) response.end() return } response.destroy() }) }) server.requestTimeout = 30_000 server.headersTimeout = 10_000 server.on("connection", (socket) => { clients.add(socket) socket.on("error", () => socket.destroy()) socket.once("close", () => clients.delete(socket)) }) const upgrade = (incoming: IncomingMessage, socket: Duplex, head: Buffer, connectMethod: boolean) => { void (async () => { if (!authorized(incoming.headers["proxy-authorization"])) { socket.end( 'HTTP/1.1 407 Proxy Authentication Required\r\nProxy-Authenticate: Basic realm="OpenCode Browser Proxy"\r\nContent-Length: 0\r\nConnection: close\r\n\r\n', ) return } const url = parseURL(connectMethod ? `https://${incoming.url ?? ""}` : incoming.url) if (!url || (!connectMethod && incoming.headers.upgrade?.toLowerCase() !== "websocket")) { socket.end("HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") return } const abort = new AbortController() const cancel = () => abort.abort() socket.once("close", cancel) socket.pause() try { const tunnel = await connect(target(url), abort.signal) if (socket.destroyed) { tunnel.destroy() return } if (connectMethod) socket.write("HTTP/1.1 200 Connection Established\r\n\r\n") if (!connectMethod) { const headers = forwardedHeaders(incoming.headers) headers.host = url.host headers.connection = "Upgrade" headers.upgrade = "websocket" tunnel.write( `${incoming.method} ${url.pathname}${url.search} HTTP/1.1\r\n${Object.entries(headers) .flatMap(([key, value]) => value === undefined ? [] : (Array.isArray(value) ? value : [value]).map((item) => `${key}: ${item}\r\n`), ) .join("")}\r\n`, ) } if (head.byteLength) tunnel.write(head) socket.once("close", () => tunnel.destroy()) tunnel.once("close", () => socket.destroy()) socket.pipe(tunnel) tunnel.pipe(socket) socket.resume() } finally { socket.off("close", cancel) } })().catch(() => { if (!socket.destroyed) socket.end("HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") }) } server.on("connect", (incoming, socket, head) => upgrade(incoming, socket, head, true)) server.on("upgrade", (incoming, socket, head) => upgrade(incoming, socket, head, false)) server.on("clientError", (_error, socket) => { if (!socket.destroyed) socket.end("HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n") }) await new Promise<void>((resolve, reject) => { server.once("error", reject) server.listen(0, "127.0.0.1", () => { server.off("error", reject) resolve() }) }) const address = server.address() if (!address || typeof address === "string") throw new Error("Browser proxy did not bind a TCP address") let closing: Promise<void> | undefined return { url: `http://127.0.0.1:${address.port}`, host: "127.0.0.1", port: address.port, credentials: { username, password }, close() { if (closing) return closing closed = true pending.forEach((abort) => abort.abort()) tunnels.forEach((socket) => socket.destroy()) clients.forEach((socket) => socket.destroy()) closing = new Promise<void>((resolve) => server.close(() => resolve())) return closing }, }}
async function forward( incoming: IncomingMessage, response: ServerResponse, connect: (target: Browser.TunnelTarget, signal: AbortSignal) => Promise<Duplex>, authorized: (value: string | undefined) => boolean,) { if (!authorized(incoming.headers["proxy-authorization"])) { response.writeHead(407, { "Proxy-Authenticate": 'Basic realm="OpenCode Browser Proxy"' }) response.end() return } const url = parseURL(incoming.url) if (!url || url.protocol !== "http:") { response.writeHead(400) response.end() return } const abort = new AbortController() const cancel = () => abort.abort() incoming.once("aborted", cancel) response.once("close", cancel) const agent = new Agent({ keepAlive: false, maxSockets: 1 }) try { const tunnel = await connect(target(url), abort.signal) agent.createConnection = () => tunnel const headers = forwardedHeaders(incoming.headers) headers.host = url.host headers.connection = "close" await new Promise<void>((resolve, reject) => { const upstream = request( { agent, hostname: url.hostname, port: url.port || 80, path: `${url.pathname}${url.search}`, method: incoming.method, headers, signal: abort.signal, }, (result) => { response.writeHead(result.statusCode ?? 502, result.statusMessage, { ...forwardedHeaders(result.headers), connection: "close", }) result.once("error", reject) response.once("finish", resolve) result.pipe(response) }, ) upstream.once("error", reject) incoming.pipe(upstream) }) } finally { incoming.off("aborted", cancel) response.off("close", cancel) agent.destroy() }}
function forwardedHeaders(input: IncomingHttpHeaders) { const headers = { ...input } headers.connection?.split(",").forEach((name) => delete headers[name.trim().toLowerCase()]) ;[ "connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "proxy-connection", "te", "trailer", "transfer-encoding", "upgrade", ].forEach((name) => delete headers[name]) return headers}
function parseURL(value: string | undefined) { if (!value || !URL.canParse(value)) return const url = new URL(value) if (!["http:", "https:", "ws:", "wss:"].includes(url.protocol) || url.username || url.password) return return url}
function target(url: URL) { return Schema.decodeUnknownSync(Browser.TunnelTarget)({ host: url.hostname.replace(/^\[|\]$/g, ""), port: url.port ? Number(url.port) : url.protocol === "https:" || url.protocol === "wss:" ? 443 : 80, })}
class TunnelSocket extends Duplex { readonly connecting = false private readonly abort = new AbortController() private pending = false
constructor( private readonly transport: Transport, private readonly id: string, ) { super({ highWaterMark: Browser.TUNNEL_CHUNK_BYTES, allowHalfOpen: true }) this.on("error", () => this.destroy()) } override _read() { if (this.pending || this.destroyed) return this.pending = true void this.transport.read(this.id, this.abort.signal).then( (result) => { this.pending = false if (this.destroyed) return if (result.eof) { this.push(null) return } if (this.push(result.data)) this._read() }, (error: unknown) => this.destroy(asError(error)), ) } override _write(chunk: Buffer | string, encoding: BufferEncoding, callback: (error?: Error | null) => void) { const data = typeof chunk === "string" ? Buffer.from(chunk, encoding) : chunk void (async () => { for (let offset = 0; offset < data.byteLength; offset += Browser.TUNNEL_CHUNK_BYTES) await this.transport.write( this.id, data.subarray(offset, offset + Browser.TUNNEL_CHUNK_BYTES), false, this.abort.signal, ) })().then( () => callback(), (error: unknown) => callback(asError(error)), ) } override _final(callback: (error?: Error | null) => void) { void this.transport.write(this.id, new Uint8Array(), true, this.abort.signal).then( () => callback(), (error: unknown) => callback(asError(error)), ) } override _destroy(error: Error | null, callback: (error?: Error | null) => void) { this.abort.abort() void this.transport .close(this.id) .catch(() => undefined) .then(() => callback(error)) } setKeepAlive() { return this } setNoDelay() { return this } setTimeout(_timeout: number, callback?: () => void) { if (callback) this.once("timeout", callback) return this } ref() { return this } unref() { return this }}
function asError(error: unknown) { return error instanceof Error ? error : new Error(String(error))}