feat: proxy subscription clients through gateway
All checks were successful
changelog-draft / changelog-draft (push) Successful in 31s
All checks were successful
changelog-draft / changelog-draft (push) Successful in 31s
This commit is contained in:
parent
f720096d3b
commit
b8543ec191
@ -993,6 +993,7 @@
|
||||
.req-status.approved { color: var(--ok); border-color: rgba(21,128,61,0.24); }
|
||||
.req-status.error, .req-status.rejected { color: var(--err); border-color: rgba(180,35,24,0.24); }
|
||||
.req-status.warning, .req-status.pending_review { color: var(--warn); border-color: rgba(180,83,9,0.24); }
|
||||
.req-status.bypassed { color: var(--err); border-color: rgba(180,35,24,0.32); }
|
||||
|
||||
.client-grid {
|
||||
display: grid;
|
||||
@ -1921,7 +1922,7 @@
|
||||
<div title="${req.request_id}">${req.request_id.substring(0, 14)}…</div>
|
||||
<div>${escapeHtml(req.caller)}</div>
|
||||
<div title="${req.model}">${req.model}</div>
|
||||
<div><span class="req-status ${req.status}">${req.status}</span></div>
|
||||
<div><span class="req-status ${formatRequestStatus(req)}">${formatRequestStatus(req)}</span></div>
|
||||
<div>${formatNumber(req.compression_tokens_before ?? req.tokens_in ?? 0)}</div>
|
||||
<div>${formatNumber(req.compression_tokens_after ?? req.tokens_in ?? 0)}</div>
|
||||
<div>${formatSavedTokens(req.compression_tokens_saved ?? 0)}</div>
|
||||
@ -1942,6 +1943,7 @@
|
||||
}
|
||||
|
||||
function formatCompression(req) {
|
||||
if (String(req.compression_mode || '').startsWith('usage-import:')) return 'bypassed';
|
||||
const mode = String(req.compression_mode || 'none:none').split(':').pop() || 'none';
|
||||
const pct = Number(req.compression_savings_pct || 0);
|
||||
if (!req.compression_mode) return 'not tracked';
|
||||
@ -1949,6 +1951,10 @@
|
||||
return `${escapeHtml(mode)} · ${pct.toFixed(1)}%`;
|
||||
}
|
||||
|
||||
function formatRequestStatus(req) {
|
||||
return String(req.request_id || '').startsWith('usage-import:') ? 'bypassed' : escapeHtml(req.status);
|
||||
}
|
||||
|
||||
function escapeHtml(s) {
|
||||
return String(s ?? '').replace(/[&<>"']/g, c => ({ '&':'&', '<':'<', '>':'>', '"':'"', "'":''' }[c]));
|
||||
}
|
||||
|
||||
@ -1047,6 +1047,7 @@
|
||||
.req-status.approved { color: var(--ok); border-color: rgba(21,128,61,0.24); }
|
||||
.req-status.error, .req-status.rejected { color: var(--err); border-color: rgba(180,35,24,0.24); }
|
||||
.req-status.warning, .req-status.pending_review { color: var(--warn); border-color: rgba(180,83,9,0.24); }
|
||||
.req-status.bypassed { color: var(--err); border-color: rgba(180,35,24,0.32); }
|
||||
|
||||
.client-grid {
|
||||
display: grid;
|
||||
@ -2228,7 +2229,7 @@
|
||||
<div title="${req.request_id}">${req.request_id.substring(0, 14)}…</div>
|
||||
<div>${escapeHtml(req.caller)}</div>
|
||||
<div title="${req.model}">${req.model}</div>
|
||||
<div><span class="req-status ${req.status}">${req.status}</span></div>
|
||||
<div><span class="req-status ${formatRequestStatus(req)}">${formatRequestStatus(req)}</span></div>
|
||||
<div>${formatNumber(req.compression_tokens_before ?? req.tokens_in ?? 0)}</div>
|
||||
<div>${formatNumber(req.compression_tokens_after ?? req.tokens_in ?? 0)}</div>
|
||||
<div>${formatSavedTokens(req.compression_tokens_saved ?? 0)}</div>
|
||||
@ -2249,6 +2250,7 @@
|
||||
}
|
||||
|
||||
function formatCompression(req) {
|
||||
if (String(req.compression_mode || '').startsWith('usage-import:')) return 'bypassed';
|
||||
const mode = String(req.compression_mode || 'none:none').split(':').pop() || 'none';
|
||||
const pct = Number(req.compression_savings_pct || 0);
|
||||
if (!req.compression_mode) return 'not tracked';
|
||||
@ -2256,6 +2258,10 @@
|
||||
return `${escapeHtml(mode)} · ${pct.toFixed(1)}%`;
|
||||
}
|
||||
|
||||
function formatRequestStatus(req) {
|
||||
return String(req.request_id || '').startsWith('usage-import:') ? 'bypassed' : escapeHtml(req.status);
|
||||
}
|
||||
|
||||
function escapeHtml(s) {
|
||||
return String(s ?? '').replace(/[&<>"']/g, c => ({ '&':'&', '<':'<', '>':'>', '"':'"', "'":''' }[c]));
|
||||
}
|
||||
|
||||
@ -0,0 +1,37 @@
|
||||
import { afterEach, describe, expect, it } from 'vitest';
|
||||
import { transformSubscriptionPayload } from '../transparent-subscription-proxy.js';
|
||||
|
||||
afterEach(() => {
|
||||
delete process.env['REDACT_PII_MODE'];
|
||||
});
|
||||
|
||||
describe('transparent subscription payload protection', () => {
|
||||
it('redacts identifiers and compresses long text without changing protocol fields', () => {
|
||||
process.env['REDACT_PII_MODE'] = 'always';
|
||||
const email = ['person', 'example.org'].join('@');
|
||||
const payload = {
|
||||
model: 'subscription-test-model',
|
||||
input: [{
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [{ type: 'input_text', text: `${email}\n${'context line\n'.repeat(6000)}` }],
|
||||
}],
|
||||
tools: [{ type: 'function', name: 'read_file' }],
|
||||
};
|
||||
|
||||
const result = transformSubscriptionPayload(payload, {
|
||||
providerName: 'codex-subscription',
|
||||
caller: 'unit-client',
|
||||
});
|
||||
const transformed = result.payload as typeof payload;
|
||||
const text = transformed.input[0]?.content[0]?.text ?? '';
|
||||
|
||||
expect(result.compressionApplied).toBe(true);
|
||||
expect(result.tokensAfter).toBeLessThan(result.tokensBefore);
|
||||
expect(result.piiRedactions).toBe(1);
|
||||
expect(text).not.toContain(email);
|
||||
expect(text).toContain('<EMAIL_001>');
|
||||
expect(text).toContain('ctxlean omitted');
|
||||
expect(transformed.tools).toEqual(payload.tools);
|
||||
});
|
||||
});
|
||||
@ -143,8 +143,7 @@ export class RequestLogger {
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT mode, tokens_before, tokens_after, savings_pct
|
||||
FROM tokenvault_metrics
|
||||
WHERE tool_used = 'gateway'
|
||||
AND file_path = rt.request_id
|
||||
WHERE file_path = rt.request_id
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 1
|
||||
) tv ON true
|
||||
|
||||
260
packages/gateway/src/modules/transparent-subscription-proxy.ts
Normal file
260
packages/gateway/src/modules/transparent-subscription-proxy.ts
Normal file
@ -0,0 +1,260 @@
|
||||
import type { FastifyReply, FastifyRequest } from 'fastify';
|
||||
import { Readable } from 'stream';
|
||||
import { compressContext } from './context-compressor.js';
|
||||
import { redactPii, getRedactMode, shouldRedactFor } from './pii-redaction.js';
|
||||
import { createRequestLogger } from './request-logger.js';
|
||||
import { getPool } from '../db/client.js';
|
||||
import { logger } from '../observability/logger.js';
|
||||
import { logCompressionMetric } from '../utils/tokenvault-hooks.js';
|
||||
|
||||
export type TransparentSubscriptionTarget = 'anthropic' | 'codex';
|
||||
|
||||
interface TransformStats {
|
||||
payload: unknown;
|
||||
tokensBefore: number;
|
||||
tokensAfter: number;
|
||||
compressionApplied: boolean;
|
||||
piiRedactions: number;
|
||||
}
|
||||
|
||||
interface ProxyOptions {
|
||||
target: TransparentSubscriptionTarget;
|
||||
upstreamPath: string;
|
||||
trackRequest?: boolean;
|
||||
}
|
||||
|
||||
const TEXT_FIELDS = new Set([
|
||||
'content',
|
||||
'input',
|
||||
'input_text',
|
||||
'output',
|
||||
'output_text',
|
||||
'result',
|
||||
'system',
|
||||
'text',
|
||||
]);
|
||||
|
||||
const INTERNAL_HEADERS = new Set([
|
||||
'connection',
|
||||
'content-length',
|
||||
'host',
|
||||
'transfer-encoding',
|
||||
'x-forwarded-for',
|
||||
'x-forwarded-host',
|
||||
'x-forwarded-proto',
|
||||
'x-llm-gateway-key',
|
||||
'x-llm-gateway-upstream',
|
||||
'x-llm-interceptor-caller',
|
||||
'x-llm-request-id',
|
||||
]);
|
||||
|
||||
const RESPONSE_HEADERS = new Set([
|
||||
'cache-control',
|
||||
'content-type',
|
||||
'openai-processing-ms',
|
||||
'request-id',
|
||||
'x-request-id',
|
||||
'x-ratelimit-limit-requests',
|
||||
'x-ratelimit-limit-tokens',
|
||||
'x-ratelimit-remaining-requests',
|
||||
'x-ratelimit-remaining-tokens',
|
||||
'x-ratelimit-reset-requests',
|
||||
'x-ratelimit-reset-tokens',
|
||||
]);
|
||||
|
||||
function estimatePayloadTokens(payload: unknown): number {
|
||||
return Math.max(1, Math.ceil(JSON.stringify(payload ?? {}).length / 4));
|
||||
}
|
||||
|
||||
function countRedactions(counts: Record<string, number>): number {
|
||||
return Object.values(counts).reduce((total, count) => total + count, 0);
|
||||
}
|
||||
|
||||
export function transformSubscriptionPayload(
|
||||
payload: unknown,
|
||||
options: { providerName: string; caller: string },
|
||||
): TransformStats {
|
||||
const tokensBefore = estimatePayloadTokens(payload);
|
||||
let compressionApplied = false;
|
||||
let piiRedactions = 0;
|
||||
const redact = shouldRedactFor(getRedactMode(), options.providerName, options.caller);
|
||||
|
||||
const visit = (value: unknown, parentField?: string): unknown => {
|
||||
if (typeof value === 'string') {
|
||||
if (!parentField || !TEXT_FIELDS.has(parentField)) return value;
|
||||
const redacted = redact ? redactPii(value) : null;
|
||||
if (redacted) piiRedactions += countRedactions(redacted.counts);
|
||||
const protectedText = redacted?.redacted ?? value;
|
||||
const compressed = compressContext(protectedText, { enabled: true, mode: 'structured-safe' });
|
||||
compressionApplied ||= compressed.applied;
|
||||
return compressed.input;
|
||||
}
|
||||
|
||||
if (Array.isArray(value)) return value.map((item) => visit(item, parentField));
|
||||
if (!value || typeof value !== 'object') return value;
|
||||
|
||||
return Object.fromEntries(
|
||||
Object.entries(value as Record<string, unknown>).map(([key, child]) => [key, visit(child, key)])
|
||||
);
|
||||
};
|
||||
|
||||
const transformed = visit(payload);
|
||||
return {
|
||||
payload: transformed,
|
||||
tokensBefore,
|
||||
tokensAfter: estimatePayloadTokens(transformed),
|
||||
compressionApplied,
|
||||
piiRedactions,
|
||||
};
|
||||
}
|
||||
|
||||
function requestHeader(request: FastifyRequest, name: string): string | undefined {
|
||||
const value = request.headers[name];
|
||||
return Array.isArray(value) ? value[0] : value;
|
||||
}
|
||||
|
||||
export function transparentTargetFromRequest(request: FastifyRequest): TransparentSubscriptionTarget | null {
|
||||
const target = requestHeader(request, 'x-llm-gateway-upstream')?.toLowerCase();
|
||||
return target === 'anthropic' || target === 'codex' ? target : null;
|
||||
}
|
||||
|
||||
function upstreamBase(target: TransparentSubscriptionTarget): string {
|
||||
if (target === 'anthropic') {
|
||||
return (process.env['ANTHROPIC_UPSTREAM_URL'] ?? 'https://api.anthropic.com/v1').replace(/\/+$/, '');
|
||||
}
|
||||
return (process.env['CODEX_UPSTREAM_URL'] ?? 'https://chatgpt.com/backend-api/codex').replace(/\/+$/, '');
|
||||
}
|
||||
|
||||
function forwardedHeaders(request: FastifyRequest): Record<string, string> {
|
||||
const headers: Record<string, string> = {};
|
||||
for (const [name, value] of Object.entries(request.headers)) {
|
||||
if (INTERNAL_HEADERS.has(name) || name.startsWith('cf-')) continue;
|
||||
const first = Array.isArray(value) ? value[0] : value;
|
||||
if (first !== undefined) headers[name] = first;
|
||||
}
|
||||
headers['content-type'] = 'application/json';
|
||||
return headers;
|
||||
}
|
||||
|
||||
function safeRequestId(request: FastifyRequest, target: TransparentSubscriptionTarget): string {
|
||||
const provided = requestHeader(request, 'x-llm-request-id');
|
||||
if (provided && /^[A-Za-z0-9:._-]{8,220}$/.test(provided)) return provided;
|
||||
return `${target}-proxy:${Date.now()}:${Math.random().toString(36).slice(2, 9)}`;
|
||||
}
|
||||
|
||||
async function trackProxyRequest(
|
||||
callId: string,
|
||||
caller: string,
|
||||
model: string,
|
||||
target: TransparentSubscriptionTarget,
|
||||
status: 'approved' | 'error',
|
||||
latencyMs: number,
|
||||
transform: TransformStats,
|
||||
): Promise<void> {
|
||||
const db = getPool();
|
||||
const saved = Math.max(0, transform.tokensBefore - transform.tokensAfter);
|
||||
await logCompressionMetric(db, {
|
||||
filePath: callId,
|
||||
mode: transform.compressionApplied ? 'structured:head-tail-excerpt' : 'structured:none',
|
||||
tokensBefore: transform.tokensBefore,
|
||||
tokensAfter: transform.tokensAfter,
|
||||
savingsPct: transform.tokensBefore > 0 ? Math.round((saved / transform.tokensBefore) * 10_000) / 100 : 0,
|
||||
toolUsed: 'gateway',
|
||||
});
|
||||
await createRequestLogger(db).logRequest(
|
||||
callId,
|
||||
caller,
|
||||
`${target}_transparent_bridge`,
|
||||
model,
|
||||
status,
|
||||
transform.tokensAfter,
|
||||
0,
|
||||
0,
|
||||
latencyMs,
|
||||
0,
|
||||
transform.compressionApplied,
|
||||
);
|
||||
}
|
||||
|
||||
export async function proxyTransparentSubscription(
|
||||
request: FastifyRequest,
|
||||
reply: FastifyReply,
|
||||
options: ProxyOptions,
|
||||
): Promise<FastifyReply> {
|
||||
const startedAt = Date.now();
|
||||
const caller = requestHeader(request, 'x-llm-interceptor-caller')
|
||||
?? requestHeader(request, 'x-caller-id')
|
||||
?? `${options.target}-transparent-client`;
|
||||
const callId = safeRequestId(request, options.target);
|
||||
const rawPayload = request.body ?? {};
|
||||
const transformed = transformSubscriptionPayload(rawPayload, {
|
||||
providerName: `${options.target}-subscription`,
|
||||
caller,
|
||||
});
|
||||
const model = typeof (rawPayload as Record<string, unknown>)?.['model'] === 'string'
|
||||
? String((rawPayload as Record<string, unknown>)['model'])
|
||||
: options.target;
|
||||
const query = request.url.includes('?') ? request.url.slice(request.url.indexOf('?')) : '';
|
||||
const targetUrl = `${upstreamBase(options.target)}${options.upstreamPath}${query}`;
|
||||
|
||||
try {
|
||||
const upstream = await fetch(targetUrl, {
|
||||
method: request.method,
|
||||
headers: forwardedHeaders(request),
|
||||
body: request.method === 'GET' || request.method === 'HEAD'
|
||||
? undefined
|
||||
: JSON.stringify(transformed.payload),
|
||||
});
|
||||
|
||||
reply.status(upstream.status).header('X-LLM-Gateway-Request-ID', callId);
|
||||
for (const [name, value] of upstream.headers.entries()) {
|
||||
if (RESPONSE_HEADERS.has(name.toLowerCase())) reply.header(name, value);
|
||||
}
|
||||
|
||||
if (options.trackRequest) {
|
||||
void trackProxyRequest(
|
||||
callId,
|
||||
caller,
|
||||
model,
|
||||
options.target,
|
||||
upstream.ok ? 'approved' : 'error',
|
||||
Date.now() - startedAt,
|
||||
transformed,
|
||||
).catch((error) => logger.warn({ error, callId }, 'Transparent bridge tracking failed'));
|
||||
}
|
||||
|
||||
logger.info({
|
||||
callId,
|
||||
caller,
|
||||
target: options.target,
|
||||
model,
|
||||
statusCode: upstream.status,
|
||||
tokensBefore: transformed.tokensBefore,
|
||||
tokensAfter: transformed.tokensAfter,
|
||||
piiRedactions: transformed.piiRedactions,
|
||||
}, 'Transparent subscription request proxied');
|
||||
|
||||
if (!upstream.body) return reply.send();
|
||||
return reply.send(Readable.fromWeb(upstream.body as any));
|
||||
} catch (error) {
|
||||
logger.error({ error, callId, target: options.target }, 'Transparent subscription proxy failed');
|
||||
if (options.trackRequest) {
|
||||
void trackProxyRequest(
|
||||
callId,
|
||||
caller,
|
||||
model,
|
||||
options.target,
|
||||
'error',
|
||||
Date.now() - startedAt,
|
||||
transformed,
|
||||
).catch(() => undefined);
|
||||
}
|
||||
return reply.status(502).send({
|
||||
error: {
|
||||
message: 'Subscription upstream unavailable',
|
||||
type: 'upstream_error',
|
||||
code: 502,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,83 @@
|
||||
import Fastify from 'fastify';
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { completionRoute } from '../completion.js';
|
||||
|
||||
const gatewayKey = ['unit', 'gateway', 'route', 'key'].join('-');
|
||||
const providerSession = ['provider', 'session', 'value'].join('-');
|
||||
|
||||
beforeEach(() => {
|
||||
process.env['LLM_GATEWAY_API_AUTH_MODE'] = 'enforce';
|
||||
process.env['LLM_GATEWAY_API_KEYS'] = gatewayKey;
|
||||
process.env['REDACT_PII_MODE'] = 'always';
|
||||
process.env['ANTHROPIC_UPSTREAM_URL'] = 'https://example.invalid/v1';
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.unstubAllGlobals();
|
||||
delete process.env['LLM_GATEWAY_API_AUTH_MODE'];
|
||||
delete process.env['LLM_GATEWAY_API_KEYS'];
|
||||
delete process.env['REDACT_PII_MODE'];
|
||||
delete process.env['ANTHROPIC_UPSTREAM_URL'];
|
||||
});
|
||||
|
||||
describe('transparent subscription routes', () => {
|
||||
it('requires the dedicated gateway key', async () => {
|
||||
const fetchMock = vi.fn();
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
const app = Fastify();
|
||||
await app.register(completionRoute, { prefix: '/v1' });
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/messages',
|
||||
headers: { 'x-llm-gateway-upstream': 'anthropic' },
|
||||
payload: { model: 'subscription-test-model', max_tokens: 16, messages: [{ role: 'user', content: 'hello' }] },
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(401);
|
||||
expect(fetchMock).not.toHaveBeenCalled();
|
||||
await app.close();
|
||||
});
|
||||
|
||||
it('forwards provider auth, strips gateway auth and protects prompt text', async () => {
|
||||
const email = ['route', 'example.org'].join('@');
|
||||
const fetchMock = vi.fn(async () => new Response(JSON.stringify({
|
||||
id: 'msg_test',
|
||||
type: 'message',
|
||||
role: 'assistant',
|
||||
model: 'subscription-test-model',
|
||||
content: [{ type: 'text', text: 'ok' }],
|
||||
usage: { input_tokens: 10, output_tokens: 1 },
|
||||
}), { status: 200, headers: { 'content-type': 'application/json' } }));
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
const app = Fastify();
|
||||
await app.register(completionRoute, { prefix: '/v1' });
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/messages',
|
||||
headers: {
|
||||
authorization: `Bearer ${providerSession}`,
|
||||
'x-llm-gateway-key': gatewayKey,
|
||||
'x-llm-gateway-upstream': 'anthropic',
|
||||
'anthropic-version': '2023-06-01',
|
||||
},
|
||||
payload: {
|
||||
model: 'subscription-test-model',
|
||||
max_tokens: 16,
|
||||
messages: [{ role: 'user', content: `Contact ${email}` }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(fetchMock).toHaveBeenCalledOnce();
|
||||
const [url, init] = fetchMock.mock.calls[0] as unknown as [string, RequestInit];
|
||||
const headers = init.headers as Record<string, string>;
|
||||
expect(url).toBe('https://example.invalid/v1/messages');
|
||||
expect(headers.authorization).toBe(`Bearer ${providerSession}`);
|
||||
expect(headers['x-llm-gateway-key']).toBeUndefined();
|
||||
expect(String(init.body)).not.toContain(email);
|
||||
expect(String(init.body)).toContain('<EMAIL_001>');
|
||||
await app.close();
|
||||
});
|
||||
});
|
||||
@ -68,6 +68,11 @@ import { runPreComplete, runPostComplete } from '../modules/plugin-system.js';
|
||||
import { getAdaptiveRecommendation } from '../modules/adaptive-routing.js';
|
||||
import { guardOutputStream, getOutputDefenseMode } from '../modules/output-defense.js';
|
||||
import { callPromptGuard, isPromptGuardConfigured, getPromptGuardThreshold, getPromptGuardMinLen } from '../modules/prompt-guard-client.js';
|
||||
import {
|
||||
proxyTransparentSubscription,
|
||||
transparentTargetFromRequest,
|
||||
} from '../modules/transparent-subscription-proxy.js';
|
||||
import { authorizeGatewayRequest } from '../security/tenant-auth.js';
|
||||
|
||||
// // Disable Ollama-dependent scanners (sentinel, constitutional, embedding, attention)
|
||||
// // to keep gateway scans fast and dependency-free
|
||||
@ -121,6 +126,11 @@ const CompletionRequestSchema = z.object({
|
||||
|
||||
type CompletionRequest = z.infer<typeof CompletionRequestSchema>;
|
||||
|
||||
const AGENT_BODY_LIMIT_BYTES = Math.min(
|
||||
64 * 1024 * 1024,
|
||||
Math.max(1024 * 1024, Number(process.env['GATEWAY_AGENT_BODY_LIMIT_BYTES'] ?? 16 * 1024 * 1024)),
|
||||
);
|
||||
|
||||
function shouldBypassResponseCache(caller: string): boolean {
|
||||
const normalized = caller.toLowerCase();
|
||||
return normalized.includes('claude-code')
|
||||
@ -1523,7 +1533,7 @@ export async function completionRoute(fastify: FastifyInstance): Promise<void> {
|
||||
return reply.send(listGatewayModels());
|
||||
});
|
||||
|
||||
fastify.post('/chat/completions', { config: { rateLimit: false } }, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
fastify.post('/chat/completions', { bodyLimit: AGENT_BODY_LIMIT_BYTES, config: { rateLimit: false } }, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const startMs = Date.now();
|
||||
const parsed = OpenAIChatCompletionRequestSchema.safeParse(request.body);
|
||||
if (!parsed.success) {
|
||||
@ -1538,6 +1548,9 @@ export async function completionRoute(fastify: FastifyInstance): Promise<void> {
|
||||
|
||||
const callId = `chatcmpl-${Date.now()}-${Math.random().toString(36).slice(2, 9)}`;
|
||||
const requestedSubscriptionId = requestedSubscriptionFromHeaders(request);
|
||||
if (requestedSubscriptionId || subscriptionBridgeTargetForModel(parsed.data.model)) {
|
||||
if (!await authorizeGatewayRequest(request, reply)) return reply;
|
||||
}
|
||||
const { caller: bridgeCaller } = detectCaller(request, 'openai-compatible', parsed.data.user);
|
||||
const subscriptionBridgeResult = await callSubscriptionBridgeChatCompletion(parsed.data, startMs, callId, requestedSubscriptionId, bridgeCaller);
|
||||
if (requestedSubscriptionId && !subscriptionBridgeResult) {
|
||||
@ -1581,8 +1594,21 @@ export async function completionRoute(fastify: FastifyInstance): Promise<void> {
|
||||
});
|
||||
|
||||
// Anthropic Messages API compatibility — accept @anthropic-ai/sdk traffic.
|
||||
fastify.post('/messages', { config: { rateLimit: false } }, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
fastify.post('/messages', { bodyLimit: AGENT_BODY_LIMIT_BYTES, config: { rateLimit: false } }, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const startMs = Date.now();
|
||||
const transparentTarget = transparentTargetFromRequest(request);
|
||||
if (transparentTarget) {
|
||||
if (transparentTarget !== 'anthropic') {
|
||||
return reply.status(400).send({ type: 'error', error: { type: 'invalid_request_error', message: 'invalid transparent upstream for Messages API' } });
|
||||
}
|
||||
if (!await authorizeGatewayRequest(request, reply)) return reply;
|
||||
return proxyTransparentSubscription(request, reply, {
|
||||
target: 'anthropic',
|
||||
upstreamPath: '/messages',
|
||||
trackRequest: false,
|
||||
});
|
||||
}
|
||||
|
||||
const parsed = AnthropicMessagesRequestSchema.safeParse(request.body);
|
||||
if (!parsed.success) {
|
||||
return reply.status(400).send({
|
||||
@ -1594,6 +1620,10 @@ export async function completionRoute(fastify: FastifyInstance): Promise<void> {
|
||||
});
|
||||
}
|
||||
|
||||
if (modelToSubscriptionId(parsed.data.model) === 'claude-code') {
|
||||
if (!await authorizeGatewayRequest(request, reply)) return reply;
|
||||
}
|
||||
|
||||
const callId = `msg_${Date.now()}_${Math.random().toString(36).slice(2, 9)}`;
|
||||
const gatewayRequest = anthropicRequestToGatewayRequest(parsed.data, request);
|
||||
const result = await executeCompletion(gatewayRequest, startMs, callId);
|
||||
@ -1622,8 +1652,33 @@ export async function completionRoute(fastify: FastifyInstance): Promise<void> {
|
||||
return reply.status(200).send(response);
|
||||
});
|
||||
|
||||
fastify.post('/responses', { config: { rateLimit: false } }, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
fastify.post('/messages/count_tokens', { bodyLimit: AGENT_BODY_LIMIT_BYTES, config: { rateLimit: false } }, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
if (transparentTargetFromRequest(request) !== 'anthropic') {
|
||||
return reply.status(400).send({ type: 'error', error: { type: 'invalid_request_error', message: 'transparent Anthropic bridge required' } });
|
||||
}
|
||||
if (!await authorizeGatewayRequest(request, reply)) return reply;
|
||||
return proxyTransparentSubscription(request, reply, {
|
||||
target: 'anthropic',
|
||||
upstreamPath: '/messages/count_tokens',
|
||||
trackRequest: false,
|
||||
});
|
||||
});
|
||||
|
||||
fastify.post('/responses', { bodyLimit: AGENT_BODY_LIMIT_BYTES, config: { rateLimit: false } }, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const startMs = Date.now();
|
||||
const transparentTarget = transparentTargetFromRequest(request);
|
||||
if (transparentTarget) {
|
||||
if (transparentTarget !== 'codex') {
|
||||
return reply.status(400).send({ error: { message: 'invalid transparent upstream for Responses API', type: 'invalid_request_error', code: 400 } });
|
||||
}
|
||||
if (!await authorizeGatewayRequest(request, reply)) return reply;
|
||||
return proxyTransparentSubscription(request, reply, {
|
||||
target: 'codex',
|
||||
upstreamPath: '/responses',
|
||||
trackRequest: true,
|
||||
});
|
||||
}
|
||||
|
||||
const parsed = OpenAIResponsesRequestSchema.safeParse(request.body);
|
||||
if (!parsed.success) {
|
||||
return reply.status(400).send({
|
||||
@ -1643,6 +1698,7 @@ export async function completionRoute(fastify: FastifyInstance): Promise<void> {
|
||||
// the right auth. Route them straight to the local codex-bridge
|
||||
// (local bridge process) which speaks codex-cli over OAuth.
|
||||
if (isCodexBridgeModel(parsed.data.model)) {
|
||||
if (!await authorizeGatewayRequest(request, reply)) return reply;
|
||||
try {
|
||||
const bridgeUrl = process.env['CODEX_BRIDGE_URL'] ?? 'http://localhost:3253';
|
||||
const bridgeCaller = (request.headers['x-llm-interceptor-caller'] as string) || 'codex-app';
|
||||
@ -1790,6 +1846,18 @@ data: [DONE]
|
||||
return reply.send(response);
|
||||
});
|
||||
|
||||
fastify.post('/responses/compact', { bodyLimit: AGENT_BODY_LIMIT_BYTES, config: { rateLimit: false } }, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
if (transparentTargetFromRequest(request) !== 'codex') {
|
||||
return reply.status(400).send({ error: { message: 'transparent Codex bridge required', type: 'invalid_request_error', code: 400 } });
|
||||
}
|
||||
if (!await authorizeGatewayRequest(request, reply)) return reply;
|
||||
return proxyTransparentSubscription(request, reply, {
|
||||
target: 'codex',
|
||||
upstreamPath: '/responses/compact',
|
||||
trackRequest: false,
|
||||
});
|
||||
});
|
||||
|
||||
// ─── Multi-Model Race Mode endpoint ────────────────────────────────────
|
||||
// Runs the same prompt against multiple models in parallel; returns
|
||||
// according to `strategy` (first | best | consensus). Audits each
|
||||
|
||||
@ -1753,22 +1753,30 @@ export async function dashboardRoute(fastify: FastifyInstance): Promise<void> {
|
||||
// Upsert by request_id (one row per caller/model/day): re-reporting an
|
||||
// in-progress day updates its totals instead of creating duplicates.
|
||||
const updated = await db.query(
|
||||
`UPDATE request_tracking SET tokens_in=$1, tokens_out=$2, cost_usd=$3, created_at=$4 WHERE request_id=$5`,
|
||||
`UPDATE request_tracking SET tokens_in=$1, tokens_out=$2, cost_usd=$3, created_at=$4, status='warning', task_type='usage_import' WHERE request_id=$5`,
|
||||
[tokensIn, tokensOut, costUsd, usedAt, requestId]
|
||||
);
|
||||
if (updated.rowCount === 0) {
|
||||
const requestLogger = createRequestLogger(db);
|
||||
await requestLogger.logRequest(requestId, caller, 'usage_import', model, 'approved', tokensIn, tokensOut, costUsd, 0);
|
||||
await requestLogger.logRequest(requestId, caller, 'usage_import', model, 'warning', tokensIn, tokensOut, costUsd, 0);
|
||||
await db.query(`UPDATE request_tracking SET created_at=$1 WHERE request_id=$2`, [usedAt, requestId]);
|
||||
}
|
||||
const updatedMetric = await db.query(
|
||||
`UPDATE tokenvault_metrics
|
||||
SET mode='usage-import:none', tokens_before=$2, tokens_after=$2, savings_pct=0, created_at=NOW(), tool_used='usage-import'
|
||||
WHERE id=(SELECT id FROM tokenvault_metrics WHERE file_path=$1 ORDER BY created_at DESC LIMIT 1)`,
|
||||
[requestId, tokensIn]
|
||||
);
|
||||
if (updatedMetric.rowCount === 0) {
|
||||
await logCompressionMetric(db, {
|
||||
filePath: requestId,
|
||||
mode: 'usage-import:none',
|
||||
tokensBefore: tokensIn,
|
||||
tokensAfter: tokensIn,
|
||||
savingsPct: 0,
|
||||
toolUsed: 'gateway',
|
||||
toolUsed: 'usage-import',
|
||||
});
|
||||
}
|
||||
return reply.status(200).send({ success: true, imported: { caller, model, day, tokensIn, tokensOut, costUsd, usedAt } });
|
||||
} catch (error) {
|
||||
logger.error({ error }, 'Failed to import usage report');
|
||||
|
||||
@ -14,6 +14,7 @@ import {
|
||||
import { getPublicSettings, saveSettings } from '../modules/settings-store.js';
|
||||
import { getAllProviders, getAvailableProviders } from '../pipeline/external-providers.js';
|
||||
import { logger } from '../observability/logger.js';
|
||||
import { requireGatewayApiKey } from '../security/tenant-auth.js';
|
||||
|
||||
interface SubscriptionBridgeBody {
|
||||
subscription_id?: SubscriptionId;
|
||||
@ -126,7 +127,7 @@ function buildAccessCard(
|
||||
` -H "X-Caller-ID: pilot-client"`,
|
||||
` -d '{"model":"${defaultModel}","messages":[{"role":"user","content":"gateway bridge smoke"}]}'`,
|
||||
].join(' \\\n'),
|
||||
note: 'Use a gateway API key/admin token from your deployment. OAuth/subscription secrets stay inside the bridge process and are never exposed here.',
|
||||
note: 'Use a dedicated gateway API key from your deployment. OAuth/subscription secrets stay inside the bridge process and are never exposed here.',
|
||||
};
|
||||
}
|
||||
|
||||
@ -329,6 +330,7 @@ async function postBridgeChat(
|
||||
|
||||
export async function subscriptionsRoute(fastify: FastifyInstance): Promise<void> {
|
||||
const auth = { preHandler: requireDashboardAuth };
|
||||
const apiAuth = { preHandler: requireGatewayApiKey };
|
||||
|
||||
fastify.get('/api/subscriptions', auth, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const query = request.query as { search?: string; q?: string };
|
||||
@ -521,7 +523,7 @@ export async function subscriptionsRoute(fastify: FastifyInstance): Promise<void
|
||||
}
|
||||
});
|
||||
|
||||
fastify.post('/api/subscriptions/:subscription_id/v1/chat/completions', auth, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
fastify.post('/api/subscriptions/:subscription_id/v1/chat/completions', apiAuth, async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const params = request.params as SubscriptionChatParams;
|
||||
const status = await getSubscriptionStatusOrReply(params.subscription_id, reply);
|
||||
if (!status) return reply;
|
||||
@ -542,6 +544,8 @@ export async function subscriptionsRoute(fastify: FastifyInstance): Promise<void
|
||||
url: '/v1/chat/completions',
|
||||
headers: {
|
||||
'content-type': 'application/json',
|
||||
...(request.headers.authorization ? { authorization: request.headers.authorization } : {}),
|
||||
...(request.headers['x-llm-gateway-key'] ? { 'x-llm-gateway-key': request.headers['x-llm-gateway-key'] } : {}),
|
||||
'x-caller-id': Array.isArray(caller) ? caller[0] : String(caller),
|
||||
'x-llm-gateway-subscription': status.descriptor.id,
|
||||
},
|
||||
|
||||
76
packages/gateway/src/routes/tracking.ts
Normal file
76
packages/gateway/src/routes/tracking.ts
Normal file
@ -0,0 +1,76 @@
|
||||
import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
|
||||
import { z } from 'zod';
|
||||
import { getPool } from '../db/client.js';
|
||||
import { createRequestLogger } from '../modules/request-logger.js';
|
||||
import { logger } from '../observability/logger.js';
|
||||
import { requireGatewayApiKey } from '../security/tenant-auth.js';
|
||||
import { logCompressionMetric } from '../utils/tokenvault-hooks.js';
|
||||
|
||||
const BridgeTrackingReportSchema = z.object({
|
||||
request_id: z.string().min(8).max(220).optional(),
|
||||
caller: z.string().min(1).max(120),
|
||||
model: z.string().min(1).max(120),
|
||||
source: z.string().min(1).max(80).default('subscription-bridge'),
|
||||
tokens_in: z.number().int().nonnegative().max(100_000_000),
|
||||
tokens_out: z.number().int().nonnegative().max(100_000_000),
|
||||
latency_ms: z.number().int().nonnegative().max(3_600_000).default(0),
|
||||
compression_tokens_before: z.number().int().nonnegative().max(100_000_000).optional(),
|
||||
compression_tokens_after: z.number().int().nonnegative().max(100_000_000).optional(),
|
||||
compression_mode: z.string().min(1).max(48).optional(),
|
||||
});
|
||||
|
||||
function safeRequestId(value: string | undefined): string {
|
||||
if (value && /^[A-Za-z0-9:._-]{8,220}$/.test(value)) return value;
|
||||
return `bridge-report:${Date.now()}:${Math.random().toString(36).slice(2, 9)}`;
|
||||
}
|
||||
|
||||
export async function trackingRoute(fastify: FastifyInstance): Promise<void> {
|
||||
fastify.post(
|
||||
'/api/tracking/report',
|
||||
{ preHandler: requireGatewayApiKey },
|
||||
async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const parsed = BridgeTrackingReportSchema.safeParse(request.body);
|
||||
if (!parsed.success) {
|
||||
return reply.status(400).send({ success: false, error: parsed.error.errors[0]?.message ?? 'invalid report' });
|
||||
}
|
||||
|
||||
const report = parsed.data;
|
||||
const requestId = safeRequestId(report.request_id);
|
||||
const before = report.compression_tokens_before ?? report.tokens_in;
|
||||
const after = report.compression_tokens_after ?? report.tokens_in;
|
||||
const saved = Math.max(0, before - after);
|
||||
const db = getPool();
|
||||
|
||||
try {
|
||||
const existing = await db.query('SELECT 1 FROM request_tracking WHERE request_id=$1 LIMIT 1', [requestId]);
|
||||
if (existing.rowCount === 0) {
|
||||
await createRequestLogger(db).logRequest(
|
||||
requestId,
|
||||
report.caller,
|
||||
`subscription_bridge:${report.source}`.slice(0, 50),
|
||||
report.model,
|
||||
'approved',
|
||||
report.tokens_in,
|
||||
report.tokens_out,
|
||||
0,
|
||||
report.latency_ms,
|
||||
0,
|
||||
saved > 0,
|
||||
);
|
||||
await logCompressionMetric(db, {
|
||||
filePath: requestId,
|
||||
mode: report.compression_mode ?? (saved > 0 ? 'bridge:compressed' : 'bridge:none'),
|
||||
tokensBefore: before,
|
||||
tokensAfter: after,
|
||||
savingsPct: before > 0 ? Math.round((saved / before) * 10_000) / 100 : 0,
|
||||
toolUsed: 'gateway',
|
||||
});
|
||||
}
|
||||
return reply.status(200).send({ success: true, requestId, duplicate: existing.rowCount !== 0 });
|
||||
} catch (error) {
|
||||
logger.error({ error, requestId }, 'Bridge tracking report failed');
|
||||
return reply.status(500).send({ success: false, error: 'tracking report failed' });
|
||||
}
|
||||
},
|
||||
);
|
||||
}
|
||||
42
packages/gateway/src/security/__tests__/tenant-auth.test.ts
Normal file
42
packages/gateway/src/security/__tests__/tenant-auth.test.ts
Normal file
@ -0,0 +1,42 @@
|
||||
import { afterEach, describe, expect, it } from 'vitest';
|
||||
import type { FastifyRequest } from 'fastify';
|
||||
import {
|
||||
gatewayApiKeyMatches,
|
||||
getGatewayApiAuthMode,
|
||||
getPresentedGatewayApiKey,
|
||||
} from '../tenant-auth.js';
|
||||
|
||||
function requestWithHeaders(headers: Record<string, string>): FastifyRequest {
|
||||
return { headers } as unknown as FastifyRequest;
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
delete process.env['LLM_GATEWAY_API_AUTH_MODE'];
|
||||
});
|
||||
|
||||
describe('gateway API authentication', () => {
|
||||
it('defaults to off and accepts explicit rollout modes', () => {
|
||||
expect(getGatewayApiAuthMode()).toBe('off');
|
||||
process.env['LLM_GATEWAY_API_AUTH_MODE'] = 'monitor';
|
||||
expect(getGatewayApiAuthMode()).toBe('monitor');
|
||||
process.env['LLM_GATEWAY_API_AUTH_MODE'] = 'enforce';
|
||||
expect(getGatewayApiAuthMode()).toBe('enforce');
|
||||
});
|
||||
|
||||
it('reads the dedicated bridge header before bearer authorization', () => {
|
||||
const dedicated = ['unit', 'gateway', 'dedicated'].join('-');
|
||||
const bearer = ['unit', 'gateway', 'bearer'].join('-');
|
||||
const request = requestWithHeaders({
|
||||
'x-llm-gateway-key': dedicated,
|
||||
authorization: `Bearer ${bearer}`,
|
||||
});
|
||||
expect(getPresentedGatewayApiKey(request)).toBe(dedicated);
|
||||
});
|
||||
|
||||
it('matches configured keys without accepting missing or partial values', () => {
|
||||
const configured = ['unit', 'gateway', 'key', 'alpha'].join('-');
|
||||
expect(gatewayApiKeyMatches(configured, [configured])).toBe(true);
|
||||
expect(gatewayApiKeyMatches(`${configured}-partial`, [configured])).toBe(false);
|
||||
expect(gatewayApiKeyMatches(undefined, [configured])).toBe(false);
|
||||
});
|
||||
});
|
||||
@ -1,9 +1,93 @@
|
||||
import fp from 'fastify-plugin';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import { timingSafeEqual } from 'crypto';
|
||||
import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
|
||||
import { logger } from '../observability/logger.js';
|
||||
|
||||
export type GatewayApiAuthMode = 'off' | 'monitor' | 'enforce';
|
||||
|
||||
export function getGatewayApiAuthMode(): GatewayApiAuthMode {
|
||||
const mode = (process.env['LLM_GATEWAY_API_AUTH_MODE'] ?? 'off').trim().toLowerCase();
|
||||
if (mode === 'monitor' || mode === 'enforce') return mode;
|
||||
return 'off';
|
||||
}
|
||||
|
||||
export function getConfiguredGatewayApiKeys(): string[] {
|
||||
return (process.env['LLM_GATEWAY_API_KEYS'] ?? '')
|
||||
.split(',')
|
||||
.map((key) => key.trim())
|
||||
.filter(Boolean);
|
||||
}
|
||||
|
||||
function firstHeader(value: string | string[] | undefined): string | undefined {
|
||||
return Array.isArray(value) ? value[0] : value;
|
||||
}
|
||||
|
||||
export function getPresentedGatewayApiKey(request: FastifyRequest): string | undefined {
|
||||
const dedicated = firstHeader(request.headers['x-llm-gateway-key']);
|
||||
if (dedicated?.trim()) return dedicated.trim();
|
||||
|
||||
const authorization = firstHeader(request.headers.authorization);
|
||||
const match = authorization?.match(/^Bearer\s+(.+)$/i);
|
||||
return match?.[1]?.trim();
|
||||
}
|
||||
|
||||
export function gatewayApiKeyMatches(presented: string | undefined, configuredKeys: readonly string[]): boolean {
|
||||
if (!presented) return false;
|
||||
const candidate = Buffer.from(presented);
|
||||
return configuredKeys.some((configured) => {
|
||||
const expected = Buffer.from(configured);
|
||||
return candidate.length === expected.length && timingSafeEqual(candidate, expected);
|
||||
});
|
||||
}
|
||||
|
||||
export async function authorizeGatewayRequest(
|
||||
request: FastifyRequest,
|
||||
reply: FastifyReply,
|
||||
): Promise<boolean> {
|
||||
const mode = getGatewayApiAuthMode();
|
||||
if (mode === 'off') return true;
|
||||
|
||||
const configuredKeys = getConfiguredGatewayApiKeys();
|
||||
const valid = gatewayApiKeyMatches(getPresentedGatewayApiKey(request), configuredKeys);
|
||||
if (valid) return true;
|
||||
|
||||
if (mode === 'monitor') {
|
||||
logger.warn({ method: request.method, path: request.url.split('?')[0] }, 'Gateway API key missing or invalid');
|
||||
return true;
|
||||
}
|
||||
|
||||
if (configuredKeys.length === 0) {
|
||||
reply.status(503).send({
|
||||
error: {
|
||||
message: 'Gateway API authentication is not configured',
|
||||
type: 'gateway_auth_configuration_error',
|
||||
code: 503,
|
||||
},
|
||||
});
|
||||
return false;
|
||||
}
|
||||
|
||||
reply.header('WWW-Authenticate', 'Bearer realm="llm-gateway"');
|
||||
reply.status(401).send({
|
||||
error: {
|
||||
message: 'Valid gateway API key required',
|
||||
type: 'authentication_error',
|
||||
code: 401,
|
||||
},
|
||||
});
|
||||
return false;
|
||||
}
|
||||
|
||||
export async function requireGatewayApiKey(
|
||||
request: FastifyRequest,
|
||||
reply: FastifyReply,
|
||||
): Promise<FastifyReply | void> {
|
||||
const authorized = await authorizeGatewayRequest(request, reply);
|
||||
if (!authorized) return reply;
|
||||
}
|
||||
|
||||
async function tenantAuth(_fastify: FastifyInstance): Promise<void> {
|
||||
// Tenant auth is intentionally permissive until tenant policies are configured.
|
||||
// Admin and dashboard routes keep their own token checks.
|
||||
logger.info({ mode: getGatewayApiAuthMode(), configuredKeys: getConfiguredGatewayApiKeys().length }, 'Gateway API authentication initialised');
|
||||
}
|
||||
|
||||
export default fp(tenantAuth, { name: 'tenant-auth' });
|
||||
|
||||
@ -3,6 +3,7 @@ import fastifyCors from '@fastify/cors';
|
||||
import fastifyRateLimit from '@fastify/rate-limit';
|
||||
import fastifyHelmet from '@fastify/helmet';
|
||||
import { completionRoute } from './routes/completion.js';
|
||||
import { trackingRoute } from './routes/tracking.js';
|
||||
import { batchRoute } from './routes/batch.js';
|
||||
import { classifyRoute } from './routes/classify.js';
|
||||
import { guardRoute } from './routes/guard.js';
|
||||
@ -111,7 +112,18 @@ async function buildServer() {
|
||||
/^https:\/\/.*\.runwerk\.app$/,
|
||||
],
|
||||
methods: ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS'],
|
||||
allowedHeaders: ['Content-Type', 'Authorization', 'X-Caller-ID', 'X-Runwerk-Caller', 'X-Runwerk-Privacy', 'X-Runwerk-Tier', 'X-Runwerk-Purpose'],
|
||||
allowedHeaders: [
|
||||
'Content-Type',
|
||||
'Authorization',
|
||||
'X-Caller-ID',
|
||||
'X-LLM-Gateway-Key',
|
||||
'X-LLM-Gateway-Upstream',
|
||||
'X-LLM-Interceptor-Caller',
|
||||
'X-Runwerk-Caller',
|
||||
'X-Runwerk-Privacy',
|
||||
'X-Runwerk-Tier',
|
||||
'X-Runwerk-Purpose',
|
||||
],
|
||||
credentials: true,
|
||||
});
|
||||
|
||||
@ -132,6 +144,7 @@ async function buildServer() {
|
||||
|
||||
await server.register(tenantAuth);
|
||||
await server.register(internalRoute);
|
||||
await server.register(trackingRoute);
|
||||
await server.register(completionRoute, { prefix: '/v1' });
|
||||
await server.register(embeddingsRoute, { prefix: '/v1' });
|
||||
await server.register(replayRoute, { prefix: '/v1' });
|
||||
|
||||
288
scripts/client/anthropic-interceptor.py
Normal file
288
scripts/client/anthropic-interceptor.py
Normal file
@ -0,0 +1,288 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fail-closed local subscription companion for Claude Code and Codex."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import time
|
||||
import uuid
|
||||
|
||||
import httpx
|
||||
from fastapi import FastAPI, Request, Response
|
||||
from fastapi.responses import JSONResponse, StreamingResponse
|
||||
|
||||
|
||||
GATEWAY_URL = os.getenv("LLM_GATEWAY_URL", "https://llm-gateway.context-x.org").rstrip("/")
|
||||
KEYCHAIN_SERVICE = os.getenv("LLM_GATEWAY_KEYCHAIN_SERVICE", "llm-gateway-api-key")
|
||||
CLAUDE_CALLER = os.getenv("INTERCEPTOR_CALLER", "claude-code-macbook")
|
||||
CODEX_CALLER = os.getenv("CODEX_INTERCEPTOR_CALLER", "codex-macbook")
|
||||
MAX_TOOL_CHARS = int(os.getenv("MAX_TOOL_CHARS", "8000"))
|
||||
|
||||
_PII = [
|
||||
(re.compile(r"\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b"), "[EMAIL]"),
|
||||
(re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"), "[IP_ADDRESS]"),
|
||||
(re.compile(r"\b(?:IBAN[:\s]*)?[A-Z]{2}\d{2}[A-Z0-9 ]{10,30}\b"), "[IBAN]"),
|
||||
(re.compile(r"\beyJ[A-Za-z0-9_\-]+\.eyJ[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+\b"), "[JWT]"),
|
||||
(re.compile(r"\b(?:sk-proj-|sk-ant-)[A-Za-z0-9_\-]{20,}\b"), "[API_KEY]"),
|
||||
(re.compile(r"\bAKIA[0-9A-Z]{16}\b"), "[AWS_KEY]"),
|
||||
(re.compile(r"\b(?:ghp_|github_pat_)[A-Za-z0-9_]{20,}\b"), "[GH_TOKEN]"),
|
||||
]
|
||||
|
||||
|
||||
def gateway_key() -> str:
|
||||
configured = os.getenv("LLM_GATEWAY_API_KEY", "").strip()
|
||||
if configured:
|
||||
return configured
|
||||
try:
|
||||
return subprocess.check_output(
|
||||
["/usr/bin/security", "find-generic-password", "-s", KEYCHAIN_SERVICE, "-w"],
|
||||
stderr=subprocess.DEVNULL,
|
||||
timeout=5,
|
||||
).decode().strip()
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def anonymize(text: str) -> str:
|
||||
for pattern, replacement in _PII:
|
||||
text = pattern.sub(replacement, text)
|
||||
return text
|
||||
|
||||
|
||||
def anonymize_content(content):
|
||||
if isinstance(content, str):
|
||||
return anonymize(content)
|
||||
if isinstance(content, list):
|
||||
result = []
|
||||
for part in content:
|
||||
if isinstance(part, dict) and part.get("type") == "text":
|
||||
part = {**part, "text": anonymize(str(part.get("text", "")))}
|
||||
result.append(part)
|
||||
return result
|
||||
return content
|
||||
|
||||
|
||||
def compress_messages(messages: list) -> list:
|
||||
result = []
|
||||
for original in messages:
|
||||
message = dict(original)
|
||||
content = message.get("content", "")
|
||||
if message.get("role") == "tool" and isinstance(content, str) and len(content) > MAX_TOOL_CHARS:
|
||||
message["content"] = content[:MAX_TOOL_CHARS] + "\n[truncated by subscription companion]"
|
||||
elif message.get("role") == "user" and isinstance(content, list):
|
||||
parts = []
|
||||
for original_part in content:
|
||||
part = dict(original_part) if isinstance(original_part, dict) else original_part
|
||||
if isinstance(part, dict) and part.get("type") == "tool_result":
|
||||
inner = part.get("content", "")
|
||||
if isinstance(inner, str) and len(inner) > MAX_TOOL_CHARS:
|
||||
part["content"] = inner[:MAX_TOOL_CHARS] + "\n[truncated by subscription companion]"
|
||||
elif isinstance(inner, list):
|
||||
trimmed = []
|
||||
for original_inner_part in inner:
|
||||
inner_part = dict(original_inner_part) if isinstance(original_inner_part, dict) else original_inner_part
|
||||
if (
|
||||
isinstance(inner_part, dict)
|
||||
and inner_part.get("type") == "text"
|
||||
and len(str(inner_part.get("text", ""))) > MAX_TOOL_CHARS
|
||||
):
|
||||
inner_part["text"] = str(inner_part["text"])[:MAX_TOOL_CHARS] + "\n[truncated by subscription companion]"
|
||||
trimmed.append(inner_part)
|
||||
part["content"] = trimmed
|
||||
parts.append(part)
|
||||
message["content"] = parts
|
||||
result.append(message)
|
||||
return result
|
||||
|
||||
|
||||
def estimate_tokens(payload) -> int:
|
||||
return max(1, math.ceil(len(json.dumps(payload, ensure_ascii=False)) / 4))
|
||||
|
||||
|
||||
def prepare_anthropic_body(raw_body: bytes) -> tuple[bytes, int, int, bool]:
|
||||
data = json.loads(raw_body or b"{}")
|
||||
before = estimate_tokens(data)
|
||||
is_stream = bool(data.get("stream"))
|
||||
if isinstance(data.get("system"), str):
|
||||
data["system"] = anonymize(data["system"])
|
||||
elif isinstance(data.get("system"), list):
|
||||
data["system"] = anonymize_content(data["system"])
|
||||
|
||||
messages = []
|
||||
for original in data.get("messages", []):
|
||||
message = dict(original)
|
||||
message["content"] = anonymize_content(message.get("content", ""))
|
||||
messages.append(message)
|
||||
data["messages"] = compress_messages(messages)
|
||||
after = estimate_tokens(data)
|
||||
return json.dumps(data, ensure_ascii=False).encode(), before, after, is_stream
|
||||
|
||||
|
||||
def response_headers(headers: httpx.Headers) -> dict[str, str]:
|
||||
blocked = {"connection", "content-encoding", "content-length", "keep-alive", "transfer-encoding"}
|
||||
return {name: value for name, value in headers.items() if name.lower() not in blocked}
|
||||
|
||||
|
||||
async def report_usage(
|
||||
key: str,
|
||||
request_id: str,
|
||||
caller: str,
|
||||
model: str,
|
||||
tokens_in: int,
|
||||
tokens_out: int,
|
||||
latency_ms: int,
|
||||
before: int,
|
||||
after: int,
|
||||
) -> None:
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=5.0) as client:
|
||||
await client.post(
|
||||
f"{GATEWAY_URL}/api/tracking/report",
|
||||
headers={"X-LLM-Gateway-Key": key},
|
||||
json={
|
||||
"request_id": request_id,
|
||||
"caller": caller,
|
||||
"model": model,
|
||||
"tokens_in": tokens_in,
|
||||
"tokens_out": tokens_out,
|
||||
"latency_ms": latency_ms,
|
||||
"source": "local-subscription-companion",
|
||||
"compression_tokens_before": before,
|
||||
"compression_tokens_after": after,
|
||||
"compression_mode": "interceptor:tool-trim" if before > after else "interceptor:none",
|
||||
},
|
||||
)
|
||||
except Exception:
|
||||
return
|
||||
|
||||
|
||||
app = FastAPI(title="llm-subscription-companion", docs_url=None, redoc_url=None)
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
configured = bool(gateway_key())
|
||||
return JSONResponse(
|
||||
status_code=200 if configured else 503,
|
||||
content={"status": "ok" if configured else "auth_required", "gateway": GATEWAY_URL, "failClosed": True},
|
||||
)
|
||||
|
||||
|
||||
@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "HEAD", "PATCH"])
|
||||
async def proxy(request: Request, path: str):
|
||||
key = gateway_key()
|
||||
if not key:
|
||||
return JSONResponse(status_code=503, content={"error": "gateway authentication unavailable"})
|
||||
|
||||
started_at = time.monotonic()
|
||||
raw_body = await request.body()
|
||||
is_codex = path.startswith("openai/")
|
||||
gateway_path = path[len("openai/"):] if is_codex else path
|
||||
target_kind = "codex" if is_codex else "anthropic"
|
||||
caller = CODEX_CALLER if is_codex else CLAUDE_CALLER
|
||||
is_messages = not is_codex and gateway_path == "v1/messages" and request.method == "POST"
|
||||
is_stream = False
|
||||
before = after = estimate_tokens({})
|
||||
|
||||
if is_messages:
|
||||
try:
|
||||
raw_body, before, after, is_stream = prepare_anthropic_body(raw_body)
|
||||
except Exception:
|
||||
return JSONResponse(status_code=400, content={"error": "invalid Anthropic request body"})
|
||||
|
||||
request_id = f"{target_kind}-companion:{int(time.time() * 1000)}:{uuid.uuid4().hex[:8]}"
|
||||
blocked = {"connection", "content-length", "host", "transfer-encoding", "x-llm-gateway-key"}
|
||||
headers = {name: value for name, value in request.headers.items() if name.lower() not in blocked}
|
||||
headers.update({
|
||||
"X-LLM-Gateway-Key": key,
|
||||
"X-LLM-Gateway-Upstream": target_kind,
|
||||
"X-LLM-Interceptor-Caller": caller,
|
||||
"X-LLM-Request-ID": request_id,
|
||||
})
|
||||
target = f"{GATEWAY_URL}/{gateway_path}"
|
||||
params = dict(request.query_params)
|
||||
|
||||
if is_stream:
|
||||
client = httpx.AsyncClient(timeout=300.0)
|
||||
upstream_request = client.build_request(request.method, target, headers=headers, content=raw_body, params=params)
|
||||
upstream = await client.send(upstream_request, stream=True)
|
||||
|
||||
async def stream_response():
|
||||
pending = ""
|
||||
model_name, tokens_in, tokens_out = "unknown", 0, 0
|
||||
try:
|
||||
async for chunk in upstream.aiter_bytes():
|
||||
pending += chunk.decode(errors="ignore")
|
||||
lines = pending.split("\n")
|
||||
pending = lines.pop()
|
||||
for line in lines:
|
||||
if not line.startswith("data: "):
|
||||
continue
|
||||
try:
|
||||
event = json.loads(line[6:])
|
||||
except Exception:
|
||||
continue
|
||||
if event.get("type") == "message_start":
|
||||
message = event.get("message", {})
|
||||
model_name = message.get("model", model_name)
|
||||
tokens_in = message.get("usage", {}).get("input_tokens", tokens_in)
|
||||
elif event.get("type") == "message_delta":
|
||||
tokens_out = event.get("usage", {}).get("output_tokens", tokens_out)
|
||||
yield chunk
|
||||
finally:
|
||||
await upstream.aclose()
|
||||
await client.aclose()
|
||||
if tokens_in or tokens_out:
|
||||
await report_usage(
|
||||
key, request_id, caller, model_name, tokens_in, tokens_out,
|
||||
int((time.monotonic() - started_at) * 1000), before, after,
|
||||
)
|
||||
|
||||
return StreamingResponse(
|
||||
stream_response(),
|
||||
status_code=upstream.status_code,
|
||||
headers=response_headers(upstream.headers),
|
||||
media_type=upstream.headers.get("content-type", "text/event-stream"),
|
||||
)
|
||||
|
||||
async with httpx.AsyncClient(timeout=300.0) as client:
|
||||
upstream = await client.request(
|
||||
request.method,
|
||||
target,
|
||||
headers=headers,
|
||||
content=raw_body if request.method not in {"GET", "HEAD"} else None,
|
||||
params=params,
|
||||
)
|
||||
|
||||
if is_messages and upstream.is_success:
|
||||
try:
|
||||
response_data = upstream.json()
|
||||
usage = response_data.get("usage", {})
|
||||
asyncio.create_task(report_usage(
|
||||
key,
|
||||
request_id,
|
||||
caller,
|
||||
response_data.get("model", "unknown"),
|
||||
int(usage.get("input_tokens", 0)),
|
||||
int(usage.get("output_tokens", 0)),
|
||||
int((time.monotonic() - started_at) * 1000),
|
||||
before,
|
||||
after,
|
||||
))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return Response(
|
||||
content=upstream.content,
|
||||
status_code=upstream.status_code,
|
||||
headers=response_headers(upstream.headers),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
|
||||
uvicorn.run(app, host="127.0.0.1", port=int(os.getenv("PORT", "3090")), log_level="warning")
|
||||
@ -3,6 +3,7 @@
|
||||
# Production fix for the recurring PM2 env-drop quirk.
|
||||
set -a
|
||||
[ -f /opt/llm-gateway/.env.defense ] && source /opt/llm-gateway/.env.defense
|
||||
[ -f /opt/llm-gateway/.env.api ] && source /opt/llm-gateway/.env.api
|
||||
[ -f /opt/llm-gateway/.env ] && source /opt/llm-gateway/.env
|
||||
set +a
|
||||
exec /usr/bin/node /opt/llm-gateway/packages/gateway/scripts/launch.mjs
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user