From ccff46c1bce3d44a6afe37cab4aa677dee5c6005 Mon Sep 17 00:00:00 2001 From: Jim Toth Date: Sun, 8 Nov 2026 14:10:01 -0500 Subject: [PATCH] Checks signers the gateway does not know as operators within a shared allowance, and limits messages per client address at the edge. --- apps/hyperbeam-gateway/README.md | 6 ++- apps/hyperbeam-gateway/src/app.test.ts | 65 ++++++++++++++++++++++- apps/hyperbeam-gateway/src/config.test.ts | 6 +++ apps/hyperbeam-gateway/src/config.ts | 4 +- apps/hyperbeam-gateway/src/main.ts | 7 +++ apps/hyperbeam-gateway/src/operators.ts | 43 +++++++++++++++ apps/hyperbeam-gateway/src/rate-limit.ts | 7 +-- apps/hyperbeam-gateway/src/writes.ts | 7 +++ operations/README.md | 2 +- operations/hyperbeam-gateway-live.hcl | 9 +++- operations/hyperbeam-gateway-stage.hcl | 9 +++- 11 files changed, 155 insertions(+), 10 deletions(-) create mode 100644 apps/hyperbeam-gateway/src/operators.ts diff --git a/apps/hyperbeam-gateway/README.md b/apps/hyperbeam-gateway/README.md index cca75f4..0eba454 100644 --- a/apps/hyperbeam-gateway/README.md +++ b/apps/hyperbeam-gateway/README.md @@ -39,7 +39,7 @@ A message is a raw ANS-104 data item sent with `content-type: application/ans104 5. The tags are exactly `type=Message`, `data-protocol=ao`, `variant=ao.N.1`, `action`, the action's own tags, and optionally `require-codec=application/json` and `ui-cache-key`. Tag names match in any case and may not repeat. The data has the shape the action takes. 6. A message whose id the gateway has already sent and the node accepted, or is sending now, is answered with that answer and not sent again. See [Message record](#message-record). 7. A signer may send `WRITES_PER_SIGNER_PER_MINUTE` messages a minute. -8. The signer is allowed to send the action, see below. +8. The signer is allowed to send the action, see below. A signer the gateway does not know as an operator is checked within an allowance of `NEW_SIGNER_CHECKS_PER_MINUTE` checks a minute shared by all such signers; past it, the message answers `503` before the node is asked. Then the gateway posts the item's original bytes to the node with only the `content-type` and `codec-device` headers. @@ -56,6 +56,8 @@ Always available. The gateway reads the signer's entry from operator-registry's An operator is an address with at least one claimable or verified fingerprint. The contract still makes its own checks, for example a blocked address or a missing registration credit. +The gateway knows as operators the addresses in operator-registry's `operators` view, read at start and every `OPERATORS_REFRESH_INTERVAL`, and any signer it has found to be an operator since the last read. Their messages do not count against the new signer allowance, so messages signed with fresh keys cannot crowd out existing operators or load the node beyond that allowance. + ### Admin actions Available only when listed in `ADMIN_ACTIONS`, as `:`, and only from a signer listed in `ADMIN_SIGNERS` who also holds the `admin` role or the role named after the action in that process's `roles` view. The `owner` is not accepted through the gateway. @@ -106,5 +108,7 @@ bun run --filter hyperbeam-gateway dev | `NODE_WRITE_TIMEOUT` | `60s` | time allowed for the node to accept a message | | `MAX_BODY_BYTES` | `1048576` | largest message accepted | | `WRITES_PER_SIGNER_PER_MINUTE` | `20` | messages a signer may send per minute | +| `NEW_SIGNER_CHECKS_PER_MINUTE` | `120` | operator checks a minute, shared by all signers the gateway does not know as operators, in bursts of up to 10 | +| `OPERATORS_REFRESH_INTERVAL` | `60s` | how often the operator list is read from operator-registry | Plus `PORT` and `LOG_LEVEL`, as for every service in this repository. diff --git a/apps/hyperbeam-gateway/src/app.test.ts b/apps/hyperbeam-gateway/src/app.test.ts index f2ffce2..02c2ec6 100644 --- a/apps/hyperbeam-gateway/src/app.test.ts +++ b/apps/hyperbeam-gateway/src/app.test.ts @@ -3,6 +3,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'bun:test'; import type { Tag } from './ans104'; import { createFetchHandler } from './app'; import { HyperbeamNode } from './node'; +import { OperatorDirectory } from './operators'; import { ProcessIds } from './processes'; import { RateLimiter } from './rate-limit'; import { ReadHandler } from './reads'; @@ -34,10 +35,20 @@ interface BuildOptions { adminActions?: string[]; adminSigners?: string[]; perMinute?: number; + newSignersPerMinute?: number; + operators?: OperatorDirectory; } -function gateway({ adminActions = [], adminSigners = [], perMinute = 100 }: BuildOptions = {}) { - const hb = new HyperbeamNode({ url: node.url, readTimeoutMs: 5_000, writeTimeoutMs: 5_000 }); +function hyperbeam() { + return new HyperbeamNode({ url: node.url, readTimeoutMs: 5_000, writeTimeoutMs: 5_000 }); +} + +function directory() { + return new OperatorDirectory({ node: hyperbeam(), processId: OPERATOR_REGISTRY, refreshMs: 60_000, logger }); +} + +function gateway({ adminActions = [], adminSigners = [], perMinute = 100, newSignersPerMinute = 100, operators = directory() }: BuildOptions = {}) { + const hb = hyperbeam(); const processes = new ProcessIds({ 'operator-registry': OPERATOR_REGISTRY, 'relay-rewards': RELAY_REWARDS, 'staking-rewards': STAKING_REWARDS }); return createFetchHandler({ maxBodyBytes: 64 * 1024, @@ -49,6 +60,8 @@ function gateway({ adminActions = [], adminSigners = [], perMinute = 100 }: Buil store, admin: { actions: new Set(adminActions), signers: new Set(adminSigners) }, limiter: new RateLimiter({ perMinute }), + operators, + newSigners: new RateLimiter({ perMinute: newSignersPerMinute, burst: newSignersPerMinute }), roles: new RolesCache(hb, 60_000), logger, }), @@ -249,6 +262,54 @@ describe('repeated messages', () => { }); }); +describe('new signers', () => { + const delegate = (address: string) => [ + { name: 'address', value: address }, + { name: 'share', value: '0.5' }, + ]; + const operatorLookups = () => node.requests.filter((request) => request.path.includes('/as/operator?')); + + it('checks signers it does not know within one shared allowance', async () => { + const handler = gateway({ newSignersPerMinute: 1 }); + expect(await refusal(await push(handler, RELAY_REWARDS, message(stranger, RELAY_REWARDS, 'Set-Delegate')))).toEqual([403, 'signer is not a registered operator']); + expect(await refusal(await push(handler, RELAY_REWARDS, message(randomSigner(), RELAY_REWARDS, 'Set-Delegate')))).toEqual([503, 'too many messages from new signers, retry shortly']); + expect(operatorLookups()).toHaveLength(1); + expect(node.pushes()).toHaveLength(0); + }); + + it('does not count operators on the list it reads from operator-registry', async () => { + node.views.set(`${OPERATOR_REGISTRY}/as/operators`, { status: 200, body: JSON.stringify({ [operator.address]: true }) }); + const operators = directory(); + await operators.refresh(); + const handler = gateway({ newSignersPerMinute: 1, operators }); + await push(handler, RELAY_REWARDS, message(stranger, RELAY_REWARDS, 'Set-Delegate')); + expect((await push(handler, RELAY_REWARDS, message(operator, RELAY_REWARDS, 'Set-Delegate'))).status).toBe(200); + }); + + it('remembers an operator after its first check', async () => { + const handler = gateway({ newSignersPerMinute: 1 }); + expect((await push(handler, RELAY_REWARDS, message(operator, RELAY_REWARDS, 'Set-Delegate'))).status).toBe(200); + expect((await push(handler, RELAY_REWARDS, message(operator, RELAY_REWARDS, 'Set-Delegate', delegate(stranger.address)))).status).toBe(200); + }); + + it('keeps the last operator list when a refresh fails', async () => { + node.views.set(`${OPERATOR_REGISTRY}/as/operators`, { status: 200, body: JSON.stringify({ [operator.address]: true }) }); + const operators = directory(); + await operators.refresh(); + node.views.set(`${OPERATOR_REGISTRY}/as/operators`, { status: 500, body: '{}' }); + await operators.refresh(); + expect(operators.has(operator.address)).toBe(true); + }); + + it('reads an empty operator list', async () => { + node.views.set(`${OPERATOR_REGISTRY}/as/operators`, { status: 200, body: '[]' }); + const operators = directory(); + operators.add(operator.address); + await operators.refresh(); + expect(operators.has(operator.address)).toBe(false); + }); +}); + describe('admin writes', () => { const team = randomSigner(); const block = () => message(team, OPERATOR_REGISTRY, 'Block-Operator-Address', [{ name: 'address', value: stranger.address }]); diff --git a/apps/hyperbeam-gateway/src/config.test.ts b/apps/hyperbeam-gateway/src/config.test.ts index 96859bd..787a129 100644 --- a/apps/hyperbeam-gateway/src/config.test.ts +++ b/apps/hyperbeam-gateway/src/config.test.ts @@ -15,6 +15,8 @@ describe('loadConfig', () => { expect(config.adminActions.size).toBe(0); expect(config.VIEW_CACHE_TTL).toBe(30_000); expect(config.NODE_WRITE_TIMEOUT).toBe(60_000); + expect(config.NEW_SIGNER_CHECKS_PER_MINUTE).toBe(120); + expect(config.OPERATORS_REFRESH_INTERVAL).toBe(60_000); }); it('reads admin actions and signers', () => { @@ -33,6 +35,10 @@ describe('loadConfig', () => { expect(() => loadConfig({ ...BASE, ADMIN_ACTIONS: 'relay-rewards:Set-Delegate', ADMIN_SIGNERS: TEAM })).toThrow('is not a known admin action'); }); + it('refuses limits that are not positive', () => { + expect(() => loadConfig({ ...BASE, NEW_SIGNER_CHECKS_PER_MINUTE: '0' })).toThrow('NEW_SIGNER_CHECKS_PER_MINUTE must be greater than 0'); + }); + it('refuses admin signers that are not checksummed and admin actions without signers', () => { expect(() => loadConfig({ ...BASE, ADMIN_SIGNERS: TEAM.toLowerCase() })).toThrow('must be a checksummed address'); expect(() => loadConfig({ ...BASE, ADMIN_ACTIONS: 'relay-rewards:Cancel-Round' })).toThrow('ADMIN_ACTIONS needs at least one ADMIN_SIGNERS entry'); diff --git a/apps/hyperbeam-gateway/src/config.ts b/apps/hyperbeam-gateway/src/config.ts index 0768336..96458bd 100644 --- a/apps/hyperbeam-gateway/src/config.ts +++ b/apps/hyperbeam-gateway/src/config.ts @@ -18,6 +18,8 @@ export const gatewayEnv = { NODE_WRITE_TIMEOUT: env.durationMs({ default: 60_000 }), MAX_BODY_BYTES: env.integer({ default: 1_048_576 }), WRITES_PER_SIGNER_PER_MINUTE: env.integer({ default: 20 }), + NEW_SIGNER_CHECKS_PER_MINUTE: env.integer({ default: 120 }), + OPERATORS_REFRESH_INTERVAL: env.durationMs({ default: 60_000 }), }; export type GatewayConfig = ReturnType; @@ -40,7 +42,7 @@ export function loadConfig(source: Environment = process.env) { ...adminActions.map((key) => !ADMIN_ACTION_KEYS.has(key) && `ADMIN_ACTIONS entry ${key} is not a known admin action, expected one of ${[...ADMIN_ACTION_KEYS].join(', ')}`), ...adminSigners.map((address) => !isChecksumAddress(address) && `ADMIN_SIGNERS entry ${address} must be a checksummed address`), adminActions.length > 0 && adminSigners.length === 0 && 'ADMIN_ACTIONS needs at least one ADMIN_SIGNERS entry', - ...(['MAX_BODY_BYTES', 'WRITES_PER_SIGNER_PER_MINUTE', 'VIEW_CACHE_TTL', 'ROLES_CACHE_TTL', 'NODE_READ_TIMEOUT', 'NODE_WRITE_TIMEOUT'] as const).map( + ...(['MAX_BODY_BYTES', 'WRITES_PER_SIGNER_PER_MINUTE', 'NEW_SIGNER_CHECKS_PER_MINUTE', 'OPERATORS_REFRESH_INTERVAL', 'VIEW_CACHE_TTL', 'ROLES_CACHE_TTL', 'NODE_READ_TIMEOUT', 'NODE_WRITE_TIMEOUT'] as const).map( (name) => config[name] <= 0 && `${name} must be greater than 0`, ), ].filter(Boolean); diff --git a/apps/hyperbeam-gateway/src/main.ts b/apps/hyperbeam-gateway/src/main.ts index e347de1..ca22c53 100644 --- a/apps/hyperbeam-gateway/src/main.ts +++ b/apps/hyperbeam-gateway/src/main.ts @@ -4,6 +4,7 @@ import { createLogger, handleShutdownSignals, handleUncaughtErrors } from '@anyo import { startServer } from './app'; import { loadConfig } from './config'; import { HyperbeamNode } from './node'; +import { OperatorDirectory } from './operators'; import { ProcessIds } from './processes'; import { RateLimiter } from './rate-limit'; import { ReadHandler } from './reads'; @@ -23,6 +24,9 @@ const processes = new ProcessIds({ 'staking-rewards': config.STAKING_REWARDS_PROCESS_ID, }); +const operators = new OperatorDirectory({ node, processId: config.OPERATOR_REGISTRY_PROCESS_ID, refreshMs: config.OPERATORS_REFRESH_INTERVAL, logger }); +operators.start(); + const server = startServer({ port: config.PORT, maxBodyBytes: config.MAX_BODY_BYTES, @@ -34,6 +38,8 @@ const server = startServer({ store, admin: { actions: config.adminActions, signers: config.adminSigners }, limiter: new RateLimiter({ perMinute: config.WRITES_PER_SIGNER_PER_MINUTE }), + operators, + newSigners: new RateLimiter({ perMinute: config.NEW_SIGNER_CHECKS_PER_MINUTE, burst: 10 }), roles: new RolesCache(node, config.ROLES_CACHE_TTL), logger, }), @@ -52,6 +58,7 @@ logger.info('Listening', { handleShutdownSignals( [ ['http server', () => server.stop()], + ['operator list', () => operators.stop()], ['message store', () => store.close()], ], { logger }, diff --git a/apps/hyperbeam-gateway/src/operators.ts b/apps/hyperbeam-gateway/src/operators.ts new file mode 100644 index 0000000..b00fa7a --- /dev/null +++ b/apps/hyperbeam-gateway/src/operators.ts @@ -0,0 +1,43 @@ +import type { Logger } from '@anyone-protocol/service-kit'; +import type { HyperbeamNode } from './node'; + +export interface OperatorDirectoryOptions { + node: HyperbeamNode; + processId: string; + refreshMs: number; + logger: Logger; +} + +export class OperatorDirectory { + private known = new Set(); + private timer: ReturnType | undefined; + + constructor(private readonly options: OperatorDirectoryOptions) {} + + has(address: string): boolean { + return this.known.has(address); + } + + add(address: string): void { + this.known.add(address); + } + + async refresh(): Promise { + try { + const operators = await this.options.node.view(this.options.processId, 'operators'); + this.known = new Set(operators && typeof operators === 'object' && !Array.isArray(operators) ? Object.keys(operators) : []); + } catch (error) { + this.options.logger.warn('operator list refresh failed, keeping the previous list', { error: error instanceof Error ? error.message : String(error), operators: this.known.size }); + } + } + + start(): void { + void this.refresh(); + this.timer = setInterval(() => void this.refresh(), this.options.refreshMs); + this.timer.unref?.(); + } + + stop(): void { + if (this.timer) clearInterval(this.timer); + } +} diff --git a/apps/hyperbeam-gateway/src/rate-limit.ts b/apps/hyperbeam-gateway/src/rate-limit.ts index 8c973f0..9ae9a28 100644 --- a/apps/hyperbeam-gateway/src/rate-limit.ts +++ b/apps/hyperbeam-gateway/src/rate-limit.ts @@ -5,6 +5,7 @@ interface Bucket { export interface RateLimiterOptions { perMinute: number; + burst?: number; now?: () => number; maxKeys?: number; } @@ -23,9 +24,9 @@ export class RateLimiter { take(key: string): boolean { const now = this.now(); - const capacity = this.options.perMinute; - const bucket = this.buckets.get(key) ?? { tokens: capacity, updatedAt: now }; - bucket.tokens = Math.min(capacity, bucket.tokens + ((now - bucket.updatedAt) / MINUTE_MS) * capacity); + const { perMinute, burst = perMinute } = this.options; + const bucket = this.buckets.get(key) ?? { tokens: burst, updatedAt: now }; + bucket.tokens = Math.min(burst, bucket.tokens + ((now - bucket.updatedAt) / MINUTE_MS) * perMinute); bucket.updatedAt = now; const allowed = bucket.tokens >= 1; diff --git a/apps/hyperbeam-gateway/src/writes.ts b/apps/hyperbeam-gateway/src/writes.ts index bc65b30..2bc5cfe 100644 --- a/apps/hyperbeam-gateway/src/writes.ts +++ b/apps/hyperbeam-gateway/src/writes.ts @@ -2,6 +2,7 @@ import type { Logger } from '@anyone-protocol/service-kit'; import { type ActionSpec, actionKey, findAction, InvalidMessage, messageTags, type OperatorEntry, type ProcessName, validateMessage } from './actions'; import { type DataItem, InvalidDataItem, parseDataItem, verifySigner } from './ans104'; import { type HyperbeamNode, NodeUnavailable, type NodeResponse } from './node'; +import type { OperatorDirectory } from './operators'; import type { ProcessIds } from './processes'; import type { RateLimiter } from './rate-limit'; import type { AcceptedMessage, MessageStore } from './store'; @@ -17,6 +18,8 @@ export interface WriteOptions { store: MessageStore; admin: AdminPolicy; limiter: RateLimiter; + operators: OperatorDirectory; + newSigners: RateLimiter; roles: RolesCache; logger: Logger; now?: () => number; @@ -26,6 +29,8 @@ export type WriteResult = { kind: 'refused'; status: number; reason: string } | type RoleTable = Record>; +const NEW_SIGNERS = 'new-signers'; + export class RolesCache { private readonly entries = new Map }>(); @@ -135,8 +140,10 @@ export class WriteHandler { private async authorize(spec: ActionSpec, signer: string, tags: ReadonlyMap): Promise { if (spec.audience === 'operator') { + if (!this.options.operators.has(signer) && !this.options.newSigners.take(NEW_SIGNERS)) throw new Refusal(503, 'too many messages from new signers, retry shortly'); const entry = await this.options.node.view | null>(this.options.processes.id('operator-registry'), 'operator', { address: signer }); const operator: OperatorEntry = { claimable: setKeys(entry?.claimable), verified: setKeys(entry?.verified) }; + if (operator.claimable.length > 0 || operator.verified.length > 0) this.options.operators.add(signer); const reason = spec.check(operator, tags); if (reason) throw new Refusal(403, reason); return; diff --git a/operations/README.md b/operations/README.md index 6e2332c..78919a0 100644 --- a/operations/README.md +++ b/operations/README.md @@ -145,7 +145,7 @@ nomad job run -var commit_sha= operations/api-service-stage.hcl What the job sets up, per environment: - One allocation on the environment's services pool. A deploy stops the running allocation before starting the new one, so only one instance writes the message record at a time, and reverts if the new one does not become healthy. -- The Consul service `hyperbeam-gateway-` on a dynamic port on the `wireguard` host network, mapped to the service's port 3000, with the Traefik tags that publish it as `hb-stage.anyone.tech` on stage and `hb.anyone.tech` on live. Traefik limits each client address to 30 requests a second on average with bursts of 60, and refuses bodies over 1 MB, the service's own `MAX_BODY_BYTES`. +- The Consul service `hyperbeam-gateway-` on a dynamic port on the `wireguard` host network, mapped to the service's port 3000, with the Traefik tags that publish it as `hb-stage.anyone.tech` on stage and `hb.anyone.tech` on live. Traefik limits each client address to 30 requests a second on average with bursts of 60, and to 2 messages (POST requests) a second with bursts of 20. It refuses bodies over 1 MB, the service's own `MAX_BODY_BYTES`. - An HTTP check on `/healthz` sent to the allocation's own address. The service reports healthy while it runs, whether or not the node answers. - The host volume `hyperbeam-gateway-` at `/data`, holding the [message record](../apps/hyperbeam-gateway/#message-record) at `/data/messages.sqlite` between restarts and deploys. The task runs as the image's `bun` user, uid 1000. - The node's address from the Consul service `hyperbeam--node`, and the three process ids from the Consul keys `smart-contracts//operator-registry-address`, `relay-rewards-address` and `staking-rewards-address`. A change to either re-renders the environment and restarts the task. diff --git a/operations/hyperbeam-gateway-live.hcl b/operations/hyperbeam-gateway-live.hcl index a7e38bb..64ce4e9 100644 --- a/operations/hyperbeam-gateway-live.hcl +++ b/operations/hyperbeam-gateway-live.hcl @@ -50,7 +50,14 @@ job "hyperbeam-gateway-live" { "traefik-ec.http.routers.hyperbeam-gateway-live.middlewares=hyperbeam-gateway-live-ratelimit,hyperbeam-gateway-live-buffering", "traefik-ec.http.middlewares.hyperbeam-gateway-live-ratelimit.ratelimit.average=30", "traefik-ec.http.middlewares.hyperbeam-gateway-live-ratelimit.ratelimit.burst=60", - "traefik-ec.http.middlewares.hyperbeam-gateway-live-buffering.buffering.maxrequestbodybytes=1048576" + "traefik-ec.http.middlewares.hyperbeam-gateway-live-buffering.buffering.maxrequestbodybytes=1048576", + "traefik-ec.http.routers.hyperbeam-gateway-live-push.rule=Host(`hb.anyone.tech`) && Method(`POST`)", + "traefik-ec.http.routers.hyperbeam-gateway-live-push.entrypoints=https", + "traefik-ec.http.routers.hyperbeam-gateway-live-push.tls=true", + "traefik-ec.http.routers.hyperbeam-gateway-live-push.tls.certresolver=anyoneresolver", + "traefik-ec.http.routers.hyperbeam-gateway-live-push.middlewares=hyperbeam-gateway-live-push-ratelimit,hyperbeam-gateway-live-buffering", + "traefik-ec.http.middlewares.hyperbeam-gateway-live-push-ratelimit.ratelimit.average=2", + "traefik-ec.http.middlewares.hyperbeam-gateway-live-push-ratelimit.ratelimit.burst=20" ] check { diff --git a/operations/hyperbeam-gateway-stage.hcl b/operations/hyperbeam-gateway-stage.hcl index 430d066..244356e 100644 --- a/operations/hyperbeam-gateway-stage.hcl +++ b/operations/hyperbeam-gateway-stage.hcl @@ -50,7 +50,14 @@ job "hyperbeam-gateway-stage" { "traefik-ec.http.routers.hyperbeam-gateway-stage.middlewares=hyperbeam-gateway-stage-ratelimit,hyperbeam-gateway-stage-buffering", "traefik-ec.http.middlewares.hyperbeam-gateway-stage-ratelimit.ratelimit.average=30", "traefik-ec.http.middlewares.hyperbeam-gateway-stage-ratelimit.ratelimit.burst=60", - "traefik-ec.http.middlewares.hyperbeam-gateway-stage-buffering.buffering.maxrequestbodybytes=1048576" + "traefik-ec.http.middlewares.hyperbeam-gateway-stage-buffering.buffering.maxrequestbodybytes=1048576", + "traefik-ec.http.routers.hyperbeam-gateway-stage-push.rule=Host(`hb-stage.anyone.tech`) && Method(`POST`)", + "traefik-ec.http.routers.hyperbeam-gateway-stage-push.entrypoints=https", + "traefik-ec.http.routers.hyperbeam-gateway-stage-push.tls=true", + "traefik-ec.http.routers.hyperbeam-gateway-stage-push.tls.certresolver=anyoneresolver", + "traefik-ec.http.routers.hyperbeam-gateway-stage-push.middlewares=hyperbeam-gateway-stage-push-ratelimit,hyperbeam-gateway-stage-buffering", + "traefik-ec.http.middlewares.hyperbeam-gateway-stage-push-ratelimit.ratelimit.average=2", + "traefik-ec.http.middlewares.hyperbeam-gateway-stage-push-ratelimit.ratelimit.burst=20" ] check {