refactor(plugin-sdk): share direct dm ingress helpers

src/plugin-sdk/channel-inbound.ts

@@ -3,6 +3,12 @@ export {
   createInboundDebouncer,
   resolveInboundDebounceMs,
 } from "../auto-reply/inbound-debounce.js";
+export {
+  createDirectDmPreCryptoGuardPolicy,
+  dispatchInboundDirectDmWithRuntime,
+  type DirectDmPreCryptoGuardPolicy,
+  type DirectDmPreCryptoGuardPolicyOverrides,
+} from "./direct-dm.js";
 export {
   formatInboundEnvelope,
   formatInboundFromLabel,

src/plugin-sdk/command-auth.ts

@@ -1,5 +1,11 @@
 import type { OpenClawConfig } from "../config/config.js";
 import { resolveDmGroupAccessWithLists } from "../security/dm-policy-shared.js";
+export {
+  createPreCryptoDirectDmAuthorizer,
+  resolveInboundDirectDmAccessWithRuntime,
+  type DirectDmCommandAuthorizationRuntime,
+  type ResolvedInboundDirectDmAccess,
+} from "./direct-dm.js";
 
 export {
   hasControlCommand,

src/plugin-sdk/direct-dm.test.ts

@@ -0,0 +1,169 @@
+import { describe, expect, it, vi } from "vitest";
+import type { OpenClawConfig } from "../config/config.js";
+import {
+  createDirectDmPreCryptoGuardPolicy,
+  createPreCryptoDirectDmAuthorizer,
+  dispatchInboundDirectDmWithRuntime,
+  resolveInboundDirectDmAccessWithRuntime,
+} from "./direct-dm.js";
+
+const baseCfg = {
+  commands: { useAccessGroups: true },
+} as unknown as OpenClawConfig;
+
+describe("plugin-sdk/direct-dm", () => {
+  it("resolves inbound DM access and command auth through one helper", async () => {
+    const result = await resolveInboundDirectDmAccessWithRuntime({
+      cfg: baseCfg,
+      channel: "nostr",
+      accountId: "default",
+      dmPolicy: "pairing",
+      allowFrom: [],
+      senderId: "paired-user",
+      rawBody: "/status",
+      isSenderAllowed: (senderId, allowFrom) => allowFrom.includes(senderId),
+      readStoreAllowFrom: async () => ["paired-user"],
+      runtime: {
+        shouldComputeCommandAuthorized: () => true,
+        resolveCommandAuthorizedFromAuthorizers: ({ authorizers }) =>
+          authorizers.some((entry) => entry.configured && entry.allowed),
+      },
+      modeWhenAccessGroupsOff: "configured",
+    });
+
+    expect(result.access.decision).toBe("allow");
+    expect(result.access.effectiveAllowFrom).toEqual(["paired-user"]);
+    expect(result.senderAllowedForCommands).toBe(true);
+    expect(result.commandAuthorized).toBe(true);
+  });
+
+  it("creates a pre-crypto authorizer that issues pairing and blocks unknown senders", async () => {
+    const issuePairingChallenge = vi.fn(async () => {});
+    const onBlocked = vi.fn();
+    const authorizer = createPreCryptoDirectDmAuthorizer({
+      resolveAccess: async (senderId) => ({
+        access:
+          senderId === "pair-me"
+            ? {
+                decision: "pairing" as const,
+                reasonCode: "dm_policy_pairing_required",
+                reason: "dmPolicy=pairing (not allowlisted)",
+                effectiveAllowFrom: [],
+              }
+            : {
+                decision: "block" as const,
+                reasonCode: "dm_policy_disabled",
+                reason: "dmPolicy=disabled",
+                effectiveAllowFrom: [],
+              },
+      }),
+      issuePairingChallenge,
+      onBlocked,
+    });
+
+    await expect(
+      authorizer({
+        senderId: "pair-me",
+        reply: async () => {},
+      }),
+    ).resolves.toBe("pairing");
+    await expect(
+      authorizer({
+        senderId: "blocked",
+        reply: async () => {},
+      }),
+    ).resolves.toBe("block");
+
+    expect(issuePairingChallenge).toHaveBeenCalledTimes(1);
+    expect(onBlocked).toHaveBeenCalledWith({
+      senderId: "blocked",
+      reason: "dmPolicy=disabled",
+      reasonCode: "dm_policy_disabled",
+    });
+  });
+
+  it("builds a shared pre-crypto guard policy with partial overrides", () => {
+    const policy = createDirectDmPreCryptoGuardPolicy({
+      maxFutureSkewSec: 30,
+      rateLimit: {
+        maxPerSenderPerWindow: 5,
+      },
+    });
+
+    expect(policy.allowedKinds).toEqual([4]);
+    expect(policy.maxFutureSkewSec).toBe(30);
+    expect(policy.maxCiphertextBytes).toBe(16 * 1024);
+    expect(policy.rateLimit.maxPerSenderPerWindow).toBe(5);
+    expect(policy.rateLimit.maxGlobalPerWindow).toBe(200);
+  });
+
+  it("dispatches direct DMs through the standard route/session/reply pipeline", async () => {
+    const recordInboundSession = vi.fn(async () => {});
+    const dispatchReplyWithBufferedBlockDispatcher = vi.fn(async ({ dispatcherOptions }) => {
+      await dispatcherOptions.deliver({ text: "reply text" });
+    });
+    const deliver = vi.fn(async () => {});
+
+    const result = await dispatchInboundDirectDmWithRuntime({
+      cfg: {
+        session: { store: { type: "jsonl" } },
+      } as never,
+      runtime: {
+        channel: {
+          routing: {
+            resolveAgentRoute: vi.fn(({ accountId, peer }) => ({
+              agentId: "agent-main",
+              accountId,
+              sessionKey: `dm:${peer.id}`,
+            })),
+          },
+          session: {
+            resolveStorePath: vi.fn(() => "/tmp/direct-dm-session-store"),
+            readSessionUpdatedAt: vi.fn(() => 1234),
+            recordInboundSession,
+          },
+          reply: {
+            resolveEnvelopeFormatOptions: vi.fn(() => ({ mode: "agent" })),
+            formatAgentEnvelope: vi.fn(({ body }) => `env:${body}`),
+            finalizeInboundContext: vi.fn((ctx) => ctx),
+            dispatchReplyWithBufferedBlockDispatcher,
+          },
+        },
+      } as never,
+      channel: "nostr",
+      channelLabel: "Nostr",
+      accountId: "default",
+      peer: { kind: "direct", id: "sender-1" },
+      senderId: "sender-1",
+      senderAddress: "nostr:sender-1",
+      recipientAddress: "nostr:bot-1",
+      conversationLabel: "sender-1",
+      rawBody: "hello world",
+      messageId: "event-123",
+      timestamp: 1_710_000_000_000,
+      commandAuthorized: true,
+      deliver,
+      onRecordError: () => {},
+      onDispatchError: () => {},
+    });
+
+    expect(result.route).toMatchObject({
+      agentId: "agent-main",
+      accountId: "default",
+      sessionKey: "dm:sender-1",
+    });
+    expect(result.storePath).toBe("/tmp/direct-dm-session-store");
+    expect(result.ctxPayload).toMatchObject({
+      Body: "env:hello world",
+      BodyForAgent: "hello world",
+      From: "nostr:sender-1",
+      To: "nostr:bot-1",
+      SenderId: "sender-1",
+      MessageSid: "event-123",
+      CommandAuthorized: true,
+    });
+    expect(recordInboundSession).toHaveBeenCalledTimes(1);
+    expect(dispatchReplyWithBufferedBlockDispatcher).toHaveBeenCalledTimes(1);
+    expect(deliver).toHaveBeenCalledWith({ text: "reply text" });
+  });
+});

src/plugin-sdk/direct-dm.ts

@@ -0,0 +1,314 @@
+import type { FinalizedMsgContext } from "../auto-reply/templating.js";
+import type { ChannelId } from "../channels/plugins/types.js";
+import type { OpenClawConfig } from "../config/config.js";
+import {
+  readStoreAllowFromForDmPolicy,
+  resolveDmGroupAccessWithLists,
+  type DmGroupAccessReasonCode,
+} from "../security/dm-policy-shared.js";
+import { resolveInboundRouteEnvelopeBuilderWithRuntime } from "./inbound-envelope.js";
+import { recordInboundSessionAndDispatchReply } from "./inbound-reply-dispatch.js";
+import type { OutboundReplyPayload } from "./reply-payload.js";
+
+export type DirectDmCommandAuthorizationRuntime = {
+  shouldComputeCommandAuthorized: (rawBody: string, cfg: OpenClawConfig) => boolean;
+  resolveCommandAuthorizedFromAuthorizers: (params: {
+    useAccessGroups: boolean;
+    authorizers: Array<{ configured: boolean; allowed: boolean }>;
+    modeWhenAccessGroupsOff?: "allow" | "deny" | "configured";
+  }) => boolean;
+};
+
+export type ResolvedInboundDirectDmAccess = {
+  access: {
+    decision: "allow" | "block" | "pairing";
+    reasonCode: DmGroupAccessReasonCode;
+    reason: string;
+    effectiveAllowFrom: string[];
+  };
+  shouldComputeAuth: boolean;
+  senderAllowedForCommands: boolean;
+  commandAuthorized: boolean | undefined;
+};
+
+/** Resolve direct-DM policy, effective allowlists, and optional command auth in one place. */
+export async function resolveInboundDirectDmAccessWithRuntime(params: {
+  cfg: OpenClawConfig;
+  channel: ChannelId;
+  accountId: string;
+  dmPolicy?: string | null;
+  allowFrom?: Array<string | number> | null;
+  senderId: string;
+  rawBody: string;
+  isSenderAllowed: (senderId: string, allowFrom: string[]) => boolean;
+  runtime: DirectDmCommandAuthorizationRuntime;
+  modeWhenAccessGroupsOff?: "allow" | "deny" | "configured";
+  readStoreAllowFrom?: (provider: ChannelId, accountId: string) => Promise<string[]>;
+}): Promise<ResolvedInboundDirectDmAccess> {
+  const dmPolicy = params.dmPolicy ?? "pairing";
+  const storeAllowFrom =
+    dmPolicy === "pairing"
+      ? await readStoreAllowFromForDmPolicy({
+          provider: params.channel,
+          accountId: params.accountId,
+          dmPolicy,
+          readStore: params.readStoreAllowFrom,
+        })
+      : [];
+
+  const access = resolveDmGroupAccessWithLists({
+    isGroup: false,
+    dmPolicy,
+    allowFrom: params.allowFrom,
+    storeAllowFrom,
+    groupAllowFromFallbackToAllowFrom: false,
+    isSenderAllowed: (allowEntries) => params.isSenderAllowed(params.senderId, allowEntries),
+  });
+
+  const shouldComputeAuth = params.runtime.shouldComputeCommandAuthorized(
+    params.rawBody,
+    params.cfg,
+  );
+  const senderAllowedForCommands = params.isSenderAllowed(
+    params.senderId,
+    access.effectiveAllowFrom,
+  );
+  const commandAuthorized = shouldComputeAuth
+    ? dmPolicy === "open"
+      ? true
+      : params.runtime.resolveCommandAuthorizedFromAuthorizers({
+          useAccessGroups: params.cfg.commands?.useAccessGroups !== false,
+          authorizers: [
+            {
+              configured: access.effectiveAllowFrom.length > 0,
+              allowed: senderAllowedForCommands,
+            },
+          ],
+          modeWhenAccessGroupsOff: params.modeWhenAccessGroupsOff,
+        })
+    : undefined;
+
+  return {
+    access: {
+      decision: access.decision,
+      reasonCode: access.reasonCode,
+      reason: access.reason,
+      effectiveAllowFrom: access.effectiveAllowFrom,
+    },
+    shouldComputeAuth,
+    senderAllowedForCommands,
+    commandAuthorized,
+  };
+}
+
+/** Convert resolved DM policy into a pre-crypto allow/block/pairing callback. */
+export function createPreCryptoDirectDmAuthorizer(params: {
+  resolveAccess: (
+    senderId: string,
+  ) => Promise<Pick<ResolvedInboundDirectDmAccess, "access"> | ResolvedInboundDirectDmAccess>;
+  issuePairingChallenge?: (params: {
+    senderId: string;
+    reply: (text: string) => Promise<void>;
+  }) => Promise<void>;
+  onBlocked?: (params: {
+    senderId: string;
+    reason: string;
+    reasonCode: DmGroupAccessReasonCode;
+  }) => void;
+}) {
+  return async (input: {
+    senderId: string;
+    reply: (text: string) => Promise<void>;
+  }): Promise<"allow" | "block" | "pairing"> => {
+    const resolved = await params.resolveAccess(input.senderId);
+    const access = "access" in resolved ? resolved.access : resolved;
+    if (access.decision === "allow") {
+      return "allow";
+    }
+    if (access.decision === "pairing") {
+      if (params.issuePairingChallenge) {
+        await params.issuePairingChallenge({
+          senderId: input.senderId,
+          reply: input.reply,
+        });
+      }
+      return "pairing";
+    }
+    params.onBlocked?.({
+      senderId: input.senderId,
+      reason: access.reason,
+      reasonCode: access.reasonCode,
+    });
+    return "block";
+  };
+}
+
+export type DirectDmPreCryptoGuardPolicy = {
+  allowedKinds: readonly number[];
+  maxFutureSkewSec: number;
+  maxCiphertextBytes: number;
+  maxPlaintextBytes: number;
+  rateLimit: {
+    windowMs: number;
+    maxPerSenderPerWindow: number;
+    maxGlobalPerWindow: number;
+    maxTrackedSenderKeys: number;
+  };
+};
+
+export type DirectDmPreCryptoGuardPolicyOverrides = Partial<
+  Omit<DirectDmPreCryptoGuardPolicy, "rateLimit">
+> & {
+  rateLimit?: Partial<DirectDmPreCryptoGuardPolicy["rateLimit"]>;
+};
+
+/** Shared policy object for DM-style pre-crypto guardrails. */
+export function createDirectDmPreCryptoGuardPolicy(
+  overrides: DirectDmPreCryptoGuardPolicyOverrides = {},
+): DirectDmPreCryptoGuardPolicy {
+  return {
+    allowedKinds: overrides.allowedKinds ?? [4],
+    maxFutureSkewSec: overrides.maxFutureSkewSec ?? 120,
+    maxCiphertextBytes: overrides.maxCiphertextBytes ?? 16 * 1024,
+    maxPlaintextBytes: overrides.maxPlaintextBytes ?? 8 * 1024,
+    rateLimit: {
+      windowMs: overrides.rateLimit?.windowMs ?? 60_000,
+      maxPerSenderPerWindow: overrides.rateLimit?.maxPerSenderPerWindow ?? 20,
+      maxGlobalPerWindow: overrides.rateLimit?.maxGlobalPerWindow ?? 200,
+      maxTrackedSenderKeys: overrides.rateLimit?.maxTrackedSenderKeys ?? 4096,
+    },
+  };
+}
+
+type DirectDmRoutePeer = {
+  kind: "direct";
+  id: string;
+};
+
+type DirectDmRoute = {
+  agentId: string;
+  sessionKey: string;
+  accountId?: string;
+};
+
+type DirectDmRuntime = {
+  channel: {
+    routing: {
+      resolveAgentRoute: (params: {
+        cfg: OpenClawConfig;
+        channel: string;
+        accountId: string;
+        peer: DirectDmRoutePeer;
+      }) => DirectDmRoute;
+    };
+    session: {
+      resolveStorePath: typeof import("../config/sessions.js").resolveStorePath;
+      readSessionUpdatedAt: (params: {
+        storePath: string;
+        sessionKey: string;
+      }) => number | undefined;
+      recordInboundSession: typeof import("../channels/session.js").recordInboundSession;
+    };
+    reply: {
+      resolveEnvelopeFormatOptions: (
+        cfg: OpenClawConfig,
+      ) => ReturnType<typeof import("../auto-reply/envelope.js").resolveEnvelopeFormatOptions>;
+      formatAgentEnvelope: typeof import("../auto-reply/envelope.js").formatAgentEnvelope;
+      finalizeInboundContext: typeof import("../auto-reply/reply/inbound-context.js").finalizeInboundContext;
+      dispatchReplyWithBufferedBlockDispatcher: typeof import("../auto-reply/reply/provider-dispatcher.js").dispatchReplyWithBufferedBlockDispatcher;
+    };
+  };
+};
+
+/** Route, envelope, record, and dispatch one direct-DM turn through the standard pipeline. */
+export async function dispatchInboundDirectDmWithRuntime(params: {
+  cfg: OpenClawConfig;
+  runtime: DirectDmRuntime;
+  channel: string;
+  channelLabel: string;
+  accountId: string;
+  peer: DirectDmRoutePeer;
+  senderId: string;
+  senderAddress: string;
+  recipientAddress: string;
+  conversationLabel: string;
+  rawBody: string;
+  messageId: string;
+  timestamp?: number;
+  commandAuthorized?: boolean;
+  bodyForAgent?: string;
+  commandBody?: string;
+  provider?: string;
+  surface?: string;
+  originatingChannel?: string;
+  originatingTo?: string;
+  extraContext?: Record<string, unknown>;
+  deliver: (payload: OutboundReplyPayload) => Promise<void>;
+  onRecordError: (err: unknown) => void;
+  onDispatchError: (err: unknown, info: { kind: string }) => void;
+}): Promise<{
+  route: DirectDmRoute;
+  storePath: string;
+  ctxPayload: FinalizedMsgContext;
+}> {
+  const { route, buildEnvelope } = resolveInboundRouteEnvelopeBuilderWithRuntime({
+    cfg: params.cfg,
+    channel: params.channel,
+    accountId: params.accountId,
+    peer: params.peer,
+    runtime: params.runtime.channel,
+    sessionStore: params.cfg.session?.store,
+  });
+
+  const { storePath, body } = buildEnvelope({
+    channel: params.channelLabel,
+    from: params.conversationLabel,
+    body: params.rawBody,
+    timestamp: params.timestamp,
+  });
+
+  const ctxPayload = params.runtime.channel.reply.finalizeInboundContext({
+    Body: body,
+    BodyForAgent: params.bodyForAgent ?? params.rawBody,
+    RawBody: params.rawBody,
+    CommandBody: params.commandBody ?? params.rawBody,
+    From: params.senderAddress,
+    To: params.recipientAddress,
+    SessionKey: route.sessionKey,
+    AccountId: route.accountId ?? params.accountId,
+    ChatType: "direct",
+    ConversationLabel: params.conversationLabel,
+    SenderId: params.senderId,
+    Provider: params.provider ?? params.channel,
+    Surface: params.surface ?? params.channel,
+    MessageSid: params.messageId,
+    MessageSidFull: params.messageId,
+    Timestamp: params.timestamp,
+    CommandAuthorized: params.commandAuthorized,
+    OriginatingChannel: params.originatingChannel ?? params.channel,
+    OriginatingTo: params.originatingTo ?? params.recipientAddress,
+    ...params.extraContext,
+  });
+
+  await recordInboundSessionAndDispatchReply({
+    cfg: params.cfg,
+    channel: params.channel,
+    accountId: route.accountId ?? params.accountId,
+    agentId: route.agentId,
+    routeSessionKey: route.sessionKey,
+    storePath,
+    ctxPayload,
+    recordInboundSession: params.runtime.channel.session.recordInboundSession,
+    dispatchReplyWithBufferedBlockDispatcher:
+      params.runtime.channel.reply.dispatchReplyWithBufferedBlockDispatcher,
+    deliver: params.deliver,
+    onRecordError: params.onRecordError,
+    onDispatchError: params.onDispatchError,
+  });
+
+  return {
+    route,
+    storePath,
+    ctxPayload,
+  };
+}

src/plugin-sdk/nostr.ts

@@ -8,6 +8,16 @@ export type { ChannelSetupAdapter } from "../channels/plugins/types.adapters.js"
 export { formatPairingApproveHint } from "../channels/plugins/helpers.js";
 export type { ChannelPlugin } from "../channels/plugins/types.plugin.js";
 export { createChannelReplyPipeline } from "./channel-reply-pipeline.js";
+export {
+  createDirectDmPreCryptoGuardPolicy,
+  dispatchInboundDirectDmWithRuntime,
+  type DirectDmPreCryptoGuardPolicy,
+  type DirectDmPreCryptoGuardPolicyOverrides,
+} from "./direct-dm.js";
+export {
+  createPreCryptoDirectDmAuthorizer,
+  resolveInboundDirectDmAccessWithRuntime,
+} from "./direct-dm.js";
 export type { OpenClawConfig } from "../config/config.js";
 export { MarkdownConfigSchema } from "../config/zod-schema.core.js";
 export { readJsonBodyWithLimit, requestBodyErrorToText } from "../infra/http-body.js";

src/plugin-sdk/subpaths.test.ts

@@ -346,8 +346,10 @@ describe("plugin-sdk subpath exports", () => {
     ]);
     expectSourceMentions("channel-inbound", [
       "buildMentionRegexes",
+      "createDirectDmPreCryptoGuardPolicy",
       "createChannelInboundDebouncer",
       "createInboundDebouncer",
+      "dispatchInboundDirectDmWithRuntime",
       "formatInboundEnvelope",
       "formatInboundFromLabel",
       "formatLocationText",
@@ -446,8 +448,10 @@ describe("plugin-sdk subpath exports", () => {
       "listNativeCommandSpecsForConfig",
       "listSkillCommandsForAgents",
       "normalizeCommandBody",
+      "createPreCryptoDirectDmAuthorizer",
       "resolveCommandAuthorization",
       "resolveCommandAuthorizedFromAuthorizers",
+      "resolveInboundDirectDmAccessWithRuntime",
       "resolveControlCommandGate",
       "resolveDualTextControlCommandGate",
       "resolveNativeCommandSessionTargets",