Repository: netty/netty
Issue: DatagramPacket.recipient() doesn't return the actual destination IP, but the IP the app is bound to

I'm using epoll (on previous versions of my app I was using NIO and I think the behaviour is the same) on an application that is listening on all available interfaces (0.0.0.0). When I try to get the actual UDP packet destination IP, DatagramPacket.recipient() returns 0.0.0.0 and not the real packet destination IP. It seems to always return the IP the app is bound to.

Please, let me know if you need some extra information.

SO: Ubuntu 14.04.4
Netty version: 4.0.32.Final


Issue comments:
Will check...

Looks like netty cannot get the udp destination address. See:

http://developerweb.net/viewtopic.php?id=4342
http://developerweb.net/viewtopic.php?id=4169

Thanks Windie for your comments. I think I have a better understanding of the problem now. IP_PKTINFO socket option should be set, in order to get the ancillary message with required extra information (Destination IP and more) available.
Java doesn't seem to provide any API to set this socket option, but I've searched this repo and epoll sockets seem to be created with C code (https://github.com/netty/netty/blob/4.0/transport-native-epoll/src/main/c/netty_unix_socket.c). So, maybe is still possible to have the UDP destination address available... 

@mpolonio yep I think I can make it work for native transport. Stay tuned.

Hi @normanmaurer, did you have the time to take a closer look on the issue?

@mpolonio not yet... sorry super busy and not high priority atm :(

Do you have any plans to address this?
Sorry but I had no cycles yet to do this and its not very high on my to do list. That said we love contributions.
Thanks a lot
Although this seems to be fixed this quite does not work for us. We are using a loading balance with KeepAlive which receives traffic via VIP address. Than it is routed to individual hosts via Direct Routing. Direct Routing works by replacing the target MAC in a package addressed to the loadbalancer to its virtual / loadbalancer IP with the MAC of the designated target server and re-sending it. This requires the servers to accept packages for the given IP, so they need to configure the corresponding IP on some local looopback / dummy device. 
For this we have setup dummy interfaces with VIP address on the server machines. UDP packages are received on the netty server that has been bootstrapped with IP_RECVORIGDSTADDR. 
Despite this msg.recipient() does not return the VIP address but the address of the server it has been routed for. I am not an expert for linux/bsd socketing so I am not sure if this is even supposed to work but I know there is software written in C (FreeRadius) which reads the correct address (VIP) as destination address of the received packet. 
We are doing this as we need to respond to the UDP packet from the same received address  specified by Radius Protocol.
Any idea why this is not working and how it can be fixed?  

Pull request review comments:
final

Patch that resolved the issue:
diff --git a/transport-native-epoll/src/main/c/netty_epoll_linuxsocket.c b/transport-native-epoll/src/main/c/netty_epoll_linuxsocket.c
index 562eaa0b537..d6e3ead94fe 100644
--- a/transport-native-epoll/src/main/c/netty_epoll_linuxsocket.c
+++ b/transport-native-epoll/src/main/c/netty_epoll_linuxsocket.c
@@ -107,6 +107,10 @@ static void netty_epoll_linuxsocket_setIpTransparent(JNIEnv* env, jclass clazz,
     netty_unix_socket_setOption(env, fd, SOL_IP, IP_TRANSPARENT, &optval, sizeof(optval));
 }
 
+static void netty_epoll_linuxsocket_setIpRecvOrigDestAddr(JNIEnv* env, jclass clazz, jint fd, jint optval) {
+    netty_unix_socket_setOption(env, fd, IPPROTO_IP, IP_RECVORIGDSTADDR, &optval, sizeof(optval));
+}
+
 static void netty_epoll_linuxsocket_setTcpMd5Sig(JNIEnv* env, jclass clazz, jint fd, jbyteArray address, jint scopeId, jbyteArray key) {
     struct sockaddr_storage addr;
     socklen_t addrSize;
@@ -193,6 +197,14 @@ static jint netty_epoll_linuxsocket_isIpTransparent(JNIEnv* env, jclass clazz, j
      return optval;
 }
 
+static jint netty_epoll_linuxsocket_isIpRecvOrigDestAddr(JNIEnv* env, jclass clazz, jint fd) {
+     int optval;
+     if (netty_unix_socket_getOption(env, fd, IPPROTO_IP, IP_RECVORIGDSTADDR, &optval, sizeof(optval)) == -1) {
+         return -1;
+     }
+     return optval;
+}
+
 static void netty_epoll_linuxsocket_getTcpInfo(JNIEnv* env, jclass clazz, jint fd, jlongArray array) {
      struct tcp_info tcp_info;
      if (netty_unix_socket_getOption(env, fd, IPPROTO_TCP, TCP_INFO, &tcp_info, sizeof(tcp_info)) == -1) {
@@ -345,12 +357,14 @@ static const JNINativeMethod fixed_method_table[] = {
   { "setTcpUserTimeout", "(II)V", (void *) netty_epoll_linuxsocket_setTcpUserTimeout },
   { "setIpFreeBind", "(II)V", (void *) netty_epoll_linuxsocket_setIpFreeBind },
   { "setIpTransparent", "(II)V", (void *) netty_epoll_linuxsocket_setIpTransparent },
+  { "setIpRecvOrigDestAddr", "(II)V", (void *) netty_epoll_linuxsocket_setIpRecvOrigDestAddr },
   { "getTcpKeepIdle", "(I)I", (void *) netty_epoll_linuxsocket_getTcpKeepIdle },
   { "getTcpKeepIntvl", "(I)I", (void *) netty_epoll_linuxsocket_getTcpKeepIntvl },
   { "getTcpKeepCnt", "(I)I", (void *) netty_epoll_linuxsocket_getTcpKeepCnt },
   { "getTcpUserTimeout", "(I)I", (void *) netty_epoll_linuxsocket_getTcpUserTimeout },
   { "isIpFreeBind", "(I)I", (void *) netty_epoll_linuxsocket_isIpFreeBind },
   { "isIpTransparent", "(I)I", (void *) netty_epoll_linuxsocket_isIpTransparent },
+  { "isIpRecvOrigDestAddr", "(I)I", (void *) netty_epoll_linuxsocket_isIpRecvOrigDestAddr },
   { "getTcpInfo", "(I[J)V", (void *) netty_epoll_linuxsocket_getTcpInfo },
   { "setTcpMd5Sig", "(I[BI[B)V", (void *) netty_epoll_linuxsocket_setTcpMd5Sig }
   // "sendFile" has a dynamic signature
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollChannelOption.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollChannelOption.java
index 3ade823275b..d03e3e72367 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollChannelOption.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollChannelOption.java
@@ -31,6 +31,7 @@ public final class EpollChannelOption<T> extends UnixChannelOption<T> {
             valueOf(EpollChannelOption.class, "TCP_USER_TIMEOUT");
     public static final ChannelOption<Boolean> IP_FREEBIND = valueOf("IP_FREEBIND");
     public static final ChannelOption<Boolean> IP_TRANSPARENT = valueOf("IP_TRANSPARENT");
+    public static final ChannelOption<Boolean> IP_RECVORIGDSTADDR = valueOf("IP_RECVORIGDSTADDR");
     public static final ChannelOption<Integer> TCP_FASTOPEN = valueOf(EpollChannelOption.class, "TCP_FASTOPEN");
     public static final ChannelOption<Boolean> TCP_FASTOPEN_CONNECT =
             valueOf(EpollChannelOption.class, "TCP_FASTOPEN_CONNECT");
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java
index 8cbc31adde7..a19191d5967 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java
@@ -470,13 +470,18 @@ void epollInReady() {
                             break;
                         }
 
+                        InetSocketAddress localAddress = remoteAddress.localAddress();
+                        if (localAddress == null) {
+                            localAddress = (InetSocketAddress) localAddress();
+                        }
+
                         allocHandle.incMessagesRead(1);
                         allocHandle.lastBytesRead(remoteAddress.receivedAmount());
                         data.writerIndex(data.writerIndex() + allocHandle.lastBytesRead());
 
                         readPending = false;
                         pipeline.fireChannelRead(
-                                new DatagramPacket(data, (InetSocketAddress) localAddress(), remoteAddress));
+                                new DatagramPacket(data, localAddress, remoteAddress));
 
                         data = null;
                     } while (allocHandle.continueReading());
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannelConfig.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannelConfig.java
index 0f8e86ec1d9..fbc44c1bcc1 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannelConfig.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannelConfig.java
@@ -49,7 +49,8 @@ public Map<ChannelOption<?>, Object> getOptions() {
                 ChannelOption.SO_REUSEADDR, ChannelOption.IP_MULTICAST_LOOP_DISABLED,
                 ChannelOption.IP_MULTICAST_ADDR, ChannelOption.IP_MULTICAST_IF, ChannelOption.IP_MULTICAST_TTL,
                 ChannelOption.IP_TOS, ChannelOption.DATAGRAM_CHANNEL_ACTIVE_ON_REGISTRATION,
-                EpollChannelOption.SO_REUSEPORT, EpollChannelOption.IP_TRANSPARENT);
+                EpollChannelOption.SO_REUSEPORT, EpollChannelOption.IP_TRANSPARENT,
+                EpollChannelOption.IP_RECVORIGDSTADDR);
     }
 
     @SuppressWarnings({ "unchecked", "deprecation" })
@@ -91,6 +92,9 @@ public <T> T getOption(ChannelOption<T> option) {
         if (option == EpollChannelOption.IP_TRANSPARENT) {
             return (T) Boolean.valueOf(isIpTransparent());
         }
+        if (option == EpollChannelOption.IP_RECVORIGDSTADDR) {
+            return (T) Boolean.valueOf(isIpRecvOrigDestAddr());
+        }
         return super.getOption(option);
     }
 
@@ -123,6 +127,8 @@ public <T> boolean setOption(ChannelOption<T> option, T value) {
             setReusePort((Boolean) value);
         } else if (option == EpollChannelOption.IP_TRANSPARENT) {
             setIpTransparent((Boolean) value);
+        } else if (option == EpollChannelOption.IP_RECVORIGDSTADDR) {
+            setIpRecvOrigDestAddr((Boolean) value);
         } else {
             return super.setOption(option, value);
         }
@@ -403,4 +409,29 @@ public EpollDatagramChannelConfig setIpTransparent(boolean ipTransparent) {
         }
     }
 
+    /**
+     * Returns {@code true} if <a href="http://man7.org/linux/man-pages/man7/ip.7.html">IP_RECVORIGDSTADDR</a> is
+     * enabled, {@code false} otherwise.
+     */
+    public boolean isIpRecvOrigDestAddr() {
+        try {
+            return datagramChannel.socket.isIpRecvOrigDestAddr();
+        } catch (IOException e) {
+            throw new ChannelException(e);
+        }
+    }
+
+    /**
+     * If {@code true} is used <a href="http://man7.org/linux/man-pages/man7/ip.7.html">IP_RECVORIGDSTADDR</a> is
+     * enabled, {@code false} for disable it. Default is disabled.
+     */
+    public EpollDatagramChannelConfig setIpRecvOrigDestAddr(boolean ipTransparent) {
+        try {
+            datagramChannel.socket.setIpRecvOrigDestAddr(ipTransparent);
+            return this;
+        } catch (IOException e) {
+            throw new ChannelException(e);
+        }
+    }
+
 }
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/LinuxSocket.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/LinuxSocket.java
index af7e47a8709..f0a5304a74c 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/LinuxSocket.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/LinuxSocket.java
@@ -99,6 +99,10 @@ void setIpTransparent(boolean enabled) throws IOException {
         setIpTransparent(intValue(), enabled ? 1 : 0);
     }
 
+    void setIpRecvOrigDestAddr(boolean enabled) throws IOException {
+        setIpRecvOrigDestAddr(intValue(), enabled ? 1 : 0);
+    }
+
     void getTcpInfo(EpollTcpInfo info) throws IOException {
         getTcpInfo(intValue(), info.info);
     }
@@ -148,6 +152,10 @@ boolean isIpTransparent() throws IOException {
         return isIpTransparent(intValue()) != 0;
     }
 
+    boolean isIpRecvOrigDestAddr() throws IOException {
+        return isIpRecvOrigDestAddr(intValue()) != 0;
+    }
+
     PeerCredentials getPeerCredentials() throws IOException {
         return getPeerCredentials(intValue());
     }
@@ -189,6 +197,7 @@ private static native long sendFile(int socketFd, DefaultFileRegion src, long ba
     private static native int getTcpUserTimeout(int fd) throws IOException;
     private static native int isIpFreeBind(int fd) throws IOException;
     private static native int isIpTransparent(int fd) throws IOException;
+    private static native int isIpRecvOrigDestAddr(int fd) throws IOException;
     private static native void getTcpInfo(int fd, long[] array) throws IOException;
     private static native PeerCredentials getPeerCredentials(int fd) throws IOException;
     private static native int isTcpFastOpenConnect(int fd) throws IOException;
@@ -205,5 +214,6 @@ private static native long sendFile(int socketFd, DefaultFileRegion src, long ba
     private static native void setTcpUserTimeout(int fd, int milliseconds)throws IOException;
     private static native void setIpFreeBind(int fd, int freeBind) throws IOException;
     private static native void setIpTransparent(int fd, int transparent) throws IOException;
+    private static native void setIpRecvOrigDestAddr(int fd, int transparent) throws IOException;
     private static native void setTcpMd5Sig(int fd, byte[] address, int scopeId, byte[] key) throws IOException;
 }
diff --git a/transport-native-unix-common/src/main/c/netty_unix_socket.c b/transport-native-unix-common/src/main/c/netty_unix_socket.c
index 7ab1a1e3c91..a9e0e11dd17 100644
--- a/transport-native-unix-common/src/main/c/netty_unix_socket.c
+++ b/transport-native-unix-common/src/main/c/netty_unix_socket.c
@@ -68,7 +68,7 @@ static int nettyNonBlockingSocket(int domain, int type, int protocol) {
 #endif
 }
 
-static jobject createDatagramSocketAddress(JNIEnv* env, const struct sockaddr_storage* addr, int len) {
+static jobject createDatagramSocketAddress(JNIEnv* env, const struct sockaddr_storage* addr, int len, jobject local) {
     char ipstr[INET6_ADDRSTRLEN];
     int port;
     jstring ipString;
@@ -93,7 +93,7 @@ static jobject createDatagramSocketAddress(JNIEnv* env, const struct sockaddr_st
             ipString = (*env)->NewStringUTF(env, ipstr);
         }
     }
-    jobject socketAddr = (*env)->NewObject(env, datagramSocketAddressClass, datagramSocketAddrMethodId, ipString, port, len);
+    jobject socketAddr = (*env)->NewObject(env, datagramSocketAddressClass, datagramSocketAddrMethodId, ipString, port, len, local);
     return socketAddr;
 }
 
@@ -289,12 +289,39 @@ static jint _sendTo(JNIEnv* env, jint fd, void* buffer, jint pos, jint limit, jb
 
 static jobject _recvFrom(JNIEnv* env, jint fd, void* buffer, jint pos, jint limit) {
     struct sockaddr_storage addr;
+    struct sockaddr_storage daddr;
     socklen_t addrlen = sizeof(addr);
+    char cntrlbuf[64];
+    struct iovec iov;
+    struct msghdr msg;
     ssize_t res;
     int err;
+    struct cmsghdr* cmsg;
+    jobject local = NULL;
+    int readLocalAddr;
+
+    if (netty_unix_socket_getOption(env, fd, IPPROTO_IP, IP_RECVORIGDSTADDR,
+            &readLocalAddr, sizeof(readLocalAddr)) < 0) {
+        readLocalAddr = 0;
+    }
+
+    if (readLocalAddr) {
+        iov.iov_base = buffer + pos;
+        iov.iov_len  = (size_t) (limit - pos);
+        msg.msg_name = (struct sockaddr*) &addr;
+        msg.msg_namelen = addrlen;
+        msg.msg_iov = &iov;
+        msg.msg_iovlen = 1;
+        msg.msg_control = cntrlbuf;
+        msg.msg_controllen = sizeof(cntrlbuf);
+    }
 
     do {
-        res = recvfrom(fd, buffer + pos, (size_t) (limit - pos), 0, (struct sockaddr*) &addr, &addrlen);
+        if (readLocalAddr) {
+            res = recvmsg(fd, &msg, 0);
+        } else {
+            res = recvfrom(fd, buffer + pos, (size_t) (limit - pos), 0, (struct sockaddr*) &addr, &addrlen);
+        }
         // Keep on reading if we was interrupted
     } while (res == -1 && ((err = errno) == EINTR));
 
@@ -315,7 +342,16 @@ static jobject _recvFrom(JNIEnv* env, jint fd, void* buffer, jint pos, jint limi
         return NULL;
     }
 
-    return createDatagramSocketAddress(env, &addr, res);
+    if (readLocalAddr) {
+        for (cmsg = CMSG_FIRSTHDR(&msg); cmsg != NULL; cmsg = CMSG_NXTHDR(&msg, cmsg)) {
+            if (cmsg->cmsg_level == SOL_IP && cmsg->cmsg_type == IP_RECVORIGDSTADDR) {
+                memcpy (&daddr, CMSG_DATA(cmsg), sizeof (struct sockaddr_storage));
+                local = createDatagramSocketAddress(env, &daddr, res, NULL);
+                break;
+            }
+        }
+    }
+    return createDatagramSocketAddress(env, &addr, res, local);
 }
 
 void netty_unix_socket_getOptionHandleError(JNIEnv* env, int err) {
@@ -977,9 +1013,9 @@ jint netty_unix_socket_JNI_OnLoad(JNIEnv* env, const char* packagePrefix) {
         netty_unix_errors_throwOutOfMemoryError(env);
         return JNI_ERR;
     }
-    datagramSocketAddrMethodId = (*env)->GetMethodID(env, datagramSocketAddressClass, "<init>", "(Ljava/lang/String;II)V");
+    datagramSocketAddrMethodId = (*env)->GetMethodID(env, datagramSocketAddressClass, "<init>", "(Ljava/lang/String;IILio/netty/channel/unix/DatagramSocketAddress;)V");
     if (datagramSocketAddrMethodId == NULL) {
-        netty_unix_errors_throwRuntimeException(env, "failed to get method ID: DatagramSocketAddress.<init>(String, int, int)");
+        netty_unix_errors_throwRuntimeException(env, "failed to get method ID: DatagramSocketAddress.<init>(String, int, int, DatagramSocketAddress)");
         return JNI_ERR;
     }
     jclass localInetSocketAddressClass = (*env)->FindClass(env, "java/net/InetSocketAddress");
diff --git a/transport-native-unix-common/src/main/java/io/netty/channel/unix/DatagramSocketAddress.java b/transport-native-unix-common/src/main/java/io/netty/channel/unix/DatagramSocketAddress.java
index 21d7215a49c..874d7054f51 100644
--- a/transport-native-unix-common/src/main/java/io/netty/channel/unix/DatagramSocketAddress.java
+++ b/transport-native-unix-common/src/main/java/io/netty/channel/unix/DatagramSocketAddress.java
@@ -28,10 +28,16 @@ public final class DatagramSocketAddress extends InetSocketAddress {
 
     // holds the amount of received bytes
     private final int receivedAmount;
+    private final DatagramSocketAddress localAddress;
 
-    DatagramSocketAddress(String addr, int port, int receivedAmount) {
+    DatagramSocketAddress(String addr, int port, int receivedAmount, DatagramSocketAddress local) {
         super(addr, port);
         this.receivedAmount = receivedAmount;
+        localAddress = local;
+    }
+
+    public DatagramSocketAddress localAddress() {
+        return localAddress;
     }
 
     public int receivedAmount() {


Test patch:
diff --git a/testsuite/src/main/java/io/netty/testsuite/transport/socket/DatagramUnicastTest.java b/testsuite/src/main/java/io/netty/testsuite/transport/socket/DatagramUnicastTest.java
index 26743ba8c70..b49f6780ef5 100644
--- a/testsuite/src/main/java/io/netty/testsuite/transport/socket/DatagramUnicastTest.java
+++ b/testsuite/src/main/java/io/netty/testsuite/transport/socket/DatagramUnicastTest.java
@@ -277,6 +277,10 @@ public void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws E
                         for (byte b : bytes) {
                             assertEquals(b, buf.readByte());
                         }
+
+                        // Test that the channel's localAddress is equal to the message's recipient
+                        assertEquals(ctx.channel().localAddress(), msg.recipient());
+
                         latch.countDown();
                     }
                 });
diff --git a/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollDatagramUnicastTest.java b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollDatagramUnicastTest.java
index 610cfb5d851..a8ed58ed257 100644
--- a/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollDatagramUnicastTest.java
+++ b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollDatagramUnicastTest.java
@@ -26,4 +26,11 @@ public class EpollDatagramUnicastTest extends DatagramUnicastTest {
     protected List<TestsuitePermutation.BootstrapComboFactory<Bootstrap, Bootstrap>> newFactories() {
         return EpollSocketTestPermutation.INSTANCE.datagram();
     }
+
+    public void testSimpleSendWithConnect(Bootstrap sb, Bootstrap cb) throws Throwable {
+        // Run this test with IP_RECVORIGDSTADDR option enabled
+        sb.option(EpollChannelOption.IP_RECVORIGDSTADDR, true);
+        super.testSimpleSendWithConnect(sb, cb);
+        sb.option(EpollChannelOption.IP_RECVORIGDSTADDR, false);
+    }
 }