feat(PoWeb): Implement parcel delivery

README.md

@@ -5,6 +5,7 @@ This README is just a placeholder for the time being
 ## NATS Streaming Channels
 
 - `pdc-parcel.${localGatewayAddress}` where `${localGatewayAddress}` is the private address of the local gateway. Parcels received via Internet-based PDCs (e.g., PoHTTP) are published on this channel.
+- `internet-parcels`. Parcels bound for an internet.
 - `crc-cargo`. Cargo received via CRC (e.g., CogRPC) are published here.
 
 ## Object Storage

package.json

@@ -11,7 +11,7 @@
     "test": "run-s static-checks test:unit",
     "test:unit": "jest --coverage",
     "test:functional": "run-s test:functional:build test:functional:jest",
-    "test:functional:build": "docker-compose -f docker-compose.yml -f src/functional_tests/docker-compose.override.yml build",
+    "test:functional:build": "docker-compose -f docker-compose.yml -f src/functional_tests/docker-compose.override.yml --project-name gw-functional-tests build",
     "test:functional:jest": "jest --config jest.config.functional.js --runInBand",
     "static-checks": "run-p static-checks:*",
     "static-checks:lint": "tslint --project .",
@@ -30,6 +30,7 @@
     "@types/jest": "^26.0.13",
     "@types/mongoose": "^5.7.36",
     "@types/pkijs": "0.0.4",
+    "@types/split2": "^2.1.6",
     "@types/verror": "^1.10.4",
     "docker-compose": "^0.23.5",
     "dotenv": "^8.2.0",
@@ -40,6 +41,7 @@
     "npm-run-all": "^4.1.5",
     "pkijs": "^2.1.84",
     "prettier": "^2.1.1",
+    "split2": "^3.2.2",
     "trash-cli": "^3.1.0",
     "ts-jest": "^25.5.1",
     "ts-node": "^9.0.0",
@@ -68,7 +70,7 @@
   "dependencies": {
     "@relaycorp/cogrpc": "^1.1.5",
     "@relaycorp/keystore-vault": "^1.1.4",
-    "@relaycorp/relaynet-core": "^1.30.0",
+    "@relaycorp/relaynet-core": "^1.32.1",
     "@relaycorp/relaynet-pohttp": "^1.5.0",
     "@typegoose/typegoose": "^7.3.5",
     "@types/pino": "^6.3.0",

src/_test_utils.ts

@@ -1,6 +1,7 @@
 import { Certificate } from '@relaycorp/relaynet-core';
 import { BinaryLike, createHash, Hash } from 'crypto';
 import pino from 'pino';
+import split2 from 'split2';
 
 export async function* arrayToAsyncIterable<T>(array: readonly T[]): AsyncIterable<T> {
   for (const item of array) {
@@ -47,6 +48,26 @@ export function mockPino(): pino.Logger {
   return mockPinoLogger as any;
 }
 
+// tslint:disable-next-line:readonly-array
+export function makeMockLogging(): { readonly logger: pino.Logger; readonly logs: object[] } {
+  // tslint:disable-next-line:readonly-array
+  const logs: object[] = [];
+  const stream = split2((data) => {
+    logs.push(JSON.parse(data));
+  });
+  const logger = pino({ level: 'debug' }, stream);
+  return { logger, logs };
+}
+
+export function partialPinoLog(level: pino.Level, message: string, extraAttributes?: any): object {
+  const levelNumber = pino.levels.values[level];
+  return expect.objectContaining({
+    level: levelNumber,
+    msg: message,
+    ...(extraAttributes && extraAttributes),
+  });
+}
+
 export interface PdaChain {
   readonly publicGatewayCert: Certificate;
   readonly publicGatewayPrivateKey: CryptoKey;

src/backingServices/objectStorage.spec.ts

@@ -120,17 +120,17 @@ describe('ObjectStore', () => {
   });
 
   describe('initFromEnv', () => {
-    const baseEnvVars = {
+    const requiredEnvVars = {
       OBJECT_STORE_ACCESS_KEY_ID: 'the-access-key-id',
       OBJECT_STORE_ENDPOINT: 'objects.example.com:9000',
       OBJECT_STORE_SECRET_KEY: 'the-secret-key',
     };
-    const mockEnvVars = configureMockEnvVars(baseEnvVars);
+    const mockEnvVars = configureMockEnvVars(requiredEnvVars);
 
-    test.each(['OBJECT_STORE_ENDPOINT', 'OBJECT_STORE_ACCESS_KEY_ID', 'OBJECT_STORE_SECRET_KEY'])(
+    test.each(Object.getOwnPropertyNames(requiredEnvVars))(
       '%s should be required',
       (envVarKey: string) => {
-        mockEnvVars({ ...baseEnvVars, [envVarKey]: undefined });
+        mockEnvVars({ ...requiredEnvVars, [envVarKey]: undefined });
 
         expect(() => ObjectStoreClient.initFromEnv()).toThrowWithMessage(
           EnvVarError,
@@ -140,7 +140,7 @@ describe('ObjectStore', () => {
     );
 
     test('OBJECT_STORE_TLS_ENABLED should be honored if present', () => {
-      mockEnvVars({ ...baseEnvVars, OBJECT_STORE_TLS_ENABLED: 'false' });
+      mockEnvVars({ ...requiredEnvVars, OBJECT_STORE_TLS_ENABLED: 'false' });
 
       ObjectStoreClient.initFromEnv();
 
@@ -150,7 +150,7 @@ describe('ObjectStore', () => {
     });
 
     test('TLS should be enabled if OBJECT_STORE_TLS_ENABLED is missing', () => {
-      mockEnvVars({ ...baseEnvVars, OBJECT_STORE_TLS_ENABLED: undefined });
+      mockEnvVars({ ...requiredEnvVars, OBJECT_STORE_TLS_ENABLED: undefined });
 
       ObjectStoreClient.initFromEnv();
 
@@ -160,21 +160,21 @@ describe('ObjectStore', () => {
     });
 
     test('Environment variables should be passed to constructor', () => {
-      mockEnvVars({ ...baseEnvVars, OBJECT_STORE_TLS_ENABLED: undefined });
+      mockEnvVars({ ...requiredEnvVars, OBJECT_STORE_TLS_ENABLED: undefined });
 
       ObjectStoreClient.initFromEnv();
 
       expect(AWS.S3).toBeCalledTimes(1);
       const s3CallArgs = getMockContext(AWS.S3).calls[0][0];
       expect(s3CallArgs).toMatchObject<AWS.S3.Types.ClientConfiguration>({
-        accessKeyId: baseEnvVars.OBJECT_STORE_ACCESS_KEY_ID,
-        endpoint: baseEnvVars.OBJECT_STORE_ENDPOINT,
-        secretAccessKey: baseEnvVars.OBJECT_STORE_SECRET_KEY,
+        accessKeyId: requiredEnvVars.OBJECT_STORE_ACCESS_KEY_ID,
+        endpoint: requiredEnvVars.OBJECT_STORE_ENDPOINT,
+        secretAccessKey: requiredEnvVars.OBJECT_STORE_SECRET_KEY,
       });
     });
 
     test('Client should be returned', () => {
-      mockEnvVars({ ...baseEnvVars, OBJECT_STORE_TLS_ENABLED: undefined });
+      mockEnvVars({ ...requiredEnvVars, OBJECT_STORE_TLS_ENABLED: undefined });
 
       expect(ObjectStoreClient.initFromEnv()).toBeInstanceOf(ObjectStoreClient);
     });

src/functional_tests/cogrpc_server.test.ts

@@ -7,23 +7,17 @@ import {
   issueGatewayCertificate,
   Parcel,
 } from '@relaycorp/relaynet-core';
+import { deliverParcel } from '@relaycorp/relaynet-pohttp';
 import bufferToArray from 'buffer-to-arraybuffer';
 
 import { asyncIterableToArray } from '../_test_utils';
-import { ObjectStoreClient } from '../backingServices/objectStorage';
-import { ParcelStore } from '../services/parcelStore';
-import { configureServices, GW_GOGRPC_URL } from './services';
-import {
-  arrayToIterable,
-  generatePdaChain,
-  getFirstQueueMessage,
-  OBJECT_STORAGE_BUCKET,
-} from './utils';
+import { configureServices, GW_GOGRPC_URL, GW_POHTTP_URL } from './services';
+import { arrayToIterable, generatePdaChain, getFirstQueueMessage, sleep } from './utils';
 
 const TOMORROW = new Date();
 TOMORROW.setDate(TOMORROW.getDate() + 1);
 
-configureServices();
+configureServices(['cogrpc']);
 
 describe('Cargo delivery', () => {
   test('Authorized cargo should be accepted', async () => {
@@ -71,14 +65,11 @@ describe('Cargo collection', () => {
       await pdaChain.peerEndpointCert.calculateSubjectPrivateAddress(),
       pdaChain.pdaCert,
       Buffer.from([]),
+      { senderCaCertificateChain: [pdaChain.peerEndpointCert, pdaChain.privateGatewayCert] },
     );
     const parcelSerialized = await parcel.serialize(pdaChain.pdaGranteePrivateKey);
-    const parcelStore = new ParcelStore(ObjectStoreClient.initFromEnv(), OBJECT_STORAGE_BUCKET);
-    await parcelStore.storeGatewayBoundParcel(
-      parcel,
-      Buffer.from(parcelSerialized),
-      await pdaChain.privateGatewayCert.calculateSubjectPrivateAddress(),
-    );
+    await deliverParcel(GW_POHTTP_URL, parcelSerialized);
+    await sleep(1);
 
     const cca = new CargoCollectionAuthorization(
       GW_GOGRPC_URL,