Bound gateway checks of unknown signers and limit messages at the edge

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 `<process>:<action>`, 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.

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 }]);

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');

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<typeof loadConfig>;
@@ -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);

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 },

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<string>();
+  private timer: ReturnType<typeof setInterval> | 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<void> {
+    try {
+      const operators = await this.options.node.view<unknown>(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);
+  }
+}

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;

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<string, Record<string, unknown>>;
 
+const NEW_SIGNERS = 'new-signers';
+
 export class RolesCache {
   private readonly entries = new Map<string, { expiresAt: number; roles: Promise<RoleTable> }>();
 
@@ -135,8 +140,10 @@ export class WriteHandler {
 
   private async authorize(spec: ActionSpec, signer: string, tags: ReadonlyMap<string, string>): Promise<void> {
     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<Record<string, unknown> | 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;

operations/README.md

@@ -145,7 +145,7 @@ nomad job run -var commit_sha=<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-<env>` 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-<env>` 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-<env>` 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-<env>-node`, and the three process ids from the Consul keys `smart-contracts/<env>/operator-registry-address`, `relay-rewards-address` and `staking-rewards-address`. A change to either re-renders the environment and restarts the task.

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 {

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 {