fix(PoWeb): Fix minor parcel delivery and collection bugs (#240)

* wip

* collection tests

* rename file

* upgrade poweb lib

* implement e2e test

* implement keep-alive test

* delete console.log

package.json

@@ -25,7 +25,7 @@
     "node": ">=12"
   },
   "devDependencies": {
-    "@relaycorp/relaynet-poweb": "^1.3.2",
+    "@relaycorp/relaynet-poweb": "^1.3.8",
     "@relaycorp/shared-config": "^1.4.5",
     "@relaycorp/ws-mock": "^1.4.5",
     "@semantic-release/exec": "^5.0.0",

src/_test_utils.ts

@@ -113,14 +113,18 @@ export function sha256(plaintext: BinaryLike): Buffer {
 
 export function iterableTake<T>(max: number): (iterable: AsyncIterable<T>) => AsyncIterable<T> {
   return async function* (iterable: AsyncIterable<T>): AsyncIterable<T> {
+    if (max <= 0) {
+      return;
+    }
+
     // tslint:disable-next-line:no-let
     let count = 0;
     for await (const item of iterable) {
-      if (max <= count) {
-        break;
-      }
       yield item;
       count++;
+      if (max === count) {
+        break;
+      }
     }
   };
 }

src/functional_tests/docker-compose.override.yml

@@ -1,7 +1,7 @@
 version: '3.7'
 services:
   pong:
-    image: relaycorp/relaynet-pong:v1.3.4
+    image: relaycorp/relaynet-pong:v1.4.0
     init: true
     links:
       - redis
@@ -10,7 +10,7 @@ services:
       POHTTP_TLS_REQUIRED: 'false'
 
   pong-queue:
-    image: relaycorp/relaynet-pong:v1.3.4
+    image: relaycorp/relaynet-pong:v1.4.0
     init: true
     command: node build/main/bin/background-queue.js
     links:

src/functional_tests/e2e.test.ts → src/functional_tests/internet_e2e.test.ts

@@ -1,5 +1,3 @@
-// tslint:disable:no-let
-
 import { CogRPCClient } from '@relaycorp/cogrpc';
 import {
   Cargo,
@@ -12,21 +10,33 @@ import {
   ServiceMessage,
   SessionEnvelopedData,
   SessionlessEnvelopedData,
+  Signer,
 } from '@relaycorp/relaynet-core';
+import { PoWebClient, StreamingMode } from '@relaycorp/relaynet-poweb';
 import bufferToArray from 'buffer-to-arraybuffer';
+import pipe from 'it-pipe';
 import uuid from 'uuid-random';
 
 import { asyncIterableToArray, PdaChain } from '../_test_utils';
 import {
   configureServices,
   GW_GOGRPC_URL,
+  GW_POWEB_LOCAL_PORT,
   PONG_ENDPOINT_ADDRESS,
   runServiceCommand,
   vaultEnableSecret,
 } from './services';
-import { arrayToIterable, generatePdaChain, IS_GITHUB, sleep, TOMORROW } from './utils';
-
-configureServices(['cogrpc', 'crc-queue-worker', 'pohttp']);
+import { arrayToIterable, generatePdaChain, IS_GITHUB, sleep } from './utils';
+
+configureServices([
+  'poweb',
+  'cogrpc',
+  'pdc-outgoing-queue-worker',
+  'crc-queue-worker',
+  'pohttp',
+  'pong',
+  'pong-queue',
+]);
 
 let GW_PDA_CHAIN: PdaChain;
 beforeEach(async () => {
@@ -37,7 +47,45 @@ beforeEach(async () => {
   await vaultEnableSecret('pong-keys');
 });
 
-test('Sending pings and receiving pongs via CogRPC and PoHTTP', async () => {
+test('Sending pings via PoWeb and receiving pongs via PoHTTP', async () => {
+  const powebClient = PoWebClient.initLocal(GW_POWEB_LOCAL_PORT);
+  const privateGatewaySigner = new Signer(
+    GW_PDA_CHAIN.privateGatewayCert,
+    GW_PDA_CHAIN.privateGatewayPrivateKey,
+  );
+
+  const pongEndpointSessionCertificate = await generatePongEndpointKeyPairs();
+  const pingId = Buffer.from(uuid());
+  const pingParcelData = await makePingParcel(
+    pingId,
+    pongEndpointSessionCertificate.identityCert,
+    pongEndpointSessionCertificate.sessionCert,
+  );
+
+  // Deliver the ping message
+  await powebClient.deliverParcel(pingParcelData.parcelSerialized, privateGatewaySigner);
+
+  await sleep(IS_GITHUB ? 10 : 5);
+
+  // Collect the pong message once it's been received
+  const incomingParcels = await pipe(
+    powebClient.collectParcels([privateGatewaySigner], StreamingMode.CLOSE_UPON_COMPLETION),
+    async function* (collections): AsyncIterable<ArrayBuffer> {
+      for await (const collection of collections) {
+        yield collection.parcelSerialized;
+        await collection.ack();
+      }
+    },
+    asyncIterableToArray,
+  );
+  expect(incomingParcels).toHaveLength(1);
+
+  await expect(deserializePong(incomingParcels[0], pingParcelData.sessionKey)).resolves.toEqual(
+    pingId,
+  );
+});
+
+test('Sending pings via CogRPC and receiving pongs via PoHTTP', async () => {
   const pongEndpointSessionCertificate = await generatePongEndpointKeyPairs();
 
   const pingId = Buffer.from(uuid());
@@ -106,7 +154,7 @@ async function makePingParcel(
     issuerCertificate: GW_PDA_CHAIN.peerEndpointCert,
     issuerPrivateKey: GW_PDA_CHAIN.peerEndpointPrivateKey,
     subjectPublicKey: await identityCert.getPublicKey(),
-    validityEndDate: TOMORROW,
+    validityEndDate: GW_PDA_CHAIN.peerEndpointCert.expiryDate,
   });
   const pingSerialized = serializePing(pingId, pongEndpointPda);
 

src/functional_tests/poweb_server.test.ts

@@ -1,12 +1,30 @@
 import {
   derSerializePublicKey,
   generateRSAKeyPair,
+  issueDeliveryAuthorization,
+  issueEndpointCertificate,
+  Parcel,
   PrivateNodeRegistrationRequest,
+  Signer,
 } from '@relaycorp/relaynet-core';
-import { PoWebClient, ServerError } from '@relaycorp/relaynet-poweb';
+import {
+  ParcelDeliveryError,
+  PoWebClient,
+  RefusedParcelError,
+  ServerError,
+  StreamingMode,
+} from '@relaycorp/relaynet-poweb';
+import pipe from 'it-pipe';
 
+import { asyncIterableToArray, iterableTake, PdaChain } from '../_test_utils';
+import { expectBuffersToEqual } from '../services/_test_utils';
 import { configureServices, GW_POWEB_LOCAL_PORT } from './services';
-import { getPublicGatewayCertificate } from './utils';
+import {
+  generatePdaChain,
+  getPublicGatewayCertificate,
+  registerPrivateGateway,
+  sleep,
+} from './utils';
 
 configureServices(['poweb']);
 
@@ -16,14 +34,7 @@ describe('PoWeb server', () => {
       const client = PoWebClient.initLocal(GW_POWEB_LOCAL_PORT);
       const privateGatewayKeyPair = await generateRSAKeyPair();
 
-      const authorizationSerialized = await client.preRegisterNode(privateGatewayKeyPair.publicKey);
-      const registrationRequest = new PrivateNodeRegistrationRequest(
-        privateGatewayKeyPair.publicKey,
-        authorizationSerialized,
-      );
-      const registration = await client.registerNode(
-        await registrationRequest.serialize(privateGatewayKeyPair.privateKey),
-      );
+      const registration = await registerPrivateGateway(privateGatewayKeyPair, client);
 
       await expect(
         derSerializePublicKey(await registration.privateNodeCertificate.getPublicKey()),
@@ -54,4 +65,139 @@ describe('PoWeb server', () => {
       await expect(client.registerNode(pnrrSerialized)).rejects.toBeInstanceOf(ServerError);
     });
   });
+
+  describe('Parcel delivery and collection', () => {
+    test('Delivering and collecting a given parcel (closing upon completion)', async () => {
+      const client = PoWebClient.initLocal(GW_POWEB_LOCAL_PORT);
+      const senderChain = await generatePdaChain();
+      const recipientChain = await generatePdaChain();
+
+      const parcelSerialized = await generateDummyParcel(senderChain, recipientChain);
+
+      await client.deliverParcel(
+        parcelSerialized,
+        new Signer(senderChain.privateGatewayCert, senderChain.privateGatewayPrivateKey),
+      );
+
+      await sleep(2);
+
+      const parcelCollection = client.collectParcels(
+        [new Signer(recipientChain.privateGatewayCert, recipientChain.privateGatewayPrivateKey)],
+        StreamingMode.CLOSE_UPON_COMPLETION,
+      );
+      const incomingParcels = await pipe(
+        parcelCollection,
+        async function* (collections): AsyncIterable<ArrayBuffer> {
+          for await (const collection of collections) {
+            yield await collection.parcelSerialized;
+            await collection.ack();
+          }
+        },
+        asyncIterableToArray,
+      );
+      expect(incomingParcels).toHaveLength(1);
+      expectBuffersToEqual(parcelSerialized, incomingParcels[0]);
+    });
+
+    test('Delivering and collecting a given parcel (keep alive)', async () => {
+      const client = PoWebClient.initLocal(GW_POWEB_LOCAL_PORT);
+      const senderChain = await generatePdaChain();
+      const recipientChain = await generatePdaChain();
+
+      const parcelSerialized = await generateDummyParcel(senderChain, recipientChain);
+
+      await client.deliverParcel(
+        parcelSerialized,
+        new Signer(senderChain.privateGatewayCert, senderChain.privateGatewayPrivateKey),
+      );
+
+      const incomingParcels = await pipe(
+        client.collectParcels(
+          [new Signer(recipientChain.privateGatewayCert, recipientChain.privateGatewayPrivateKey)],
+          StreamingMode.KEEP_ALIVE,
+        ),
+        async function* (collections): AsyncIterable<ArrayBuffer> {
+          for await (const collection of collections) {
+            yield await collection.parcelSerialized;
+            await collection.ack();
+          }
+        },
+        iterableTake(1),
+        asyncIterableToArray,
+      );
+      expect(incomingParcels).toHaveLength(1);
+      expectBuffersToEqual(parcelSerialized, incomingParcels[0]);
+    });
+
+    test('Invalid parcel deliveries should be refused', async () => {
+      const client = PoWebClient.initLocal(GW_POWEB_LOCAL_PORT);
+      const privateGatewayKeyPair = await generateRSAKeyPair();
+      const privateGatewayRegistration = await registerPrivateGateway(
+        privateGatewayKeyPair,
+        client,
+      );
+
+      const invalidKeyPair = await generateRSAKeyPair();
+
+      await expect(
+        client.deliverParcel(
+          new ArrayBuffer(0),
+          new Signer(privateGatewayRegistration.privateNodeCertificate, invalidKeyPair.privateKey),
+        ),
+      ).rejects.toBeInstanceOf(ParcelDeliveryError);
+    });
+
+    test('Invalid parcels should be refused', async () => {
+      const client = PoWebClient.initLocal(GW_POWEB_LOCAL_PORT);
+      const privateGatewayKeyPair = await generateRSAKeyPair();
+      const privateGatewayRegistration = await registerPrivateGateway(
+        privateGatewayKeyPair,
+        client,
+      );
+
+      const sendingEndpointKeyPair = await generateRSAKeyPair();
+      const sendingEndpointCertificate = await issueEndpointCertificate({
+        issuerCertificate: privateGatewayRegistration.privateNodeCertificate,
+        issuerPrivateKey: privateGatewayKeyPair.privateKey,
+        subjectPublicKey: sendingEndpointKeyPair.publicKey,
+        validityEndDate: privateGatewayRegistration.privateNodeCertificate.expiryDate,
+      });
+
+      const parcel = new Parcel('0deadbeef', sendingEndpointCertificate, Buffer.from([]));
+      const parcelSerialized = await parcel.serialize(sendingEndpointKeyPair.privateKey);
+
+      await expect(
+        client.deliverParcel(
+          parcelSerialized,
+          new Signer(
+            privateGatewayRegistration.privateNodeCertificate,
+            privateGatewayKeyPair.privateKey,
+          ),
+        ),
+      ).rejects.toBeInstanceOf(RefusedParcelError);
+    });
+  });
 });
+
+async function generateDummyParcel(
+  senderChain: PdaChain,
+  recipientChain: PdaChain,
+): Promise<ArrayBuffer> {
+  const recipientEndpointCertificate = recipientChain.peerEndpointCert;
+  const sendingEndpointCertificate = await issueDeliveryAuthorization({
+    issuerCertificate: recipientEndpointCertificate,
+    issuerPrivateKey: recipientChain.peerEndpointPrivateKey,
+    subjectPublicKey: await senderChain.peerEndpointCert.getPublicKey(),
+    validityEndDate: recipientEndpointCertificate.expiryDate,
+  });
+
+  const parcel = new Parcel(
+    await recipientEndpointCertificate.calculateSubjectPrivateAddress(),
+    sendingEndpointCertificate,
+    Buffer.from([]),
+    {
+      senderCaCertificateChain: [recipientEndpointCertificate, recipientChain.privateGatewayCert],
+    },
+  );
+  return parcel.serialize(senderChain.peerEndpointPrivateKey);
+}