relayer-gateway

#!/usr/bin/env node
'use strict';
// This is a proof of concept. The code below is ugly, inefficient and has no tests.

const cosocketServer = require('../CoSocket/server');
const EventEmitter = require('events');
const fs = require('fs');
const os = require('os');
const path = require('path');
const pogrpcGateway = require('../PogRPC/gateway');
const PogRPCClient = require('../PogRPC/client');
const {createHash} = require('crypto');
const {getAddressFromCert} = require('../core/pki');
const {PARCEL_SERIALIZER} = require('../core/serialization');

const DATA_DIR_PATH = path.join(os.tmpdir(), 'relayer-gateway-data');
const CRC_PARCELS_DIR_PATH = path.join(DATA_DIR_PATH, 'crc-parcels');

const COSOCKET_PATH = path.join(DATA_DIR_PATH, 'cosocket.sock');
const POGRPC_NETLOC = '127.0.0.1:21473'; // 21473? Too late!

const CERTS_DIR_PATH = path.join(path.dirname(__dirname), 'certs');
// This gateway is reusing the asymmetric keys for parcel delivery and cargo relay *for expediency*
// in the context of this PoC.
const CERT_PATH = path.join(CERTS_DIR_PATH, 'relayer-gateway.cert.pem');
const KEY_PATH = path.join(CERTS_DIR_PATH, 'relayer-gateway.key.pem');

const GATEWAY_CRC_ADDRESS = getAddressFromCert(fs.readFileSync(CERT_PATH));
const GATEWAY_PDC_ADDRESS = `rne+grpc://${POGRPC_NETLOC}`;

// This is *WRONG*. The the relaying gateway MUST take the target gateway's certificate from the
// Cargo Collection Authorization (CCA), which MUST be attached to the request to collect cargo.
// However, this PoC doesn't (yet) support CCAs, so we can't extract the certificate from any CCA.
const USER_GW_CERT_PATH = path.join(CERTS_DIR_PATH, 'user-gateway.cert.pem');

// We're whitelisting the self-signed cert for the API endpoint's gRPC server for expediency, but this MUST NOT be
// done in production.
const TWITTER_API_ENDPOINT_SERVER_CERT = fs.readFileSync(path.join(
    CERTS_DIR_PATH,
    'twitter-endpoint-server.cert.pem',
));

function main() {
    fs.mkdirSync(DATA_DIR_PATH, {recursive: true});
    fs.mkdirSync(CRC_PARCELS_DIR_PATH, {recursive: true});

    const parcelNotifier = new EventEmitter();
    parcelNotifier.on('pdc', deliverPndParcel);
    parcelNotifier.on('crc', deliverCrcParcel);
    parcelNotifier.on('crcCollection', ackCrcParcelCollection);

    cosocketServer(COSOCKET_PATH, CERT_PATH, KEY_PATH, parcelNotifier, fetchCargoesPayloads);
    pogrpcGateway.runServer(POGRPC_NETLOC, parcelNotifier);

    console.log(`Running relayer's gateway ${GATEWAY_CRC_ADDRESS}
CoSocket listening on ${COSOCKET_PATH}
PogRPC listening on ${POGRPC_NETLOC}
`);
}

/**
 * Deliver parcel (received through CRC) to its corresponding public endpoint.
 *
 * This function is called for each parcel contained in a cargo relayed to the this gateway.
 *
 * The production-ready equivalent of this function would be different in multiple ways. For example:
 *
 * - It could support multiple parcel delivery bindings, not just PogRPC.
 * - Whatever the binding, it should _queue_ the delivery, not do it in band.
 * - It should support parcels sent to private endpoints (those with opaque
 *   addresses). Whilst unlikely, it'd be necessary if/when the relayer exposes
 *   private endpoints.
 *
 * @param parcelSerialized
 * @returns {Promise<void>}
 */
async function deliverCrcParcel(parcelSerialized) {
    const {id, recipient} = await PARCEL_SERIALIZER.deserialize(parcelSerialized);
    const recipientMatch = recipient.match(/^rne(\+(?<binding>[\w]+))?:\/\/(?<address>.+)$/);

    if (!recipientMatch) {
        console.error(`Invalid public endpoint address: ${recipient}.`);
        return;
    }

    const {binding, address} = recipientMatch.groups;

    if (binding && binding !== 'grpc') {
        // The address includes the binding hint, but PogRPC is the only supported binding in this PoC.
        console.error(`This PoC can only deliver parcels via PogRPC. Parcel ${id}  will be ignored.`);
        return;

        // Note: When the binding hint is missing from the address, we should use
        // Application-Layer Protocol Negotiation (ALPN) to determine the binding.
        // This PoC doesn't use ALPN for expediency and it only supports PogRPC.
    }

    const pdcClient = new PogRPCClient(
        address,
        GATEWAY_PDC_ADDRESS,
        TWITTER_API_ENDPOINT_SERVER_CERT, // Don't do in production
    );
    try {
        await pdcClient.deliverParcels([parcelSerialized]);
    } catch (error) {
        // We should retry in production, depending on the type of error.
        console.error(`Could not deliver parcel ${id} to ${recipient}:`, error.toString());
        return;
    }

    console.log(`[CRC] Delivered parcel ${id} to ${recipient}`);
}

async function deliverPndParcel(parcelSerialized) {
    const parcel = await PARCEL_SERIALIZER.deserialize(parcelSerialized);

    const hash = createHash('sha256');
    hash.update(parcel.senderCert);
    hash.update(parcel.id);
    const internalParcelId = hash.digest('hex');

    const parcelPath = path.join(CRC_PARCELS_DIR_PATH, `${internalParcelId}.parcel`);
    fs.writeFileSync(parcelPath, parcelSerialized);
    console.log(`[PDC] Added parcel ${parcel.id} to CRC queue.`);
}

/**
 * Return all the parcels that should be relayed in a single cargo.
 *
 * In production, the equivalent of this function MUST yield multiple values if the parcels could not
 * fit in one cargo.
 *
 * The function is a generator and takes the target gateway addresses as input because the generic code
 * in the CRC server should be agnostic of the storage medium. This implementation uses the file system,
 * but a relaying gateway could use S3 or a DB to store the parcels, for example.
 *
 * @param {Array<string>} gatewayAddresses
 * @returns {AsyncIterableIterator<{gatewayCertPath: string, parcels}>}
 */
async function* fetchCargoesPayloads(gatewayAddresses) {
    // This PoC doesn't support Cargo Collection Authorizations (CCAs), so there's nothing we can do with
    // gatewayAddresses. In production, we MUST only yield cargo for addresses in gatewayAddresses.
    // In other words, all the cargo will be encrypted using the user's gateway's certificate.

    const parcelFileNames = fs.readdirSync(CRC_PARCELS_DIR_PATH).map(f => path.join(CRC_PARCELS_DIR_PATH, f));

    if (parcelFileNames.length === 0) {
        return;
    }

    const parcels = {};
    for (const parcelFileName of parcelFileNames) {
        // Use fs.createReadStream() in production.
        parcels[parcelFileName] = fs.readFileSync(parcelFileName);
    }
    yield {
        gatewayCertPath: USER_GW_CERT_PATH, // Again, we MUSTN'T do this, but the PoC doesn't support CCAs
        parcels,
    };
}

function ackCrcParcelCollection(parcelId) {
    // In this case, the parcelId is the path on disk, but it could be an S3 object URL
    // or a DB PK (if the parcel is stored in a DB), for example. This scenario is likely
    // with a relayer gateway.

    // We should check that the path actually corresponds to a collected parcel in
    // production.

    console.log(`[CRC] Cargo containing parcel ${parcelId} was collected. The parcel will be removed.`);
    fs.unlinkSync(parcelId);
}

main();