PVA: Handle CMD_ORIGIN_TAG

kasemir · kasemir · commit a9c3a3aff7e9 · 2025-06-16T14:14:25.000-04:00
When forwarding UDP unicast search requests, prefix them with origin
tag.
When server receives search request, check for optional preceding origin
tag to identify forwarded message.

On forward, 'reply_to_src_port' flag is copied, but server ignores it in
forwarded messages.
diff --git a/core/pva/src/main/java/org/epics/pva/client/ClientUDPHandler.java b/core/pva/src/main/java/org/epics/pva/client/ClientUDPHandler.java
@@ -22,6 +22,7 @@
 import org.epics.pva.PVASettings;
 import org.epics.pva.common.AddressInfo;
 import org.epics.pva.common.Network;
+import org.epics.pva.common.OriginTag;
 import org.epics.pva.common.PVAHeader;
 import org.epics.pva.common.SearchRequest;
 import org.epics.pva.common.SearchResponse;
@@ -197,6 +198,9 @@ protected boolean handleMessage(final InetSocketAddress from, final byte version
         {
         case PVAHeader.CMD_BEACON:
             return handleBeacon(from, version, payload, buffer);
+        case PVAHeader.CMD_ORIGIN_TAG:
+            // Will be decoded with CMD_SEARCH
+            break;
         case PVAHeader.CMD_SEARCH:
             return handleSearchRequest(from, version, payload, buffer);
         case PVAHeader.CMD_SEARCH_RESPONSE:
@@ -290,7 +294,8 @@ private boolean handleBeacon(final InetSocketAddress from, final byte version,
     private boolean handleSearchRequest(final InetSocketAddress from, final byte version,
                                         final int payload, final ByteBuffer buffer)
     {
-        final SearchRequest search = SearchRequest.decode(from, version, payload, buffer);
+        final OriginTag origin = OriginTag.testForOriginOfSearch(from, buffer);
+        final SearchRequest search = SearchRequest.decode(origin, from, version, payload, buffer);
         try
         {
             if (local_multicast != null  &&  search != null  &&  search.unicast)
@@ -300,7 +305,8 @@ private boolean handleSearchRequest(final InetSocketAddress from, final byte ver
                     if (search.reply_required)
                     {
                         forward_buffer.clear();
-                        SearchRequest.encode(false, false, 0, null, search.client, search.tls, forward_buffer);
+                        OriginTag.encode(udp_search4, forward_buffer);
+                        SearchRequest.encode(false, search.reply_to_src_port, 0, null, search.client, search.tls, forward_buffer);
                         forward_buffer.flip();
                         logger.log(Level.FINER, () -> "Forward search to list servers to " + local_multicast + "\n" + Hexdump.toHexdump(forward_buffer));
                         send(forward_buffer, local_multicast);
@@ -309,7 +315,8 @@ private boolean handleSearchRequest(final InetSocketAddress from, final byte ver
                 else
                 {
                     forward_buffer.clear();
-                    SearchRequest.encode(false, false, search.seq, search.channels, search.client, search.tls, forward_buffer);
+                    OriginTag.encode(udp_search4, forward_buffer);
+                    SearchRequest.encode(false, search.reply_to_src_port, search.seq, search.channels, search.client, search.tls, forward_buffer);
                     forward_buffer.flip();
                     logger.log(Level.FINER, () -> "Forward search to " + local_multicast + "\n" + Hexdump.toHexdump(forward_buffer));
                     send(forward_buffer, local_multicast);
diff --git a/core/pva/src/main/java/org/epics/pva/common/OriginTag.java b/core/pva/src/main/java/org/epics/pva/common/OriginTag.java
@@ -0,0 +1,128 @@
+/*******************************************************************************
+ * Copyright (c) 2025 Oak Ridge National Laboratory.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ ******************************************************************************/
+package org.epics.pva.common;
+
+import static org.epics.pva.PVASettings.logger;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
+import java.util.logging.Level;
+
+import org.epics.pva.data.PVAAddress;
+
+/** Helper for CMD_ORIGIN_TAG
+ *
+ *  <p>UDP search messages that are forwarded via
+ *  the local 224.0.0.128 multicast are prefixed
+ *  with CMD_ORIGIN_TAG.
+ *
+ *  @author Kay Kasemir
+ */
+@SuppressWarnings("nls")
+public class OriginTag
+{
+    /** Size of CMD_ORIGIN_TAG payload (address) */
+    public static final byte PAYLOAD_SIZE = 16;
+
+    /** Size of CMD_ORIGIN_TAG message */
+    public static final byte TOTAL_SIZE = PVAHeader.HEADER_SIZE + PAYLOAD_SIZE;
+
+    /** "address to which the receiving socket was bound ..may be 0.0.0.0" */
+    public InetAddress address;
+
+    /** Check for optional CMD_ORIGIN_TAG before CMD_SEARCH
+     *  @param from Peer address
+     *  @param buffer UDP message buffer ready to decode CMD_SEARCH
+     *  @return {@link OriginTag} or <code>null</code>
+     */
+    public static OriginTag testForOriginOfSearch(final InetSocketAddress from, final ByteBuffer buffer)
+    {
+        // Check for optional origin tag.
+        // For a normal search packet, the buffer is positioned at PVAHeader.HEADER_SIZE (8).
+        // For a forwarded packet, it's at CMD_ORIGIN_TAG message (8+16) + HEADER_SIZE (8), i.e., 32
+        // For now assuming no or exactly one CMD_ORIGIN_TAG before CMD_SEARCH.
+        // Not supporting multiple CMD_ORIGIN_TAGs nor anything else before CMD_SEARCH.
+
+        // pos is just after the CMD_SEARCH message header
+        final int pos = buffer.position();
+        if (// Is there room for CMD_ORIGIN_TAG before this message header?
+            pos == OriginTag.TOTAL_SIZE + PVAHeader.HEADER_SIZE  &&
+            // Is command of previous message indeed CMD_ORIGIN_TAG?
+            buffer.get(PVAHeader.HEADER_OFFSET_COMMAND) == PVAHeader.CMD_ORIGIN_TAG)
+        {
+            // Move back to origin message, decode it, restore buffer position
+            buffer.position(PVAHeader.HEADER_SIZE);
+            OriginTag origin = OriginTag.decode(from, OriginTag.PAYLOAD_SIZE, buffer);
+            buffer.position(pos);
+            return origin;
+        }
+
+        return null;
+    }
+
+    /** Decode origin tag
+     *
+     *  @param from Peer address
+     *  @param payload Payload size
+     *  @param buffer Buffer positioned on payload
+     *  @return Decoded origin tag or <code>null</code> if not a valid
+     */
+    public static OriginTag decode(final InetSocketAddress from,
+                                   final int payload, final ByteBuffer buffer)
+    {
+        if (payload < PAYLOAD_SIZE)
+        {
+            logger.log(Level.WARNING, "PVA client " + from + " sent only " + payload + " bytes for origin tag");
+            return null;
+        }
+        final OriginTag origin = new OriginTag();
+
+        // responseAddress, IPv6 address in case of IP based transport, UDP
+        try
+        {
+            origin.address = PVAAddress.decode(buffer);
+        }
+        catch (Exception ex)
+        {
+            logger.log(Level.WARNING, "PVA Client " + from + " sent origin tag with invalid address");
+            return null;
+        }
+        logger.log(Level.FINER, () -> "PVA client " + from + " sent " + origin);
+        return origin;
+    }
+
+    /** Encode origin tag
+     *  @param udp UDP socket from which to pick the forwarder's address
+     *  @param buffer Buffer into which to encode
+     *  @return Encoded forwarder's address
+     */
+    public static InetAddress encode(final DatagramChannel udp, final ByteBuffer buffer)
+    {
+        PVAHeader.encodeMessageHeader(buffer, PVAHeader.FLAG_NONE, PVAHeader.CMD_ORIGIN_TAG, PAYLOAD_SIZE);
+        InetAddress this_end;
+        try
+        {
+            this_end = ((InetSocketAddress) udp.getLocalAddress()).getAddress();
+        }
+        catch (Exception ex)
+        {
+            logger.log(Level.WARNING, "Invalid address for CMD_ORIGIN_TAG", ex);
+            this_end = new InetSocketAddress(0).getAddress();
+        }
+        PVAAddress.encode(this_end, buffer);
+        return this_end;
+    }
+
+    @Override
+    public String toString()
+    {
+        return "Origin tag with address " + address;
+    }
+}
diff --git a/core/pva/src/main/java/org/epics/pva/common/PVAHeader.java b/core/pva/src/main/java/org/epics/pva/common/PVAHeader.java
@@ -27,7 +27,7 @@ public class PVAHeader
     public static final byte PVA_MAGIC = (byte)0xCA;
 
     /** PVA protocol revision (implemented by this library)
-     * 
+     *
      *  <br>v2: Server's Echo reply includes the request payload
      *  <br>v3: SearchRequest FLAG_REPLY_SRC_PORT
      */
@@ -137,12 +137,18 @@ public class PVAHeader
     /** Offset from start of common PVA message header to byte version */
     public static final int HEADER_OFFSET_VERSION = 1;
 
+    /** Offset from start of common PVA message header to byte flags */
+    public static final int HEADER_OFFSET_FLAGS = 2;
+
+    /** Offset from start of common PVA message header to byte command */
+    public static final int HEADER_OFFSET_COMMAND = 3;
+
     /** Offset from start of common PVA message header to int payload_size */
     public static final int HEADER_OFFSET_PAYLOAD_SIZE = 4;
 
 
     /** Encode common PVA message header
-     *  @param buffer Buffer into which to encode
+     *  @param buffer Buffer into which to encode, must be positioned at desired address (usually 'clear()'ed)
      *  @param flags  Combination of FLAG_
      *  @param command Command
      *  @param payload_size Size of payload that follows
@@ -154,7 +160,6 @@ public static void encodeMessageHeader(final ByteBuffer buffer, byte flags, fina
             flags |= FLAG_BIG_ENDIAN;
         else
             flags &= ~FLAG_BIG_ENDIAN;
-        buffer.clear();
         buffer.put(PVA_MAGIC);
         buffer.put(PVA_PROTOCOL_REVISION);
         buffer.put(flags);
diff --git a/core/pva/src/main/java/org/epics/pva/common/SearchRequest.java b/core/pva/src/main/java/org/epics/pva/common/SearchRequest.java
@@ -70,7 +70,17 @@ public String toString()
 
     /** Client should ignore the 'port' in the reply and
      *  simply use the port of the 'source', that is the peer port
-     *  of the UDP message or TCP connection
+     *  of the UDP message or TCP connection.
+     *
+     *  <p>In a forwarded message, the 'client' (reply-to)
+     *  address has been updated to reflect the original client.
+     *  The 'reply_to_src_port' flag is copied, but server
+     *  must NOT reply to the source port because that would
+     *  be the port of the forwarder, not the real client.
+     *  The presence of an {@link OriginTag} will indicate
+     *  that the 'reply_to_src_port' flag is copied and needs
+     *  to be ignored.
+     *
      *  @since Version 3
      */
     public static final byte FLAG_REPLY_SRC_PORT = 0x02;
@@ -94,14 +104,14 @@ public String toString()
     public List<Channel> channels;
 
     /** Check search request
-     *
+     *  @param origin Optional CMD_ORIGIN_TAG that preceded the search message
      *  @param from Peer address
      *  @param version Message version
      *  @param payload Payload size
      *  @param buffer Buffer positioned on payload
      *  @return Decoded search request or <code>null</code> if not a valid search request
      */
-    public static SearchRequest decode(final InetSocketAddress from, final byte version,
+    public static SearchRequest decode(final OriginTag origin, final InetSocketAddress from, final byte version,
                                        final int payload, final ByteBuffer buffer)
     {
         // pvinfo sends 0x1D=29 bytes:
@@ -150,12 +160,12 @@ public static SearchRequest decode(final InetSocketAddress from, final byte vers
         // and instead use the peer's port.
         // This should help with NAT where we get the message from an intermediate
         // and need to reply via that same intermediate
-        if (version >= 3  &&  search.reply_to_src_port)
+        if (version >= 3  &&  search.reply_to_src_port  &&  origin == null)
             port = from.getPort();
 
         // Use address from message unless it's a generic local address
         if (addr.isAnyLocalAddress() || port <= 0)
-            search.client = from;
+            search.client = new InetSocketAddress(from.getAddress(), port);
         else
             search.client = new InetSocketAddress(addr, port);
 
@@ -200,7 +210,7 @@ else if ("tcp".equals(protocol))
                                             + (search.tls               ? " (TLS)" : "")
                                             + (search.unicast           ? " (unicast)" : "")
                                             + (search.reply_required    ? " (reply required)" : "")
-                                            + (search.reply_to_src_port ? " (reply to source port)" : ""));
+                                            + (search.reply_to_src_port ? (origin == null ?  " (reply to source port)"  : " (reply to source port ignored because of origin tag)") : ""));
                 search.channels.add(new Channel(cid, name));
             }
         }
@@ -221,10 +231,11 @@ public static void encode(final boolean unicast, final boolean use_src_port,
                               final InetSocketAddress address, final boolean tls,
                               final ByteBuffer buffer)
     {
+        final int start = buffer.position();
         // Create with zero payload size, to be patched later
         PVAHeader.encodeMessageHeader(buffer, PVAHeader.FLAG_NONE, PVAHeader.CMD_SEARCH, 0);
 
-        final int payload_start = buffer.position();
+        final int payload_start = start + PVAHeader.HEADER_SIZE;
 
         // SEARCH message sequence
         // PVXS sends "find".getBytes() instead
@@ -282,6 +293,6 @@ public static void encode(final boolean unicast, final boolean use_src_port,
         }
 
         // Update payload size
-        buffer.putInt(PVAHeader.HEADER_OFFSET_PAYLOAD_SIZE, buffer.position() - payload_start);
+        buffer.putInt(start + PVAHeader.HEADER_OFFSET_PAYLOAD_SIZE, buffer.position() - payload_start);
     }
 }
diff --git a/core/pva/src/main/java/org/epics/pva/common/TCPHandler.java b/core/pva/src/main/java/org/epics/pva/common/TCPHandler.java
@@ -394,14 +394,14 @@ private ByteBuffer assertBufferSize(final ByteBuffer buffer, final int size)
      */
     private void handleMessage(final ByteBuffer buffer) throws Exception
     {
-        final byte flags = buffer.get(2);
+        final byte flags = buffer.get(PVAHeader.HEADER_OFFSET_FLAGS);
         final byte segemented = (byte) (flags & PVAHeader.FLAG_SEGMENT_MASK);
         if (segemented != 0)
             handleSegmentedMessage(segemented, buffer);
         else
         {
             final boolean control = (flags & PVAHeader.FLAG_CONTROL) != 0;
-            final byte command = buffer.get(3);
+            final byte command = buffer.get(PVAHeader.HEADER_OFFSET_COMMAND);
             // Move to start of potential payload
             if (buffer.limit() >= 8)
                 buffer.position(8);
@@ -462,11 +462,11 @@ else if (segments.position() > 0)
             if (segments == null  ||  segments.position() <= 0)
                 throw new Exception("Received " + (last ? "last" : "middle") + " message segment without first segment");
             // Check if command matches the one in first segment
-            final byte seg_command = segments.get(3);
-            if (seg_command != buffer.get(3))
+            final byte seg_command = segments.get(PVAHeader.HEADER_OFFSET_COMMAND);
+            if (seg_command != buffer.get(PVAHeader.HEADER_OFFSET_COMMAND))
                 throw new Exception(String.format("Received " + (last ? "last" : "middle") +
                                                   " message segment for command 0x%02X after first segment for command 0x%02X",
-                                                  buffer.get(3), seg_command));
+                                                  buffer.get(PVAHeader.HEADER_OFFSET_COMMAND), seg_command));
 
             // Size of segments accumulated so far..
             final int seg_size = segments.getInt(PVAHeader.HEADER_OFFSET_PAYLOAD_SIZE);
diff --git a/core/pva/src/main/java/org/epics/pva/server/SearchCommandHandler.java b/core/pva/src/main/java/org/epics/pva/server/SearchCommandHandler.java
@@ -30,7 +30,7 @@ public void handleCommand(final ServerTCPHandler tcp, final ByteBuffer buffer) t
         final byte version = buffer.get(PVAHeader.HEADER_OFFSET_VERSION);
         final int payload_size = buffer.getInt(PVAHeader.HEADER_OFFSET_PAYLOAD_SIZE);
 
-        final SearchRequest search = SearchRequest.decode(tcp.getRemoteAddress(), version, payload_size, buffer);
+        final SearchRequest search = SearchRequest.decode(null, tcp.getRemoteAddress(), version, payload_size, buffer);
 
         if (search.channels != null)
             for (SearchRequest.Channel channel : search.channels)
diff --git a/core/pva/src/main/java/org/epics/pva/server/ServerUDPHandler.java b/core/pva/src/main/java/org/epics/pva/server/ServerUDPHandler.java
diff --git a/core/pva/src/test/java/org/epics/pva/server/BoolDemo.java b/core/pva/src/test/java/org/epics/pva/server/BoolDemo.java
