feat: IPv6 outbound rotation via curl --interface

- Add IPv6SourcePool for round-robin IP rotation
- fetchViaCurl: Bun.spawn + curl -6 --interface for source binding
- Auto-fallback to regular fetch when IPv6 connection fails
- Response decompression via --compressed flag
- Docker + GHCR build/deploy workflows
- 8 routable IPv6 addresses on VPS
This commit is contained in:
MythEclipse
2026-06-19 19:07:46 +07:00
parent cb6191902e
commit 17338a7dca
13 changed files with 673 additions and 47 deletions
+23 -5
View File
@@ -31,6 +31,7 @@ import { checkBodySize } from "./middleware/body-limiter";
import { createRateLimiter } from "./middleware/rate-limiter";
import { logRelayEvent } from "./middleware/logger";
import { ProxyPool, SessionProxyPool } from "./lib/proxy-pool";
import { IPv6SourcePool } from "./lib/ipv6-pool";
import { handleChatCompletion, listModels } from "./lib/ai-proxy";
import { handleAnthropicMessages } from "./lib/anthropic-proxy";
import { fetchWithRetry, closeAllActiveReaders, isDevMode } from "./lib/fetch-utils";
@@ -40,6 +41,7 @@ import type { Server, ServerWebSocket } from "bun";
// --- Configuration ------------------------------------------------------------
const PORT = Number.parseInt(process.env.PORT ?? "3000", 10);
const HOST = process.env.HOST ?? "::";
const RELAY_TIMEOUT_MS = Number.parseInt(
process.env.RELAY_TIMEOUT_MS ?? "30000",
10,
@@ -87,6 +89,14 @@ proxyPool.tryLoad(
const sessionPool = new SessionProxyPool(proxyPool);
sessionPool.setFailureThreshold(3);
// --- IPv6 source pool (optional) ----------------------------------------------
const ipv6Pool = new IPv6SourcePool();
ipv6Pool.loadFromEnv();
if (ipv6Pool.configured) {
console.log(`[relay] IPv6 source pool loaded: ${ipv6Pool.size} addresses`);
}
// --- SSRF DNS rebinding protection --------------------------------------------
if (process.env.SSRF_DNS_CHECK === "true") {
@@ -388,12 +398,16 @@ async function handleRelay(
const targetUrlString = targetUrl.toString();
// -- Get IPv6 source for outbound binding ----------------------------------
const ipv6Source = ipv6Pool.getNext() ?? undefined;
// -- Execute upstream fetch with shared retry -------------------------------
const result = await fetchWithRetry(
targetUrlString,
fetchOptions,
proxyPool,
"relay",
ipv6Source,
);
if (result.errorClassification) {
@@ -477,6 +491,8 @@ function handleWebSocketUpgrade(
// --- Server ------------------------------------------------------------------
const server: Server<WSRelayData> = Bun.serve<WSRelayData>({
hostname: HOST,
ipv6Only: false,
port: PORT,
development: isDevMode() ? { hmr: true, console: true } : undefined,
@@ -509,8 +525,9 @@ const server: Server<WSRelayData> = Bun.serve<WSRelayData>({
try {
const body = await req.json();
const sessionId = crypto.randomUUID();
console.log(`[index] POST /v1/chat/completions session=${sessionId.slice(0, 8)} model=${(body as any).model} poolSize=${proxyPool.size} sessionPool.active=${sessionPool.activeSessions}`);
return handleChatCompletion(body, proxyPool, sessionPool, sessionId);
const ipv6Source = ipv6Pool.getNext() ?? undefined;
console.log(`[index] POST /v1/chat/completions session=${sessionId.slice(0, 8)} model=${(body as any).model} poolSize=${proxyPool.size} sessionPool.active=${sessionPool.activeSessions} ipv6=${ipv6Source ?? "none"}`);
return handleChatCompletion(body, proxyPool, sessionPool, sessionId, ipv6Source);
} catch {
return new Response(
JSON.stringify({ error: { message: "Invalid JSON body", type: "invalid_request_error" } }),
@@ -532,8 +549,9 @@ const server: Server<WSRelayData> = Bun.serve<WSRelayData>({
try {
const body = await req.json();
const sessionId = crypto.randomUUID();
console.log(`[index] POST /v1/messages session=${sessionId.slice(0, 8)} model=${(body as any).model} poolSize=${proxyPool.size} sessionPool.active=${sessionPool.activeSessions}`);
return handleAnthropicMessages(body, proxyPool, sessionPool, sessionId);
const ipv6Source = ipv6Pool.getNext() ?? undefined;
console.log(`[index] POST /v1/messages session=${sessionId.slice(0, 8)} model=${(body as any).model} poolSize=${proxyPool.size} sessionPool.active=${sessionPool.activeSessions} ipv6=${ipv6Source ?? "none"}`);
return handleAnthropicMessages(body, proxyPool, sessionPool, sessionId, ipv6Source);
} catch {
return new Response(
JSON.stringify({
@@ -659,7 +677,7 @@ const server: Server<WSRelayData> = Bun.serve<WSRelayData>({
// --- Startup -----------------------------------------------------------------
console.log(
`[relay] Edge Proxy Relay v${RELAY_VERSION} listening on http://localhost:${server.port}`,
`[relay] Edge Proxy Relay v${RELAY_VERSION} listening on http://${HOST}:${server.port}`,
);
if (isDevMode()) {
console.log("[relay] Development mode: HMR enabled");
+7 -6
View File
@@ -376,6 +376,7 @@ export async function handleChatCompletion(
proxyPool?: ProxyPool,
sessionPool?: SessionProxyPool,
sessionId?: string,
ipv6Source?: string,
): Promise<Response> {
// -- Input validation -------------------------------------------------------
const validationError = validateChatRequest(body);
@@ -420,8 +421,8 @@ export async function handleChatCompletion(
// -- Execute with session-aware or standard retry --------------------------
let result: FetchWithRetryResult =
sessionPool && sessionId
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`)
: await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`);
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined, ipv6Source)
: await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`, ipv6Source);
// -- Mimo Free: auth failure → invalidate JWT and retry once ---------------
if (
@@ -437,8 +438,8 @@ export async function handleChatCompletion(
};
result =
sessionPool && sessionId
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`)
: await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`);
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined, ipv6Source)
: await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`, ipv6Source);
}
// -- aichat.org: session expiry → invalidate session and retry once ---------
@@ -456,8 +457,8 @@ export async function handleChatCompletion(
};
result =
sessionPool && sessionId
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`)
: await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`);
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `openai:${req.model}`, undefined, ipv6Source)
: await fetchWithRetry(url, init, proxyPool, `openai:${req.model}`, ipv6Source);
}
if (result.errorClassification) {
+3 -2
View File
@@ -473,6 +473,7 @@ export async function handleAnthropicMessages(
proxyPool?: ProxyPool,
sessionPool?: SessionProxyPool,
sessionId?: string,
ipv6Source?: string,
): Promise<Response> {
// -- Input validation -------------------------------------------------------
const validationError = validateAnthropicRequest(body);
@@ -506,8 +507,8 @@ export async function handleAnthropicMessages(
// -- Execute with session-aware or standard retry --------------------------
const result: FetchWithRetryResult =
sessionPool && sessionId
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `anthropic:${req.model}`)
: await fetchWithRetry(url, init, proxyPool, `anthropic:${req.model}`);
? await fetchWithSessionRetry(url, init, sessionPool, sessionId, `anthropic:${req.model}`, undefined, ipv6Source)
: await fetchWithRetry(url, init, proxyPool, `anthropic:${req.model}`, ipv6Source);
if (result.errorClassification) {
if (sessionPool && sessionId) {
+147 -10
View File
@@ -6,6 +6,12 @@
*/
import type { ProxyPool, SessionProxyPool } from "./proxy-pool";
import type { IPv6SourcePool } from "./ipv6-pool";
// ─── Constants ─────────────────────────────────────────────────────────
/** Default relay timeout (30 seconds). Used for curl IPv6 source requests. */
const DEFAULT_TIMEOUT_MS = 30_000;
// ─── Active stream tracking (for graceful shutdown) ───────────────────────
@@ -127,11 +133,109 @@ export function sanitizeErrorMessage(raw: string): string {
return "Upstream error";
}
// ─── Fetch via curl (for IPv6 source binding) ────────────────────────────
/**
* Execute an HTTP request via curl with a specific source IPv6 address.
* Uses `Bun.spawn` to run curl with `--interface` to bind to the given address.
*
* This is used when outbound IPv6 source rotation is needed, since
* Bun's built-in fetch() does not support specifying a local address.
*
* @param url - Target URL
* @param init - Request init (method, headers, body)
* @param ipv6Source - IPv6 source address to bind to
* @param timeoutMs - Request timeout in milliseconds
*/
export async function fetchViaCurl(
url: string,
init: RequestInit,
ipv6Source: string,
timeoutMs: number,
): Promise<Response> {
const method = (init.method ?? "GET").toUpperCase();
const args = [
"curl",
"-6",
"--interface", ipv6Source,
"-X", method,
"-s", // silent mode
"--compressed", // auto-decompress gzip/brotli
"-o", "-", // output body to stdout
"-w", "\n%{http_code}", // append status code on new line after body
"--max-time", String(Math.ceil(timeoutMs / 1000)),
"--connect-timeout", "10",
];
// Add headers
if (init.headers) {
const headers = init.headers instanceof Headers
? Object.fromEntries(init.headers.entries())
: init.headers;
for (const [key, value] of Object.entries(headers)) {
if (key.toLowerCase() === "host") continue; // curl sets Host automatically
args.push("-H", `${key}: ${value}`);
}
}
// Add body for non-GET/HEAD methods
if (init.body && method !== "GET" && method !== "HEAD") {
if (typeof init.body === "string") {
args.push("-d", init.body);
} else if (init.body instanceof ArrayBuffer) {
args.push("--data-binary", "@-");
}
// Note: streaming bodies not supported via curl path
}
args.push(url);
logProxy("fetchViaCurl", `${args.slice(0, 6).join(" ")}...`, { ipv6Source, url });
const proc = Bun.spawn(args, {
stdout: "pipe",
stderr: "pipe",
stdin: "pipe",
});
// Collect output with timeout
const timeout = setTimeout(() => {
proc.kill();
}, timeoutMs + 5000);
try {
const [stdout, stderr] = await Promise.all([
new Response(proc.stdout).text(),
new Response(proc.stderr).text(),
]);
clearTimeout(timeout);
// Parse: curl -w "\n%{http_code}" appends status code on a new line after body
const lastNewline = stdout.lastIndexOf("\n");
const rawCode = parseInt(stdout.slice(lastNewline + 1), 10);
const statusCode = rawCode > 0 ? rawCode : 502;
const body = lastNewline >= 0 ? stdout.slice(0, lastNewline) : stdout;
logProxy("fetchViaCurl", `response status=${statusCode}`, { ipv6Source, url });
return new Response(body, {
status: statusCode,
statusText: statusCode === 200 ? "OK" : "Error",
headers: { "Content-Type": "text/plain" },
});
} catch (err) {
clearTimeout(timeout);
proc.kill();
throw err;
}
}
// ─── Fetch with retry (direct → proxy fallback) ─────────────────────────
export interface FetchWithRetryResult {
response?: Response;
errorClassification?: { code: string; status: number; message: string };
response?: Response;
errorClassification?: { code: string; status: number; message: string };
}
/**
@@ -141,12 +245,17 @@ export interface FetchWithRetryResult {
* Falls back to direct when no pool is available.
* Logs every failure to `console.warn` so the operator can diagnose without
* the error body leaking to the downstream client.
*
* @param ipv6Source - Optional IPv6 source address for outbound binding.
* When provided, uses curl instead of fetch() to bind
* to the specified source address.
*/
export async function fetchWithRetry(
url: string,
init: RequestInit & { proxy?: string },
proxyPool?: ProxyPool,
context?: string,
url: string,
init: RequestInit & { proxy?: string },
proxyPool?: ProxyPool,
context?: string,
ipv6Source?: string,
): Promise<FetchWithRetryResult> {
let response: Response | undefined;
let lastError: unknown;
@@ -184,8 +293,22 @@ export async function fetchWithRetry(
const proxyShort = init.proxy ? init.proxy.replace(/https?:\/\//, "").replace(/@.*/, "@***") : "direct";
logProxy("fetchWithRetry", `attempt=${attempt + 1}/${maxAttempts} proxy=${proxyShort}`, { context });
try {
response = await fetch(url, init);
try {
// Use curl for IPv6 source binding (only for direct connections)
if (ipv6Source && !init.proxy) {
response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS);
// If curl failed with connection error (502, empty body), fallback to regular fetch
if (response.status === 502) {
const cloned = response.clone();
const bodyText = await cloned.text();
if (!bodyText) {
logProxy("fetchWithRetry", `IPv6 connection failed, falling back to regular fetch`, { ipv6Source, url });
response = await fetch(url, init);
}
}
} else {
response = await fetch(url, init);
}
if (response.ok) {
if (usedProxy && proxyPool && proxyPool.size > 0) {
proxyPool.markSuccess();
@@ -281,6 +404,8 @@ function classifyFetchErrorSafe(error: unknown): {
* SSE streams: the initial request is retried normally. Once the response body
* starts streaming, mid-stream errors are **not** retried; the session is
* released and the error is returned to the caller.
*
* @param ipv6Source - Optional IPv6 source address for outbound binding.
*/
export async function fetchWithSessionRetry(
url: string,
@@ -289,12 +414,18 @@ export async function fetchWithSessionRetry(
sessionId: string,
context?: string,
maxRetries?: number,
ipv6Source?: string,
): Promise<FetchWithRetryResult> {
// Fallback when no session pool is available
if (!sessionPool) {
logProxy("fetchWithSessionRetry", "no session pool — direct fetch", { context, sessionId: sessionId.slice(0, 8) });
try {
const response = await fetch(url, init);
let response: Response;
if (ipv6Source) {
response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS);
} else {
response = await fetch(url, init);
}
return { response };
} catch (err) {
return { errorClassification: classifyFetchErrorSafe(err) };
@@ -333,7 +464,13 @@ export async function fetchWithSessionRetry(
});
try {
const response = await fetch(url, init);
// Use curl for IPv6 source binding (only for direct connections)
let response: Response;
if (ipv6Source && !init.proxy) {
response = await fetchViaCurl(url, init, ipv6Source, DEFAULT_TIMEOUT_MS);
} else {
response = await fetch(url, init);
}
if (response.ok) {
sessionPool.markSuccess(sessionId);
+118
View File
@@ -0,0 +1,118 @@
import { test, expect, describe } from "bun:test";
import { IPv6SourcePool } from "./ipv6-pool";
describe("IPv6SourcePool", () => {
test("should load addresses from comma-separated string", () => {
const pool = new IPv6SourcePool();
pool.loadFromString("2001:db8::1,2001:db8::2,2001:db8::3");
expect(pool.size).toBe(3);
expect(pool.configured).toBe(true);
});
test("should return null when empty", () => {
const pool = new IPv6SourcePool();
expect(pool.size).toBe(0);
expect(pool.configured).toBe(false);
expect(pool.getNext()).toBeNull();
});
test("should rotate through addresses round-robin", () => {
const pool = new IPv6SourcePool();
pool.loadFromString("2001:db8::1,2001:db8::2,2001:db8::3");
expect(pool.getNext()).toBe("2001:db8::1");
expect(pool.getNext()).toBe("2001:db8::2");
expect(pool.getNext()).toBe("2001:db8::3");
expect(pool.getNext()).toBe("2001:db8::1"); // wraps around
});
test("should skip failed sources", () => {
const pool = new IPv6SourcePool();
pool.loadFromString("2001:db8::1,2001:db8::2,2001:db8::3");
pool.setFailureThreshold(1);
pool.markFailed("2001:db8::2");
const results = [pool.getNext(), pool.getNext(), pool.getNext(), pool.getNext()];
// Should skip 2001:db8::2 (disabled after 1 failure with threshold=1)
expect(results).not.toContain("2001:db8::2");
});
test("should reset after all sources fail", () => {
const pool = new IPv6SourcePool();
pool.loadFromString("2001:db8::1,2001:db8::2");
// Mark all as failed
pool.markFailed("2001:db8::1");
pool.markFailed("2001:db8::1");
pool.markFailed("2001:db8::1");
pool.markFailed("2001:db8::2");
pool.markFailed("2001:db8::2");
pool.markFailed("2001:db8::2");
// Should auto-reset and return first
const addr = pool.getNext();
expect(addr).toBe("2001:db8::1");
});
test("should reset failure count on success", () => {
const pool = new IPv6SourcePool();
pool.loadFromString("2001:db8::1,2001:db8::2");
pool.markFailed("2001:db8::1");
pool.markFailed("2001:db8::1");
pool.markSuccess("2001:db8::1");
// Should not be disabled after success
const addr = pool.getNext();
expect(addr).toBe("2001:db8::1");
});
test("should get address by index", () => {
const pool = new IPv6SourcePool();
pool.loadFromString("2001:db8::1,2001:db8::2");
expect(pool.getAtIndex(0)).toBe("2001:db8::1");
expect(pool.getAtIndex(1)).toBe("2001:db8::2");
expect(pool.getAtIndex(2)).toBeNull();
});
test("should handle whitespace in input", () => {
const pool = new IPv6SourcePool();
pool.loadFromString(" 2001:db8::1 , 2001:db8::2 ");
expect(pool.size).toBe(2);
expect(pool.getNext()).toBe("2001:db8::1");
});
test("should handle empty input", () => {
const pool = new IPv6SourcePool();
pool.loadFromString("");
expect(pool.size).toBe(0);
});
test("should reset all sources", () => {
const pool = new IPv6SourcePool();
pool.loadFromString("2001:db8::1,2001:db8::2");
pool.markFailed("2001:db8::1");
pool.markFailed("2001:db8::1");
pool.markFailed("2001:db8::1");
pool.reset();
expect(pool.getNext()).toBe("2001:db8::1");
});
test("should change failure threshold", () => {
const pool = new IPv6SourcePool();
pool.loadFromString("2001:db8::1");
pool.setFailureThreshold(1);
pool.markFailed("2001:db8::1");
// With threshold 1, should be disabled after 1 failure
// getNext should auto-reset since all failed
const addr = pool.getNext();
expect(addr).toBe("2001:db8::1");
});
});
+186
View File
@@ -0,0 +1,186 @@
/**
* IPv6 source address pool for outbound rotation.
*
* Loads multiple IPv6 source addresses and rotates through them
* round-robin. Each outbound request can pick the next source IP
* via getNext(), ensuring requests are distributed across addresses.
*
* --- Environment Variables ----------------------------------------------------
* IPV6_SOURCES — Comma-separated list of IPv6 source addresses
* e.g. "2001:df4:c140:1f::d6,2001:df4:c140:1f:ffff:ffff:ffff:ffff"
*/
const POOL_PREFIX = "[ipv6-pool]";
function logPool(msg: string, extra?: Record<string, unknown>): void {
const ts = new Date().toISOString().slice(11, 23);
const parts = [`${POOL_PREFIX} ${ts}`, msg];
if (extra) {
for (const [k, v] of Object.entries(extra)) {
parts.push(`${k}=${v ?? "null"}`);
}
}
console.log(parts.join(" "));
}
// --- Types -------------------------------------------------------------------
export interface IPv6SourceEntry {
address: string;
/** Number of consecutive failures for this source */
failures: number;
/** Whether this source is temporarily disabled */
disabled: boolean;
}
// --- IPv6SourcePool ----------------------------------------------------------
export class IPv6SourcePool {
private sources: IPv6SourceEntry[] = [];
private currentIndex = 0;
private failureThreshold = 3;
/**
* Load IPv6 source addresses from a comma-separated string.
*
* @param csv - Comma-separated IPv6 addresses (e.g. "addr1,addr2,addr3")
*/
loadFromString(csv: string): void {
if (!csv) return;
const addrs = csv
.split(",")
.map((a) => a.trim())
.filter(Boolean);
this.sources = addrs.map((addr) => ({
address: addr,
failures: 0,
disabled: false,
}));
this.currentIndex = 0;
logPool(`loaded ${this.sources.length} IPv6 source addresses`);
for (const src of this.sources) {
logPool(` source: ${src.address}`);
}
}
/**
* Load from environment variable IPV6_SOURCES.
*/
loadFromEnv(): void {
const env = process.env.IPV6_SOURCES;
if (env) {
this.loadFromString(env);
}
}
// -- Access ------------------------------------------------------------------
/** Total number of source addresses. */
get size(): number {
return this.sources.length;
}
/** Whether any source addresses are configured. */
get configured(): boolean {
return this.sources.length > 0;
}
/**
* Get the next IPv6 source address (round-robin).
* Skips sources that have exceeded the failure threshold.
*
* @returns The next IPv6 address, or null if pool is empty or all failed.
*/
getNext(): string | null {
if (this.sources.length === 0) return null;
const startIndex = this.currentIndex;
let checked = 0;
do {
const entry = this.sources[this.currentIndex];
if (entry && !entry.disabled) {
const addr = entry.address;
// Advance index for next call
this.currentIndex = (this.currentIndex + 1) % this.sources.length;
logPool(`getNext -> ${addr} (index=${this.currentIndex})`);
return addr;
}
this.currentIndex = (this.currentIndex + 1) % this.sources.length;
checked++;
} while (this.currentIndex !== startIndex && checked <= this.sources.length);
// All sources failed — reset and return first
logPool("all IPv6 sources failed, resetting");
for (const src of this.sources) {
src.failures = 0;
src.disabled = false;
}
this.currentIndex = 0;
return this.sources[0]?.address ?? null;
}
/**
* Get a specific source address by index.
* Returns null if out of bounds.
*/
getAtIndex(index: number): string | null {
return this.sources[index]?.address ?? null;
}
// -- Failure tracking --------------------------------------------------------
/**
* Mark a source address as failed.
* If failures exceed threshold, the source is disabled until reset.
*
* @param address - The IPv6 address that failed
*/
markFailed(address: string): void {
const entry = this.sources.find((s) => s.address === address);
if (!entry) return;
entry.failures++;
logPool(`markFailed ${address} (${entry.failures}/${this.failureThreshold})`);
if (entry.failures >= this.failureThreshold) {
entry.disabled = true;
console.warn(
`[ipv6-pool] Source ${address} failed ${entry.failures}/${this.failureThreshold} times — disabled`,
);
}
}
/**
* Mark a source address as successful (resets failure count).
*
* @param address - The IPv6 address that succeeded
*/
markSuccess(address: string): void {
const entry = this.sources.find((s) => s.address === address);
if (!entry) return;
if (entry.failures > 0) {
logPool(`markSuccess ${address} (was ${entry.failures} failures)`);
}
entry.failures = 0;
entry.disabled = false;
}
/** Set the failure threshold (default 3). */
setFailureThreshold(n: number): void {
this.failureThreshold = n;
}
/** Reset all failure states. */
reset(): void {
for (const src of this.sources) {
src.failures = 0;
src.disabled = false;
}
this.currentIndex = 0;
logPool("reset all sources");
}
}