diff --git a/.gitignore b/.gitignore index 1c5f07a05..d2f257c21 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,9 @@ logs *.log npm-debug.log* +# Linter +.lint-reports/ + # Coverage .nyc_output .test-reports @@ -17,8 +20,10 @@ node_modules/ # Optional npm cache directory .npm -# Optional eslint cache +# Caches .eslintcache +.dccache +.DS_Store # generate output dist diff --git a/CONFIGURATION.md b/CONFIGURATION.md index cbff94a5b..c2c7166b0 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -37,5 +37,11 @@ Running `nostr-ts-relay` for the first time creates the settings file in `~/.nos | limits.event.pubkey.blacklist | List of public keys to reject. Public keys in this list will not be able to post to this relay. | | limits.event.createdAt.maxPositiveDelta | Maximum number of seconds an event's `created_at` can be in the future. Defaults to 900 (15 minutes). Disabled when set to zero. | | limits.event.createdAt.minNegativeDelta | Maximum number of secodns an event's `created_at` can be in the past. Defaults to zero. Disabled when set to zero. | +| limits.event.rateLimits[].kinds | List of event kinds rate limited. Use `[min, max]` for ranges. Optional. | +| limits.event.rateLimits[].period | Rate limiting period in milliseconds. | +| limits.event.rateLimits[].rate | Maximum number of events during period. | | limits.client.subscription.maxSubscriptions | Maximum number of subscriptions per connected client. Defaults to 10. Disabled when set to zero. | | limits.client.subscription.maxFilters | Maximum number of filters per subscription. Defaults to 10. Disabled when set to zero. | +| limits.message.rateLimits[].period | Rate limit period in milliseconds. | +| limits.message.rateLimits[].rate | Maximum number of messages during period. | +| limits.message.ipWhitelist | List of IPs (IPv4 or IPv6) without rate limit. | diff --git a/README.md b/README.md index 63e7d664e..db7409c3a 100644 --- a/README.md +++ b/README.md @@ -51,18 +51,25 @@ NIPs with a relay-specific implementation are listed here. - [x] NIP-13: Proof of Work - [x] NIP-15: End of Stored Events Notice - [x] NIP-16: Event Treatment +- [x] NIP-20: Command Results - [x] NIP-22: Event `created_at` Limits - [x] NIP-26: Delegated Event Signing - [x] NIP-33: Parameterized Replaceable Events ## Requirements -- PostgreSQL 15.0 (For standalone steps only) +### Standalone setup +- PostgreSQL 15.0 +- Redis - Node v18 - Typescript -- Docker (For docker steps only, version 20 or higher) -## Quick Start (Docker Compose) (Best) +### Docker setups +- Node v18 +- Docker v20.10 +- Docker compose v2.10 + +## Quick Start (Docker Compose) Install Docker following the [official guide](https://docs.docker.com/engine/install/). You may have to uninstall Docker if you installed it using a different guide. @@ -75,7 +82,7 @@ Clone repository and enter directory: Start with: ``` - npm run docker:compose:start & + npm run docker:compose:start -- --detach ``` Stop the server with: @@ -84,6 +91,7 @@ Stop the server with: ``` ## Quick Start (over Tor) +`Docker` `Tor` Install Docker following the [official guide](https://docs.docker.com/engine/install/). You may have to uninstall Docker if you installed it using a different guide. @@ -96,7 +104,7 @@ Clone repository and enter directory: Start with: ``` - npm run tor:docker:compose:start & + npm run tor:docker:compose:start ``` Print the Tor hostname: @@ -119,6 +127,10 @@ Set the following environment variables: DB_NAME=nostr_ts_relay DB_USER=postgres DB_PASSWORD=postgres + REDIS_HOST=localhost + REDIS_PORT=6379 + REDIS_USER=default + REDIS_PASSWORD=nostr_ts_relay ``` Create `nostr_ts_relay` database: @@ -129,6 +141,15 @@ Create `nostr_ts_relay` database: postgres=# quit ``` +Start Redis and use `redis-cli` to set the default password and verify: + ``` + $ redis-cli + 127.0.0.1:6379> CONFIG SET requirepass "nostr_ts_relay" + OK + 127.0.0.1:6379> AUTH nostr_ts_relay + Ok + ``` + Clone repository and enter directory: ``` git clone git@github.com:Cameri/nostr-ts-relay.git diff --git a/docker-compose.yml b/docker-compose.yml index ba5a01df6..78fcff346 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -3,6 +3,8 @@ services: build: . container_name: nostr-ts-relay environment: + PORT: 8008 + NOSTR_CONFIG_DIR: /home/node/ DB_HOST: db DB_PORT: 5432 DB_USER: nostr_ts_relay @@ -10,14 +12,19 @@ services: DB_NAME: nostr_ts_relay DB_MIN_POOL_SIZE: 2 DB_MAX_POOL_SIZE: 10 - NOSTR_CONFIG_DIR: /home/node/ - PORT: 8008 + REDIS_HOST: cache + REDIS_PORT: 6379 + REDIS_USER: default + REDIS_PASSWORD: nostr_ts_relay + DEBUG: "worker:*" user: node:node volumes: - $HOME/.nostr:/home/node/ ports: - 8008:8008 depends_on: + cache: + condition: service_healthy db: condition: service_healthy migrations: @@ -43,6 +50,21 @@ services: interval: 5s timeout: 5s retries: 5 + cache: + image: redis:7.0.5-alpine3.16 + container_name: cache + volumes: + - cache:/data + command: redis-server --save 20 1 --loglevel warning --requirepass nostr_ts_relay + networks: + default: + ipv4_address: 10.10.10.4 + restart: always + healthcheck: + test: [ "CMD", "redis-cli", "ping", "|", "grep", "PONG" ] + interval: 1s + timeout: 5s + retries: 5 migrations: image: node:18-alpine3.16 container_name: migrations @@ -75,3 +97,4 @@ networks: volumes: pgdata: + cache: \ No newline at end of file diff --git a/package-lock.json b/package-lock.json index 50f58e46d..a779d23f3 100644 --- a/package-lock.json +++ b/package-lock.json @@ -16,6 +16,7 @@ "pg": "8.8.0", "pg-query-stream": "4.2.4", "ramda": "0.28.0", + "redis": "4.5.0", "ws": "^8.10.0" }, "devDependencies": { @@ -1141,6 +1142,59 @@ "@octokit/openapi-types": "^14.0.0" } }, + "node_modules/@redis/bloom": { + "version": "1.1.0", + "resolved": "https://registry.npmjs.org/@redis/bloom/-/bloom-1.1.0.tgz", + "integrity": "sha512-9QovlxmpRtvxVbN0UBcv8WfdSMudNZZTFqCsnBszcQXqaZb/TVe30ScgGEO7u1EAIacTPAo7/oCYjYAxiHLanQ==", + "peerDependencies": { + "@redis/client": "^1.0.0" + } + }, + "node_modules/@redis/client": { + "version": "1.4.0", + "resolved": "https://registry.npmjs.org/@redis/client/-/client-1.4.0.tgz", + "integrity": "sha512-1gEj1AkyXPlkcC/9/T5xpDcQF8ntERURjLBgEWMTdUZqe181zfI9BY3jc2OzjTLkvZh5GV7VT4ktoJG2fV2ufw==", + "dependencies": { + "cluster-key-slot": "1.1.1", + "generic-pool": "3.9.0", + "yallist": "4.0.0" + }, + "engines": { + "node": ">=14" + } + }, + "node_modules/@redis/graph": { + "version": "1.1.0", + "resolved": "https://registry.npmjs.org/@redis/graph/-/graph-1.1.0.tgz", + "integrity": "sha512-16yZWngxyXPd+MJxeSr0dqh2AIOi8j9yXKcKCwVaKDbH3HTuETpDVPcLujhFYVPtYrngSco31BUcSa9TH31Gqg==", + "peerDependencies": { + "@redis/client": "^1.0.0" + } + }, + "node_modules/@redis/json": { + "version": "1.0.4", + "resolved": "https://registry.npmjs.org/@redis/json/-/json-1.0.4.tgz", + "integrity": "sha512-LUZE2Gdrhg0Rx7AN+cZkb1e6HjoSKaeeW8rYnt89Tly13GBI5eP4CwDVr+MY8BAYfCg4/N15OUrtLoona9uSgw==", + "peerDependencies": { + "@redis/client": "^1.0.0" + } + }, + "node_modules/@redis/search": { + "version": "1.1.0", + "resolved": "https://registry.npmjs.org/@redis/search/-/search-1.1.0.tgz", + "integrity": "sha512-NyFZEVnxIJEybpy+YskjgOJRNsfTYqaPbK/Buv6W2kmFNaRk85JiqjJZA5QkRmWvGbyQYwoO5QfDi2wHskKrQQ==", + "peerDependencies": { + "@redis/client": "^1.0.0" + } + }, + "node_modules/@redis/time-series": { + "version": "1.0.4", + "resolved": "https://registry.npmjs.org/@redis/time-series/-/time-series-1.0.4.tgz", + "integrity": "sha512-ThUIgo2U/g7cCuZavucQTQzA9g9JbDDY2f64u3AbAoz/8vE2lt2U37LamDUVChhaDA3IRT9R6VvJwqnUfTJzng==", + "peerDependencies": { + "@redis/client": "^1.0.0" + } + }, "node_modules/@semantic-release/commit-analyzer": { "version": "9.0.2", "resolved": "https://registry.npmjs.org/@semantic-release/commit-analyzer/-/commit-analyzer-9.0.2.tgz", @@ -2526,6 +2580,14 @@ "wrap-ansi": "^7.0.0" } }, + "node_modules/cluster-key-slot": { + "version": "1.1.1", + "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.1.tgz", + "integrity": "sha512-rwHwUfXL40Chm1r08yrhU3qpUvdVlgkKNeyeGPOxnW8/SyVDvgRaed/Uz54AqWNaTCAThlj6QAs3TZcKI0xDEw==", + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/color-convert": { "version": "2.0.1", "resolved": "https://registry.npmjs.org/color-convert/-/color-convert-2.0.1.tgz", @@ -3820,6 +3882,14 @@ "integrity": "sha1-GwqzvVU7Kg1jmdKcDj6gslIHgyc=", "dev": true }, + "node_modules/generic-pool": { + "version": "3.9.0", + "resolved": "https://registry.npmjs.org/generic-pool/-/generic-pool-3.9.0.tgz", + "integrity": "sha512-hymDOu5B53XvN4QT9dBmZxPX4CWhBPPLguTZ9MMFeFa/Kg0xWVfylOVNlJji/E7yTZWFd/q9GO5TxDLq156D7g==", + "engines": { + "node": ">= 4" + } + }, "node_modules/gensync": { "version": "1.0.0-beta.2", "resolved": "https://registry.npmjs.org/gensync/-/gensync-1.0.0-beta.2.tgz", @@ -9629,6 +9699,19 @@ "esprima": "~4.0.0" } }, + "node_modules/redis": { + "version": "4.5.0", + "resolved": "https://registry.npmjs.org/redis/-/redis-4.5.0.tgz", + "integrity": "sha512-oZGAmOKG+RPnHo0UxM5GGjJ0dBd/Vi4fs3MYwM1p2baDoXC0wpm0yOdpxVS9K+0hM84ycdysp2eHg2xGoQ4FEw==", + "dependencies": { + "@redis/bloom": "1.1.0", + "@redis/client": "1.4.0", + "@redis/graph": "1.1.0", + "@redis/json": "1.0.4", + "@redis/search": "1.1.0", + "@redis/time-series": "1.0.4" + } + }, "node_modules/reflect-metadata": { "version": "0.1.13", "resolved": "https://registry.npmjs.org/reflect-metadata/-/reflect-metadata-0.1.13.tgz", @@ -11289,8 +11372,7 @@ "node_modules/yallist": { "version": "4.0.0", "resolved": "https://registry.npmjs.org/yallist/-/yallist-4.0.0.tgz", - "integrity": "sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==", - "dev": true + "integrity": "sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==" }, "node_modules/yaml": { "version": "1.10.2", @@ -12269,6 +12351,46 @@ "@octokit/openapi-types": "^14.0.0" } }, + "@redis/bloom": { + "version": "1.1.0", + "resolved": "https://registry.npmjs.org/@redis/bloom/-/bloom-1.1.0.tgz", + "integrity": "sha512-9QovlxmpRtvxVbN0UBcv8WfdSMudNZZTFqCsnBszcQXqaZb/TVe30ScgGEO7u1EAIacTPAo7/oCYjYAxiHLanQ==", + "requires": {} + }, + "@redis/client": { + "version": "1.4.0", + "resolved": "https://registry.npmjs.org/@redis/client/-/client-1.4.0.tgz", + "integrity": "sha512-1gEj1AkyXPlkcC/9/T5xpDcQF8ntERURjLBgEWMTdUZqe181zfI9BY3jc2OzjTLkvZh5GV7VT4ktoJG2fV2ufw==", + "requires": { + "cluster-key-slot": "1.1.1", + "generic-pool": "3.9.0", + "yallist": "4.0.0" + } + }, + "@redis/graph": { + "version": "1.1.0", + "resolved": "https://registry.npmjs.org/@redis/graph/-/graph-1.1.0.tgz", + "integrity": "sha512-16yZWngxyXPd+MJxeSr0dqh2AIOi8j9yXKcKCwVaKDbH3HTuETpDVPcLujhFYVPtYrngSco31BUcSa9TH31Gqg==", + "requires": {} + }, + "@redis/json": { + "version": "1.0.4", + "resolved": "https://registry.npmjs.org/@redis/json/-/json-1.0.4.tgz", + "integrity": "sha512-LUZE2Gdrhg0Rx7AN+cZkb1e6HjoSKaeeW8rYnt89Tly13GBI5eP4CwDVr+MY8BAYfCg4/N15OUrtLoona9uSgw==", + "requires": {} + }, + "@redis/search": { + "version": "1.1.0", + "resolved": "https://registry.npmjs.org/@redis/search/-/search-1.1.0.tgz", + "integrity": "sha512-NyFZEVnxIJEybpy+YskjgOJRNsfTYqaPbK/Buv6W2kmFNaRk85JiqjJZA5QkRmWvGbyQYwoO5QfDi2wHskKrQQ==", + "requires": {} + }, + "@redis/time-series": { + "version": "1.0.4", + "resolved": "https://registry.npmjs.org/@redis/time-series/-/time-series-1.0.4.tgz", + "integrity": "sha512-ThUIgo2U/g7cCuZavucQTQzA9g9JbDDY2f64u3AbAoz/8vE2lt2U37LamDUVChhaDA3IRT9R6VvJwqnUfTJzng==", + "requires": {} + }, "@semantic-release/commit-analyzer": { "version": "9.0.2", "resolved": "https://registry.npmjs.org/@semantic-release/commit-analyzer/-/commit-analyzer-9.0.2.tgz", @@ -13319,6 +13441,11 @@ "wrap-ansi": "^7.0.0" } }, + "cluster-key-slot": { + "version": "1.1.1", + "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.1.tgz", + "integrity": "sha512-rwHwUfXL40Chm1r08yrhU3qpUvdVlgkKNeyeGPOxnW8/SyVDvgRaed/Uz54AqWNaTCAThlj6QAs3TZcKI0xDEw==" + }, "color-convert": { "version": "2.0.1", "resolved": "https://registry.npmjs.org/color-convert/-/color-convert-2.0.1.tgz", @@ -14325,6 +14452,11 @@ "integrity": "sha1-GwqzvVU7Kg1jmdKcDj6gslIHgyc=", "dev": true }, + "generic-pool": { + "version": "3.9.0", + "resolved": "https://registry.npmjs.org/generic-pool/-/generic-pool-3.9.0.tgz", + "integrity": "sha512-hymDOu5B53XvN4QT9dBmZxPX4CWhBPPLguTZ9MMFeFa/Kg0xWVfylOVNlJji/E7yTZWFd/q9GO5TxDLq156D7g==" + }, "gensync": { "version": "1.0.0-beta.2", "resolved": "https://registry.npmjs.org/gensync/-/gensync-1.0.0-beta.2.tgz", @@ -18566,6 +18698,19 @@ "esprima": "~4.0.0" } }, + "redis": { + "version": "4.5.0", + "resolved": "https://registry.npmjs.org/redis/-/redis-4.5.0.tgz", + "integrity": "sha512-oZGAmOKG+RPnHo0UxM5GGjJ0dBd/Vi4fs3MYwM1p2baDoXC0wpm0yOdpxVS9K+0hM84ycdysp2eHg2xGoQ4FEw==", + "requires": { + "@redis/bloom": "1.1.0", + "@redis/client": "1.4.0", + "@redis/graph": "1.1.0", + "@redis/json": "1.0.4", + "@redis/search": "1.1.0", + "@redis/time-series": "1.0.4" + } + }, "reflect-metadata": { "version": "0.1.13", "resolved": "https://registry.npmjs.org/reflect-metadata/-/reflect-metadata-0.1.13.tgz", @@ -19814,8 +19959,7 @@ "yallist": { "version": "4.0.0", "resolved": "https://registry.npmjs.org/yallist/-/yallist-4.0.0.tgz", - "integrity": "sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==", - "dev": true + "integrity": "sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==" }, "yaml": { "version": "1.10.2", diff --git a/package.json b/package.json index 6a43edb59..6f3461f21 100644 --- a/package.json +++ b/package.json @@ -11,6 +11,7 @@ 12, 15, 16, + 20, 22, 26, 33 @@ -23,7 +24,7 @@ "prestart": "npm run build", "start": "cd dist && node src/index.js", "build:check": "npm run build -- --noEmit", - "lint": "eslint ./src ./test --ext .ts", + "lint": "eslint -o .lint-reports/eslint.json -f json --ext .ts ./src ./test", "lint:fix": "npm run lint -- --fix", "db:migrate": "knex migrate:latest", "db:migrate:rollback": "knex migrate:rollback", @@ -108,6 +109,7 @@ "pg": "8.8.0", "pg-query-stream": "4.2.4", "ramda": "0.28.0", + "redis": "4.5.0", "ws": "8.10.0" } } diff --git a/settings.sample.json b/settings.sample.json index 7e9c3eed4..5a58f744a 100644 --- a/settings.sample.json +++ b/settings.sample.json @@ -23,13 +23,53 @@ "createdAt": { "maxPositiveDelta": 900, "maxNegativeDelta": 0 - } + }, + "rateLimits": [ + { + "kinds": [[0, 5], 7, [40, 49], [10000, 19999], [30000, 39999]], + "period": 60000, + "rate": 60 + }, + { + "kinds": [[20000, 29999]], + "period": 60000, + "rate": 600 + }, + { + "period": 3600000, + "rate": 3600 + }, + { + "period": 86400000, + "rate": 86400 + } + ] }, "client": { "subscription": { "maxSubscriptions": 10, "maxFilters": 10 } + }, + "message": { + "rateLimits": [ + { + "period": 60000, + "rate": 600 + }, + { + "period": 3600000, + "rate": 3600 + }, + { + "period": 86400000, + "rate": 86400 + } + ], + "ipWhitelist": [ + "::1", + "::ffff:10.10.10.1" + ] } } } \ No newline at end of file diff --git a/src/@types/adapters.ts b/src/@types/adapters.ts index d30a4caf3..ebddfe738 100644 --- a/src/@types/adapters.ts +++ b/src/@types/adapters.ts @@ -15,3 +15,10 @@ export type IWebSocketAdapter = EventEmitter & { getClientId(): string getSubscriptions(): Map } + +export interface ICacheAdapter { + addToSortedSet(key: string, set: Record | Record[]): Promise + removeRangeByScoreFromSortedSet(key: string, min: number, max: number): Promise + getRangeFromSortedSet(key: string, start: number, stop: number): Promise + setKeyExpiry(key: string, expiry: number): Promise +} diff --git a/src/@types/base.ts b/src/@types/base.ts index 5e11f846a..16e23cd1f 100644 --- a/src/@types/base.ts +++ b/src/@types/base.ts @@ -23,7 +23,7 @@ export type Range = Exclude< Enumerate > -export type Factory = (input: TInput) => TOutput +export type Factory = (input: TInput) => TOutput export type DatabaseClient = Knex diff --git a/src/@types/cache.ts b/src/@types/cache.ts new file mode 100644 index 000000000..2c2245467 --- /dev/null +++ b/src/@types/cache.ts @@ -0,0 +1,8 @@ +import { + RedisClientType, + RedisFunctions, + RedisModules, + RedisScripts, +} from 'redis' + +export type CacheClient = RedisClientType diff --git a/src/@types/messages.ts b/src/@types/messages.ts index 037e6819e..49a818992 100644 --- a/src/@types/messages.ts +++ b/src/@types/messages.ts @@ -1,6 +1,6 @@ +import { EventId, Range } from './base' import { SubscriptionFilter, SubscriptionId } from './subscription' import { Event } from './event' -import { Range } from './base' export enum MessageType { REQ = 'REQ', @@ -8,6 +8,7 @@ export enum MessageType { CLOSE = 'CLOSE', NOTICE = 'NOTICE', EOSE = 'EOSE', + OK = 'OK' } export type IncomingMessage = @@ -20,6 +21,7 @@ export type OutgoingMessage = | OutgoingEventMessage | EndOfStoredEventsNotice | NoticeMessage + | CommandResult export type SubscribeMessage = { [index in Range<2, 100>]: SubscriptionFilter @@ -51,6 +53,13 @@ export interface NoticeMessage { 1: string } +export interface CommandResult { + 0: MessageType.OK + 1: EventId + 2: boolean + 3: string +} + export interface EndOfStoredEventsNotice { 0: MessageType.EOSE 1: SubscriptionId diff --git a/src/@types/settings.ts b/src/@types/settings.ts index 649b2adee..98cb99741 100644 --- a/src/@types/settings.ts +++ b/src/@types/settings.ts @@ -21,6 +21,12 @@ export interface PubkeyLimits { export type EventKindsRange = [EventKinds, EventKinds] +export interface EventRateLimit { + kinds?: (EventKinds | [EventKinds, EventKinds])[] + rate: number + period: number +} + export interface KindLimits { whitelist?: (EventKinds | EventKindsRange)[] blacklist?: (EventKinds | EventKindsRange)[] @@ -42,6 +48,7 @@ export interface EventLimits { pubkey?: PubkeyLimits kind?: KindLimits createdAt?: CreatedAtLimits + rateLimits?: EventRateLimit[] } export interface ClientSubscriptionLimits { @@ -53,9 +60,20 @@ export interface ClientLimits { subscription?: ClientSubscriptionLimits } +export interface MessageRateLimit { + rate: number + period: number +} + +export interface MessageLimits { + rateLimits?: MessageRateLimit[] + ipWhitelist?: string[] +} + export interface Limits { client?: ClientLimits event?: EventLimits + message?: MessageLimits } export interface Worker { diff --git a/src/@types/utils.ts b/src/@types/utils.ts new file mode 100644 index 000000000..1548deb64 --- /dev/null +++ b/src/@types/utils.ts @@ -0,0 +1,8 @@ +export interface IRateLimiterOptions { + period: number; + rate: number; +} + +export interface IRateLimiter { + hit(key: string, step: number, options: IRateLimiterOptions): Promise +} diff --git a/src/adapters/redis-adapter.ts b/src/adapters/redis-adapter.ts new file mode 100644 index 000000000..63c8dd410 --- /dev/null +++ b/src/adapters/redis-adapter.ts @@ -0,0 +1,80 @@ +import { CacheClient } from '../@types/cache' +import { createLogger } from '../factories/logger-factory' +import { ICacheAdapter } from '../@types/adapters' + +const debug = createLogger('redis-adapter') + +export class RedisAdapter implements ICacheAdapter { + private connection: Promise + + public constructor(private readonly client: CacheClient) { + this.connection = client.connect() + + this.connection.catch((error) => this.onClientError(error)) + + this.client + .on('connect', () => debug('connecting')) + .on('ready', () => debug('connected')) + .on('error', (error) => this.onClientError(error)) + .on('reconnecting', () => { + debug('reconnecting') + this.connection = new Promise((resolve, reject) => { + const cleanup = () => { + this.client.removeListener('ready', onReady) + this.client.removeListener('error', onError) + } + + const onError = (error: Error) => { + cleanup() + reject(error) + } + + const onReady = () => { + cleanup() + resolve() + } + + this.client.once('ready', onReady) + + this.client.once('error', onError) + }) + }) + } + + private onClientError(error: Error) { + console.error('Unable to connect to Redis.', error) + // throw error + } + + public async removeRangeByScoreFromSortedSet(key: string, min: number, max: number): Promise { + await this.connection + debug('remove %d..%d range from sorted set %s', min, max, key) + return this.client.zRemRangeByScore(key, min, max) + } + + public async getRangeFromSortedSet(key: string, min: number, max: number): Promise { + await this.connection + debug('get %d..%d range from sorted set %s', min, max, key) + return this.client.zRange(key, min, max) + } + + public async setKeyExpiry(key: string, expiry: number): Promise { + await this.connection + debug('expire at %d from sorted set %s', expiry, key) + await this.client.expire(key, expiry) + } + + public async addToSortedSet( + key: string, + set: Record + ): Promise { + await this.connection + debug('add %o to sorted set %s', set, key) + const members = Object + .entries(set) + .map(([value, score]) => ({ score: Number(score), value })) + + return this.client.zAdd(key, members) + } + +} \ No newline at end of file diff --git a/src/adapters/web-server-adapter.ts b/src/adapters/web-server-adapter.ts index 28be6475b..ccb4ece1c 100644 --- a/src/adapters/web-server-adapter.ts +++ b/src/adapters/web-server-adapter.ts @@ -33,7 +33,7 @@ export class WebServerAdapter extends EventEmitter implements IWebServerAdapter } private onRequest(request: IncomingMessage, response: ServerResponse) { - debug('request received: %O', request.rawHeaders) + debug('request received: %O', request.headers) if (request.method === 'GET' && request.headers['accept'] === 'application/nostr+json') { const { info: { name, description, pubkey, contact }, diff --git a/src/adapters/web-socket-adapter.ts b/src/adapters/web-socket-adapter.ts index e7c13f6e4..77d4f4b70 100644 --- a/src/adapters/web-socket-adapter.ts +++ b/src/adapters/web-socket-adapter.ts @@ -13,6 +13,8 @@ import { attemptValidation } from '../utils/validation' import { createLogger } from '../factories/logger-factory' import { Event } from '../@types/event' import { Factory } from '../@types/base' +import { IRateLimiter } from '../@types/utils' +import { ISettings } from '../@types/settings' import { isEventMatchingFilter } from '../utils/event' import { messageSchema } from '../schemas/message-schema' @@ -21,7 +23,7 @@ const debugHeartbeat = debug.extend('heartbeat') export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter { public clientId: string - // private clientAddress: string + private clientAddress: string private alive: boolean private subscriptions: Map @@ -30,13 +32,17 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter private readonly request: IncomingHttpMessage, private readonly webSocketServer: IWebSocketServerAdapter, private readonly createMessageHandler: Factory, + private readonly slidingWindowRateLimiter: Factory, + private readonly settingsFactory: Factory, ) { super() this.alive = true this.subscriptions = new Map() this.clientId = Buffer.from(this.request.headers['sec-websocket-key'], 'base64').toString('hex') - // this.clientAddress = this.request.headers['x-forwarded-for'] as string + this.clientAddress = (this.request.headers['x-forwarded-for'] ?? this.request.socket.remoteAddress) as string + + debug('client %s from address %s', this.clientId, this.clientAddress) this.client .on('message', this.onClientMessage.bind(this)) @@ -120,10 +126,15 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter private async onClientMessage(raw: Buffer) { let abort: () => void try { + if (await this.isRateLimited(this.clientAddress)) { + this.sendMessage(createNoticeMessage('rate limited')) + return + } + const message = attemptValidation(messageSchema)(JSON.parse(raw.toString('utf8'))) const messageHandler = this.createMessageHandler([message, this]) as IMessageHandler & IAbortable - if (typeof messageHandler.abort === 'function') { + if (typeof messageHandler?.abort === 'function') { abort = messageHandler.abort.bind(messageHandler) this.client.prependOnceListener('close', abort) } @@ -145,6 +156,36 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter } } + private async isRateLimited(client: string): Promise { + const { + rateLimits, + ipWhitelist = [], + } = this.settingsFactory().limits?.message ?? {} + + if (ipWhitelist.includes(client)) { + debug('rate limit check %s: skipped', client) + return false + } + + const rateLimiter = this.slidingWindowRateLimiter() + + const hit = (period: number, rate: number) => + rateLimiter.hit( + `${client}:message:${period}`, + 1, + { period: period, rate: rate }, + ) + + const hits = await Promise.all( + rateLimits + .map(({ period, rate }) => hit(period, rate)) + ) + + debug('rate limit check %s: %o = %o', client, rateLimits.map(({ period }) => period), hits) + + return hits.some((thresholdCrossed) => thresholdCrossed) + } + private onClientPong() { debugHeartbeat('client %s pong', this.clientId) this.alive = true diff --git a/src/cache/client.ts b/src/cache/client.ts new file mode 100644 index 000000000..7a19fff63 --- /dev/null +++ b/src/cache/client.ts @@ -0,0 +1,19 @@ +import { createClient, RedisClientOptions } from 'redis' +import { CacheClient } from '../@types/cache' + +export const getCacheConfig = (): RedisClientOptions => ({ + url: `redis://${process.env.REDIS_USER}:${process.env.REDIS_PASSWORD}@${process.env.REDIS_HOST}:${process.env.REDIS_PORT}`, + password: process.env.REDIS_PASSWORD, +}) + +let instance: CacheClient = undefined + +export const getCacheClient = (): CacheClient => { + if (!instance) { + const config = getCacheConfig() + + instance = createClient(config) + } + + return instance +} diff --git a/src/constants/base.ts b/src/constants/base.ts index f5cee9f0c..3397d251a 100644 --- a/src/constants/base.ts +++ b/src/constants/base.ts @@ -6,10 +6,21 @@ export enum EventKinds { ENCRYPTED_DIRECT_MESSAGE = 4, DELETE = 5, REACTION = 7, + // Channels + CHANNEL_CREATION = 40, + CHANNEL_METADATA = 41, + CHANNEL_MESSAGE = 42, + CHANNEL_HIDE_MESSAGE = 43, + CHANNEL_MUTE_USER = 44, + CHANNEL_RESERVED_FIRST = 45, + CHANNEL_RESERVED_LAST = 49, + // Replaceable events REPLACEABLE_FIRST = 10000, REPLACEABLE_LAST = 19999, + // Ephemeral events EPHEMERAL_FIRST = 20000, EPHEMERAL_LAST = 29999, + // Parameterized replaceable events PARAMETERIZED_REPLACEABLE_FIRST = 30000, PARAMETERIZED_REPLACEABLE_LAST = 39999, } diff --git a/src/factories/message-handler-factory.ts b/src/factories/message-handler-factory.ts index c3bcbbac1..9ebda77c5 100644 --- a/src/factories/message-handler-factory.ts +++ b/src/factories/message-handler-factory.ts @@ -7,6 +7,7 @@ import { eventStrategyFactory } from './event-strategy-factory' import { IEventRepository } from '../@types/repositories' import { isDelegatedEvent } from '../utils/event' import { IWebSocketAdapter } from '../@types/adapters' +import { slidingWindowRateLimiterFactory } from './rate-limiter-factory' import { SubscribeMessageHandler } from '../handlers/subscribe-message-handler' import { UnsubscribeMessageHandler } from '../handlers/unsubscribe-message-handler' @@ -20,11 +21,17 @@ export const messageHandlerFactory = ( return new DelegatedEventMessageHandler( adapter, delegatedEventStrategyFactory(eventRepository), - createSettings + createSettings, + slidingWindowRateLimiterFactory, ) } - return new EventMessageHandler(adapter, eventStrategyFactory(eventRepository), createSettings) + return new EventMessageHandler( + adapter, + eventStrategyFactory(eventRepository), + createSettings, + slidingWindowRateLimiterFactory, + ) } case MessageType.REQ: return new SubscribeMessageHandler(adapter, eventRepository, createSettings) diff --git a/src/factories/rate-limiter-factory.ts b/src/factories/rate-limiter-factory.ts new file mode 100644 index 000000000..5e33f8c70 --- /dev/null +++ b/src/factories/rate-limiter-factory.ts @@ -0,0 +1,16 @@ +import { getCacheClient } from '../cache/client' +import { ICacheAdapter } from '../@types/adapters' +import { IRateLimiter } from '../@types/utils' +import { RedisAdapter } from '../adapters/redis-adapter' +import { SlidingWindowRateLimiter } from '../utils/sliding-window-rate-limiter' + +let instance: IRateLimiter = undefined + +export const slidingWindowRateLimiterFactory = () => { + if (!instance) { + const cache: ICacheAdapter = new RedisAdapter(getCacheClient()) + instance = new SlidingWindowRateLimiter(cache) + } + + return instance +} diff --git a/src/factories/websocket-adapter-factory.ts b/src/factories/websocket-adapter-factory.ts index c0d5c3676..481321f30 100644 --- a/src/factories/websocket-adapter-factory.ts +++ b/src/factories/websocket-adapter-factory.ts @@ -1,9 +1,11 @@ import { IncomingMessage } from 'http' import { WebSocket } from 'ws' +import { createSettings } from './settings-factory' import { IEventRepository } from '../@types/repositories' import { IWebSocketServerAdapter } from '../@types/adapters' import { messageHandlerFactory } from './message-handler-factory' +import { slidingWindowRateLimiterFactory } from './rate-limiter-factory' import { WebSocketAdapter } from '../adapters/web-socket-adapter' @@ -14,5 +16,7 @@ export const webSocketAdapterFactory = ( client, request, webSocketServerAdapter, - messageHandlerFactory(eventRepository) + messageHandlerFactory(eventRepository), + slidingWindowRateLimiterFactory, + createSettings, ) diff --git a/src/handlers/delegated-event-message-handler.ts b/src/handlers/delegated-event-message-handler.ts index 15f5ac7f8..09e743895 100644 --- a/src/handlers/delegated-event-message-handler.ts +++ b/src/handlers/delegated-event-message-handler.ts @@ -1,9 +1,7 @@ -import { mergeDeepLeft } from 'ramda' - -import { DelegatedEvent, Event } from '../@types/event' import { EventDelegatorMetadataKey, EventTags } from '../constants/base' +import { createCommandResult } from '../utils/messages' import { createLogger } from '../factories/logger-factory' -import { createNoticeMessage } from '../utils/messages' +import { DelegatedEvent } from '../@types/event' import { EventMessageHandler } from './event-message-handler' import { IMessageHandler } from '../@types/message-handlers' import { IncomingEventMessage } from '../@types/messages' @@ -14,33 +12,39 @@ const debug = createLogger('delegated-event-message-handler') export class DelegatedEventMessageHandler extends EventMessageHandler implements IMessageHandler { public async handleMessage(message: IncomingEventMessage): Promise { + debug('received message: %o', message) const [, event] = message - let reason = this.canAcceptEvent(event) + let reason = await this.isEventValid(event) if (reason) { debug('event %s rejected: %s', event.id, reason) - this.webSocket.emit(WebSocketAdapterEvent.Message, createNoticeMessage(`Event rejected: ${reason}`)) + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, reason)) + return + } + + if (await this.isRateLimited(event)) { + debug('event %s rejected: rate-limited') + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, 'rate-limited: slow down')) return } - reason = await this.isEventValid(event) + reason = this.canAcceptEvent(event) if (reason) { debug('event %s rejected: %s', event.id, reason) - this.webSocket.emit(WebSocketAdapterEvent.Message, createNoticeMessage(`Event rejected: ${reason}`)) + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, reason)) return } const [, delegator] = event.tags.find((tag) => tag.length === 4 && tag[0] === EventTags.Delegation) - const delegatedEvent: DelegatedEvent = mergeDeepLeft( - event, - { + const delegatedEvent: DelegatedEvent = { + ...event, [EventDelegatorMetadataKey]: delegator, - } - ) + } const strategy = this.strategyFactory([delegatedEvent, this.webSocket]) if (typeof strategy?.execute !== 'function') { + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, 'error: event not supported')) return } @@ -48,16 +52,18 @@ export class DelegatedEventMessageHandler extends EventMessageHandler implements await strategy.execute(delegatedEvent) } catch (error) { debug('error handling message %o: %o', message, error) + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, 'error: unable to process event')) } } - protected async isEventValid(event: Event): Promise { + protected async isEventValid(event: DelegatedEvent): Promise { const reason = await super.isEventValid(event) if (reason) { return reason } + if (!await isDelegatedEventValid(event)) { - return `Event with id ${event.id} from ${event.pubkey} is invalid delegated event` + return 'invalid: delegation verification failed' } } } diff --git a/src/handlers/event-message-handler.ts b/src/handlers/event-message-handler.ts index 0a2198ea0..183b1c3d3 100644 --- a/src/handlers/event-message-handler.ts +++ b/src/handlers/event-message-handler.ts @@ -1,12 +1,13 @@ -import { EventKindsRange, ISettings } from '../@types/settings' +import { EventKindsRange, EventRateLimit, ISettings } from '../@types/settings' import { getEventProofOfWork, getPubkeyProofOfWork, isEventIdValid, isEventSignatureValid } from '../utils/event' import { IEventStrategy, IMessageHandler } from '../@types/message-handlers' +import { createCommandResult } from '../utils/messages' import { createLogger } from '../factories/logger-factory' -import { createNoticeMessage } from '../utils/messages' import { Event } from '../@types/event' import { EventKinds } from '../constants/base' import { Factory } from '../@types/base' import { IncomingEventMessage } from '../@types/messages' +import { IRateLimiter } from '../@types/utils' import { IWebSocketAdapter } from '../@types/adapters' import { WebSocketAdapterEvent } from '../constants/adapter' @@ -16,7 +17,8 @@ export class EventMessageHandler implements IMessageHandler { public constructor( protected readonly webSocket: IWebSocketAdapter, protected readonly strategyFactory: Factory>, [Event, IWebSocketAdapter]>, - private readonly settings: () => ISettings + private readonly settings: () => ISettings, + private readonly slidingWindowRateLimiter: Factory, ) { } public async handleMessage(message: IncomingEventMessage): Promise { @@ -26,20 +28,27 @@ export class EventMessageHandler implements IMessageHandler { let reason = await this.isEventValid(event) if (reason) { debug('event %s rejected: %s', event.id, reason) - this.webSocket.emit(WebSocketAdapterEvent.Message, createNoticeMessage(`Event rejected: ${reason}`)) + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, reason)) + return + } + + if (await this.isRateLimited(event)) { + debug('event %s rejected: rate-limited') + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, 'rate-limited: slow down')) return } reason = this.canAcceptEvent(event) if (reason) { debug('event %s rejected: %s', event.id, reason) - this.webSocket.emit(WebSocketAdapterEvent.Message, createNoticeMessage(`Event rejected: ${reason}`)) + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, reason)) return } const strategy = this.strategyFactory([event, this.webSocket]) if (typeof strategy?.execute !== 'function') { + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, 'error: event not supported')) return } @@ -47,46 +56,47 @@ export class EventMessageHandler implements IMessageHandler { await strategy.execute(event) } catch (error) { debug('error handling message %o: %o', message, error) + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, 'error: unable to process event')) } } protected canAcceptEvent(event: Event): string | undefined { const now = Math.floor(Date.now()/1000) const limits = this.settings().limits.event - if (limits.createdAt.maxPositiveDelta > 0) { - if (event.created_at > now + limits.createdAt.maxPositiveDelta) { - return `created_at is more than ${limits.createdAt.maxPositiveDelta} seconds in the future` - } + if (limits.createdAt.maxPositiveDelta > 0 && event.created_at > now + limits.createdAt.maxPositiveDelta) { + return `rejected: created_at is more than ${limits.createdAt.maxPositiveDelta} seconds in the future` } - if (limits.createdAt.maxNegativeDelta > 0) { - if (event.created_at < now - limits.createdAt.maxNegativeDelta) { - return `created_at is more than ${limits.createdAt.maxNegativeDelta} seconds in the past` - } + if (limits.createdAt.maxNegativeDelta > 0 && event.created_at < now - limits.createdAt.maxNegativeDelta) { + return `rejected: created_at is more than ${limits.createdAt.maxNegativeDelta} seconds in the past` } if (limits.eventId.minLeadingZeroBits > 0) { - if (getEventProofOfWork(event) < limits.eventId.minLeadingZeroBits) { - return `insufficient proof of work: event Id has less than ${limits.eventId.minLeadingZeroBits} leading zero bits` + const pow = getEventProofOfWork(event.id) + if (pow < limits.eventId.minLeadingZeroBits) { + return `pow: difficulty ${pow}<${limits.eventId.minLeadingZeroBits}` } } if (limits.pubkey.minLeadingZeroBits > 0) { - if (getPubkeyProofOfWork(event.pubkey) < limits.pubkey.minLeadingZeroBits) { - return `insufficient proof of work: pubkey has less than ${limits.pubkey.minLeadingZeroBits} leading zero bits` + const pow = getPubkeyProofOfWork(event.pubkey) + if (pow < limits.pubkey.minLeadingZeroBits) { + return `pow: pubkey difficulty ${pow}<${limits.pubkey.minLeadingZeroBits}` } } - if (limits.pubkey.whitelist.length > 0) { - if (!limits.pubkey.whitelist.some((prefix) => event.pubkey.startsWith(prefix))) { - return `pubkey ${event.pubkey} is not allowed` - } + if ( + limits.pubkey.whitelist.length > 0 + && !limits.pubkey.whitelist.some((prefix) => event.pubkey.startsWith(prefix)) + ) { + return 'blocked: pubkey not allowed' } - if (limits.pubkey.blacklist.length > 0) { - if (limits.pubkey.blacklist.some((prefix) => event.pubkey.startsWith(prefix))) { - return `pubkey ${event.pubkey} is not allowed` - } + if ( + limits.pubkey.blacklist.length > 0 + && limits.pubkey.blacklist.some((prefix) => event.pubkey.startsWith(prefix)) + ) { + return 'blocked: pubkey not allowed' } const isEventKindMatch = (item: EventKinds | EventKindsRange) => @@ -94,25 +104,52 @@ export class EventMessageHandler implements IMessageHandler { ? item === event.kind : event.kind >= item[0] && event.kind <= item[1] - if (limits.kind.whitelist.length > 0) { - if (!limits.kind.whitelist.some(isEventKindMatch)) { - return `event kind ${event.kind} is not allowed` - } + if (limits.kind.whitelist.length > 0 && !limits.kind.whitelist.some(isEventKindMatch)) { + return `blocked: event kind ${event.kind} not allowed` } - if (limits.kind.blacklist.length > 0) { - if (limits.kind.blacklist.some(isEventKindMatch)) { - return `event kind ${event.kind} is not allowed` - } + if (limits.kind.blacklist.length > 0 && limits.kind.blacklist.some(isEventKindMatch)) { + return `blocked: event kind ${event.kind} not allowed` } } protected async isEventValid(event: Event): Promise { if (!await isEventIdValid(event)) { - return `Event with id ${event.id} from ${event.pubkey} is not valid` + return 'invalid: event id does not match' } if (!await isEventSignatureValid(event)) { - return `Event with id ${event.id} from ${event.pubkey} has invalid signature` + return 'invalid: event signature verification failed' + } + } + + protected async isRateLimited(event: Event): Promise { + const rateLimits = this.settings().limits?.event?.rateLimits + if (!rateLimits || !rateLimits.length) { + return + } + + const rateLimiter = this.slidingWindowRateLimiter() + + const toString = (input: any | any[]): string => { + return Array.isArray(input) ? `[${input.map(toString)}]` : input.toString() } + + const hit = ({ period, rate, kinds = undefined }: EventRateLimit) => { + const key = Array.isArray(kinds) + ? `${event.pubkey}:events:${period}:${toString(kinds)}` + : `${event.pubkey}:events:${period}` + + return rateLimiter.hit( + key, + 1, + { period, rate }, + ) + } + + const hits = await Promise.all(rateLimits.map(hit)) + + debug('rate limit check %s: %o', event.pubkey, hits) + + return hits.some((active) => active) } } diff --git a/src/handlers/event-strategies/default-event-strategy.ts b/src/handlers/event-strategies/default-event-strategy.ts index 2cf0bcc47..48b5b886c 100644 --- a/src/handlers/event-strategies/default-event-strategy.ts +++ b/src/handlers/event-strategies/default-event-strategy.ts @@ -1,3 +1,4 @@ +import { createCommandResult } from '../../utils/messages' import { createLogger } from '../../factories/logger-factory' import { Event } from '../../@types/event' import { IEventRepository } from '../../@types/repositories' @@ -16,9 +17,10 @@ export class DefaultEventStrategy implements IEventStrategy public async execute(event: Event): Promise { debug('received event: %o', event) const count = await this.eventRepository.create(event) - if (!count) { - return + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, (count) ? '' : 'duplicate:')) + + if (count) { + this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event) } - this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event) } } diff --git a/src/handlers/event-strategies/delete-event-strategy.ts b/src/handlers/event-strategies/delete-event-strategy.ts index 4533475be..b6b594466 100644 --- a/src/handlers/event-strategies/delete-event-strategy.ts +++ b/src/handlers/event-strategies/delete-event-strategy.ts @@ -1,3 +1,4 @@ +import { createCommandResult } from '../../utils/messages' import { createLogger } from '../../factories/logger-factory' import { Event } from '../../@types/event' import { EventTags } from '../../constants/base' @@ -16,17 +17,25 @@ export class DeleteEventStrategy implements IEventStrategy> public async execute(event: Event): Promise { debug('received event: %o', event) - await this.eventRepository.create(event) + const count = await this.eventRepository.create(event) + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, (count) ? '' : 'duplicate:')) - const eTags = event.tags.filter((tag) => tag[0] === EventTags.Event) + const ids = event.tags.reduce( + (eventIds, tag) => (tag.length >= 2 && tag[0] === EventTags.Event) + ? [...eventIds, tag[1]] + : eventIds, + [] as string[] + ) - if (eTags.length) { + if (ids.length) { await this.eventRepository.deleteByPubkeyAndIds( event.pubkey, - eTags.map((tag) => tag[1]) + ids ) } - this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event) + if (count) { + this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event) + } } } diff --git a/src/handlers/event-strategies/parameterized-replaceable-event-strategy.ts b/src/handlers/event-strategies/parameterized-replaceable-event-strategy.ts index 09099d212..60204c8c3 100644 --- a/src/handlers/event-strategies/parameterized-replaceable-event-strategy.ts +++ b/src/handlers/event-strategies/parameterized-replaceable-event-strategy.ts @@ -1,5 +1,6 @@ import { Event, ParameterizedReplaceableEvent } from '../../@types/event' import { EventDeduplicationMetadataKey, EventTags } from '../../constants/base' +import { createCommandResult } from '../../utils/messages' import { createLogger } from '../../factories/logger-factory' import { IEventRepository } from '../../@types/repositories' import { IEventStrategy } from '../../@types/message-handlers' @@ -8,7 +9,8 @@ import { WebSocketAdapterEvent } from '../../constants/adapter' const debug = createLogger('parameterized-replaceable-event-strategy') -export class ParameterizedReplaceableEventStrategy implements IEventStrategy> { +export class ParameterizedReplaceableEventStrategy + implements IEventStrategy> { public constructor( private readonly webSocket: IWebSocketAdapter, private readonly eventRepository: IEventRepository, @@ -25,10 +27,10 @@ export class ParameterizedReplaceableEventStrategy implements IEventStrategy { debug('received event: %o', event) const count = await this.eventRepository.upsert(event) - if (!count) { - return + this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, true, (count) ? '' : 'duplicate:')) + if (count) { + this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event) } - - this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event) } } diff --git a/src/handlers/subscribe-message-handler.ts b/src/handlers/subscribe-message-handler.ts index c074ad923..3262906e4 100644 --- a/src/handlers/subscribe-message-handler.ts +++ b/src/handlers/subscribe-message-handler.ts @@ -33,7 +33,7 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable { public async handleMessage(message: SubscribeMessage): Promise { debug('received message: %o', message) - const subscriptionId = message[1] as SubscriptionId + const subscriptionId = message[1] const filters = uniqWith(equals, message.slice(2)) as SubscriptionFilter[] const reason = this.canSubscribe(subscriptionId, filters) diff --git a/src/repositories/event-repository.ts b/src/repositories/event-repository.ts index 76410027b..e29eee455 100644 --- a/src/repositories/event-repository.ts +++ b/src/repositories/event-repository.ts @@ -73,11 +73,11 @@ export class EventRepository implements IEventRepository { evolve({ exact: (pubkeys: string[]) => tableFields.forEach((tableField) => - void bd.orWhereIn(tableField, pubkeys.map(toBuffer)) + bd.orWhereIn(tableField, pubkeys.map(toBuffer)) ), even: forEach((prefix: string) => tableFields.forEach((tableField) => - void bd.orWhereRaw( + bd.orWhereRaw( `substring("${tableField}" from 1 for ?) = ?`, [prefix.length >> 1, toBuffer(prefix)] ) @@ -85,7 +85,7 @@ export class EventRepository implements IEventRepository { ), odd: forEach((prefix: string) => tableFields.forEach((tableField) => - void bd.orWhereRaw( + bd.orWhereRaw( `substring("${tableField}" from 1 for ?) BETWEEN ? AND ?`, [ (prefix.length >> 1) + 1, diff --git a/src/utils/event.ts b/src/utils/event.ts index cad7fba7c..f3101844a 100644 --- a/src/utils/event.ts +++ b/src/utils/event.ts @@ -2,11 +2,11 @@ import * as secp256k1 from '@noble/secp256k1' import { applySpec, converge, curry, mergeLeft, nth, omit, pipe, prop, reduceBy } from 'ramda' import { CanonicalEvent, Event } from '../@types/event' +import { EventId, Pubkey } from '../@types/base' import { EventKinds, EventTags } from '../constants/base' import { fromBuffer } from './transform' import { getLeadingZeroBits } from './proof-of-work' import { isGenericTagQuery } from './filter' -import { Pubkey } from '../@types/base' import { RuneLike } from './runes/rune-like' import { SubscriptionFilter } from '../@types/subscription' @@ -143,16 +143,6 @@ export const isDelegatedEventValid = async (event: Event): Promise => { const token = await secp256k1.utils.sha256(Buffer.from(serializedDelegationTag)) - // Token generation to be decided: - // const serializedDelegationTag = [ - // delegation[0], // 'delegation' - // delegation[1], // - // event.pubkey, // - // delegation[2], // - // ] - // const token = await secp256k1.utils.sha256(Buffer.from(JSON.stringify(serializedDelegationTag))) - - // Validate delegation signature return secp256k1.schnorr.verify(delegation[3], token, delegation[1]) } @@ -185,8 +175,8 @@ export const isDeleteEvent = (event: Event): boolean => { return event.kind === EventKinds.DELETE } -export const getEventProofOfWork = (event: Event): number => { - return getLeadingZeroBits(Buffer.from(event.id, 'hex')) +export const getEventProofOfWork = (eventId: EventId): number => { + return getLeadingZeroBits(Buffer.from(eventId, 'hex')) } export const getPubkeyProofOfWork = (pubkey: Pubkey): number => { diff --git a/src/utils/messages.ts b/src/utils/messages.ts index 536028a71..6f5e8afd6 100644 --- a/src/utils/messages.ts +++ b/src/utils/messages.ts @@ -5,6 +5,7 @@ import { OutgoingMessage, } from '../@types/messages' import { Event } from '../@types/event' +import { EventId } from '../@types/base' import { SubscriptionId } from '../@types/subscription' export const createNoticeMessage = (notice: string): NoticeMessage => { @@ -18,8 +19,14 @@ export const createOutgoingEventMessage = ( return [MessageType.EVENT, subscriptionId, event] } +// NIP-15 export const createEndOfStoredEventsNoticeMessage = ( subscriptionId: SubscriptionId, ): EndOfStoredEventsNotice => { return [MessageType.EOSE, subscriptionId] } + +// NIP-20 +export const createCommandResult = (eventId: EventId, successful: boolean, message: string) => { + return [MessageType.OK, eventId, successful, message] +} diff --git a/src/utils/settings.ts b/src/utils/settings.ts index 4086482e3..9d609dac5 100644 --- a/src/utils/settings.ts +++ b/src/utils/settings.ts @@ -4,10 +4,12 @@ import { join } from 'path' import { mergeDeepRight } from 'ramda' import { createLogger } from '../factories/logger-factory' +import { EventKinds } from '../constants/base' import { ISettings } from '../@types/settings' import packageJson from '../../package.json' const debug = createLogger('settings') + export class SettingsStatic { static _settings: ISettings @@ -48,6 +50,32 @@ export class SettingsStatic { maxPositiveDelta: 900, maxNegativeDelta: 0, // disabled }, + rateLimits: [ + { + kinds: [ + [EventKinds.SET_METADATA, EventKinds.DELETE], + EventKinds.REACTION, + [EventKinds.CHANNEL_CREATION, EventKinds.CHANNEL_RESERVED_LAST], + [EventKinds.REPLACEABLE_FIRST, EventKinds.REPLACEABLE_LAST], + [EventKinds.PARAMETERIZED_REPLACEABLE_FIRST, EventKinds.PARAMETERIZED_REPLACEABLE_LAST], + ], + period: 60000, + rate: 60, + }, + { + kinds: [[EventKinds.EPHEMERAL_FIRST, EventKinds.EPHEMERAL_LAST]], + period: 60000, + rate: 600, + }, + { + period: 3600000, + rate: 3600, + }, + { + period: 86400000, + rate: 86400, + }, + ], }, client: { subscription: { @@ -55,6 +83,26 @@ export class SettingsStatic { maxFilters: 10, }, }, + message: { + rateLimits: [ + { + period: 60000, + rate: 600, + }, + { + period: 3600000, + rate: 3600, + }, + { + period: 86400000, + rate: 86400, + }, + ], + ipWhitelist: [ + '::1', // local host + '::ffff:10.10.10.1', // host running docker + ], + }, }, } } diff --git a/src/utils/sliding-window-rate-limiter.ts b/src/utils/sliding-window-rate-limiter.ts new file mode 100644 index 000000000..fdc3fdf6b --- /dev/null +++ b/src/utils/sliding-window-rate-limiter.ts @@ -0,0 +1,35 @@ +import { IRateLimiter, IRateLimiterOptions } from '../@types/utils' +import { createLogger } from '../factories/logger-factory' +import { ICacheAdapter } from '../@types/adapters' + +const debug = createLogger('sliding-window-rate-limiter') + +export class SlidingWindowRateLimiter implements IRateLimiter { + public constructor( + private readonly cache: ICacheAdapter, + ) {} + + public async hit( + key: string, + step: number, + options: IRateLimiterOptions, + ): Promise { + const timestamp = Date.now() + const { period } = options + + debug('add %d hits on %s bucket', step, key) + + const [,, entries] = await Promise.all([ + this.cache.removeRangeByScoreFromSortedSet(key, 0, timestamp - period), + this.cache.addToSortedSet(key, { [`${timestamp}:${step}`]: timestamp.toString() }), + this.cache.getRangeFromSortedSet(key, 0, -1), + this.cache.setKeyExpiry(key, period), + ]) + + const hits = entries.reduce((acc, timestampAndStep) => acc + Number(timestampAndStep.split(':')[1]), 0) + + debug('hit count on %s bucket: %d', key, hits) + + return hits > options.rate + } +} \ No newline at end of file diff --git a/test/integration/docker-compose.yml b/test/integration/docker-compose.yml index 811363cb3..c85ad6953 100644 --- a/test/integration/docker-compose.yml +++ b/test/integration/docker-compose.yml @@ -11,6 +11,10 @@ services: DB_PASSWORD: postgres DB_MIN_POOL_SIZE: 1 DB_MAX_POOL_SIZE: 2 + REDIS_HOST: cache-test + REDIS_PORT: 6379 + REDIS_USER: default + REDIS_PASSWORD: nostr_ts_relay_test NOSTR_CONFIG_DIR: /code volumes: - ../../src:/code/src @@ -19,6 +23,8 @@ services: - ../../.test-reports:/code/.test-reports working_dir: /code depends_on: + cache-test: + condition: service_healthy db-test: condition: service_healthy migrations-test: @@ -40,6 +46,17 @@ services: timeout: 5s start_period: 5s retries: 0 + cache-test: + image: redis:7.0.5-alpine3.16 + command: redis-server --save 20 1 --loglevel warning --requirepass nostr_ts_relay_test + networks: + - nostr-ts-relay-test + restart: always + healthcheck: + test: [ "CMD", "redis-cli", "ping", "|", "grep", "PONG" ] + interval: 1s + timeout: 5s + retries: 5 migrations-test: image: node:18-alpine3.16 environment: diff --git a/test/integration/features/helpers.ts b/test/integration/features/helpers.ts index b27b382da..7f678c0a2 100644 --- a/test/integration/features/helpers.ts +++ b/test/integration/features/helpers.ts @@ -223,3 +223,28 @@ export async function waitForNotice(ws: WebSocket): Promise { ws.on('message', onMessage) }) } + +export async function waitForCommand(ws: WebSocket): Promise { + return new Promise((resolve, reject) => { + function cleanup() { + ws.removeListener('message', onMessage) + ws.removeListener('error', onError) + } + + function onError(error: Error) { + reject(error) + cleanup() + } + ws.once('error', onError) + + function onMessage(raw: RawData) { + const message = JSON.parse(raw.toString('utf8')) + if (message[0] === MessageType.OK) { + resolve(message) + cleanup() + } + } + + ws.on('message', onMessage) + }) +} diff --git a/test/integration/features/nip-01/nip-01.feature b/test/integration/features/nip-01/nip-01.feature index 1bf307435..b58006af9 100644 --- a/test/integration/features/nip-01/nip-01.feature +++ b/test/integration/features/nip-01/nip-01.feature @@ -27,7 +27,7 @@ Feature: NIP-01 Scenario: Alice can't post a text_note event with an invalid signature Given someone called Alice When Alice sends a text_note event with invalid signature - Then Alice receives a notice with invalid signature + Then Alice receives an unsuccessful result Scenario: Alice and Bob exchange text_note events Given someone called Alice diff --git a/test/integration/features/nip-01/nip-01.feature.ts b/test/integration/features/nip-01/nip-01.feature.ts index 4c07e776e..5f71d59d8 100644 --- a/test/integration/features/nip-01/nip-01.feature.ts +++ b/test/integration/features/nip-01/nip-01.feature.ts @@ -11,6 +11,7 @@ import { createEvent, createSubscription, sendEvent, + waitForCommand, waitForEOSE, waitForEventCount, waitForNextEvent, @@ -237,3 +238,10 @@ Then(/(\w+) receives a notice with (.*)/, async function(name: string, pattern: expect(actualNotice).to.contain(pattern) }) + +Then(/(\w+) receives an? (\w+) result/, async function(name: string, successful: string) { + const ws = this.parameters.clients[name] as WebSocket + const command = await waitForCommand(ws) + + expect(command[2]).to.equal(successful === 'successful') +}) diff --git a/test/integration/features/shared.ts b/test/integration/features/shared.ts index 82a3fe0e6..6e2115e92 100644 --- a/test/integration/features/shared.ts +++ b/test/integration/features/shared.ts @@ -12,7 +12,9 @@ import WebSocket from 'ws' import { connect, createIdentity, createSubscription } from './helpers' import { AppWorker } from '../../../src/app/worker' +import { CacheClient } from '../../../src/@types/cache' import { DatabaseClient } from '../../../src/@types/base' +import { getCacheClient } from '../../../src/cache/client' import { getDbClient } from '../../../src/database/client' import { SettingsStatic } from '../../../src/utils/settings' import { workerFactory } from '../../../src/factories/worker-factory' @@ -20,9 +22,11 @@ import { workerFactory } from '../../../src/factories/worker-factory' let worker: AppWorker let dbClient: DatabaseClient +let cacheClient: CacheClient BeforeAll({ timeout: 6000 }, async function () { process.env.PORT = '18808' + cacheClient = getCacheClient() dbClient = getDbClient() await dbClient.raw('SELECT 1=1') @@ -35,7 +39,10 @@ BeforeAll({ timeout: 6000 }, async function () { AfterAll(async function() { worker.close(async () => { - await dbClient.destroy() + await Promise.all([ + cacheClient.disconnect(), + dbClient.destroy(), + ]) }) }) diff --git a/test/unit/factories/websocket-adapter-factory.spec.ts b/test/unit/factories/websocket-adapter-factory.spec.ts index c654902b7..7f0ada338 100644 --- a/test/unit/factories/websocket-adapter-factory.spec.ts +++ b/test/unit/factories/websocket-adapter-factory.spec.ts @@ -30,6 +30,9 @@ describe('webSocketAdapterFactory', () => { headers: { 'sec-websocket-key': Buffer.from('key', 'utf8').toString('base64'), }, + socket: { + remoteAddress: '::1', + }, } as any const webSocketServerAdapter: IWebSocketServerAdapter = {} as any diff --git a/test/unit/handlers/delegated-event-message-handler.spec.ts b/test/unit/handlers/delegated-event-message-handler.spec.ts index 1db7fe35c..a5b9bf4be 100644 --- a/test/unit/handlers/delegated-event-message-handler.spec.ts +++ b/test/unit/handlers/delegated-event-message-handler.spec.ts @@ -50,6 +50,7 @@ describe('DelegatedEventMessageHandler', () => { let strategyFactoryStub: Sinon.SinonStub let onMessageSpy: Sinon.SinonSpy let strategyExecuteStub: Sinon.SinonStub + let isRateLimitedStub: Sinon.SinonStub beforeEach(() => { canAcceptEventStub = sandbox.stub(DelegatedEventMessageHandler.prototype, 'canAcceptEvent' as any) @@ -62,10 +63,12 @@ describe('DelegatedEventMessageHandler', () => { webSocket = new EventEmitter() webSocket.on(WebSocketAdapterEvent.Message, onMessageSpy) message = [MessageType.EVENT, event] + isRateLimitedStub = sandbox.stub(EventMessageHandler.prototype, 'isRateLimited' as any) handler = new DelegatedEventMessageHandler( webSocket as any, strategyFactoryStub, () => ({}) as any, + () => ({ hit: async () => false }), ) }) @@ -81,7 +84,14 @@ describe('DelegatedEventMessageHandler', () => { await handler.handleMessage(message) expect(canAcceptEventStub).to.have.been.calledOnceWithExactly(event) - expect(onMessageSpy).to.have.been.calledOnceWithExactly(['NOTICE', 'Event rejected: reason']) + expect(onMessageSpy).to.have.been.calledOnceWithExactly( + [ + MessageType.OK, + 'ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff', + false, + 'reason', + ], + ) expect(strategyFactoryStub).not.to.have.been.called }) @@ -95,6 +105,18 @@ describe('DelegatedEventMessageHandler', () => { expect(strategyFactoryStub).not.to.have.been.called }) + it('rejects event if rate-limited', async () => { + isRateLimitedStub.resolves(true) + + await handler.handleMessage(message) + + expect(isRateLimitedStub).to.have.been.calledOnceWithExactly(event) + expect(onMessageSpy).to.have.been.calledOnceWithExactly( + [MessageType.OK, event.id, false, 'rate-limited: slow down'], + ) + expect(strategyFactoryStub).not.to.have.been.called + }) + it('does not call strategy if none given', async () => { isEventValidStub.returns(undefined) canAcceptEventStub.returns(undefined) @@ -177,7 +199,7 @@ describe('DelegatedEventMessageHandler', () => { parentIsEventValidStub.resolves(undefined) event.tags[0][3] = 'wrong sig' - return expect((handler as any).isEventValid(event)).to.eventually.equal('Event with id a080fd288b60ac2225ff2e2d815291bd730911e583e177302cc949a15dc2b2dc from 62903b1ff41559daf9ee98ef1ae67cc52f301bb5ce26d14baba3052f649c3f49 is invalid delegated event') + return expect((handler as any).isEventValid(event)).to.eventually.equal('invalid: delegation verification failed') }) }) }) diff --git a/test/unit/handlers/event-message-handler.spec.ts b/test/unit/handlers/event-message-handler.spec.ts index 8a0e7c3b0..6bdb691c7 100644 --- a/test/unit/handlers/event-message-handler.spec.ts +++ b/test/unit/handlers/event-message-handler.spec.ts @@ -1,6 +1,6 @@ import EventEmitter from 'events' -import Sinon, { SinonFakeTimers } from 'sinon' +import Sinon, { SinonFakeTimers, SinonStub } from 'sinon' import chai from 'chai' import chaiAsPromised from 'chai-as-promised' @@ -49,6 +49,7 @@ describe('EventMessageHandler', () => { let strategyFactoryStub: Sinon.SinonStub let onMessageSpy: Sinon.SinonSpy let strategyExecuteStub: Sinon.SinonStub + let isRateLimitedStub: Sinon.SinonStub beforeEach(() => { canAcceptEventStub = sandbox.stub(EventMessageHandler.prototype, 'canAcceptEvent' as any) @@ -61,10 +62,12 @@ describe('EventMessageHandler', () => { webSocket = new EventEmitter() webSocket.on(WebSocketAdapterEvent.Message, onMessageSpy) message = [MessageType.EVENT, event] + isRateLimitedStub = sandbox.stub(EventMessageHandler.prototype, 'isRateLimited' as any) handler = new EventMessageHandler( webSocket as any, strategyFactoryStub, () => ({}) as any, + () => ({ hit: async () => false }) ) }) @@ -80,7 +83,21 @@ describe('EventMessageHandler', () => { await handler.handleMessage(message) expect(canAcceptEventStub).to.have.been.calledOnceWithExactly(event) - expect(onMessageSpy).to.have.been.calledOnceWithExactly(['NOTICE', 'Event rejected: reason']) + expect(onMessageSpy).to.have.been.calledOnceWithExactly( + [MessageType.OK, event.id, false, 'reason'], + ) + expect(strategyFactoryStub).not.to.have.been.called + }) + + it('rejects event if rate-limited', async () => { + isRateLimitedStub.resolves(true) + + await handler.handleMessage(message) + + expect(isRateLimitedStub).to.have.been.calledOnceWithExactly(event) + expect(onMessageSpy).to.have.been.calledOnceWithExactly( + [MessageType.OK, event.id, false, 'rate-limited: slow down'], + ) expect(strategyFactoryStub).not.to.have.been.called }) @@ -128,6 +145,7 @@ describe('EventMessageHandler', () => { it('does not reject if strategy rejects', async () => { isEventValidStub.returns(undefined) canAcceptEventStub.returns(undefined) + strategyExecuteStub.rejects() return expect(handler.handleMessage(message)).to.eventually.be.fulfilled @@ -169,6 +187,7 @@ describe('EventMessageHandler', () => { {} as any, () => null, () => settings, + () => ({ hit: async () => false }) ) }) @@ -192,7 +211,7 @@ describe('EventMessageHandler', () => { expect( (handler as any).canAcceptEvent(event) - ).to.equal('created_at is more than 100 seconds in the future') + ).to.equal('rejected: created_at is more than 100 seconds in the future') }) }) @@ -211,7 +230,7 @@ describe('EventMessageHandler', () => { expect( (handler as any).canAcceptEvent(event) - ).to.equal('created_at is more than 100 seconds in the past') + ).to.equal('rejected: created_at is more than 100 seconds in the past') }) }) }) @@ -237,7 +256,7 @@ describe('EventMessageHandler', () => { event.id = '00' + 'f'.repeat(62) expect( (handler as any).canAcceptEvent(event) - ).to.equal('insufficient proof of work: event Id has less than 16 leading zero bits') + ).to.equal('pow: difficulty 8<16') }) }) }) @@ -263,7 +282,7 @@ describe('EventMessageHandler', () => { event.pubkey = '0'.repeat(2) + 'f'.repeat(62) expect( (handler as any).canAcceptEvent(event) - ).to.equal('insufficient proof of work: pubkey has less than 16 leading zero bits') + ).to.equal('pow: pubkey difficulty 8<16') }) }) @@ -296,7 +315,7 @@ describe('EventMessageHandler', () => { event.pubkey = 'aabbcc' expect( (handler as any).canAcceptEvent(event) - ).to.equal('pubkey aabbcc is not allowed') + ).to.equal('blocked: pubkey not allowed') }) it('returns reason if pubkey is blacklisted by prefix', () => { @@ -304,7 +323,7 @@ describe('EventMessageHandler', () => { event.pubkey = 'aa55ccddeeff' expect( (handler as any).canAcceptEvent(event) - ).to.equal('pubkey aa55ccddeeff is not allowed') + ).to.equal('blocked: pubkey not allowed') }) }) @@ -337,7 +356,7 @@ describe('EventMessageHandler', () => { event.pubkey = 'aabbcc' expect( (handler as any).canAcceptEvent(event) - ).to.equal('pubkey aabbcc is not allowed') + ).to.equal('blocked: pubkey not allowed') }) it('returns reason if pubkey is not whitelisted by prefix', () => { @@ -345,7 +364,7 @@ describe('EventMessageHandler', () => { event.pubkey = 'aabbccddeeff' expect( (handler as any).canAcceptEvent(event) - ).to.equal('pubkey aabbccddeeff is not allowed') + ).to.equal('blocked: pubkey not allowed') }) }) }) @@ -380,7 +399,7 @@ describe('EventMessageHandler', () => { event.kind = 4 expect( (handler as any).canAcceptEvent(event) - ).to.equal('event kind 4 is not allowed') + ).to.equal('blocked: event kind 4 not allowed') }) }) @@ -414,7 +433,7 @@ describe('EventMessageHandler', () => { event.kind = 3 expect( (handler as any).canAcceptEvent(event) - ).to.equal('event kind 3 is not allowed') + ).to.equal('blocked: event kind 3 not allowed') }) it('returns reason if kind is blacklisted and whitelisted', () => { @@ -423,7 +442,7 @@ describe('EventMessageHandler', () => { event.kind = 3 expect( (handler as any).canAcceptEvent(event) - ).to.equal('event kind 3 is not allowed') + ).to.equal('blocked: event kind 3 not allowed') }) it('returns reason if kind is not whitelisted', () => { @@ -431,7 +450,7 @@ describe('EventMessageHandler', () => { event.kind = 4 expect( (handler as any).canAcceptEvent(event) - ).to.equal('event kind 4 is not allowed') + ).to.equal('blocked: event kind 4 not allowed') }) it('returns reason if kind is not whitelisted in range', () => { @@ -439,7 +458,7 @@ describe('EventMessageHandler', () => { event.kind = 6 expect( (handler as any).canAcceptEvent(event) - ).to.equal('event kind 6 is not allowed') + ).to.equal('blocked: event kind 6 not allowed') }) }) }) @@ -464,12 +483,148 @@ describe('EventMessageHandler', () => { it('returns reason if event id is not valid', () => { event.id = 'wrong' - return expect((handler as any).isEventValid(event)).to.eventually.equal('Event with id wrong from 55b702c167c85eb1c2d5ab35d68bedd1a35b94c01147364d2395c2f66f35a503 is not valid') + return expect((handler as any).isEventValid(event)).to.eventually.equal('invalid: event id does not match') }) it('returns reason if event signature is not valid', () => { event.sig = 'wrong' - return expect((handler as any).isEventValid(event)).to.eventually.equal('Event with id e527fe8b0f64a38c6877f943a9e8841074056ba72aceb31a4c85e6d10b27095a from 55b702c167c85eb1c2d5ab35d68bedd1a35b94c01147364d2395c2f66f35a503 has invalid signature') + return expect((handler as any).isEventValid(event)).to.eventually.equal('invalid: event signature verification failed') + }) + }) + + describe('isRateLimited', () => { + let eventLimits: EventLimits + let settings: ISettings + let rateLimiterHitStub: SinonStub + + beforeEach(() => { + eventLimits = { + rateLimits: [], + } + settings = { + limits: { + event: eventLimits, + }, + } as any + rateLimiterHitStub = sandbox.stub() + handler = new EventMessageHandler( + {} as any, + () => null, + () => settings, + () => ({ hit: rateLimiterHitStub }) + ) + }) + + it('returns undefined if rate limits setting is not set', async () => { + eventLimits.rateLimits = undefined + return expect((handler as any).isRateLimited(event)).to.eventually.be.undefined + }) + + it('returns undefined if rate limits setting is empty', async () => { + eventLimits.rateLimits = [] + return expect((handler as any).isRateLimited(event)).to.eventually.be.undefined + }) + + it('calls hit with given rate limit settings', async () => { + eventLimits.rateLimits = [ + { + period: 60000, + rate: 1, + }, + { + kinds: [0], + period: 60000, + rate: 2, + }, + { + kinds: [[10, 20]], + period: 86400000, + rate: 3, + }, + ] + + await (handler as any).isRateLimited(event) + + expect(rateLimiterHitStub).to.have.been.calledThrice + expect(rateLimiterHitStub.firstCall).to.have.been.calledWithExactly( + 'ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff:events:60000', + 1, + { + period: 60000, + rate: 1, + } + ) + expect(rateLimiterHitStub.secondCall).to.have.been.calledWithExactly( + 'ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff:events:60000:[0]', + 1, + { + period: 60000, + rate: 2, + } + ) + expect(rateLimiterHitStub.thirdCall).to.have.been.calledWithExactly( + 'ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff:events:86400000:[[10,20]]', + 1, + { + period: 86400000, + rate: 3, + } + ) + }) + + + it('fulfills with false if not rate limited', async () => { + eventLimits.rateLimits = [ + { + period: 60000, + rate: 1, + }, + { + kinds: [0], + period: 60000, + rate: 2, + }, + { + kinds: [[10, 20]], + period: 86400000, + rate: 3, + }, + ] + + rateLimiterHitStub.resolves(false) + + const actualResult = await (handler as any).isRateLimited(event) + + expect(rateLimiterHitStub).to.have.been.calledThrice + expect(actualResult).to.be.false + }) + + it('fulfills with true if rate limited by second rate limit setting', async () => { + eventLimits.rateLimits = [ + { + period: 60000, + rate: 1, + }, + { + kinds: [0], + period: 60000, + rate: 2, + }, + { + kinds: [[10, 20]], + period: 86400000, + rate: 3, + }, + ] + + rateLimiterHitStub.onFirstCall().resolves(false) + rateLimiterHitStub.onSecondCall().resolves(true) + rateLimiterHitStub.onThirdCall().resolves(false) + + const actualResult = await (handler as any).isRateLimited(event) + + expect(rateLimiterHitStub).to.have.been.calledThrice + expect(actualResult).to.be.true }) }) }) diff --git a/test/unit/handlers/event-strategies/default-event-strategy.spec.ts b/test/unit/handlers/event-strategies/default-event-strategy.spec.ts index c23287030..cd88258e1 100644 --- a/test/unit/handlers/event-strategies/default-event-strategy.spec.ts +++ b/test/unit/handlers/event-strategies/default-event-strategy.spec.ts @@ -11,12 +11,15 @@ import { EventRepository } from '../../../../src/repositories/event-repository' import { IEventRepository } from '../../../../src/@types/repositories' import { IEventStrategy } from '../../../../src/@types/message-handlers' import { IWebSocketAdapter } from '../../../../src/@types/adapters' +import { MessageType } from '../../../../src/@types/messages' import { WebSocketAdapterEvent } from '../../../../src/constants/adapter' const { expect } = chai describe('DefaultEventStrategy', () => { - const event: Event = {} as any + const event: Event = { + id: 'id', + } as any let webSocket: IWebSocketAdapter let eventRepository: IEventRepository @@ -59,12 +62,29 @@ describe('DefaultEventStrategy', () => { await strategy.execute(event) expect(eventRepositoryCreateStub).to.have.been.calledOnceWithExactly(event) - expect(webSocketEmitStub).to.have.been.calledOnceWithExactly( + expect(webSocketEmitStub).to.have.been.calledTwice + expect(webSocketEmitStub).to.have.been.calledWithExactly( + WebSocketAdapterEvent.Message, + [MessageType.OK, 'id', true, ''] + ) + expect(webSocketEmitStub).to.have.been.calledWithExactly( WebSocketAdapterEvent.Broadcast, event ) }) + it('does not broadcast event if event is duplicate', async () => { + eventRepositoryCreateStub.resolves(0) + + await strategy.execute(event) + + expect(eventRepositoryCreateStub).to.have.been.calledOnceWithExactly(event) + expect(webSocketEmitStub).to.have.been.calledOnceWithExactly( + WebSocketAdapterEvent.Message, + ['OK', 'id', true, 'duplicate:'] + ) + }) + it('rejects if unable to create event', async () => { const error = new Error() eventRepositoryCreateStub.rejects(error) diff --git a/test/unit/handlers/event-strategies/delete-event-strategy.spec.ts b/test/unit/handlers/event-strategies/delete-event-strategy.spec.ts index 4cb313389..636f68703 100644 --- a/test/unit/handlers/event-strategies/delete-event-strategy.spec.ts +++ b/test/unit/handlers/event-strategies/delete-event-strategy.spec.ts @@ -12,12 +12,14 @@ import { EventTags } from '../../../../src/constants/base' import { IEventRepository } from '../../../../src/@types/repositories' import { IEventStrategy } from '../../../../src/@types/message-handlers' import { IWebSocketAdapter } from '../../../../src/@types/adapters' +import { MessageType } from '../../../../src/@types/messages' import { WebSocketAdapterEvent } from '../../../../src/constants/adapter' const { expect } = chai describe('DeleteEventStrategy', () => { const event: Event = { + id: 'id', pubkey: 'pubkey', tags: [ [EventTags.Event, 'event id 1'], @@ -79,18 +81,34 @@ describe('DeleteEventStrategy', () => { expect(eventRepositoryDeleteByPubkeyAndIdsStub).not.to.have.been.called }) - it('broadcast event', async () => { - eventRepositoryCreateStub.resolves() + it('broadcast event if created', async () => { + eventRepositoryCreateStub.resolves(1) await strategy.execute(event) expect(eventRepositoryCreateStub).to.have.been.calledOnceWithExactly(event) - expect(webSocketEmitStub).to.have.been.calledOnceWithExactly( + expect(webSocketEmitStub).to.have.been.calledTwice + expect(webSocketEmitStub).to.have.been.calledWithExactly( + WebSocketAdapterEvent.Message, + [MessageType.OK, 'id', true, ''] + ) + expect(webSocketEmitStub).to.have.been.calledWithExactly( WebSocketAdapterEvent.Broadcast, event ) }) + it('does not broadcast event if duplicate', async () => { + eventRepositoryCreateStub.resolves(0) + + await strategy.execute(event) + + expect(webSocketEmitStub).to.have.been.calledOnceWithExactly( + WebSocketAdapterEvent.Message, + [MessageType.OK, 'id', true, 'duplicate:'] + ) + }) + it('rejects if unable to create event', async () => { const error = new Error() eventRepositoryCreateStub.rejects(error) diff --git a/test/unit/handlers/event-strategies/parameterized-replaceable-event-strategy.spec.ts b/test/unit/handlers/event-strategies/parameterized-replaceable-event-strategy.spec.ts index 040557436..5a7335143 100644 --- a/test/unit/handlers/event-strategies/parameterized-replaceable-event-strategy.spec.ts +++ b/test/unit/handlers/event-strategies/parameterized-replaceable-event-strategy.spec.ts @@ -11,6 +11,7 @@ import { EventRepository } from '../../../../src/repositories/event-repository' import { IEventRepository } from '../../../../src/@types/repositories' import { IEventStrategy } from '../../../../src/@types/message-handlers' import { IWebSocketAdapter } from '../../../../src/@types/adapters' +import { MessageType } from '../../../../src/@types/messages' import { ParameterizedReplaceableEventStrategy } from '../../../../src/handlers/event-strategies/parameterized-replaceable-event-strategy' import { WebSocketAdapterEvent } from '../../../../src/constants/adapter' @@ -18,6 +19,7 @@ const { expect } = chai describe('ParameterizedReplaceableEventStrategy', () => { const event: Event = { + id: 'id', tags: [ [EventTags.Deduplication, 'dedup'], ], @@ -82,12 +84,28 @@ describe('ParameterizedReplaceableEventStrategy', () => { await strategy.execute(event) expect(eventRepositoryUpsertStub).to.have.been.calledOnceWithExactly(event) - expect(webSocketEmitStub).to.have.been.calledOnceWithExactly( + expect(webSocketEmitStub).to.have.been.calledTwice + expect(webSocketEmitStub).to.have.been.calledWithExactly( + WebSocketAdapterEvent.Message, + [MessageType.OK, 'id', true, ''] + ) + expect(webSocketEmitStub).to.have.been.calledWithExactly( WebSocketAdapterEvent.Broadcast, event ) }) + it('does not broadcast event if event is duplicate', async () => { + eventRepositoryUpsertStub.resolves(0) + + await strategy.execute(event) + + expect(webSocketEmitStub).to.have.been.calledOnceWithExactly( + WebSocketAdapterEvent.Message, + [MessageType.OK, 'id', true, 'duplicate:'] + ) + }) + it('rejects if unable to upsert event', async () => { const error = new Error() eventRepositoryUpsertStub.rejects(error) diff --git a/test/unit/handlers/event-strategies/replaceable-event-strategy.spec.ts b/test/unit/handlers/event-strategies/replaceable-event-strategy.spec.ts index af1ef27cc..c3731975f 100644 --- a/test/unit/handlers/event-strategies/replaceable-event-strategy.spec.ts +++ b/test/unit/handlers/event-strategies/replaceable-event-strategy.spec.ts @@ -10,13 +10,16 @@ import { EventRepository } from '../../../../src/repositories/event-repository' import { IEventRepository } from '../../../../src/@types/repositories' import { IEventStrategy } from '../../../../src/@types/message-handlers' import { IWebSocketAdapter } from '../../../../src/@types/adapters' +import { MessageType } from '../../../../src/@types/messages' import { ReplaceableEventStrategy } from '../../../../src/handlers/event-strategies/replaceable-event-strategy' import { WebSocketAdapterEvent } from '../../../../src/constants/adapter' const { expect } = chai describe('ReplaceableEventStrategy', () => { - const event: Event = {} as any + const event: Event = { + id: 'id', + } as any let webSocket: IWebSocketAdapter let eventRepository: IEventRepository @@ -59,12 +62,29 @@ describe('ReplaceableEventStrategy', () => { await strategy.execute(event) expect(eventRepositoryUpsertStub).to.have.been.calledOnceWithExactly(event) - expect(webSocketEmitStub).to.have.been.calledOnceWithExactly( + expect(webSocketEmitStub).to.have.been.calledTwice + expect(webSocketEmitStub).to.have.been.calledWithExactly( + WebSocketAdapterEvent.Message, + [MessageType.OK, 'id', true, ''] + ) + expect(webSocketEmitStub).to.have.been.calledWithExactly( WebSocketAdapterEvent.Broadcast, event ) }) + it('does not broadcast event if event is duplicate', async () => { + eventRepositoryUpsertStub.resolves(0) + + await strategy.execute(event) + + expect(eventRepositoryUpsertStub).to.have.been.calledOnceWithExactly(event) + expect(webSocketEmitStub).to.have.been.calledOnceWithExactly( + WebSocketAdapterEvent.Message, + ['OK', 'id', true, 'duplicate:'] + ) + }) + it('rejects if unable to upsert event', async () => { const error = new Error() eventRepositoryUpsertStub.rejects(error) diff --git a/test/unit/utils/settings.spec.ts b/test/unit/utils/settings.spec.ts index bd7b8aa25..481c16700 100644 --- a/test/unit/utils/settings.spec.ts +++ b/test/unit/utils/settings.spec.ts @@ -72,6 +72,26 @@ describe('SettingsStatic', () => { maxPositiveDelta: 900, // +15 min maxNegativeDelta: 0, // disabled }, + 'rateLimits': [ + { + 'kinds': [[0, 5], 7, [40, 49], [10000, 19999], [30000, 39999]], + 'period': 60000, + 'rate': 60, + }, + { + 'kinds': [[20000, 29999]], + 'period': 60000, + 'rate': 600, + }, + { + 'period': 3600000, + 'rate': 3600, + }, + { + 'period': 86400000, + 'rate': 86400, + }, + ], }, client: { subscription: { @@ -79,6 +99,26 @@ describe('SettingsStatic', () => { maxFilters: 10, }, }, + message: { + 'rateLimits': [ + { + 'period': 60000, + 'rate': 600, + }, + { + 'period': 3600000, + 'rate': 3600, + }, + { + 'period': 86400000, + 'rate': 86400, + }, + ], + ipWhitelist: [ + '::1', + '::ffff:10.10.10.1', + ], + }, }) }) }) diff --git a/test/unit/utils/sliding-window-rate-limiter.spec.ts b/test/unit/utils/sliding-window-rate-limiter.spec.ts new file mode 100644 index 000000000..bee006862 --- /dev/null +++ b/test/unit/utils/sliding-window-rate-limiter.spec.ts @@ -0,0 +1,64 @@ +import { expect } from 'chai' +import Sinon from 'sinon' + +import { ICacheAdapter } from '../../../src/@types/adapters' +import { IRateLimiter } from '../../../src/@types/utils' +import { SlidingWindowRateLimiter } from '../../../src/utils/sliding-window-rate-limiter' + +describe('SlidingWindowRateLimiter', () => { + let clock: Sinon.SinonFakeTimers + let cache: ICacheAdapter + let rateLimiter: IRateLimiter + + let removeRangeByScoreFromSortedSetStub: Sinon.SinonStub + let addToSortedSetStub: Sinon.SinonStub + let getRangeFromSortedSetStub: Sinon.SinonStub + let setKeyExpiryStub: Sinon.SinonStub + + let sandbox: Sinon.SinonSandbox + + beforeEach(() => { + sandbox = Sinon.createSandbox() + clock = sandbox.useFakeTimers(1665546189000) + removeRangeByScoreFromSortedSetStub = sandbox.stub() + addToSortedSetStub = sandbox.stub() + getRangeFromSortedSetStub = sandbox.stub() + setKeyExpiryStub = sandbox.stub() + cache = { + removeRangeByScoreFromSortedSet: removeRangeByScoreFromSortedSetStub, + addToSortedSet: addToSortedSetStub, + getRangeFromSortedSet: getRangeFromSortedSetStub, + setKeyExpiry: setKeyExpiryStub, + } + rateLimiter = new SlidingWindowRateLimiter(cache) + }) + + afterEach(() => { + clock.restore() + sandbox.restore() + }) + + it('returns true if rate limited', async () => { + const now = Date.now() + getRangeFromSortedSetStub.resolves([ + `${now}:6`, + `${now}:4`, + `${now}:1`, + ]) + + const actualResult = await rateLimiter.hit('key', 1, { period: 60000, rate: 10 }) + + expect(actualResult).to.be.true + }) + + it('returns false if not rate limited',async () => { + const now = Date.now() + getRangeFromSortedSetStub.resolves([ + `${now}:10`, + ]) + + const actualResult = await rateLimiter.hit('key', 1, { period: 60000, rate: 10 }) + + expect(actualResult).to.be.false + }) +})