user-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 powebsocketGateway = require('../PoWebSocket/gateway');
const {createHash} = require('crypto');
const {getAddressFromCert} = require('../core/pki');
const {PARCEL_SERIALIZER} = require('../core/serialization');

// 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.normalize(__dirname + '/../certs/user-gateway.cert.pem');
const KEY_PATH = path.normalize(__dirname + '/../certs/user-gateway.key.pem');

const ADDRESS = getAddressFromCert(fs.readFileSync(CERT_PATH));

const DATA_DIR_PATH = path.join(os.tmpdir(), 'user-gateway-data');
const POWEBSOCKET_SOCKET_PATH = path.join(DATA_DIR_PATH, 'powebsocket.sock');
const COSOCKET_SOCKET_PATH = path.join(DATA_DIR_PATH, 'cosocket.sock');

const PDC_PARCELS_DIR_PATH = path.join(DATA_DIR_PATH, 'pdc-parcels');
const CRC_PARCELS_DIR_PATH = path.join(DATA_DIR_PATH, 'crc-parcels');

const RELAYER_GW_CERT_PATH = path.normalize(__dirname + '/../certs/relayer-gateway.cert.pem');
const RELAYER_GW_ADDRESS = getAddressFromCert(fs.readFileSync(RELAYER_GW_CERT_PATH));

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

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

    cosocketServer(COSOCKET_SOCKET_PATH, CERT_PATH, KEY_PATH, parcelNotifier, fetchCargoesPayloads);
    powebsocketGateway.runUnixSocketServer(
        POWEBSOCKET_SOCKET_PATH,
        parcelNotifier,
        collectPdcParcels,
    );

    console.log(`Running user's gateway ${ADDRESS}

CRC:
 * Binding: CoSocket (${COSOCKET_SOCKET_PATH})
 * Relayer gateway address: ${RELAYER_GW_ADDRESS}

PDC:
 * Binding: PoWebSocket (${POWEBSOCKET_SOCKET_PATH})
`);
}

/**
 * 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.
 *
 * @param parcelSerialized
 * @returns {Promise<void>}
 */
async function deliverCrcParcel(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(PDC_PARCELS_DIR_PATH, `${internalParcelId}.parcel`);
    fs.writeFileSync(parcelPath, parcelSerialized);
    console.log(`[CRC] Added parcel ${parcel.id} to PDC queue.`);
}

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.`);
}

function ackPdcParcelCollection(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(`[PDC] Parcel ${parcelId} was collected and will be removed.`);
    fs.unlinkSync(parcelId);
}

async function* collectPdcParcels() {
    const parcelFileNames = fs.readdirSync(PDC_PARCELS_DIR_PATH)
        .map(f => path.join(PDC_PARCELS_DIR_PATH, f));

    for (const parcelFileName of parcelFileNames) {
        // Use fs.createReadStream() in production.
        yield {id: parcelFileName, parcel: fs.readFileSync(parcelFileName)};
    }
}

/**
 * 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 relayer's 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) {
    // The following must be in place in production. This PoC doesn't support Cargo Collection Authorizations (CCAs).
    // if (!gatewayAddresses.includes(RELAYER_GW_ADDRESS)) {
    //     // Because this is the user's gateway, the relayer's gateway is the only recipient of the cargo.
    //     // Any other target gateway is ignored.
    //     console.warn("[CRC] Ignoring parcel collection request because it isn't for our relayer");
    //     return;
    // }

    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: RELAYER_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();