import { createServer } from 'node:http';
import { randomUUID } from 'node:crypto';
import { WebSocketServer } from 'ws';
export const ACCESS = 'fake-access';
export const PROXY_TOKEN = 'proxy-secret';
export const RUNTIME = 'ep-1';
function frame(msgType, parent, content) {
return {
header: { msg_id: randomUUID(), msg_type: msgType, version: '5.3' },
parent_header: { msg_id: parent },
channel: msgType === 'execute_reply' ? 'shell' : 'iopub',
content,
};
}
function defaultExecute({ send, msgId, code }) {
send(frame('status', msgId, { execution_state: 'busy' }));
if (code === 'boom') {
send(frame('error', msgId, { ename: 'ValueError', evalue: 'boom', traceback: [] }));
send(frame('execute_reply', msgId, { status: 'error' }));
} else {
send(frame('stream', msgId, { name: 'stdout', text: `out:${code}` }));
send(frame('execute_reply', msgId, { status: 'ok', execution_count: 1 }));
}
send(frame('status', msgId, { execution_state: 'idle' }));
}
async function readBody(req) {
const chunks = [];
for await (const chunk of req) chunks.push(chunk);
const text = Buffer.concat(chunks).toString();
return text ? JSON.parse(text) : undefined;
}
// A local Colab control plane plus Jupyter runtime proxy on one 127.0.0.1 port.
// Tests tweak the exported object: `routes` (return true to claim a request), `execute`,
// `assignments`, `kernels` and `files`.
export async function startFake() {
const fake = {
requests: [],
sockets: [],
kernels: [],
files: new Map([['', { type: 'directory' }]]),
assignments: [RUNTIME],
routes: () => false,
execute: defaultExecute,
binary: false,
proxyToken: PROXY_TOKEN,
xsrf: 'xsrf-1',
dealloc: 'dealloc-1',
};
const wss = new WebSocketServer({ noServer: true });
const json = (res, status, body) => {
res.writeHead(status, { 'content-type': 'application/json' });
res.end(typeof body === 'string' ? body : JSON.stringify(body));
return true;
};
const count = (method, path) =>
fake.requests.filter((r) => r.method === method && r.path === path).length;
Object.assign(fake, { count });
const server = createServer(async (req, res) => {
const url = new URL(req.url, 'http://127.0.0.1');
const path = decodeURIComponent(url.pathname);
const body = await readBody(req);
const entry = {
method: req.method,
path,
query: Object.fromEntries(url.searchParams),
headers: req.headers,
body,
};
fake.requests.push(entry);
if (await fake.routes(entry, res, json)) return;
const control = path.startsWith('/tun/m/');
if (control && req.headers.authorization !== `Bearer ${ACCESS}`) return json(res, 401, {});
if (!control && req.headers['x-colab-runtime-proxy-token'] !== fake.proxyToken)
return json(res, 401, {});
if (path === '/tun/m/assign') {
if (req.method === 'GET') return json(res, 200, { token: fake.xsrf });
if (req.headers['x-goog-colab-token'] !== fake.xsrf) return json(res, 403, {});
return json(res, 200, {
endpoint: RUNTIME,
variant: 'GPU',
accelerator: entry.query.accelerator,
});
}
if (path.startsWith('/tun/m/unassign/')) {
if (req.method === 'GET') return json(res, 200, { token: fake.dealloc });
return req.headers['x-goog-colab-token'] === fake.dealloc
? json(res, 200, {})
: json(res, 403, {});
}
if (path === '/tun/m/assignments') {
const base = `http://127.0.0.1:${server.address().port}`;
return json(res, 200, {
assignments: fake.assignments.map((endpoint) => ({
endpoint,
variant: 'GPU',
accelerator: 'T4',
runtimeProxyInfo: { url: base, token: fake.proxyToken },
})),
});
}
if (path === '/tun/m/ccu-info')
return json(res, 200, { currentBalance: 42, consumptionRateHourly: 1.5 });
if (path === '/api/kernels') {
if (req.method === 'POST') {
const kernel = {
id: `k${fake.kernels.length + 1}`,
name: 'python3',
execution_state: 'idle',
};
fake.kernels.push(kernel);
return json(res, 201, kernel);
}
return json(res, 200, fake.kernels);
}
if (/^\/api\/kernels\/[^/]+\/interrupt$/.test(path)) return json(res, 204, '');
if (path.startsWith('/api/contents/')) {
const name = path.slice('/api/contents/'.length);
if (req.method === 'PUT') {
const buf =
body.format === 'base64'
? Buffer.from(body.content, 'base64')
: Buffer.from(body.content);
fake.files.set(name, { type: 'file', data: buf });
return json(res, 200, { name, path: name, type: 'file' });
}
const item = fake.files.get(name);
if (!item) return json(res, 404, { message: 'SECRET-RAW-BODY' });
if (item.type === 'directory') {
const prefix = name ? `${name}/` : '';
const content = [...fake.files]
.filter(([p]) => p && p.startsWith(prefix) && !p.slice(prefix.length).includes('/'))
.map(([p, f]) => ({
name: p.slice(prefix.length),
path: p,
type: f.type,
size: f.data?.length ?? null,
last_modified: '2026-01-01T00:00:00Z',
content: 'must-not-leak',
}));
return json(res, 200, { type: 'directory', path: name, content });
}
const format = entry.query.format;
return json(res, 200, {
type: 'file',
path: name,
format,
content: item.data.toString(format === 'base64' ? 'base64' : 'utf8'),
});
}
json(res, 404, {});
});
server.on('upgrade', (req, socket, head) => {
const url = new URL(req.url, 'http://127.0.0.1');
fake.requests.push({
method: 'WS',
path: url.pathname,
query: Object.fromEntries(url.searchParams),
headers: req.headers,
});
if (req.headers['x-colab-runtime-proxy-token'] !== fake.proxyToken)
return void socket.destroy();
wss.handleUpgrade(req, socket, head, (ws) => {
fake.sockets.push(ws);
ws.on('message', (data) => {
const msg = JSON.parse(data.toString());
const send = (frameBody) => {
const text = JSON.stringify(frameBody);
if (!fake.binary) return ws.send(text);
const json = Buffer.from(text);
const head = Buffer.alloc(8);
head.writeUInt32BE(1, 0);
head.writeUInt32BE(8, 4);
ws.send(Buffer.concat([head, json]));
};
fake.execute({ ws, send, msg, msgId: msg.header.msg_id, code: msg.content.code, frame });
});
});
});
await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));
fake.url = `http://127.0.0.1:${server.address().port}`;
fake.close = async () => {
for (const ws of fake.sockets) ws.terminate();
server.closeAllConnections();
await new Promise((resolve) => server.close(resolve));
};
return fake;
}