calvin.erfmann/pangolin
1
0
Fork 0
mirror of https://github.com/fosrl/pangolin.git synced 2026-08-26 05:55:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3641 from fosrl/dev

Browse Source 
1.22.0
This commit is contained in:
Owen Schwartz
2026-08-25 17:19:46 -04:00
committed by GitHub
parent 7319bf84f7 72d2c79793
commit f3474dac98
101 changed files with 2209 additions and 886 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/routers/aiGateway/createAiGatewayRouter.ts
+22 -4
View File
@@ -1,19 +1,37 @@
import { Router } from "express";
import { Router, type Request, type Response } from "express";
import {
AI_CAPABILITY_DEFS,
type AiCapability
} from "@server/lib/aiCapabilities";
import { handleAiGatewayProxy } from "@server/routers/aiGateway/pipeline";
import { handleV1Models } from "@server/routers/aiGateway";
type CapabilityHandler = (
req: Request,
res: Response,
capability: AiCapability
) => Promise<any>;
// Capabilities the gateway answers itself instead of proxying upstream.
// Everything else goes through the inference pipeline.
const LOCAL_HANDLERS: Partial<Record<AiCapability, CapabilityHandler>> = {
v1_models: handleV1Models
};
export function createAiGatewayRouter() {
const router = Router();
for (const def of Object.values(AI_CAPABILITY_DEFS)) {
const capability = def.id as AiCapability;
const handler = LOCAL_HANDLERS[capability] ?? handleAiGatewayProxy;
for (const route of def.routes) {
router.post(route.path, (req, res) =>
handleAiGatewayProxy(req, res, capability)
);
const bind = (req: Request, res: Response) =>
handler(req, res, capability);
if (route.method === "GET") {
router.get(route.path, bind);
} else {
router.post(route.path, bind);
}
}
}
server/routers/aiGateway/index.ts
+1
View File
@@ -1,2 +1,3 @@
export { handleAiGatewayProxy } from "./pipeline";
export { handleV1Models } from "./v1Models";
export { createAiGatewayRouter } from "./createAiGatewayRouter";
server/routers/aiGateway/logAiSession.ts
+5 -255
View File
@@ -1,171 +1,12 @@
import { logsDb, db, orgs, aiSessionLog, type AiProvider } from "@server/db";
import type { InferInsertModel } from "drizzle-orm";
import logger from "@server/logger";
import { and, eq, lt } from "drizzle-orm";
import cache from "#dynamic/lib/cache";
import { calculateCutoffTimestamp } from "@server/lib/cleanupLogs";
import { sanitizeString } from "@server/lib/sanitize";
import type { AiCapability } from "@server/lib/aiCapabilities";
import {
normalizeAiRequest,
normalizeAiResponse
} from "@server/lib/aiMessageNormalization";
// Caps how much of the request/response body we keep per row, so a single
// huge multimodal payload can't blow up buffer memory or storage.
const AI_SESSION_LOG_MAX_BODY_CHARS = 200_000;
type AiSessionLogInsert = InferInsertModel<typeof aiSessionLog>;
// In-memory buffer for batching AI session log inserts, mirroring the
// approach in server/routers/badger/logRequestAudit.ts.
const sessionLogBuffer: AiSessionLogInsert[] = [];
const BATCH_SIZE = 100; // Write to DB every 100 logs
const BATCH_INTERVAL_MS = 5000; // Or every 5 seconds, whichever comes first
const MAX_BUFFER_SIZE = 10000; // Prevent unbounded memory growth
let flushTimer: NodeJS.Timeout | null = null;
let isFlushInProgress = false;
/**
* Flush buffered logs to database
*/
async function flushSessionLogs() {
if (sessionLogBuffer.length === 0 || isFlushInProgress) {
return;
}
isFlushInProgress = true;
// Take all current logs and clear buffer
const logsToWrite = sessionLogBuffer.splice(0, sessionLogBuffer.length);
try {
// Use a transaction to ensure all inserts succeed or fail together
await logsDb.transaction(async (tx) => {
// Batch insert logs in groups of 25 to avoid overwhelming the database
const BATCH_DB_SIZE = 25;
for (let i = 0; i < logsToWrite.length; i += BATCH_DB_SIZE) {
const batch = logsToWrite.slice(i, i + BATCH_DB_SIZE);
await tx.insert(aiSessionLog).values(batch);
}
});
logger.debug(
`Flushed ${logsToWrite.length} AI session logs to database`
);
} catch (error) {
logger.error("Error flushing AI session logs:", error);
// On transaction error, put logs back at the front of the buffer to retry
// but only if buffer isn't too large
if (sessionLogBuffer.length < MAX_BUFFER_SIZE - logsToWrite.length) {
sessionLogBuffer.unshift(...logsToWrite);
logger.info(
`Re-queued ${logsToWrite.length} AI session logs for retry`
);
} else {
logger.error(
`Buffer full, dropped ${logsToWrite.length} AI session logs`
);
}
} finally {
isFlushInProgress = false;
// If buffer filled up while we were flushing, flush again
if (sessionLogBuffer.length >= BATCH_SIZE) {
flushSessionLogs().catch((err) =>
logger.error("Error in follow-up AI session log flush:", err)
);
}
}
}
/**
* Schedule a flush if not already scheduled
*/
function scheduleFlush() {
if (flushTimer === null) {
flushTimer = setTimeout(() => {
flushTimer = null;
flushSessionLogs().catch((err) =>
logger.error("Error in scheduled AI session log flush:", err)
);
}, BATCH_INTERVAL_MS);
}
}
import { AiCapability } from "@app/lib/aiCapabilities";
import { AiProvider } from "@server/db";
/**
* Gracefully flush all pending logs (call this on shutdown)
*/
export async function shutdownAiSessionLogger() {
if (flushTimer) {
clearTimeout(flushTimer);
flushTimer = null;
}
// Force flush even if one is in progress by waiting and retrying
while (isFlushInProgress) {
await new Promise((resolve) => setTimeout(resolve, 100));
}
await flushSessionLogs();
}
export async function shutdownAiSessionLogger() {}
async function getRetentionDays(orgId: string): Promise<number> {
// check cache first
const cached = await cache.get<number>(`org_${orgId}_aiSessionsDays`);
if (cached !== undefined) {
return cached;
}
const [org] = await db
.select({
settingsLogRetentionDaysAISessions:
orgs.settingsLogRetentionDaysAISessions
})
.from(orgs)
.where(eq(orgs.orgId, orgId))
.limit(1);
if (!org) {
return 0;
}
// store the result in cache
await cache.set(
`org_${orgId}_aiSessionsDays`,
org.settingsLogRetentionDaysAISessions,
300
);
return org.settingsLogRetentionDaysAISessions;
}
export async function cleanUpOldLogs(orgId: string, retentionDays: number) {
// calculateCutoffTimestamp returns a seconds-epoch cutoff (built for
// requestAuditLog.timestamp), but aiSessionLog.createdAt is ms-epoch to
// match aiUsageRecords - convert before comparing.
const cutoffTimestampMs = calculateCutoffTimestamp(retentionDays) * 1000;
try {
await logsDb
.delete(aiSessionLog)
.where(
and(
lt(aiSessionLog.createdAt, cutoffTimestampMs),
eq(aiSessionLog.orgId, orgId)
)
);
} catch (error) {
logger.error("Error cleaning up old AI session logs:", error);
}
}
function truncateBody(value: string): { value: string; truncated: boolean } {
if (value.length <= AI_SESSION_LOG_MAX_BODY_CHARS) {
return { value, truncated: false };
}
return {
value: value.slice(0, AI_SESSION_LOG_MAX_BODY_CHARS),
truncated: true
};
}
export async function cleanUpOldLogs(orgId: string, retentionDays: number) {}
export function logAiSession(data: {
sessionId: string;
@@ -181,95 +22,4 @@ export function logAiSession(data: {
siteResourceId: number | null;
requestUserId: string | null;
virtualApiKeyId: string | null;
}): void {
(async () => {
try {
// Check retention before buffering any logs
if (data.orgId) {
const retentionDays = await getRetentionDays(data.orgId);
if (retentionDays === 0) {
// do not log
return;
}
} else {
// No org resolved for this request - nothing to govern
// retention with, so don't log it.
return;
}
const requestBodyText = truncateBody(
JSON.stringify(data.requestBody ?? "")
);
const responseBodyText = truncateBody(data.responseText ?? "");
// Uniform, capability-agnostic transcript for search/display -
// computed from the untruncated originals so normalization sees
// the full content; the normalized result gets its own
// (typically much smaller) truncation pass below.
const normalizedRequestMessages = normalizeAiRequest(
data.capability,
data.requestBody
);
const normalizedResponseMessages = normalizeAiResponse(
data.capability,
data.responseText ?? "",
data.isStream
);
const normalizedRequestText = normalizedRequestMessages
? truncateBody(JSON.stringify(normalizedRequestMessages))
: null;
const normalizedResponseText = normalizedResponseMessages
? truncateBody(JSON.stringify(normalizedResponseMessages))
: null;
// Prevent unbounded buffer growth - drop oldest entries if buffer is too large
if (sessionLogBuffer.length >= MAX_BUFFER_SIZE) {
const dropped = sessionLogBuffer.splice(0, BATCH_SIZE);
logger.warn(
`AI session log buffer exceeded max size (${MAX_BUFFER_SIZE}), dropped ${dropped.length} oldest entries`
);
}
sessionLogBuffer.push({
sessionId: data.sessionId,
orgId: sanitizeString(data.orgId),
providerId: data.provider.providerId,
capability: data.capability,
resourceId: data.resourceId ?? undefined,
siteResourceId: data.siteResourceId ?? undefined,
userId: sanitizeString(data.requestUserId ?? undefined),
virtualApiKeyId: sanitizeString(
data.virtualApiKeyId ?? undefined
),
requestedModel: sanitizeString(data.requestedModel),
isStream: data.isStream,
requestBody: sanitizeString(requestBodyText.value),
responseBody: sanitizeString(responseBodyText.value),
normalizedRequest: normalizedRequestText
? sanitizeString(normalizedRequestText.value)
: undefined,
normalizedResponse: normalizedResponseText
? sanitizeString(normalizedResponseText.value)
: undefined,
truncated:
requestBodyText.truncated ||
responseBodyText.truncated ||
(normalizedRequestText?.truncated ?? false) ||
(normalizedResponseText?.truncated ?? false),
statusCode: data.statusCode,
createdAt: Date.now()
});
// Flush immediately if buffer is full, otherwise schedule a flush
if (sessionLogBuffer.length >= BATCH_SIZE) {
flushSessionLogs().catch((err) =>
logger.error("Error flushing AI session logs:", err)
);
} else {
scheduleFlush();
}
} catch (error) {
logger.error("Failed to log AI session", { error });
}
})();
}
}): void {}
server/routers/aiGateway/pipeline.ts
+20 -11
View File
@@ -86,7 +86,7 @@ import {
type AiUsage
} from "@server/lib/aiUsageExtraction";
import { streamAiGatewayResponse } from "@server/routers/aiGateway/streamAiGatewayResponse";
import { logAiSession } from "@server/routers/aiGateway/logAiSession";
import { logAiSession } from "#dynamic/routers/aiGateway/logAiSession";
const EXIT_NODE_RANGES_CACHE_KEY = "aiGateway:exitNodeRanges";
const EXIT_NODE_RANGES_TTL_SEC = 6000;
@@ -137,7 +137,7 @@ async function findClientByIp(ip: string): Promise<CachedClient> {
return result;
}
type ProviderAttachment = {
export type ProviderAttachment = {
provider: AiProvider;
accessMode: AccessMode;
};
@@ -149,12 +149,12 @@ type ResourceModelPattern = {
enabled: boolean;
};
type ProviderPatternLists = {
export type ProviderPatternLists = {
allows: string[];
blocks: string[];
};
type ResolvedTarget = {
export type ResolvedTarget = {
resourceId: number | null;
siteResourceId: number | null;
orgId: string | null;
@@ -362,7 +362,7 @@ function getRequestHeader(req: Request, name: string): string | undefined {
// request came through, per the trust middleware's resource-type header -
// falls back to checking both (public preferred on overlap) only when that
// header is absent, e.g. a request that reached the gateway outside Traefik.
async function resolveTarget(
export async function resolveTarget(
host: string,
resourceType: AiGatewayResourceType | null
): Promise<ResolvedTarget | null> {
@@ -728,7 +728,9 @@ export function recordAiGatewayCompletion(args: {
let cost: ReturnType<typeof calculateAiCost> = null;
if (upstreamSucceeded) {
usage = extractUsage(capability, responseText, isStream, headers) ?? emptyUsage();
usage =
extractUsage(capability, responseText, isStream, headers) ??
emptyUsage();
if (isUsageEmpty(usage)) {
usage = estimateUsage(
JSON.stringify(requestBody ?? ""),
@@ -810,6 +812,17 @@ export function recordAiGatewayCompletion(args: {
});
}
// p-host is only used sometimes when overriding the host header for some
// middleware proxy. Shared with the model-discovery endpoint so both resolve
// the inference resource off the same hostname.
export function resolveGatewayHost(req: Request): string {
return (
(req.headers["p-host"] as string | undefined) ||
req.headers.host ||
""
).split(":")[0];
}
export async function handleAiGatewayProxy(
req: Request,
res: Response,
@@ -818,11 +831,7 @@ export async function handleAiGatewayProxy(
try {
const def = AI_CAPABILITY_DEFS[capability];
const host = (
(req.headers["p-host"] as string | undefined) || // p-host is only used sometimes when overriding the host header for some middleware proxy
req.headers.host ||
""
).split(":")[0];
const host = resolveGatewayHost(req);
if (!host) {
return res
.status(HttpCode.BAD_REQUEST)
server/routers/aiGateway/targetRouting.ts
+7
View File
@@ -99,6 +99,7 @@ async function fetchProviderTargets(
method: targets.method,
exitNodeSubnet: sites.exitNodeSubnet,
reachableAt: exitNodes.reachableAt,
exitNodeType: exitNodes.type,
hcHealth: targetHealthCheck.hcHealth
})
.from(targets)
@@ -119,6 +120,12 @@ async function fetchProviderTargets(
if (!row.exitNodeSubnet || !row.reachableAt) {
continue;
}
// Sites connected to a remote exit node aren't reachable via a
// gerbil sidecar's /router/* proxy - only "gerbil" type exit nodes
// run that endpoint.
if (row.exitNodeType !== "gerbil") {
continue;
}
// A target with an active health check that's currently failing is
// taken out of rotation. No health check (null) or "unknown" (check
// hasn't run yet / hcEnabled is off) still routes normally, matching
server/routers/aiGateway/v1Models.ts
+308
View File
@@ -0,0 +1,308 @@
import { Request, Response } from "express";
import { inArray } from "drizzle-orm";
import { z } from "zod";
import { aiModels, db } from "@server/db";
import {
providerHasCapability,
type AiCapability
} from "@server/lib/aiCapabilities";
import {
buildAiCapabilityErrorBody,
type AiCapabilityErrorKind
} from "@server/lib/aiGatewayAuthError";
import {
getAiGatewayResourceType,
isAiGatewayTrustHeaderValid
} from "@server/lib/aiGatewayTrust";
import { resolveEffectiveLists } from "@server/lib/aiInferenceResource";
import { listCatalogEntriesForType } from "@server/lib/aiModelCatalog";
import {
listPermittedModels,
paginateModels,
MODEL_PAGE_DEFAULT_LIMIT,
MODEL_PAGE_MAX_LIMIT,
type CatalogModelMetadata,
type ConfiguredModel,
type ModelDiscoveryProvider
} from "@server/lib/aiModelDiscovery";
import type { AiProviderType } from "@server/lib/aiProviderDefaults";
import {
resolveGatewayHost,
resolveTarget,
type ProviderAttachment,
type ProviderPatternLists
} from "@server/routers/aiGateway/pipeline";
import logger from "@server/logger";
import HttpCode from "@server/types/HttpCode";
const CAPABILITY: AiCapability = "v1_models";
const querySchema = z.object({
limit: z.coerce.number().int().min(1).max(MODEL_PAGE_MAX_LIMIT).optional(),
after_id: z.string().min(1).optional(),
before_id: z.string().min(1).optional()
});
type ProviderModelLists = {
allowsByProvider: Map<number, string[]>;
blocksByProvider: Map<number, string[]>;
configuredByProvider: Map<number, Map<string, ConfiguredModel>>;
};
function errorResponse(
res: Response,
status: number,
kind: AiCapabilityErrorKind,
message: string
) {
return res
.status(status)
.json(buildAiCapabilityErrorBody(CAPABILITY, kind, message, status));
}
// Provider-level allow/block lists, plus the display name and creation time of
// every catalog row, so explicitly configured models are reported with the name
// the administrator gave them rather than a bare model id.
async function loadProviderModelLists(
providerIds: number[]
): Promise<ProviderModelLists> {
const lists: ProviderModelLists = {
allowsByProvider: new Map(),
blocksByProvider: new Map(),
configuredByProvider: new Map()
};
if (providerIds.length === 0) {
return lists;
}
const rows = await db
.select({
providerId: aiModels.providerId,
modelKey: aiModels.modelKey,
name: aiModels.name,
listType: aiModels.listType,
enabled: aiModels.enabled,
createdAt: aiModels.createdAt
})
.from(aiModels)
.where(inArray(aiModels.providerId, providerIds));
for (const row of rows) {
if (!row.enabled) {
continue;
}
const targetMap =
row.listType === "allow"
? lists.allowsByProvider
: lists.blocksByProvider;
const existing = targetMap.get(row.providerId) ?? [];
existing.push(row.modelKey);
targetMap.set(row.providerId, existing);
let configured = lists.configuredByProvider.get(row.providerId);
if (!configured) {
configured = new Map();
lists.configuredByProvider.set(row.providerId, configured);
}
configured.set(row.modelKey, {
name: row.name,
createdAt: row.createdAt
});
}
return lists;
}
function catalogMetadataForType(
type: AiProviderType
): Map<string, CatalogModelMetadata> {
const metadata = new Map<string, CatalogModelMetadata>();
for (const entry of listCatalogEntriesForType(type)) {
metadata.set(entry.model, {
maxInputTokens: entry.limits.input,
maxOutputTokens: entry.limits.output,
capabilities: entry.capabilities
});
}
return metadata;
}
function buildDiscoveryProviders(
attachments: ProviderAttachment[],
resourceListsByProvider: Map<number, ProviderPatternLists>,
lists: ProviderModelLists
): ModelDiscoveryProvider[] {
return attachments.map((attachment) => {
const providerId = attachment.provider.providerId;
const resourceLists = resourceListsByProvider.get(providerId);
const { allows, blocks } = resolveEffectiveLists({
accessMode: attachment.accessMode,
providerAllows: lists.allowsByProvider.get(providerId) ?? [],
providerBlocks: lists.blocksByProvider.get(providerId) ?? [],
resourceAllows: resourceLists?.allows ?? [],
resourceBlocks: resourceLists?.blocks ?? []
});
return {
providerId,
allows,
blocks,
catalog: catalogMetadataForType(
attachment.provider.type as AiProviderType
),
configured: lists.configuredByProvider.get(providerId) ?? new Map()
};
});
}
/**
* Serves Anthropic's model-discovery endpoints (`GET /v1/models` and
* `GET /v1/models/{id}`) for an inference resource. The gateway answers these
* itself rather than proxying: upstream providers either don't expose a model
* list at all or would expose models the resource's allow/block lists forbid,
* so the response is built from the same effective lists that gate inference.
*/
export async function handleV1Models(
req: Request,
res: Response
): Promise<any> {
try {
const host = resolveGatewayHost(req);
if (!host) {
return errorResponse(
res,
HttpCode.BAD_REQUEST,
"invalid_request",
"Missing Host header"
);
}
const resourceType = getAiGatewayResourceType(
req.headers as Record<string, string>
);
const target = await resolveTarget(host, resourceType);
if (!target) {
return errorResponse(
res,
HttpCode.NOT_FOUND,
"not_found",
"No inference resource found for this host"
);
}
// Same gate as the inference pipeline: public inference must pass
// Badger verify-session first, which is what stamps the trust header.
if (
target.resourceId != null &&
!isAiGatewayTrustHeaderValid(req.headers as Record<string, string>)
) {
return errorResponse(
res,
HttpCode.UNAUTHORIZED,
"authentication",
"Request must be authenticated via the inference resource"
);
}
if (target.attachments.length === 0) {
return errorResponse(
res,
HttpCode.FORBIDDEN,
"permission",
"No AI providers configured for this resource"
);
}
const capableAttachments = target.attachments.filter((a) =>
providerHasCapability(a.provider.capabilities, CAPABILITY)
);
if (capableAttachments.length === 0) {
return errorResponse(
res,
HttpCode.FORBIDDEN,
"permission",
`No AI provider on this resource supports ${CAPABILITY}`
);
}
const lists = await loadProviderModelLists(
capableAttachments.map((a) => a.provider.providerId)
);
const models = listPermittedModels(
buildDiscoveryProviders(
capableAttachments,
target.resourceListsByProvider,
lists
)
);
// `GET /v1/models/{id}` - a single model, 404 when this resource
// doesn't permit it.
const requestedModel = req.params?.model;
if (typeof requestedModel === "string" && requestedModel.length > 0) {
const model = models.find((m) => m.id === requestedModel);
if (!model) {
return errorResponse(
res,
HttpCode.NOT_FOUND,
"not_found",
`Model "${requestedModel}" is not available on this resource`
);
}
return res.status(HttpCode.OK).json(model);
}
const parsedQuery = querySchema.safeParse(req.query);
if (!parsedQuery.success) {
return errorResponse(
res,
HttpCode.BAD_REQUEST,
"invalid_request",
parsedQuery.error.issues[0]?.message ??
"Invalid pagination parameters"
);
}
const page = paginateModels(
models,
parsedQuery.data.limit ?? MODEL_PAGE_DEFAULT_LIMIT,
{
afterId: parsedQuery.data.after_id,
beforeId: parsedQuery.data.before_id
}
);
if ("error" in page) {
return errorResponse(
res,
HttpCode.BAD_REQUEST,
"invalid_request",
page.error
);
}
logger.debug("AI gateway model discovery", {
host,
resourceId: target.resourceId,
siteResourceId: target.siteResourceId,
providers: capableAttachments.length,
total: models.length,
returned: page.data.length
});
return res.status(HttpCode.OK).json({
data: page.data,
has_more: page.has_more,
first_id: page.data[0]?.id ?? null,
last_id: page.data[page.data.length - 1]?.id ?? null
});
} catch (error) {
logger.error(error);
return errorResponse(
res,
HttpCode.INTERNAL_SERVER_ERROR,
"internal",
"Failed to list models"
);
}
}
server/routers/certificates/createCertificate.ts
+32 -28
View File
@@ -23,6 +23,12 @@ export async function createCertificate(
throw new Error(`Domain with ID ${domainId} not found`);
}
// Note: certificates.domain has a global UNIQUE constraint (it is not
// scoped per-domainId), so existence must be checked by domain value
// alone. Filtering on domainId here as well can cause this check to
// miss an existing cert (e.g. if it was stored under a different but
// still-valid domainId), leading to an INSERT that then fails on the
// unique constraint.
let existing: Certificate[] = [];
if (domainRecord.type == "ns" || domainRecord.type == "wildcard") {
const domainLevelDown = domain.split(".").slice(1).join(".");
@@ -32,16 +38,13 @@ export async function createCertificate(
.select()
.from(certificates)
.where(
and(
eq(certificates.domainId, domainId),
or(
eq(certificates.domain, domain),
and(
eq(certificates.wildcard, true),
or(
eq(certificates.domain, domainLevelDown),
eq(certificates.domain, wildcardPrefixed)
)
or(
eq(certificates.domain, domain),
and(
eq(certificates.wildcard, true),
or(
eq(certificates.domain, domainLevelDown),
eq(certificates.domain, wildcardPrefixed)
)
)
)
@@ -51,12 +54,7 @@ export async function createCertificate(
existing = await trx
.select()
.from(certificates)
.where(
and(
eq(certificates.domainId, domainId),
eq(certificates.domain, domain) // exact match for non-NS domains
)
);
.where(eq(certificates.domain, domain)); // exact match for non-NS domains
}
if (existing.length > 0) {
@@ -87,16 +85,22 @@ export async function createCertificate(
}
}
// No cert found, create a new one in pending state
await trx.insert(certificates).values({
domain: domainToWrite,
domainId,
wildcard:
domainRecord.type == "ns" ||
(domainRecord.type == "wildcard" &&
domainRecord.preferWildcardCert), // we can only create wildcard certs for NS domains
status: "pending",
updatedAt: Math.floor(Date.now() / 1000),
createdAt: Math.floor(Date.now() / 1000)
});
// No cert found, create a new one in pending state. onConflictDoNothing
// guards against the domain having been inserted concurrently (or under
// a different domainId) between the existence check above and this
// insert, since certificates.domain is globally unique.
await trx
.insert(certificates)
.values({
domain: domainToWrite,
domainId,
wildcard:
domainRecord.type == "ns" ||
(domainRecord.type == "wildcard" &&
domainRecord.preferWildcardCert), // we can only create wildcard certs for NS domains
status: "pending",
updatedAt: Math.floor(Date.now() / 1000),
createdAt: Math.floor(Date.now() / 1000)
})
.onConflictDoNothing();
}
server/routers/external.ts
-15
View File
@@ -1490,21 +1490,6 @@ authenticated.get(
logs.exportRequestAuditLogs
);
authenticated.get(
"/org/:orgId/logs/ai",
verifyOrgAccess,
verifyUserHasAction(ActionsEnum.viewLogs),
logs.queryAiSessionLogs
);
authenticated.get(
"/org/:orgId/logs/ai/export",
verifyOrgAccess,
verifyUserHasAction(ActionsEnum.exportLogs),
logActionAudit(ActionsEnum.exportLogs),
logs.exportAiSessionLogs
);
authenticated.get(
"/org/:orgId/logs/ai/usage/filters",
verifyOrgAccess,
server/routers/gerbil/createExitNode.ts
+1 -7
View File
@@ -15,13 +15,7 @@ export async function createExitNode(
if (!exitNodeQuery) {
const { value: address, release } = await getNextAvailableSubnet();
try {
// TODO: eventually we will want to get the next available port so that we can multiple exit nodes
// const listenPort = await getNextAvailablePort();
const listenPort = config.getRawConfig().gerbil.start_port;
let subEndpoint = "";
if (config.getRawConfig().gerbil.use_subdomain) {
subEndpoint = await getUniqueExitNodeEndpointName();
}
const exitNodeName =
config.getRawConfig().gerbil.exit_node_name ||
@@ -32,7 +26,7 @@ export async function createExitNode(
.insert(exitNodes)
.values({
publicKey,
endpoint: `${subEndpoint}${subEndpoint != "" ? "." : ""}${config.getRawConfig().gerbil.base_endpoint}`,
endpoint: config.getRawConfig().gerbil.base_endpoint,
address,
online: true,
listenPort,
server/routers/integration.ts
-15
View File
@@ -1532,21 +1532,6 @@ authenticated.get(
logs.exportRequestAuditLogs
);
authenticated.get(
"/org/:orgId/logs/ai",
verifyApiKeyOrgAccess,
verifyApiKeyHasAction(ActionsEnum.viewLogs),
logs.queryAiSessionLogs
);
authenticated.get(
"/org/:orgId/logs/ai/export",
verifyApiKeyOrgAccess,
verifyApiKeyHasAction(ActionsEnum.exportLogs),
logActionAudit(ActionsEnum.exportLogs),
logs.exportAiSessionLogs
);
authenticated.get(
"/org/:orgId/logs/ai/usage/filters",
verifyApiKeyOrgAccess,
server/routers/newt/handleRequestLogMessage.ts
+241 -2
View File
@@ -1,9 +1,248 @@
/*
* This file is part of a proprietary work.
*
* Copyright (c) 2025-2026 Fossorial, Inc.
* All rights reserved.
*
* This file is licensed under the Fossorial Commercial License.
* You may not use this file except in compliance with the License.
* Unauthorized use, copying, modification, or distribution is strictly prohibited.
*
* This file is not licensed under the AGPLv3.
*/
import { db } from "@server/db";
import { MessageHandler } from "@server/routers/ws";
import { sites, Newt, orgs, clients, clientSitesAssociationsCache, users } from "@server/db";
import { and, eq, inArray } from "drizzle-orm";
import logger from "@server/logger";
import { inflate } from "zlib";
import { promisify } from "util";
import { logRequestAudit } from "@server/routers/badger/logRequestAudit";
import { getCountryCodeForIp } from "@server/lib/geoip";
export async function flushRequestLogToDb(): Promise<void> {
return;
}
const zlibInflate = promisify(inflate);
interface HTTPRequestLogData {
requestId: string;
resourceId: number; // siteResourceId
timestamp: string; // ISO 8601
method: string;
scheme: string; // "http" or "https"
host: string;
path: string;
rawQuery?: string;
userAgent?: string;
sourceAddr: string; // ip:port
tls: boolean;
}
/**
* Decompress a base64-encoded zlib-compressed string into parsed JSON.
*/
async function decompressRequestLog(
compressed: string
): Promise<HTTPRequestLogData[]> {
const compressedBuffer = Buffer.from(compressed, "base64");
const decompressed = await zlibInflate(compressedBuffer);
const jsonString = decompressed.toString("utf-8");
const parsed = JSON.parse(jsonString);
if (!Array.isArray(parsed)) {
throw new Error("Decompressed request log data is not an array");
}
return parsed;
}
export const handleRequestLogMessage: MessageHandler = async (context) => {
return;
};
const { message, client } = context;
const newt = client as Newt;
if (!newt) {
logger.warn("Request log received but no newt client in context");
return;
}
if (!newt.siteId) {
logger.warn("Request log received but newt has no siteId");
return;
}
if (!message.data?.compressed) {
logger.warn("Request log message missing compressed data");
return;
}
// Look up the org for this site and check retention settings
const [site] = await db
.select({
orgId: sites.orgId,
orgSubnet: orgs.subnet,
settingsLogRetentionDaysRequest:
orgs.settingsLogRetentionDaysRequest
})
.from(sites)
.innerJoin(orgs, eq(sites.orgId, orgs.orgId))
.where(eq(sites.siteId, newt.siteId));
if (!site) {
logger.warn(
`Request log received but site ${newt.siteId} not found in database`
);
return;
}
const orgId = site.orgId;
if (site.settingsLogRetentionDaysRequest === 0) {
logger.debug(
`Request log retention is disabled for org ${orgId}, skipping`
);
return;
}
let entries: HTTPRequestLogData[];
try {
entries = await decompressRequestLog(message.data.compressed);
} catch (error) {
logger.error("Failed to decompress request log data:", error);
return;
}
if (entries.length === 0) {
return;
}
logger.debug(`Request log entries: ${JSON.stringify(entries)}`);
// Build a map from sourceIp → external endpoint string by joining clients
// with clientSitesAssociationsCache. The endpoint is the real-world IP:port
// of the client device and is used for GeoIP lookup.
const ipToEndpoint = new Map<string, string>();
// Build a map from sourceIp → the user associated with the client (if any)
const ipToUser = new Map<string, { username: string; userId: string }>();
const cidrSuffix = site.orgSubnet?.includes("/")
? site.orgSubnet.substring(site.orgSubnet.indexOf("/"))
: null;
if (cidrSuffix) {
const uniqueSourceAddrs = new Set<string>();
for (const entry of entries) {
if (entry.sourceAddr) {
uniqueSourceAddrs.add(entry.sourceAddr);
}
}
if (uniqueSourceAddrs.size > 0) {
const subnetQueries = Array.from(uniqueSourceAddrs).map((addr) => {
const ip = addr.includes(":") ? addr.split(":")[0] : addr;
return `${ip}${cidrSuffix}`;
});
const matchedClients = await db
.select({
subnet: clients.subnet,
endpoint: clientSitesAssociationsCache.endpoint,
username: users.username,
userId: users.userId
})
.from(clients)
.innerJoin(
clientSitesAssociationsCache,
and(
eq(
clientSitesAssociationsCache.clientId,
clients.clientId
),
eq(clientSitesAssociationsCache.siteId, newt.siteId)
)
)
.leftJoin(users, eq(clients.userId, users.userId))
.where(
and(
eq(clients.orgId, orgId),
inArray(clients.subnet, subnetQueries)
)
);
for (const c of matchedClients) {
const ip = c.subnet.split("/")[0];
if (c.endpoint) {
ipToEndpoint.set(ip, c.endpoint);
}
if (c.userId && c.username) {
ipToUser.set(ip, { userId: c.userId, username: c.username });
}
}
}
}
for (const entry of entries) {
if (
!entry.requestId ||
!entry.resourceId ||
!entry.method ||
!entry.scheme ||
!entry.host ||
!entry.path ||
!entry.sourceAddr
) {
logger.debug(
`Skipping request log entry with missing required fields: ${JSON.stringify(entry)}`
);
continue;
}
const originalRequestURL =
entry.scheme +
"://" +
entry.host +
entry.path +
(entry.rawQuery ? "?" + entry.rawQuery : "");
// Resolve the client's external endpoint for GeoIP lookup.
// sourceAddr is the WireGuard IP (possibly ip:port), so strip the port.
const sourceIp = entry.sourceAddr.includes(":")
? entry.sourceAddr.split(":")[0]
: entry.sourceAddr;
const endpoint = ipToEndpoint.get(sourceIp);
let location: string | undefined;
if (endpoint) {
const endpointIp = endpoint.includes(":")
? endpoint.split(":")[0]
: endpoint;
location = await getCountryCodeForIp(endpointIp);
}
const user = ipToUser.get(sourceIp);
await logRequestAudit(
{
action: true,
reason: 108,
siteResourceId: entry.resourceId,
orgId,
location,
user
},
{
path: entry.path,
originalRequestURL,
scheme: entry.scheme,
host: entry.host,
method: entry.method,
tls: entry.tls,
requestIp: entry.sourceAddr
}
);
}
logger.debug(
`Buffered ${entries.length} request log entry/entries from newt ${newt.newtId} (site ${newt.siteId})`
);
};
server/routers/olm/handleOlmExitNodesRequestMessage.ts
+2 -1
View File
@@ -42,7 +42,8 @@ export const handleOlmExitNodesRequestMessage: MessageHandler = async (
client.orgId,
true,
noCloud || false,
olm.clientId
olm.clientId,
true // don't select remote exit nodes for clients
); // filter for only the online ones
let lastExitNodeId = null;
server/routers/org/updateOrg.ts
+36
View File
@@ -147,6 +147,42 @@ export async function updateOrg(
parsedBody.data.settingsEnableGlobalNewtAutoUpdate = false; // force it off
}
// Check access logs feature
const hasAccessLogsFeature = await isLicensedOrSubscribed(
orgId,
tierMatrix[TierFeature.AccessLogs]
);
if (!hasAccessLogsFeature) {
parsedBody.data.settingsLogRetentionDaysAccess = undefined;
}
// Check action logs feature
const hasActionLogsFeature = await isLicensedOrSubscribed(
orgId,
tierMatrix[TierFeature.ActionLogs]
);
if (!hasActionLogsFeature) {
parsedBody.data.settingsLogRetentionDaysAction = undefined;
}
// Check connection logs feature
const hasConnectionLogsFeature = await isLicensedOrSubscribed(
orgId,
tierMatrix[TierFeature.ConnectionLogs]
);
if (!hasConnectionLogsFeature) {
parsedBody.data.settingsLogRetentionDaysConnection = undefined;
}
// Check AI session logs feature
const hasAISessionLogsFeature = await isLicensedOrSubscribed(
orgId,
tierMatrix[TierFeature.AISessionLogs]
);
if (!hasAISessionLogsFeature) {
parsedBody.data.settingsLogRetentionDaysAISessions = undefined;
}
if (build == "saas") {
const { tier } = await getOrgTierData(orgId);
server/routers/resource/listResourceUsers.ts
+38 -3
View File
@@ -1,7 +1,7 @@
import { Request, Response, NextFunction } from "express";
import { z } from "zod";
import { db } from "@server/db";
import { idp, userResources, users } from "@server/db"; // Assuming these are the correct tables
import { idp, resources, userPolicies, userResources, users } from "@server/db"; // Assuming these are the correct tables
import { eq } from "drizzle-orm";
import response from "@server/lib/response";
import HttpCode from "@server/types/HttpCode";
@@ -14,7 +14,23 @@ const listResourceUsersSchema = z.strictObject({
resourceId: z.coerce.number().int().positive()
});
async function queryUsers(resourceId: number) {
async function queryUsers(resourceId: number, policyId: number | null) {
if (policyId !== null) {
return await db
.select({
userId: userPolicies.userId,
username: users.username,
type: users.type,
idpName: idp.name,
idpId: users.idpId,
email: users.email
})
.from(userPolicies)
.innerJoin(users, eq(userPolicies.userId, users.userId))
.leftJoin(idp, eq(users.idpId, idp.idpId))
.where(eq(userPolicies.resourcePolicyId, policyId));
}
return await db
.select({
userId: userResources.userId,
@@ -104,7 +120,26 @@ export async function listResourceUsers(
const { resourceId } = parsedParams.data;
const resourceUsersList = await queryUsers(resourceId);
const [resource] = await db
.select()
.from(resources)
.where(eq(resources.resourceId, resourceId))
.limit(1);
if (!resource) {
return next(
createHttpError(HttpCode.NOT_FOUND, "Resource not found")
);
}
const isInlinePolicy =
resource.resourcePolicyId === null &&
resource.defaultResourcePolicyId !== null;
const resourceUsersList = await queryUsers(
resourceId,
isInlinePolicy ? resource.defaultResourcePolicyId! : null
);
return response<ListResourceUsersResponse>(res, {
data: {
server/routers/siteResource/addClientToSiteResource.ts
+1 -1
View File
@@ -64,7 +64,7 @@ registry.registerPath({
registry.registerPath({
method: "post",
path: "/private-resource/{siteResourceId}/clients/add",
path: "/private-resource/{resourceId}/clients/add",
description:
"Add a single client to a site resource. Clients with a userId cannot be added.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.Client],
server/routers/siteResource/addRoleToSiteResource.ts
+1 -1
View File
@@ -64,7 +64,7 @@ registry.registerPath({
registry.registerPath({
method: "post",
path: "/private-resource/{siteResourceId}/roles/add",
path: "/private-resource/{resourceId}/roles/add",
description: "Add a single role to a site resource.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.Role],
request: {
server/routers/siteResource/addUserToSiteResource.ts
+1 -1
View File
@@ -64,7 +64,7 @@ registry.registerPath({
registry.registerPath({
method: "post",
path: "/private-resource/{siteResourceId}/users/add",
path: "/private-resource/{resourceId}/users/add",
description: "Add a single user to a site resource.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.User],
request: {
server/routers/siteResource/listSiteResourceClients.ts
+1 -1
View File
@@ -63,7 +63,7 @@ registry.registerPath({
registry.registerPath({
method: "get",
path: "/private-resource/{siteResourceId}/clients",
path: "/private-resource/{resourceId}/clients",
description: "List all clients for a site resource.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.Client],
request: {
server/routers/siteResource/listSiteResourceRoles.ts
+1 -1
View File
@@ -64,7 +64,7 @@ registry.registerPath({
registry.registerPath({
method: "get",
path: "/private-resource/{siteResourceId}/roles",
path: "/private-resource/{resourceId}/roles",
description: "List all roles for a site resource.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.Role],
request: {
server/routers/siteResource/listSiteResourceUsers.ts
+1 -1
View File
@@ -67,7 +67,7 @@ registry.registerPath({
registry.registerPath({
method: "get",
path: "/private-resource/{siteResourceId}/users",
path: "/private-resource/{resourceId}/users",
description: "List all users for a site resource.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.User],
request: {
server/routers/siteResource/removeClientFromSiteResource.ts
+1 -1
View File
@@ -64,7 +64,7 @@ registry.registerPath({
registry.registerPath({
method: "post",
path: "/private-resource/{siteResourceId}/clients/remove",
path: "/private-resource/{resourceId}/clients/remove",
description:
"Remove a single client from a site resource. Clients with a userId cannot be removed.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.Client],
server/routers/siteResource/removeRoleFromSiteResource.ts
+1 -1
View File
@@ -64,7 +64,7 @@ registry.registerPath({
registry.registerPath({
method: "post",
path: "/private-resource/{siteResourceId}/roles/remove",
path: "/private-resource/{resourceId}/roles/remove",
description: "Remove a single role from a site resource.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.Role],
request: {
server/routers/siteResource/removeUserFromSiteResource.ts
+1 -1
View File
@@ -64,7 +64,7 @@ registry.registerPath({
registry.registerPath({
method: "post",
path: "/private-resource/{siteResourceId}/users/remove",
path: "/private-resource/{resourceId}/users/remove",
description: "Remove a single user from a site resource.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.User],
request: {
server/routers/siteResource/setSiteResourceClients.ts
+1 -1
View File
@@ -64,7 +64,7 @@ registry.registerPath({
registry.registerPath({
method: "post",
path: "/private-resource/{siteResourceId}/clients",
path: "/private-resource/{resourceId}/clients",
description:
"Set clients for a site resource. This will replace all existing clients. Clients with a userId cannot be added.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.Client],
server/routers/siteResource/setSiteResourceRoles.ts
+1 -1
View File
@@ -65,7 +65,7 @@ registry.registerPath({
registry.registerPath({
method: "post",
path: "/private-resource/{siteResourceId}/roles",
path: "/private-resource/{resourceId}/roles",
description:
"Set roles for a site resource. This will replace all existing roles.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.Role],
server/routers/siteResource/setSiteResourceUsers.ts
+1 -1
View File
@@ -66,7 +66,7 @@ registry.registerPath({
registry.registerPath({
method: "post",
path: "/private-resource/{siteResourceId}/users",
path: "/private-resource/{resourceId}/users",
description:
"Set users for a site resource. This will replace all existing users.",
tags: [OpenAPITags.PrivateResource, OpenAPITags.User],
server/routers/ws/messageHandlers.ts
+3 -1
View File
@@ -7,7 +7,8 @@ import {
handleNewtExitNodesRequestMessage,
handleApplyBlueprintMessage,
handleNewtPingMessage,
handleNewtDisconnectingMessage
handleNewtDisconnectingMessage,
handleRequestLogMessage
} from "../newt";
import {
handleOlmRegisterMessage,
@@ -46,5 +47,6 @@ export const messageHandlers: Record<string, MessageHandler> = {
"newt/ping/request": handleNewtExitNodesRequestMessage,
"newt/blueprint/apply": handleApplyBlueprintMessage,
"newt/healthcheck/status": handleHealthcheckStatusMessage,
"newt/request-log": handleRequestLogMessage,
"ws/round-trip/complete": handleRoundTripMessage
};
server/routers/ws/ws.ts
+1 -1
View File
@@ -388,7 +388,7 @@ const setupConnection = async (
}
}
} catch (error) {
logger.error("Message handling error:", error);
logger.warn("Message handling error:", error);
ws.send(
JSON.stringify({
type: "error",