mercuryworkshop-mirror/adrift
1
0
Fork 0
mirror of https://github.com/MercuryWorkshop/adrift.git synced 2025-05-13 06:10:01 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

monorepo part 1

Browse source
This commit is contained in:
Spencer Pogorzelski 2023-08-12 11:52:31 -07:00
parent aa838cbff6
commit 2138e02613
26 changed files with 681 additions and 14437 deletions
Download patch file Download diff file Expand all files Collapse all files

188
server/client.ts
View file

@ -1,188 +0,0 @@
import {
ClientRequest,
Agent as HTTPAgent,
IncomingMessage,
RequestOptions,
STATUS_CODES,
request as httpRequest,
} from "http";
import { Readable } from "stream";
import EventEmitter from "events";
import {
C2SRequestTypes,
HTTPRequestPayload,
HTTPResponsePayload,
ProtoBareHeaders,
S2CRequestTypes,
} from "../protocol/index.js";
import { BareError, bareFetch, options } from "./http.js";
export class Client {
send: (msg: Buffer) => void;
events: EventEmitter;
constructor(send) {
this.send = send;
this.events = new EventEmitter();
}
static parseMsgInit(
msg: Buffer
): { cursor: number; seq: number; op: number } | undefined {
try {
let cursor = 0;
const seq = msg.readUint16BE(cursor);
cursor += 2;
const op = msg.readUint8(cursor);
cursor += 1;
return { cursor, seq, op };
} catch (e) {
if (e instanceof RangeError) {
// malformed message
return;
}
throw e;
}
}
static parseHttpReqPayload(
payloadRaw: Buffer
): HTTPRequestPayload | undefined {
let payload;
try {
payload = JSON.parse(payloadRaw.toString());
} catch (e) {
if (e instanceof SyntaxError) {
return;
}
throw e;
}
console.log({ payload });
return payload;
}
static bareErrorToResponse(e: BareError): {
payload: HTTPResponsePayload;
body: Buffer;
} {
return {
payload: {
status: e.status,
statusText: STATUS_CODES[e.status] || "",
headers: {},
},
body: Buffer.from(JSON.stringify(e.body)),
};
}
async handleHTTPRequest(payload: HTTPRequestPayload): Promise<{
payload: HTTPResponsePayload;
body: Buffer;
}> {
const abort = new AbortController();
const onClose = () => {
abort.abort();
this.events.off("close", onClose);
};
this.events.on("close", onClose);
let resp: IncomingMessage;
try {
resp = await bareFetch(
payload,
abort.signal,
new URL(payload.remote),
options
);
} catch (e) {
if (e instanceof BareError) {
return Client.bareErrorToResponse(e);
}
this.events.off("close", onClose);
throw e;
}
this.events.off("close", onClose);
const buffers: any[] = [];
// node.js readable streams implement the async iterator protocol
for await (const data of resp) {
buffers.push(data);
}
const body = Buffer.concat(buffers);
return {
payload: {
status: resp.statusCode || 500,
statusText: resp.statusMessage || "",
headers: Object.fromEntries(
Object.entries(resp.headersDistinct).filter(([k, v]) => Boolean(v))
) as ProtoBareHeaders,
},
body,
};
}
sendHTTPResponse(seq: number, payload: HTTPResponsePayload, body: Buffer) {
const payloadBuffer = Buffer.from(JSON.stringify(payload));
const buf = Buffer.alloc(2 + 1 + 4 + payloadBuffer.length + body.length);
let cursor = 0;
cursor = buf.writeUInt16BE(seq, cursor);
cursor = buf.writeUInt8(S2CRequestTypes.HTTPResponse, cursor);
cursor = buf.writeUInt32BE(payloadBuffer.length, cursor);
cursor += payloadBuffer.copy(buf, cursor);
body.copy(buf, cursor);
this.send(buf);
}
async onMsg(msg: Buffer) {
const init = Client.parseMsgInit(msg);
if (!init) return;
const { cursor, seq, op } = init;
switch (op) {
case C2SRequestTypes.HTTPRequest:
let resp;
const reqPayload = Client.parseHttpReqPayload(msg.subarray(cursor));
if (!reqPayload) return;
try {
resp = await this.handleHTTPRequest(reqPayload);
} catch (e) {
if (options.logErrors) console.error(e);
let bareError;
if (e instanceof BareError) {
bareError = e;
} else if (e instanceof Error) {
bareError = new BareError(500, {
code: "UNKNOWN",
id: `error.${e.name}`,
message: e.message,
stack: e.stack,
});
} else {
bareError = new BareError(500, {
code: "UNKNOWN",
id: "error.Exception",
message: "Error: " + e,
stack: new Error(<string | undefined>e).stack,
});
}
resp = Client.bareErrorToResponse(bareError);
}
const { payload, body } = resp;
this.sendHTTPResponse(seq, payload, body);
break;
default:
// not implemented
break;
}
}
onClose() {
this.events.emit("close");
}
}

195
server/http.ts
View file

@ -1,195 +0,0 @@
import { HTTPRequestPayload } from "../protocol/index.js";
import { LookupAddress, LookupAllOptions, lookup } from "dns";
import {
ClientRequest,
Agent as HTTPAgent,
IncomingMessage,
RequestOptions,
STATUS_CODES,
request as httpRequest,
} from "http";
import { Agent as HTTPSAgent, request as httpsRequest } from "https";
import fuck from "ipaddr.js";
const { isValid, parse } = fuck;
import { Readable } from "stream";
export interface BareErrorBody {
code: string;
id: string;
message?: string;
stack?: string;
}
export class BareError extends Error {
status: number;
body: BareErrorBody;
constructor(status: number, body: BareErrorBody) {
super(body.message || body.code);
this.status = status;
this.body = body;
}
}
export const options: BareServerOptions = {
logErrors: true,
filterRemote: (url) => {
// if the remote is an IP then it didn't go through the init.lookup hook
// isValid determines if this is so
if (isValid(url.hostname) && parse(url.hostname).range() !== "unicast")
throw new RangeError("Forbidden IP");
},
lookup: (hostname, options, callback) =>
lookup(hostname, options, (err, address, family) => {
if (
address &&
toAddressArray(address, family).some(
({ address }) => parse(address).range() !== "unicast"
)
)
callback(new RangeError("Forbidden IP"), [], -1);
else callback(err, address, family);
}),
httpAgent: new HTTPAgent({
keepAlive: true,
}),
httpsAgent: new HTTPSAgent({
keepAlive: true,
}),
database: new Map<string, string>(),
};
export interface BareServerOptions {
logErrors: boolean;
/**
* Callback for filtering the remote URL.
* @returns Nothing
* @throws An error if the remote is bad.
*/
filterRemote?: (remote: Readonly<URL>) => Promise<void> | void;
/**
* DNS lookup
* May not get called when remote.host is an IP
* Use in combination with filterRemote to block IPs
*/
lookup: (
hostname: string,
options: LookupAllOptions,
callback: (
err: NodeJS.ErrnoException | null,
addresses: LookupAddress[],
family: number
) => void
) => void;
localAddress?: string;
family?: number;
httpAgent: HTTPAgent;
httpsAgent: HTTPSAgent;
database: Map<string, string>;
}
export interface Address {
address: string;
family: number;
}
/**
* Converts the address and family of a DNS lookup callback into an array if it wasn't already
*/
export function toAddressArray(address: string | Address[], family?: number) {
if (typeof address === "string")
return [
{
address,
family,
},
] as Address[];
else return address;
}
function outgoingError<T>(error: T): T | BareError {
if (error instanceof Error) {
switch ((<Error & { code?: string }>error).code) {
case "ENOTFOUND":
return new BareError(500, {
code: "HOST_NOT_FOUND",
id: "request",
message: "The specified host could not be resolved.",
});
case "ECONNREFUSED":
return new BareError(500, {
code: "CONNECTION_REFUSED",
id: "response",
message: "The remote rejected the request.",
});
case "ECONNRESET":
return new BareError(500, {
code: "CONNECTION_RESET",
id: "response",
message: "The request was forcibly closed.",
});
case "ETIMEOUT":
return new BareError(500, {
code: "CONNECTION_TIMEOUT",
id: "response",
message: "The response timed out.",
});
}
}
return error;
}
export async function bareFetch(
request: HTTPRequestPayload,
signal: AbortSignal,
remote: URL,
options: BareServerOptions
): Promise<IncomingMessage> {
if (options.filterRemote) await options.filterRemote(remote);
const req: RequestOptions = {
method: request.method,
headers: request.requestHeaders,
setHost: false,
signal,
localAddress: options.localAddress,
family: options.family,
lookup: options.lookup,
};
let outgoing: ClientRequest;
// NodeJS will convert the URL into HTTP options automatically
// see https://github.com/nodejs/node/blob/e30e71665cab94118833cc536a43750703b19633/lib/internal/url.js#L1277
if (remote.protocol === "https:")
outgoing = httpsRequest(remote, {
...req,
agent: options.httpsAgent,
});
else if (remote.protocol === "http:")
outgoing = httpRequest(remote, {
...req,
agent: options.httpAgent,
});
else throw new RangeError(`Unsupported protocol: '${remote.protocol}'`);
if (request.body) Readable.from([request.body]).pipe(outgoing);
else outgoing.end();
return await new Promise((resolve, reject) => {
outgoing.on("response", (response: IncomingMessage) => {
resolve(response);
});
outgoing.on("upgrade", (req, socket) => {
reject("Remote did not send a response");
socket.destroy();
});
outgoing.on("error", (error: Error) => {
reject(outgoingError(error));
});
});
}

33
server/package.json Normal file
View file

@ -0,0 +1,33 @@
{
"name": "adrift",
"version": "1.0.0",
"description": "",
"main": "src/main.ts",
"scripts": {
"dev": "vite",
"build": "vite build",
"watch": "vite build --watch"
},
"keywords": [],
"author": "",
"license": "ISC",
"dependencies": {
"dotenv": "^16.3.1",
"express": "^4.18.2",
"express-ws": "^5.0.2",
"firebase": "^10.1.0",
"ipaddr.js": "^2.1.0",
"ts-node": "^10.9.1",
"typescript": "^5.1.6",
"wrtc": "^0.4.7",
"firebase-config": "workspace:*",
"protocol": "workspace:*"
},
"devDependencies": {
"@types/express": "^4.17.17",
"@types/express-ws": "^3.0.1",
"@types/node": "^20.4.10",
"@types/webrtc": "^0.0.36",
"nodemon": "^3.0.1"
}
}

180
server/src/client.ts Normal file
View file

@ -0,0 +1,180 @@
import { IncomingMessage, STATUS_CODES } from "http";
import EventEmitter from "events";
import {
C2SRequestTypes,
HTTPRequestPayload,
HTTPResponsePayload,
ProtoBareHeaders,
S2CRequestTypes,
} from "protocol";
import { BareError, bareFetch, options } from "./http";
export class Client {
send: (msg: Buffer) => void;
events: EventEmitter;
constructor(send: (msg: Buffer) => void) {
this.send = send;
this.events = new EventEmitter();
}
static parseMsgInit(
msg: Buffer
): { cursor: number; seq: number; op: number } | undefined {
try {
let cursor = 0;
const seq = msg.readUint16BE(cursor);
cursor += 2;
const op = msg.readUint8(cursor);
cursor += 1;
return { cursor, seq, op };
} catch (e) {
if (e instanceof RangeError) {
// malformed message
return;
}
throw e;
}
}
static parseHttpReqPayload(
payloadRaw: Buffer
): HTTPRequestPayload | undefined {
let payload;
try {
payload = JSON.parse(payloadRaw.toString());
} catch (e) {
if (e instanceof SyntaxError) {
return;
}
throw e;
}
console.log({ payload });
return payload;
}
static bareErrorToResponse(e: BareError): {
payload: HTTPResponsePayload;
body: Buffer;
} {
return {
payload: {
status: e.status,
statusText: STATUS_CODES[e.status] || "",
headers: {},
},
body: Buffer.from(JSON.stringify(e.body)),
};
}
async handleHTTPRequest(payload: HTTPRequestPayload): Promise<{
payload: HTTPResponsePayload;
body: Buffer;
}> {
const abort = new AbortController();
const onClose = () => {
abort.abort();
this.events.off("close", onClose);
};
this.events.on("close", onClose);
let resp: IncomingMessage;
try {
resp = await bareFetch(
payload,
abort.signal,
new URL(payload.remote),
options
);
} catch (e) {
if (e instanceof BareError) {
return Client.bareErrorToResponse(e);
}
this.events.off("close", onClose);
throw e;
}
this.events.off("close", onClose);
const buffers: any[] = [];
// node.js readable streams implement the async iterator protocol
for await (const data of resp) {
buffers.push(data);
}
const body = Buffer.concat(buffers);
return {
payload: {
status: resp.statusCode || 500,
statusText: resp.statusMessage || "",
headers: Object.fromEntries(
Object.entries(resp.headersDistinct).filter(([_k, v]) => Boolean(v))
) as ProtoBareHeaders,
},
body,
};
}
sendHTTPResponse(seq: number, payload: HTTPResponsePayload, body: Buffer) {
const payloadBuffer = Buffer.from(JSON.stringify(payload));
const buf = Buffer.alloc(2 + 1 + 4 + payloadBuffer.length + body.length);
let cursor = 0;
cursor = buf.writeUInt16BE(seq, cursor);
cursor = buf.writeUInt8(S2CRequestTypes.HTTPResponse, cursor);
cursor = buf.writeUInt32BE(payloadBuffer.length, cursor);
cursor += payloadBuffer.copy(buf, cursor);
body.copy(buf, cursor);
this.send(buf);
}
async onMsg(msg: Buffer) {
const init = Client.parseMsgInit(msg);
if (!init) return;
const { cursor, seq, op } = init;
switch (op) {
case C2SRequestTypes.HTTPRequest:
let resp;
const reqPayload = Client.parseHttpReqPayload(msg.subarray(cursor));
if (!reqPayload) return;
try {
resp = await this.handleHTTPRequest(reqPayload);
} catch (e) {
if (options.logErrors) console.error(e);
let bareError;
if (e instanceof BareError) {
bareError = e;
} else if (e instanceof Error) {
bareError = new BareError(500, {
code: "UNKNOWN",
id: `error.${e.name}`,
message: e.message,
stack: e.stack,
});
} else {
bareError = new BareError(500, {
code: "UNKNOWN",
id: "error.Exception",
message: "Error: " + e,
stack: new Error(<string | undefined>e).stack,
});
}
resp = Client.bareErrorToResponse(bareError);
}
const { payload, body } = resp;
this.sendHTTPResponse(seq, payload, body);
break;
default:
// not implemented
break;
}
}
onClose() {
this.events.emit("close");
}
}

0
server/dev.ts → server/src/dev.ts
View file

192
server/src/http.ts Normal file
View file

@ -0,0 +1,192 @@
import { LookupAddress, LookupAllOptions, lookup } from "dns";
import {
ClientRequest,
Agent as HTTPAgent,
IncomingMessage,
RequestOptions,
request as httpRequest,
} from "http";
import { Agent as HTTPSAgent, request as httpsRequest } from "https";
import fuck from "ipaddr.js";
import { HTTPRequestPayload } from "protocol";
import { Readable } from "stream";
const { isValid, parse } = fuck;
export interface BareErrorBody {
code: string;
id: string;
message?: string;
stack?: string;
}
export class BareError extends Error {
status: number;
body: BareErrorBody;
constructor(status: number, body: BareErrorBody) {
super(body.message || body.code);
this.status = status;
this.body = body;
}
}
export const options: BareServerOptions = {
logErrors: true,
filterRemote: (url) => {
// if the remote is an IP then it didn't go through the init.lookup hook
// isValid determines if this is so
if (isValid(url.hostname) && parse(url.hostname).range() !== "unicast")
throw new RangeError("Forbidden IP");
},
lookup: (hostname, options, callback) =>
lookup(hostname, options, (err: any, address: any, family: any) => {
if (
address &&
toAddressArray(address, family).some(
({ address }) => parse(address).range() !== "unicast"
)
)
callback(new RangeError("Forbidden IP"), [], -1);
else callback(err, address, family);
}),
httpAgent: new HTTPAgent({
keepAlive: true,
}),
httpsAgent: new HTTPSAgent({
keepAlive: true,
}),
database: new Map<string, string>(),
};
export interface BareServerOptions {
logErrors: boolean;
/**
* Callback for filtering the remote URL.
* @returns Nothing
* @throws An error if the remote is bad.
*/
filterRemote?: (remote: Readonly<URL>) => Promise<void> | void;
/**
* DNS lookup
* May not get called when remote.host is an IP
* Use in combination with filterRemote to block IPs
*/
lookup: (
hostname: string,
options: LookupAllOptions,
callback: (
err: NodeJS.ErrnoException | null,
addresses: LookupAddress[],
family: number
) => void
) => void;
localAddress?: string;
family?: number;
httpAgent: HTTPAgent;
httpsAgent: HTTPSAgent;
database: Map<string, string>;
}
export interface Address {
address: string;
family: number;
}
/**
* Converts the address and family of a DNS lookup callback into an array if it wasn't already
*/
export function toAddressArray(address: string | Address[], family?: number) {
if (typeof address === "string")
return [
{
address,
family,
},
] as Address[];
else return address;
}
function outgoingError<T>(error: T): T | BareError {
if (error instanceof Error) {
switch ((<Error & { code?: string }>error).code) {
case "ENOTFOUND":
return new BareError(500, {
code: "HOST_NOT_FOUND",
id: "request",
message: "The specified host could not be resolved.",
});
case "ECONNREFUSED":
return new BareError(500, {
code: "CONNECTION_REFUSED",
id: "response",
message: "The remote rejected the request.",
});
case "ECONNRESET":
return new BareError(500, {
code: "CONNECTION_RESET",
id: "response",
message: "The request was forcibly closed.",
});
case "ETIMEOUT":
return new BareError(500, {
code: "CONNECTION_TIMEOUT",
id: "response",
message: "The response timed out.",
});
}
}
return error;
}
export async function bareFetch(
request: HTTPRequestPayload,
signal: AbortSignal,
remote: URL,
options: BareServerOptions
): Promise<IncomingMessage> {
if (options.filterRemote) await options.filterRemote(remote);
const req: RequestOptions = {
method: request.method,
headers: request.requestHeaders,
setHost: false,
signal,
localAddress: options.localAddress,
family: options.family,
lookup: options.lookup,
};
let outgoing: ClientRequest;
// NodeJS will convert the URL into HTTP options automatically
// see https://github.com/nodejs/node/blob/e30e71665cab94118833cc536a43750703b19633/lib/internal/url.js#L1277
if (remote.protocol === "https:")
outgoing = httpsRequest(remote, {
...req,
agent: options.httpsAgent,
});
else if (remote.protocol === "http:")
outgoing = httpRequest(remote, {
...req,
agent: options.httpAgent,
});
else throw new RangeError(`Unsupported protocol: '${remote.protocol}'`);
if (request.body) Readable.from([request.body]).pipe(outgoing);
else outgoing.end();
return await new Promise((resolve, reject) => {
outgoing.on("response", (response: IncomingMessage) => {
resolve(response);
});
outgoing.on("upgrade", (_req, socket) => {
reject("Remote did not send a response");
socket.destroy();
});
outgoing.on("error", (error: Error) => {
reject(outgoingError(error));
});
});
}

42
server/main.ts → server/src/main.ts
View file

@ -2,11 +2,11 @@ import dotenv from "dotenv";
import express from "express";
import expressWs from "express-ws";
import wrtc from "wrtc";
import { Client } from "./client.js";
import { signInWithEmailAndPassword } from "firebase/auth";
import wrtc from "wrtc";
import { Client } from "./client";
import { auth } from "../firebase-config.js";
import { auth } from "firebase-config";
import { getDatabase, onValue, ref, set } from "firebase/database";
const configuration = {
@ -19,14 +19,13 @@ const configuration = {
dotenv.config();
async function connect(
offer,
candidates,
onAnswer: (answer: Record<string, any>) => void,
offer: RTCSessionDescriptionInit,
candidates: RTCIceCandidateInit[],
onAnswer: (answer: Record<string, any>) => void
): Promise<RTCDataChannel> {
const localCandidates: any[] = [];
let dataChannel;
const peer = new wrtc.RTCPeerConnection(configuration);
const peer: RTCPeerConnection = new wrtc.RTCPeerConnection(configuration);
let promise = new Promise((resolve) => {
peer.ondatachannel = (event) => {
dataChannel = event.channel;
@ -73,7 +72,7 @@ const app = express() as unknown as expressWs.Application;
expressWs(app);
app.use(express.json());
app.use((req, res, next) => {
app.use((_req, res, next) => {
res.header("x-robots-tag", "noindex");
res.header("access-control-allow-headers", "*");
res.header("access-control-allow-origin", "*");
@ -82,16 +81,12 @@ app.use((req, res, next) => {
next();
});
async function answerRtc(data: any, onrespond: (answer: any) => void) {
if (data && data.offer && data.localCandidates) {
const { offer, localCandidates } = data;
let didAnswer = false;
let dataChannel = await connect(offer, localCandidates, (answer) => {
if (!didAnswer) {
didAnswer = true;
onrespond(answer);
@ -104,9 +99,8 @@ async function answerRtc(data: any, onrespond: (answer: any) => void) {
dataChannel.onopen = () => {
console.log("opened");
client = new Client((msg) => dataChannel.send(msg));
};
dataChannel.onclose = (event) => {
dataChannel.onclose = () => {
console.log("closed");
client.onClose();
};
@ -114,7 +108,6 @@ async function answerRtc(data: any, onrespond: (answer: any) => void) {
console.log("messaged");
client.onMsg(Buffer.from(event.data));
};
}
}
@ -125,10 +118,7 @@ app.post("/connect", (req, res) => {
});
});
app.ws("/dev-ws", (ws, req) => {
app.ws("/dev-ws", (ws, _req) => {
console.log("ws connect");
const client = new Client((msg) => ws.send(msg));
@ -146,17 +136,11 @@ app.ws("/dev-ws", (ws, req) => {
});
async function connectFirebase() {
let creds = await signInWithEmailAndPassword(
auth,
"test@test.com",
"123456"
);
let creds = await signInWithEmailAndPassword(auth, "test@test.com", "123456");
const db = getDatabase();
let peer = ref(db, `/peers/${creds.user.uid}`);
set(peer, "");
onValue(peer, (snapshot) => {
@ -172,7 +156,7 @@ async function connectFirebase() {
answerRtc(data, (answer) => {
console.log("answering");
set(peer, JSON.stringify(answer));
})
});
}
}
});
@ -180,5 +164,3 @@ async function connectFirebase() {
connectFirebase();
// app.listen(3000, () => console.log("listening"));

0
server/rtc.ts → server/src/rtc.ts
View file

4
server/src/wrtc.d.ts vendored Normal file
View file

@ -0,0 +1,4 @@
declare module "wrtc" {
import * as RTC from "@types/webrtc";
export = RTC;
}

28
server/tsconfig.json Normal file
View file

@ -0,0 +1,28 @@
{
"compilerOptions": {
"target": "es2017",
"module": "commonjs",
"lib": ["dom", "es6", "es2017", "esnext.asynciterable"],
"skipLibCheck": true,
"sourceMap": true,
"outDir": "./dist",
"moduleResolution": "node",
"removeComments": true,
"noImplicitAny": true,
"strictNullChecks": true,
"strictFunctionTypes": true,
"noImplicitThis": true,
"noUnusedLocals": true,
"noUnusedParameters": true,
"noImplicitReturns": true,
"noFallthroughCasesInSwitch": true,
"allowSyntheticDefaultImports": true,
"esModuleInterop": true,
"emitDecoratorMetadata": true,
"experimentalDecorators": true,
"resolveJsonModule": true,
"baseUrl": "."
},
"exclude": ["node_modules"],
"include": ["./src/**/*.ts"]
}