Improve transport safety and log redaction

plugin/core/runtime.ts

@@ -17,12 +17,15 @@ import {
   buildPairRequest,
   buildPairSuccess,
   buildRePairRequired,
+  CodecError,
   decodeBuiltin,
   encodeBuiltin,
+  encodeRuleMessage,
   extractAuthRequestSigningInput,
   isBuiltinMessage,
   isTimestampFresh,
   isValidAuthNonce,
+  parseRuleMessage,
   type AuthRequestPayload,
   type HeartbeatPayload
 } from "../../../Yonexus.Protocol/src/index.js";
@@ -43,6 +46,7 @@ import {
   createDiscordNotificationService,
   type DiscordNotificationService
 } from "../notifications/discord.js";
+import { safeErrorMessage } from "./logging.js";
 
 export interface YonexusServerRuntimeOptions {
   config: YonexusServerConfig;
@@ -143,10 +147,27 @@ export class YonexusServerRuntime {
 
   async handleMessage(connection: ClientConnection, raw: string): Promise<void> {
     if (!isBuiltinMessage(raw)) {
+      // Handle rule message - rewrite and dispatch
+      await this.handleRuleMessage(connection, raw);
       return;
     }
 
-    const envelope = decodeBuiltin(raw);
+    let envelope: BuiltinEnvelope;
+    try {
+      envelope = decodeBuiltin(raw);
+    } catch (error) {
+      const message = error instanceof CodecError ? error.message : "Invalid builtin message";
+      this.options.transport.sendToConnection(
+        connection,
+        encodeBuiltin(
+          buildError(
+            { code: "MALFORMED_MESSAGE", message },
+            { timestamp: this.now() }
+          )
+        )
+      );
+      return;
+    }
 
     if (envelope.type === "hello") {
       await this.handleHello(connection, envelope as BuiltinEnvelope<"hello", HelloPayload>);
@@ -174,7 +195,21 @@ export class YonexusServerRuntime {
         connection,
         envelope as BuiltinEnvelope<"heartbeat", HeartbeatPayload>
       );
+      return;
     }
+
+    this.options.transport.sendToConnection(
+      connection,
+      encodeBuiltin(
+        buildError(
+          {
+            code: "MALFORMED_MESSAGE",
+            message: `Unsupported builtin type: ${String(envelope.type)}`
+          },
+          { requestId: envelope.requestId, timestamp: this.now() }
+        )
+      )
+    );
   }
 
   private async handleHello(
@@ -803,6 +838,123 @@ export class YonexusServerRuntime {
   private async persist(): Promise<void> {
     await this.options.store.save(this.registry.clients.values());
   }
+
+  /**
+   * Send a rule message to a specific client.
+   *
+   * @param identifier - The target client identifier
+   * @param message - The complete rule message with identifier and content
+   * @returns True if message was sent, false if client not connected/authenticated
+   */
+  sendMessageToClient(identifier: string, message: string): boolean {
+    const session = this.registry.sessions.get(identifier);
+    if (!session || !session.isAuthenticated) {
+      return false;
+    }
+
+    // Validate the message is a properly formatted rule message
+    try {
+      // Quick check: must not be a builtin message and must have :: delimiter
+      if (message.startsWith("builtin::")) {
+        return false;
+      }
+      const delimiterIndex = message.indexOf("::");
+      if (delimiterIndex === -1) {
+        return false;
+      }
+      parseRuleMessage(message);
+    } catch {
+      return false;
+    }
+
+    return this.options.transport.send(identifier, message);
+  }
+
+  /**
+   * Send a rule message to a specific client using separate rule identifier and content.
+   *
+   * @param identifier - The target client identifier
+   * @param ruleIdentifier - The rule identifier
+   * @param content - The message content
+   * @returns True if message was sent, false if client not connected/authenticated or invalid format
+   */
+  sendRuleMessageToClient(identifier: string, ruleIdentifier: string, content: string): boolean {
+    const session = this.registry.sessions.get(identifier);
+    if (!session || !session.isAuthenticated) {
+      return false;
+    }
+
+    try {
+      const encoded = encodeRuleMessage(ruleIdentifier, content);
+      return this.options.transport.send(identifier, encoded);
+    } catch {
+      return false;
+    }
+  }
+
+  /**
+   * Handle incoming rule message from a client.
+   * Rewrites the message to include sender identifier before dispatch.
+   *
+   * @param connection - The client connection
+   * @param raw - The raw rule message
+   */
+  private async handleRuleMessage(connection: ClientConnection, raw: string): Promise<void> {
+    // Get sender identifier from connection or session
+    let senderIdentifier = connection.identifier;
+    if (!senderIdentifier) {
+      // Try to find identifier from WebSocket
+      for (const [id, session] of this.registry.sessions.entries()) {
+        if (session.socket === connection.ws) {
+          senderIdentifier = id;
+          break;
+        }
+      }
+    }
+
+    if (!senderIdentifier) {
+      // Cannot determine sender - close connection
+      connection.ws.close(1008, "Cannot identify sender");
+      return;
+    }
+
+    const session = this.registry.sessions.get(senderIdentifier);
+    if (!session || !session.isAuthenticated) {
+      // Only accept rule messages from authenticated clients
+      connection.ws.close(1008, "Not authenticated");
+      return;
+    }
+
+    try {
+      const parsed = parseRuleMessage(raw);
+      const rewritten = `${parsed.ruleIdentifier}::${senderIdentifier}::${parsed.content}`;
+
+      // TODO: Dispatch to registered rules via rule registry
+      // For now, just log the rewritten message
+      // this.ruleRegistry.dispatch(rewritten);
+
+      // Update last activity
+      session.lastActivityAt = this.now();
+
+      // Future: dispatch to rule registry
+      // eslint-disable-next-line @typescript-eslint/no-unused-vars
+      void rewritten;
+    } catch (error) {
+      // Malformed rule message
+      this.options.transport.sendToConnection(
+        connection,
+        encodeBuiltin(
+          buildError(
+            {
+              code: "MALFORMED_MESSAGE",
+              message: safeErrorMessage(error) || "Invalid rule message format"
+            },
+            { timestamp: this.now() }
+          )
+        )
+      );
+    }
+  }
 }
 
 export function createYonexusServerRuntime(