Propagate root keypair changes to domains external to KeyManager

[emmacasolin]

Specification

When the root keypair is changed, this needs to be propagated to other places that rely on knowing this information. In particular, the new Node Id generated by a new keypair needs to be propagated to the NodeManager and Status, among other places. This can be achieved either through the use of the Observer Pattern, or by using Event Emitters (preferred due to looser coupling).

Additional context

• Example implementation of the Observer Pattern for this purpose: f097671
• Discussion of Observer Pattern vs. Event Emitters (Pub/Sub Pattern): CLI and Client & Agent Service test splitting #311 (comment)
• Discussion of dependency injection + coupling issues: CLI and Client & Agent Service test splitting #311 (comment)
• Discussion of propagation of keypair renewal: https://gitlab.com/MatrixAI/Engineering/Polykey/js-polykey/-/merge_requests/209#note_676525054

Tasks

1. Find all places where keypair renewal/resetting must be propagated to
2. Expose a property in KeyManager that allows events to be propagated to dependent classes
3. Register keypair renewal and resetting as such events
4. Ensure that errors are correctly handled asynchronously

[CMCDragonkai]

Some notes regarding the recent discussion about observer and event emitter.

The problem with using event emitter is that handlers become asynchronously asynchronous. Explained here: https://stackoverflow.com/questions/67961514/event-emitter-emit-in-sequence-or-in-parallel-and-behaviour-when-they-are-async

In order to simplify testing and also ensure that side-effects can be controlled, it's better for when the keypair changes that all observers also complete their side-effects.

However the loose coupling of event emitter is also desirable, especially by not propagating exceptions to the subject call site.

Therefore it's a good idea to create our own "Observer" system that combines properties of event emitter and observer pattern.

So basically we can create a Observer class that acts like an event emitter. The main difference is that during the emit method call, rather than just emitting the event, it looks for all the handlers and awaits them simultaneously using Promise.all.

Here is some pseudo code:

class Observer {
    handlers: Array<[() => Promise<void>, ((e: Error) => Promise<void>)?]> = [];
    registerTopic (topic, f, onFailure) {
        const handlers = this.handlers.get(topic);
        handlers.push([f, onFailure]);
    }
    async updateTopic(topic) {
        const handlers = this.handlers.get(topic);
        // Change this to Promise.all instead, so that they are all run simultaneously
        for (const [f, onFailure] of fs) {
            try {
                await f();
            } catch (e) {
                if (onFailure) {
                    await onFailure(e);
                } else {
                    await this.handleError(e);
                }
            }
        }
    }
    async handleError (e) {
        this.logger.error(e);
    }
}

This way, when say KeyManager updates on a topic after a key renewal or reset, they will call something like await observer.updateTopic('key', data);.

The data is made available to all the listeners on the key topic, and then we can be guaranteed that all side effects are executed when the key change has been done.

Exceptions that are thrown are however not thrown to the updateTopic, that's kept separate with custom error handlers associated on each handler.

Then this Observer can be constructed in PolykeyAgent which bootstraps all the handlers together and also addresses any exceptions.

The reason for not propagating these exceptions to the emission site, is that those exceptions are not part of KeyManager, it shouldn't need to know about them. The only thing we are doing here is ensuring that we get an asynchronous-synchronous workflow, or a pub/sub pattern that isn't eventually consistent. EC is nice when these are completely separate systems, but this is 1 single application and there are consistency expectations when one asks the application to change its keypair, so full EC is not desirable here.

This merges the 2 patterns together.

It looks like somebody has done something similar here:

• https://github.com/yanickrochon/promise-events
• https://github.com/neuronetio/eventorjs

Both are pretty old implementations and aren't in TS, so I think we build our own, and have a look at those for inspiration.

The Observer can follow the event emitter API a little bit, but I think we can keep it simple too.

[CMCDragonkai]

I've been testing out extending the EventEmitter to include an emitAsync method. I noticed this pattern being used in eventemitter2 library.

Anyway I found a bug in upstream node, which will be solved by this PR: nodejs/node#41414

So right now captureRejections option won't work inside ts-node due to the importing of repl. But it will work correctly once compiled into JS and executed with just node since we don't use the REPL anywhere. But with this bug, we would need to upgrade our Node version to the latest soon.

[CMCDragonkai]

I've prototyped a working "EventBus" that is an extension of EventEmitter that is already available in Nodejs. It has adds one additional method called emitAsync, that operates based on the emit logic already in the Node source code: https://github.com/nodejs/node/blob/master/lib/events.js

import { EventEmitter } from 'events';

class EventBus extends EventEmitter {

  protected kCapture: symbol;

  constructor (...args: ConstructorParameters<typeof EventEmitter>) {
    super(...args);
    // EventEmitter's captureRejections option is only accessible through a private symbol
    // Here we augment the construction and save it as a property
    const symbols = Object.getOwnPropertySymbols(this);
    this.kCapture = symbols[0];
  }

  public async emitAsync(event: string | symbol, ...args: Array<any>): Promise<boolean> {
    const listeners = this.rawListeners(event);
    if (listeners.length < 1) {
      return false;
    }
    let result;
    try {
      for (const listener of listeners) {
        result = listener.apply(this, args);
        await result;
      }
    } catch (e) {
      if (!this[this.kCapture]) {
        throw e;
      }
      // The capture rejections mechanism only applies to promises
      if (!(result instanceof Promise)) {
        throw e;
      }
      // Queues error handling asynchronously to avoid bubbling up rejections
      // This matches the behaviour of EventEmitter which uses `process.nextTick`
      queueMicrotask(() => {
        if (typeof this[EventEmitter.captureRejectionSymbol] === 'function') {
          this[EventEmitter.captureRejectionSymbol](e, event, ...args);
        } else {
          // Disable the capture rejections mechanism to avoid infinite loop
          const prev = this[this.kCapture];
          // If the error handler throws, it results in `uncaughtException`
          try {
            this[this.kCapture] = false;
            this.emit('error', e);
          } finally {
            this[this.kCapture] = prev;
          }
        }
      });
    }
    return true;
  }

}

It turns out that the behaviour of error handling is quite complex. Some notes:

1. Node's event emitter would be described as "synchronously synchronous". This is because emit synchronously iterates over all the handlers and executes them. This means any synchronous exceptions will interrupt the loop and emit will bubble up a synchronous exception. Furthermore it also runs in sequence of the registration. Which means it's not really parallel.
2. Nodejs documents that to make it asynchronous, one can add a setImmediate wrapper around the handlers, and this would make emit a sort of "synchronously asynchronous". This would make event emitter much closer to pub-sub, where errors don't interrupt the loop of running each handler.
3. The captureRejections mechanism ONLY applies to promises. Which means if errors are emitted synchronously, they are always just going to bubble up the emit. It only works if the handlers are returning promises. And if it is true, then promise rejections will instead be handled by the error handler.
4. In creating emitAsync, I decided to create a sort of "asynchronously synchronous" system. It follows the emit logic, in that it loops over the handlers, and executes them one at a time. It also awaits for each handler, which means it does not execute all handlers concurrently. Just like emit, any synchronous exception will be thrown and bubble up, however asynchronous exceptions will now be handled by the captureRejections mechanism.
5. In order to implement captureRejections mechanism, I had to look over the node source code and observe how they are handling errors, but I selected to use tools that were more portable like queueMicrotask over process.nextTick.
6. One major difference between emit and emitAsync is that when asynchronous errors occur in emit, they are only handled after all handlers have already executed. This is because emit is not promise-aware. But emitAsync is, so both synchronous and asynchronous exceptions will interrupt the handler loop.
7. An "asynchronously asynchronous" system would be interesting, and there are many more variants of emitAsync that could be done, one which does proper broadcasting using Promise.allSettled or even cascading. But this goes into more complicating messaging system architecture, which we can extend later in the future.

For now we can use EventBus as our application-global event bus across the PK codebase. Won't create a separate package for it yet, will keep it just in js-polykey's codebase for now.

[CMCDragonkai]

Example usage:

async function main () {

  const e = new EventBus({
    captureRejections: true
  });

  e.on('error', (e) => {
    console.log('ERROR HANDLED', e);
  });

  e.on('sync', async () => {
    console.log('1');
  });
  e.on('sync', async () => {
    throw new Error('sync error');
  });
  e.on('sync', async () => {
    console.log(2);
  });

  console.log('EMITTING');

  const result = await e.emitAsync('sync');

  console.log('result', result);

}

main();

Note that this must be run like this tsc ./test-events.ts && node ./test-events.js. The ts-node cannot show this behaviour properly with captureRejections. This is because of an upstream bug in NodeJS: nodejs/node#41414

However it should work fine in our jest testing.

[CMCDragonkai]

@emmacasolin Some valid tests for EventBus would be dealing with multiple async handlers or a mix of sync and async handlers... etc. Compare it with emit to understand how it should be working.

[CMCDragonkai]

Also please write down the plan regarding this usage, and what handlers you'll be registering and whether they will be asynchronous, and how this should impact testing...

Note that getting the node id is still "pull-based" so that isn't affected by this new "push-based" mechanism.

I can imagine the ability to push events as a stream to the GUI now as things are updated.

[emmacasolin]

Plan for using the EventBus class for propagating root key pair changes

The areas that need to be updated when the keypair is changed are:

• Status (nodeId)
• NodeGraph (refresh buckets)
• ForwardProxy (TLS config)
• ReverseProxy (TLS config)
• GRPCServerClient (TLS config)

nodeGraph.refreshBuckets() doesn't require any data about the key pair change, so the only data that needs to be emitted is the new node id, root key pair pem (private key), and root cert chain pem. The key pair pem and cert chain pem could just be bundled into a TLSConfig type, so there really only needs to be two fields:

type rootKeyPairChangeData = {
  nodeId: NodeId;
  tlsConfig: TLSConfig;
};

As for the topic string for the event, this can be a symbol, for example:

const rootKeyPairChangeSymbol = Symbol("rootKeyPairChange");

In terms of the handlers that would need to be registered to this event, the handlers for updating the TLSConfigs can be synchronous, however the ones for updating the Status file and refreshing node buckets must be asynchronous due to the underlying operations that would need to be called. Ideally it would be best for event emission and registration to be coordinated by the Polykey Agent, so I'll see if I can prototype a way of doing that.

[CMCDragonkai]

I was thinking that all event emission and registration/handling can be done in src/PolykeyAgent.ts.

Instead domains expose "hookpoints" via callbacks. For example during the creation of KeyManager, there may be a parameter: rootKeyPairChange(), which is an async callback (rootKeyPair) => Promise<void>.

Then KeyManager calls and awaits after the keypair change is done inside the relevant functions.

Now in PolykeyAgent, it can inject a callback like this:

// get this exported out of the `events` index.ts` as well
import { EventBus, captureRejectionSymbol } from '../events';

const events = new EventBus({ captureRejections: true });

events[Symbol.for('nodejs.rejection')] = (err, event, ...args) => {
  // log out the error here
};

const keyManager = await KeyManager.createKeyManager({
  ...,
  rootKeyPairChange: (keyPair) => {
    await events.emitAsync(keyUtils.eventRootKeyPairChange, keyPair);
  }
}

events.on(keyUtils.eventRootKeyPairChange, async (keyPair) => {
  await domain1.update(...);
  await domain2.update(...);
});

Then subsequently register handlers for each event too.

This keeps all of the event logic in PolykeyAgent.ts, and domains only need to expose hook points or, public methods that need to be triggered.

One might ask if we are using hookpoints, why use the eventbus at all, just dispense with it and just call the updating functions directly in the hook point. This is a good point. One of the reasons to use eventbus is for "decentralised" event logic, where each domain emits events and then listens for events from other domains. This is best used when the domains are separated packages, or software components that are managed by separate people/teams. Whereas in PK, it's one single application codebase. So here we could just do direct calls, but we create an eventbus in case in the future, we are separating into separate components. So it's just a bit of future proofing.

---

Note that you should switch on captureRejections, however I believe each handler should handle their own exceptions and avoid bubbling it to the root error handler. Make sure to use this https://nodejs.org/api/events.html#emittersymbolfornodejsrejectionerr-eventname-args. The only you can do there is to log out the error there.

[CMCDragonkai]

@tegefaulkes any thoughts on how nodegraph should be updated in relation to the root keypair changing based on this comment #312 (comment)

[tegefaulkes]

From what I can tell, you just need to call refreshBuckets() after the NodeId has been updated in the KeyManager.

[emmacasolin]

Doing a quick test in a test.ts is showing that this is looking promising! I'm creating the EventBus inside createPolykeyAgent(), using it to generate the callback that gets injected into KeyManager, and then setting events as a property in the constructor for PolykeyAgent. Then inside PolykeyAgent.start() I have this:

this.events.on(keysUtils.eventRootKeyPairChange, async (keyChangeData: RootKeyPairChangeData) => {
  await this.status.updateNodeId(keyChangeData.nodeId);
  await this.nodeManager.refreshBuckets();
  this.fwdProxy.setTLSConfig(keyChangeData.tlsConfig);
  this.revProxy.setTLSConfig(keyChangeData.tlsConfig);
  this.grpcServerClient.setTLSConfig(keyChangeData.tlsConfig);
});

And then this is what gets logged in the terminal when renewing the key pair in a test:

{
  status: 'LIVE',
  data: {
    pid: 1716087,
    nodeId: 'vmcb55dv2853bd52sfhs64jlpp76gtms4v4ifmels9c3n9aksbll0',
    clientHost: '127.0.0.1',
    clientPort: 38431,
    ingressHost: '0.0.0.0',
    ingressPort: 35458
  }
}
INFO:KeyManager:Renewing root key pair
INFO:KeyManager:Reading /home/emma/.local/share/polykey/state/keys/db.key
INFO:KeyManager:Reading /home/emma/.local/share/polykey/state/keys/vault.key
INFO:KeyManager:Copying old root key pair to /home/emma/.local/share/polykey/state/keys/root_certs/root.crt.1642388046
INFO:KeyManager:Performing garbage collection of root certificates
INFO:KeyManager:Writing /home/emma/.local/share/polykey/state/keys/root.pub and /home/emma/.local/share/polykey/state/keys/root.key
INFO:KeyManager:Writing /home/emma/.local/share/polykey/state/keys/root.crt
INFO:KeyManager:Writing /home/emma/.local/share/polykey/state/keys/db.key
INFO:KeyManager:Writing /home/emma/.local/share/polykey/state/keys/vault.key
INFO:Status:Updating Node Id
INFO:Status:Writing Status file to /home/emma/.local/share/polykey/status.json
INFO:NodeManager:Refreshing buckets
INFO:ForwardProxy:Updating ForwardProxy TLS Config
INFO:ReverseProxy:Updating ReverseProxy TLS Config
INFO:GRPCServerClient:Updating GRPCServer TLS Config
{
  status: 'LIVE',
  data: {
    pid: 1716087,
    nodeId: 'vio5621rbf2u5svcb5ddldcpjjt96bn3862pa2q2ef8khcde35q5g',
    clientHost: '127.0.0.1',
    clientPort: 38431,
    ingressHost: '0.0.0.0',
    ingressPort: 35458
  }
}

I put logger.info()'s in each of the functions as a guide for the order that things are being called in, and it matches up with the order inside the event handler. There was already a logger message inside GRPCServer.setTLSConfig() so not sure if I should keep the other logs as well for cohesion.

I copied this example #312 (comment) for the most part, but the only thing I'm unsure about is what this line is for and what I should be putting inside it:

events[Symbol.for('nodejs.rejection')] = (err, event, ...args) => {
  // log out the error here
};

Will now start writing some proper tests to check that the event bus is working as expected.

[emmacasolin]

Implementation and tests have now been pushed up here: c394331. I've written tests for the EventBus class and I've tested keypair changing in a test.ts, however I still need to ammend the existing tests in the codebase to check that they're all passing. The only changes needed should be making sure that the KeyManager is constructed correctly (including the rootKeyChange callback) and removing the previous setup that was needed for the Observer Pattern implementation.

[CMCDragonkai]

Nice work!

I suggest adding logger info messages at the beginning and end of your event handler, so we can know that those updating messages are grouped together.

As for the error handler, all you need to do is log out the error. That's the only thing you can do.

The err will be an exception with exception message properties. Have a look at our root exception handler to see how it should be done.

It's also possible we don't bother doing that if those errors bubble up to root exception handlers. This requires some testing.

In our src/bin/utils/ExitHandlers.ts we already register global process handlers for unhandled rejections:

    // Both synchronous and asynchronous errors are handled
    process.once('unhandledRejection', this.errorHandler);
    process.once('uncaughtException', this.errorHandler);

This means even if you don't captureRejections that should mean those exceptions will bubble up to the ExitHandlers handlers anyway.

---

Oh I just realised that the errorHandler inside ExitHandlers will exit the process if this occurs. If we think that any unhandled error going into the eventbus is a fatal error, then that should happen. However as I said, all of our handlers in the event bus should catch relevant exceptions.

For example:

this.events.on(keysUtils.eventRootKeyPairChange, async (keyChangeData: RootKeyPairChangeData) => {
  try {
    await this.status.updateNodeId(keyChangeData.nodeId);
    await this.nodeManager.refreshBuckets();
    this.fwdProxy.setTLSConfig(keyChangeData.tlsConfig);
    this.revProxy.setTLSConfig(keyChangeData.tlsConfig);
    this.grpcServerClient.setTLSConfig(keyChangeData.tlsConfig);
  } catch (e) {
    // catch relevant exceptions
    // if they are not "relevant", then throw it upwards
  }
});

So this requires some planning:

1. Domain function error handling is only for when the domain itself can handle the error and recover.
2. Domain functions will throw it up to the caller, hence the PolykeyAgent can catch the error and recover.
3. If PolykeyAgent cannot recover, then it bubbles up again, this time to the EventBus's captureRejections mechanism. But EventBus cannot recover... so it just has to log it out, and bubble it up again.
4. This would eventually hit the ExitHandlers error handling which prints out the error formatted to STDERR and does a process.exit() recognising it as a "fatal error".

[CMCDragonkai]

From what I can tell, you just need to call refreshBuckets() after the NodeId has been updated in the KeyManager.

From memory, other nodes will need to use the new node Id eventually. However this is done in a "lazy way". Which is that old node ids are still valid, but when the establish a connection, they are meant to be updated with the new node id if it has been changed.

There are some cases where we need to push out to everybody, that may include all the nodes that are part of the gestalt, or in the DHT, we need to propagate updates, other node's node graph are now out of date. This is where #190 may be relevant. Just something to think about, I don't think we have fully specced out this behaviour yet.

[emmacasolin]

At the moment, event handlers are being registered in PolykeyAgent's start method, so would this mean that if an error was thrown during this process and wasn't caught that it would be caught by start's catch block? Resulting in the Failed Starting PolykeyAgent logger message and every domain being stopped?

[CMCDragonkai]

If there's an error in start, all event handlers should be explicitly deregistered.

You could do this with removeAllListeners.

[CMCDragonkai]

Make sure to do that when stop is called too.

Otherwise event listeners would be duplicated.

[emmacasolin]

I'm not sure what would be considered a relevant error for keypair propagation, since any error caught by the event handler will have already halted the propagation to domains. In this case would it be better to just do something like this.logger.warn('Keypair could not be propagated to all domains'); after catching the error (and then continue) or should the error be thrown again, e.g.

this.events.on(
  keysUtils.eventRootKeyPairChange,
    async (keyChangeData: RootKeyPairChangeData) => {
    try {
      this.logger.info('Propagating root keypair change');
      await this.status.updateNodeId(keyChangeData.nodeId);
      await this.nodeManager.refreshBuckets();
      this.fwdProxy.setTLSConfig(keyChangeData.tlsConfig);
      this.revProxy.setTLSConfig(keyChangeData.tlsConfig);
      this.grpcServerClient.setTLSConfig(keyChangeData.tlsConfig);
      this.logger.info('Propagated root keypair change');
    } catch (e) {
      this.logger.warn('Keypair could not be propagated to all domains');
      throw new errors.ErrorKeyPairPropagate(e.message, {
        errno: e.errno,
        syscall: e.syscall,
        code: e.code,
        path: e.path,
      });
    }
  },
);

[CMCDragonkai]

I think it needs to be done bottom up first. You want to catch exceptions for each method that you're calling like await this.nodeManager.refreshBuckets. Those should be catching exceptions if they can recover.

Remember this order:

1. Domain function error handling is only for when the domain itself can handle the error and recover.
2. Domain functions will throw it up to the caller, hence the PolykeyAgent can catch the error and recover.
3. If PolykeyAgent cannot recover, then it bubbles up again, this time to the EventBus's captureRejections mechanism. But EventBus cannot recover... so it just has to log it out, and bubble it up again.
4. This would eventually hit the ExitHandlers error handling which prints out the error formatted to STDERR and does a process.exit() recognising it as a "fatal error".

If PolykeyAgent cannot recover from the error, don't bother catching it on eventRootKeyPairChange event.

Your eventbus error handler should instead use this.logger.error and report the exact exception from the particular event type. Then rethrow the exception up so that it gets caught by exit handlers.

[CMCDragonkai]

// use the captureRejectionsSymbol instead of Symbol.for
this.events[captureRejectionSymbol] = (err, event, ...args) => {
  this.logger.error(...);
  throw err;
}

this.events.on(
  keysUtils.eventRootKeyPairChange,
  async () => { ... }
);

[emmacasolin]

Attempting to catch errors this way is working fine in tests (i.e. EventBus.test.ts), however it's not working inside the PolykeyAgent... I'm wondering if it could be because the EventBus isn't being injected into KeyManger (just a callback)? I'll see if injecting it makes a difference

this.events[captureRejectionSymbol] = (err, event, ...args) => {
  this.logger.error(...);
  throw err;
}

[CMCDragonkai]

No the reason is that there's an upstream bug in NodeJS that prevents captureRejections from working. The ts-node adds an extra library that affects the captureRejections thing. It will work however once you compile with tsc and run with nodejs directly. So when you say it's not working in PolykeyAgent, that's probably because you're using ts-node right? Jest doesn't use ts-node, so I should be fine. #312 (comment)

[emmacasolin]

Yeah I just remembered that and figured it was probably the issue. Will do some more tests to make sure though

[CMCDragonkai]

Changing the node id will impact communications between:

• Agent to Agent - this is mTLS
• CLI to Agent - normal TLS (client does not provide cert)
• GUI to Agent - normal TLS (client does not provide cert)

Under Agent to Agent, this is going through the forward and reverse proxies, and also the GRPCClientAgent. The GRPC client does not use TLS but the TLS is being done between forward and reverse proxies. This means that when we perform a keyManager.renewRootKeyPair, the old NodeId is still valid until the certificate expiration date. This means nodes contacting the changing agent with the old NodeId should still work, but receive an automatic update to the NodeGraph when the node id changes. I believe the logic to check the cert chain is in network/utils.ts verifyServerCertificateChain, but the logic to update the node id in the nodegraph is not verified yet.

Under CLI to Agent and GUI to Agent, they are using direct TLS on the GRPCClient. Existing GRPCClient connections should continue to work and not be interrupted by the TLS change. I believe I had tests for this under tests/grpc/GRPCServer.test.ts as changing the private key and certificate on the fly. However new connections will need to be made with a new target Node ID. This can occur by reading the status each time a new connection is made. However I think a similar TLS verification logic could take place where the old node ID can still be valid.

See that the verifyServerCertificateChain is used for both proxies and direct TLS:

[nix-shell:~/Projects/js-polykey/src]$ ag 'verifyServerCertificateChain'
network/utils.ts
181:function verifyServerCertificateChain(
370:  verifyServerCertificateChain,

network/ConnectionForward.ts
191:      networkUtils.verifyServerCertificateChain(this.nodeId, serverCertChain);

grpc/GRPCClient.ts
135:        networkUtils.verifyServerCertificateChain(nodeId, serverCertChain);

So we need tests to ensure this is the case. One for tests/grpc/GRPCClient.test.ts to ensure that the GRPCClient node id can still work. Another for tests/network as well.

Then we would extend these tests into tests/nodes for NodeConnection, and also in tests/bin to test CLI situation.

---

I remember a prior issue where I wrote that there 3 modes of connecting to the agent. Details:

• Implementing TLS for ClientService #229 (comment)
• Implementing TLS for ClientService #229 (comment)

This means there was another mode where during CLI to agent, the client provides a cert as well... but we scrapped that idea simply because the client may not be on the same host as the agent, and therefore not have access to the certificate files. That's why we built the session authentication logic.

[CMCDragonkai]

This has been implemented here: fc0e83b.

However a new issue should be created to address the "Test communications when root key pair is renewed and NodeID is changed".

@emmacasolin please create new issue for #311 (comment) and the above comment, and then close this issue, while referencing this issue in the new issue for additional context.

[emmacasolin]

Closing this issue. Comms testing to be tracked here: #317