Merge pull request #870 from MatrixAI/feature-eng-527-adding-agent-rpc-handlers-for-getting-sigchain-and-audit-data

CDeltakai · web-flow · commit bcad31fcdbd5 · 2025-03-07T14:31:36.000+11:00
feature/ENG-527 Adding agent RPC handlers for getting sigchain and audit data.
diff --git a/src/PolykeyAgent.ts b/src/PolykeyAgent.ts
@@ -804,6 +804,7 @@ class PolykeyAgent {
         port: optionsDefaulted.agentServicePort,
         ipv6Only: optionsDefaulted.ipv6Only,
         agentService: agentServerManifest({
+          audit: this.audit,
           acl: this.acl,
           db: this.db,
           keyRing: this.keyRing,
diff --git a/src/nodes/NodeManager.ts b/src/nodes/NodeManager.ts
@@ -17,7 +17,7 @@ import type { Host, Port } from '../network/types';
 import type {
   Claim,
   ClaimId,
-  ClaimIdEncoded,
+  // ClaimIdEncoded,
   SignedClaim,
 } from '../claims/types';
 import type { ClaimLinkNode } from '../claims/payloads';
@@ -1257,12 +1257,19 @@ class NodeManager {
     return await this.withConnF(targetNodeId, ctx, async (connection) => {
       const claims: Record<ClaimId, SignedClaim> = {};
       const client = connection.getClient();
+
+      // Let claimIdEncoded: ClaimIdEncoded | undefined;
+
+      // if (claimId != null) {
+      //   claimIdEncoded = claimsUtils.encodeClaimId(claimId);
+      // } else {
+      //   claimIdEncoded = undefined;
+      // }
+
       for await (const agentClaim of await client.methods.nodesClaimsGet(
         {
-          claimIdEncoded:
-            claimId != null
-              ? claimsUtils.encodeClaimId(claimId)
-              : ('' as ClaimIdEncoded),
+          // Needs to be addressed later - causes test failures in Discovery.test.ts
+          // seek: claimIdEncoded,
         },
         ctx,
       )) {
diff --git a/src/nodes/agent/callers/index.ts b/src/nodes/agent/callers/index.ts
@@ -1,4 +1,5 @@
 import nodesAuthenticateConnection from './nodesAuthenticateConnection';
+import nodesAuditEventsGet from './nodesAuditEventsGet';
 import nodesClaimsGet from './nodesClaimsGet';
 import nodesClosestActiveConnectionsGet from './nodesClosestActiveConnectionsGet';
 import nodesClosestLocalNodesGet from './nodesClosestLocalNodesGet';
@@ -17,6 +18,7 @@ import vaultsScan from './vaultsScan';
  */
 const manifestClient = {
   nodesAuthenticateConnection,
+  nodesAuditEventsGet,
   nodesClaimsGet,
   nodesClosestActiveConnectionsGet,
   nodesClosestLocalNodesGet,
@@ -37,6 +39,7 @@ export default manifestClient;
 
 export {
   nodesAuthenticateConnection,
+  nodesAuditEventsGet,
   nodesClaimsGet,
   nodesClosestActiveConnectionsGet,
   nodesClosestLocalNodesGet,
diff --git a/src/nodes/agent/callers/nodesAuditEventsGet.ts b/src/nodes/agent/callers/nodesAuditEventsGet.ts
@@ -0,0 +1,12 @@
+import type { HandlerTypes } from '@matrixai/rpc';
+import type NodesAuditEventsGet from '../handlers/NodesAuditEventsGet';
+import { ServerCaller } from '@matrixai/rpc';
+
+type CallerTypes = HandlerTypes<NodesAuditEventsGet>;
+
+const nodesAuditEventsGet = new ServerCaller<
+  CallerTypes['input'],
+  CallerTypes['output']
+>();
+
+export default nodesAuditEventsGet;
diff --git a/src/nodes/agent/handlers/NodesAuditEventsGet.ts b/src/nodes/agent/handlers/NodesAuditEventsGet.ts
@@ -0,0 +1,70 @@
+import type { ContextTimed } from '@matrixai/contexts';
+import type { DB } from '@matrixai/db';
+import type { JSONValue } from '@matrixai/rpc';
+import type {
+  AgentRPCRequestParams,
+  AgentRPCResponseResult,
+  AuditIdMessage,
+  AgentAuditMessage,
+} from '../types';
+import type Audit from '../../../audit/Audit';
+import type { AuditEvent } from '../../../audit/types';
+import type { AuditEventId } from '../../../ids';
+import { ServerHandler } from '@matrixai/rpc';
+import * as auditUtils from '../../../audit/utils';
+
+/**
+ * Gets audit events from a node
+ */
+class NodesAuditEventsGet extends ServerHandler<
+  {
+    audit: Audit;
+    db: DB;
+  },
+  AgentRPCRequestParams<AuditIdMessage>,
+  AgentRPCResponseResult<AgentAuditMessage<AuditEvent>>
+> {
+  public handle = async function* (
+    input: AgentRPCRequestParams<AuditIdMessage>,
+    _cancel: (reason?: any) => void,
+    _meta: Record<string, JSONValue> | undefined,
+    ctx: ContextTimed,
+  ): AsyncGenerator<AgentRPCResponseResult<AgentAuditMessage<AuditEvent>>> {
+    let seek_: AuditEventId | number | undefined;
+    let seekEnd_: AuditEventId | number | undefined;
+
+    const { seek, seekEnd, limit } = input;
+
+    if (typeof seek !== 'number') {
+      seek_ = auditUtils.decodeAuditEventId(seek);
+    }
+    if (typeof seekEnd !== 'number') {
+      seekEnd_ = auditUtils.decodeAuditEventId(seekEnd);
+    }
+
+    const { audit, db }: { audit: Audit; db: DB } = this.container;
+
+    yield* db.withTransactionG(async function* (tran): AsyncGenerator<
+      AgentRPCResponseResult<AgentAuditMessage<AuditEvent>>
+    > {
+      for await (const auditEvent of audit.getAuditEvents(
+        [],
+        {
+          seek: seek_,
+          seekEnd: seekEnd_,
+          limit,
+        },
+        tran,
+      )) {
+        ctx.signal.throwIfAborted();
+        yield {
+          id: auditUtils.encodeAuditEventId(auditEvent.id),
+          path: auditEvent.path,
+          data: auditEvent.data,
+        };
+      }
+    });
+  };
+}
+
+export default NodesAuditEventsGet;
diff --git a/src/nodes/agent/handlers/NodesClaimsGet.ts b/src/nodes/agent/handlers/NodesClaimsGet.ts
@@ -5,11 +5,12 @@ import type Sigchain from '../../../sigchain/Sigchain';
 import type {
   AgentRPCRequestParams,
   AgentRPCResponseResult,
-  ClaimIdMessage,
   AgentClaimMessage,
+  NodesClaimsGetMessage,
 } from '../types';
 import { ServerHandler } from '@matrixai/rpc';
 import * as claimsUtils from '../../../claims/utils';
+import * as ids from '../../../ids';
 
 /**
  * Gets the sigchain claims of a node
@@ -19,21 +20,32 @@ class NodesClaimsGet extends ServerHandler<
     sigchain: Sigchain;
     db: DB;
   },
-  AgentRPCRequestParams<ClaimIdMessage>,
+  AgentRPCRequestParams<NodesClaimsGetMessage>,
   AgentRPCResponseResult<AgentClaimMessage>
 > {
   public handle = async function* (
-    _input: ClaimIdMessage,
+    input: NodesClaimsGetMessage,
     _cancel: (reason?: any) => void,
     _meta: Record<string, JSONValue> | undefined,
     ctx: ContextTimed,
   ): AsyncGenerator<AgentRPCResponseResult<AgentClaimMessage>> {
+    const { seek, order, limit } = input;
     const { sigchain, db }: { sigchain: Sigchain; db: DB } = this.container;
+
+    let decodedClaimId = ids.decodeClaimId(seek);
+    if (decodedClaimId == null) {
+      decodedClaimId = undefined;
+    }
+
     yield* db.withTransactionG(async function* (tran): AsyncGenerator<
       AgentRPCResponseResult<AgentClaimMessage>
     > {
       for await (const [claimId, signedClaim] of sigchain.getSignedClaims(
-        { order: 'asc' },
+        {
+          seek: decodedClaimId,
+          order: order,
+          limit: limit,
+        },
         tran,
       )) {
         ctx.signal.throwIfAborted();
diff --git a/src/nodes/agent/handlers/index.ts b/src/nodes/agent/handlers/index.ts
@@ -1,6 +1,7 @@
 import type { DB } from '@matrixai/db';
 import type Logger from '@matrixai/logger';
 import type KeyRing from '../../../keys/KeyRing';
+import type Audit from '../../../audit/Audit';
 import type Sigchain from '../../../sigchain/Sigchain';
 import type ACL from '../../../acl/ACL';
 import type NodeGraph from '../../../nodes/NodeGraph';
@@ -9,6 +10,7 @@ import type NodeConnectionManager from '../../../nodes/NodeConnectionManager';
 import type NotificationsManager from '../../../notifications/NotificationsManager';
 import type VaultManager from '../../../vaults/VaultManager';
 import NodesAuthenticateConnection from './NodesAuthenticateConnection';
+import NodesAuditEventsGet from './NodesAuditEventsGet';
 import NodesClaimsGet from './NodesClaimsGet';
 import NodesClosestActiveConnectionsGet from './NodesClosestActiveConnectionsGet';
 import NodesClosestLocalNodesGet from './NodesClosestLocalNodesGet';
@@ -25,6 +27,7 @@ import VaultsScan from './VaultsScan';
  * Server manifest factory.
  */
 const manifestServer = (container: {
+  audit: Audit;
   db: DB;
   sigchain: Sigchain;
   nodeGraph: NodeGraph;
@@ -38,6 +41,7 @@ const manifestServer = (container: {
 }) => {
   return {
     nodesAuthenticateConnection: new NodesAuthenticateConnection(container),
+    nodesAuditEventsGet: new NodesAuditEventsGet(container),
     nodesClaimsGet: new NodesClaimsGet(container),
     nodesClosestActiveConnectionsGet: new NodesClosestActiveConnectionsGet(
       container,
@@ -60,6 +64,7 @@ export default manifestServer;
 
 export {
   NodesAuthenticateConnection,
+  NodesAuditEventsGet,
   NodesClaimsGet,
   NodesClosestActiveConnectionsGet,
   NodesClosestLocalNodesGet,
diff --git a/src/nodes/agent/types.ts b/src/nodes/agent/types.ts
@@ -4,18 +4,41 @@ import type {
   JSONRPCResponseResult,
 } from '@matrixai/rpc';
 import type { SignedTokenEncoded } from '../../tokens/types';
-import type { ClaimIdEncoded, NodeIdEncoded, VaultIdEncoded } from '../../ids';
+import type {
+  AuditEventIdEncoded,
+  ClaimIdEncoded,
+  NodeIdEncoded,
+  VaultIdEncoded,
+} from '../../ids';
 import type { VaultAction, VaultName } from '../../vaults/types';
 import type { SignedNotification } from '../../notifications/types';
 import type { Host, Hostname, Port } from '../../network/types';
 import type { NetworkId, NodeContact } from '../../nodes/types';
+import type { AuditEvent } from '../../audit/types';
 
 type AgentRPCRequestParams<T extends JSONObject = JSONObject> =
   JSONRPCRequestParams<T>;
 
 type AgentRPCResponseResult<T extends JSONObject = JSONObject> =
   JSONRPCResponseResult<T>;
 
+type AuditIdMessage = {
+  seek?: AuditEventIdEncoded | number;
+  seekEnd?: AuditEventIdEncoded | number;
+  order?: 'asc' | 'desc';
+  limit?: number;
+};
+
+type AgentAuditMessage<T extends AuditEvent> = Omit<T, 'id'> & {
+  id: AuditEventIdEncoded;
+};
+
+type NodesClaimsGetMessage = {
+  seek?: ClaimIdEncoded | number;
+  order?: 'asc' | 'desc';
+  limit?: number;
+};
+
 type ClaimIdMessage = {
   claimIdEncoded: ClaimIdEncoded;
 };
@@ -97,6 +120,9 @@ type NodesAuthenticateConnectionMessageNone = {
 export type {
   AgentRPCRequestParams,
   AgentRPCResponseResult,
+  AuditIdMessage,
+  AgentAuditMessage,
+  NodesClaimsGetMessage,
   ClaimIdMessage,
   AgentClaimMessage,
   NodeIdMessage,
diff --git a/src/sigchain/Sigchain.ts b/src/sigchain/Sigchain.ts
@@ -354,10 +354,10 @@ class Sigchain {
       seekOptions =
         order === 'asc'
           ? {
-              lte: [seek.toBuffer()],
+              gte: [seek.toBuffer()],
             }
           : {
-              gte: [seek.toBuffer()],
+              lte: [seek.toBuffer()],
             };
     }
     for await (const [kP, claim] of tran.iterator<Claim>(this.dbClaimsPath, {
diff --git a/tests/nodes/agent/handlers/nodesAuditsGet.test.ts b/tests/nodes/agent/handlers/nodesAuditsGet.test.ts
diff --git a/tests/nodes/agent/handlers/nodesClaimsGet.test.ts b/tests/nodes/agent/handlers/nodesClaimsGet.test.ts
diff --git a/tests/nodes/utils.ts b/tests/nodes/utils.ts

Original file line number	Diff line number	Diff line change
`@@ -354,10 +354,10 @@ class Sigchain {`
`354`	`354`	`seekOptions =`
`355`	`355`	`order === 'asc'`
`356`	`356`	`? {`
`357`		`- lte: [seek.toBuffer()],`
	`357`	`+ gte: [seek.toBuffer()],`
`358`	`358`	`}`
`359`	`359`	`: {`
`360`		`- gte: [seek.toBuffer()],`
	`360`	`+ lte: [seek.toBuffer()],`
`361`	`361`	`};`
`362`	`362`	`}`
`363`	`363`	`for await (const [kP, claim] of tran.iterator<Claim>(this.dbClaimsPath, {`