wip: client handler usage examples

Related #500
Related #501

[ci skip]

src/RPC/RPCServer.ts

@@ -13,11 +13,7 @@ import type { ReadableWritablePair } from 'stream/web';
 import type { JSONValue, POJO } from '../types';
 import type { ConnectionInfo } from '../network/types';
 import type { RPCErrorEvent } from './utils';
-import type {
-  MiddlewareFactory,
-  MiddlewareShort,
-  Middleware,
-} from 'tokens/types';
+import type { MiddlewareFactory, MiddlewareShort, Middleware } from './types';
 import { ReadableStream } from 'stream/web';
 import { CreateDestroy, ready } from '@matrixai/async-init/dist/CreateDestroy';
 import Logger from '@matrixai/logger';

src/clientRPC/handlers/agentStatus.ts

@@ -0,0 +1,49 @@
+import type { UnaryHandler } from '../../RPC/types';
+import type KeyRing from '../../keys/KeyRing';
+import type CertManager from '../../keys/CertManager';
+import type Logger from '@matrixai/logger';
+import type { NodeIdEncoded } from '../../ids';
+import type RPCClient from '../../RPC/RPCClient';
+import type { POJO } from '../../types';
+import * as nodesUtils from '../../nodes/utils';
+import * as keysUtils from '../../keys/utils';
+
+type StatusResult = {
+  pid: number;
+  nodeId: NodeIdEncoded;
+  publicJwk: string;
+};
+const agentStatusName = 'agentStatus';
+const agentStatusHandler: UnaryHandler<null, StatusResult> = async (
+  input,
+  container: {
+    keyRing: KeyRing;
+    certManager: CertManager;
+    logger: Logger;
+  },
+  _connectionInfo,
+  _ctx,
+) => {
+  return {
+    pid: process.pid,
+    nodeId: nodesUtils.encodeNodeId(container.keyRing.getNodeId()),
+    publicJwk: JSON.stringify(
+      keysUtils.publicKeyToJWK(container.keyRing.keyPair.publicKey),
+    ),
+  };
+};
+
+const agentStatusCaller = async (metadata: POJO, rpcClient: RPCClient) => {
+  const result = await rpcClient.unaryCaller<null, StatusResult>(
+    agentStatusName,
+    null,
+    metadata,
+  );
+  return {
+    pid: result.pid,
+    nodeId: nodesUtils.decodeNodeId(result.nodeId),
+    publicJwk: result.publicJwk,
+  };
+};
+
+export { agentStatusName, agentStatusHandler, agentStatusCaller };

src/clientRPC/handlers/agentUnlock.ts

@@ -0,0 +1,24 @@
+import type { UnaryHandler } from '../../RPC/types';
+import type Logger from '@matrixai/logger';
+import type RPCClient from '../../RPC/RPCClient';
+import type { POJO } from '../../types';
+
+const agentUnlockName = 'agentStatus';
+const agentUnlockHandler: UnaryHandler<null, null> = async (
+  _input,
+  _container: {
+    logger: Logger;
+  },
+  _connectionInfo,
+  _ctx,
+) => {
+  // This is a NOP handler,
+  // authentication and unlocking is handled via middleware
+  return null;
+};
+
+const agentUnlockCaller = async (metadata: POJO, rpcClient: RPCClient) => {
+  await rpcClient.unaryCaller<null, null>(agentUnlockName, null, metadata);
+};
+
+export { agentUnlockName, agentUnlockHandler, agentUnlockCaller };

src/clientRPC/utils.ts

@@ -0,0 +1,58 @@
+import type { SessionToken } from '../sessions/types';
+import type KeyRing from '../keys/KeyRing';
+import type SessionManager from '../sessions/SessionManager';
+import type { Authenticate } from '../client/types';
+import * as grpc from '@grpc/grpc-js';
+import * as clientErrors from '../client/errors';
+
+/**
+ * Encodes an Authorization header from session token
+ * Assumes token is already encoded
+ * Will mutate metadata if it is passed in
+ */
+function encodeAuthFromSession(
+  token: SessionToken,
+  metadata: grpc.Metadata = new grpc.Metadata(),
+): grpc.Metadata {
+  metadata.set('Authorization', `Bearer ${token}`);
+  return metadata;
+}
+
+function authenticator(
+  sessionManager: SessionManager,
+  keyRing: KeyRing,
+): Authenticate {
+  return async (
+    forwardMetadata: grpc.Metadata,
+    reverseMetadata: grpc.Metadata = new grpc.Metadata(),
+  ) => {
+    const auth = forwardMetadata.get('Authorization')[0] as string | undefined;
+    if (auth == null) {
+      throw new clientErrors.ErrorClientAuthMissing();
+    }
+    if (auth.startsWith('Bearer ')) {
+      const token = auth.substring(7) as SessionToken;
+      if (!(await sessionManager.verifyToken(token))) {
+        throw new clientErrors.ErrorClientAuthDenied();
+      }
+    } else if (auth.startsWith('Basic ')) {
+      const encoded = auth.substring(6);
+      const decoded = Buffer.from(encoded, 'base64').toString('utf-8');
+      const match = decoded.match(/:(.*)/);
+      if (match == null) {
+        throw new clientErrors.ErrorClientAuthFormat();
+      }
+      const password = match[1];
+      if (!(await keyRing.checkPassword(password))) {
+        throw new clientErrors.ErrorClientAuthDenied();
+      }
+    } else {
+      throw new clientErrors.ErrorClientAuthMissing();
+    }
+    const token = await sessionManager.createToken();
+    encodeAuthFromSession(token, reverseMetadata);
+    return reverseMetadata;
+  };
+}
+
+export { authenticator };

tests/clientRPC/handlers/agentStatus.test.ts

@@ -0,0 +1,97 @@
+import type { ConnectionInfo } from '@/network/types';
+import fs from 'fs';
+import path from 'path';
+import os from 'os';
+import Logger, { LogLevel, StreamHandler } from '@matrixai/logger';
+import { DB } from '@matrixai/db';
+import KeyRing from '@/keys/KeyRing';
+import * as keysUtils from '@/keys/utils';
+import RPCServer from '@/RPC/RPCServer';
+import TaskManager from '@/tasks/TaskManager';
+import CertManager from '@/keys/CertManager';
+import {
+  agentStatusName,
+  agentStatusHandler,
+  agentStatusCaller,
+} from '@/clientRPC/handlers/agentStatus';
+import RPCClient from '@/RPC/RPCClient';
+import * as rpcTestUtils from '../../RPC/utils';
+
+describe('agentStatus', () => {
+  const logger = new Logger('agentStatus test', LogLevel.WARN, [
+    new StreamHandler(),
+  ]);
+  const password = 'helloworld';
+  let dataDir: string;
+  let db: DB;
+  let keyRing: KeyRing;
+  let taskManager: TaskManager;
+  let certManager: CertManager;
+
+  beforeEach(async () => {
+    dataDir = await fs.promises.mkdtemp(
+      path.join(os.tmpdir(), 'polykey-test-'),
+    );
+    const keysPath = path.join(dataDir, 'keys');
+    const dbPath = path.join(dataDir, 'db');
+    db = await DB.createDB({
+      dbPath,
+      logger,
+    });
+    keyRing = await KeyRing.createKeyRing({
+      password,
+      keysPath,
+      logger,
+      passwordOpsLimit: keysUtils.passwordOpsLimits.min,
+      passwordMemLimit: keysUtils.passwordMemLimits.min,
+      strictMemoryLock: false,
+    });
+    taskManager = await TaskManager.createTaskManager({ db, logger });
+    certManager = await CertManager.createCertManager({
+      db,
+      keyRing,
+      taskManager,
+      logger,
+    });
+  });
+  afterEach(async () => {
+    await certManager.stop();
+    await taskManager.stop();
+    await keyRing.stop();
+    await db.stop();
+    await fs.promises.rm(dataDir, {
+      force: true,
+      recursive: true,
+    });
+  });
+  test('get status', async () => {
+    // Setup
+    const rpcServer = await RPCServer.createRPCServer({
+      container: {
+        // KeyRing,
+        // certManager,
+        logger,
+      },
+      logger,
+    });
+    rpcServer.registerUnaryHandler(agentStatusName, agentStatusHandler);
+    const rpcClient = await RPCClient.createRPCClient({
+      streamPairCreateCallback: async () => {
+        const { clientPair, serverPair } = rpcTestUtils.createTapPairs();
+        rpcServer.handleStream(serverPair, {} as ConnectionInfo);
+        return clientPair;
+      },
+      logger,
+    });
+
+    // Doing the test
+    const result = await agentStatusCaller({}, rpcClient);
+    expect(result).toStrictEqual({
+      pid: process.pid,
+      nodeId: keyRing.getNodeId(),
+      publicJwk: JSON.stringify(
+        keysUtils.publicKeyToJWK(keyRing.keyPair.publicKey),
+      ),
+    });
+  });
+});

tests/clientRPC/handlers/agentUnlock.test.ts

@@ -0,0 +1,110 @@
+import type { ConnectionInfo } from '@/network/types';
+import fs from 'fs';
+import path from 'path';
+import os from 'os';
+import { TransformStream } from 'stream/web';
+import Logger, { LogLevel, StreamHandler } from '@matrixai/logger';
+import { DB } from '@matrixai/db';
+import KeyRing from '@/keys/KeyRing';
+import * as keysUtils from '@/keys/utils';
+import RPCServer from '@/RPC/RPCServer';
+import TaskManager from '@/tasks/TaskManager';
+import CertManager from '@/keys/CertManager';
+import {
+  agentUnlockName,
+  agentUnlockHandler,
+  agentUnlockCaller,
+} from '@/clientRPC/handlers/agentUnlock';
+import RPCClient from '@/RPC/RPCClient';
+import * as rpcTestUtils from '../../RPC/utils';
+
+describe('agentStatus', () => {
+  const logger = new Logger('agentStatus test', LogLevel.WARN, [
+    new StreamHandler(),
+  ]);
+  const password = 'helloworld';
+  let dataDir: string;
+  let db: DB;
+  let keyRing: KeyRing;
+  let taskManager: TaskManager;
+  let certManager: CertManager;
+
+  beforeEach(async () => {
+    dataDir = await fs.promises.mkdtemp(
+      path.join(os.tmpdir(), 'polykey-test-'),
+    );
+    const keysPath = path.join(dataDir, 'keys');
+    const dbPath = path.join(dataDir, 'db');
+    db = await DB.createDB({
+      dbPath,
+      logger,
+    });
+    keyRing = await KeyRing.createKeyRing({
+      password,
+      keysPath,
+      logger,
+      passwordOpsLimit: keysUtils.passwordOpsLimits.min,
+      passwordMemLimit: keysUtils.passwordMemLimits.min,
+      strictMemoryLock: false,
+    });
+    taskManager = await TaskManager.createTaskManager({ db, logger });
+    certManager = await CertManager.createCertManager({
+      db,
+      keyRing,
+      taskManager,
+      logger,
+    });
+  });
+  afterEach(async () => {
+    await certManager.stop();
+    await taskManager.stop();
+    await keyRing.stop();
+    await db.stop();
+    await fs.promises.rm(dataDir, {
+      force: true,
+      recursive: true,
+    });
+  });
+  test('get status', async () => {
+    // Setup
+    const rpcServer = await RPCServer.createRPCServer({
+      container: {
+        // KeyRing,
+        // certManager,
+        logger,
+      },
+      logger,
+    });
+    rpcServer.registerUnaryHandler(agentUnlockName, agentUnlockHandler);
+    rpcServer.registerForwardMiddleware(() => {
+      return (input) => {
+        // This middleware needs to check the first message for the token
+        return input.pipeThrough(
+          new TransformStream({
+            transform: (chunk, controller) => {
+              controller.enqueue(chunk);
+            },
+          }),
+        );
+      };
+    });
+    const rpcClient = await RPCClient.createRPCClient({
+      streamPairCreateCallback: async () => {
+        const { clientPair, serverPair } = rpcTestUtils.createTapPairs();
+        rpcServer.handleStream(serverPair, {} as ConnectionInfo);
+        return clientPair;
+      },
+      logger,
+    });
+
+    // Doing the test
+    const result = await agentUnlockCaller({}, rpcClient);
+    expect(result).toStrictEqual({
+      pid: process.pid,
+      nodeId: keyRing.getNodeId(),
+      publicJwk: JSON.stringify(
+        keysUtils.publicKeyToJWK(keyRing.keyPair.publicKey),
+      ),
+    });
+  });
+});