mirror of
https://github.com/fosrl/pangolin.git
synced 2025-08-21 09:48:39 +02:00
Move exit node comms to new file
This commit is contained in:
parent
6600de7320
commit
04ecf41c5a
5 changed files with 295 additions and 197 deletions
|
@ -1,6 +1,3 @@
|
|||
import next from "next";
|
||||
import express from "express";
|
||||
import { parse } from "url";
|
||||
import logger from "@server/logger";
|
||||
import config from "@server/lib/config";
|
||||
import { createWebSocketClient } from "./routers/ws/client";
|
||||
|
@ -9,6 +6,7 @@ import { db, exitNodes } from "./db";
|
|||
import { TraefikConfigManager } from "./lib/remoteTraefikConfig";
|
||||
import { tokenManager } from "./lib/tokenManager";
|
||||
import { APP_VERSION } from "./lib/consts";
|
||||
import axios from "axios";
|
||||
|
||||
export async function createHybridClientServer() {
|
||||
logger.info("Starting hybrid client server...");
|
||||
|
@ -34,7 +32,7 @@ export async function createHybridClientServer() {
|
|||
);
|
||||
|
||||
// Register message handlers
|
||||
client.registerHandler("remote/peers/add", async (message) => {
|
||||
client.registerHandler("remoteExitNode/peers/add", async (message) => {
|
||||
const { pubKey, allowedIps } = message.data;
|
||||
|
||||
// TODO: we are getting the exit node twice here
|
||||
|
@ -46,7 +44,7 @@ export async function createHybridClientServer() {
|
|||
});
|
||||
});
|
||||
|
||||
client.registerHandler("remote/peers/remove", async (message) => {
|
||||
client.registerHandler("remoteExitNode/peers/remove", async (message) => {
|
||||
const { pubKey } = message.data;
|
||||
|
||||
// TODO: we are getting the exit node twice here
|
||||
|
@ -55,7 +53,69 @@ export async function createHybridClientServer() {
|
|||
await deletePeer(exitNode.exitNodeId, pubKey);
|
||||
});
|
||||
|
||||
client.registerHandler("remote/traefik/reload", async (message) => {
|
||||
// /update-proxy-mapping
|
||||
client.registerHandler("remoteExitNode/update-proxy-mapping", async (message) => {
|
||||
try {
|
||||
const [exitNode] = await db.select().from(exitNodes).limit(1);
|
||||
if (!exitNode) {
|
||||
logger.error("No exit node found for proxy mapping update");
|
||||
return;
|
||||
}
|
||||
|
||||
const response = await axios.post(`${exitNode.endpoint}/update-proxy-mapping`, message.data);
|
||||
logger.info(`Successfully updated proxy mapping: ${response.status}`);
|
||||
} catch (error) {
|
||||
// Extract useful information from axios error without circular references
|
||||
if (error && typeof error === 'object' && 'response' in error) {
|
||||
const axiosError = error as any;
|
||||
logger.error("Failed to update proxy mapping:", {
|
||||
status: axiosError.response?.status,
|
||||
statusText: axiosError.response?.statusText,
|
||||
data: axiosError.response?.data,
|
||||
message: axiosError.message,
|
||||
url: axiosError.config?.url
|
||||
});
|
||||
} else {
|
||||
logger.error("Failed to update proxy mapping:", {
|
||||
message: error instanceof Error ? error.message : String(error),
|
||||
stack: error instanceof Error ? error.stack : undefined
|
||||
});
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// /update-destinations
|
||||
client.registerHandler("remoteExitNode/update-destinations", async (message) => {
|
||||
try {
|
||||
const [exitNode] = await db.select().from(exitNodes).limit(1);
|
||||
if (!exitNode) {
|
||||
logger.error("No exit node found for destinations update");
|
||||
return;
|
||||
}
|
||||
|
||||
const response = await axios.post(`${exitNode.endpoint}/update-destinations`, message.data);
|
||||
logger.info(`Successfully updated destinations: ${response.status}`);
|
||||
} catch (error) {
|
||||
// Extract useful information from axios error without circular references
|
||||
if (error && typeof error === 'object' && 'response' in error) {
|
||||
const axiosError = error as any;
|
||||
logger.error("Failed to update destinations:", {
|
||||
status: axiosError.response?.status,
|
||||
statusText: axiosError.response?.statusText,
|
||||
data: axiosError.response?.data,
|
||||
message: axiosError.message,
|
||||
url: axiosError.config?.url
|
||||
});
|
||||
} else {
|
||||
logger.error("Failed to update proxy mapping:", {
|
||||
message: error instanceof Error ? error.message : String(error),
|
||||
stack: error instanceof Error ? error.stack : undefined
|
||||
});
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
client.registerHandler("remoteExitNode/traefik/reload", async (message) => {
|
||||
await monitor.HandleTraefikConfig();
|
||||
});
|
||||
|
||||
|
@ -72,7 +132,9 @@ export async function createHybridClientServer() {
|
|||
});
|
||||
|
||||
client.on("message", (message) => {
|
||||
logger.info(`Received message: ${message.type} ${JSON.stringify(message.data)}`);
|
||||
logger.info(
|
||||
`Received message: ${message.type} ${JSON.stringify(message.data)}`
|
||||
);
|
||||
});
|
||||
|
||||
// Connect to the server
|
||||
|
|
86
server/lib/exitNodeComms.ts
Normal file
86
server/lib/exitNodeComms.ts
Normal file
|
@ -0,0 +1,86 @@
|
|||
import axios from "axios";
|
||||
import logger from "@server/logger";
|
||||
import { ExitNode } from "@server/db";
|
||||
|
||||
interface ExitNodeRequest {
|
||||
remoteType: string;
|
||||
localPath: string;
|
||||
method?: "POST" | "DELETE" | "GET" | "PUT";
|
||||
data?: any;
|
||||
queryParams?: Record<string, string>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a request to an exit node, handling both remote and local exit nodes
|
||||
* @param exitNode The exit node to send the request to
|
||||
* @param request The request configuration
|
||||
* @returns Promise<any> Response data for local nodes, undefined for remote nodes
|
||||
*/
|
||||
export async function sendToExitNode(
|
||||
exitNode: ExitNode,
|
||||
request: ExitNodeRequest
|
||||
): Promise<any> {
|
||||
if (!exitNode.reachableAt) {
|
||||
throw new Error(
|
||||
`Exit node with ID ${exitNode.exitNodeId} is not reachable`
|
||||
);
|
||||
}
|
||||
|
||||
// Handle local exit node with HTTP API
|
||||
const method = request.method || "POST";
|
||||
let url = `${exitNode.reachableAt}${request.localPath}`;
|
||||
|
||||
// Add query parameters if provided
|
||||
if (request.queryParams) {
|
||||
const params = new URLSearchParams(request.queryParams);
|
||||
url += `?${params.toString()}`;
|
||||
}
|
||||
|
||||
try {
|
||||
let response;
|
||||
|
||||
switch (method) {
|
||||
case "POST":
|
||||
response = await axios.post(url, request.data, {
|
||||
headers: {
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
});
|
||||
break;
|
||||
case "DELETE":
|
||||
response = await axios.delete(url);
|
||||
break;
|
||||
case "GET":
|
||||
response = await axios.get(url);
|
||||
break;
|
||||
case "PUT":
|
||||
response = await axios.put(url, request.data, {
|
||||
headers: {
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
});
|
||||
break;
|
||||
default:
|
||||
throw new Error(`Unsupported HTTP method: ${method}`);
|
||||
}
|
||||
|
||||
logger.info(`Exit node request successful:`, {
|
||||
method,
|
||||
url,
|
||||
status: response.data.status
|
||||
});
|
||||
|
||||
return response.data;
|
||||
} catch (error) {
|
||||
if (axios.isAxiosError(error)) {
|
||||
logger.error(
|
||||
`Error making ${method} request (can Pangolin see Gerbil HTTP API?) for exit node at ${exitNode.reachableAt} (status: ${error.response?.status}): ${error.message}`
|
||||
);
|
||||
} else {
|
||||
logger.error(
|
||||
`Error making ${method} request for exit node at ${exitNode.reachableAt}: ${error}`
|
||||
);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
|
@ -17,7 +17,7 @@ import {
|
|||
addPeer as olmAddPeer,
|
||||
deletePeer as olmDeletePeer
|
||||
} from "../olm/peers";
|
||||
import axios from "axios";
|
||||
import { sendToExitNode } from "../../lib/exitNodeComms";
|
||||
|
||||
const updateClientParamsSchema = z
|
||||
.object({
|
||||
|
@ -144,10 +144,12 @@ export async function updateClient(
|
|||
const [clientSite] = await db
|
||||
.select()
|
||||
.from(clientSites)
|
||||
.where(and(
|
||||
.where(
|
||||
and(
|
||||
eq(clientSites.clientId, client.clientId),
|
||||
eq(clientSites.siteId, siteId)
|
||||
))
|
||||
)
|
||||
)
|
||||
.limit(1);
|
||||
|
||||
if (!clientSite || !clientSite.endpoint) {
|
||||
|
@ -274,18 +276,14 @@ export async function updateClient(
|
|||
const sitesData = await db
|
||||
.select()
|
||||
.from(sites)
|
||||
.innerJoin(
|
||||
clientSites,
|
||||
eq(sites.siteId, clientSites.siteId)
|
||||
)
|
||||
.leftJoin(
|
||||
exitNodes,
|
||||
eq(sites.exitNodeId, exitNodes.exitNodeId)
|
||||
)
|
||||
.innerJoin(clientSites, eq(sites.siteId, clientSites.siteId))
|
||||
.leftJoin(exitNodes, eq(sites.exitNodeId, exitNodes.exitNodeId))
|
||||
.where(eq(clientSites.clientId, client.clientId));
|
||||
|
||||
let exitNodeDestinations: {
|
||||
reachableAt: string;
|
||||
exitNodeId: number;
|
||||
type: string;
|
||||
sourceIp: string;
|
||||
sourcePort: number;
|
||||
destinations: PeerDestination[];
|
||||
|
@ -314,12 +312,15 @@ export async function updateClient(
|
|||
if (!destinations) {
|
||||
destinations = {
|
||||
reachableAt: site.exitNodes?.reachableAt || "",
|
||||
exitNodeId: site.exitNodes?.exitNodeId || 0,
|
||||
type: site.exitNodes?.type || "",
|
||||
sourceIp: site.clientSites.endpoint.split(":")[0] || "",
|
||||
sourcePort: parseInt(site.clientSites.endpoint.split(":")[1]) || 0,
|
||||
sourcePort:
|
||||
parseInt(site.clientSites.endpoint.split(":")[1]) ||
|
||||
0,
|
||||
destinations: [
|
||||
{
|
||||
destinationIP:
|
||||
site.sites.subnet.split("/")[0],
|
||||
destinationIP: site.sites.subnet.split("/")[0],
|
||||
destinationPort: site.sites.listenPort || 0
|
||||
}
|
||||
]
|
||||
|
@ -340,7 +341,6 @@ export async function updateClient(
|
|||
}
|
||||
|
||||
for (const destination of exitNodeDestinations) {
|
||||
try {
|
||||
logger.info(
|
||||
`Updating destinations for exit node at ${destination.reachableAt}`
|
||||
);
|
||||
|
@ -352,30 +352,20 @@ export async function updateClient(
|
|||
logger.info(
|
||||
`Payload for update-destinations: ${JSON.stringify(payload, null, 2)}`
|
||||
);
|
||||
const response = await axios.post(
|
||||
`${destination.reachableAt}/update-destinations`,
|
||||
payload,
|
||||
{
|
||||
headers: {
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
logger.info("Destinations updated:", {
|
||||
peer: response.data.status
|
||||
// Create an ExitNode-like object for sendToExitNode
|
||||
const exitNodeForComm = {
|
||||
exitNodeId: destination.exitNodeId,
|
||||
type: destination.type,
|
||||
reachableAt: destination.reachableAt
|
||||
} as any; // Using 'as any' since we know sendToExitNode will handle this correctly
|
||||
|
||||
await sendToExitNode(exitNodeForComm, {
|
||||
remoteType: "remoteExitNode/update-destinations",
|
||||
localPath: "/update-destinations",
|
||||
method: "POST",
|
||||
data: payload
|
||||
});
|
||||
} catch (error) {
|
||||
if (axios.isAxiosError(error)) {
|
||||
logger.error(
|
||||
`Error updating destinations (can Pangolin see Gerbil HTTP API?) for exit node at ${destination.reachableAt} (status: ${error.response?.status}): ${JSON.stringify(error.response?.data, null, 2)}`
|
||||
);
|
||||
} else {
|
||||
logger.error(
|
||||
`Error updating destinations for exit node at ${destination.reachableAt}: ${error}`
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Fetch the updated client
|
||||
|
|
|
@ -1,8 +1,8 @@
|
|||
import axios from "axios";
|
||||
import logger from "@server/logger";
|
||||
import { db } from "@server/db";
|
||||
import { exitNodes } from "@server/db";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { sendToExitNode } from "../../lib/exitNodeComms";
|
||||
|
||||
export async function addPeer(
|
||||
exitNodeId: number,
|
||||
|
@ -22,34 +22,13 @@ export async function addPeer(
|
|||
if (!exitNode) {
|
||||
throw new Error(`Exit node with ID ${exitNodeId} not found`);
|
||||
}
|
||||
if (!exitNode.reachableAt) {
|
||||
throw new Error(`Exit node with ID ${exitNodeId} is not reachable`);
|
||||
}
|
||||
|
||||
try {
|
||||
const response = await axios.post(
|
||||
`${exitNode.reachableAt}/peer`,
|
||||
peer,
|
||||
{
|
||||
headers: {
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
logger.info("Peer added successfully:", { peer: response.data.status });
|
||||
return response.data;
|
||||
} catch (error) {
|
||||
if (axios.isAxiosError(error)) {
|
||||
logger.error(
|
||||
`Error adding peer (can Pangolin see Gerbil HTTP API?) for exit node at ${exitNode.reachableAt} (status: ${error.response?.status}): ${error.message}`
|
||||
);
|
||||
} else {
|
||||
logger.error(
|
||||
`Error adding peer for exit node at ${exitNode.reachableAt}: ${error}`
|
||||
);
|
||||
}
|
||||
}
|
||||
return await sendToExitNode(exitNode, {
|
||||
remoteType: "remoteExitNode/peers/add",
|
||||
localPath: "/peer",
|
||||
method: "POST",
|
||||
data: peer
|
||||
});
|
||||
}
|
||||
|
||||
export async function deletePeer(exitNodeId: number, publicKey: string) {
|
||||
|
@ -64,24 +43,16 @@ export async function deletePeer(exitNodeId: number, publicKey: string) {
|
|||
if (!exitNode) {
|
||||
throw new Error(`Exit node with ID ${exitNodeId} not found`);
|
||||
}
|
||||
if (!exitNode.reachableAt) {
|
||||
throw new Error(`Exit node with ID ${exitNodeId} is not reachable`);
|
||||
}
|
||||
try {
|
||||
const response = await axios.delete(
|
||||
`${exitNode.reachableAt}/peer?public_key=${encodeURIComponent(publicKey)}`
|
||||
);
|
||||
logger.info("Peer deleted successfully:", response.data.status);
|
||||
return response.data;
|
||||
} catch (error) {
|
||||
if (axios.isAxiosError(error)) {
|
||||
logger.error(
|
||||
`Error deleting peer (can Pangolin see Gerbil HTTP API?) for exit node at ${exitNode.reachableAt} (status: ${error.response?.status}): ${error.message}`
|
||||
);
|
||||
} else {
|
||||
logger.error(
|
||||
`Error deleting peer for exit node at ${exitNode.reachableAt}: ${error}`
|
||||
);
|
||||
}
|
||||
|
||||
return await sendToExitNode(exitNode, {
|
||||
remoteType: "remoteExitNode/peers/remove",
|
||||
localPath: "/peer",
|
||||
method: "DELETE",
|
||||
data: {
|
||||
publicKey: publicKey
|
||||
},
|
||||
queryParams: {
|
||||
public_key: publicKey
|
||||
}
|
||||
});
|
||||
}
|
||||
|
|
|
@ -13,7 +13,7 @@ import {
|
|||
import { clients, clientSites, Newt, sites } from "@server/db";
|
||||
import { eq, and, inArray } from "drizzle-orm";
|
||||
import { updatePeer } from "../olm/peers";
|
||||
import axios from "axios";
|
||||
import { sendToExitNode } from "../../lib/exitNodeComms";
|
||||
|
||||
const inputSchema = z.object({
|
||||
publicKey: z.string(),
|
||||
|
@ -102,11 +102,12 @@ export const handleGetConfigMessage: MessageHandler = async (context) => {
|
|||
.from(exitNodes)
|
||||
.where(eq(exitNodes.exitNodeId, site.exitNodeId))
|
||||
.limit(1);
|
||||
if (exitNode.reachableAt && existingSite.subnet && existingSite.listenPort) {
|
||||
try {
|
||||
const response = await axios.post(
|
||||
`${exitNode.reachableAt}/update-proxy-mapping`,
|
||||
{
|
||||
if (
|
||||
exitNode.reachableAt &&
|
||||
existingSite.subnet &&
|
||||
existingSite.listenPort
|
||||
) {
|
||||
const payload = {
|
||||
oldDestination: {
|
||||
destinationIP: existingSite.subnet?.split("/")[0],
|
||||
destinationPort: existingSite.listenPort
|
||||
|
@ -115,28 +116,14 @@ export const handleGetConfigMessage: MessageHandler = async (context) => {
|
|||
destinationIP: site.subnet?.split("/")[0],
|
||||
destinationPort: site.listenPort
|
||||
}
|
||||
},
|
||||
{
|
||||
headers: {
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
}
|
||||
);
|
||||
};
|
||||
|
||||
logger.info("Destinations updated:", {
|
||||
peer: response.data.status
|
||||
await sendToExitNode(exitNode, {
|
||||
remoteType: "remoteExitNode/update-proxy-mapping",
|
||||
localPath: "/update-proxy-mapping",
|
||||
method: "POST",
|
||||
data: payload
|
||||
});
|
||||
} catch (error) {
|
||||
if (axios.isAxiosError(error)) {
|
||||
logger.error(
|
||||
`Error updating proxy mapping (can Pangolin see Gerbil HTTP API?) for exit node at ${exitNode.reachableAt} (status: ${error.response?.status}): ${error.message}`
|
||||
);
|
||||
} else {
|
||||
logger.error(
|
||||
`Error updating proxy mapping for exit node at ${exitNode.reachableAt}: ${error}`
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -237,7 +224,9 @@ export const handleGetConfigMessage: MessageHandler = async (context) => {
|
|||
protocol: resources.protocol
|
||||
})
|
||||
.from(resources)
|
||||
.where(and(eq(resources.siteId, siteId), eq(resources.http, false)));
|
||||
.where(
|
||||
and(eq(resources.siteId, siteId), eq(resources.http, false))
|
||||
);
|
||||
|
||||
// Get all enabled targets for these resources in a single query
|
||||
const resourceIds = resourcesList.map((r) => r.resourceId);
|
||||
|
@ -251,7 +240,7 @@ export const handleGetConfigMessage: MessageHandler = async (context) => {
|
|||
method: targets.method,
|
||||
port: targets.port,
|
||||
internalPort: targets.internalPort,
|
||||
enabled: targets.enabled,
|
||||
enabled: targets.enabled
|
||||
})
|
||||
.from(targets)
|
||||
.where(
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue