* as public key may be from the web client, need to use new _getPublicKey utility that uses maybeGetAsyncProperty
This commit is contained in:
parent
10a7b4ebc6
commit
96dd1ad46e
|
@ -5,7 +5,11 @@ import { deserializeError } from "serialize-error";
|
||||||
import b4a from "b4a";
|
import b4a from "b4a";
|
||||||
import type { TcpSocketConnectOpts } from "net";
|
import type { TcpSocketConnectOpts } from "net";
|
||||||
import { DataSocketOptions, PeerOptions } from "../peer.js";
|
import { DataSocketOptions, PeerOptions } from "../peer.js";
|
||||||
import { roundRobinFactory, idFactory } from "../util.js";
|
import {
|
||||||
|
roundRobinFactory,
|
||||||
|
idFactory,
|
||||||
|
maybeGetAsyncProperty,
|
||||||
|
} from "../util.js";
|
||||||
import {
|
import {
|
||||||
CloseSocketRequest,
|
CloseSocketRequest,
|
||||||
ErrorSocketRequest,
|
ErrorSocketRequest,
|
||||||
|
@ -119,14 +123,16 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
return this._sockets;
|
return this._sockets;
|
||||||
}
|
}
|
||||||
|
|
||||||
handleNewPeerChannel(peer: Peer) {
|
async handleNewPeerChannel(peer: Peer) {
|
||||||
this.update(peer.stream.remotePublicKey, { peer });
|
this.update(await this._getPublicKey(peer), {
|
||||||
|
peer,
|
||||||
|
});
|
||||||
|
|
||||||
this._registerOpenSocketMessage(peer);
|
await this._registerOpenSocketMessage(peer);
|
||||||
this._registerWriteSocketMessage(peer);
|
await this._registerWriteSocketMessage(peer);
|
||||||
this._registerCloseSocketMessage(peer);
|
await this._registerCloseSocketMessage(peer);
|
||||||
this._registerTimeoutSocketMessage(peer);
|
await this._registerTimeoutSocketMessage(peer);
|
||||||
this._registerErrorSocketMessage(peer);
|
await this._registerErrorSocketMessage(peer);
|
||||||
}
|
}
|
||||||
|
|
||||||
async handleClosePeer(peer: Peer) {
|
async handleClosePeer(peer: Peer) {
|
||||||
|
@ -136,7 +142,7 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
const pubkey = this._toString(peer.stream.remotePublicKey);
|
const pubkey = this._toString(await this._getPublicKey(peer));
|
||||||
|
|
||||||
if (this._peers.has(pubkey)) {
|
if (this._peers.has(pubkey)) {
|
||||||
this._peers.delete(pubkey);
|
this._peers.delete(pubkey);
|
||||||
|
@ -181,7 +187,7 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
return socket;
|
return socket;
|
||||||
}
|
}
|
||||||
|
|
||||||
private _registerOpenSocketMessage(peer: Peer) {
|
private async _registerOpenSocketMessage(peer: Peer) {
|
||||||
const self = this;
|
const self = this;
|
||||||
const message = peer.channel.addMessage({
|
const message = peer.channel.addMessage({
|
||||||
encoding: {
|
encoding: {
|
||||||
|
@ -194,7 +200,7 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
self._allowedPorts.length &&
|
self._allowedPorts.length &&
|
||||||
!self._allowedPorts.includes((m as TcpSocketConnectOpts).port)
|
!self._allowedPorts.includes((m as TcpSocketConnectOpts).port)
|
||||||
) {
|
) {
|
||||||
self.get(peer.stream.remotePublicKey).messages.errorSocket.send({
|
self.get(await this._getPublicKey(peer)).messages.errorSocket.send({
|
||||||
id: (m as SocketRequest).id,
|
id: (m as SocketRequest).id,
|
||||||
err: new Error(
|
err: new Error(
|
||||||
`port ${(m as TcpSocketConnectOpts).port} not allowed`
|
`port ${(m as TcpSocketConnectOpts).port} not allowed`
|
||||||
|
@ -210,7 +216,7 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
nextSocketId(),
|
nextSocketId(),
|
||||||
m,
|
m,
|
||||||
self,
|
self,
|
||||||
self.get(peer.stream.remotePublicKey) as PeerEntity,
|
self.get(await this._getPublicKey(peer)) as PeerEntity,
|
||||||
m
|
m
|
||||||
).connect();
|
).connect();
|
||||||
return;
|
return;
|
||||||
|
@ -224,12 +230,12 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
this.update(peer.stream.remotePublicKey, {
|
this.update(await this._getPublicKey(peer), {
|
||||||
messages: { openSocket: message },
|
messages: { openSocket: message },
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
private _registerWriteSocketMessage(peer: Peer) {
|
private async _registerWriteSocketMessage(peer: Peer) {
|
||||||
const self = this;
|
const self = this;
|
||||||
const message = peer.channel.addMessage({
|
const message = peer.channel.addMessage({
|
||||||
encoding: writeSocketEncoding,
|
encoding: writeSocketEncoding,
|
||||||
|
@ -237,12 +243,12 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
self._sockets.get(m.id)?.push(m.data);
|
self._sockets.get(m.id)?.push(m.data);
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
this.update(peer.stream.remotePublicKey, {
|
this.update(await this._getPublicKey(peer), {
|
||||||
messages: { writeSocket: message },
|
messages: { writeSocket: message },
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
private _registerCloseSocketMessage(peer: Peer) {
|
private async _registerCloseSocketMessage(peer: Peer) {
|
||||||
const self = this;
|
const self = this;
|
||||||
const message = peer.channel.addMessage({
|
const message = peer.channel.addMessage({
|
||||||
encoding: socketEncoding,
|
encoding: socketEncoding,
|
||||||
|
@ -250,12 +256,12 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
self._sockets.get(m.id)?.end();
|
self._sockets.get(m.id)?.end();
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
this.update(peer.stream.remotePublicKey, {
|
this.update(await this._getPublicKey(peer), {
|
||||||
messages: { closeSocket: message },
|
messages: { closeSocket: message },
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
private _registerTimeoutSocketMessage(peer: Peer) {
|
private async _registerTimeoutSocketMessage(peer: Peer) {
|
||||||
const self = this;
|
const self = this;
|
||||||
const message = peer.channel.addMessage({
|
const message = peer.channel.addMessage({
|
||||||
encoding: socketEncoding,
|
encoding: socketEncoding,
|
||||||
|
@ -264,12 +270,12 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
self._sockets.get(m.id)?.emit("timeout");
|
self._sockets.get(m.id)?.emit("timeout");
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
this.update(peer.stream.remotePublicKey, {
|
this.update(await this._getPublicKey(peer), {
|
||||||
messages: { timeoutSocket: message },
|
messages: { timeoutSocket: message },
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
private _registerErrorSocketMessage(peer: Peer) {
|
private async _registerErrorSocketMessage(peer: Peer) {
|
||||||
const self = this;
|
const self = this;
|
||||||
const message = peer.channel.addMessage({
|
const message = peer.channel.addMessage({
|
||||||
encoding: errorSocketEncoding,
|
encoding: errorSocketEncoding,
|
||||||
|
@ -278,7 +284,7 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
self._sockets.get(m.id)?.emit("error", m.err);
|
self._sockets.get(m.id)?.emit("error", m.err);
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
this.update(peer.stream.remotePublicKey, {
|
this.update(await this._getPublicKey(peer), {
|
||||||
messages: { errorSocket: message },
|
messages: { errorSocket: message },
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
@ -286,4 +292,8 @@ export default class MultiSocketProxy extends Proxy {
|
||||||
private _toString(pubkey: Uint8Array) {
|
private _toString(pubkey: Uint8Array) {
|
||||||
return b4a.from(pubkey).toString("hex");
|
return b4a.from(pubkey).toString("hex");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private async _getPublicKey(peer: Peer) {
|
||||||
|
return maybeGetAsyncProperty(peer.stream.remotePublicKey);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue