mostly complete implementation

.dockerignore

@@ -3,6 +3,7 @@
 !src
 src/functional_tests
 src/**/*.spec.ts
+src/**/_test_utils.ts
 
 !tsconfig.json
 !package.json

package.json

@@ -69,7 +69,7 @@
   "dependencies": {
     "@relaycorp/cogrpc": "^1.1.7",
     "@relaycorp/keystore-vault": "^1.2.0",
-    "@relaycorp/relaynet-core": "^1.40.0",
+    "@relaycorp/relaynet-core": "^1.40.1",
     "@relaycorp/relaynet-pohttp": "^1.5.4",
     "@typegoose/typegoose": "^7.4.5",
     "@types/pino": "^6.3.4",

src/_test_utils.ts

@@ -1,9 +1,21 @@
-import { Certificate } from '@relaycorp/relaynet-core';
+import {
+  CargoCollectionAuthorization,
+  CargoCollectionRequest,
+  Certificate,
+  issueDeliveryAuthorization,
+  issueGatewayCertificate,
+  SessionlessEnvelopedData,
+} from '@relaycorp/relaynet-core';
 import bufferToArray from 'buffer-to-arraybuffer';
 import { BinaryLike, createHash, Hash } from 'crypto';
 import pino, { symbols as PinoSymbols } from 'pino';
 import split2 from 'split2';
 
+import { reSerializeCertificate } from './services/_test_utils';
+
+export const TOMORROW = new Date();
+TOMORROW.setDate(TOMORROW.getDate() + 1);
+
 export const UUID4_REGEX = expect.stringMatching(/^[0-9a-f-]+$/);
 
 export const MONGO_ENV_VARS = {
@@ -32,6 +44,15 @@ export function arrayBufferFrom(value: string): ArrayBuffer {
   return bufferToArray(Buffer.from(value));
 }
 
+export async function getPromiseRejection<E extends Error>(promise: Promise<any>): Promise<E> {
+  try {
+    await promise;
+  } catch (error) {
+    return error;
+  }
+  throw new Error('Expected project to reject');
+}
+
 // tslint:disable-next-line:readonly-array
 export function mockSpy<T, Y extends any[]>(
   spy: jest.MockInstance<T, Y>,
@@ -95,17 +116,6 @@ export function partialPinoLog(level: pino.Level, message: string, extraAttribut
   });
 }
 
-export interface PdaChain {
-  readonly publicGatewayCert: Certificate;
-  readonly publicGatewayPrivateKey: CryptoKey;
-  readonly privateGatewayCert: Certificate;
-  readonly privateGatewayPrivateKey: CryptoKey;
-  readonly peerEndpointCert: Certificate;
-  readonly peerEndpointPrivateKey: CryptoKey;
-  readonly pdaCert: Certificate;
-  readonly pdaGranteePrivateKey: CryptoKey;
-}
-
 function makeSHA256Hash(plaintext: BinaryLike): Hash {
   return createHash('sha256').update(plaintext);
 }
@@ -135,3 +145,58 @@ export function iterableTake<T>(max: number): (iterable: AsyncIterable<T>) => As
     }
   };
 }
+
+export interface CDAChain {
+  readonly publicGatewayCert: Certificate;
+  readonly privateGatewayCert: Certificate;
+}
+
+export interface ExternalPdaChain extends CDAChain {
+  readonly privateGatewayPrivateKey: CryptoKey;
+  readonly peerEndpointCert: Certificate;
+  readonly peerEndpointPrivateKey: CryptoKey;
+  readonly pdaCert: Certificate;
+  readonly pdaGranteePrivateKey: CryptoKey;
+}
+
+export interface PdaChain extends ExternalPdaChain {
+  readonly publicGatewayPrivateKey: CryptoKey;
+}
+
+export async function generateCDAChain(pdaChain: ExternalPdaChain): Promise<CDAChain> {
+  const privateGatewayCert = reSerializeCertificate(
+    await issueGatewayCertificate({
+      issuerPrivateKey: pdaChain.privateGatewayPrivateKey,
+      subjectPublicKey: await pdaChain.privateGatewayCert.getPublicKey(),
+      validityEndDate: pdaChain.privateGatewayCert.expiryDate,
+    }),
+  );
+  const publicGatewayCert = reSerializeCertificate(
+    await issueDeliveryAuthorization({
+      issuerCertificate: privateGatewayCert,
+      issuerPrivateKey: pdaChain.privateGatewayPrivateKey,
+      subjectPublicKey: await pdaChain.publicGatewayCert.getPublicKey(),
+      validityEndDate: TOMORROW,
+    }),
+  );
+  return { privateGatewayCert, publicGatewayCert };
+}
+
+export async function generateCCA(
+  recipientAddress: string,
+  chain: CDAChain,
+  publicGatewaySelfIssuedCertificate: Certificate,
+  privateGatewayPrivateKey: CryptoKey,
+): Promise<Buffer> {
+  const ccr = new CargoCollectionRequest(chain.publicGatewayCert);
+  const ccaPayload = await SessionlessEnvelopedData.encrypt(
+    ccr.serialize(),
+    publicGatewaySelfIssuedCertificate,
+  );
+  const cca = new CargoCollectionAuthorization(
+    recipientAddress,
+    chain.privateGatewayCert,
+    Buffer.from(await ccaPayload.serialize()),
+  );
+  return Buffer.from(await cca.serialize(privateGatewayPrivateKey));
+}

src/functional_tests/cogrpc_server.test.ts

@@ -1,17 +1,25 @@
-// tslint:disable:no-let
 import { CogRPCClient, CogRPCError } from '@relaycorp/cogrpc';
 import {
   Cargo,
   CargoCollectionAuthorization,
   generateRSAKeyPair,
   issueGatewayCertificate,
   Parcel,
+  RecipientAddressType,
 } from '@relaycorp/relaynet-core';
 import { deliverParcel } from '@relaycorp/relaynet-pohttp';
 import bufferToArray from 'buffer-to-arraybuffer';
+import grpc from 'grpc';
 import { Message, Stan, Subscription } from 'node-nats-streaming';
 
-import { asyncIterableToArray } from '../_test_utils';
+import {
+  asyncIterableToArray,
+  ExternalPdaChain,
+  generateCCA,
+  generateCDAChain,
+  getPromiseRejection,
+} from '../_test_utils';
+import { expectBuffersToEqual } from '../services/_test_utils';
 import { GW_GOGRPC_URL, GW_POHTTP_URL } from './services';
 import { arrayToIterable, connectToNatsStreaming, generatePdaChain, sleep } from './utils';
 
@@ -68,29 +76,21 @@ describe('Cargo delivery', () => {
 describe('Cargo collection', () => {
   test('Authorized CCA should be accepted', async () => {
     const pdaChain = await generatePdaChain();
-    const parcel = new Parcel(
-      await pdaChain.peerEndpointCert.calculateSubjectPrivateAddress(),
-      pdaChain.pdaCert,
-      Buffer.from([]),
-      { senderCaCertificateChain: [pdaChain.peerEndpointCert, pdaChain.privateGatewayCert] },
-    );
-    const parcelSerialized = await parcel.serialize(pdaChain.pdaGranteePrivateKey);
+    const parcelSerialized = await generateDummyParcel(pdaChain);
     await deliverParcel(GW_POHTTP_URL, parcelSerialized);
+
     await sleep(1);
 
-    const cca = new CargoCollectionAuthorization(
+    const ccaSerialized = await generateCCA(
       GW_GOGRPC_URL,
-      pdaChain.privateGatewayCert,
-      Buffer.from([]),
+      await generateCDAChain(pdaChain),
+      pdaChain.publicGatewayCert,
+      pdaChain.privateGatewayPrivateKey,
     );
     const cogrpcClient = await CogRPCClient.init(GW_GOGRPC_URL);
     let collectedCargoes: readonly Buffer[];
     try {
-      collectedCargoes = await asyncIterableToArray(
-        cogrpcClient.collectCargo(
-          Buffer.from(await cca.serialize(pdaChain.privateGatewayPrivateKey)),
-        ),
-      );
+      collectedCargoes = await asyncIterableToArray(cogrpcClient.collectCargo(ccaSerialized));
     } finally {
       cogrpcClient.close();
     }
@@ -104,10 +104,36 @@ describe('Cargo collection', () => {
     const { payload: cargoMessageSet } = await cargo.unwrapPayload(
       pdaChain.privateGatewayPrivateKey,
     );
-    await expect(Array.from(cargoMessageSet.messages)).toEqual([parcelSerialized]);
+    expect(cargoMessageSet.messages).toHaveLength(1);
+    expectBuffersToEqual(cargoMessageSet.messages[0], parcelSerialized);
+  });
+
+  test('Cargo should be signed with Cargo Delivery Authorization', async () => {
+    const pdaChain = await generatePdaChain();
+    await deliverParcel(GW_POHTTP_URL, await generateDummyParcel(pdaChain));
+
+    await sleep(1);
+
+    const cdaChain = await generateCDAChain(pdaChain);
+    const ccaSerialized = await generateCCA(
+      GW_GOGRPC_URL,
+      cdaChain,
+      pdaChain.publicGatewayCert,
+      pdaChain.privateGatewayPrivateKey,
+    );
+    const cogrpcClient = await CogRPCClient.init(GW_GOGRPC_URL);
+    let collectedCargoes: readonly Buffer[];
+    try {
+      collectedCargoes = await asyncIterableToArray(cogrpcClient.collectCargo(ccaSerialized));
+    } finally {
+      cogrpcClient.close();
+    }
+
+    const cargo = await Cargo.deserialize(bufferToArray(collectedCargoes[0]));
+    await cargo.validate(RecipientAddressType.PRIVATE, [cdaChain.privateGatewayCert]);
   });
 
-  test('Unauthorized CCA should return zero cargoes', async () => {
+  test('Unauthorized CCA should be refused', async () => {
     const unauthorizedSenderKeyPair = await generateRSAKeyPair();
     const unauthorizedCertificate = await issueGatewayCertificate({
       issuerPrivateKey: unauthorizedSenderKeyPair.privateKey,
@@ -123,35 +149,48 @@ describe('Cargo collection', () => {
     const ccaSerialized = Buffer.from(await cca.serialize(unauthorizedSenderKeyPair.privateKey));
     const cogrpcClient = await CogRPCClient.init(GW_GOGRPC_URL);
     try {
-      await expect(
+      const error = await getPromiseRejection<CogRPCError>(
         asyncIterableToArray(cogrpcClient.collectCargo(ccaSerialized)),
-      ).resolves.toHaveLength(0);
+      );
+      expect(error.cause()).toHaveProperty('code', grpc.status.UNAUTHENTICATED);
     } finally {
       cogrpcClient.close();
     }
   });
 
   test('CCAs should not be reusable', async () => {
     const pdaChain = await generatePdaChain();
-    const cca = new CargoCollectionAuthorization(
+    const cdaChain = await generateCDAChain(pdaChain);
+    const ccaSerialized = await generateCCA(
       GW_GOGRPC_URL,
-      pdaChain.privateGatewayCert,
-      Buffer.from([]),
+      cdaChain,
+      pdaChain.publicGatewayCert,
+      pdaChain.privateGatewayPrivateKey,
     );
-    const ccaSerialized = Buffer.from(await cca.serialize(pdaChain.privateGatewayPrivateKey));
 
     const cogrpcClient = await CogRPCClient.init(GW_GOGRPC_URL);
     try {
       await expect(asyncIterableToArray(cogrpcClient.collectCargo(ccaSerialized))).toResolve();
-      await expect(
+      const error = await getPromiseRejection<CogRPCError>(
         asyncIterableToArray(cogrpcClient.collectCargo(ccaSerialized)),
-      ).rejects.toBeInstanceOf(CogRPCError);
+      );
+      expect(error.cause()).toHaveProperty('code', grpc.status.PERMISSION_DENIED);
     } finally {
       cogrpcClient.close();
     }
   });
 });
 
+async function generateDummyParcel(pdaChain: ExternalPdaChain): Promise<ArrayBuffer> {
+  const parcel = new Parcel(
+    await pdaChain.peerEndpointCert.calculateSubjectPrivateAddress(),
+    pdaChain.pdaCert,
+    Buffer.from([]),
+    { senderCaCertificateChain: [pdaChain.peerEndpointCert, pdaChain.privateGatewayCert] },
+  );
+  return parcel.serialize(pdaChain.pdaGranteePrivateKey);
+}
+
 async function getLastQueueMessage(): Promise<Buffer | undefined> {
   const stanConnection = await connectToNatsStreaming();
   const subscription = subscribeToCRCChannel(stanConnection);

src/functional_tests/internet_e2e.test.ts

@@ -2,7 +2,6 @@ import { CogRPCClient } from '@relaycorp/cogrpc';
 import { VaultPrivateKeyStore } from '@relaycorp/keystore-vault';
 import {
   Cargo,
-  CargoCollectionAuthorization,
   CargoMessageSet,
   Certificate,
   issueDeliveryAuthorization,
@@ -19,9 +18,15 @@ import { get as getEnvVar } from 'env-var';
 import pipe from 'it-pipe';
 import uuid from 'uuid-random';
 
-import { asyncIterableToArray, iterableTake } from '../_test_utils';
+import {
+  asyncIterableToArray,
+  ExternalPdaChain,
+  generateCCA,
+  generateCDAChain,
+  iterableTake,
+} from '../_test_utils';
 import { GW_GOGRPC_URL, GW_POWEB_LOCAL_PORT, PONG_ENDPOINT_ADDRESS } from './services';
-import { arrayToIterable, ExternalPdaChain, generatePdaChain, IS_GITHUB, sleep } from './utils';
+import { arrayToIterable, generatePdaChain, IS_GITHUB, sleep } from './utils';
 
 test('Sending pings via PoWeb and receiving pongs via PoHTTP', async () => {
   const powebClient = PoWebClient.initLocal(GW_POWEB_LOCAL_PORT);
@@ -64,22 +69,22 @@ test('Sending pings via PoWeb and receiving pongs via PoHTTP', async () => {
 
 test('Sending pings via CogRPC and receiving pongs via PoHTTP', async () => {
   const pongEndpointSessionCertificate = await getPongEndpointKeyPairs();
-  const gwPDAChain = await generatePdaChain();
+  const pdaChain = await generatePdaChain();
 
   const pingId = Buffer.from(uuid());
   const pingParcelData = await makePingParcel(
     pingId,
     pongEndpointSessionCertificate.identityCert,
     pongEndpointSessionCertificate.sessionCert,
-    gwPDAChain,
+    pdaChain,
   );
 
   const cogRPCClient = await CogRPCClient.init(GW_GOGRPC_URL);
   try {
     // Deliver the ping message encapsulated in a cargo
     const cargoSerialized = await encapsulateParcelsInCargo(
       [pingParcelData.parcelSerialized],
-      gwPDAChain,
+      pdaChain,
     );
     await asyncIterableToArray(
       cogRPCClient.deliverCargo(
@@ -90,11 +95,16 @@ test('Sending pings via CogRPC and receiving pongs via PoHTTP', async () => {
     await sleep(IS_GITHUB ? 4 : 2);
 
     // Collect the pong message encapsulated in a cargo
-    const collectedCargoes = await asyncIterableToArray(
-      cogRPCClient.collectCargo(await makeCCA(gwPDAChain)),
+    const cdaChain = await generateCDAChain(pdaChain);
+    const ccaSerialized = await generateCCA(
+      GW_GOGRPC_URL,
+      cdaChain,
+      pdaChain.publicGatewayCert,
+      pdaChain.privateGatewayPrivateKey,
     );
+    const collectedCargoes = await asyncIterableToArray(cogRPCClient.collectCargo(ccaSerialized));
     expect(collectedCargoes).toHaveLength(1);
-    const collectedMessages = await extractMessagesFromCargo(collectedCargoes[0], gwPDAChain);
+    const collectedMessages = await extractMessagesFromCargo(collectedCargoes[0], pdaChain);
     expect(collectedMessages).toHaveLength(2);
     expect(ParcelCollectionAck.deserialize(collectedMessages[0])).toHaveProperty(
       'parcelId',
@@ -200,15 +210,6 @@ async function encapsulateParcelsInCargo(
   return Buffer.from(await cargo.serialize(gwPDAChain.privateGatewayPrivateKey));
 }
 
-async function makeCCA(gwPDAChain: ExternalPdaChain): Promise<Buffer> {
-  const cca = new CargoCollectionAuthorization(
-    GW_GOGRPC_URL,
-    gwPDAChain.privateGatewayCert,
-    Buffer.from([]),
-  );
-  return Buffer.from(await cca.serialize(gwPDAChain.privateGatewayPrivateKey));
-}
-
 async function extractMessagesFromCargo(
   cargoSerialized: Buffer,
   gwPDAChain: ExternalPdaChain,