From 314e4308ee5d8b33705e1ba3a977e4e01b51eced Mon Sep 17 00:00:00 2001 From: Priyanshubhartistm Date: Mon, 12 Oct 2026 20:35:28 +0530 Subject: [PATCH 1/3] feat(nip77): add negentropy wire messages and settings Signed-off-by: Priyanshubhartistm --- CONFIGURATION.md | 4 +++ resources/default-settings.yaml | 9 ++++++ src/@types/messages.ts | 55 +++++++++++++++++++++++++++++++-- src/@types/settings.ts | 19 ++++++++++++ src/schemas/message-schema.ts | 25 ++++++++++++++- src/utils/messages.ts | 17 ++++++++++ src/utils/settings-config.ts | 11 +++++++ 7 files changed, 137 insertions(+), 3 deletions(-) diff --git a/CONFIGURATION.md b/CONFIGURATION.md index 0cdd83603..1b76a3d8b 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -232,6 +232,10 @@ The settings below are listed in alphabetical order by name. Please keep this ta | nip66.timeouts.nip11Ms | NIP-11 fetch timeout in milliseconds. Defaults to 10000. | | nip66.timeouts.tlsMs | TLS probe timeout in milliseconds. Defaults to 10000. | | nip66.timeouts.wsRttMs | WebSocket open RTT probe timeout in milliseconds. Defaults to 10000. | +| nip77.enabled | Enable NIP-77 (Negentropy) set-reconciliation sync, so clients and other relays can find which events differ without downloading them all. Defaults to false. Only answers when `eventStore.backend` is `postgres`; other backends bring their own negentropy implementation. Each open session holds the ids of every event matching its filter in memory, so size the limits below for your hardware. | +| nip77.maxEventsPerSession | Most events one session may cover. A `NEG-OPEN` whose filter matches more is rejected with `NEG-ERR blocked:` (clients should split the sync into `since`/`until` windows). The filter's `limit` is ignored, since reconciliation needs the whole set. Defaults to 100000. | +| nip77.maxSessionsPerConnection | Concurrent negentropy sessions one websocket connection may hold; opening an id that is already open replaces it. Further `NEG-OPEN`s are rejected with `NEG-ERR blocked:`. Defaults to 3. | +| nip77.sessionIdleTimeoutMs | A session that receives no `NEG-MSG` for this many milliseconds is closed with `NEG-ERR closed:`. Defaults to 60000. | | paymentProcessors.lnbits.baseURL | Base URL of your Lnbits instance. | | paymentProcessors.lnbits.callbackBaseURL | Public-facing Nostream's Lnbits Callback URL. (e.g. https://relay.your-domain.com/callbacks/lnbits) | | paymentProcessors.lnurl.invoiceURL | [LUD-06 Pay Request](https://github.com/lnurl/luds/blob/luds/06.md) provider URL. (e.g. https://getalby.com/lnurlp/your-username) | diff --git a/resources/default-settings.yaml b/resources/default-settings.yaml index 115ea8454..d3d1cae69 100755 --- a/resources/default-settings.yaml +++ b/resources/default-settings.yaml @@ -148,6 +148,15 @@ nip56: # When true, events matching an actionable report are excluded from # REQ/COUNT results. Has no effect unless enabled is also true. hideActionableReports: false +nip77: + # NIP-77 negentropy sync. Only answers when eventStore.backend is postgres. + enabled: false + # Most events a single session may hold in memory. + maxEventsPerSession: 100000 + # Concurrent negentropy sessions per connection. + maxSessionsPerConnection: 3 + # Close a session with no NEG-MSG for this long. + sessionIdleTimeoutMs: 60000 network: maxPayloadSize: 524288 # Uncomment only when using a trusted reverse proxy and configuring trustedProxies. diff --git a/src/@types/messages.ts b/src/@types/messages.ts index 591d5d94c..312e3b48f 100644 --- a/src/@types/messages.ts +++ b/src/@types/messages.ts @@ -13,13 +13,35 @@ export enum MessageType { COUNT = 'COUNT', CLOSED = 'CLOSED', AUTH = 'AUTH', + NEG_OPEN = 'NEG-OPEN', + NEG_MSG = 'NEG-MSG', + NEG_CLOSE = 'NEG-CLOSE', + NEG_ERR = 'NEG-ERR', } -export type IncomingMessage = (SubscribeMessage | IncomingEventMessage | UnsubscribeMessage | CountMessage | AuthMessage) & { +export type IncomingMessage = ( + | SubscribeMessage + | IncomingEventMessage + | UnsubscribeMessage + | CountMessage + | AuthMessage + | NegOpenMessage + | NegMsgMessage + | NegCloseMessage +) & { [ContextMetadataKey]?: ContextMetadata } -export type OutgoingMessage = OutgoingEventMessage | EndOfStoredEventsNotice | NoticeMessage | CommandResult | CountResultMessage | ClosedMessage | AuthChallengeMessage +export type OutgoingMessage = + | OutgoingEventMessage + | EndOfStoredEventsNotice + | NoticeMessage + | CommandResult + | CountResultMessage + | ClosedMessage + | AuthChallengeMessage + | NegMsgMessage + | NegErrMessage export type SubscribeMessage = { [index in Range<2, 100>]: SubscriptionFilter @@ -103,3 +125,32 @@ export interface AuthChallengeMessage { 0: MessageType.AUTH 1: string } + +// NIP-77 +export interface NegOpenMessage { + 0: MessageType.NEG_OPEN + 1: SubscriptionId + 2: SubscriptionFilter + /** Hex-encoded initial negentropy message. */ + 3: string +} + +/** Sent in both directions; the payload is a hex-encoded negentropy message. */ +export interface NegMsgMessage { + 0: MessageType.NEG_MSG + 1: SubscriptionId + 2: string +} + +export interface NegCloseMessage { + 0: MessageType.NEG_CLOSE + 1: SubscriptionId +} + +export interface NegErrMessage { + 0: MessageType.NEG_ERR + 1: SubscriptionId + 2: string + /** For `blocked:` replies, the most records a session may hold. */ + 3?: number +} diff --git a/src/@types/settings.ts b/src/@types/settings.ts index 885ecdb11..9d9848ed6 100644 --- a/src/@types/settings.ts +++ b/src/@types/settings.ts @@ -314,6 +314,24 @@ export interface Nip45Settings { enabled?: boolean } +export interface Nip77Settings { + /** + * NIP-77 negentropy sync. Defaults to false. Even when true it only answers + * if `eventStore.backend` is postgres: other backends (e.g. strfry) bring + * their own negentropy implementation. + */ + enabled?: boolean + /** + * Most events one session may hold in memory; larger queries are rejected + * with `NEG-ERR blocked:`. Defaults to 100000. + */ + maxEventsPerSession?: number + /** Concurrent negentropy sessions allowed on one connection. Defaults to 3. */ + maxSessionsPerConnection?: number + /** A session with no `NEG-MSG` for this long is closed with `NEG-ERR closed:`. Defaults to 60000. */ + sessionIdleTimeoutMs?: number +} + export interface Nip50Settings { enabled?: boolean language?: string @@ -491,5 +509,6 @@ export interface Settings { nip50?: Nip50Settings nip56?: Nip56Settings nip66?: Nip66Settings + nip77?: Nip77Settings wot?: WoTSettings } diff --git a/src/schemas/message-schema.ts b/src/schemas/message-schema.ts index c246b4eee..6a928d4e4 100644 --- a/src/schemas/message-schema.ts +++ b/src/schemas/message-schema.ts @@ -55,7 +55,30 @@ export const countMessageSchema = z export const closeMessageSchema = z.tuple([z.literal(MessageType.CLOSE), subscriptionSchema]) +// NIP-77 +const negentropyHexSchema = z.string().regex(/^(?:[0-9a-fA-F]{2})+$/, 'negentropy message must be hex-encoded bytes') + +export const negOpenMessageSchema = z.tuple([ + z.literal(MessageType.NEG_OPEN), + z.string().max(256).min(1), + filterSchema, + negentropyHexSchema, +]) + +export const negMsgMessageSchema = z.tuple([z.literal(MessageType.NEG_MSG), z.string().max(256).min(1), negentropyHexSchema]) + +export const negCloseMessageSchema = z.tuple([z.literal(MessageType.NEG_CLOSE), z.string().max(256).min(1)]) + // NIP-42 export const authMessageSchema = z.tuple([z.literal(MessageType.AUTH), eventSchema]) -export const messageSchema = z.union([eventMessageSchema, reqMessageSchema, closeMessageSchema, countMessageSchema, authMessageSchema]) +export const messageSchema = z.union([ + eventMessageSchema, + reqMessageSchema, + closeMessageSchema, + countMessageSchema, + authMessageSchema, + negOpenMessageSchema, + negMsgMessageSchema, + negCloseMessageSchema, +]) diff --git a/src/utils/messages.ts b/src/utils/messages.ts index 3c2544421..bc6b3ff6b 100644 --- a/src/utils/messages.ts +++ b/src/utils/messages.ts @@ -7,6 +7,8 @@ import { IncomingEventMessage, IncomingRelayedEventMessage, MessageType, + NegErrMessage, + NegMsgMessage, NoticeMessage, OutgoingMessage, SubscribeMessage, @@ -50,6 +52,21 @@ export const createAuthChallengeMessage = (challenge: string): AuthChallengeMess return [MessageType.AUTH, challenge] } +// NIP-77 +export const createNegMsgMessage = (subscriptionId: SubscriptionId, message: string): NegMsgMessage => { + return [MessageType.NEG_MSG, subscriptionId, message] +} + +export const createNegErrMessage = ( + subscriptionId: SubscriptionId, + reason: string, + maxRecords?: number, +): NegErrMessage => { + return maxRecords === undefined + ? [MessageType.NEG_ERR, subscriptionId, reason] + : [MessageType.NEG_ERR, subscriptionId, reason, maxRecords] +} + export const createSubscriptionMessage = ( subscriptionId: SubscriptionId, filters: SubscriptionFilter[], diff --git a/src/utils/settings-config.ts b/src/utils/settings-config.ts index 93fb91498..937449b16 100644 --- a/src/utils/settings-config.ts +++ b/src/utils/settings-config.ts @@ -621,6 +621,17 @@ export const validateSettings = (settings: Settings): ValidationIssue[] => { }) } + const nip77 = settings.nip77 + if (nip77?.enabled) { + for (const key of ['maxEventsPerSession', 'maxSessionsPerConnection', 'sessionIdleTimeoutMs'] as const) { + const value = nip77[key] + // Wrong types (e.g. a string) are already reported against the default settings schema. + if (typeof value === 'number' && !(Number.isInteger(value) && value > 0)) { + issues.push({ path: `nip77.${key}`, message: `${key} must be a positive integer` }) + } + } + } + const pow = settings.limits?.event?.pow if (pow?.enabled) { if (!(pow.floorBits >= 0) || !(pow.floorBits <= pow.ceilingBits)) { From 89ebc37c18959f02503964d5d6f2180fdbaf5554 Mon Sep 17 00:00:00 2001 From: Priyanshubhartistm Date: Mon, 12 Oct 2026 20:36:51 +0530 Subject: [PATCH 2/3] feat(nip77): serve negentropy sessions over the websocket Signed-off-by: Priyanshubhartistm --- .changeset/nip77-wire-protocol.md | 5 + .knip.json | 1 - CONFIGURATION.md | 2 +- src/@types/adapters.ts | 3 + src/@types/negentropy.ts | 42 ++++++ src/@types/repositories.ts | 2 +- src/adapters/web-socket-adapter.ts | 19 +++ src/factories/message-handler-factory.ts | 9 ++ src/handlers/neg-close-message-handler.ts | 12 ++ src/handlers/neg-msg-message-handler.ts | 60 ++++++++ src/handlers/neg-open-message-handler.ts | 160 ++++++++++++++++++++++ src/repositories/event-repository.ts | 9 +- src/utils/negentropy-sessions.ts | 117 ++++++++++++++++ src/utils/nip42.ts | 26 ++++ src/utils/nip77.ts | 21 +++ 15 files changed, 483 insertions(+), 5 deletions(-) create mode 100644 .changeset/nip77-wire-protocol.md create mode 100644 src/@types/negentropy.ts create mode 100644 src/handlers/neg-close-message-handler.ts create mode 100644 src/handlers/neg-msg-message-handler.ts create mode 100644 src/handlers/neg-open-message-handler.ts create mode 100644 src/utils/negentropy-sessions.ts create mode 100644 src/utils/nip77.ts diff --git a/.changeset/nip77-wire-protocol.md b/.changeset/nip77-wire-protocol.md new file mode 100644 index 000000000..1fce17bf4 --- /dev/null +++ b/.changeset/nip77-wire-protocol.md @@ -0,0 +1,5 @@ +--- +"nostream": minor +--- + +feat(nip77): serve NEG-OPEN, NEG-MSG and NEG-CLOSE negentropy sync for the PostgreSQL event store diff --git a/.knip.json b/.knip.json index 4151e7b65..b42e6a95b 100644 --- a/.knip.json +++ b/.knip.json @@ -15,7 +15,6 @@ ], "ignore": [ ".nostr/**", - "src/utils/negentropy.ts", "src/utils/relay-probe/**" ], "commitlint": false, diff --git a/CONFIGURATION.md b/CONFIGURATION.md index 1b76a3d8b..21be4c16b 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -232,7 +232,7 @@ The settings below are listed in alphabetical order by name. Please keep this ta | nip66.timeouts.nip11Ms | NIP-11 fetch timeout in milliseconds. Defaults to 10000. | | nip66.timeouts.tlsMs | TLS probe timeout in milliseconds. Defaults to 10000. | | nip66.timeouts.wsRttMs | WebSocket open RTT probe timeout in milliseconds. Defaults to 10000. | -| nip77.enabled | Enable NIP-77 (Negentropy) set-reconciliation sync, so clients and other relays can find which events differ without downloading them all. Defaults to false. Only answers when `eventStore.backend` is `postgres`; other backends bring their own negentropy implementation. Each open session holds the ids of every event matching its filter in memory, so size the limits below for your hardware. | +| nip77.enabled | Enable NIP-77 (Negentropy) set-reconciliation sync, so clients and other relays can find which events differ without downloading them all. Defaults to false. Only answers when `eventStore.backend` is `postgres`; other backends bring their own negentropy implementation. When `nip42.restrictedReads` is enabled, reconciliation would list the ids of restricted events, so a `NEG-OPEN` filter that could match one (it names a restricted kind or no kinds at all) is only accepted when scoped by `authors` or `#p` to pubkeys the client has authenticated as, and is re-checked on every `NEG-MSG`; otherwise the session is closed with `NEG-ERR auth-required:`. Each open session holds the ids of every event matching its filter in memory, so size the limits below for your hardware. | | nip77.maxEventsPerSession | Most events one session may cover. A `NEG-OPEN` whose filter matches more is rejected with `NEG-ERR blocked:` (clients should split the sync into `since`/`until` windows). The filter's `limit` is ignored, since reconciliation needs the whole set. Defaults to 100000. | | nip77.maxSessionsPerConnection | Concurrent negentropy sessions one websocket connection may hold; opening an id that is already open replaces it. Further `NEG-OPEN`s are rejected with `NEG-ERR blocked:`. Defaults to 3. | | nip77.sessionIdleTimeoutMs | A session that receives no `NEG-MSG` for this many milliseconds is closed with `NEG-ERR closed:`. Defaults to 60000. | diff --git a/src/@types/adapters.ts b/src/@types/adapters.ts index f84f542cf..b8bb2dae6 100644 --- a/src/@types/adapters.ts +++ b/src/@types/adapters.ts @@ -1,4 +1,5 @@ import { EventEmitter } from 'node:stream' +import { INegentropySessions } from './negentropy' import { SubscriptionFilter } from './subscription' export interface IWebSocketServerAdapter extends EventEmitter, IWebServerAdapter { @@ -20,6 +21,8 @@ export type IWebSocketAdapter = EventEmitter & { /** Returns false if this AUTH event id was already accepted on this socket. */ addAuthenticatedPubkey(pubkey: string, authEventId: string): boolean drainAndClose(reason?: string): void + /** NIP-77 sessions of this connection. */ + getNegentropySessions(): INegentropySessions } export interface ICacheAdapter { diff --git a/src/@types/negentropy.ts b/src/@types/negentropy.ts new file mode 100644 index 000000000..011fe9a81 --- /dev/null +++ b/src/@types/negentropy.ts @@ -0,0 +1,42 @@ +import { Negentropy } from '../utils/negentropy' +import { SubscriptionFilter, SubscriptionId } from './subscription' + +/** Proves a caller still owns a subscription id after awaiting; see `INegentropySessions.reserve`. */ +export type NegentropySessionToken = symbol + +export interface NegentropySession { + negentropy: Negentropy + /** The filter the session was opened with, so access can be re-checked on every message. */ + filter: SubscriptionFilter +} + +export interface INegentropySessions { + /** + * Claims `subscriptionId` for a new session, closing any session already open under that id. + * Returns undefined when the connection already holds `maxLimit` other sessions, or already has + * `maxLimit` event reads in flight -- including reads of requests that have since been replaced, + * which keep running until `endRead` is called. The caller must call `endRead` for a token it got. + */ + reserve(subscriptionId: SubscriptionId, maxLimit: number): NegentropySessionToken | undefined + /** Marks the event read started for a reservation as finished, successfully or not. */ + endRead(token: NegentropySessionToken): void + /** Whether `token` is still the reservation for `subscriptionId`, i.e. nobody replaced or closed it. */ + isCurrent(subscriptionId: SubscriptionId, token: NegentropySessionToken): boolean + /** + * Turns a reservation into a live session. Returns false when the reservation was closed or + * replaced while the caller was busy (e.g. reading events), in which case nothing is stored. + */ + activate( + subscriptionId: SubscriptionId, + token: NegentropySessionToken, + session: NegentropySession, + idleTimeoutMs: number, + ): boolean + /** The live session for an id, with its idle timer restarted. */ + touch(subscriptionId: SubscriptionId): NegentropySession | undefined + /** Drops a reservation that never became a session; a no-op if it has since been replaced. */ + release(subscriptionId: SubscriptionId, token: NegentropySessionToken): void + close(subscriptionId: SubscriptionId): boolean + closeAll(): void + getSessionIds(): SubscriptionId[] +} diff --git a/src/@types/repositories.ts b/src/@types/repositories.ts index 7d45cf102..d4ce0d4f3 100644 --- a/src/@types/repositories.ts +++ b/src/@types/repositories.ts @@ -44,7 +44,7 @@ export interface IEventRepository { upsertMany(events: Event[]): Promise findByFilters(filters: SubscriptionFilter[]): IQueryResult countByFilters(filters: SubscriptionFilter[]): Promise - findReconciliationItems(filter: SubscriptionFilter): IQueryResult + findReconciliationItems(filter: SubscriptionFilter, limit?: number): IQueryResult deleteByPubkeyAndIds(pubkey: Pubkey, ids: EventId[]): Promise deleteByPubkeyExceptKinds(pubkey: Pubkey, excludedKinds: number[]): Promise hasActiveRequestToVanish(pubkey: Pubkey): Promise diff --git a/src/adapters/web-socket-adapter.ts b/src/adapters/web-socket-adapter.ts index 8f107f3c2..6ac493849 100644 --- a/src/adapters/web-socket-adapter.ts +++ b/src/adapters/web-socket-adapter.ts @@ -8,11 +8,14 @@ import { ContextMetadata, Factory } from '../@types/base' import { createAuthChallengeMessage, createClosedMessage, + createNegErrMessage, createNoticeMessage, createOutgoingEventMessage, } from '../utils/messages' import { IAbortable, IMessageHandler } from '../@types/message-handlers' import { IncomingMessage, MessageType, OutgoingMessage } from '../@types/messages' +import { INegentropySessions } from '../@types/negentropy' +import { NegentropySessions } from '../utils/negentropy-sessions' import { IWebSocketAdapter, IWebSocketServerAdapter } from '../@types/adapters' import { SubscriptionFilter, SubscriptionId } from '../@types/subscription' import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../constants/adapter' @@ -42,6 +45,8 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter private alive: boolean private subscriptions: Map private readonly session: Nip42SessionManager + // NIP-77 + private readonly negentropySessions: NegentropySessions public constructor( private readonly client: WebSocket, @@ -54,6 +59,9 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter super() this.alive = true this.subscriptions = new Map() + this.negentropySessions = new NegentropySessions((subscriptionId) => + this.sendMessage(createNegErrMessage(subscriptionId, 'closed: negentropy session timed out')), + ) this.clientId = Buffer.from(this.request.headers['sec-websocket-key'] as string, 'base64').toString('hex') @@ -175,11 +183,21 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter this.sendMessage(createClosedMessage(subscriptionId, `error: ${reason}`)) }) + this.negentropySessions.getSessionIds().forEach((subscriptionId) => { + this.sendMessage(createNegErrMessage(subscriptionId, `closed: ${reason}`)) + }) + this.negentropySessions.closeAll() + if (this.client.readyState === WebSocket.OPEN) { this.client.close(1001, reason) } } + // NIP-77 + public getNegentropySessions(): INegentropySessions { + return this.negentropySessions + } + // NIP-42 public getChallenge(): string { return this.session.getChallenge() @@ -349,6 +367,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter recordWebsocketConnectionClosed() this.alive = false this.subscriptions.clear() + this.negentropySessions.closeAll() this.session.clear() const handlers = abortableMessageHandlers.get(this.client) diff --git a/src/factories/message-handler-factory.ts b/src/factories/message-handler-factory.ts index b11f3339e..b3bf9ccd2 100644 --- a/src/factories/message-handler-factory.ts +++ b/src/factories/message-handler-factory.ts @@ -13,6 +13,9 @@ import { getCacheClient } from '../cache/client' import { AuthMessageHandler } from '../handlers/auth-message-handler' import { CountMessageHandler } from '../handlers/count-message-handler' import { EventMessageHandler } from '../handlers/event-message-handler' +import { NegCloseMessageHandler } from '../handlers/neg-close-message-handler' +import { NegMsgMessageHandler } from '../handlers/neg-msg-message-handler' +import { NegOpenMessageHandler } from '../handlers/neg-open-message-handler' import { SubscribeMessageHandler } from '../handlers/subscribe-message-handler' import { UnsubscribeMessageHandler } from '../handlers/unsubscribe-message-handler' import { eventStrategyFactory } from './event-strategy-factory' @@ -74,6 +77,12 @@ export const messageHandlerFactory = return new CountMessageHandler(adapter, eventRepository, createSettings) case MessageType.AUTH: return new AuthMessageHandler(adapter, createSettings) + case MessageType.NEG_OPEN: + return new NegOpenMessageHandler(adapter, eventRepository, createSettings) + case MessageType.NEG_MSG: + return new NegMsgMessageHandler(adapter, createSettings) + case MessageType.NEG_CLOSE: + return new NegCloseMessageHandler(adapter) default: throw new Error(`Unknown message type: ${String(message[0]).substring(0, 64)}`) } diff --git a/src/handlers/neg-close-message-handler.ts b/src/handlers/neg-close-message-handler.ts new file mode 100644 index 000000000..e4ef1f314 --- /dev/null +++ b/src/handlers/neg-close-message-handler.ts @@ -0,0 +1,12 @@ +import { IMessageHandler } from '../@types/message-handlers' +import { IWebSocketAdapter } from '../@types/adapters' +import { NegCloseMessage } from '../@types/messages' + +export class NegCloseMessageHandler implements IMessageHandler { + public constructor(private readonly webSocket: IWebSocketAdapter) {} + + // NIP-77 defines no reply to NEG-CLOSE, and closing an unknown id is harmless. + public async handleMessage(message: NegCloseMessage): Promise { + this.webSocket.getNegentropySessions().close(message[1]) + } +} diff --git a/src/handlers/neg-msg-message-handler.ts b/src/handlers/neg-msg-message-handler.ts new file mode 100644 index 000000000..158536240 --- /dev/null +++ b/src/handlers/neg-msg-message-handler.ts @@ -0,0 +1,60 @@ +import { createNegErrMessage, createNegMsgMessage } from '../utils/messages' +import { IMessageHandler } from '../@types/message-handlers' +import { IWebSocketAdapter } from '../@types/adapters' +import { NegMsgMessage, OutgoingMessage } from '../@types/messages' +import { Settings } from '../@types/settings' +import { WebSocketAdapterEvent } from '../constants/adapter' +import { createLogger } from '../factories/logger-factory' +import { isNip77Active } from '../utils/nip77' +import { isReconciliationAuthorized } from '../utils/nip42' + +const logger = createLogger('neg-msg-message-handler') + +export class NegMsgMessageHandler implements IMessageHandler { + public constructor( + private readonly webSocket: IWebSocketAdapter, + private readonly settings: () => Settings, + ) {} + + public async handleMessage(message: NegMsgMessage): Promise { + const subscriptionId = message[1] + const sessions = this.webSocket.getNegentropySessions() + + if (!isNip77Active(this.settings())) { + // Turned off while a session was open: end it instead of leaving it half-served. + sessions.close(subscriptionId) + this.reply(createNegErrMessage(subscriptionId, 'blocked: NIP-77 is disabled by relay configuration')) + return + } + + const session = sessions.touch(subscriptionId) + if (!session) { + this.reply(createNegErrMessage(subscriptionId, 'closed: unknown negentropy subscription')) + return + } + + // Access can lapse while a session is open (e.g. NIP-42 auth expiring), so check it every time. + if (!isReconciliationAuthorized(this.settings(), session.filter, () => this.webSocket.getAuthenticatedPubkeys())) { + sessions.close(subscriptionId) + this.reply( + createNegErrMessage(subscriptionId, 'auth-required: authentication is required to sync these event kinds'), + ) + return + } + + try { + const { output } = session.negentropy.reconcile(Buffer.from(message[2], 'hex')) + + this.reply(createNegMsgMessage(subscriptionId, Buffer.from(output as Uint8Array).toString('hex'))) + } catch (error) { + // NIP-77: after a NEG-ERR the subscription is closed. + sessions.close(subscriptionId) + logger('NEG-MSG %s sent a malformed negentropy message: %o', subscriptionId, error) + this.reply(createNegErrMessage(subscriptionId, 'invalid: malformed negentropy message')) + } + } + + private reply(message: OutgoingMessage): void { + this.webSocket.emit(WebSocketAdapterEvent.Message, message) + } +} diff --git a/src/handlers/neg-open-message-handler.ts b/src/handlers/neg-open-message-handler.ts new file mode 100644 index 000000000..03c0e7ce4 --- /dev/null +++ b/src/handlers/neg-open-message-handler.ts @@ -0,0 +1,160 @@ +import { IWebSocketAdapter } from '../@types/adapters' +import { createNegErrMessage, createNegMsgMessage } from '../utils/messages' +import { getNip77Limits, isNip77Active, NEGENTROPY_FRAME_SIZE_LIMIT } from '../utils/nip77' +import { INegentropySessions } from '../@types/negentropy' +import { Negentropy, NegentropyStorageVector } from '../utils/negentropy' +import { IEventRepository, ReconciliationItem } from '../@types/repositories' +import { IMessageHandler } from '../@types/message-handlers' +import { NegOpenMessage, OutgoingMessage } from '../@types/messages' +import { Settings } from '../@types/settings' +import { SubscriptionFilter, SubscriptionId } from '../@types/subscription' +import { WebSocketAdapterEvent } from '../constants/adapter' +import { countFilterValues } from '../utils/filter' +import { createLogger } from '../factories/logger-factory' +import { isReconciliationAuthorized } from '../utils/nip42' + +const logger = createLogger('neg-open-message-handler') + +// Negentropy protocol versions are 0x60-0x6f; anything else is not a negentropy message at all. +const isNegentropyVersionByte = (byte: number | undefined): boolean => + byte !== undefined && byte >= 0x60 && byte <= 0x6f + +export class NegOpenMessageHandler implements IMessageHandler { + public constructor( + private readonly webSocket: IWebSocketAdapter, + private readonly eventRepository: IEventRepository, + private readonly settings: () => Settings, + ) {} + + public async handleMessage(message: NegOpenMessage): Promise { + const subscriptionId = message[1] + const filter = message[2] + const initialMessage = Buffer.from(message[3], 'hex') + + const settings = this.settings() + const sessions = this.webSocket.getNegentropySessions() + + if (!isNip77Active(settings)) { + this.reject(sessions, subscriptionId, 'blocked: NIP-77 is disabled by relay configuration') + return + } + + const rejection = this.getRejection(subscriptionId, filter, initialMessage) + if (rejection) { + logger('NEG-OPEN %s with %o rejected: %s', subscriptionId, filter, rejection) + this.reject(sessions, subscriptionId, rejection) + return + } + + const { maxEventsPerSession, maxSessionsPerConnection, sessionIdleTimeoutMs } = getNip77Limits(settings) + + const token = sessions.reserve(subscriptionId, maxSessionsPerConnection) + if (!token) { + this.reject( + sessions, + subscriptionId, + `blocked: too many negentropy sessions or reads in progress on this connection (max ${maxSessionsPerConnection})`, + ) + return + } + + let items: ReconciliationItem[] + try { + // One row past the limit tells us the set is too big without loading all of it. + items = await this.eventRepository.findReconciliationItems(filter, maxEventsPerSession + 1) + } catch (error) { + if (sessions.isCurrent(subscriptionId, token)) { + sessions.release(subscriptionId, token) + logger('NEG-OPEN %s failed: %o', subscriptionId, error) + // Keep this generic so internal errors are not leaked to clients. + this.reply(createNegErrMessage(subscriptionId, 'error: unable to process negentropy request')) + } + return + } finally { + sessions.endRead(token) + } + + // The client may have closed or replaced this NEG-OPEN while events were being read. Then the + // newer request owns the id: say nothing, since an error here would tell the client to drop it. + if (!sessions.isCurrent(subscriptionId, token)) { + return + } + + // Nothing below awaits, so ownership cannot change before the reply. + if (items.length > maxEventsPerSession) { + sessions.release(subscriptionId, token) + this.reply(createNegErrMessage(subscriptionId, 'blocked: this query is too big', maxEventsPerSession)) + return + } + + try { + const storage = new NegentropyStorageVector() + for (const item of items) { + storage.insert(item.event_created_at, item.event_id) + } + storage.seal() + + const negentropy = new Negentropy(storage, NEGENTROPY_FRAME_SIZE_LIMIT) + + let output: Uint8Array + try { + output = negentropy.reconcile(initialMessage).output as Uint8Array + } catch (error) { + // The only input from the client here is the initial message, so this is its fault. + sessions.release(subscriptionId, token) + logger('NEG-OPEN %s sent a malformed negentropy message: %o', subscriptionId, error) + this.reply(createNegErrMessage(subscriptionId, 'invalid: malformed negentropy message')) + return + } + + sessions.activate(subscriptionId, token, { negentropy, filter }, sessionIdleTimeoutMs) + this.reply(createNegMsgMessage(subscriptionId, Buffer.from(output).toString('hex'))) + } catch (error) { + sessions.release(subscriptionId, token) + logger('NEG-OPEN %s failed: %o', subscriptionId, error) + this.reply(createNegErrMessage(subscriptionId, 'error: unable to process negentropy request')) + } + } + + private getRejection( + subscriptionId: SubscriptionId, + filter: SubscriptionFilter, + initialMessage: Buffer, + ): string | undefined { + const settings = this.settings() + const subscriptionLimits = settings.limits?.client?.subscription + + const maxFilterValues = subscriptionLimits?.maxFilterValues ?? 0 + if (maxFilterValues > 0 && countFilterValues(filter) > maxFilterValues) { + return `blocked: too many filter values (max ${maxFilterValues})` + } + + if ( + typeof subscriptionLimits?.maxSubscriptionIdLength === 'number' && + subscriptionId.length > subscriptionLimits.maxSubscriptionIdLength + ) { + return `invalid: subscription id too long (max ${subscriptionLimits.maxSubscriptionIdLength})` + } + + if (!isNegentropyVersionByte(initialMessage[0])) { + return 'invalid: malformed negentropy message' + } + + // NIP-42: reconciling reveals the id of every event the filter matches, so a filter that could + // match a restricted kind must be scoped to the client's own pubkeys. + if (!isReconciliationAuthorized(settings, filter, () => this.webSocket.getAuthenticatedPubkeys())) { + return 'auth-required: authentication is required to sync these event kinds' + } + } + + // NIP-77: after a NEG-ERR the subscription is closed, so a rejected NEG-OPEN must also end any + // session the client already had under that id; otherwise it keeps serving, and holding a slot. + private reject(sessions: INegentropySessions, subscriptionId: SubscriptionId, reason: string): void { + sessions.close(subscriptionId) + this.reply(createNegErrMessage(subscriptionId, reason)) + } + + private reply(message: OutgoingMessage): void { + this.webSocket.emit(WebSocketAdapterEvent.Message, message) + } +} diff --git a/src/repositories/event-repository.ts b/src/repositories/event-repository.ts index 3ac583162..26431db60 100644 --- a/src/repositories/event-repository.ts +++ b/src/repositories/event-repository.ts @@ -138,9 +138,10 @@ export class EventRepository implements IEventRepository { * Every live event matching the filter as (created_at, id), oldest first, for NIP-77 set * reconciliation. The filter's limit is ignored on purpose: reconciliation compares whole sets, so * truncating one side would report the cut-off events as missing. Deleted and expired events are - * excluded because they must never be advertised to a peer. + * excluded because they must never be advertised to a peer. `limit` is a safety cap for callers + * that refuse oversized sets (they ask for one more row than they accept); it is never a window. */ - public findReconciliationItems(filter: SubscriptionFilter): IQueryResult { + public findReconciliationItems(filter: SubscriptionFilter, limit?: number): IQueryResult { logger('querying reconciliation items for %o', filter) const now = Math.floor(Date.now() / 1000) @@ -156,6 +157,10 @@ export class EventRepository implements IEventRepository { .orderBy('events.event_created_at', 'asc') .orderBy('events.event_id', 'asc') + if (typeof limit === 'number') { + builder.limit(limit) + } + return builder as unknown as IQueryResult } diff --git a/src/utils/negentropy-sessions.ts b/src/utils/negentropy-sessions.ts new file mode 100644 index 000000000..952a213fc --- /dev/null +++ b/src/utils/negentropy-sessions.ts @@ -0,0 +1,117 @@ +import { INegentropySessions, NegentropySession, NegentropySessionToken } from '../@types/negentropy' +import { SubscriptionId } from '../@types/subscription' + +interface Entry { + token: NegentropySessionToken + session?: NegentropySession + idleTimeoutMs?: number + idleTimer?: NodeJS.Timeout +} + +/** The NIP-77 sessions of one websocket connection. Ids are a separate namespace from REQ ids. */ +export class NegentropySessions implements INegentropySessions { + private readonly entries = new Map() + // Event reads still running. A replaced request's read is not cancelled, so it stays counted here + // until it finishes; otherwise repeating NEG-OPEN on one id would stack up unbounded queries. + private readonly pendingReads = new Set() + + public constructor(private readonly onIdleTimeout: (subscriptionId: SubscriptionId) => void) {} + + public reserve(subscriptionId: SubscriptionId, maxLimit: number): NegentropySessionToken | undefined { + // NIP-77: opening an id that is already open closes the old session first. + this.close(subscriptionId) + + if (this.entries.size >= maxLimit || this.pendingReads.size >= maxLimit) { + return undefined + } + + const token = Symbol(subscriptionId) + this.entries.set(subscriptionId, { token }) + this.pendingReads.add(token) + + return token + } + + public endRead(token: NegentropySessionToken): void { + this.pendingReads.delete(token) + } + + public isCurrent(subscriptionId: SubscriptionId, token: NegentropySessionToken): boolean { + return this.entries.get(subscriptionId)?.token === token + } + + public activate( + subscriptionId: SubscriptionId, + token: NegentropySessionToken, + session: NegentropySession, + idleTimeoutMs: number, + ): boolean { + const entry = this.entries.get(subscriptionId) + if (!entry || entry.token !== token) { + return false + } + + // Activating means the read is over. + this.pendingReads.delete(token) + entry.session = session + entry.idleTimeoutMs = idleTimeoutMs + this.startIdleTimer(subscriptionId, entry) + + return true + } + + public touch(subscriptionId: SubscriptionId): NegentropySession | undefined { + const entry = this.entries.get(subscriptionId) + if (!entry?.session) { + return undefined + } + + this.startIdleTimer(subscriptionId, entry) + + return entry.session + } + + public release(subscriptionId: SubscriptionId, token: NegentropySessionToken): void { + if (this.entries.get(subscriptionId)?.token === token) { + this.close(subscriptionId) + } + } + + public close(subscriptionId: SubscriptionId): boolean { + const entry = this.entries.get(subscriptionId) + if (!entry) { + return false + } + + clearTimeout(entry.idleTimer) + this.entries.delete(subscriptionId) + + return true + } + + public closeAll(): void { + for (const subscriptionId of [...this.entries.keys()]) { + this.close(subscriptionId) + } + } + + public getSessionIds(): SubscriptionId[] { + return [...this.entries.entries()].filter(([, entry]) => entry.session).map(([subscriptionId]) => subscriptionId) + } + + private startIdleTimer(subscriptionId: SubscriptionId, entry: Entry): void { + clearTimeout(entry.idleTimer) + if (!entry.idleTimeoutMs || entry.idleTimeoutMs <= 0) { + return + } + + entry.idleTimer = setTimeout(() => { + if (this.entries.get(subscriptionId) === entry) { + this.entries.delete(subscriptionId) + this.onIdleTimeout(subscriptionId) + } + }, entry.idleTimeoutMs) + // An idle session must never keep the process alive. + entry.idleTimer.unref() + } +} diff --git a/src/utils/nip42.ts b/src/utils/nip42.ts index 144695436..6d20f824d 100644 --- a/src/utils/nip42.ts +++ b/src/utils/nip42.ts @@ -89,6 +89,32 @@ export const isSubscriptionAuthRequired = ( return getAuthenticatedPubkeys().size === 0 } +// NIP-77 reconciliation, like COUNT, can't be filtered per event: it reveals the id of every event the +// filter matches. REQ hides restricted events per event, so a filter that *could* match one -- one +// naming a restricted kind or naming no kinds at all -- must be scoped to the client's own pubkeys. +export const isReconciliationAuthorized = ( + settings: Settings | undefined, + filter: SubscriptionFilter, + getAuthenticatedPubkeys: () => ReadonlySet, +): boolean => { + const restrictedKinds = getRestrictedReadKinds(settings) + if (!restrictedKinds.length) { + return true + } + + const mayMatchRestricted = + !Array.isArray(filter.kinds) || filter.kinds.some((kind) => isKindRestricted(restrictedKinds, kind)) + if (!mayMatchRestricted) { + return true + } + + const authenticatedPubkeys = getAuthenticatedPubkeys() + const isScopedToClient = (values?: Pubkey[]) => + Array.isArray(values) && values.length > 0 && values.every((value) => authenticatedPubkeys.has(value)) + + return isScopedToClient(filter.authors) || isScopedToClient(filter['#p']) +} + // COUNT can't be filtered per event, so a restricted-kind filter must be // scoped to the client's own pubkeys via authors/#p. export const isCountAuthorized = ( diff --git a/src/utils/nip77.ts b/src/utils/nip77.ts new file mode 100644 index 000000000..3bdd670f1 --- /dev/null +++ b/src/utils/nip77.ts @@ -0,0 +1,21 @@ +import { EventStoreBackend } from '../constants/base' +import { getEventStoreBackend } from './event-store' +import { Settings } from '../@types/settings' + +export const DEFAULT_MAX_EVENTS_PER_SESSION = 100000 +export const DEFAULT_MAX_SESSIONS_PER_CONNECTION = 3 +export const DEFAULT_SESSION_IDLE_TIMEOUT_MS = 60000 + +// Binary size of one negentropy message we send, before hex encoding. Larger answers are cut short +// and finished over further rounds. 60 kB matches strfry's default. +export const NEGENTROPY_FRAME_SIZE_LIMIT = 60000 + +// NIP-77 is served natively only from PostgreSQL; other backends (e.g. strfry) bring their own. +export const isNip77Active = (settings: Settings): boolean => + settings.nip77?.enabled === true && getEventStoreBackend(settings) === EventStoreBackend.POSTGRES + +export const getNip77Limits = (settings: Settings) => ({ + maxEventsPerSession: settings.nip77?.maxEventsPerSession ?? DEFAULT_MAX_EVENTS_PER_SESSION, + maxSessionsPerConnection: settings.nip77?.maxSessionsPerConnection ?? DEFAULT_MAX_SESSIONS_PER_CONNECTION, + sessionIdleTimeoutMs: settings.nip77?.sessionIdleTimeoutMs ?? DEFAULT_SESSION_IDLE_TIMEOUT_MS, +}) From 32d46c371a97453b08350626b689f7fd507757c0 Mon Sep 17 00:00:00 2001 From: Priyanshubhartistm Date: Mon, 12 Oct 2026 20:38:21 +0530 Subject: [PATCH 3/3] test(nip77): cover negentropy wire protocol and sessions Signed-off-by: Priyanshubhartistm --- test/unit/adapters/web-socket-adapter.spec.ts | 88 +++ .../factories/message-handler-factory.spec.ts | 21 + .../neg-close-message-handler.spec.ts | 60 ++ .../handlers/neg-msg-message-handler.spec.ts | 192 ++++++ .../handlers/neg-open-message-handler.spec.ts | 576 ++++++++++++++++++ .../repositories/event-repository.spec.ts | 6 + test/unit/schemas/message-schema.spec.ts | 81 +++ test/unit/utils/messages.spec.ts | 31 + test/unit/utils/negentropy-sessions.spec.ts | 232 +++++++ test/unit/utils/nip42.spec.ts | 59 ++ test/unit/utils/nip77.spec.ts | 59 ++ test/unit/utils/settings-config.spec.ts | 41 ++ 12 files changed, 1446 insertions(+) create mode 100644 test/unit/handlers/neg-close-message-handler.spec.ts create mode 100644 test/unit/handlers/neg-msg-message-handler.spec.ts create mode 100644 test/unit/handlers/neg-open-message-handler.spec.ts create mode 100644 test/unit/utils/negentropy-sessions.spec.ts create mode 100644 test/unit/utils/nip77.spec.ts diff --git a/test/unit/adapters/web-socket-adapter.spec.ts b/test/unit/adapters/web-socket-adapter.spec.ts index b5039996d..751094dbe 100644 --- a/test/unit/adapters/web-socket-adapter.spec.ts +++ b/test/unit/adapters/web-socket-adapter.spec.ts @@ -13,6 +13,7 @@ const { expect } = chai import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../../../src/constants/adapter' import { IWebSocketServerAdapter } from '../../../src/@types/adapters' +import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy' import { WebSocketAdapter } from '../../../src/adapters/web-socket-adapter' import { markActionableTarget, resetHiddenContentCache } from '../../../src/utils/hidden-content-cache' @@ -976,4 +977,91 @@ describe('WebSocketAdapter', () => { adapter2.removeAllListeners() }) }) + + describe('NIP-77 negentropy sessions', () => { + const emptyStorage = () => { + const storage = new NegentropyStorageVector() + storage.seal() + + return new Negentropy(storage) + } + + const openSession = (id: string, idleTimeoutMs = 60000) => { + const sessions = adapter.getNegentropySessions() + sessions.activate( + id, + sessions.reserve(id, 3) as symbol, + { negentropy: emptyStorage(), filter: {} }, + idleTimeoutMs, + ) + } + + const sentMessages = () => client.send.getCalls().map((call: any) => JSON.parse(call.args[0])) + + const triggerMessage = async (payload: unknown) => { + const call = client.on.getCalls().find((c: any) => c.args[0] === 'message') + await call.args[1](Buffer.from(JSON.stringify(payload))) + } + + it('hands the same sessions to every caller', () => { + expect(adapter.getNegentropySessions()).to.equal(adapter.getNegentropySessions()) + }) + + it('dispatches NEG-OPEN, NEG-MSG and NEG-CLOSE to a message handler', async () => { + const handleMessage = sandbox.stub().resolves() + createMessageHandler.returns({ handleMessage }) + + await triggerMessage(['NEG-OPEN', 'sync1', { kinds: [1] }, '6100000200']) + await triggerMessage(['NEG-MSG', 'sync1', '6100']) + await triggerMessage(['NEG-CLOSE', 'sync1']) + + expect(handleMessage.getCalls().map((call: any) => call.args[0][0])).to.deep.equal([ + 'NEG-OPEN', + 'NEG-MSG', + 'NEG-CLOSE', + ]) + }) + + it('replies with a NOTICE to a NEG-OPEN whose initial message is not hex', async () => { + await triggerMessage(['NEG-OPEN', 'sync1', {}, 'not hex']) + + expect(createMessageHandler).to.not.have.been.called + expect(sentMessages()).to.have.lengthOf(1) + expect(sentMessages()[0][0]).to.equal('NOTICE') + expect(sentMessages()[0][1]).to.match(/^invalid:/) + }) + + it('tells the client when a session times out', () => { + const clock = sandbox.useFakeTimers() + openSession('sync1', 1000) + + clock.tick(1000) + + expect(sentMessages()).to.deep.include(['NEG-ERR', 'sync1', 'closed: negentropy session timed out']) + expect(adapter.getNegentropySessions().getSessionIds()).to.deep.equal([]) + }) + + it('drops every session and its timer when the client disconnects', () => { + const clock = sandbox.useFakeTimers() + openSession('sync1', 1000) + const onClose = client.on.getCalls().find((call: any) => call.args[0] === 'close').args[1] + + onClose() + clock.tick(5000) + + expect(adapter.getNegentropySessions().getSessionIds()).to.deep.equal([]) + expect(sentMessages().filter((message: any[]) => message[0] === 'NEG-ERR')).to.deep.equal([]) + }) + + it('tells the client its sessions are closed when the relay drains', () => { + openSession('a') + openSession('b') + + adapter.drainAndClose('relay shutting down') + + expect(sentMessages()).to.deep.include(['NEG-ERR', 'a', 'closed: relay shutting down']) + expect(sentMessages()).to.deep.include(['NEG-ERR', 'b', 'closed: relay shutting down']) + expect(adapter.getNegentropySessions().getSessionIds()).to.deep.equal([]) + }) + }) }) diff --git a/test/unit/factories/message-handler-factory.spec.ts b/test/unit/factories/message-handler-factory.spec.ts index 6e5374d98..e9c30dd4c 100644 --- a/test/unit/factories/message-handler-factory.spec.ts +++ b/test/unit/factories/message-handler-factory.spec.ts @@ -15,6 +15,9 @@ import { EventMessageHandler } from '../../../src/handlers/event-message-handler import { IWebSocketAdapter } from '../../../src/@types/adapters' import { messageHandlerFactory } from '../../../src/factories/message-handler-factory' import { CountMessageHandler } from '../../../src/handlers/count-message-handler' +import { NegCloseMessageHandler } from '../../../src/handlers/neg-close-message-handler' +import { NegMsgMessageHandler } from '../../../src/handlers/neg-msg-message-handler' +import { NegOpenMessageHandler } from '../../../src/handlers/neg-open-message-handler' import { SubscribeMessageHandler } from '../../../src/handlers/subscribe-message-handler' import { UnsubscribeMessageHandler } from '../../../src/handlers/unsubscribe-message-handler' import * as cacheModule from '../../../src/cache/client' @@ -101,6 +104,24 @@ describe('messageHandlerFactory', () => { expect(factory([message, adapter])).to.be.an.instanceOf(AuthMessageHandler) }) + it('returns NegOpenMessageHandler when given a NEG-OPEN message', () => { + message = [MessageType.NEG_OPEN, 'sync1', {}, '6100000200'] as any + + expect(factory([message, adapter])).to.be.an.instanceOf(NegOpenMessageHandler) + }) + + it('returns NegMsgMessageHandler when given a NEG-MSG message', () => { + message = [MessageType.NEG_MSG, 'sync1', '6100'] as any + + expect(factory([message, adapter])).to.be.an.instanceOf(NegMsgMessageHandler) + }) + + it('returns NegCloseMessageHandler when given a NEG-CLOSE message', () => { + message = [MessageType.NEG_CLOSE, 'sync1'] as any + + expect(factory([message, adapter])).to.be.an.instanceOf(NegCloseMessageHandler) + }) + it('throws when given an invalid message', () => { message = [] as any diff --git a/test/unit/handlers/neg-close-message-handler.spec.ts b/test/unit/handlers/neg-close-message-handler.spec.ts new file mode 100644 index 000000000..b483fc20d --- /dev/null +++ b/test/unit/handlers/neg-close-message-handler.spec.ts @@ -0,0 +1,60 @@ +import { expect } from 'chai' +import EventEmitter from 'events' + +import { IWebSocketAdapter } from '../../../src/@types/adapters' +import { MessageType } from '../../../src/@types/messages' +import { NegCloseMessageHandler } from '../../../src/handlers/neg-close-message-handler' +import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy' +import { NegentropySessions } from '../../../src/utils/negentropy-sessions' +import { WebSocketAdapterEvent } from '../../../src/constants/adapter' + +describe('NegCloseMessageHandler', () => { + let sessions: NegentropySessions + let webSocket: IWebSocketAdapter + let handler: NegCloseMessageHandler + let emitted: unknown[][] + + const openSession = (id: string) => { + const storage = new NegentropyStorageVector() + storage.seal() + sessions.activate(id, sessions.reserve(id, 3) as symbol, { negentropy: new Negentropy(storage), filter: {} }, 60000) + } + + beforeEach(() => { + sessions = new NegentropySessions(() => undefined) + webSocket = new EventEmitter() as any + webSocket.getNegentropySessions = () => sessions + emitted = [] + webSocket.on(WebSocketAdapterEvent.Message, (...args) => emitted.push(args)) + handler = new NegCloseMessageHandler(webSocket) + }) + + afterEach(() => { + sessions.closeAll() + webSocket.removeAllListeners() + }) + + it('closes the session without replying', async () => { + openSession('sync1') + + await handler.handleMessage([MessageType.NEG_CLOSE, 'sync1'] as any) + + expect(sessions.getSessionIds()).to.deep.equal([]) + expect(emitted).to.deep.equal([]) + }) + + it('only closes the named session', async () => { + openSession('a') + openSession('b') + + await handler.handleMessage([MessageType.NEG_CLOSE, 'a'] as any) + + expect(sessions.getSessionIds()).to.deep.equal(['b']) + }) + + it('ignores an unknown id', async () => { + await handler.handleMessage([MessageType.NEG_CLOSE, 'nope'] as any) + + expect(emitted).to.deep.equal([]) + }) +}) diff --git a/test/unit/handlers/neg-msg-message-handler.spec.ts b/test/unit/handlers/neg-msg-message-handler.spec.ts new file mode 100644 index 000000000..925fe21da --- /dev/null +++ b/test/unit/handlers/neg-msg-message-handler.spec.ts @@ -0,0 +1,192 @@ +import chai from 'chai' +import { createHash } from 'crypto' +import EventEmitter from 'events' +import Sinon from 'sinon' +import sinonChai from 'sinon-chai' + +import { IWebSocketAdapter } from '../../../src/@types/adapters' +import { MessageType } from '../../../src/@types/messages' +import { Settings } from '../../../src/@types/settings' +import { WebSocketAdapterEvent } from '../../../src/constants/adapter' +import { NegMsgMessageHandler } from '../../../src/handlers/neg-msg-message-handler' +import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy' +import { NegentropySessions } from '../../../src/utils/negentropy-sessions' + +chai.use(sinonChai) +const { expect } = chai + +const storageOf = (count: number): NegentropyStorageVector => { + const storage = new NegentropyStorageVector() + for (let i = 0; i < count; i++) { + storage.insert(1700000000 + i, createHash('sha256').update(`neg-msg-${i}`).digest()) + } + storage.seal() + + return storage +} + +describe('NegMsgMessageHandler', () => { + let sandbox: Sinon.SinonSandbox + let webSocket: IWebSocketAdapter + let sessions: NegentropySessions + let settings: Settings + let emitted: Sinon.SinonStub + let handler: NegMsgMessageHandler + + const send = (id: string, hex: string) => handler.handleMessage([MessageType.NEG_MSG, id, hex] as any) + + const openSession = (id: string, itemCount = 5, filter: any = { kinds: [1] }) => { + const token = sessions.reserve(id, 3) as symbol + sessions.activate(id, token, { negentropy: new Negentropy(storageOf(itemCount)), filter }, 60000) + } + + const validMessageHex = Buffer.from(new Negentropy(storageOf(0)).initiate()).toString('hex') + + beforeEach(() => { + sandbox = Sinon.createSandbox() + sessions = new NegentropySessions(() => undefined) + settings = { nip77: { enabled: true } } as Settings + + webSocket = new EventEmitter() as any + webSocket.getNegentropySessions = () => sessions + + emitted = sandbox.stub() + webSocket.on(WebSocketAdapterEvent.Message, emitted) + + handler = new NegMsgMessageHandler(webSocket, () => settings) + }) + + afterEach(() => { + sessions.closeAll() + webSocket.removeAllListeners() + sandbox.restore() + }) + + it('answers a message on an open session with a NEG-MSG', async () => { + openSession('sync1') + + await send('sync1', validMessageHex) + + expect(emitted).to.have.been.calledOnce + const [type, id, payload] = emitted.firstCall.args[0] + expect(type).to.equal(MessageType.NEG_MSG) + expect(id).to.equal('sync1') + expect(payload).to.match(/^61(?:[0-9a-f]{2})+$/) + expect(sessions.getSessionIds()).to.deep.equal(['sync1']) + }) + + it('replies NEG-ERR closed for an unknown subscription id', async () => { + await send('nope', validMessageHex) + + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'nope', + 'closed: unknown negentropy subscription', + ]) + }) + + it('does not answer for a REQ subscription id: the namespaces are separate', async () => { + // A REQ subscription named "sync1" lives in the adapter's subscriptions, not in the negentropy sessions. + await send('sync1', validMessageHex) + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_ERR) + }) + + it('closes the session and replies NEG-ERR when the message is malformed', async () => { + openSession('sync1') + + await send('sync1', '6100') + + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'sync1', + 'invalid: malformed negentropy message', + ]) + expect(sessions.getSessionIds()).to.deep.equal([]) + }) + + it('keeps other sessions when one is closed by an error', async () => { + openSession('a') + openSession('b') + + await send('a', '6100') + + expect(sessions.getSessionIds()).to.deep.equal(['b']) + }) + + it('ends the session with NEG-ERR blocked when NIP-77 is turned off while it is open', async () => { + openSession('sync1') + settings = { nip77: { enabled: false } } as Settings + + await send('sync1', validMessageHex) + + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'sync1', + 'blocked: NIP-77 is disabled by relay configuration', + ]) + expect(sessions.getSessionIds()).to.deep.equal([]) + }) + + describe('authorization is checked on every message', () => { + const pubkey = 'f'.repeat(64) + let authenticated: Set + + beforeEach(() => { + authenticated = new Set([pubkey]) + webSocket.getAuthenticatedPubkeys = () => authenticated + settings = { nip77: { enabled: true }, nip42: { restrictedReads: { enabled: true } } } as Settings + }) + + it('keeps answering while the client is still authorized', async () => { + openSession('sync1', 5, { authors: [pubkey] }) + + await send('sync1', validMessageHex) + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) + }) + + it('closes the session with auth-required once authentication lapses', async () => { + openSession('sync1', 5, { authors: [pubkey] }) + authenticated.clear() // e.g. nip42.sessionExpirySeconds elapsed + + await send('sync1', validMessageHex) + + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'sync1', + 'auth-required: authentication is required to sync these event kinds', + ]) + expect(sessions.getSessionIds()).to.deep.equal([]) + }) + + it('closes the session when restricted reads were switched on after it opened', async () => { + settings = { nip77: { enabled: true } } as Settings + openSession('sync1', 5, {}) + settings = { nip77: { enabled: true }, nip42: { restrictedReads: { enabled: true } } } as Settings + authenticated.clear() + + await send('sync1', validMessageHex) + + expect(emitted.firstCall.args[0][2]).to.match(/^auth-required:/) + expect(sessions.getSessionIds()).to.deep.equal([]) + }) + + it('does not ask for auth when the session only covers ordinary kinds', async () => { + openSession('sync1', 5, { kinds: [1] }) + authenticated.clear() + + await send('sync1', validMessageHex) + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) + }) + }) + + it('answers an unsupported protocol version with just the supported version', async () => { + openSession('sync1') + + await send('sync1', '62') + + expect(emitted).to.have.been.calledOnceWithExactly([MessageType.NEG_MSG, 'sync1', '61']) + }) +}) diff --git a/test/unit/handlers/neg-open-message-handler.spec.ts b/test/unit/handlers/neg-open-message-handler.spec.ts new file mode 100644 index 000000000..6212ce5bd --- /dev/null +++ b/test/unit/handlers/neg-open-message-handler.spec.ts @@ -0,0 +1,576 @@ +import chai from 'chai' +import { createHash } from 'crypto' +import EventEmitter from 'events' +import Sinon from 'sinon' +import sinonChai from 'sinon-chai' + +import { IWebSocketAdapter } from '../../../src/@types/adapters' +import { MessageType } from '../../../src/@types/messages' +import { IEventRepository } from '../../../src/@types/repositories' +import { Settings } from '../../../src/@types/settings' +import { WebSocketAdapterEvent } from '../../../src/constants/adapter' +import { NegMsgMessageHandler } from '../../../src/handlers/neg-msg-message-handler' +import { NegOpenMessageHandler } from '../../../src/handlers/neg-open-message-handler' +import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy' +import { NegentropySessions } from '../../../src/utils/negentropy-sessions' + +chai.use(sinonChai) +const { expect } = chai + +const BASE_TIMESTAMP = 1700000000 + +const idFor = (n: number): Buffer => createHash('sha256').update(`neg-open-${n}`).digest() + +const rowsFor = (from: number, to: number) => { + const rows: { event_created_at: number; event_id: Buffer }[] = [] + for (let i = from; i < to; i++) { + rows.push({ event_created_at: BASE_TIMESTAMP + i, event_id: idFor(i) }) + } + + return rows +} + +const clientFor = (from: number, to: number): Negentropy => { + const storage = new NegentropyStorageVector() + for (const row of rowsFor(from, to)) { + storage.insert(row.event_created_at, row.event_id) + } + storage.seal() + + return new Negentropy(storage) +} + +const emptyClientMessageHex = Buffer.from(clientFor(0, 0).initiate()).toString('hex') + +describe('NegOpenMessageHandler', () => { + let sandbox: Sinon.SinonSandbox + let webSocket: IWebSocketAdapter + let sessions: NegentropySessions + let eventRepository: IEventRepository + let findReconciliationItems: Sinon.SinonStub + let settings: Settings + let emitted: Sinon.SinonStub + let handler: NegOpenMessageHandler + + const open = (id = 'sync1', filter: any = { kinds: [1] }, hex = emptyClientMessageHex) => + handler.handleMessage([MessageType.NEG_OPEN, id, filter, hex] as any) + + beforeEach(() => { + sandbox = Sinon.createSandbox() + sessions = new NegentropySessions(() => undefined) + findReconciliationItems = sandbox.stub().resolves(rowsFor(0, 5)) + eventRepository = { findReconciliationItems } as any + settings = { nip77: { enabled: true, maxEventsPerSession: 100, maxSessionsPerConnection: 2 } } as Settings + + webSocket = new EventEmitter() as any + webSocket.getNegentropySessions = () => sessions + webSocket.getAuthenticatedPubkeys = () => new Set() + + emitted = sandbox.stub() + webSocket.on(WebSocketAdapterEvent.Message, emitted) + + handler = new NegOpenMessageHandler(webSocket, eventRepository, () => settings) + }) + + afterEach(() => { + sessions.closeAll() + webSocket.removeAllListeners() + sandbox.restore() + }) + + it('answers the initial message with a NEG-MSG and keeps the session open', async () => { + await open() + + expect(emitted).to.have.been.calledOnce + const [type, id, payload] = emitted.firstCall.args[0] + expect(type).to.equal(MessageType.NEG_MSG) + expect(id).to.equal('sync1') + expect(payload).to.match(/^61(?:[0-9a-f]{2})+$/) + expect(sessions.getSessionIds()).to.deep.equal(['sync1']) + }) + + it('reads the matching events with one row of headroom over the limit', async () => { + await open('sync1', { kinds: [1], limit: 3 }) + + expect(findReconciliationItems).to.have.been.calledOnceWithExactly({ kinds: [1], limit: 3 }, 101) + }) + + describe('when disabled', () => { + it('replies NEG-ERR blocked and does not read events', async () => { + settings = { nip77: { enabled: false } } as Settings + + await open() + + expect(findReconciliationItems).to.not.have.been.called + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'sync1', + 'blocked: NIP-77 is disabled by relay configuration', + ]) + expect(sessions.getSessionIds()).to.deep.equal([]) + }) + + it('is disabled when the nip77 section is missing', async () => { + settings = {} as Settings + + await open() + + expect(emitted.firstCall.args[0][2]).to.match(/^blocked: NIP-77 is disabled/) + }) + }) + + describe('limits', () => { + it('rejects a query that matches more events than allowed, with the limit as 4th element', async () => { + findReconciliationItems.resolves(rowsFor(0, 101)) + + await open() + + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'sync1', + 'blocked: this query is too big', + 100, + ]) + expect(sessions.getSessionIds()).to.deep.equal([]) + }) + + it('accepts a query that matches exactly the limit', async () => { + findReconciliationItems.resolves(rowsFor(0, 100)) + + await open() + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) + }) + + it('frees the slot of a rejected oversized query', async () => { + findReconciliationItems.resolves(rowsFor(0, 101)) + await open('big') + + findReconciliationItems.resolves(rowsFor(0, 5)) + await open('a') + await open('b') + + expect(sessions.getSessionIds()).to.have.members(['a', 'b']) + }) + + it('rejects a new session beyond the per-connection limit', async () => { + await open('a') + await open('b') + emitted.resetHistory() + + await open('c') + + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'c', + 'blocked: too many negentropy sessions or reads in progress on this connection (max 2)', + ]) + expect(sessions.getSessionIds()).to.have.members(['a', 'b']) + }) + + it('lets a client reopen an id it already has when at the limit, replacing the session', async () => { + await open('a') + await open('b') + emitted.resetHistory() + + await open('a') + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) + expect(sessions.getSessionIds()).to.have.members(['a', 'b']) + }) + + it('uses the defaults when no limits are configured', async () => { + settings = { nip77: { enabled: true } } as Settings + + await open() + + expect(findReconciliationItems.firstCall.args[1]).to.equal(100001) + }) + + it('rejects subscription ids over the configured length', async () => { + settings = { + nip77: { enabled: true }, + limits: { client: { subscription: { maxSubscriptionIdLength: 4 } } }, + } as Settings + + await open('too-long') + + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'too-long', + 'invalid: subscription id too long (max 4)', + ]) + expect(findReconciliationItems).to.not.have.been.called + }) + + it('rejects filters with too many values', async () => { + settings = { + nip77: { enabled: true }, + limits: { client: { subscription: { maxFilterValues: 2 } } }, + } as Settings + + await open('sync1', { authors: ['a', 'b', 'c'] }) + + expect(emitted.firstCall.args[0][2]).to.equal('blocked: too many filter values (max 2)') + expect(findReconciliationItems).to.not.have.been.called + }) + }) + + describe('NIP-42', () => { + const pubkey = 'f'.repeat(64) + const otherPubkey = 'e'.repeat(64) + + beforeEach(() => { + settings = { nip77: { enabled: true }, nip42: { restrictedReads: { enabled: true } } } as Settings + }) + + const expectAuthRequired = () => { + expect(findReconciliationItems).to.not.have.been.called + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'sync1', + 'auth-required: authentication is required to sync these event kinds', + ]) + } + + it('requires auth to sync restricted kinds', async () => { + await open('sync1', { kinds: [1059], '#p': [pubkey] }) + + expectAuthRequired() + }) + + // REQ hides restricted events one by one; reconciliation would list their ids, so a filter that + // could match one has to be scoped to the client. + it('requires auth for a filter that names no kinds, which could match restricted events', async () => { + await open('sync1', {}) + + expectAuthRequired() + }) + + it('requires auth for an authors-only filter of someone else', async () => { + webSocket.getAuthenticatedPubkeys = () => new Set([pubkey]) + + await open('sync1', { authors: [otherPubkey] }) + + expectAuthRequired() + }) + + it('requires auth for a filter mixing a restricted and an ordinary kind', async () => { + await open('sync1', { kinds: [1, 1059] }) + + expectAuthRequired() + }) + + it('requires every author to be authenticated', async () => { + webSocket.getAuthenticatedPubkeys = () => new Set([pubkey]) + + await open('sync1', { authors: [pubkey, otherPubkey] }) + + expectAuthRequired() + }) + + it('allows restricted kinds scoped to the authenticated pubkey', async () => { + webSocket.getAuthenticatedPubkeys = () => new Set([pubkey]) + + await open('sync1', { kinds: [1059], '#p': [pubkey] }) + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) + }) + + it('allows a filter without kinds when it is scoped to the authenticated pubkey', async () => { + webSocket.getAuthenticatedPubkeys = () => new Set([pubkey]) + + await open('sync1', { authors: [pubkey] }) + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) + }) + + it('does not restrict filters whose kinds are all ordinary', async () => { + await open('sync1', { kinds: [1, 7] }) + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) + }) + + it('does not restrict anything when restricted reads are off', async () => { + settings = { nip77: { enabled: true } } as Settings + + await open('sync1', {}) + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) + }) + }) + + describe('malformed input', () => { + it('rejects an initial message that is not a negentropy message, before reading events', async () => { + await open('sync1', {}, 'ff00') + + expect(findReconciliationItems).to.not.have.been.called + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'sync1', + 'invalid: malformed negentropy message', + ]) + }) + + it('rejects a truncated initial message and keeps no session', async () => { + await open('sync1', {}, '6100') + + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'sync1', + 'invalid: malformed negentropy message', + ]) + expect(sessions.getSessionIds()).to.deep.equal([]) + }) + + it('answers an unsupported protocol version with just the supported version', async () => { + await open('sync1', {}, '62') + + expect(emitted).to.have.been.calledOnceWithExactly([MessageType.NEG_MSG, 'sync1', '61']) + }) + }) + + describe('failures', () => { + it('replies with a generic NEG-ERR when reading events fails, without leaking the cause', async () => { + findReconciliationItems.rejects(new Error('connection to 10.0.0.5 refused')) + + await open() + + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'sync1', + 'error: unable to process negentropy request', + ]) + expect(sessions.getSessionIds()).to.deep.equal([]) + }) + + it('frees the slot after a failure', async () => { + findReconciliationItems.rejects(new Error('boom')) + await open('a') + await open('b') + findReconciliationItems.resolves(rowsFor(0, 5)) + + await open('c') + await open('d') + + expect(sessions.getSessionIds()).to.have.members(['c', 'd']) + }) + + it('stays silent when the client closed the session while events were being read', async () => { + let finishRead: (rows: any[]) => void = () => undefined + findReconciliationItems.returns(new Promise((resolve) => (finishRead = resolve))) + + const pending = open('sync1') + sessions.close('sync1') + finishRead(rowsFor(0, 5)) + await pending + + expect(emitted).to.not.have.been.called + expect(sessions.getSessionIds()).to.deep.equal([]) + }) + + it('lets a newer NEG-OPEN win when two race on the same id', async () => { + const finishers: ((rows: any[]) => void)[] = [] + findReconciliationItems.callsFake(() => new Promise((resolve) => finishers.push(resolve))) + + const first = open('sync1', { kinds: [1] }) + const second = open('sync1', { kinds: [7] }) + finishers[1](rowsFor(0, 5)) + await second + finishers[0](rowsFor(0, 5)) + await first + + expect(emitted).to.have.been.calledOnce + expect(sessions.getSessionIds()).to.deep.equal(['sync1']) + }) + }) + + describe('replacing an open session', () => { + const pendingReads = () => { + const finishers: { resolve: (rows: any[]) => void; reject: (error: Error) => void }[] = [] + findReconciliationItems.callsFake(() => new Promise((resolve, reject) => finishers.push({ resolve, reject }))) + + return finishers + } + + // NIP-77: after a NEG-ERR the subscription is considered closed. + for (const [name, arrange, hex, filter] of [ + ['a bad version byte', () => undefined, 'ff00', {}], + [ + 'too many filter values', + () => + (settings = { + nip77: { enabled: true }, + limits: { client: { subscription: { maxFilterValues: 1 } } }, + } as Settings), + emptyClientMessageHex, + { authors: ['a', 'b'] }, + ], + [ + 'NIP-77 being turned off', + () => (settings = { nip77: { enabled: false } } as Settings), + emptyClientMessageHex, + {}, + ], + [ + 'an unauthorized filter', + () => (settings = { nip77: { enabled: true }, nip42: { restrictedReads: { enabled: true } } } as Settings), + emptyClientMessageHex, + {}, + ], + ] as [string, () => void, string, any][]) { + it(`closes the session the client already had when the replacement is rejected for ${name}`, async () => { + await open('sync1') + expect(sessions.getSessionIds()).to.deep.equal(['sync1']) + arrange() + emitted.resetHistory() + + await open('sync1', filter, hex) + + expect(emitted).to.have.been.calledOnce + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_ERR) + expect(sessions.getSessionIds()).to.deep.equal([]) + }) + } + + it('frees the slot of a session whose replacement was rejected', async () => { + await open('a') + await open('b') + + await open('a', {}, 'ff00') + + await open('c') + expect(sessions.getSessionIds()).to.have.members(['b', 'c']) + }) + + it('stays silent when a replaced request turns out to be too big', async () => { + const reads = pendingReads() + const first = open('sync1', { kinds: [1] }) + const second = open('sync1', { kinds: [7] }) + reads[1].resolve(rowsFor(0, 5)) + await second + emitted.resetHistory() + + reads[0].resolve(rowsFor(0, 101)) + await first + + expect(emitted).to.not.have.been.called + expect(sessions.getSessionIds()).to.deep.equal(['sync1']) + }) + + it('stays silent when a replaced request fails to read', async () => { + const reads = pendingReads() + const first = open('sync1', { kinds: [1] }) + const second = open('sync1', { kinds: [7] }) + reads[1].resolve(rowsFor(0, 5)) + await second + emitted.resetHistory() + + reads[0].reject(new Error('boom')) + await first + + expect(emitted).to.not.have.been.called + expect(sessions.getSessionIds()).to.deep.equal(['sync1']) + }) + + it('stays silent when a replaced request had a malformed message', async () => { + const reads = pendingReads() + const first = open('sync1', { kinds: [1] }, '6100') + const second = open('sync1', { kinds: [7] }) + reads[1].resolve(rowsFor(0, 5)) + await second + emitted.resetHistory() + + reads[0].resolve(rowsFor(0, 5)) + await first + + expect(emitted).to.not.have.been.called + expect(sessions.getSessionIds()).to.deep.equal(['sync1']) + }) + + it('does no work for a replaced request once its read returns', async () => { + const reads = pendingReads() + const first = open('sync1', { kinds: [1] }) + const second = open('sync1', { kinds: [7] }) + reads[1].resolve(rowsFor(0, 5)) + await second + emitted.resetHistory() + const sealSpy = sandbox.spy(NegentropyStorageVector.prototype, 'seal') + + reads[0].resolve(rowsFor(0, 50)) + await first + + expect(sealSpy).to.not.have.been.called + }) + + it('does not let repeated NEG-OPENs on one id pile up event reads', async () => { + const reads = pendingReads() + + const opens = [open('sync1'), open('sync1')] + await open('sync1') // a third read while the first two have not returned + + expect(findReconciliationItems).to.have.been.calledTwice + expect(emitted).to.have.been.calledOnceWithExactly([ + MessageType.NEG_ERR, + 'sync1', + 'blocked: too many negentropy sessions or reads in progress on this connection (max 2)', + ]) + + reads.forEach((read) => read.resolve(rowsFor(0, 5))) + await Promise.all(opens) + }) + + it('accepts new reads again once the earlier ones have returned', async () => { + const reads = pendingReads() + const opens = [open('sync1'), open('sync1')] + reads.forEach((read) => read.resolve(rowsFor(0, 5))) + await Promise.all(opens) + findReconciliationItems.resolves(rowsFor(0, 5)) + emitted.resetHistory() + + await open('sync1') + + expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_MSG) + }) + }) + + describe('a full reconciliation through the handlers', () => { + const decode = (message: any[]) => new Uint8Array(Buffer.from(message[2], 'hex')) + + // The relay holds events 10-59; the client holds 0-39. The client must learn that it lacks 40-59 + // and that the relay lacks 0-9. + it('lets a client learn exactly which events differ', async () => { + findReconciliationItems.resolves(rowsFor(10, 60)) + const msgHandler = new NegMsgMessageHandler(webSocket, () => settings) + const client = clientFor(0, 40) + const haveIds = new Set() + const needIds = new Set() + + let outgoing: Uint8Array | null = client.initiate() + let rounds = 0 + while (outgoing !== null) { + expect(++rounds).to.be.lessThan(20) + emitted.resetHistory() + const hex = Buffer.from(outgoing).toString('hex') + if (rounds === 1) { + await open('sync1', { kinds: [1] }, hex) + } else { + await msgHandler.handleMessage([MessageType.NEG_MSG, 'sync1', hex] as any) + } + + expect(emitted).to.have.been.calledOnce + const reply = emitted.firstCall.args[0] + expect(reply[0]).to.equal(MessageType.NEG_MSG) + + const result = client.reconcile(decode(reply)) + result.haveIds.forEach((id) => haveIds.add(id)) + result.needIds.forEach((id) => needIds.add(id)) + outgoing = result.output + } + + const hexOf = (from: number, to: number) => rowsFor(from, to).map((row) => row.event_id.toString('hex')) + expect([...haveIds].sort()).to.deep.equal(hexOf(0, 10).sort()) + expect([...needIds].sort()).to.deep.equal(hexOf(40, 60).sort()) + }) + }) +}) diff --git a/test/unit/repositories/event-repository.spec.ts b/test/unit/repositories/event-repository.spec.ts index 38c91e20a..0e89d0934 100644 --- a/test/unit/repositories/event-repository.spec.ts +++ b/test/unit/repositories/event-repository.spec.ts @@ -728,6 +728,12 @@ describe('EventRepository', () => { expect(query).to.include(ORDER_CLAUSE) }) + it('applies a row cap when the caller asks for one, after the ordering', () => { + const query = repository.findReconciliationItems({ kinds: [1] }, 101).toString() + + expect(query).to.match(/order by "events"\."event_created_at" asc, "events"\."event_id" asc limit 101$/) + }) + it('applies tag filters', () => { const query = repository.findReconciliationItems({ '#e': ['b'.repeat(64)] } as any).toString() diff --git a/test/unit/schemas/message-schema.spec.ts b/test/unit/schemas/message-schema.spec.ts index 562e5b906..f60c45271 100644 --- a/test/unit/schemas/message-schema.spec.ts +++ b/test/unit/schemas/message-schema.spec.ts @@ -206,3 +206,84 @@ describe('NIP-01', () => { }) }) }) + +describe('NIP-77', () => { + const valid = { + open: ['NEG-OPEN', 'sync1', { kinds: [1], since: 1000 }, '61000002'], + msg: ['NEG-MSG', 'sync1', '6100000200'], + close: ['NEG-CLOSE', 'sync1'], + } + const check = (message: unknown) => validateSchema(messageSchema)(message as any) + + describe('NEG-OPEN', () => { + it('returns same message if valid', () => { + const result = check(valid.open) + + expect(result.error).to.be.undefined + expect(result).to.have.deep.property('value', valid.open) + }) + + it('accepts uppercase hex', () => { + expect(check(['NEG-OPEN', 'sync1', {}, 'ABCDEF01']).error).to.be.undefined + }) + + for (const [name, mutate] of [ + ['subscription id is missing', (m: any[]) => m.splice(1, 1)], + ['subscription id is empty', (m: any[]) => (m[1] = '')], + ['subscription id is longer than 256 characters', (m: any[]) => (m[1] = 'a'.repeat(257))], + ['filter is missing', (m: any[]) => m.splice(2, 1)], + ['filter is an array of filters', (m: any[]) => (m[2] = [{}])], + ['filter is not an object', (m: any[]) => (m[2] = null)], + ['initial message is missing', (m: any[]) => m.splice(3, 1)], + ['initial message is empty', (m: any[]) => (m[3] = '')], + ['initial message is not hex', (m: any[]) => (m[3] = 'zz')], + ['initial message has an odd number of hex digits', (m: any[]) => (m[3] = '610')], + ['there are extra elements', (m: any[]) => m.push('extra')], + ] as [string, (m: any[]) => void][]) { + it(`returns error if ${name}`, () => { + const message = JSON.parse(JSON.stringify(valid.open)) + mutate(message) + + expect(check(message)).to.have.property('error').that.is.not.undefined + }) + } + }) + + describe('NEG-MSG', () => { + it('returns same message if valid', () => { + const result = check(valid.msg) + + expect(result.error).to.be.undefined + expect(result).to.have.deep.property('value', valid.msg) + }) + + for (const [name, message] of [ + ['subscription id is empty', ['NEG-MSG', '', '6100']], + ['payload is not hex', ['NEG-MSG', 'sync1', 'not hex']], + ['payload is empty', ['NEG-MSG', 'sync1', '']], + ['payload is missing', ['NEG-MSG', 'sync1']], + ] as [string, unknown[]][]) { + it(`returns error if ${name}`, () => { + expect(check(message)).to.have.property('error').that.is.not.undefined + }) + } + }) + + describe('NEG-CLOSE', () => { + it('returns same message if valid', () => { + const result = check(valid.close) + + expect(result.error).to.be.undefined + expect(result).to.have.deep.property('value', valid.close) + }) + + it('returns error if subscription id is missing or empty', () => { + expect(check(['NEG-CLOSE'])).to.have.property('error').that.is.not.undefined + expect(check(['NEG-CLOSE', ''])).to.have.property('error').that.is.not.undefined + }) + }) + + it('does not accept NEG-ERR from a client', () => { + expect(check(['NEG-ERR', 'sync1', 'blocked: no'])).to.have.property('error').that.is.not.undefined + }) +}) diff --git a/test/unit/utils/messages.spec.ts b/test/unit/utils/messages.spec.ts index cff5ad242..0252166ef 100644 --- a/test/unit/utils/messages.spec.ts +++ b/test/unit/utils/messages.spec.ts @@ -4,6 +4,8 @@ import { createClosedMessage, createCommandResult, createEndOfStoredEventsNoticeMessage, + createNegErrMessage, + createNegMsgMessage, createNoticeMessage, createOutgoingEventMessage, createRelayedEventMessage, @@ -147,3 +149,32 @@ describe('createRelayedEventMessage', () => { ]) }) }) + +describe('NIP-77 messages', () => { + it('createNegMsgMessage returns a NEG-MSG with the hex payload', () => { + expect(createNegMsgMessage('sync1', '6100')).to.deep.equal([MessageType.NEG_MSG, 'sync1', '6100']) + }) + + it('createNegErrMessage returns a NEG-ERR with the reason', () => { + expect(createNegErrMessage('sync1', 'closed: bye')).to.deep.equal([MessageType.NEG_ERR, 'sync1', 'closed: bye']) + }) + + it('createNegErrMessage appends the record limit when given', () => { + expect(createNegErrMessage('sync1', 'blocked: this query is too big', 5000)).to.deep.equal([ + MessageType.NEG_ERR, + 'sync1', + 'blocked: this query is too big', + 5000, + ]) + }) + + it('createNegErrMessage keeps a record limit of 0', () => { + expect(createNegErrMessage('sync1', 'blocked: no', 0)).to.have.lengthOf(4) + }) + + it('serialises to the JSON array NIP-77 describes', () => { + expect(JSON.stringify(createNegErrMessage('sync1', 'blocked: no', 3))).to.equal( + '["NEG-ERR","sync1","blocked: no",3]', + ) + }) +}) diff --git a/test/unit/utils/negentropy-sessions.spec.ts b/test/unit/utils/negentropy-sessions.spec.ts new file mode 100644 index 000000000..ba1ce6183 --- /dev/null +++ b/test/unit/utils/negentropy-sessions.spec.ts @@ -0,0 +1,232 @@ +import chai from 'chai' +import Sinon from 'sinon' +import sinonChai from 'sinon-chai' + +import { NegentropySession } from '../../../src/@types/negentropy' +import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy' +import { NegentropySessions } from '../../../src/utils/negentropy-sessions' + +chai.use(sinonChai) +const { expect } = chai + +const newSession = (filter: NegentropySession['filter'] = {}): NegentropySession => { + const storage = new NegentropyStorageVector() + storage.seal() + + return { negentropy: new Negentropy(storage), filter } +} + +describe('NegentropySessions', () => { + let clock: Sinon.SinonFakeTimers + let onIdleTimeout: Sinon.SinonStub + let sessions: NegentropySessions + + beforeEach(() => { + clock = Sinon.useFakeTimers() + onIdleTimeout = Sinon.stub() + sessions = new NegentropySessions(onIdleTimeout) + }) + + afterEach(() => { + sessions.closeAll() + clock.restore() + }) + + describe('reserve and activate', () => { + it('stores a session once it is activated', () => { + const session = newSession() + + const token = sessions.reserve('a', 3) + expect(token).to.not.be.undefined + expect(sessions.activate('a', token as symbol, session, 1000)).to.equal(true) + + expect(sessions.touch('a')).to.equal(session) + expect(sessions.getSessionIds()).to.deep.equal(['a']) + }) + + it('does not list a reservation that is not active yet', () => { + sessions.reserve('a', 3) + + expect(sessions.getSessionIds()).to.deep.equal([]) + expect(sessions.touch('a')).to.be.undefined + }) + + it('refuses a new id once the limit is reached, counting reservations', () => { + sessions.reserve('a', 2) + sessions.reserve('b', 2) + + expect(sessions.reserve('c', 2)).to.be.undefined + }) + + it('lets an id that is already open be reopened at the limit, replacing the old session', () => { + const first = newSession() + const second = newSession() + const token1 = sessions.reserve('a', 1) as symbol + sessions.activate('a', token1, first, 1000) + + const token2 = sessions.reserve('a', 1) + expect(token2).to.not.be.undefined + sessions.activate('a', token2 as symbol, second, 1000) + + expect(sessions.touch('a')).to.equal(second) + }) + + it('refuses to activate a reservation that was replaced meanwhile', () => { + const stale = sessions.reserve('a', 3) as symbol + sessions.reserve('a', 3) + + expect(sessions.activate('a', stale, newSession(), 1000)).to.equal(false) + expect(sessions.touch('a')).to.be.undefined + }) + + it('refuses to activate a reservation that was closed meanwhile', () => { + const token = sessions.reserve('a', 3) as symbol + sessions.close('a') + + expect(sessions.activate('a', token, newSession(), 1000)).to.equal(false) + }) + + it('frees the slot when a reservation is released and its read has ended', () => { + const token = sessions.reserve('a', 1) as symbol + + sessions.release('a', token) + sessions.endRead(token) + + expect(sessions.reserve('b', 1)).to.not.be.undefined + }) + + it('does not release a newer reservation with an old token', () => { + const stale = sessions.reserve('a', 3) as symbol + const current = sessions.reserve('a', 3) as symbol + + sessions.release('a', stale) + + expect(sessions.activate('a', current, newSession(), 1000)).to.equal(true) + }) + }) + + describe('reads in flight', () => { + it('keeps counting the read of a replaced request until it ends', () => { + sessions.reserve('a', 2) + sessions.reserve('a', 2) // replaces the first, whose read is still running + + expect(sessions.reserve('a', 2)).to.be.undefined + }) + + it('stops counting a read once it ends', () => { + const first = sessions.reserve('a', 2) as symbol + sessions.reserve('a', 2) + + sessions.endRead(first) + + expect(sessions.reserve('a', 2)).to.not.be.undefined + }) + + it('does not count a read twice or after it was never started', () => { + const token = sessions.reserve('a', 2) as symbol + + sessions.endRead(token) + sessions.endRead(token) + + expect(sessions.reserve('b', 2)).to.not.be.undefined + expect(sessions.reserve('c', 2)).to.be.undefined // the two sessions a and b fill the limit + }) + + it('counts a read as over once its session is activated', () => { + const token = sessions.reserve('a', 1) as symbol + sessions.activate('a', token, newSession(), 1000) + + expect(sessions.reserve('a', 1)).to.not.be.undefined + }) + + it('reports whether a token still owns its id', () => { + const first = sessions.reserve('a', 3) as symbol + expect(sessions.isCurrent('a', first)).to.equal(true) + + const second = sessions.reserve('a', 3) as symbol + + expect(sessions.isCurrent('a', first)).to.equal(false) + expect(sessions.isCurrent('a', second)).to.equal(true) + sessions.close('a') + expect(sessions.isCurrent('a', second)).to.equal(false) + }) + + it('keeps the filter a session was opened with', () => { + const filter = { kinds: [1], authors: ['a'.repeat(64)] } + sessions.activate('a', sessions.reserve('a', 3) as symbol, newSession(filter), 1000) + + expect(sessions.touch('a')?.filter).to.deep.equal(filter) + }) + }) + + describe('close', () => { + it('removes a session and reports whether there was one', () => { + sessions.activate('a', sessions.reserve('a', 3) as symbol, newSession(), 1000) + + expect(sessions.close('a')).to.equal(true) + expect(sessions.close('a')).to.equal(false) + expect(sessions.touch('a')).to.be.undefined + }) + + it('closeAll removes every session and reservation', () => { + sessions.activate('a', sessions.reserve('a', 3) as symbol, newSession(), 1000) + const pending = sessions.reserve('b', 3) as symbol + + sessions.closeAll() + sessions.endRead(pending) + + expect(sessions.getSessionIds()).to.deep.equal([]) + expect(sessions.reserve('c', 1)).to.not.be.undefined + }) + }) + + describe('idle timeout', () => { + it('closes an idle session and tells the caller', () => { + sessions.activate('a', sessions.reserve('a', 3) as symbol, newSession(), 1000) + + clock.tick(1000) + + expect(onIdleTimeout).to.have.been.calledOnceWithExactly('a') + expect(sessions.touch('a')).to.be.undefined + }) + + it('restarts the timer whenever the session is touched', () => { + sessions.activate('a', sessions.reserve('a', 3) as symbol, newSession(), 1000) + + clock.tick(900) + sessions.touch('a') + clock.tick(900) + + expect(onIdleTimeout).to.not.have.been.called + clock.tick(100) + expect(onIdleTimeout).to.have.been.calledOnceWithExactly('a') + }) + + it('does not fire for a session closed before the timeout', () => { + sessions.activate('a', sessions.reserve('a', 3) as symbol, newSession(), 1000) + + sessions.close('a') + clock.tick(5000) + + expect(onIdleTimeout).to.not.have.been.called + }) + + it('does not fire for a replaced session', () => { + sessions.activate('a', sessions.reserve('a', 3) as symbol, newSession(), 1000) + sessions.activate('a', sessions.reserve('a', 3) as symbol, newSession(), 10000) + + clock.tick(1000) + + expect(onIdleTimeout).to.not.have.been.called + }) + + it('never times out when the timeout is not positive', () => { + sessions.activate('a', sessions.reserve('a', 3) as symbol, newSession(), 0) + + clock.tick(10 * 60 * 1000) + + expect(onIdleTimeout).to.not.have.been.called + expect(sessions.getSessionIds()).to.deep.equal(['a']) + }) + }) +}) diff --git a/test/unit/utils/nip42.spec.ts b/test/unit/utils/nip42.spec.ts index efba87f8e..b2310b995 100644 --- a/test/unit/utils/nip42.spec.ts +++ b/test/unit/utils/nip42.spec.ts @@ -7,6 +7,7 @@ import { isAuthRequired, isClientAuthorizedToReadMention, isCountAuthorized, + isReconciliationAuthorized, isSubscriptionAuthRequired, } from '../../../src/utils/nip42' import { Event } from '../../../src/@types/event' @@ -194,4 +195,62 @@ describe('nip42', () => { expect(isCountAuthorized(enabledSettings(), filters, () => new Set([author]))).to.be.false }) }) + + describe('isReconciliationAuthorized', () => { + const noOne = () => new Set() + const onlyAuthor = () => new Set([author]) + + it('allows everything when restricted reads are off', () => { + expect(isReconciliationAuthorized(undefined, {}, noOne)).to.be.true + expect(isReconciliationAuthorized({} as Settings, { kinds: [4] }, noOne)).to.be.true + }) + + it('allows a filter whose kinds are all ordinary', () => { + expect(isReconciliationAuthorized(enabledSettings(), { kinds: [1, 7] }, noOne)).to.be.true + }) + + it('allows a filter that matches no kinds', () => { + expect(isReconciliationAuthorized(enabledSettings(), { kinds: [] }, noOne)).to.be.true + }) + + it('rejects a filter with no kinds, which could match a restricted event', () => { + expect(isReconciliationAuthorized(enabledSettings(), {}, noOne)).to.be.false + expect(isReconciliationAuthorized(enabledSettings(), { authors: [author] }, noOne)).to.be.false + expect(isReconciliationAuthorized(enabledSettings(), { since: 1 }, onlyAuthor)).to.be.false + }) + + it('rejects a filter naming a restricted kind alongside ordinary ones', () => { + expect(isReconciliationAuthorized(enabledSettings(), { kinds: [1, 4] }, noOne)).to.be.false + }) + + it('honours restricted kind ranges', () => { + const settings = enabledSettings([[1059, 1060]]) + + expect(isReconciliationAuthorized(settings, { kinds: [1060] }, noOne)).to.be.false + expect(isReconciliationAuthorized(settings, { kinds: [1061] }, noOne)).to.be.true + }) + + it('allows a possibly-restricted filter scoped to the authenticated pubkey', () => { + expect(isReconciliationAuthorized(enabledSettings(), { authors: [author] }, onlyAuthor)).to.be.true + expect(isReconciliationAuthorized(enabledSettings(), { kinds: [4], '#p': [author] }, onlyAuthor)).to.be.true + }) + + it('rejects a scope that includes a pubkey the client has not authenticated as', () => { + expect(isReconciliationAuthorized(enabledSettings(), { authors: [author, stranger] }, onlyAuthor)).to.be.false + expect(isReconciliationAuthorized(enabledSettings(), { authors: [stranger] }, onlyAuthor)).to.be.false + }) + + it('rejects an empty scope', () => { + expect(isReconciliationAuthorized(enabledSettings(), { authors: [] }, onlyAuthor)).to.be.false + }) + + it('reads the authenticated pubkeys when asked, so lapsed auth is noticed', () => { + const authenticated = new Set([author]) + const filter: SubscriptionFilter = { authors: [author] } + + expect(isReconciliationAuthorized(enabledSettings(), filter, () => authenticated)).to.be.true + authenticated.clear() + expect(isReconciliationAuthorized(enabledSettings(), filter, () => authenticated)).to.be.false + }) + }) }) diff --git a/test/unit/utils/nip77.spec.ts b/test/unit/utils/nip77.spec.ts new file mode 100644 index 000000000..6def73a28 --- /dev/null +++ b/test/unit/utils/nip77.spec.ts @@ -0,0 +1,59 @@ +import { expect } from 'chai' + +import { + DEFAULT_MAX_EVENTS_PER_SESSION, + DEFAULT_MAX_SESSIONS_PER_CONNECTION, + DEFAULT_SESSION_IDLE_TIMEOUT_MS, + getNip77Limits, + isNip77Active, + NEGENTROPY_FRAME_SIZE_LIMIT, +} from '../../../src/utils/nip77' +import { Settings } from '../../../src/@types/settings' + +describe('NIP-77 helpers', () => { + describe('isNip77Active', () => { + it('is off by default', () => { + expect(isNip77Active({} as Settings)).to.equal(false) + }) + + it('is off when nip77.enabled is false', () => { + expect(isNip77Active({ nip77: { enabled: false } } as Settings)).to.equal(false) + }) + + it('is on when enabled and the event store is postgres', () => { + expect(isNip77Active({ nip77: { enabled: true }, eventStore: { backend: 'postgres' } } as Settings)).to.equal( + true, + ) + }) + + it('treats an unset event store as postgres', () => { + expect(isNip77Active({ nip77: { enabled: true } } as Settings)).to.equal(true) + }) + + it('only accepts a literal true', () => { + expect(isNip77Active({ nip77: { enabled: 'yes' } } as any)).to.equal(false) + }) + }) + + describe('getNip77Limits', () => { + it('falls back to the defaults', () => { + expect(getNip77Limits({} as Settings)).to.deep.equal({ + maxEventsPerSession: DEFAULT_MAX_EVENTS_PER_SESSION, + maxSessionsPerConnection: DEFAULT_MAX_SESSIONS_PER_CONNECTION, + sessionIdleTimeoutMs: DEFAULT_SESSION_IDLE_TIMEOUT_MS, + }) + }) + + it('uses configured values', () => { + expect( + getNip77Limits({ + nip77: { maxEventsPerSession: 10, maxSessionsPerConnection: 2, sessionIdleTimeoutMs: 500 }, + } as Settings), + ).to.deep.equal({ maxEventsPerSession: 10, maxSessionsPerConnection: 2, sessionIdleTimeoutMs: 500 }) + }) + }) + + it('keeps the frame size limit above the 4096 bytes negentropy requires', () => { + expect(NEGENTROPY_FRAME_SIZE_LIMIT).to.be.at.least(4096) + }) +}) diff --git a/test/unit/utils/settings-config.spec.ts b/test/unit/utils/settings-config.spec.ts index d9179f35d..9ab468bcd 100644 --- a/test/unit/utils/settings-config.spec.ts +++ b/test/unit/utils/settings-config.spec.ts @@ -167,6 +167,47 @@ describe('settings-config', () => { }) }) + describe('nip77 settings', () => { + const baseSettings = (nip77: any) => + ({ + info: { relay_url: 'wss://test.relay', name: 'test' }, + network: {}, + nip77, + }) as any + + const nip77Issues = (nip77: any) => + validateSettings(baseSettings(nip77)).filter((issue) => issue.path.startsWith('nip77')) + + it('accepts the shipped defaults', () => { + expect( + nip77Issues({ + enabled: true, + maxEventsPerSession: 100000, + maxSessionsPerConnection: 3, + sessionIdleTimeoutMs: 60000, + }), + ).to.deep.equal([]) + }) + + it('accepts an enabled section that leaves the limits to their defaults', () => { + expect(nip77Issues({ enabled: true })).to.deep.equal([]) + }) + + for (const key of ['maxEventsPerSession', 'maxSessionsPerConnection', 'sessionIdleTimeoutMs']) { + for (const bad of [0, -1, 1.5, '10', Number.NaN]) { + it(`rejects ${key} = ${String(bad)}`, () => { + const issues = nip77Issues({ enabled: true, [key]: bad }) + + expect(issues.map((issue) => issue.path)).to.deep.equal([`nip77.${key}`]) + }) + } + } + + it('does not validate the limits while the feature is off', () => { + expect(nip77Issues({ enabled: false, maxEventsPerSession: 0 })).to.deep.equal([]) + }) + }) + describe('adaptive pow settings', () => { const baseSettings = () => ({