Guard ticket signing against observed epoch changes

README.md

@@ -142,6 +142,15 @@ Connection retry behavior can be configured via environment variables:
 | DIODE_TICKET_BYTES_THRESHOLD | Bytes threshold for ticket updates | 4194304 (4 MiB) |
 | DIODE_TICKET_UPDATE_INTERVAL | Time interval for ticket updates (ms) | 30000 (30s) |
 
+With relay `hello` version 1001, ticket usage is an absolute device byte count.
+The client checks the ticket epoch before and after fetching it, and again
+before signing. A change between those steps cancels the ticket so the next
+attempt fetches a fresh report. The relay report contains no epoch, however,
+and the relay's usage counter can follow a different clock from the ticket
+chain. This check cannot prove which epoch a high report belongs to after a
+cold start or reconnect. Complete protection requires the relay to return an
+epoch alongside usage.
+
 Example `.env` file:
 ```
 DIODE_MAX_RETRIES=10

connection.js

@@ -201,6 +201,8 @@ class DiodeConnection extends EventEmitter {
     this._relayUsage = null;
     this._relayUsageGeneration = -1;
     this._relayUsageSequence = 0;
+    this._ticketUsageEpoch = null;
+    this._ticketUsageSequence = null;
     this._supportsRelayUsage = null;
     this._relayUsageQueryInFlight = false;
     this._ticketAcceptedOnTransport = false;
@@ -278,6 +280,8 @@ class DiodeConnection extends EventEmitter {
     this.lastRelayMeasuredBytes = 0;
     this._relayUsage = null;
     this._relayUsageGeneration = -1;
+    this._ticketUsageEpoch = null;
+    this._ticketUsageSequence = null;
     this._supportsRelayUsage = null;
     this._relayUsageQueryInFlight = false;
     this._ticketAcceptedOnTransport = false;
@@ -929,6 +933,10 @@ class DiodeConnection extends EventEmitter {
         this._relayUsage = deviceUsage;
         this._relayUsageGeneration = this._socketGeneration;
         this._relayUsageSequence += 1;
+        // An unsolicited report can supersede the one checked against the
+        // ticket epoch. Require a fresh check before signing it.
+        this._ticketUsageEpoch = null;
+        this._ticketUsageSequence = null;
         this.emit('relay_usage', {
           generation: this._socketGeneration,
           sequence: this._relayUsageSequence,
@@ -1088,12 +1096,12 @@ class DiodeConnection extends EventEmitter {
       return Promise.reject(new DiodeDisconnectedError('Diode relay disconnected before reporting usage'));
     }
     if (this._relayUsageGeneration === generation && this._relayUsageSequence > sequence) {
-      return Promise.resolve(this._relayUsage);
+      return Promise.resolve({ usage: this._relayUsage, sequence: this._relayUsageSequence });
     }
     return new Promise((resolve, reject) => {
       let timer;
       let settleImmediate;
-      let latestUsage;
+      let latestReport;
       const cleanup = () => {
         clearTimeout(timer);
         if (settleImmediate) clearImmediate(settleImmediate);
@@ -1102,11 +1110,11 @@ class DiodeConnection extends EventEmitter {
       };
       const onUsage = (event) => {
         if (event.generation !== generation || event.sequence <= sequence) return;
-        latestUsage = event.usage;
+        latestReport = { usage: event.usage, sequence: event.sequence };
         if (settleImmediate) return;
         settleImmediate = setImmediate(() => {
           cleanup();
-          resolve(latestUsage);
+          resolve(latestReport);
         });
       };
       const onDisconnect = (event) => {
@@ -1134,13 +1142,24 @@ class DiodeConnection extends EventEmitter {
     const send = allowTransportReady
       ? (command) => this._sendCommandTransportReady(command, commandOptions)
       : (command) => this.sendCommand(command, commandOptions);
+    this._ticketUsageEpoch = null;
+    this._ticketUsageSequence = null;
 
     // The relay's paid ticket is scoped to an epoch; never use the previous
     // epoch's floor when interpreting current device usage.
     const epoch = allowTransportReady
       ? await this.RPC._getEpochWithSender(send)
       : await this.RPC.getEpoch();
     if (hasExpectedTransport) this._assertCurrentTransport(expectedSocket, expectedGeneration, 'Diode usage epoch');
+    const assertUsageEpoch = async () => {
+      const currentEpoch = allowTransportReady
+        ? await this.RPC._getEpochWithSender(send)
+        : await this.RPC.getEpoch();
+      if (hasExpectedTransport) this._assertCurrentTransport(expectedSocket, expectedGeneration, 'Diode usage epoch');
+      if (generation !== this._socketGeneration || currentEpoch !== epoch) {
+        throw new DiodeConnectionError('Diode ticket epoch changed while measuring usage', 'DIODE_USAGE_EPOCH_CHANGED');
+      }
+    };
     if (epoch !== this.lastAcceptedTicketEpoch) {
       this.lastAcceptedTicketEpoch = epoch;
       this.lastAcceptedTicketBytes = 0;
@@ -1153,9 +1172,14 @@ class DiodeConnection extends EventEmitter {
         // ambiguous pre-ticket `bytes` reply to that floor.
         this.totalBytes = Math.max(128000, this.lastAcceptedTicketBytes);
         this.accumulatedBytes = Math.max(0, this.totalBytes - this.lastAcceptedTicketBytes);
+        await assertUsageEpoch();
+        this._ticketUsageEpoch = epoch;
         return null;
       }
-      return this._syncMeasuredBytesWithRelay({ allowTransportReady, expectedSocket, expectedGeneration });
+      const measuredBytes = await this._syncMeasuredBytesWithRelay({ allowTransportReady, expectedSocket, expectedGeneration });
+      await assertUsageEpoch();
+      this._ticketUsageEpoch = epoch;
+      return measuredBytes;
     }
 
     const sequence = this._relayUsageSequence;
@@ -1177,12 +1201,19 @@ class DiodeConnection extends EventEmitter {
       // but unsolicited dispatch is deferred. Drain that batch so an older
       // queued request cannot win over the fresh report.
       await new Promise(setImmediate);
-      const usage = await this._waitForRelayUsageAfter(sequence, generation);
+      const report = await this._waitForRelayUsageAfter(sequence, generation);
       if (hasExpectedTransport) this._assertCurrentTransport(expectedSocket, expectedGeneration, 'Diode usage report');
+      const usage = report.usage;
       if (!Number.isSafeInteger(usage) || usage < 0) {
         throw new DiodeConnectionError('Diode relay reported invalid usage', 'DIODE_USAGE_INVALID');
       }
       this._supportsRelayUsage = true;
+      await assertUsageEpoch();
+      if (this._relayUsageSequence !== report.sequence || this._relayUsageGeneration !== generation) {
+        throw new DiodeConnectionError('Diode relay usage changed during epoch check', 'DIODE_USAGE_STALE');
+      }
+      this._ticketUsageEpoch = epoch;
+      this._ticketUsageSequence = report.sequence;
       this.totalBytes = this.lastAcceptedTicketBytes > 0 && usage <= this.lastAcceptedTicketBytes
         ? this.lastAcceptedTicketBytes
         : usage + this._ticketHeadroomBytes;
@@ -1627,6 +1658,9 @@ class DiodeConnection extends EventEmitter {
     if (hasExpectedTransport) {
       this._assertCurrentTransport(expectedSocket, expectedGeneration, 'Diode ticket epoch');
     }
+    if (this._supportsRelayUsage !== null && this._ticketUsageEpoch !== epoch) {
+      throw new DiodeConnectionError('Diode ticket usage belongs to another epoch', 'DIODE_USAGE_EPOCH_CHANGED');
+    }
     // Relay ticket floors are scoped to an epoch. The previous epoch's
     // accepted total cannot bound a ticket in the new epoch.
     if (epoch !== this.lastAcceptedTicketEpoch) {
@@ -1640,7 +1674,11 @@ class DiodeConnection extends EventEmitter {
       ? Math.max(0, this.lastAcceptedTicketBytes)
       : 0;
     let totalBytes;
-    if (this._supportsRelayUsage === true && this._relayUsageGeneration === this._socketGeneration) {
+    if (this._supportsRelayUsage === true) {
+      if (this._relayUsageGeneration !== this._socketGeneration ||
+          this._ticketUsageSequence !== this._relayUsageSequence) {
+        throw new DiodeConnectionError('Diode ticket usage is stale', 'DIODE_USAGE_STALE');
+      }
       totalBytes = acceptedBytes > 0 && this._relayUsage <= acceptedBytes
         ? acceptedBytes
         : this._relayUsage + this._ticketHeadroomBytes;
@@ -1656,6 +1694,16 @@ class DiodeConnection extends EventEmitter {
       localAddress,
       epoch
     );
+    // Signing may yield. Do not return a ticket based on a report that was
+    // superseded while its signature was being created.
+    if (this._supportsRelayUsage === true &&
+        (this._relayUsageGeneration !== this._socketGeneration ||
+         this._ticketUsageSequence !== this._relayUsageSequence)) {
+      throw new DiodeConnectionError('Diode ticket usage is stale', 'DIODE_USAGE_STALE');
+    }
+    if (this._supportsRelayUsage !== null && this._ticketUsageEpoch !== epoch) {
+      throw new DiodeConnectionError('Diode ticket usage belongs to another epoch', 'DIODE_USAGE_EPOCH_CHANGED');
+    }
     if (hasExpectedTransport) {
       this._assertCurrentTransport(expectedSocket, expectedGeneration, 'Diode ticket signature');
       this.totalConnections = totalConnections;

package.json

@@ -1,6 +1,6 @@
 {
   "name": "diodejs",
-  "version": "0.5.9",
+  "version": "0.5.10",
   "description": "A JavaScript client for interacting with the Diode network. It provides functionalities to bind and publish ports, send RPC commands, and handle responses.",
   "main": "index.js",
   "files": [

test/connection.ticketRetry.test.js

@@ -308,6 +308,8 @@ test('too_low repair reconciles against authoritative relay usage without adding
   connection._supportsRelayUsage = true;
   connection._relayUsageGeneration = connection._socketGeneration;
   connection._relayUsage = 820_000_000;
+  connection._ticketUsageEpoch = 687;
+  connection._ticketUsageSequence = connection._relayUsageSequence;
   const ticket = await connection.createTicketCommand();
   assert.equal(ticket[5], 820_001_024);
   assert.equal(connection.totalBytes, 820_001_024);
@@ -462,6 +464,8 @@ test('active stream too_low retries use fresh usage and bounded temporary margin
     connection._supportsRelayUsage = true;
     connection._relayUsageGeneration = connection._socketGeneration;
     connection._relayUsage = reports[refreshes];
+    connection._ticketUsageEpoch = 687;
+    connection._ticketUsageSequence = connection._relayUsageSequence;
     refreshes += 1;
   };
   connection.sendCommand = (command, options) => {
@@ -521,6 +525,117 @@ test('new epoch usage replaces the previous epoch paid floor', async () => {
   assert.equal(connection.lastAcceptedTicketBytes, 0);
 });
 
+test('ticket creation refuses a usage report from the previous epoch', async () => {
+  const connection = makeConnection();
+  stubTicketSigning(connection);
+  let epoch = 687;
+  connection.RPC.getEpoch = async () => epoch;
+  stubHelloUsage(connection, 500_000_000);
+
+  await connection._refreshTicketUsage();
+  epoch = 688; // Epoch rolls over after hello but before ticket signing.
+
+  await assert.rejects(connection.createTicketCommand(), (error) =>
+    error.code === 'DIODE_USAGE_EPOCH_CHANGED');
+  assert.equal(connection.lastAcceptedTicketEpoch, 687);
+  assert.equal(connection.lastAcceptedTicketBytes, 0);
+
+  stubHelloUsage(connection, 20_000_000);
+  await connection._refreshTicketUsage();
+  const retry = await connection.createTicketCommand();
+  assert.equal(retry[2], 688);
+  assert.equal(retry[5], 20_001_024);
+});
+
+test('hello refuses usage when the epoch changes while the report is in flight', async () => {
+  const connection = makeConnection();
+  stubTicketSigning(connection);
+  const epochs = [687, 688];
+  connection.RPC.getEpoch = async () => epochs.shift() ?? 688;
+  stubHelloUsage(connection, 500_000_000);
+
+  await assert.rejects(connection._refreshTicketUsage(), (error) =>
+    error.code === 'DIODE_USAGE_EPOCH_CHANGED');
+  assert.equal(connection._ticketUsageEpoch, null);
+  await assert.rejects(connection.createTicketCommand(), (error) =>
+    error.code === 'DIODE_USAGE_EPOCH_CHANGED');
+});
+
+test('too_low retry after rollover signs fresh-epoch usage', async () => {
+  const connection = makeConnection();
+  stubTicketSigning(connection);
+  let epoch = 687;
+  connection.RPC.getEpoch = async () => epoch;
+  stubHelloUsage(connection, 500_000_000);
+  await connection._refreshTicketUsage();
+  const previousEpochTicket = await connection.createTicketCommand();
+  assert.equal(previousEpochTicket[5], 500_001_024);
+  epoch = 688;
+
+  let retryTicket;
+  connection.sendCommand = async (command) => {
+    if (command[0] === 'hello') {
+      reportRelayUsage(connection, 20_000_000);
+      return ['ok'];
+    }
+    retryTicket = command;
+    return ['thanks!'];
+  };
+  const response = new Promise((resolve, reject) => {
+    connection.pendingRequests.set(1, {
+      resolve,
+      reject,
+      commandArray: previousEpochTicket,
+      ticketRetryCount: 0,
+    });
+  });
+  connection._handleData(encodeResponse(1, tooLowResponse({ epoch: 688, totalBytes: 0 })));
+
+  assert.deepEqual(await response, ['thanks!']);
+  assert.equal(retryTicket[2], 688);
+  assert.equal(retryTicket[5], 20_000_000 + 64 * 1024);
+});
+
+test('an extra usage request during epoch validation retries with a fresh report', async () => {
+  const connection = makeConnection();
+  stubTicketSigning(connection);
+  let reads = 0;
+  connection.RPC.getEpoch = async () => {
+    reads += 1;
+    if (reads === 2) {
+      reportRelayUsage(connection, 10_000_100);
+      await new Promise(setImmediate);
+    }
+    return 687;
+  };
+  stubHelloUsage(connection, 10_000_000);
+  await assert.rejects(connection._refreshTicketUsage(), (error) =>
+    error.code === 'DIODE_USAGE_STALE');
+
+  stubHelloUsage(connection, 10_000_100);
+  await connection._refreshTicketUsage();
+  assert.equal((await connection.createTicketCommand())[5], 10_001_124);
+});
+
+test('signing refuses a usage report superseded during an async signature', async () => {
+  const connection = makeConnection();
+  stubTicketSigning(connection);
+  stubHelloUsage(connection, 10_000_000);
+  await connection._refreshTicketUsage();
+  connection.createTicketSignature = async () => {
+    reportRelayUsage(connection, 11_000_000);
+    await new Promise(setImmediate);
+    return Buffer.alloc(65, 1);
+  };
+
+  await assert.rejects(connection.createTicketCommand(), (error) =>
+    error.code === 'DIODE_USAGE_STALE');
+  stubHelloUsage(connection, 11_000_000);
+  connection.createTicketSignature = async () => Buffer.alloc(65, 1);
+  await connection._refreshTicketUsage();
+  assert.equal((await connection.createTicketCommand())[5], 11_001_024);
+});
+
 test('unsupported hello wire error probes the paid floor without ambiguous pre-ticket bytes', async () => {
   const connection = makeConnection();
   stubTicketSigning(connection);