fix: secure assistant generation telemetry
This commit is contained in:
1 parent
ca8163c061
commit
4060f9299f
11 files changed
+181
-10
No files matched your search
@@ -0,0 +1,34 @@
|
||||
import { createHmac, hkdfSync, timingSafeEqual } from "node:crypto";
|
||||
|
||||
import { env } from "@homarr/common/env";
|
||||
|
||||
const getGenerationTelemetrySecret = () =>
|
||||
Buffer.from(
|
||||
hkdfSync("sha256", Buffer.from(env.SECRET_ENCRYPTION_KEY, "hex"), "", "assistant-generation-telemetry", 32),
|
||||
);
|
||||
|
||||
const getGenerationTelemetryPayload = ({
|
||||
userId,
|
||||
threadId,
|
||||
generationId,
|
||||
}: {
|
||||
userId: string;
|
||||
threadId: string;
|
||||
generationId: string;
|
||||
}) => `${userId}\0${threadId}\0${generationId}`;
|
||||
|
||||
export const createAssistantGenerationAccessToken = (input: {
|
||||
userId: string;
|
||||
threadId: string;
|
||||
generationId: string;
|
||||
}) =>
|
||||
createHmac("sha256", getGenerationTelemetrySecret()).update(getGenerationTelemetryPayload(input)).digest("base64url");
|
||||
|
||||
export const verifyAssistantGenerationAccessToken = (
|
||||
input: { userId: string; threadId: string; generationId: string },
|
||||
accessToken: string,
|
||||
) => {
|
||||
const expected = Buffer.from(createAssistantGenerationAccessToken(input));
|
||||
const actual = Buffer.from(accessToken);
|
||||
return expected.length === actual.length && timingSafeEqual(expected, actual);
|
||||
};
|
||||
@@ -1,5 +1,9 @@
|
||||
import { fetchWithTrustedCertificatesAsync } from "@homarr/core/infrastructure/http";
|
||||
|
||||
const generationTelemetryCache = new Map<string, { expiresAt: number; value: OpenRouterGenerationTelemetry }>();
|
||||
const generationTelemetryCacheTtlMs = 5 * 60_000;
|
||||
const generationTelemetryCacheMaxEntries = 512;
|
||||
|
||||
export type OpenRouterGenerationTelemetry = {
|
||||
generationId?: string;
|
||||
routedProvider?: string;
|
||||
@@ -50,13 +54,17 @@ export const parseOpenRouterGenerationTelemetry = (value: unknown): OpenRouterGe
|
||||
const generationTimeMs = asFiniteNonNegativeNumber(data.generation_time);
|
||||
const providerResponses = Array.isArray(data.provider_responses) ? data.provider_responses : undefined;
|
||||
const parsedProviderResponses = providerResponses?.map(asRecord).filter((response) => response !== null);
|
||||
const providerResponseStatuses = parsedProviderResponses?.map((response) =>
|
||||
asFiniteNonNegativeNumber(response.status),
|
||||
);
|
||||
const failedProviderResponses = parsedProviderResponses?.filter((response) => {
|
||||
const status = asFiniteNonNegativeNumber(response.status);
|
||||
return status !== undefined && (status < 200 || status >= 300);
|
||||
});
|
||||
const fallbackCount = providerResponses
|
||||
? failedProviderResponses && failedProviderResponses.length > 0
|
||||
? failedProviderResponses.length
|
||||
? providerResponseStatuses?.length === providerResponses.length &&
|
||||
providerResponseStatuses.every((status) => status !== undefined)
|
||||
? (failedProviderResponses?.length ?? 0)
|
||||
: Math.max(providerResponses.length - 1, 0)
|
||||
: undefined;
|
||||
const fallbackLatencies = failedProviderResponses?.map((response) => asFiniteNonNegativeNumber(response.latency));
|
||||
@@ -128,12 +136,31 @@ export const fetchOpenRouterGenerationTelemetryAsync = async ({
|
||||
generationId: string;
|
||||
headers: Record<string, string>;
|
||||
}): Promise<OpenRouterGenerationTelemetry | null> => {
|
||||
const cacheKey = `${baseUrl}\0${generationId}`;
|
||||
const cached = generationTelemetryCache.get(cacheKey);
|
||||
if (cached && cached.expiresAt > Date.now()) {
|
||||
generationTelemetryCache.delete(cacheKey);
|
||||
generationTelemetryCache.set(cacheKey, cached);
|
||||
return cached.value;
|
||||
}
|
||||
if (cached) generationTelemetryCache.delete(cacheKey);
|
||||
|
||||
try {
|
||||
const endpoint = new URL(`${baseUrl.replace(/\/$/, "")}/generation`);
|
||||
endpoint.searchParams.set("id", generationId);
|
||||
const response = await fetchWithTrustedCertificatesAsync(endpoint, { headers, timeout: 3000 });
|
||||
if (!response.ok) return null;
|
||||
return parseOpenRouterGenerationTelemetry(await response.json());
|
||||
const telemetry = parseOpenRouterGenerationTelemetry(await response.json());
|
||||
if (!telemetry) return null;
|
||||
if (generationTelemetryCache.size >= generationTelemetryCacheMaxEntries) {
|
||||
const oldestKey = generationTelemetryCache.keys().next().value;
|
||||
if (oldestKey !== undefined) generationTelemetryCache.delete(oldestKey);
|
||||
}
|
||||
generationTelemetryCache.set(cacheKey, {
|
||||
expiresAt: Date.now() + generationTelemetryCacheTtlMs,
|
||||
value: telemetry,
|
||||
});
|
||||
return telemetry;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -5,3 +5,4 @@ export {
|
||||
type AssistantContextEntity,
|
||||
} from "./router/assistant";
|
||||
export type { OpenRouterGenerationTelemetry } from "./assistant-generation-telemetry";
|
||||
export { createAssistantGenerationAccessToken } from "./assistant-generation-access";
|
||||
@@ -27,6 +27,7 @@ import {
|
||||
|
||||
import type { createTRPCContext } from "../trpc";
|
||||
import { fetchOpenRouterGenerationTelemetryAsync } from "../assistant-generation-telemetry";
|
||||
import { verifyAssistantGenerationAccessToken } from "../assistant-generation-access";
|
||||
import { createTRPCRouter, permissionRequiredProcedure, protectedProcedure } from "../trpc";
|
||||
import { boardRouter } from "./board";
|
||||
|
||||
@@ -174,7 +175,13 @@ const getGenerationIdsFromMessageContent = (content: unknown) => {
|
||||
.flatMap((step) => {
|
||||
if (!step || typeof step !== "object" || Array.isArray(step)) return [];
|
||||
const generationId = (step as { generationId?: unknown }).generationId;
|
||||
return typeof generationId === "string" && /^gen-[A-Za-z0-9_-]{1,128}$/.test(generationId) ? [generationId] : [];
|
||||
const accessToken = (step as { generationAccessToken?: unknown }).generationAccessToken;
|
||||
return typeof generationId === "string" &&
|
||||
/^gen-[A-Za-z0-9_-]{1,128}$/.test(generationId) &&
|
||||
typeof accessToken === "string" &&
|
||||
/^[A-Za-z0-9_-]{43}$/.test(accessToken)
|
||||
? [{ generationId, accessToken }]
|
||||
: [];
|
||||
})
|
||||
.slice(0, 8);
|
||||
};
|
||||
@@ -714,10 +721,19 @@ export const assistantRouter = createTRPCRouter({
|
||||
return { complete: true, generations: [] };
|
||||
}
|
||||
|
||||
const generationIds = getGenerationIdsFromMessageContent(parse(message.content));
|
||||
if (generationIds.length === 0) return { complete: true, generations: [] };
|
||||
const generationReferences = getGenerationIdsFromMessageContent(parse(message.content)).filter((reference) =>
|
||||
verifyAssistantGenerationAccessToken(
|
||||
{
|
||||
userId: ctx.session.user.id,
|
||||
threadId: thread.id,
|
||||
generationId: reference.generationId,
|
||||
},
|
||||
reference.accessToken,
|
||||
),
|
||||
);
|
||||
if (generationReferences.length === 0) return { complete: true, generations: [] };
|
||||
const generations = await Promise.all(
|
||||
generationIds.map((generationId) =>
|
||||
generationReferences.map(({ generationId }) =>
|
||||
fetchOpenRouterGenerationTelemetryAsync({
|
||||
baseUrl: configuration.baseUrl,
|
||||
generationId,
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
|
||||
vi.mock("@homarr/common/env", () => ({
|
||||
env: { SECRET_ENCRYPTION_KEY: "a".repeat(64) },
|
||||
}));
|
||||
|
||||
import {
|
||||
createAssistantGenerationAccessToken,
|
||||
verifyAssistantGenerationAccessToken,
|
||||
} from "../assistant-generation-access";
|
||||
|
||||
describe("assistant generation access token", () => {
|
||||
const input = {
|
||||
userId: "user-1",
|
||||
threadId: "thread-1",
|
||||
generationId: "gen-example",
|
||||
};
|
||||
|
||||
test("binds a generation to its user and conversation", () => {
|
||||
const accessToken = createAssistantGenerationAccessToken(input);
|
||||
|
||||
expect(accessToken).toMatch(/^[A-Za-z0-9_-]{43}$/);
|
||||
expect(verifyAssistantGenerationAccessToken(input, accessToken)).toBe(true);
|
||||
expect(verifyAssistantGenerationAccessToken({ ...input, userId: "user-2" }, accessToken)).toBe(false);
|
||||
expect(verifyAssistantGenerationAccessToken({ ...input, threadId: "thread-2" }, accessToken)).toBe(false);
|
||||
expect(verifyAssistantGenerationAccessToken({ ...input, generationId: "gen-other" }, accessToken)).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -1,8 +1,13 @@
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
|
||||
import { fetchWithTrustedCertificatesAsync } from "@homarr/core/infrastructure/http";
|
||||
|
||||
vi.mock("@homarr/core/infrastructure/http", () => ({ fetchWithTrustedCertificatesAsync: vi.fn() }));
|
||||
|
||||
import { parseOpenRouterGenerationTelemetry } from "../assistant-generation-telemetry";
|
||||
import {
|
||||
fetchOpenRouterGenerationTelemetryAsync,
|
||||
parseOpenRouterGenerationTelemetry,
|
||||
} from "../assistant-generation-telemetry";
|
||||
|
||||
describe("parseOpenRouterGenerationTelemetry", () => {
|
||||
test("uses provider-native tokens for context and throughput", () => {
|
||||
@@ -63,4 +68,38 @@ describe("parseOpenRouterGenerationTelemetry", () => {
|
||||
expect(parseOpenRouterGenerationTelemetry({ data: [] })).toBeNull();
|
||||
expect(parseOpenRouterGenerationTelemetry({ data: { latency: -1, total_cost: "invalid" } })).toBeNull();
|
||||
});
|
||||
|
||||
test("does not count additional successful provider responses as fallbacks", () => {
|
||||
expect(
|
||||
parseOpenRouterGenerationTelemetry({
|
||||
id: "gen-success",
|
||||
provider_responses: [
|
||||
{ status: 200, latency: 100 },
|
||||
{ status: 200, latency: 120 },
|
||||
],
|
||||
}),
|
||||
).toMatchObject({ fallbackCount: 0 });
|
||||
});
|
||||
|
||||
test("caches completed generation records", async () => {
|
||||
const mockedFetch = vi.mocked(fetchWithTrustedCertificatesAsync);
|
||||
mockedFetch.mockResolvedValue({
|
||||
ok: true,
|
||||
json: () => Promise.resolve({ data: { id: "gen-cached", native_tokens_completion: 4 } }),
|
||||
} as Awaited<ReturnType<typeof fetchWithTrustedCertificatesAsync>>);
|
||||
const input = {
|
||||
baseUrl: "https://openrouter.ai/api/v1",
|
||||
generationId: "gen-cached",
|
||||
headers: { Authorization: "Bearer test" },
|
||||
};
|
||||
|
||||
await expect(fetchOpenRouterGenerationTelemetryAsync(input)).resolves.toMatchObject({
|
||||
generationId: "gen-cached",
|
||||
outputTokens: 4,
|
||||
});
|
||||
await expect(fetchOpenRouterGenerationTelemetryAsync(input)).resolves.toMatchObject({
|
||||
generationId: "gen-cached",
|
||||
});
|
||||
expect(mockedFetch).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user