mirror of
https://github.com/fosrl/pangolin.git
synced 2026-03-24 03:26:39 +00:00
Compare commits
20 Commits
1.16.2-s.1
...
logging-pr
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0d4edcd1c7 | ||
|
|
7d8797840a | ||
|
|
19f8c1772f | ||
|
|
37d331e813 | ||
|
|
c660df55cd | ||
|
|
7c8b865379 | ||
|
|
3cca0c09c0 | ||
|
|
7c2b4f422a | ||
|
|
ad2a0ae127 | ||
|
|
6c2c620c99 | ||
|
|
f643abf19a | ||
|
|
a1729033cf | ||
|
|
7311766512 | ||
|
|
17105f3a51 | ||
|
|
edcfbd26e4 | ||
|
|
0c4d9ea164 | ||
|
|
b01fcc70fe | ||
|
|
35fed74e49 | ||
|
|
6cf1b9b010 | ||
|
|
dae169540b |
@@ -1,9 +1,11 @@
|
||||
import { flushBandwidthToDb } from "@server/routers/newt/handleReceiveBandwidthMessage";
|
||||
import { flushConnectionLogToDb } from "#dynamic/routers/newt";
|
||||
import { flushSiteBandwidthToDb } from "@server/routers/gerbil/receiveBandwidth";
|
||||
import { cleanup as wsCleanup } from "#dynamic/routers/ws";
|
||||
|
||||
async function cleanup() {
|
||||
await flushBandwidthToDb();
|
||||
await flushConnectionLogToDb();
|
||||
await flushSiteBandwidthToDb();
|
||||
await wsCleanup();
|
||||
|
||||
|
||||
@@ -17,7 +17,9 @@ import {
|
||||
users,
|
||||
exitNodes,
|
||||
sessions,
|
||||
clients
|
||||
clients,
|
||||
siteResources,
|
||||
sites
|
||||
} from "./schema";
|
||||
|
||||
export const certificates = pgTable("certificates", {
|
||||
@@ -302,6 +304,39 @@ export const accessAuditLog = pgTable(
|
||||
]
|
||||
);
|
||||
|
||||
export const connectionAuditLog = pgTable(
|
||||
"connectionAuditLog",
|
||||
{
|
||||
id: serial("id").primaryKey(),
|
||||
sessionId: text("sessionId").notNull(),
|
||||
siteResourceId: integer("siteResourceId").references(
|
||||
() => siteResources.siteResourceId,
|
||||
{ onDelete: "cascade" }
|
||||
),
|
||||
orgId: text("orgId").references(() => orgs.orgId, {
|
||||
onDelete: "cascade"
|
||||
}),
|
||||
siteId: integer("siteId").references(() => sites.siteId, {
|
||||
onDelete: "cascade"
|
||||
}),
|
||||
sourceAddr: text("sourceAddr").notNull(),
|
||||
destAddr: text("destAddr").notNull(),
|
||||
protocol: text("protocol").notNull(),
|
||||
startedAt: integer("startedAt").notNull(),
|
||||
endedAt: integer("endedAt"),
|
||||
bytesTx: integer("bytesTx"),
|
||||
bytesRx: integer("bytesRx")
|
||||
},
|
||||
(table) => [
|
||||
index("idx_accessAuditLog_startedAt").on(table.startedAt),
|
||||
index("idx_accessAuditLog_org_startedAt").on(
|
||||
table.orgId,
|
||||
table.startedAt
|
||||
),
|
||||
index("idx_accessAuditLog_siteResourceId").on(table.siteResourceId)
|
||||
]
|
||||
);
|
||||
|
||||
export const approvals = pgTable("approvals", {
|
||||
approvalId: serial("approvalId").primaryKey(),
|
||||
timestamp: integer("timestamp").notNull(), // this is EPOCH time in seconds
|
||||
@@ -357,3 +392,4 @@ export type LoginPage = InferSelectModel<typeof loginPage>;
|
||||
export type LoginPageBranding = InferSelectModel<typeof loginPageBranding>;
|
||||
export type ActionAuditLog = InferSelectModel<typeof actionAuditLog>;
|
||||
export type AccessAuditLog = InferSelectModel<typeof accessAuditLog>;
|
||||
export type ConnectionAuditLog = InferSelectModel<typeof connectionAuditLog>;
|
||||
|
||||
@@ -55,6 +55,9 @@ export const orgs = pgTable("orgs", {
|
||||
settingsLogRetentionDaysAction: integer("settingsLogRetentionDaysAction") // where 0 = dont keep logs and -1 = keep forever and 9001 = end of the following year
|
||||
.notNull()
|
||||
.default(0),
|
||||
settingsLogRetentionDaysConnection: integer("settingsLogRetentionDaysConnection") // where 0 = dont keep logs and -1 = keep forever and 9001 = end of the following year
|
||||
.notNull()
|
||||
.default(0),
|
||||
sshCaPrivateKey: text("sshCaPrivateKey"), // Encrypted SSH CA private key (PEM format)
|
||||
sshCaPublicKey: text("sshCaPublicKey"), // SSH CA public key (OpenSSH format)
|
||||
isBillingOrg: boolean("isBillingOrg"),
|
||||
|
||||
@@ -6,7 +6,7 @@ import {
|
||||
sqliteTable,
|
||||
text
|
||||
} from "drizzle-orm/sqlite-core";
|
||||
import { clients, domains, exitNodes, orgs, sessions, users } from "./schema";
|
||||
import { clients, domains, exitNodes, orgs, sessions, siteResources, sites, users } from "./schema";
|
||||
|
||||
export const certificates = sqliteTable("certificates", {
|
||||
certId: integer("certId").primaryKey({ autoIncrement: true }),
|
||||
@@ -294,6 +294,39 @@ export const accessAuditLog = sqliteTable(
|
||||
]
|
||||
);
|
||||
|
||||
export const connectionAuditLog = sqliteTable(
|
||||
"connectionAuditLog",
|
||||
{
|
||||
id: integer("id").primaryKey({ autoIncrement: true }),
|
||||
sessionId: text("sessionId").notNull(),
|
||||
siteResourceId: integer("siteResourceId").references(
|
||||
() => siteResources.siteResourceId,
|
||||
{ onDelete: "cascade" }
|
||||
),
|
||||
orgId: text("orgId").references(() => orgs.orgId, {
|
||||
onDelete: "cascade"
|
||||
}),
|
||||
siteId: integer("siteId").references(() => sites.siteId, {
|
||||
onDelete: "cascade"
|
||||
}),
|
||||
sourceAddr: text("sourceAddr").notNull(),
|
||||
destAddr: text("destAddr").notNull(),
|
||||
protocol: text("protocol").notNull(),
|
||||
startedAt: integer("startedAt").notNull(),
|
||||
endedAt: integer("endedAt"),
|
||||
bytesTx: integer("bytesTx"),
|
||||
bytesRx: integer("bytesRx")
|
||||
},
|
||||
(table) => [
|
||||
index("idx_accessAuditLog_startedAt").on(table.startedAt),
|
||||
index("idx_accessAuditLog_org_startedAt").on(
|
||||
table.orgId,
|
||||
table.startedAt
|
||||
),
|
||||
index("idx_accessAuditLog_siteResourceId").on(table.siteResourceId)
|
||||
]
|
||||
);
|
||||
|
||||
export const approvals = sqliteTable("approvals", {
|
||||
approvalId: integer("approvalId").primaryKey({ autoIncrement: true }),
|
||||
timestamp: integer("timestamp").notNull(), // this is EPOCH time in seconds
|
||||
@@ -348,3 +381,4 @@ export type LoginPage = InferSelectModel<typeof loginPage>;
|
||||
export type LoginPageBranding = InferSelectModel<typeof loginPageBranding>;
|
||||
export type ActionAuditLog = InferSelectModel<typeof actionAuditLog>;
|
||||
export type AccessAuditLog = InferSelectModel<typeof accessAuditLog>;
|
||||
export type ConnectionAuditLog = InferSelectModel<typeof connectionAuditLog>;
|
||||
|
||||
@@ -47,6 +47,9 @@ export const orgs = sqliteTable("orgs", {
|
||||
settingsLogRetentionDaysAction: integer("settingsLogRetentionDaysAction") // where 0 = dont keep logs and -1 = keep forever and 9001 = end of the following year
|
||||
.notNull()
|
||||
.default(0),
|
||||
settingsLogRetentionDaysConnection: integer("settingsLogRetentionDaysConnection") // where 0 = dont keep logs and -1 = keep forever and 9001 = end of the following year
|
||||
.notNull()
|
||||
.default(0),
|
||||
sshCaPrivateKey: text("sshCaPrivateKey"), // Encrypted SSH CA private key (PEM format)
|
||||
sshCaPublicKey: text("sshCaPublicKey"), // SSH CA public key (OpenSSH format)
|
||||
isBillingOrg: integer("isBillingOrg", { mode: "boolean" }),
|
||||
|
||||
@@ -2,6 +2,7 @@ import { db, orgs } from "@server/db";
|
||||
import { cleanUpOldLogs as cleanUpOldAccessLogs } from "#dynamic/lib/logAccessAudit";
|
||||
import { cleanUpOldLogs as cleanUpOldActionLogs } from "#dynamic/middlewares/logActionAudit";
|
||||
import { cleanUpOldLogs as cleanUpOldRequestLogs } from "@server/routers/badger/logRequestAudit";
|
||||
import { cleanUpOldLogs as cleanUpOldConnectionLogs } from "#dynamic/routers/newt";
|
||||
import { gt, or } from "drizzle-orm";
|
||||
import { cleanUpOldFingerprintSnapshots } from "@server/routers/olm/fingerprintingUtils";
|
||||
import { build } from "@server/build";
|
||||
@@ -20,14 +21,17 @@ export function initLogCleanupInterval() {
|
||||
settingsLogRetentionDaysAccess:
|
||||
orgs.settingsLogRetentionDaysAccess,
|
||||
settingsLogRetentionDaysRequest:
|
||||
orgs.settingsLogRetentionDaysRequest
|
||||
orgs.settingsLogRetentionDaysRequest,
|
||||
settingsLogRetentionDaysConnection:
|
||||
orgs.settingsLogRetentionDaysConnection
|
||||
})
|
||||
.from(orgs)
|
||||
.where(
|
||||
or(
|
||||
gt(orgs.settingsLogRetentionDaysAction, 0),
|
||||
gt(orgs.settingsLogRetentionDaysAccess, 0),
|
||||
gt(orgs.settingsLogRetentionDaysRequest, 0)
|
||||
gt(orgs.settingsLogRetentionDaysRequest, 0),
|
||||
gt(orgs.settingsLogRetentionDaysConnection, 0)
|
||||
)
|
||||
);
|
||||
|
||||
@@ -37,7 +41,8 @@ export function initLogCleanupInterval() {
|
||||
orgId,
|
||||
settingsLogRetentionDaysAction,
|
||||
settingsLogRetentionDaysAccess,
|
||||
settingsLogRetentionDaysRequest
|
||||
settingsLogRetentionDaysRequest,
|
||||
settingsLogRetentionDaysConnection
|
||||
} = org;
|
||||
|
||||
if (settingsLogRetentionDaysAction > 0) {
|
||||
@@ -60,6 +65,13 @@ export function initLogCleanupInterval() {
|
||||
settingsLogRetentionDaysRequest
|
||||
);
|
||||
}
|
||||
|
||||
if (settingsLogRetentionDaysConnection > 0) {
|
||||
await cleanUpOldConnectionLogs(
|
||||
orgId,
|
||||
settingsLogRetentionDaysConnection
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
await cleanUpOldFingerprintSnapshots(365);
|
||||
|
||||
123
server/lib/ip.ts
123
server/lib/ip.ts
@@ -571,6 +571,129 @@ export function generateSubnetProxyTargets(
|
||||
return targets;
|
||||
}
|
||||
|
||||
export type SubnetProxyTargetV2 = {
|
||||
sourcePrefixes: string[]; // must be cidrs
|
||||
destPrefix: string; // must be a cidr
|
||||
disableIcmp?: boolean;
|
||||
rewriteTo?: string; // must be a cidr
|
||||
portRange?: {
|
||||
min: number;
|
||||
max: number;
|
||||
protocol: "tcp" | "udp";
|
||||
}[];
|
||||
};
|
||||
|
||||
export function generateSubnetProxyTargetV2(
|
||||
siteResource: SiteResource,
|
||||
clients: {
|
||||
clientId: number;
|
||||
pubKey: string | null;
|
||||
subnet: string | null;
|
||||
}[]
|
||||
): SubnetProxyTargetV2 | undefined {
|
||||
if (clients.length === 0) {
|
||||
logger.debug(
|
||||
`No clients have access to site resource ${siteResource.siteResourceId}, skipping target generation.`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
let target: SubnetProxyTargetV2 | null = null;
|
||||
|
||||
const portRange = [
|
||||
...parsePortRangeString(siteResource.tcpPortRangeString, "tcp"),
|
||||
...parsePortRangeString(siteResource.udpPortRangeString, "udp")
|
||||
];
|
||||
const disableIcmp = siteResource.disableIcmp ?? false;
|
||||
|
||||
if (siteResource.mode == "host") {
|
||||
let destination = siteResource.destination;
|
||||
// check if this is a valid ip
|
||||
const ipSchema = z.union([z.ipv4(), z.ipv6()]);
|
||||
if (ipSchema.safeParse(destination).success) {
|
||||
destination = `${destination}/32`;
|
||||
|
||||
target = {
|
||||
sourcePrefixes: [],
|
||||
destPrefix: destination,
|
||||
portRange,
|
||||
disableIcmp
|
||||
};
|
||||
}
|
||||
|
||||
if (siteResource.alias && siteResource.aliasAddress) {
|
||||
// also push a match for the alias address
|
||||
target = {
|
||||
sourcePrefixes: [],
|
||||
destPrefix: `${siteResource.aliasAddress}/32`,
|
||||
rewriteTo: destination,
|
||||
portRange,
|
||||
disableIcmp
|
||||
};
|
||||
}
|
||||
} else if (siteResource.mode == "cidr") {
|
||||
target = {
|
||||
sourcePrefixes: [],
|
||||
destPrefix: siteResource.destination,
|
||||
portRange,
|
||||
disableIcmp
|
||||
};
|
||||
}
|
||||
|
||||
if (!target) {
|
||||
return;
|
||||
}
|
||||
|
||||
for (const clientSite of clients) {
|
||||
if (!clientSite.subnet) {
|
||||
logger.debug(
|
||||
`Client ${clientSite.clientId} has no subnet, skipping for site resource ${siteResource.siteResourceId}.`
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
const clientPrefix = `${clientSite.subnet.split("/")[0]}/32`;
|
||||
|
||||
// add client prefix to source prefixes
|
||||
target.sourcePrefixes.push(clientPrefix);
|
||||
}
|
||||
|
||||
// print a nice representation of the targets
|
||||
// logger.debug(
|
||||
// `Generated subnet proxy targets for: ${JSON.stringify(targets, null, 2)}`
|
||||
// );
|
||||
|
||||
return target;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Converts a SubnetProxyTargetV2 to an array of SubnetProxyTarget (v1)
|
||||
* by expanding each source prefix into its own target entry.
|
||||
* @param targetV2 - The v2 target to convert
|
||||
* @returns Array of v1 SubnetProxyTarget objects
|
||||
*/
|
||||
export function convertSubnetProxyTargetsV2ToV1(
|
||||
targetsV2: SubnetProxyTargetV2[]
|
||||
): SubnetProxyTarget[] {
|
||||
return targetsV2.flatMap((targetV2) =>
|
||||
targetV2.sourcePrefixes.map((sourcePrefix) => ({
|
||||
sourcePrefix,
|
||||
destPrefix: targetV2.destPrefix,
|
||||
...(targetV2.disableIcmp !== undefined && {
|
||||
disableIcmp: targetV2.disableIcmp
|
||||
}),
|
||||
...(targetV2.rewriteTo !== undefined && {
|
||||
rewriteTo: targetV2.rewriteTo
|
||||
}),
|
||||
...(targetV2.portRange !== undefined && {
|
||||
portRange: targetV2.portRange
|
||||
})
|
||||
}))
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
// Custom schema for validating port range strings
|
||||
// Format: "80,443,8000-9000" or "*" for all ports, or empty string
|
||||
export const portRangeStringSchema = z
|
||||
|
||||
@@ -302,8 +302,8 @@ export const configSchema = z
|
||||
.optional()
|
||||
.default({
|
||||
block_size: 24,
|
||||
subnet_group: "100.90.128.0/24",
|
||||
utility_subnet_group: "100.96.128.0/24"
|
||||
subnet_group: "100.90.128.0/20",
|
||||
utility_subnet_group: "100.96.128.0/20"
|
||||
}),
|
||||
rate_limits: z
|
||||
.object({
|
||||
|
||||
@@ -32,7 +32,7 @@ import logger from "@server/logger";
|
||||
import {
|
||||
generateAliasConfig,
|
||||
generateRemoteSubnets,
|
||||
generateSubnetProxyTargets,
|
||||
generateSubnetProxyTargetV2,
|
||||
parseEndpoint,
|
||||
formatEndpoint
|
||||
} from "@server/lib/ip";
|
||||
@@ -660,19 +660,16 @@ async function handleSubnetProxyTargetUpdates(
|
||||
);
|
||||
|
||||
if (addedClients.length > 0) {
|
||||
const targetsToAdd = generateSubnetProxyTargets(
|
||||
const targetToAdd = generateSubnetProxyTargetV2(
|
||||
siteResource,
|
||||
addedClients
|
||||
);
|
||||
|
||||
if (targetsToAdd.length > 0) {
|
||||
logger.info(
|
||||
`Adding ${targetsToAdd.length} subnet proxy targets for siteResource ${siteResource.siteResourceId}`
|
||||
);
|
||||
if (targetToAdd) {
|
||||
proxyJobs.push(
|
||||
addSubnetProxyTargets(
|
||||
newt.newtId,
|
||||
targetsToAdd,
|
||||
[targetToAdd],
|
||||
newt.version
|
||||
)
|
||||
);
|
||||
@@ -700,19 +697,16 @@ async function handleSubnetProxyTargetUpdates(
|
||||
);
|
||||
|
||||
if (removedClients.length > 0) {
|
||||
const targetsToRemove = generateSubnetProxyTargets(
|
||||
const targetToRemove = generateSubnetProxyTargetV2(
|
||||
siteResource,
|
||||
removedClients
|
||||
);
|
||||
|
||||
if (targetsToRemove.length > 0) {
|
||||
logger.info(
|
||||
`Removing ${targetsToRemove.length} subnet proxy targets for siteResource ${siteResource.siteResourceId}`
|
||||
);
|
||||
if (targetToRemove) {
|
||||
proxyJobs.push(
|
||||
removeSubnetProxyTargets(
|
||||
newt.newtId,
|
||||
targetsToRemove,
|
||||
[targetToRemove],
|
||||
newt.version
|
||||
)
|
||||
);
|
||||
@@ -1169,7 +1163,7 @@ async function handleMessagesForClientResources(
|
||||
}
|
||||
|
||||
for (const resource of resources) {
|
||||
const targets = generateSubnetProxyTargets(resource, [
|
||||
const target = generateSubnetProxyTargetV2(resource, [
|
||||
{
|
||||
clientId: client.clientId,
|
||||
pubKey: client.pubKey,
|
||||
@@ -1177,11 +1171,11 @@ async function handleMessagesForClientResources(
|
||||
}
|
||||
]);
|
||||
|
||||
if (targets.length > 0) {
|
||||
if (target) {
|
||||
proxyJobs.push(
|
||||
addSubnetProxyTargets(
|
||||
newt.newtId,
|
||||
targets,
|
||||
[target],
|
||||
newt.version
|
||||
)
|
||||
);
|
||||
@@ -1246,7 +1240,7 @@ async function handleMessagesForClientResources(
|
||||
}
|
||||
|
||||
for (const resource of resources) {
|
||||
const targets = generateSubnetProxyTargets(resource, [
|
||||
const target = generateSubnetProxyTargetV2(resource, [
|
||||
{
|
||||
clientId: client.clientId,
|
||||
pubKey: client.pubKey,
|
||||
@@ -1254,11 +1248,11 @@ async function handleMessagesForClientResources(
|
||||
}
|
||||
]);
|
||||
|
||||
if (targets.length > 0) {
|
||||
if (target) {
|
||||
proxyJobs.push(
|
||||
removeSubnetProxyTargets(
|
||||
newt.newtId,
|
||||
targets,
|
||||
[target],
|
||||
newt.version
|
||||
)
|
||||
);
|
||||
|
||||
40
server/lib/sanitize.ts
Normal file
40
server/lib/sanitize.ts
Normal file
@@ -0,0 +1,40 @@
|
||||
/**
|
||||
* Sanitize a string field before inserting into a database TEXT column.
|
||||
*
|
||||
* Two passes are applied:
|
||||
*
|
||||
* 1. Lone UTF-16 surrogates – JavaScript strings can hold unpaired surrogates
|
||||
* (e.g. \uD800 without a following \uDC00-\uDFFF codepoint). These are
|
||||
* valid in JS but cannot be encoded as UTF-8, triggering
|
||||
* `report_invalid_encoding` in SQLite / Postgres. They are replaced with
|
||||
* the Unicode replacement character U+FFFD so the data is preserved as a
|
||||
* visible signal that something was malformed.
|
||||
*
|
||||
* 2. Null bytes and C0 control characters – SQLite stores TEXT as
|
||||
* null-terminated C strings, so \x00 in a value causes
|
||||
* `report_invalid_encoding`. Bots and scanners routinely inject null bytes
|
||||
* into URLs (e.g. `/path\u0000.jpg`). All C0 control characters in the
|
||||
* range \x00-\x1F are stripped except for the three that are legitimate in
|
||||
* text payloads: HT (\x09), LF (\x0A), and CR (\x0D). DEL (\x7F) is also
|
||||
* stripped.
|
||||
*/
|
||||
export function sanitizeString(value: string): string;
|
||||
export function sanitizeString(
|
||||
value: string | null | undefined
|
||||
): string | undefined;
|
||||
export function sanitizeString(
|
||||
value: string | null | undefined
|
||||
): string | undefined {
|
||||
if (value == null) return undefined;
|
||||
return (
|
||||
value
|
||||
// Replace lone high surrogates (not followed by a low surrogate)
|
||||
// and lone low surrogates (not preceded by a high surrogate).
|
||||
.replace(
|
||||
/[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(?<![\uD800-\uDBFF])[\uDC00-\uDFFF]/g,
|
||||
"\uFFFD"
|
||||
)
|
||||
// Strip null bytes, C0 control chars (except HT/LF/CR), and DEL.
|
||||
.replace(/[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]/g, "")
|
||||
);
|
||||
}
|
||||
@@ -14,10 +14,12 @@
|
||||
import { rateLimitService } from "#private/lib/rateLimit";
|
||||
import { cleanup as wsCleanup } from "#private/routers/ws";
|
||||
import { flushBandwidthToDb } from "@server/routers/newt/handleReceiveBandwidthMessage";
|
||||
import { flushConnectionLogToDb } from "#dynamic/routers/newt";
|
||||
import { flushSiteBandwidthToDb } from "@server/routers/gerbil/receiveBandwidth";
|
||||
|
||||
async function cleanup() {
|
||||
await flushBandwidthToDb();
|
||||
await flushConnectionLogToDb();
|
||||
await flushSiteBandwidthToDb();
|
||||
await rateLimitService.cleanup();
|
||||
await wsCleanup();
|
||||
|
||||
@@ -24,23 +24,31 @@ setInterval(() => {
|
||||
*/
|
||||
class AdaptiveCache {
|
||||
private useRedis(): boolean {
|
||||
return redisManager.isRedisEnabled() && redisManager.getHealthStatus().isHealthy;
|
||||
return (
|
||||
redisManager.isRedisEnabled() &&
|
||||
redisManager.getHealthStatus().isHealthy
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set a value in the cache
|
||||
* @param key - Cache key
|
||||
* @param value - Value to cache (will be JSON stringified for Redis)
|
||||
* @param ttl - Time to live in seconds (0 = no expiration)
|
||||
* @param ttl - Time to live in seconds (0 = no expiration; omit = 3600s for Redis)
|
||||
* @returns boolean indicating success
|
||||
*/
|
||||
async set(key: string, value: any, ttl?: number): Promise<boolean> {
|
||||
const effectiveTtl = ttl === 0 ? undefined : ttl;
|
||||
const redisTtl = ttl === 0 ? undefined : (ttl ?? 3600);
|
||||
|
||||
if (this.useRedis()) {
|
||||
try {
|
||||
const serialized = JSON.stringify(value);
|
||||
const success = await redisManager.set(key, serialized, effectiveTtl);
|
||||
const success = await redisManager.set(
|
||||
key,
|
||||
serialized,
|
||||
redisTtl
|
||||
);
|
||||
|
||||
if (success) {
|
||||
logger.debug(`Set key in Redis: ${key}`);
|
||||
@@ -48,7 +56,9 @@ class AdaptiveCache {
|
||||
}
|
||||
|
||||
// Redis failed, fall through to local cache
|
||||
logger.debug(`Redis set failed for key ${key}, falling back to local cache`);
|
||||
logger.debug(
|
||||
`Redis set failed for key ${key}, falling back to local cache`
|
||||
);
|
||||
} catch (error) {
|
||||
logger.error(`Redis set error for key ${key}:`, error);
|
||||
// Fall through to local cache
|
||||
@@ -120,9 +130,14 @@ class AdaptiveCache {
|
||||
}
|
||||
|
||||
// Some Redis deletes failed, fall through to local cache
|
||||
logger.debug(`Some Redis deletes failed, falling back to local cache`);
|
||||
logger.debug(
|
||||
`Some Redis deletes failed, falling back to local cache`
|
||||
);
|
||||
} catch (error) {
|
||||
logger.error(`Redis del error for keys ${keys.join(", ")}:`, error);
|
||||
logger.error(
|
||||
`Redis del error for keys ${keys.join(", ")}:`,
|
||||
error
|
||||
);
|
||||
// Fall through to local cache
|
||||
deletedCount = 0;
|
||||
}
|
||||
@@ -195,7 +210,9 @@ class AdaptiveCache {
|
||||
*/
|
||||
async flushAll(): Promise<void> {
|
||||
if (this.useRedis()) {
|
||||
logger.warn("Adaptive cache flushAll called - Redis flush not implemented, only local cache will be flushed");
|
||||
logger.warn(
|
||||
"Adaptive cache flushAll called - Redis flush not implemented, only local cache will be flushed"
|
||||
);
|
||||
}
|
||||
|
||||
localCache.flushAll();
|
||||
@@ -239,7 +256,9 @@ class AdaptiveCache {
|
||||
getTtl(key: string): number {
|
||||
// Note: This only works for local cache, Redis TTL is not supported
|
||||
if (this.useRedis()) {
|
||||
logger.warn(`getTtl called for key ${key} but Redis TTL lookup is not implemented`);
|
||||
logger.warn(
|
||||
`getTtl called for key ${key} but Redis TTL lookup is not implemented`
|
||||
);
|
||||
}
|
||||
|
||||
const ttl = localCache.getTtl(key);
|
||||
@@ -255,7 +274,9 @@ class AdaptiveCache {
|
||||
*/
|
||||
keys(): string[] {
|
||||
if (this.useRedis()) {
|
||||
logger.warn("keys() called but Redis keys are not included, only local cache keys returned");
|
||||
logger.warn(
|
||||
"keys() called but Redis keys are not included, only local cache keys returned"
|
||||
);
|
||||
}
|
||||
return localCache.keys();
|
||||
}
|
||||
|
||||
@@ -57,7 +57,10 @@ export const privateConfigSchema = z.object({
|
||||
.object({
|
||||
host: z.string(),
|
||||
port: portSchema,
|
||||
password: z.string().optional(),
|
||||
password: z
|
||||
.string()
|
||||
.optional()
|
||||
.transform(getEnvOrYaml("REDIS_PASSWORD")),
|
||||
db: z.int().nonnegative().optional().default(0),
|
||||
replicas: z
|
||||
.array(
|
||||
|
||||
@@ -15,6 +15,7 @@ import { verifySessionRemoteExitNodeMiddleware } from "#private/middlewares/veri
|
||||
import { Router } from "express";
|
||||
import {
|
||||
db,
|
||||
logsDb,
|
||||
exitNodes,
|
||||
Resource,
|
||||
ResourcePassword,
|
||||
@@ -81,6 +82,7 @@ import { verifyResourceAccessToken } from "@server/auth/verifyResourceAccessToke
|
||||
import semver from "semver";
|
||||
import { maxmindAsnLookup } from "@server/db/maxmindAsn";
|
||||
import { checkOrgAccessPolicy } from "@server/lib/checkOrgAccessPolicy";
|
||||
import { sanitizeString } from "@server/lib/sanitize";
|
||||
|
||||
// Zod schemas for request validation
|
||||
const getResourceByDomainParamsSchema = z.strictObject({
|
||||
@@ -1859,24 +1861,24 @@ hybridRouter.post(
|
||||
})
|
||||
.map((logEntry) => ({
|
||||
timestamp: logEntry.timestamp,
|
||||
orgId: logEntry.orgId,
|
||||
actorType: logEntry.actorType,
|
||||
actor: logEntry.actor,
|
||||
actorId: logEntry.actorId,
|
||||
metadata: logEntry.metadata,
|
||||
orgId: sanitizeString(logEntry.orgId),
|
||||
actorType: sanitizeString(logEntry.actorType),
|
||||
actor: sanitizeString(logEntry.actor),
|
||||
actorId: sanitizeString(logEntry.actorId),
|
||||
metadata: sanitizeString(logEntry.metadata),
|
||||
action: logEntry.action,
|
||||
resourceId: logEntry.resourceId,
|
||||
reason: logEntry.reason,
|
||||
location: logEntry.location,
|
||||
location: sanitizeString(logEntry.location),
|
||||
// userAgent: data.userAgent, // TODO: add this
|
||||
// headers: data.body.headers,
|
||||
// query: data.body.query,
|
||||
originalRequestURL: logEntry.originalRequestURL,
|
||||
scheme: logEntry.scheme,
|
||||
host: logEntry.host,
|
||||
path: logEntry.path,
|
||||
method: logEntry.method,
|
||||
ip: logEntry.ip,
|
||||
originalRequestURL: sanitizeString(logEntry.originalRequestURL) ?? "",
|
||||
scheme: sanitizeString(logEntry.scheme) ?? "",
|
||||
host: sanitizeString(logEntry.host) ?? "",
|
||||
path: sanitizeString(logEntry.path) ?? "",
|
||||
method: sanitizeString(logEntry.method) ?? "",
|
||||
ip: sanitizeString(logEntry.ip),
|
||||
tls: logEntry.tls
|
||||
}));
|
||||
|
||||
@@ -1884,7 +1886,7 @@ hybridRouter.post(
|
||||
const batchSize = 100;
|
||||
for (let i = 0; i < logEntries.length; i += batchSize) {
|
||||
const batch = logEntries.slice(i, i + batchSize);
|
||||
await db.insert(requestAuditLog).values(batch);
|
||||
await logsDb.insert(requestAuditLog).values(batch);
|
||||
}
|
||||
|
||||
return response(res, {
|
||||
|
||||
324
server/private/routers/newt/handleConnectionLogMessage.ts
Normal file
324
server/private/routers/newt/handleConnectionLogMessage.ts
Normal file
@@ -0,0 +1,324 @@
|
||||
import { db, logsDb } from "@server/db";
|
||||
import { MessageHandler } from "@server/routers/ws";
|
||||
import { connectionAuditLog, sites, Newt } from "@server/db";
|
||||
import { and, eq, lt } from "drizzle-orm";
|
||||
import logger from "@server/logger";
|
||||
import { inflate } from "zlib";
|
||||
import { promisify } from "util";
|
||||
import { calculateCutoffTimestamp } from "@server/lib/cleanupLogs";
|
||||
|
||||
const zlibInflate = promisify(inflate);
|
||||
|
||||
// Retry configuration for deadlock handling
|
||||
const MAX_RETRIES = 3;
|
||||
const BASE_DELAY_MS = 50;
|
||||
|
||||
// How often to flush accumulated connection log data to the database
|
||||
const FLUSH_INTERVAL_MS = 30_000; // 30 seconds
|
||||
|
||||
// Maximum number of records to buffer before forcing a flush
|
||||
const MAX_BUFFERED_RECORDS = 500;
|
||||
|
||||
// Maximum number of records to insert in a single batch
|
||||
const INSERT_BATCH_SIZE = 100;
|
||||
|
||||
interface ConnectionSessionData {
|
||||
sessionId: string;
|
||||
resourceId: number;
|
||||
sourceAddr: string;
|
||||
destAddr: string;
|
||||
protocol: string;
|
||||
startedAt: string; // ISO 8601 timestamp
|
||||
endedAt?: string; // ISO 8601 timestamp
|
||||
bytesTx?: number;
|
||||
bytesRx?: number;
|
||||
}
|
||||
|
||||
interface ConnectionLogRecord {
|
||||
sessionId: string;
|
||||
siteResourceId: number;
|
||||
orgId: string;
|
||||
siteId: number;
|
||||
sourceAddr: string;
|
||||
destAddr: string;
|
||||
protocol: string;
|
||||
startedAt: number; // epoch seconds
|
||||
endedAt: number | null;
|
||||
bytesTx: number | null;
|
||||
bytesRx: number | null;
|
||||
}
|
||||
|
||||
// In-memory buffer of records waiting to be flushed
|
||||
let buffer: ConnectionLogRecord[] = [];
|
||||
|
||||
/**
|
||||
* Check if an error is a deadlock error
|
||||
*/
|
||||
function isDeadlockError(error: any): boolean {
|
||||
return (
|
||||
error?.code === "40P01" ||
|
||||
error?.cause?.code === "40P01" ||
|
||||
(error?.message && error.message.includes("deadlock"))
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a function with retry logic for deadlock handling
|
||||
*/
|
||||
async function withDeadlockRetry<T>(
|
||||
operation: () => Promise<T>,
|
||||
context: string
|
||||
): Promise<T> {
|
||||
let attempt = 0;
|
||||
while (true) {
|
||||
try {
|
||||
return await operation();
|
||||
} catch (error: any) {
|
||||
if (isDeadlockError(error) && attempt < MAX_RETRIES) {
|
||||
attempt++;
|
||||
const baseDelay = Math.pow(2, attempt - 1) * BASE_DELAY_MS;
|
||||
const jitter = Math.random() * baseDelay;
|
||||
const delay = baseDelay + jitter;
|
||||
logger.warn(
|
||||
`Deadlock detected in ${context}, retrying attempt ${attempt}/${MAX_RETRIES} after ${delay.toFixed(0)}ms`
|
||||
);
|
||||
await new Promise((resolve) => setTimeout(resolve, delay));
|
||||
continue;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Decompress a base64-encoded zlib-compressed string into parsed JSON.
|
||||
*/
|
||||
async function decompressConnectionLog(
|
||||
compressed: string
|
||||
): Promise<ConnectionSessionData[]> {
|
||||
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 connection log data is not an array");
|
||||
}
|
||||
|
||||
return parsed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert an ISO 8601 timestamp string to epoch seconds.
|
||||
* Returns null if the input is falsy.
|
||||
*/
|
||||
function toEpochSeconds(isoString: string | undefined | null): number | null {
|
||||
if (!isoString) {
|
||||
return null;
|
||||
}
|
||||
const ms = new Date(isoString).getTime();
|
||||
if (isNaN(ms)) {
|
||||
return null;
|
||||
}
|
||||
return Math.floor(ms / 1000);
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush all buffered connection log records to the database.
|
||||
*
|
||||
* Swaps out the buffer before writing so that any records added during the
|
||||
* flush are captured in the new buffer rather than being lost. Entries that
|
||||
* fail to write are re-queued back into the buffer so they will be retried
|
||||
* on the next flush.
|
||||
*
|
||||
* This function is exported so that the application's graceful-shutdown
|
||||
* cleanup handler can call it before the process exits.
|
||||
*/
|
||||
export async function flushConnectionLogToDb(): Promise<void> {
|
||||
if (buffer.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Atomically swap out the buffer so new data keeps flowing in
|
||||
const snapshot = buffer;
|
||||
buffer = [];
|
||||
|
||||
logger.debug(
|
||||
`Flushing ${snapshot.length} connection log record(s) to the database`
|
||||
);
|
||||
|
||||
// Insert in batches to avoid overly large SQL statements
|
||||
for (let i = 0; i < snapshot.length; i += INSERT_BATCH_SIZE) {
|
||||
const batch = snapshot.slice(i, i + INSERT_BATCH_SIZE);
|
||||
|
||||
try {
|
||||
await withDeadlockRetry(async () => {
|
||||
await logsDb.insert(connectionAuditLog).values(batch);
|
||||
}, `flush connection log batch (${batch.length} records)`);
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
`Failed to flush connection log batch of ${batch.length} records:`,
|
||||
error
|
||||
);
|
||||
|
||||
// Re-queue the failed batch so it is retried on the next flush
|
||||
buffer = [...batch, ...buffer];
|
||||
|
||||
// Cap buffer to prevent unbounded growth if DB is unreachable
|
||||
if (buffer.length > MAX_BUFFERED_RECORDS * 5) {
|
||||
const dropped = buffer.length - MAX_BUFFERED_RECORDS * 5;
|
||||
buffer = buffer.slice(0, MAX_BUFFERED_RECORDS * 5);
|
||||
logger.warn(
|
||||
`Connection log buffer overflow, dropped ${dropped} oldest records`
|
||||
);
|
||||
}
|
||||
|
||||
// Stop trying further batches from this snapshot — they'll be
|
||||
// picked up by the next flush via the re-queued records above
|
||||
const remaining = snapshot.slice(i + INSERT_BATCH_SIZE);
|
||||
if (remaining.length > 0) {
|
||||
buffer = [...remaining, ...buffer];
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const flushTimer = setInterval(async () => {
|
||||
try {
|
||||
await flushConnectionLogToDb();
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
"Unexpected error during periodic connection log flush:",
|
||||
error
|
||||
);
|
||||
}
|
||||
}, FLUSH_INTERVAL_MS);
|
||||
|
||||
// Calling unref() means this timer will not keep the Node.js event loop alive
|
||||
// on its own — the process can still exit normally when there is no other work
|
||||
// left. The graceful-shutdown path will call flushConnectionLogToDb() explicitly
|
||||
// before process.exit(), so no data is lost.
|
||||
flushTimer.unref();
|
||||
|
||||
export async function cleanUpOldLogs(orgId: string, retentionDays: number) {
|
||||
const cutoffTimestamp = calculateCutoffTimestamp(retentionDays);
|
||||
|
||||
try {
|
||||
await logsDb
|
||||
.delete(connectionAuditLog)
|
||||
.where(
|
||||
and(
|
||||
lt(connectionAuditLog.startedAt, cutoffTimestamp),
|
||||
eq(connectionAuditLog.orgId, orgId)
|
||||
)
|
||||
);
|
||||
|
||||
// logger.debug(
|
||||
// `Cleaned up connection audit logs older than ${retentionDays} days`
|
||||
// );
|
||||
} catch (error) {
|
||||
logger.error("Error cleaning up old connection audit logs:", error);
|
||||
}
|
||||
}
|
||||
|
||||
export const handleConnectionLogMessage: MessageHandler = async (context) => {
|
||||
const { message, client } = context;
|
||||
const newt = client as Newt;
|
||||
|
||||
if (!newt) {
|
||||
logger.warn("Connection log received but no newt client in context");
|
||||
return;
|
||||
}
|
||||
|
||||
if (!newt.siteId) {
|
||||
logger.warn("Connection log received but newt has no siteId");
|
||||
return;
|
||||
}
|
||||
|
||||
if (!message.data?.compressed) {
|
||||
logger.warn("Connection log message missing compressed data");
|
||||
return;
|
||||
}
|
||||
|
||||
// Look up the org for this site
|
||||
const [site] = await db
|
||||
.select({ orgId: sites.orgId })
|
||||
.from(sites)
|
||||
.where(eq(sites.siteId, newt.siteId));
|
||||
|
||||
if (!site) {
|
||||
logger.warn(
|
||||
`Connection log received but site ${newt.siteId} not found in database`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const orgId = site.orgId;
|
||||
|
||||
let sessions: ConnectionSessionData[];
|
||||
try {
|
||||
sessions = await decompressConnectionLog(message.data.compressed);
|
||||
} catch (error) {
|
||||
logger.error("Failed to decompress connection log data:", error);
|
||||
return;
|
||||
}
|
||||
|
||||
if (sessions.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Convert to DB records and add to the buffer
|
||||
for (const session of sessions) {
|
||||
// Validate required fields
|
||||
if (
|
||||
!session.sessionId ||
|
||||
!session.resourceId ||
|
||||
!session.sourceAddr ||
|
||||
!session.destAddr ||
|
||||
!session.protocol
|
||||
) {
|
||||
logger.debug(
|
||||
`Skipping connection log session with missing required fields: ${JSON.stringify(session)}`
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
const startedAt = toEpochSeconds(session.startedAt);
|
||||
if (startedAt === null) {
|
||||
logger.debug(
|
||||
`Skipping connection log session with invalid startedAt: ${session.startedAt}`
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
buffer.push({
|
||||
sessionId: session.sessionId,
|
||||
siteResourceId: session.resourceId,
|
||||
orgId,
|
||||
siteId: newt.siteId,
|
||||
sourceAddr: session.sourceAddr,
|
||||
destAddr: session.destAddr,
|
||||
protocol: session.protocol,
|
||||
startedAt,
|
||||
endedAt: toEpochSeconds(session.endedAt),
|
||||
bytesTx: session.bytesTx ?? null,
|
||||
bytesRx: session.bytesRx ?? null
|
||||
});
|
||||
}
|
||||
|
||||
logger.debug(
|
||||
`Buffered ${sessions.length} connection log session(s) from newt ${newt.newtId} (site ${newt.siteId})`
|
||||
);
|
||||
|
||||
// If the buffer has grown large enough, trigger an immediate flush
|
||||
if (buffer.length >= MAX_BUFFERED_RECORDS) {
|
||||
// Fire and forget — errors are handled inside flushConnectionLogToDb
|
||||
flushConnectionLogToDb().catch((error) => {
|
||||
logger.error(
|
||||
"Unexpected error during size-triggered connection log flush:",
|
||||
error
|
||||
);
|
||||
});
|
||||
}
|
||||
};
|
||||
1
server/private/routers/newt/index.ts
Normal file
1
server/private/routers/newt/index.ts
Normal file
@@ -0,0 +1 @@
|
||||
export * from "./handleConnectionLogMessage";
|
||||
@@ -38,7 +38,7 @@ export const startRemoteExitNodeOfflineChecker = (): void => {
|
||||
);
|
||||
|
||||
// Find clients that haven't pinged in the last 2 minutes and mark them as offline
|
||||
const newlyOfflineNodes = await db
|
||||
const offlineNodes = await db
|
||||
.update(exitNodes)
|
||||
.set({ online: false })
|
||||
.where(
|
||||
@@ -53,32 +53,15 @@ export const startRemoteExitNodeOfflineChecker = (): void => {
|
||||
)
|
||||
.returning();
|
||||
|
||||
// Update the sites to offline if they have not pinged either
|
||||
const exitNodeIds = newlyOfflineNodes.map(
|
||||
(node) => node.exitNodeId
|
||||
if (offlineNodes.length > 0) {
|
||||
logger.info(
|
||||
`checkRemoteExitNodeOffline: Marked ${offlineNodes.length} remoteExitNode client(s) offline due to inactivity`
|
||||
);
|
||||
|
||||
const sitesOnNode = await db
|
||||
.select()
|
||||
.from(sites)
|
||||
.where(
|
||||
and(
|
||||
eq(sites.online, true),
|
||||
inArray(sites.exitNodeId, exitNodeIds)
|
||||
)
|
||||
for (const offlineClient of offlineNodes) {
|
||||
logger.debug(
|
||||
`checkRemoteExitNodeOffline: Client ${offlineClient.exitNodeId} marked offline (lastPing: ${offlineClient.lastPing})`
|
||||
);
|
||||
|
||||
// loop through the sites and process their lastBandwidthUpdate as an iso string and if its more than 1 minute old then mark the site offline
|
||||
for (const site of sitesOnNode) {
|
||||
if (!site.lastBandwidthUpdate) {
|
||||
continue;
|
||||
}
|
||||
const lastBandwidthUpdate = new Date(site.lastBandwidthUpdate);
|
||||
if (Date.now() - lastBandwidthUpdate.getTime() > 60 * 1000) {
|
||||
await db
|
||||
.update(sites)
|
||||
.set({ online: false })
|
||||
.where(eq(sites.siteId, site.siteId));
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
|
||||
@@ -18,10 +18,12 @@ import {
|
||||
} from "#private/routers/remoteExitNode";
|
||||
import { MessageHandler } from "@server/routers/ws";
|
||||
import { build } from "@server/build";
|
||||
import { handleConnectionLogMessage } from "#dynamic/routers/newt";
|
||||
|
||||
export const messageHandlers: Record<string, MessageHandler> = {
|
||||
"remoteExitNode/register": handleRemoteExitNodeRegisterMessage,
|
||||
"remoteExitNode/ping": handleRemoteExitNodePingMessage
|
||||
"remoteExitNode/ping": handleRemoteExitNodePingMessage,
|
||||
"newt/access-log": handleConnectionLogMessage,
|
||||
};
|
||||
|
||||
if (build != "saas") {
|
||||
|
||||
@@ -5,25 +5,7 @@ import cache from "#dynamic/lib/cache";
|
||||
import { calculateCutoffTimestamp } from "@server/lib/cleanupLogs";
|
||||
import { stripPortFromHost } from "@server/lib/ip";
|
||||
|
||||
/**
|
||||
* Sanitize a string field by replacing lone UTF-16 surrogates (which cannot
|
||||
* be encoded as valid UTF-8) with the Unicode replacement character, and
|
||||
* stripping ASCII control characters that are invalid in most text columns.
|
||||
*/
|
||||
function sanitizeString(value: string | undefined | null): string | undefined {
|
||||
if (value == null) return undefined;
|
||||
return (
|
||||
value
|
||||
// Replace lone high surrogates (not followed by a low surrogate)
|
||||
// and lone low surrogates (not preceded by a high surrogate)
|
||||
.replace(
|
||||
/[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(?<![\uD800-\uDBFF])[\uDC00-\uDFFF]/g,
|
||||
"\uFFFD"
|
||||
)
|
||||
// Strip C0 control characters except HT (\x09), LF (\x0A), CR (\x0D)
|
||||
.replace(/[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]/g, "")
|
||||
);
|
||||
}
|
||||
import { sanitizeString } from "@server/lib/sanitize";
|
||||
|
||||
/**
|
||||
|
||||
|
||||
@@ -70,7 +70,7 @@ async function getLatestOlmVersion(): Promise<string | null> {
|
||||
tags = tags.filter((version) => !version.name.includes("rc"));
|
||||
const latestVersion = tags[0].name;
|
||||
|
||||
olmVersionCache.set("latestOlmVersion", latestVersion);
|
||||
olmVersionCache.set("latestOlmVersion", latestVersion, 3600);
|
||||
|
||||
return latestVersion;
|
||||
} catch (error: any) {
|
||||
|
||||
@@ -71,7 +71,7 @@ async function getLatestOlmVersion(): Promise<string | null> {
|
||||
tags = tags.filter((version) => !version.name.includes("rc"));
|
||||
const latestVersion = tags[0].name;
|
||||
|
||||
olmVersionCache.set("latestOlmVersion", latestVersion);
|
||||
olmVersionCache.set("latestOlmVersion", latestVersion, 3600);
|
||||
|
||||
return latestVersion;
|
||||
} catch (error: any) {
|
||||
|
||||
@@ -1,15 +1,54 @@
|
||||
import { sendToClient } from "#dynamic/routers/ws";
|
||||
import { db, olms, Transaction } from "@server/db";
|
||||
import { db, newts, olms } from "@server/db";
|
||||
import {
|
||||
Alias,
|
||||
convertSubnetProxyTargetsV2ToV1,
|
||||
SubnetProxyTarget,
|
||||
SubnetProxyTargetV2
|
||||
} from "@server/lib/ip";
|
||||
import { canCompress } from "@server/lib/clientVersionChecks";
|
||||
import { Alias, SubnetProxyTarget } from "@server/lib/ip";
|
||||
import logger from "@server/logger";
|
||||
import { eq } from "drizzle-orm";
|
||||
import semver from "semver";
|
||||
|
||||
const NEWT_V2_TARGETS_VERSION = ">=1.10.3";
|
||||
|
||||
export async function convertTargetsIfNessicary(
|
||||
newtId: string,
|
||||
targets: SubnetProxyTarget[] | SubnetProxyTargetV2[]
|
||||
) {
|
||||
// get the newt
|
||||
const [newt] = await db
|
||||
.select()
|
||||
.from(newts)
|
||||
.where(eq(newts.newtId, newtId));
|
||||
if (!newt) {
|
||||
throw new Error(`No newt found for id: ${newtId}`);
|
||||
}
|
||||
|
||||
// check the semver
|
||||
if (
|
||||
newt.version &&
|
||||
!semver.satisfies(newt.version, NEWT_V2_TARGETS_VERSION)
|
||||
) {
|
||||
logger.debug(
|
||||
`addTargets Newt version ${newt.version} does not support targets v2 falling back`
|
||||
);
|
||||
targets = convertSubnetProxyTargetsV2ToV1(
|
||||
targets as SubnetProxyTargetV2[]
|
||||
);
|
||||
}
|
||||
|
||||
return targets;
|
||||
}
|
||||
|
||||
export async function addTargets(
|
||||
newtId: string,
|
||||
targets: SubnetProxyTarget[],
|
||||
targets: SubnetProxyTarget[] | SubnetProxyTargetV2[],
|
||||
version?: string | null
|
||||
) {
|
||||
targets = await convertTargetsIfNessicary(newtId, targets);
|
||||
|
||||
await sendToClient(
|
||||
newtId,
|
||||
{
|
||||
@@ -22,9 +61,11 @@ export async function addTargets(
|
||||
|
||||
export async function removeTargets(
|
||||
newtId: string,
|
||||
targets: SubnetProxyTarget[],
|
||||
targets: SubnetProxyTarget[] | SubnetProxyTargetV2[],
|
||||
version?: string | null
|
||||
) {
|
||||
targets = await convertTargetsIfNessicary(newtId, targets);
|
||||
|
||||
await sendToClient(
|
||||
newtId,
|
||||
{
|
||||
@@ -38,11 +79,39 @@ export async function removeTargets(
|
||||
export async function updateTargets(
|
||||
newtId: string,
|
||||
targets: {
|
||||
oldTargets: SubnetProxyTarget[];
|
||||
newTargets: SubnetProxyTarget[];
|
||||
oldTargets: SubnetProxyTarget[] | SubnetProxyTargetV2[];
|
||||
newTargets: SubnetProxyTarget[] | SubnetProxyTargetV2[];
|
||||
},
|
||||
version?: string | null
|
||||
) {
|
||||
// get the newt
|
||||
const [newt] = await db
|
||||
.select()
|
||||
.from(newts)
|
||||
.where(eq(newts.newtId, newtId));
|
||||
if (!newt) {
|
||||
logger.error(`addTargetsL No newt found for id: ${newtId}`);
|
||||
return;
|
||||
}
|
||||
|
||||
// check the semver
|
||||
if (
|
||||
newt.version &&
|
||||
!semver.satisfies(newt.version, NEWT_V2_TARGETS_VERSION)
|
||||
) {
|
||||
logger.debug(
|
||||
`addTargets Newt version ${newt.version} does not support targets v2 falling back`
|
||||
);
|
||||
targets = {
|
||||
oldTargets: convertSubnetProxyTargetsV2ToV1(
|
||||
targets.oldTargets as SubnetProxyTargetV2[]
|
||||
),
|
||||
newTargets: convertSubnetProxyTargetsV2ToV1(
|
||||
targets.newTargets as SubnetProxyTargetV2[]
|
||||
)
|
||||
};
|
||||
}
|
||||
|
||||
await sendToClient(
|
||||
newtId,
|
||||
{
|
||||
|
||||
@@ -16,8 +16,8 @@ import { eq, and } from "drizzle-orm";
|
||||
import config from "@server/lib/config";
|
||||
import {
|
||||
formatEndpoint,
|
||||
generateSubnetProxyTargets,
|
||||
SubnetProxyTarget
|
||||
generateSubnetProxyTargetV2,
|
||||
SubnetProxyTargetV2
|
||||
} from "@server/lib/ip";
|
||||
|
||||
export async function buildClientConfigurationForNewtClient(
|
||||
@@ -143,7 +143,7 @@ export async function buildClientConfigurationForNewtClient(
|
||||
.from(siteResources)
|
||||
.where(eq(siteResources.siteId, siteId));
|
||||
|
||||
const targetsToSend: SubnetProxyTarget[] = [];
|
||||
const targetsToSend: SubnetProxyTargetV2[] = [];
|
||||
|
||||
for (const resource of allSiteResources) {
|
||||
// Get clients associated with this specific resource
|
||||
@@ -168,12 +168,14 @@ export async function buildClientConfigurationForNewtClient(
|
||||
)
|
||||
);
|
||||
|
||||
const resourceTargets = generateSubnetProxyTargets(
|
||||
const resourceTarget = generateSubnetProxyTargetV2(
|
||||
resource,
|
||||
resourceClients
|
||||
);
|
||||
|
||||
targetsToSend.push(...resourceTargets);
|
||||
if (resourceTarget) {
|
||||
targetsToSend.push(resourceTarget);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
|
||||
13
server/routers/newt/handleConnectionLogMessage.ts
Normal file
13
server/routers/newt/handleConnectionLogMessage.ts
Normal file
@@ -0,0 +1,13 @@
|
||||
import { MessageHandler } from "@server/routers/ws";
|
||||
|
||||
export async function flushConnectionLogToDb(): Promise<void> {
|
||||
return;
|
||||
}
|
||||
|
||||
export async function cleanUpOldLogs(orgId: string, retentionDays: number) {
|
||||
return;
|
||||
}
|
||||
|
||||
export const handleConnectionLogMessage: MessageHandler = async (context) => {
|
||||
return;
|
||||
};
|
||||
@@ -6,6 +6,7 @@ import { db, ExitNode, exitNodes, Newt, sites } from "@server/db";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { sendToExitNode } from "#dynamic/lib/exitNodes";
|
||||
import { buildClientConfigurationForNewtClient } from "./buildConfiguration";
|
||||
import { convertTargetsIfNessicary } from "../client/targets";
|
||||
import { canCompress } from "@server/lib/clientVersionChecks";
|
||||
|
||||
const inputSchema = z.object({
|
||||
@@ -127,13 +128,15 @@ export const handleGetConfigMessage: MessageHandler = async (context) => {
|
||||
exitNode
|
||||
);
|
||||
|
||||
const targetsToSend = await convertTargetsIfNessicary(newt.newtId, targets);
|
||||
|
||||
return {
|
||||
message: {
|
||||
type: "newt/wg/receive-config",
|
||||
data: {
|
||||
ipAddress: site.address,
|
||||
peers,
|
||||
targets
|
||||
targets: targetsToSend
|
||||
}
|
||||
},
|
||||
options: {
|
||||
|
||||
@@ -6,7 +6,9 @@ import logger from "@server/logger";
|
||||
/**
|
||||
* Handles disconnecting messages from sites to show disconnected in the ui
|
||||
*/
|
||||
export const handleNewtDisconnectingMessage: MessageHandler = async (context) => {
|
||||
export const handleNewtDisconnectingMessage: MessageHandler = async (
|
||||
context
|
||||
) => {
|
||||
const { message, client: c, sendToClient } = context;
|
||||
const newt = c as Newt;
|
||||
|
||||
@@ -27,7 +29,7 @@ export const handleNewtDisconnectingMessage: MessageHandler = async (context) =>
|
||||
.set({
|
||||
online: false
|
||||
})
|
||||
.where(eq(sites.siteId, sites.siteId));
|
||||
.where(eq(sites.siteId, newt.siteId));
|
||||
} catch (error) {
|
||||
logger.error("Error handling disconnecting message", { error });
|
||||
}
|
||||
|
||||
@@ -8,3 +8,4 @@ export * from "./handleNewtPingRequestMessage";
|
||||
export * from "./handleApplyBlueprintMessage";
|
||||
export * from "./handleNewtPingMessage";
|
||||
export * from "./handleNewtDisconnectingMessage";
|
||||
export * from "./handleConnectionLogMessage";
|
||||
|
||||
@@ -55,7 +55,7 @@ async function getLatestNewtVersion(): Promise<string | null> {
|
||||
tags = tags.filter((version) => !version.name.includes("rc"));
|
||||
const latestVersion = tags[0].name;
|
||||
|
||||
await cache.set("latestNewtVersion", latestVersion);
|
||||
await cache.set("latestNewtVersion", latestVersion, 3600);
|
||||
|
||||
return latestVersion;
|
||||
} catch (error: any) {
|
||||
@@ -180,7 +180,7 @@ registry.registerPath({
|
||||
method: "get",
|
||||
path: "/org/{orgId}/sites",
|
||||
description: "List all sites in an organization",
|
||||
tags: [OpenAPITags.Site],
|
||||
tags: [OpenAPITags.Org, OpenAPITags.Site],
|
||||
request: {
|
||||
params: listSitesParamsSchema,
|
||||
query: listSitesSchema
|
||||
|
||||
@@ -88,7 +88,7 @@ const createSiteResourceSchema = z
|
||||
},
|
||||
{
|
||||
message:
|
||||
"Destination must be a valid IP address or valid domain AND alias is required"
|
||||
"Destination must be a valid IPV4 address or valid domain AND alias is required"
|
||||
}
|
||||
)
|
||||
.refine(
|
||||
|
||||
@@ -24,7 +24,7 @@ import { updatePeerData, updateTargets } from "@server/routers/client/targets";
|
||||
import {
|
||||
generateAliasConfig,
|
||||
generateRemoteSubnets,
|
||||
generateSubnetProxyTargets,
|
||||
generateSubnetProxyTargetV2,
|
||||
isIpInCidr,
|
||||
portRangeStringSchema
|
||||
} from "@server/lib/ip";
|
||||
@@ -608,18 +608,18 @@ export async function handleMessagingForUpdatedSiteResource(
|
||||
|
||||
// Only update targets on newt if destination changed
|
||||
if (destinationChanged || portRangesChanged) {
|
||||
const oldTargets = generateSubnetProxyTargets(
|
||||
const oldTarget = generateSubnetProxyTargetV2(
|
||||
existingSiteResource,
|
||||
mergedAllClients
|
||||
);
|
||||
const newTargets = generateSubnetProxyTargets(
|
||||
const newTarget = generateSubnetProxyTargetV2(
|
||||
updatedSiteResource,
|
||||
mergedAllClients
|
||||
);
|
||||
|
||||
await updateTargets(newt.newtId, {
|
||||
oldTargets: oldTargets,
|
||||
newTargets: newTargets
|
||||
oldTargets: oldTarget ? [oldTarget] : [],
|
||||
newTargets: newTarget ? [newTarget] : []
|
||||
}, newt.version);
|
||||
}
|
||||
|
||||
|
||||
@@ -201,7 +201,7 @@ export async function inviteUser(
|
||||
);
|
||||
}
|
||||
|
||||
await cache.set(email, attempts + 1);
|
||||
await cache.set("regenerateInvite:" + email, attempts + 1, 3600);
|
||||
|
||||
const inviteId = existingInvite[0].inviteId; // Retrieve the original inviteId
|
||||
const token = generateRandomString(
|
||||
|
||||
@@ -29,6 +29,7 @@ import { usePathname, useRouter } from "next/navigation";
|
||||
import { useMemo, useState } from "react";
|
||||
import { useUserContext } from "@app/hooks/useUserContext";
|
||||
import { useTranslations } from "next-intl";
|
||||
import { build } from "@server/build";
|
||||
|
||||
interface OrgSelectorProps {
|
||||
orgId?: string;
|
||||
@@ -50,6 +51,11 @@ export function OrgSelector({
|
||||
|
||||
const selectedOrg = orgs?.find((org) => org.orgId === orgId);
|
||||
|
||||
let canCreateOrg = !env.flags.disableUserCreateOrg || user.serverAdmin;
|
||||
if (build === "saas" && user.type !== "internal") {
|
||||
canCreateOrg = false;
|
||||
}
|
||||
|
||||
const sortedOrgs = useMemo(() => {
|
||||
if (!orgs?.length) return orgs ?? [];
|
||||
return [...orgs].sort((a, b) => {
|
||||
@@ -161,7 +167,7 @@ export function OrgSelector({
|
||||
</CommandGroup>
|
||||
</CommandList>
|
||||
</Command>
|
||||
{(!env.flags.disableUserCreateOrg || user.serverAdmin) && (
|
||||
{canCreateOrg && (
|
||||
<div className="p-2 border-t border-border">
|
||||
<Button
|
||||
variant="ghost"
|
||||
|
||||
Reference in New Issue
Block a user