diff --git a/packages/contracts/source/contracts/p2p/api-node-processor.ts b/packages/contracts/source/contracts/p2p/api-node-processor.ts index 42fa17344b..d1b75e442c 100644 --- a/packages/contracts/source/contracts/p2p/api-node-processor.ts +++ b/packages/contracts/source/contracts/p2p/api-node-processor.ts @@ -6,4 +6,6 @@ export interface AcceptNewApiNodeOptions { export interface ApiNodeProcessor { validateAndAcceptApiNode(apiNode: ApiNode, options?: AcceptNewApiNodeOptions): Promise; + + revalidateApiNode(apiNode: ApiNode): Promise; } diff --git a/packages/p2p/source/actions/index.ts b/packages/p2p/source/actions/index.ts deleted file mode 100644 index 77384b9636..0000000000 --- a/packages/p2p/source/actions/index.ts +++ /dev/null @@ -1,3 +0,0 @@ -export { RevalidateApiNodeAction } from "./revalidate-api-node.js"; -export { ValidateAndAcceptApiNodeAction } from "./validate-and-accept-api-node.js"; -export { ValidateAndAcceptPeerAction } from "./validate-and-accept-peer.js"; diff --git a/packages/p2p/source/actions/revalidate-api-node.ts b/packages/p2p/source/actions/revalidate-api-node.ts deleted file mode 100644 index 24b23976ca..0000000000 --- a/packages/p2p/source/actions/revalidate-api-node.ts +++ /dev/null @@ -1,23 +0,0 @@ -import type { Contracts } from "@mainsail/contracts"; - -import { Identifiers } from "@mainsail/constants"; -import { injectable } from "@mainsail/container"; -import { Services } from "@mainsail/kernel"; - -import { ApiNodeProcessor } from "../api-node-processor.js"; - -@injectable() -export class RevalidateApiNodeAction extends Services.Triggers.Action { - #app: Contracts.Kernel.Application; - - public constructor(app: Contracts.Kernel.Application) { - super(); - this.#app = app; - } - - public async execute(arguments_: { apiNode: Contracts.P2P.ApiNode }): Promise { - const apiNode: Contracts.P2P.ApiNode = arguments_.apiNode; - - return this.#app.get(Identifiers.P2P.ApiNode.Processor).revalidateApiNode(apiNode); - } -} diff --git a/packages/p2p/source/actions/validate-and-accept-api-node.ts b/packages/p2p/source/actions/validate-and-accept-api-node.ts deleted file mode 100644 index 499ffe8369..0000000000 --- a/packages/p2p/source/actions/validate-and-accept-api-node.ts +++ /dev/null @@ -1,29 +0,0 @@ -import type { Contracts } from "@mainsail/contracts"; - -import { Identifiers } from "@mainsail/constants"; -import { injectable } from "@mainsail/container"; -import { Services } from "@mainsail/kernel"; - -import { ApiNodeProcessor } from "../api-node-processor.js"; - -@injectable() -export class ValidateAndAcceptApiNodeAction extends Services.Triggers.Action { - #app: Contracts.Kernel.Application; - - public constructor(app: Contracts.Kernel.Application) { - super(); - this.#app = app; - } - - public async execute(arguments_: { - apiNode: Contracts.P2P.ApiNode; - options: Contracts.P2P.AcceptNewPeerOptions; - }): Promise { - const apiNode: Contracts.P2P.ApiNode = arguments_.apiNode; - const options: Contracts.P2P.AcceptNewPeerOptions = arguments_.options; - - return this.#app - .get(Identifiers.P2P.ApiNode.Processor) - .validateAndAcceptApiNode(apiNode, options); - } -} diff --git a/packages/p2p/source/actions/validate-and-accept-peer.test.ts b/packages/p2p/source/actions/validate-and-accept-peer.test.ts deleted file mode 100644 index eb41d13308..0000000000 --- a/packages/p2p/source/actions/validate-and-accept-peer.test.ts +++ /dev/null @@ -1,31 +0,0 @@ -import { Identifiers } from "@mainsail/constants"; - -import { Application } from "@mainsail/kernel"; -import { describe } from "@mainsail/test-runner"; -import { ValidateAndAcceptPeerAction } from "./validate-and-accept-peer"; - -describe<{ - app: Application; - action: ValidateAndAcceptPeerAction; -}>("ValidateAndAcceptPeerAction", ({ it, spy, beforeEach }) => { - const peerProcessor = { validateAndAcceptPeer: () => {} }; - - beforeEach((context) => { - context.app = new Application(); - - context.app.bind(Identifiers.P2P.Peer.Processor).toConstantValue(peerProcessor); - - context.action = new ValidateAndAcceptPeerAction(context.app); - }); - - it("#execute - should call peerProcessor.validateAndAcceptPeer with arguments provided", async ({ action }) => { - const spyValidateAndAcceptPeer = spy(peerProcessor, "validateAndAcceptPeer"); - - const options = { someParam: 1 }; - - await action.execute({ ip: "187.165.33.2", options }); - - spyValidateAndAcceptPeer.calledOnce(); - spyValidateAndAcceptPeer.calledWith("187.165.33.2", options); - }); -}); diff --git a/packages/p2p/source/actions/validate-and-accept-peer.ts b/packages/p2p/source/actions/validate-and-accept-peer.ts deleted file mode 100644 index 56737f2729..0000000000 --- a/packages/p2p/source/actions/validate-and-accept-peer.ts +++ /dev/null @@ -1,24 +0,0 @@ -import type { Contracts } from "@mainsail/contracts"; - -import { Identifiers } from "@mainsail/constants"; -import { injectable } from "@mainsail/container"; -import { Services } from "@mainsail/kernel"; - -import { PeerProcessor } from "../peer-processor.js"; - -@injectable() -export class ValidateAndAcceptPeerAction extends Services.Triggers.Action { - #app: Contracts.Kernel.Application; - - public constructor(app: Contracts.Kernel.Application) { - super(); - this.#app = app; - } - - public async execute(arguments_: { ip: string; options: Contracts.P2P.AcceptNewPeerOptions }): Promise { - const ip = arguments_.ip; - const options = arguments_.options; - - return this.#app.get(Identifiers.P2P.Peer.Processor).validateAndAcceptPeer(ip, options); - } -} diff --git a/packages/p2p/source/api-node-discoverer.test.ts b/packages/p2p/source/api-node-discoverer.test.ts new file mode 100644 index 0000000000..bec871818c --- /dev/null +++ b/packages/p2p/source/api-node-discoverer.test.ts @@ -0,0 +1,82 @@ +import { Identifiers } from "@mainsail/constants"; +import { Application } from "@mainsail/kernel"; +import { describe } from "@mainsail/test-runner"; +import dayjs from "dayjs"; + +import { ApiNodeDiscoverer } from "./api-node-discoverer"; + +describe<{ + discoverer: ApiNodeDiscoverer; + apiNodeRepository: any; + apiNodeProcessor: any; + communicator: any; +}>("ApiNodeDiscoverer", ({ it, beforeEach, spy }) => { + const peer = { ip: "1.2.3.4" }; + + beforeEach((context) => { + context.apiNodeRepository = { getApiNodes: () => [] }; + context.apiNodeProcessor = { revalidateApiNode: async () => {}, validateAndAcceptApiNode: async () => {} }; + context.communicator = { getApiNodes: async () => ({ apiNodes: [] }) }; + + const app = new Application(); + app.bind(Identifiers.ServiceProvider.Configuration) + .toConstantValue({ getOptional: (key: string, defaultValue: unknown) => defaultValue }) + .whenTagged("plugin", "p2p"); + app.bind(Identifiers.P2P.ApiNode.Factory).toConstantValue((url: string) => ({ url })); + app.bind(Identifiers.P2P.ApiNode.Processor).toConstantValue(context.apiNodeProcessor); + app.bind(Identifiers.P2P.Peer.Communicator).toConstantValue(context.communicator); + app.bind(Identifiers.P2P.ApiNode.Repository).toConstantValue(context.apiNodeRepository); + app.bind(Identifiers.P2P.Peer.Repository).toConstantValue({ getPeers: () => [peer] }); + app.bind(Identifiers.Services.Log.Service).toConstantValue({ debug: () => {} }); + + context.discoverer = app.resolve(ApiNodeDiscoverer); + }); + + it("#discoverApiNodes - should hand every advertised node to the processor", async ({ + discoverer, + communicator, + apiNodeProcessor, + }) => { + const apiNodes = [{ url: "http://1.2.3.4:4003/" }, { url: "http://5.6.7.8:4003/" }]; + communicator.getApiNodes = async () => ({ apiNodes }); + const validateAndAcceptApiNode = spy(apiNodeProcessor, "validateAndAcceptApiNode"); + + await discoverer.discoverApiNodes(peer); + + validateAndAcceptApiNode.calledTimes(2); + validateAndAcceptApiNode.calledWith(apiNodes[0], {}); + }); + + it("#discoverApiNodes - should swallow a failed request", async ({ + discoverer, + communicator, + apiNodeProcessor, + }) => { + communicator.getApiNodes = async () => { + throw new Error("boom"); + }; + const validateAndAcceptApiNode = spy(apiNodeProcessor, "validateAndAcceptApiNode"); + + await discoverer.discoverApiNodes(peer); + + validateAndAcceptApiNode.neverCalled(); + }); + + it("#refreshApiNodes - should revalidate nodes that were never pinged or were pinged long ago", async ({ + discoverer, + apiNodeRepository, + apiNodeProcessor, + }) => { + const neverPinged = { url: "http://1.1.1.1:4003/" }; + const pingedLongAgo = { lastPinged: dayjs().subtract(11, "minutes"), url: "http://2.2.2.2:4003/" }; + const pingedRecently = { lastPinged: dayjs().subtract(1, "minute"), url: "http://3.3.3.3:4003/" }; + apiNodeRepository.getApiNodes = () => [neverPinged, pingedLongAgo, pingedRecently]; + const revalidateApiNode = spy(apiNodeProcessor, "revalidateApiNode"); + + await discoverer.refreshApiNodes(); + + revalidateApiNode.calledTimes(2); + revalidateApiNode.calledWith(neverPinged); + revalidateApiNode.calledWith(pingedLongAgo); + }); +}); diff --git a/packages/p2p/source/api-node-discoverer.ts b/packages/p2p/source/api-node-discoverer.ts index 118afc3bf8..a51e69b9d5 100644 --- a/packages/p2p/source/api-node-discoverer.ts +++ b/packages/p2p/source/api-node-discoverer.ts @@ -2,7 +2,6 @@ import type { Contracts } from "@mainsail/contracts"; import { Identifiers } from "@mainsail/constants"; import { inject, injectable, tagged } from "@mainsail/container"; -import { Services } from "@mainsail/kernel"; import { ensureError, shuffle } from "@mainsail/utils"; import dayjs from "dayjs"; @@ -10,9 +9,6 @@ import { normalizeUrl } from "./utils/index.js"; @injectable() export class ApiNodeDiscoverer implements Contracts.P2P.ApiNodeDiscoverer { - @inject(Identifiers.Application.Instance) - private readonly app!: Contracts.Kernel.Application; - @inject(Identifiers.ServiceProvider.Configuration) @tagged("plugin", "p2p") private readonly configuration!: Contracts.Kernel.PluginConfiguration; @@ -20,6 +16,9 @@ export class ApiNodeDiscoverer implements Contracts.P2P.ApiNodeDiscoverer { @inject(Identifiers.P2P.ApiNode.Factory) private readonly ApiNodeFactory!: Contracts.P2P.ApiNodeFactory; + @inject(Identifiers.P2P.ApiNode.Processor) + private readonly apiNodeProcessor!: Contracts.P2P.ApiNodeProcessor; + @inject(Identifiers.P2P.Peer.Communicator) private readonly communicator!: Contracts.P2P.PeerCommunicator; @@ -37,10 +36,7 @@ export class ApiNodeDiscoverer implements Contracts.P2P.ApiNodeDiscoverer { const { apiNodes } = await this.communicator.getApiNodes(peer); for (const apiNode of apiNodes) { - await this.app.get(Identifiers.Services.Trigger.Service).call<{ - apiNode: Contracts.P2P.ApiNode; - options: Contracts.P2P.AcceptNewPeerOptions; - }>("validateAndAcceptApiNode", { apiNode, options: {} }); + await this.apiNodeProcessor.validateAndAcceptApiNode(apiNode, {}); } } catch (rawError) { const error = ensureError(rawError); @@ -56,10 +52,7 @@ export class ApiNodeDiscoverer implements Contracts.P2P.ApiNodeDiscoverer { await Promise.all( Object.values(apiNodes).map((apiNode: Contracts.P2P.ApiNode) => - this.app.get(Identifiers.Services.Trigger.Service).call<{ - apiNode: Contracts.P2P.ApiNode; - options: Contracts.P2P.AcceptNewPeerOptions; - }>("validateAndAcceptApiNode", { apiNode, options: { seed: true } }), + this.apiNodeProcessor.validateAndAcceptApiNode(apiNode, { seed: true }), ), ); } @@ -73,15 +66,11 @@ export class ApiNodeDiscoverer implements Contracts.P2P.ApiNodeDiscoverer { await Promise.all( this.apiNodeRepository .getApiNodes() - .filter((apiNode) => + .filter( // ignore nodes that were pinged recently - (apiNode.lastPinged ?? dayjs()).isAfter(dayjs().add(10, "minutes")), + (apiNode) => !apiNode.lastPinged || apiNode.lastPinged.isBefore(dayjs().subtract(10, "minutes")), ) - .map((apiNode) => - this.app.get(Identifiers.Services.Trigger.Service).call<{ - apiNode: Contracts.P2P.ApiNode; - }>("revalidateApiNode", { apiNode }), - ), + .map((apiNode) => this.apiNodeProcessor.revalidateApiNode(apiNode)), ); } } diff --git a/packages/p2p/source/api-node-processor.test.ts b/packages/p2p/source/api-node-processor.test.ts new file mode 100644 index 0000000000..e728872787 --- /dev/null +++ b/packages/p2p/source/api-node-processor.test.ts @@ -0,0 +1,109 @@ +import { Events, Identifiers } from "@mainsail/constants"; +import { Application } from "@mainsail/kernel"; +import { describe } from "@mainsail/test-runner"; +import dayjs from "dayjs"; + +import { ApiNodeProcessor } from "./api-node-processor"; + +describe<{ + processor: ApiNodeProcessor; + repository: any; + apiNodeVerifier: any; + events: any; +}>("ApiNodeProcessor", ({ it, beforeEach, spy }) => { + const apiNode = { url: "http://1.2.3.4:4003/" } as any; + + beforeEach((context) => { + context.repository = { + forgetApiNode: () => {}, + forgetPendingApiNode: () => {}, + hasApiNode: () => true, + hasPendingApiNode: () => false, + setApiNode: () => {}, + setPendingApiNode: () => {}, + }; + context.apiNodeVerifier = { verify: async () => true }; + context.events = { dispatch: async () => {} }; + + const app = new Application(); + app.bind(Identifiers.P2P.ApiNode.Repository).toConstantValue(context.repository); + app.bind(Identifiers.P2P.ApiNode.Verifier).toConstantValue(context.apiNodeVerifier); + app.bind(Identifiers.Services.EventDispatcher.Service).toConstantValue(context.events); + app.bind(Identifiers.P2P.Logger).toConstantValue({ debugExtra: () => {} }); + + context.processor = app.resolve(ApiNodeProcessor); + + delete apiNode.lastPinged; + }); + + it("#validateAndAcceptApiNode - should accept a node that verifies", async ({ processor, repository, events }) => { + repository.hasApiNode = () => false; + const setApiNode = spy(repository, "setApiNode"); + const forgetPendingApiNode = spy(repository, "forgetPendingApiNode"); + const dispatch = spy(events, "dispatch"); + + await processor.validateAndAcceptApiNode(apiNode); + + setApiNode.calledWith(apiNode); + dispatch.calledWith(Events.ApiNodeEvent.Added, apiNode); + forgetPendingApiNode.calledWith(apiNode); + }); + + it("#revalidateApiNode - should verify a node that was never pinged", async ({ processor, apiNodeVerifier }) => { + const verify = spy(apiNodeVerifier, "verify"); + + await processor.revalidateApiNode(apiNode); + + verify.calledWith(apiNode); + }); + + it("#revalidateApiNode - should verify a node whose last ping is older than ten minutes", async ({ + processor, + apiNodeVerifier, + }) => { + apiNode.lastPinged = dayjs().subtract(11, "minutes"); + const verify = spy(apiNodeVerifier, "verify"); + + await processor.revalidateApiNode(apiNode); + + verify.calledOnce(); + }); + + it("#revalidateApiNode - should skip a node that was pinged recently", async ({ processor, apiNodeVerifier }) => { + apiNode.lastPinged = dayjs().subtract(1, "minute"); + const verify = spy(apiNodeVerifier, "verify"); + + await processor.revalidateApiNode(apiNode); + + verify.neverCalled(); + }); + + it("#revalidateApiNode - should remove a node that fails verification", async ({ + processor, + repository, + apiNodeVerifier, + events, + }) => { + apiNodeVerifier.verify = async () => false; + const forgetApiNode = spy(repository, "forgetApiNode"); + const dispatch = spy(events, "dispatch"); + + await processor.revalidateApiNode(apiNode); + + forgetApiNode.calledWith(apiNode); + dispatch.calledWith(Events.ApiNodeEvent.Removed, apiNode); + }); + + it("#revalidateApiNode - should ignore a node that is not in the repository", async ({ + processor, + repository, + apiNodeVerifier, + }) => { + repository.hasApiNode = () => false; + const verify = spy(apiNodeVerifier, "verify"); + + await processor.revalidateApiNode(apiNode); + + verify.neverCalled(); + }); +}); diff --git a/packages/p2p/source/api-node-processor.ts b/packages/p2p/source/api-node-processor.ts index 6f6484da0b..2790165aac 100644 --- a/packages/p2p/source/api-node-processor.ts +++ b/packages/p2p/source/api-node-processor.ts @@ -44,8 +44,7 @@ export class ApiNodeProcessor implements Contracts.P2P.ApiNodeProcessor { return; } - const lastPinged = apiNode.lastPinged ?? dayjs(); - if (lastPinged.isBefore(dayjs().subtract(10, "minutes"))) { + if (apiNode.lastPinged && apiNode.lastPinged.isAfter(dayjs().subtract(10, "minutes"))) { return; } diff --git a/packages/p2p/source/api-node-verifier.test.ts b/packages/p2p/source/api-node-verifier.test.ts new file mode 100644 index 0000000000..0993381fb2 --- /dev/null +++ b/packages/p2p/source/api-node-verifier.test.ts @@ -0,0 +1,92 @@ +import { Identifiers } from "@mainsail/constants"; +import { Application } from "@mainsail/kernel"; +import { describe } from "@mainsail/test-runner"; +import esmock from "esmock"; + +import { ApiNodeVerifier } from "./api-node-verifier"; + +let httpGet: (url: string, options: unknown) => Promise; + +const { ApiNodeVerifier: ApiNodeVerifierProxy } = await esmock( + "./api-node-verifier", + import.meta.url, + { "@mainsail/utils": { http: { get: (url: string, options: unknown) => httpGet(url, options) } } }, + {}, +); + +describe<{ + verifier: ApiNodeVerifier; +}>("ApiNodeVerifier", ({ it, assert, beforeEach }) => { + const helloWorld = { data: "Hello World from Public API!" }; + const contentLength = String(JSON.stringify(helloWorld).length); + const apiNode = () => ({ url: "http://1.2.3.4:4003/" }) as any; + + const respond = (statusCode: number, data: unknown, headers: string[]) => { + httpGet = async () => ({ data, headers, method: "GET", statusCode, statusMessage: "" }); + }; + + beforeEach((context) => { + respond(200, helloWorld, ["content-length", contentLength]); + + const app = new Application(); + app.bind(Identifiers.P2P.Logger).toConstantValue({ debugExtra: () => {} }); + app.bind(Identifiers.ServiceProvider.Configuration) + .toConstantValue({ getRequired: () => 1024 }) + .whenTagged("plugin", "p2p"); + + context.verifier = app.resolve(ApiNodeVerifierProxy); + }); + + it("#verify - should accept the default api response", async ({ verifier }) => { + const node = apiNode(); + + assert.true(await verifier.verify(node)); + assert.equal(node.statusCode, 200); + assert.defined(node.lastPinged); + }); + + it("#verify - should match the content-length header regardless of its case", async ({ verifier }) => { + respond(200, helloWorld, ["Content-Type", "application/json", "Content-Length", contentLength]); + + assert.true(await verifier.verify(apiNode())); + }); + + it("#verify - should only match header names, not values", async ({ verifier }) => { + respond(200, helloWorld, ["x-echo", "content-length", "content-length", contentLength]); + + assert.true(await verifier.verify(apiNode())); + }); + + it("#verify - should reject a missing content length", async ({ verifier }) => { + respond(200, helloWorld, ["content-type", "application/json"]); + + assert.false(await verifier.verify(apiNode())); + }); + + it("#verify - should reject a wrong content length", async ({ verifier }) => { + respond(200, helloWorld, ["content-length", "1"]); + + assert.false(await verifier.verify(apiNode())); + }); + + it("#verify - should reject a wrong body", async ({ verifier }) => { + const body = { data: "Hello World from Nowhere!!!!" }; + respond(200, body, ["content-length", String(JSON.stringify(body).length)]); + + assert.false(await verifier.verify(apiNode())); + }); + + it("#verify - should reject a non-200 status", async ({ verifier }) => { + respond(201, helloWorld, ["content-length", contentLength]); + + assert.false(await verifier.verify(apiNode())); + }); + + it("#verify - should reject an unreachable node", async ({ verifier }) => { + httpGet = async () => { + throw new Error("connection refused"); + }; + + assert.false(await verifier.verify(apiNode())); + }); +}); diff --git a/packages/p2p/source/api-node-verifier.ts b/packages/p2p/source/api-node-verifier.ts index 4365636ddd..2fc4e66c2b 100644 --- a/packages/p2p/source/api-node-verifier.ts +++ b/packages/p2p/source/api-node-verifier.ts @@ -55,7 +55,14 @@ export class ApiNodeVerifier implements Contracts.P2P.ApiNodeVerifier { } #verifyHeaders(response: HttpResponse): void { - const contentLength = Number(response.headers[response.headers.indexOf("content-length") + 1]); + const index = response.headers.findIndex( + (header, index) => index % 2 === 0 && header.toLowerCase() === "content-length", + ); + if (index === -1) { + throw new Error("missing content length"); + } + + const contentLength = Number(response.headers[index + 1]); if (contentLength !== helloWorldLength) { throw new Error("invalid content length"); } diff --git a/packages/p2p/source/constants.ts b/packages/p2p/source/constants.ts index 79d539c8b8..46fea4af77 100644 --- a/packages/p2p/source/constants.ts +++ b/packages/p2p/source/constants.ts @@ -10,6 +10,8 @@ export const constants = { MAX_DOWNLOAD_BLOCKS_JOBS: 10, // maximum number of blocks jobs MAX_PAYLOAD_CLIENT: 5 * Units.MEGABYTE, // max size of a complete serialized response frame the WS client accepts + // A serialized prevote/precommit is ~140 bytes and the request headers ~200 bytes; 1 KB leaves ample room. + MAX_PAYLOAD_POST_MESSAGE: Units.KILOBYTE, MAX_PAYLOAD_SERVER: 3 * Units.MEGABYTE, MAX_PEERS_GET_API_NODES: 100, diff --git a/packages/p2p/source/downloader/block-downloader.test.ts b/packages/p2p/source/downloader/block-downloader.test.ts index 15fc0fbb54..5e65bf08f8 100644 --- a/packages/p2p/source/downloader/block-downloader.test.ts +++ b/packages/p2p/source/downloader/block-downloader.test.ts @@ -412,6 +412,50 @@ describe<{ assert.true(downloader.isDownloading()); }); + it("#download - should process a ready successor after dropping a replayed job that is already applied", async ({ + downloader, + }) => { + // Blocks above 255 need two bytes. + const makeWideBlock = (number: number): Buffer => { + const buffer = Buffer.alloc(2); + buffer.writeUInt16BE(number, 0); + return buffer; + }; + stub(commitFactory, "fromBytes").callsFake(async (buffer: Buffer) => { + const number = buffer.readUInt16BE(0); + return { block: { hash: `hash${number}`, number } }; + }); + // Blocks 401-402 form their own round, so the second job is processed as one slice. + stub(roundCalculator, "calculateRound").callsFake((number: number) => ({ + maxValidators: 2, + roundHeight: number <= 400 ? 1 : 401, + })); + const process = stub(commitProcessor, "process").returnValue(Enums.Consensus.ProcessorResult.Accepted); + + const requests: { resolve: any; reject: any }[] = []; + stub(communicator, "getBlocks").callsFake( + () => new Promise((resolve, reject) => requests.push({ reject, resolve })), + ); + stub(repository, "getPeers").returnValue([makePeer(1000, "2.2.2.2")]); + + // Jobs for blocks 1-400 and 401-402. The second one arrives first and waits for the first. + downloader.download(makePeer(401)); + downloader.download(makePeer(403)); + requests[1].resolve({ blocks: [makeWideBlock(401), makeWideBlock(402)] }); + await sleep(10); + process.neverCalled(); + + // Consensus applies blocks 1-400 meanwhile, then the first request fails: its replay has + // nothing left to fetch and must hand over to the successor that is already ready. + stub(stateStore, "getBlockNumber").returnValue(400); + stub(stateStore, "getLastBlock").returnValue({ hash: "hash400" }); + requests[0].reject(new Error("boom")); + await sleep(10); + + process.calledTimes(2); + assert.false(downloader.isDownloading()); + }); + it("#download - should not ban when the short reply was near the payload limit", async ({ downloader }) => { // 3 MiB of block data: adding another maxPayload (2 MiB) block could exceed the 5 MiB // response limit, so the short reply is legitimate. diff --git a/packages/p2p/source/downloader/block-downloader.ts b/packages/p2p/source/downloader/block-downloader.ts index 42a7c2effd..496916d687 100644 --- a/packages/p2p/source/downloader/block-downloader.ts +++ b/packages/p2p/source/downloader/block-downloader.ts @@ -288,6 +288,7 @@ export class BlockDownloader implements Contracts.P2P.Downloader { // Skip if blockNumberFrom is higher than blockNumberTo if (isFirstJob && blockNumberFrom > blockNumberTo) { this.#downloadJobs.shift(); + this.#processNextJob(); return; } diff --git a/packages/p2p/source/peer-communicator.ts b/packages/p2p/source/peer-communicator.ts index f1b9546d16..d5e9ccc468 100644 --- a/packages/p2p/source/peer-communicator.ts +++ b/packages/p2p/source/peer-communicator.ts @@ -2,7 +2,7 @@ import type { Contracts } from "@mainsail/contracts"; import { Identifiers } from "@mainsail/constants"; import { inject, injectable, tagged } from "@mainsail/container"; -import { assert, ensureError, http } from "@mainsail/utils"; +import { assert, ensureError, http, IpAddress } from "@mainsail/utils"; import { performance } from "perf_hooks"; import { constants } from "./constants.js"; @@ -13,9 +13,6 @@ import { Throttle } from "./throttle.js"; @injectable() export class PeerCommunicator implements Contracts.P2P.PeerCommunicator { - @inject(Identifiers.Application.Instance) - private readonly app!: Contracts.Kernel.Application; - @inject(Identifiers.ServiceProvider.Configuration) @tagged("plugin", "p2p") private readonly configuration!: Contracts.Kernel.PluginConfiguration; @@ -23,6 +20,9 @@ export class PeerCommunicator implements Contracts.P2P.PeerCommunicator { @inject(Identifiers.P2P.Peer.Connector) private readonly connector!: Contracts.P2P.PeerConnector; + @inject(Identifiers.P2P.Peer.Disposer) + private readonly peerDisposer!: Contracts.P2P.PeerDisposer; + @inject(Identifiers.P2P.Header.Factory) private readonly headerFactory!: Contracts.P2P.HeaderFactory; @@ -41,14 +41,13 @@ export class PeerCommunicator implements Contracts.P2P.PeerCommunicator { @inject(Identifiers.P2P.Statistic.Service) private readonly statisticService!: Contracts.P2P.StatisticService; - #throttle?: Throttle; + #throttle?: Promise; public async postProposal(peer: Contracts.P2P.Peer, proposal: Buffer): Promise { try { await this.#emit(peer, Routes.PostProposal, { proposal }, { timeout: 6000 }); } catch (rawError) { - const error = ensureError(rawError); - this.#handleSocketError(peer, error); + this.peerDisposer.banPeer(peer.ip, ensureError(rawError)); } } @@ -56,8 +55,7 @@ export class PeerCommunicator implements Contracts.P2P.PeerCommunicator { try { await this.#emit(peer, Routes.PostMessage, { message }, { timeout: 6000 }); } catch (rawError) { - const error = ensureError(rawError); - this.#handleSocketError(peer, error); + this.peerDisposer.banPeer(peer.ip, ensureError(rawError)); } } @@ -66,7 +64,9 @@ export class PeerCommunicator implements Contracts.P2P.PeerCommunicator { Object.entries(peer.plugins).map(async ([name, plugin]) => { peer.ports[name] = -1; try { - const { statusCode } = await http.head(`http://${peer.ip}:${plugin.port}/`); + const { statusCode } = await http.head( + `http://${IpAddress.normalizeAddress(peer.ip)}:${plugin.port}/`, + ); if (statusCode === 200) { peer.ports[name] = plugin.port; @@ -163,24 +163,17 @@ export class PeerCommunicator implements Contracts.P2P.PeerCommunicator { return result.data; } - #validateReply(peer: Contracts.P2P.Peer, reply: T, endpoint: string) { + #validateReply(reply: T, endpoint: string): string | undefined { const schema = replySchemas[endpoint]; if (schema === undefined) { this.logger.error( `Can't validate reply from "${endpoint}": none of the predefined schemas matches.`, "p2p", ); - return false; + return "no reply schema"; } - const { error } = this.validator.validate(schema, reply); - if (error) { - this.logger.debugExtra(`Got unexpected reply from ${peer.url}/${endpoint}: ${error}`, "p2p"); - - return false; - } - - return true; + return this.validator.validate(schema, reply).error; } async #emit( @@ -210,18 +203,10 @@ export class PeerCommunicator implements Contracts.P2P.PeerCommunicator { const timeBeforeSocketCall = performance.now(); - await this.connector.connect(peer); - const response = await this.connector.emit( peer, event, - codec.request.serialize({ - ...payload, - // @ts-ignore - headers: { - ...this.headerFactory().toData(), - }, - }), + codec.request.serialize({ ...payload, headers: this.headerFactory().toData() }), options.timeout, ); statistic.responseTime = Math.round(performance.now() - timeBeforeSocketCall); @@ -236,9 +221,10 @@ export class PeerCommunicator implements Contracts.P2P.PeerCommunicator { // Validate peer.setPinged(Math.floor(statistic.responseTime + statistic.deserializeTime)); - if (!this.#validateReply(peer, data, event)) { + const replyError = this.#validateReply(data, event); + if (replyError) { const validationError = new Error( - `Response validation failed for ${event} from peer ${peer.ip}: ${JSON.stringify(data)}`, + `Response validation failed for ${event} from peer ${peer.ip}: ${replyError}`, ); validationError.name = SocketErrors.Validation; @@ -260,14 +246,8 @@ export class PeerCommunicator implements Contracts.P2P.PeerCommunicator { } } - #handleSocketError(peer: Contracts.P2P.Peer, error: Error): void { - this.app.get(Identifiers.P2P.Peer.Disposer).banPeer(peer.ip, error); - } - - async #getThrottle(): Promise { - if (!this.#throttle) { - this.#throttle = await this.throttleFactory(); - } + #getThrottle(): Promise { + this.#throttle ??= this.throttleFactory(); return this.#throttle; } diff --git a/packages/p2p/source/peer-connector.test.ts b/packages/p2p/source/peer-connector.test.ts index 7b334b31a8..f6cae17fd0 100644 --- a/packages/p2p/source/peer-connector.test.ts +++ b/packages/p2p/source/peer-connector.test.ts @@ -1,228 +1,224 @@ +import type { Contracts } from "@mainsail/contracts"; + import { Identifiers } from "@mainsail/constants"; +import { Application } from "@mainsail/kernel"; +import { describe } from "@mainsail/test-runner"; import esmock from "esmock"; -import { Application } from "@mainsail/kernel"; -import { describeSkip } from "@mainsail/test-runner"; -import { Peer } from "./peer"; import { PeerConnector } from "./peer-connector"; -let onDelay = (timeout: number) => {}; +type Deferred = { promise: Promise; resolve: () => void; reject: (error: Error) => void }; + +const deferred = (): Deferred => { + let resolve!: () => void; + let reject!: (error: Error) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + + return { promise, reject, resolve }; +}; class ClientMock { - static onConstructor = (...arguments_) => {}; + static instances: ClientMock[] = []; + static connectImpl: (client: ClientMock) => Promise = async () => {}; + + public onDisconnect: (...arguments_: unknown[]) => void = () => {}; + public onError: (error: Error) => void = () => {}; + public connectOptions: unknown; + public timeout: number | undefined; + public terminated = 0; + + public constructor( + public readonly url: string, + public readonly options: unknown, + ) { + ClientMock.instances.push(this); + } + + public connect(options: unknown): Promise { + this.connectOptions = options; + return ClientMock.connectImpl(this); + } - constructor(...arguments_) { - ClientMock.onConstructor(...arguments_); + public async request(): Promise<{ payload: Buffer }> { + return { payload: Buffer.from("response") }; + } + + public setTimeout(timeout: number): void { + this.timeout = timeout; + } + + public async terminate(): Promise { + this.terminated++; } - async connect() {} - async request() {} - async terminate() {} } +let delayCalls: number[] = []; + const { PeerConnector: PeerConnectorProxy } = await esmock("./peer-connector", { "./hapi-nes": { Client: ClientMock, }, delay: async (timeout: number) => { - onDelay(timeout); + delayCalls.push(timeout); }, }); -describeSkip<{ +describe<{ app: Application; peerConnector: PeerConnector; -}>("PeerConnector", ({ it, assert, beforeEach, stub, spy, spyFn }) => { + peerDisposer: { banPeer: (ip: string, error: Error) => void; disposePeer: (ip: string) => void }; +}>("PeerConnector", ({ it, assert, beforeEach, spy }) => { const logger = { debug: () => {}, error: () => {}, info: () => {}, warn: () => {} }; + const peer = { ip: "178.165.55.11", port: 4000 } as Contracts.P2P.Peer; + beforeEach((context) => { - onDelay = () => {}; - ClientMock.onConstructor = () => {}; + ClientMock.instances = []; + ClientMock.connectImpl = async () => {}; + delayCalls = []; - context.app = new Application(); + context.peerDisposer = { banPeer: () => {}, disposePeer: () => {} }; - context.app.bind(Identifiers.Services.Log.Service).toConstantValue(logger); + context.app = new Application(); + context.app.bind(Identifiers.P2P.Logger).toConstantValue(logger); + context.app.bind(Identifiers.P2P.Peer.Disposer).toConstantValue(context.peerDisposer); context.peerConnector = context.app.resolve(PeerConnectorProxy); }); - it("#all - should return a empty array when there are no connections", ({ peerConnector }) => { - assert.length(peerConnector.all(), 0); - }); - - it("#all - should return the connections", async ({ peerConnector }) => { - const peers = [new Peer("178.165.55.44", 4000), new Peer("178.165.55.33", 4000)]; - await peerConnector.connect(peers[0]); - await peerConnector.connect(peers[1]); - - assert.length(peerConnector.all(), 2); - }); - - it("#connection - should return the connection", async ({ peerConnector }) => { - const peers = [ - new Peer("178.165.55.44", 4000), - new Peer("178.165.55.33", 4000), - new Peer("2001:3984:3989::104", 4000), - ]; - await peerConnector.connect(peers[0]); - await peerConnector.connect(peers[1]); - await peerConnector.connect(peers[2]); - - assert.instance(peerConnector.connection(peers[0]), ClientMock); - assert.instance(peerConnector.connection(peers[1]), ClientMock); - assert.instance(peerConnector.connection(peers[2]), ClientMock); - }); - - it("#connection - should return undefined if there is no connection", async ({ peerConnector }) => { - const peerNotAdded = new Peer("178.0.0.0", 4000); - assert.undefined(peerConnector.connection(peerNotAdded)); - }); - - it("#connect - should set the connection in the connections and return it", async ({ peerConnector }) => { - const spyClientConstructor = spyFn(); - ClientMock.onConstructor = (...arguments_) => spyClientConstructor.call(...arguments_); + it("#connect - should create a client, connect it without reconnect and reuse it", async ({ peerConnector }) => { + const connection = await peerConnector.connect(peer); - const peer = new Peer("178.165.55.11", 4000); - const peerConnection = await peerConnector.connect(peer); + assert.length(ClientMock.instances, 1); + assert.equal(ClientMock.instances[0].url, "ws://178.165.55.11:4000"); + assert.equal(ClientMock.instances[0].options, { timeout: 10_000 }); + assert.equal(ClientMock.instances[0].connectOptions, { reconnect: false, timeout: 10_000 }); - spyClientConstructor.calledOnce(); - spyClientConstructor.calledWith("ws://178.165.55.11:4000", { timeout: 10_000 }); - assert.instance(peerConnection, ClientMock); + assert.equal(await peerConnector.connect(peer), connection); + assert.length(ClientMock.instances, 1); }); - it("#connect - should set the connection with brackets IPv6", async ({ peerConnector }) => { - const spyClientConstructor = spyFn(); - ClientMock.onConstructor = (...arguments_) => spyClientConstructor.call(...arguments_); + it("#connect - should bracket IPv6 addresses", async ({ peerConnector }) => { + await peerConnector.connect({ ip: "2001:3984:3989::104", port: 4000 } as Contracts.P2P.Peer); - const peer = new Peer("2001:3984:3989::104", 4000); - const peerConnection = await peerConnector.connect(peer); - - spyClientConstructor.calledOnce(); - spyClientConstructor.calledWith("ws://[2001:3984:3989::104]:4000", { timeout: 10_000 }); - assert.instance(peerConnection, ClientMock); + assert.equal(ClientMock.instances[0].url, "ws://[2001:3984:3989::104]:4000"); }); - it.skip("#connect - should log and remove if error on connection", async ({ peerConnector }) => { - const spyLoggerDebug = spy(logger, "debug"); + it("#connect - should share one in-flight creation between concurrent callers", async ({ peerConnector }) => { + const connecting = deferred(); + ClientMock.connectImpl = () => connecting.promise; - const peer = new Peer("178.165.55.11", 4000); - const peerConnection = await peerConnector.connect(peer); + const first = peerConnector.connect(peer); + const second = peerConnector.connect(peer); - peerConnection.onError(new Error("dummy")); + connecting.resolve(); - spyLoggerDebug.calledOnce(); - assert.instance(peerConnection, ClientMock); + assert.equal(await first, await second); + assert.length(ClientMock.instances, 1); }); - it.skip("#connect - should delay connection create if re-connecting within 10 seconds", async ({ - peerConnector, - }) => { - const spyDelay = spyFn(); - onDelay = (timeout) => { - spyDelay.call(timeout); + it("#connect - should not keep a client whose connect failed", async ({ peerConnector }) => { + ClientMock.connectImpl = async () => { + throw new Error("refused"); }; - const peer = new Peer("178.165.55.11", 4000); - await peerConnector.connect(peer); - peerConnector.disconnect(peer); - await peerConnector.connect(peer); - - spyDelay.calledOnce(); - assert.gte(spyDelay.getCallArgs(0)[0], 9000); - }); + await assert.rejects(() => peerConnector.connect(peer), "refused"); - it.skip("#disconnect - should call terminate on the connection and forget it", async ({ peerConnector }) => { - const peer = new Peer("178.165.55.11", 4000); - const peerConnection = await peerConnector.connect(peer); - const spyTerminate = spy(peerConnection, "terminate"); + assert.length(ClientMock.instances, 1); - assert.instance(peerConnector.connection(peer), ClientMock); + ClientMock.connectImpl = async () => {}; + const connection = await peerConnector.connect(peer); - peerConnector.disconnect(peer); - assert.undefined(peerConnector.connection(peer)); - spyTerminate.calledOnce(); + assert.length(ClientMock.instances, 2); + assert.equal(connection, ClientMock.instances[1]); }); - it("#disconnect - should not do anything if the peer is not defined", async ({ peerConnector }) => { - const peer = new Peer("178.165.0.0", 4000); - - assert.undefined(peerConnector.connection(peer)); + it("#connect - should delay re-creation within ten seconds of the previous one", async ({ peerConnector }) => { + await peerConnector.connect(peer); + await peerConnector.disconnect(peer.ip); + await peerConnector.connect(peer); - peerConnector.disconnect(peer); - assert.undefined(peerConnector.connection(peer)); + assert.length(delayCalls, 2); + assert.equal(delayCalls[0], 0); + assert.gte(delayCalls[1], 9000); + assert.length(ClientMock.instances, 2); }); - it("#emit - should connect to the peer and call connection.request", async ({ peerConnector }) => { - const peer = new Peer("178.165.11.12", 4000); - - const peerConnection = await peerConnector.connect(peer); - - const mockResponse = { payload: "mock payload" }; - const spyRequest = stub(peerConnection, "request").returnValue(mockResponse); + it("#disconnect - should terminate and forget the connection", async ({ peerConnector }) => { + const connection = await peerConnector.connect(peer); - const response = await peerConnector.emit(peer, "p2p.peer.getStatus", {}); + await peerConnector.disconnect(peer.ip); - spyRequest.calledOnce(); - assert.equal(response, mockResponse); + assert.equal(ClientMock.instances[0].terminated, 1); + assert.not.equal(await peerConnector.connect(peer), connection); }); - it("#getError - should return the error set for the peer", ({ peerConnector }) => { - const peer = new Peer("178.165.11.12", 4000); + it("#disconnect - should terminate a connection that is still being created", async ({ peerConnector }) => { + const connecting = deferred(); + ClientMock.connectImpl = () => connecting.promise; - const peerError = `some random error for the peer ${peer.ip}`; - peerConnector.setError(peer, peerError); + const connectPromise = peerConnector.connect(peer); + const disconnectPromise = peerConnector.disconnect(peer.ip); - assert.equal(peerConnector.getError(peer), peerError); - }); - - it("#getError - should return undefined when the peer has no error set", ({ peerConnector }) => { - const peer = new Peer("178.165.11.12", 4000); + connecting.resolve(); + const connection = await connectPromise; + await disconnectPromise; - assert.undefined(peerConnector.getError(peer)); + assert.equal(connection.terminated, 1); + assert.not.equal(await peerConnector.connect(peer), connection); + assert.length(ClientMock.instances, 2); }); - it("#setError - should set the error for the peer", ({ peerConnector }) => { - const peer = new Peer("178.165.11.12", 4000); + it("#disconnect - should ignore a creation that fails", async ({ peerConnector }) => { + const connecting = deferred(); + ClientMock.connectImpl = () => connecting.promise; - const peerError = `some random error for the peer ${peer.ip}`; - peerConnector.setError(peer, peerError); + const connectPromise = peerConnector.connect(peer); + const disconnectPromise = peerConnector.disconnect(peer.ip); - assert.equal(peerConnector.getError(peer), peerError); - }); + connecting.reject(new Error("refused")); - it("#hasError - should return true if the peer has the error specified set", ({ peerConnector }) => { - const peer = new Peer("178.165.11.12", 4000); - - const peerError = `some random error for the peer ${peer.ip}`; - peerConnector.setError(peer, peerError); - - assert.true(peerConnector.hasError(peer, peerError)); + await assert.rejects(() => connectPromise, "refused"); + await assert.resolves(() => disconnectPromise); }); - it("#hasError - should return false if the peer has not the error specified set", ({ peerConnector }) => { - const peer = new Peer("178.165.11.12", 4000); - - const peerError = `some random error for the peer ${peer.ip}`; - peerConnector.setError(peer, peerError); + it("#disconnect - should do nothing for an unknown peer", async ({ peerConnector }) => { + await peerConnector.disconnect(peer.ip); - assert.false(peerConnector.hasError(peer, "a different error")); + assert.length(ClientMock.instances, 0); }); - it("#hasError - should return false if the peer has no error", ({ peerConnector }) => { - const peer = new Peer("178.165.11.12", 4000); + it("#emit - should connect, set the timeout and send the request", async ({ peerConnector }) => { + const connection = await peerConnector.connect(peer); + const request = spy(connection, "request"); - const peerError = `some random error for the peer ${peer.ip}`; + const payload = Buffer.from("payload"); + const response = await peerConnector.emit(peer, "getStatus", payload, 5000); - assert.false(peerConnector.hasError(peer, peerError)); + request.calledOnce(); + request.calledWith({ headers: {}, method: "POST", path: "getStatus", payload }); + assert.equal(connection.timeout, 5000); + assert.equal(response.payload, Buffer.from("response")); }); - it("#forgetError - should forget the error set for the peer", ({ peerConnector }) => { - const peer = new Peer("178.165.11.12", 4000); + it("should dispose the peer when the client disconnects and ban it on client errors", async ({ + peerConnector, + peerDisposer, + }) => { + const disposePeer = spy(peerDisposer, "disposePeer"); + const banPeer = spy(peerDisposer, "banPeer"); - const peerError = `some random error for the peer ${peer.ip}`; - peerConnector.setError(peer, peerError); + const connection = await peerConnector.connect(peer); - assert.equal(peerConnector.getError(peer), peerError); + connection.onDisconnect(); + disposePeer.calledOnce(); + disposePeer.calledWith(peer.ip); - peerConnector.forgetError(peer); - assert.undefined(peerConnector.getError(peer)); + const error = new Error("boom"); + connection.onError(error); + banPeer.calledOnce(); + banPeer.calledWith(peer.ip, error); }); }); diff --git a/packages/p2p/source/peer-connector.ts b/packages/p2p/source/peer-connector.ts index 6b2ffc52db..9255719669 100644 --- a/packages/p2p/source/peer-connector.ts +++ b/packages/p2p/source/peer-connector.ts @@ -17,20 +17,29 @@ export class PeerConnector implements Contracts.P2P.PeerConnector { @inject(Identifiers.P2P.Logger) private readonly logger!: Contracts.P2P.Logger; - private readonly connections: Map = new Map(); - readonly #lastConnectionCreate: Map = new Map(); + readonly #connections = new Map>(); + readonly #lastConnectionCreate = new Map(); public async connect(peer: Contracts.P2P.Peer): Promise { - return this.connections.get(peer.ip) || (await this.#create(peer)); + return this.#connections.get(peer.ip) ?? this.#create(peer); } public async disconnect(ip: string): Promise { - const connection = this.connections.get(ip); + const connection = this.#connections.get(ip); + if (!connection) { + return; + } + + this.#connections.delete(ip); - if (connection) { - await connection.terminate(); - this.connections.delete(ip); + let client: Client; + try { + client = await connection; + } catch { + return; // the creation failed, so there is nothing to close } + + await client.terminate(); } public async emit( @@ -56,6 +65,21 @@ export class PeerConnector implements Contracts.P2P.PeerConnector { } async #create(peer: Contracts.P2P.Peer): Promise { + const connection = this.#open(peer); + this.#connections.set(peer.ip, connection); + + try { + return await connection; + } catch (error) { + if (this.#connections.get(peer.ip) === connection) { + this.#connections.delete(peer.ip); + } + + throw error; + } + } + + async #open(peer: Contracts.P2P.Peer): Promise { // delay a bit if last connection create was less than 10 sec ago to prevent possible abuse of reconnection const timeSinceLastConnectionCreate = Date.now() - (this.#lastConnectionCreate.get(peer.ip) ?? 0); await delay(Math.max(0, TEN_SECONDS - timeSinceLastConnectionCreate)); @@ -63,7 +87,6 @@ export class PeerConnector implements Contracts.P2P.PeerConnector { const connection = new Client(`ws://${IpAddress.normalizeAddress(peer.ip)}:${peer.port}`, { timeout: 10_000, }); - this.connections.set(peer.ip, connection); this.#lastConnectionCreate.set(peer.ip, Date.now()); connection.onDisconnect = () => { @@ -77,7 +100,8 @@ export class PeerConnector implements Contracts.P2P.PeerConnector { this.app.get(Identifiers.P2P.Peer.Disposer).banPeer(peer.ip, error); }; - await connection.connect({ reconnect: false }); + // Without a timeout a black-holed address waits for the kernel TCP timeout (minutes). + await connection.connect({ reconnect: false, timeout: TEN_SECONDS }); return connection; } diff --git a/packages/p2p/source/peer-discoverer.ts b/packages/p2p/source/peer-discoverer.ts index 8e7e03ae0b..cc7a26e554 100644 --- a/packages/p2p/source/peer-discoverer.ts +++ b/packages/p2p/source/peer-discoverer.ts @@ -2,7 +2,6 @@ import type { Contracts } from "@mainsail/contracts"; import { Identifiers } from "@mainsail/constants"; import { inject, injectable } from "@mainsail/container"; -import { Services } from "@mainsail/kernel"; import { ensureError, http } from "@mainsail/utils"; import { readJSONSync } from "fs-extra/esm"; @@ -28,10 +27,7 @@ export class PeerDiscoverer implements Contracts.P2P.PeerDiscoverer { const { peers } = await this.communicator.getPeers(peer); for (const peer of peers) { - await this.app.get(Identifiers.Services.Trigger.Service).call<{ - ip: string; - options: Contracts.P2P.AcceptNewPeerOptions; - }>("validateAndAcceptPeer", { ip: peer.ip, options: {} }); + await this.#getProcessor().validateAndAcceptPeer(peer.ip, {}); } } catch (rawError) { const error = ensureError(rawError); @@ -67,14 +63,15 @@ export class PeerDiscoverer implements Contracts.P2P.PeerDiscoverer { await Promise.all( Object.values(peers).map((peer: Contracts.P2P.Peer) => - this.app.get(Identifiers.Services.Trigger.Service).call<{ - ip: string; - options: Contracts.P2P.AcceptNewPeerOptions; - }>("validateAndAcceptPeer", { ip: peer.ip, options: { seed: true } }), + this.#getProcessor().validateAndAcceptPeer(peer.ip, { seed: true }), ), ); } + #getProcessor(): Contracts.P2P.PeerProcessor { + return this.app.get(Identifiers.P2P.Peer.Processor); + } + // TODO: Get from all sources async #loadPeersFromUrlList(): Promise> { const urls: string[] = this.app.config<{ sources: string[] }>("peers")?.sources || []; diff --git a/packages/p2p/source/peer-processor.test.ts b/packages/p2p/source/peer-processor.test.ts index 90031b4fc5..2ae9273656 100644 --- a/packages/p2p/source/peer-processor.test.ts +++ b/packages/p2p/source/peer-processor.test.ts @@ -16,7 +16,8 @@ describe<{ peerCommunicator: any; peerDiscoverer: any; apiNodeDiscoverer: any; -}>("PeerProcessor", ({ it, beforeEach, spy }) => { + configuration: any; +}>("PeerProcessor", ({ it, assert, beforeEach, spy }) => { beforeEach((context) => { context.logger = { debug: () => {}, warn: () => {}, warnExtra: () => {} }; context.repository = { @@ -37,10 +38,11 @@ describe<{ context.peerDiscoverer = { discoverPeers: async () => {} }; context.apiNodeDiscoverer = { discoverApiNodes: async () => {} }; - const configuration = { + context.configuration = { getOptional: (key: string, defaultValue: unknown) => defaultValue, getRequired: (key: string) => ({ blacklist: [], maxSameSubnetPeers: 5, whitelist: ["*"] })[key], }; + const configuration = context.configuration; const app = new Application(); app.bind(Identifiers.ServiceProvider.Configuration).toConstantValue(configuration).whenTagged("plugin", "p2p"); @@ -76,6 +78,24 @@ describe<{ forgetPendingPeer.calledOnce(); }); + it("#validateAndAcceptPeer - should forget the pending peer and dispose it when accepting throws", async ({ + processor, + repository, + transactionPoolWorker, + peerDisposer, + }) => { + transactionPoolWorker.setPeer = async () => { + throw new Error("worker down"); + }; + const forgetPendingPeer = spy(repository, "forgetPendingPeer"); + const disposePeer = spy(peerDisposer, "disposePeer"); + + await processor.validateAndAcceptPeer("178.165.55.55"); + + forgetPendingPeer.calledOnce(); + disposePeer.calledWith("178.165.55.55"); + }); + it("#validateAndAcceptPeer - should not accept a peer that got banned while its acceptance was being verified", async ({ processor, repository, @@ -181,6 +201,44 @@ describe<{ setPendingPeer.neverCalled(); }); + it("#validateAndAcceptPeer - should not learn new peers in non-discovery mode", async ({ + processor, + repository, + peerVerifier, + configuration, + }) => { + configuration.getOptional = (key: string, defaultValue: unknown) => + key === "disableDiscovery" ? true : defaultValue; + const verify = spy(peerVerifier, "verify"); + const setPendingPeer = spy(repository, "setPendingPeer"); + + await processor.validateAndAcceptPeer("178.165.55.55"); + + verify.neverCalled(); + setPendingPeer.neverCalled(); + }); + + it("#validateAndAcceptPeer - should still accept seed peers in non-discovery mode", async ({ + processor, + repository, + configuration, + }) => { + configuration.getOptional = (key: string, defaultValue: unknown) => + key === "disableDiscovery" ? true : defaultValue; + const setPeer = spy(repository, "setPeer"); + + await processor.validateAndAcceptPeer("178.165.55.55", { seed: true }); + + setPeer.calledOnce(); + }); + + it("#validatePeerIp - should not reject inbound requests in non-discovery mode", ({ processor, configuration }) => { + configuration.getOptional = (key: string, defaultValue: unknown) => + key === "disableDiscovery" ? true : defaultValue; + + assert.true(processor.validatePeerIp("178.165.55.55")); + }); + it("#validateAndAcceptPeer - should reject localhost addresses", async ({ processor, peerVerifier }) => { const verify = spy(peerVerifier, "verify"); diff --git a/packages/p2p/source/peer-processor.ts b/packages/p2p/source/peer-processor.ts index 3d48f7248b..8c2da8363e 100644 --- a/packages/p2p/source/peer-processor.ts +++ b/packages/p2p/source/peer-processor.ts @@ -69,17 +69,17 @@ export class PeerProcessor implements Contracts.P2P.PeerProcessor { return; } + if (!options.seed && this.configuration.getOptional("disableDiscovery", false)) { + this.logger.debug(`Rejected ${ip} because the relay is in non-discovery mode.`, "p2p"); + return; + } + if (this.validatePeerIp(ip, options)) { await this.#acceptNewPeer(ip); } } public validatePeerIp(ip: string, options: Contracts.P2P.AcceptNewPeerOptions = {}): boolean { - if (this.configuration.getOptional("disableDiscovery", false)) { - this.logger.warn(`Rejected ${ip} because the relay is in non-discovery mode.`, "p2p"); - return false; - } - if (!isValidPeerIp(ip)) { return false; } @@ -115,13 +115,17 @@ export class PeerProcessor implements Contracts.P2P.PeerProcessor { this.repository.setPendingPeer(peer); - const txPoolNode = this.txPoolNodeFactory(ip); + try { + const txPoolNode = this.txPoolNodeFactory(ip); - if ((await this.peerVerifier.verify(peer)) && (await this.txPoolNodeVerifier.verify(txPoolNode))) { - // Re-check after the awaits: the peer may have been banned while its + // The ban is re-checked after the awaits: the peer may have been banned while its // acceptance was being verified, and a banned peer must never enter the // repository - nothing would remove it before the ban expires. - if (!this.peerDisposer.isBanned(peer.ip)) { + if ( + (await this.peerVerifier.verify(peer)) && + (await this.txPoolNodeVerifier.verify(txPoolNode)) && + !this.peerDisposer.isBanned(peer.ip) + ) { this.repository.setPeer(peer); this.logger.debug(`Accepted new peer ${peer.ip}:${peer.port} (v${peer.version})`, "p2p"); @@ -133,9 +137,11 @@ export class PeerProcessor implements Contracts.P2P.PeerProcessor { await this.peerDiscoverer.discoverPeers(peer); await this.ApiNodeDiscoverer.discoverApiNodes(peer); } + } catch (rawError) { + this.peerDisposer.disposePeer(ip); + } finally { + this.repository.forgetPendingPeer(peer); } - - this.repository.forgetPendingPeer(peer); } async #disconnectInvalidPeers(): Promise { diff --git a/packages/p2p/source/service-provider.test.ts b/packages/p2p/source/service-provider.test.ts index d74a59403e..bcc8af6f79 100644 --- a/packages/p2p/source/service-provider.test.ts +++ b/packages/p2p/source/service-provider.test.ts @@ -14,7 +14,6 @@ describe<{ app: Application; serviceProvider: ServiceProvider; }>("ServiceProvider", ({ it, assert, beforeEach, stub }) => { - const triggerService = { bind: () => {} }; const validator = { addFormat: () => {}, addKeyword: () => {} }; const server = { boot: async () => {}, dispose: async () => {}, initialize: async () => {} }; const statisticService = { boot: async () => {} }; @@ -25,7 +24,6 @@ describe<{ beforeEach((context) => { context.app = new Application(); - context.app.bind(Identifiers.Services.Trigger.Service).toConstantValue(triggerService); context.app.bind(Identifiers.Cryptography.Validator).toConstantValue(validator); context.app.bind(Identifiers.Services.EventDispatcher.Service).toConstantValue(eventDispatcher); @@ -93,13 +91,11 @@ describe<{ }>("ServiceProvider.configSchema", ({ it, assert, beforeEach }) => { const importDefaults = async () => (await importFresh("../distribution/defaults.js")).defaults; - const triggerService = { bind: () => {} }; const validator = { addFormat: () => {} }; beforeEach((context) => { context.app = new Application(); - context.app.bind(Identifiers.Services.Trigger.Service).toConstantValue(triggerService); context.app.bind(Identifiers.Cryptography.Configuration).toConstantValue({ getMilestone: () => ({ roundValidators: 2, diff --git a/packages/p2p/source/service-provider.ts b/packages/p2p/source/service-provider.ts index ccea3f3a6a..769e49db38 100644 --- a/packages/p2p/source/service-provider.ts +++ b/packages/p2p/source/service-provider.ts @@ -2,15 +2,10 @@ import type { Contracts } from "@mainsail/contracts"; import { Identifiers } from "@mainsail/constants"; import { injectable } from "@mainsail/container"; -import { Providers, Services } from "@mainsail/kernel"; +import { Providers } from "@mainsail/kernel"; import { assert } from "@mainsail/utils"; import Joi from "joi"; -import { - RevalidateApiNodeAction, - ValidateAndAcceptApiNodeAction, - ValidateAndAcceptPeerAction, -} from "./actions/index.js"; import { ApiNodeDiscoverer } from "./api-node-discoverer.js"; import { ApiNodeProcessor } from "./api-node-processor.js"; import { ApiNodeRepository } from "./api-node-repository.js"; @@ -60,8 +55,6 @@ export class ServiceProvider extends Providers.ServiceProvider { this.#registerFactories(); this.#registerServices(); - - this.#registerActions(); } public async boot(): Promise { @@ -213,20 +206,6 @@ export class ServiceProvider extends Providers.ServiceProvider { await server.initialize("P2P Server", serverConfig); } - #registerActions(): void { - this.app - .get(Identifiers.Services.Trigger.Service) - .bind("validateAndAcceptPeer", new ValidateAndAcceptPeerAction(this.app)); - - this.app - .get(Identifiers.Services.Trigger.Service) - .bind("validateAndAcceptApiNode", new ValidateAndAcceptApiNodeAction(this.app)); - - this.app - .get(Identifiers.Services.Trigger.Service) - .bind("revalidateApiNode", new RevalidateApiNodeAction(this.app)); - } - #registerValidation(): void { for (const [name, format] of Object.entries(makeFormats())) { this.app.get(Identifiers.Cryptography.Validator).addFormat(name, format); diff --git a/packages/p2p/source/socket-server/codecs/get-api-nodes.ts b/packages/p2p/source/socket-server/codecs/get-api-nodes.ts index 3ae20160a8..4ccb225aec 100644 --- a/packages/p2p/source/socket-server/codecs/get-api-nodes.ts +++ b/packages/p2p/source/socket-server/codecs/get-api-nodes.ts @@ -1,25 +1,25 @@ -import { packHeadersBitmaps, unpackHeadersBitmaps } from "./headers.js"; +import { packHeadersBitmaps, unpackHeadersBitmaps, type WithUnpackedHeaders } from "./headers.js"; import { getApiNodes as proto } from "./proto/protos.js"; export const getApiNodes = { request: { - deserialize: (payload: Buffer): proto.IGetApiNodesRequest => { + deserialize: (payload: Buffer): proto.GetApiNodesRequest.$Properties => { const decoded = proto.GetApiNodesRequest.decode(payload); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetApiNodesRequest): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetApiNodesRequest.encode(packHeadersBitmaps(object)).finish()), }, response: { - deserialize: (payload: Buffer): proto.IGetApiNodesResponse => { + deserialize: (payload: Buffer): proto.GetApiNodesResponse.$Properties => { const decoded = proto.GetApiNodesResponse.toObject(proto.GetApiNodesResponse.decode(payload), { defaults: true, }); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetApiNodesResponse): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetApiNodesResponse.encode(packHeadersBitmaps(object)).finish()), }, }; diff --git a/packages/p2p/source/socket-server/codecs/get-blocks.ts b/packages/p2p/source/socket-server/codecs/get-blocks.ts index fce56fb26e..90940f6092 100644 --- a/packages/p2p/source/socket-server/codecs/get-blocks.ts +++ b/packages/p2p/source/socket-server/codecs/get-blocks.ts @@ -1,25 +1,25 @@ -import { packHeadersBitmaps, unpackHeadersBitmaps } from "./headers.js"; +import { packHeadersBitmaps, unpackHeadersBitmaps, type WithUnpackedHeaders } from "./headers.js"; import { getBlocks as proto } from "./proto/protos.js"; export const getBlocks = { request: { - deserialize: (payload: Buffer): proto.IGetBlocksRequest => { + deserialize: (payload: Buffer): proto.GetBlocksRequest.$Properties => { const decoded = proto.GetBlocksRequest.decode(payload); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetBlocksRequest): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetBlocksRequest.encode(packHeadersBitmaps(object)).finish()), }, response: { - deserialize: (payload: Buffer): proto.IGetBlocksResponse => { + deserialize: (payload: Buffer): proto.GetBlocksResponse.$Properties => { const decoded = proto.GetBlocksResponse.toObject(proto.GetBlocksResponse.decode(payload), { defaults: true, }); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetBlocksResponse): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetBlocksResponse.encode(packHeadersBitmaps(object)).finish()), }, }; diff --git a/packages/p2p/source/socket-server/codecs/get-messages.ts b/packages/p2p/source/socket-server/codecs/get-messages.ts index 0abd2d5aa6..056f22bcdd 100644 --- a/packages/p2p/source/socket-server/codecs/get-messages.ts +++ b/packages/p2p/source/socket-server/codecs/get-messages.ts @@ -1,30 +1,44 @@ -import { packBitmaps, packHeadersBitmaps, unpackBitmapsInPlace, unpackHeadersBitmaps } from "./headers.js"; +import { + packBitmaps, + packHeadersBitmaps, + type Unpacked, + unpackBitmapsInPlace, + unpackHeadersBitmaps, + type WithUnpackedHeaders, +} from "./headers.js"; import { getMessages as proto } from "./proto/protos.js"; +type GetMessagesRequest = Omit, "query"> & { + query?: Unpacked | null; +}; + export const getMessages = { request: { - deserialize: (payload: Buffer): proto.IGetMessagesRequest => { + deserialize: (payload: Buffer): proto.GetMessagesRequest.$Properties => { const decoded = proto.GetMessagesRequest.decode(payload); unpackHeadersBitmaps(decoded); unpackBitmapsInPlace(decoded.query); return decoded; }, - serialize: (object: proto.IGetMessagesRequest): Buffer => + serialize: (object: GetMessagesRequest): Buffer => Buffer.from( proto.GetMessagesRequest.encode( - packHeadersBitmaps(object.query ? { ...object, query: packBitmaps(object.query) } : object), + packHeadersBitmaps({ + ...object, + query: object.query ? packBitmaps(object.query) : object.query, + }), ).finish(), ), }, response: { - deserialize: (payload: Buffer): proto.IGetMessagesResponse => { + deserialize: (payload: Buffer): proto.GetMessagesResponse.$Properties => { const decoded = proto.GetMessagesResponse.toObject(proto.GetMessagesResponse.decode(payload), { defaults: true, }); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetMessagesResponse): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetMessagesResponse.encode(packHeadersBitmaps(object)).finish()), }, }; diff --git a/packages/p2p/source/socket-server/codecs/get-peers.ts b/packages/p2p/source/socket-server/codecs/get-peers.ts index 985a47f010..1f4ed63ebc 100644 --- a/packages/p2p/source/socket-server/codecs/get-peers.ts +++ b/packages/p2p/source/socket-server/codecs/get-peers.ts @@ -1,25 +1,25 @@ -import { packHeadersBitmaps, unpackHeadersBitmaps } from "./headers.js"; +import { packHeadersBitmaps, unpackHeadersBitmaps, type WithUnpackedHeaders } from "./headers.js"; import { getPeers as proto } from "./proto/protos.js"; export const getPeers = { request: { - deserialize: (payload: Buffer): proto.IGetPeersRequest => { + deserialize: (payload: Buffer): proto.GetPeersRequest.$Properties => { const decoded = proto.GetPeersRequest.decode(payload); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetPeersRequest): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetPeersRequest.encode(packHeadersBitmaps(object)).finish()), }, response: { - deserialize: (payload: Buffer): proto.IGetPeersResponse => { + deserialize: (payload: Buffer): proto.GetPeersResponse.$Properties => { const decoded = proto.GetPeersResponse.toObject(proto.GetPeersResponse.decode(payload), { defaults: true, }); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetPeersResponse): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetPeersResponse.encode(packHeadersBitmaps(object)).finish()), }, }; diff --git a/packages/p2p/source/socket-server/codecs/get-proposal.ts b/packages/p2p/source/socket-server/codecs/get-proposal.ts index 96a1945f37..cfa36766c0 100644 --- a/packages/p2p/source/socket-server/codecs/get-proposal.ts +++ b/packages/p2p/source/socket-server/codecs/get-proposal.ts @@ -1,25 +1,25 @@ -import { packHeadersBitmaps, unpackHeadersBitmaps } from "./headers.js"; +import { packHeadersBitmaps, unpackHeadersBitmaps, type WithUnpackedHeaders } from "./headers.js"; import { getProposal as proto } from "./proto/protos.js"; export const getProposal = { request: { - deserialize: (payload: Buffer): proto.IGetProposalRequest => { + deserialize: (payload: Buffer): proto.GetProposalRequest.$Properties => { const decoded = proto.GetProposalRequest.decode(payload); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetProposalRequest): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetProposalRequest.encode(packHeadersBitmaps(object)).finish()), }, response: { - deserialize: (payload: Buffer): proto.IGetProposalResponse => { + deserialize: (payload: Buffer): proto.GetProposalResponse.$Properties => { const decoded = proto.GetProposalResponse.toObject(proto.GetProposalResponse.decode(payload), { defaults: true, }); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetProposalResponse): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetProposalResponse.encode(packHeadersBitmaps(object)).finish()), }, }; diff --git a/packages/p2p/source/socket-server/codecs/get-status.ts b/packages/p2p/source/socket-server/codecs/get-status.ts index 05962241f0..f9e6cdf393 100644 --- a/packages/p2p/source/socket-server/codecs/get-status.ts +++ b/packages/p2p/source/socket-server/codecs/get-status.ts @@ -1,25 +1,25 @@ -import { packHeadersBitmaps, unpackHeadersBitmaps } from "./headers.js"; +import { packHeadersBitmaps, unpackHeadersBitmaps, type WithUnpackedHeaders } from "./headers.js"; import { getStatus as proto } from "./proto/protos.js"; export const getStatus = { request: { - deserialize: (payload: Buffer): proto.IGetStatusRequest => { + deserialize: (payload: Buffer): proto.GetStatusRequest.$Properties => { const decoded = proto.GetStatusRequest.decode(payload); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetStatusRequest): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetStatusRequest.encode(packHeadersBitmaps(object)).finish()), }, response: { - deserialize: (payload: Buffer): proto.IGetStatusResponse => { + deserialize: (payload: Buffer): proto.GetStatusResponse.$Properties => { const decoded = proto.GetStatusResponse.toObject(proto.GetStatusResponse.decode(payload), { defaults: true, }); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IGetStatusResponse): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.GetStatusResponse.encode(packHeadersBitmaps(object)).finish()), }, }; diff --git a/packages/p2p/source/socket-server/codecs/headers.ts b/packages/p2p/source/socket-server/codecs/headers.ts index 8ab237f146..4b995f3ad0 100644 --- a/packages/p2p/source/socket-server/codecs/headers.ts +++ b/packages/p2p/source/socket-server/codecs/headers.ts @@ -3,6 +3,16 @@ import { validatorSetPack, validatorSetUnpack } from "@mainsail/utils"; type Bitmap = readonly boolean[]; type BitmapFields = { validatorsSignedPrevote?: unknown; validatorsSignedPrecommit?: unknown }; +// The wire shape carries the bitmaps packed as bytes; callers hand in the boolean arrays and the codec packs them. +export type Unpacked = Omit & { + validatorsSignedPrevote?: Bitmap | null; + validatorsSignedPrecommit?: Bitmap | null; +}; + +export type WithUnpackedHeaders = Omit & { + headers?: Unpacked> | null; +}; + export const packBitmap = (bitmap: Bitmap): Buffer => { if (bitmap.length > 255) { throw new RangeError(`Bitmap of ${bitmap.length} validators does not fit the count byte`); @@ -40,15 +50,15 @@ export const unpackBitmap = (packed: Uint8Array | null | undefined): boolean[] = return validatorSetUnpack(value, count); }; -export const packBitmaps = (object: T): T => +export const packBitmaps = (object: Unpacked): T => ({ ...object, - validatorsSignedPrecommit: packBitmap((object.validatorsSignedPrecommit ?? []) as Bitmap), - validatorsSignedPrevote: packBitmap((object.validatorsSignedPrevote ?? []) as Bitmap), + validatorsSignedPrecommit: packBitmap(object.validatorsSignedPrecommit ?? []), + validatorsSignedPrevote: packBitmap(object.validatorsSignedPrevote ?? []), }) as T; -export const packHeadersBitmaps = (object: T): T => - object.headers ? ({ ...object, headers: packBitmaps(object.headers as BitmapFields) } as T) : object; +export const packHeadersBitmaps = (object: WithUnpackedHeaders): T => + (object.headers ? { ...object, headers: packBitmaps(object.headers) } : object) as T; export const unpackBitmapsInPlace = (object: BitmapFields | null | undefined): void => { if (object) { diff --git a/packages/p2p/source/socket-server/codecs/post-message.ts b/packages/p2p/source/socket-server/codecs/post-message.ts index 88ff091213..59af0d0373 100644 --- a/packages/p2p/source/socket-server/codecs/post-message.ts +++ b/packages/p2p/source/socket-server/codecs/post-message.ts @@ -1,9 +1,9 @@ -import { packHeadersBitmaps, unpackHeadersBitmaps } from "./headers.js"; +import { packHeadersBitmaps, unpackHeadersBitmaps, type WithUnpackedHeaders } from "./headers.js"; import { postMessage as proto } from "./proto/protos.js"; export const postMessage = { request: { - deserialize: (payload: Buffer): proto.IPostMessageRequest => { + deserialize: (payload: Buffer): proto.PostMessageRequest.$Properties => { const decoded = proto.PostMessageRequest.decode(payload); unpackHeadersBitmaps(decoded); return { @@ -11,18 +11,18 @@ export const postMessage = { message: Buffer.from(decoded.message), }; }, - serialize: (object: proto.IPostMessageRequest): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.PostMessageRequest.encode(packHeadersBitmaps(object)).finish()), }, response: { - deserialize: (payload: Buffer): proto.IPostMessageResponse => { + deserialize: (payload: Buffer): proto.PostMessageResponse.$Properties => { const decoded = proto.PostMessageResponse.toObject(proto.PostMessageResponse.decode(payload), { defaults: true, }); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IPostMessageResponse): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.PostMessageResponse.encode(packHeadersBitmaps(object)).finish()), }, }; diff --git a/packages/p2p/source/socket-server/codecs/post-proposal.ts b/packages/p2p/source/socket-server/codecs/post-proposal.ts index 97a8fab22e..ec094d48b1 100644 --- a/packages/p2p/source/socket-server/codecs/post-proposal.ts +++ b/packages/p2p/source/socket-server/codecs/post-proposal.ts @@ -1,9 +1,9 @@ -import { packHeadersBitmaps, unpackHeadersBitmaps } from "./headers.js"; +import { packHeadersBitmaps, unpackHeadersBitmaps, type WithUnpackedHeaders } from "./headers.js"; import { postProposal as proto } from "./proto/protos.js"; export const postProposal = { request: { - deserialize: (payload: Buffer): proto.IPostProposalRequest => { + deserialize: (payload: Buffer): proto.PostProposalRequest.$Properties => { const decoded = proto.PostProposalRequest.decode(payload); unpackHeadersBitmaps(decoded); return { @@ -11,18 +11,18 @@ export const postProposal = { proposal: Buffer.from(decoded.proposal), }; }, - serialize: (object: proto.IPostProposalRequest): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.PostProposalRequest.encode(packHeadersBitmaps(object)).finish()), }, response: { - deserialize: (payload: Buffer): proto.IPostProposalResponse => { + deserialize: (payload: Buffer): proto.PostProposalResponse.$Properties => { const decoded = proto.PostProposalResponse.toObject(proto.PostProposalResponse.decode(payload), { defaults: true, }); unpackHeadersBitmaps(decoded); return decoded; }, - serialize: (object: proto.IPostProposalResponse): Buffer => + serialize: (object: WithUnpackedHeaders): Buffer => Buffer.from(proto.PostProposalResponse.encode(packHeadersBitmaps(object)).finish()), }, }; diff --git a/packages/p2p/source/socket-server/controllers/get-blocks.ts b/packages/p2p/source/socket-server/controllers/get-blocks.ts index a5c96f86fd..c068c8e6f1 100644 --- a/packages/p2p/source/socket-server/controllers/get-blocks.ts +++ b/packages/p2p/source/socket-server/controllers/get-blocks.ts @@ -48,7 +48,7 @@ export class GetBlocksController implements Contracts.P2P.Controller { maxBytes, ); - this.logger.info( + this.logger.debug( `${mapAddr(request.info.remoteAddress)} has downloaded ${pluralize( "block", blocks.length, diff --git a/packages/p2p/source/socket-server/plugins/validate-ip.test.ts b/packages/p2p/source/socket-server/plugins/validate-ip.test.ts new file mode 100644 index 0000000000..c38b067674 --- /dev/null +++ b/packages/p2p/source/socket-server/plugins/validate-ip.test.ts @@ -0,0 +1,106 @@ +import { Server } from "@hapi/hapi"; +import { Identifiers } from "@mainsail/constants"; +import { Application } from "@mainsail/kernel"; +import { describe } from "@mainsail/test-runner"; + +import { ValidateIpPlugin } from "./validate-ip"; + +describe<{ + app: Application; + validateIpPlugin: ValidateIpPlugin; + peerDisposer: { + banPeer: (ip: string, error: Error) => void; + disposePeer: (ip: string) => void; + isBanned: (ip: string) => boolean; + }; + peerProcessor: { validatePeerIp: (ip: string) => boolean }; + peerRepository: { hasPeer: (ip: string) => boolean }; + server: Server; +}>("ValidateIpPlugin", ({ it, assert, beforeEach, spy }) => { + const responsePayload = { status: "ok" }; + + beforeEach((context) => { + context.app = new Application(); + + context.peerDisposer = { banPeer: () => {}, disposePeer: () => {}, isBanned: () => false }; + context.peerProcessor = { validatePeerIp: () => true }; + context.peerRepository = { hasPeer: () => false }; + + context.app + .bind(Identifiers.ServiceProvider.Configuration) + .toConstantValue({ getRequired: () => false }) + .whenTagged("plugin", "p2p"); + context.app.bind(Identifiers.P2P.Peer.Disposer).toConstantValue(context.peerDisposer); + context.app.bind(Identifiers.P2P.Peer.Processor).toConstantValue(context.peerProcessor); + context.app.bind(Identifiers.P2P.Peer.Repository).toConstantValue(context.peerRepository); + + context.validateIpPlugin = context.app.resolve(ValidateIpPlugin); + + const controller = { handle: () => responsePayload }; + context.server = new Server({ port: 4104 }); + context.server.bind(controller); + context.server.route({ + method: "POST", + options: { handler: controller.handle.bind(controller) }, + path: "/p2p/peer/mockroute", + }); + + context.validateIpPlugin.register(context.server); + }); + + const inject = (server: Server) => server.inject({ method: "POST", payload: {}, url: "/p2p/peer/mockroute" }); + + it("should serve a request from an ip that passes the acceptance policy", async ({ server, peerDisposer }) => { + const disposePeer = spy(peerDisposer, "disposePeer"); + + const response = await inject(server); + + assert.equal(response.statusCode, 200); + assert.equal(JSON.parse(response.payload), responsePayload); + disposePeer.neverCalled(); + }); + + it("should serve a request from an accepted peer without re-applying the acceptance policy", async ({ + server, + peerDisposer, + peerProcessor, + peerRepository, + }) => { + // A known peer whose /24 subnet has meanwhile reached the limit. + peerRepository.hasPeer = () => true; + peerProcessor.validatePeerIp = () => false; + const validatePeerIp = spy(peerProcessor, "validatePeerIp"); + const disposePeer = spy(peerDisposer, "disposePeer"); + + const response = await inject(server); + + assert.equal(response.statusCode, 200); + validatePeerIp.neverCalled(); + disposePeer.neverCalled(); + }); + + it("should dispose an unknown ip that fails the acceptance policy", async ({ + server, + peerDisposer, + peerProcessor, + }) => { + peerProcessor.validatePeerIp = () => false; + const disposePeer = spy(peerDisposer, "disposePeer"); + + const response = await inject(server); + + assert.equal(response.statusCode, 400); + disposePeer.calledWith("127.0.0.1"); + }); + + it("should reject a banned ip even when it is a known peer", async ({ server, peerDisposer, peerRepository }) => { + peerRepository.hasPeer = () => true; + peerDisposer.isBanned = () => true; + const banPeer = spy(peerDisposer, "banPeer"); + + const response = await inject(server); + + assert.equal(response.statusCode, 400); + banPeer.calledOnce(); + }); +}); diff --git a/packages/p2p/source/socket-server/plugins/validate-ip.ts b/packages/p2p/source/socket-server/plugins/validate-ip.ts index b3ccedaf8c..b7113e8ba5 100644 --- a/packages/p2p/source/socket-server/plugins/validate-ip.ts +++ b/packages/p2p/source/socket-server/plugins/validate-ip.ts @@ -16,6 +16,9 @@ export class ValidateIpPlugin extends BasePlugin { @inject(Identifiers.P2P.Peer.Processor) private readonly peerProcessor!: Contracts.P2P.PeerProcessor; + @inject(Identifiers.P2P.Peer.Repository) + private readonly peerRepository!: Contracts.P2P.PeerRepository; + public register(server: Server): void { if (this.configuration.getRequired("developmentMode.enabled")) { return; @@ -30,6 +33,10 @@ export class ValidateIpPlugin extends BasePlugin { return this.banAndReturnBadRequest(peerRequest, "Validation failed (peer is bannned)"); } + if (this.peerRepository.hasPeer(ip)) { + return h.continue; + } + if (!this.peerProcessor.validatePeerIp(ip)) { return this.disposeAndReturnBadRequest(peerRequest, "Validation failed (bad ip)"); } diff --git a/packages/p2p/source/socket-server/routes/post-message.ts b/packages/p2p/source/socket-server/routes/post-message.ts index dd70c3321a..92550321c1 100644 --- a/packages/p2p/source/socket-server/routes/post-message.ts +++ b/packages/p2p/source/socket-server/routes/post-message.ts @@ -16,7 +16,7 @@ export class PostMessageRoute extends Route { "/postMessage": { codec: Codecs.postMessage, id: Routes.PostMessage, - maxBytes: constants.MAX_PAYLOAD_SERVER, + maxBytes: constants.MAX_PAYLOAD_POST_MESSAGE, validation: Schemas.postMessage(this.cryptoConfiguration), }, }; diff --git a/packages/p2p/source/socket-server/schemas/post-message.test.ts b/packages/p2p/source/socket-server/schemas/post-message.test.ts new file mode 100644 index 0000000000..6d29921e50 --- /dev/null +++ b/packages/p2p/source/socket-server/schemas/post-message.test.ts @@ -0,0 +1,35 @@ +import { describe } from "@mainsail/test-runner"; + +import { constants } from "../../constants"; +import { postMessage } from "./post-message"; + +describe("postMessage schema", ({ it, assert }) => { + const makeSchema = () => postMessage({ getMaxRoundValidators: () => 2 } as any); + + const headers = { + blockNumber: 2, + // eslint-disable-next-line unicorn/no-null + proposedBlockHash: null, + round: 0, + step: 1, + validatorsSignedPrecommit: [false, false], + validatorsSignedPrevote: [false, false], + version: "0.0.1", + }; + + it("should accept a message that fits the cap", () => { + const message = Buffer.alloc(constants.MAX_PAYLOAD_POST_MESSAGE); + + assert.undefined(makeSchema().validate({ headers, message }).error); + }); + + it("should reject a message above the cap", () => { + const message = Buffer.alloc(constants.MAX_PAYLOAD_POST_MESSAGE + 1); + + assert.defined(makeSchema().validate({ headers, message }).error); + }); + + it("should reject a message that is not binary", () => { + assert.defined(makeSchema().validate({ headers, message: {} }).error); + }); +}); diff --git a/packages/p2p/source/socket-server/schemas/post-message.ts b/packages/p2p/source/socket-server/schemas/post-message.ts index 762b14e4f8..703e59b7a4 100644 --- a/packages/p2p/source/socket-server/schemas/post-message.ts +++ b/packages/p2p/source/socket-server/schemas/post-message.ts @@ -2,10 +2,11 @@ import type { Contracts } from "@mainsail/contracts"; import Joi from "joi"; +import { constants } from "../../constants.js"; import { makeHeaders } from "./shared.js"; export const postMessage = (configuration: Contracts.Crypto.Configuration): Joi.ObjectSchema => Joi.object({ headers: makeHeaders(configuration), - message: Joi.binary(), + message: Joi.binary().max(constants.MAX_PAYLOAD_POST_MESSAGE), }); diff --git a/packages/p2p/source/socket-server/schemas/shared.ts b/packages/p2p/source/socket-server/schemas/shared.ts index cdd8b55209..b5365768bb 100644 --- a/packages/p2p/source/socket-server/schemas/shared.ts +++ b/packages/p2p/source/socket-server/schemas/shared.ts @@ -7,12 +7,18 @@ export const makeHeaders = (configuration: Contracts.Crypto.Configuration): Joi. return Joi.object({ blockNumber: Joi.number().integer().min(1).required(), - // eslint-disable-next-line unicorn/no-null - proposedBlockHash: Joi.string().allow(null).required(), + proposedBlockHash: Joi.string() + .pattern(/^[0-9a-f]{64}$/) + // eslint-disable-next-line unicorn/no-null + .allow(null) + .required(), round: Joi.number().integer().min(0).required(), step: Joi.number().integer().min(0).max(2).required(), validatorsSignedPrecommit: Joi.array().items(Joi.boolean()).max(roundValidators).required(), validatorsSignedPrevote: Joi.array().items(Joi.boolean()).max(roundValidators).required(), - version: Joi.string().required(), + version: Joi.string() + .max(24) + .pattern(/^\d+\.\d+\.\d+(-[a-zA-Z0-9.-]+\.\d+)?$/) + .required(), }).required(); }; diff --git a/packages/p2p/source/tx-pool-node-verifier.test.ts b/packages/p2p/source/tx-pool-node-verifier.test.ts new file mode 100644 index 0000000000..64aaa57234 --- /dev/null +++ b/packages/p2p/source/tx-pool-node-verifier.test.ts @@ -0,0 +1,80 @@ +import { Identifiers } from "@mainsail/constants"; +import { Application } from "@mainsail/kernel"; +import { describe } from "@mainsail/test-runner"; +import esmock from "esmock"; + +import { TxPoolNodeVerifier } from "./tx-pool-node-verifier"; + +let httpGet: (url: string, options: unknown) => Promise; + +const { TxPoolNodeVerifier: TxPoolNodeVerifierProxy } = await esmock( + "./tx-pool-node-verifier", + import.meta.url, + { "@mainsail/utils": { http: { get: (url: string, options: unknown) => httpGet(url, options) } } }, + {}, +); + +describe<{ + verifier: TxPoolNodeVerifier; + configuration: any; +}>("TxPoolNodeVerifier", ({ it, assert, beforeEach }) => { + const helloWorld = { data: "Hello World from Transaction Pool API!" }; + const contentLength = String(JSON.stringify(helloWorld).length); + const node = { url: "http://1.2.3.4:4007/" } as any; + + const respond = (statusCode: number, data: unknown, headers: string[]) => { + httpGet = async () => ({ data, headers, method: "GET", statusCode, statusMessage: "" }); + }; + + beforeEach((context) => { + respond(200, helloWorld, ["content-length", contentLength]); + context.configuration = { getRequired: () => false }; + + const app = new Application(); + app.bind(Identifiers.P2P.Logger).toConstantValue({ debugExtra: () => {} }); + app.bind(Identifiers.ServiceProvider.Configuration) + .toConstantValue(context.configuration) + .whenTagged("plugin", "p2p"); + + context.verifier = app.resolve(TxPoolNodeVerifierProxy); + }); + + it("#verify - should accept the default tx pool response", async ({ verifier }) => { + assert.true(await verifier.verify(node)); + }); + + it("#verify - should match the content-length header regardless of its case", async ({ verifier }) => { + respond(200, helloWorld, ["Content-Type", "application/json", "Content-Length", contentLength]); + + assert.true(await verifier.verify(node)); + }); + + it("#verify - should reject a missing content length", async ({ verifier }) => { + respond(200, helloWorld, ["content-type", "application/json"]); + + assert.false(await verifier.verify(node)); + }); + + it("#verify - should reject a wrong content length", async ({ verifier }) => { + respond(200, helloWorld, ["content-length", "1"]); + + assert.false(await verifier.verify(node)); + }); + + it("#verify - should reject an unreachable node", async ({ verifier }) => { + httpGet = async () => { + throw new Error("connection refused"); + }; + + assert.false(await verifier.verify(node)); + }); + + it("#verify - should skip verification when configured", async ({ verifier, configuration }) => { + configuration.getRequired = () => true; + httpGet = async () => { + throw new Error("must not be called"); + }; + + assert.true(await verifier.verify(node)); + }); +}); diff --git a/packages/p2p/source/tx-pool-node-verifier.ts b/packages/p2p/source/tx-pool-node-verifier.ts index af559573a1..f16c1a4e88 100644 --- a/packages/p2p/source/tx-pool-node-verifier.ts +++ b/packages/p2p/source/tx-pool-node-verifier.ts @@ -47,7 +47,14 @@ export class TxPoolNodeVerifier implements Contracts.P2P.TxPoolNodeVerifier { } #verifyHeaders(response: HttpResponse): void { - const contentLength = Number(response.headers[response.headers.indexOf("content-length") + 1]); + const index = response.headers.findIndex( + (header, index) => index % 2 === 0 && header.toLowerCase() === "content-length", + ); + if (index === -1) { + throw new Error("missing content length"); + } + + const contentLength = Number(response.headers[index + 1]); if (contentLength !== helloWorldLength) { throw new Error("invalid content length"); } diff --git a/packages/p2p/source/tx-pool-node.ts b/packages/p2p/source/tx-pool-node.ts index 201bedf106..236b47296c 100644 --- a/packages/p2p/source/tx-pool-node.ts +++ b/packages/p2p/source/tx-pool-node.ts @@ -1,6 +1,7 @@ import type { Contracts } from "@mainsail/contracts"; import { injectable } from "@mainsail/container"; +import { IpAddress } from "@mainsail/utils"; @injectable() export class TxPoolNode implements Contracts.P2P.TxPoolNode { @@ -15,6 +16,6 @@ export class TxPoolNode implements Contracts.P2P.TxPoolNode { } public get url(): string { - return `http://${this.ip}:${this.port}/`; + return `http://${IpAddress.normalizeAddress(this.ip)}:${this.port}/`; } } diff --git a/packages/p2p/source/utils/get-peer-url.test.ts b/packages/p2p/source/utils/get-peer-url.test.ts index 7522800aaf..0f287d7bb1 100644 --- a/packages/p2p/source/utils/get-peer-url.test.ts +++ b/packages/p2p/source/utils/get-peer-url.test.ts @@ -6,7 +6,7 @@ import { getPeerUrl } from "./get-peer-url"; describe("getPeerUrl", ({ each, assert }) => { each( "should return peer url", - ({ dataset, context }) => { + ({ dataset }) => { assert.equal(getPeerUrl(dataset[0]), dataset[1]); }, [ @@ -18,6 +18,14 @@ describe("getPeerUrl", ({ each, assert }) => { [{ ip: "127.0.0.1", port: 443, protocol: Enums.Api.Protocol.Http }, "https://127.0.0.1:443"], // defaults to HTTP if unknown protocol [{ ip: "127.0.0.1", port: 80, protocol: 5 }, "http://127.0.0.1:80"], + [ + { ip: "2001:3984:3989::104", port: 4003, protocol: Enums.Api.Protocol.Http }, + "http://[2001:3984:3989::104]:4003", + ], + [ + { ip: "2001:3984:3989::104", port: 443, protocol: Enums.Api.Protocol.Https }, + "https://[2001:3984:3989::104]:443", + ], ], ); }); diff --git a/packages/p2p/source/utils/get-peer-url.ts b/packages/p2p/source/utils/get-peer-url.ts index fbfab79332..287592c146 100644 --- a/packages/p2p/source/utils/get-peer-url.ts +++ b/packages/p2p/source/utils/get-peer-url.ts @@ -1,9 +1,11 @@ -import { Enums } from "@mainsail/constants"; +import type { Contracts } from "@mainsail/contracts"; -import { type shared } from "../socket-server/codecs/proto/protos.js"; +import { Enums } from "@mainsail/constants"; +import { IpAddress } from "@mainsail/utils"; -export const getPeerUrl = (peer: shared.IPeerLike): string => { +export const getPeerUrl = (peer: Contracts.P2P.PeerBroadcast): string => { let protocol = peer.protocol; + const host = IpAddress.normalizeAddress(peer.ip); // Heuristically check based on port first to match existing behavior. switch (peer.port) { @@ -22,14 +24,14 @@ export const getPeerUrl = (peer: shared.IPeerLike): string => { switch (protocol) { case Enums.Api.Protocol.Http: { - return `http://${peer.ip}:${peer.port}`; + return `http://${host}:${peer.port}`; } case Enums.Api.Protocol.Https: { - return `https://${peer.ip}:${peer.port}`; + return `https://${host}:${peer.port}`; } default: { // fallback to HTTP just in case - return `http://${peer.ip}:${peer.port}`; + return `http://${host}:${peer.port}`; } } }; diff --git a/packages/p2p/source/utils/normalize-url.test.ts b/packages/p2p/source/utils/normalize-url.test.ts index b7176d79b2..e7f36cd2d6 100644 --- a/packages/p2p/source/utils/normalize-url.test.ts +++ b/packages/p2p/source/utils/normalize-url.test.ts @@ -15,6 +15,8 @@ describe<{}>("normalizeUrl", ({ assert, each }) => { { given: "http://example.com/", normalized: "http://example.com/" }, { given: "http://example.com/path/", normalized: "http://example.com/path" }, { given: "http://example.com/path?123534", normalized: "http://example.com/path" }, + { given: "http://example.com/path#section", normalized: "http://example.com/path" }, + { given: "http://example.com/#section", normalized: "http://example.com/" }, { given: "https://example.com:4003/", normalized: "https://example.com:4003/" }, { given: "https://example.com:4003/path", normalized: "https://example.com:4003/path" }, ], diff --git a/packages/p2p/source/utils/normalize-url.ts b/packages/p2p/source/utils/normalize-url.ts index e4014b1177..e20fea5ed1 100644 --- a/packages/p2p/source/utils/normalize-url.ts +++ b/packages/p2p/source/utils/normalize-url.ts @@ -4,8 +4,9 @@ export function normalizeUrl(urlString: string): string { try { const url = new URL(urlString); - // Trim search parameters + // Trim search parameters and the fragment url.search = ""; + url.hash = ""; // Remove trailing slash for non-root paths if (url.pathname !== "/") {