fix: Add strict hostname resolution and loopback check for advertised address

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieImpl.java

@@ -254,49 +254,56 @@ public static BookieId getBookieId(ServerConfiguration conf) throws UnknownHostE
      */
     public static BookieSocketAddress getBookieAddress(ServerConfiguration conf)
             throws UnknownHostException {
+        String hostAddress;
         // Advertised address takes precedence over the listening interface and the
         // useHostNameAsBookieID settings
         if (conf.getAdvertisedAddress() != null && conf.getAdvertisedAddress().trim().length() > 0) {
-            String hostAddress = conf.getAdvertisedAddress().trim();
-            return new BookieSocketAddress(hostAddress, conf.getBookiePort());
-        }
-
-        String iface = conf.getListeningInterface();
-        if (iface == null) {
-            iface = "default";
-        }
-
-        String hostAddress = DNS.getDefaultIP(iface);
-        if (conf.getUseHostNameAsBookieID()) {
-            try {
-                hostAddress = InetAddress.getByName(hostAddress).getCanonicalHostName();
-            } catch (Exception e) {
-                UnknownHostException unknownHostException =
-                        new UnknownHostException("Unable to resolve hostname for interface: " + iface);
-                unknownHostException.initCause(e);
-                throw unknownHostException;
+            hostAddress = conf.getAdvertisedAddress().trim();
+        } else {
+            String iface = conf.getListeningInterface();
+            if (iface == null) {
+                iface = "default";
             }
-            if (conf.getUseShortHostName()) {
-                /*
-                 * if short hostname is used, then FQDN is not used. Short
-                 * hostname is the hostname cut at the first dot.
-                 */
-                hostAddress = hostAddress.split("\\.", 2)[0];
+            String defaultIP = DNS.getDefaultIP(iface);
+            if (conf.getUseHostNameAsBookieID()) {
+                try {
+                    hostAddress = InetAddress.getByName(defaultIP).getCanonicalHostName();
+                } catch (Exception e) {
+                    UnknownHostException unknownHostException =
+                            new UnknownHostException("Unable to resolve hostname for interface: " + iface);
+                    unknownHostException.initCause(e);
+                    throw unknownHostException;
+                }
+                if (defaultIP.equals(hostAddress)) {
+                    throw new UnknownHostException("Unable to resolve hostname for ip address: " + defaultIP);
+                }
+                if (conf.getUseShortHostName()) {
+                    /*
+                     * if short hostname is used, then FQDN is not used. Short
+                     * hostname is the hostname cut at the first dot.
+                     */
+                    hostAddress = hostAddress.split("\\.", 2)[0];
+                }
+            } else {
+                hostAddress = defaultIP;
             }
         }
 
-        BookieSocketAddress addr =
+        BookieSocketAddress bookieSocketAddress =
                 new BookieSocketAddress(hostAddress, conf.getBookiePort());
-        if (addr.getSocketAddress().getAddress().isLoopbackAddress()
-            && !conf.getAllowLoopback()) {
+        InetAddress inetAddress = bookieSocketAddress.getSocketAddress().getAddress();
+        if (inetAddress == null) {
+            throw new UnknownHostException("Failed to resolve InetAddress for host address: " + hostAddress);
+        }
+        if (inetAddress.isLoopbackAddress() && !conf.getAllowLoopback()) {
             throw new UnknownHostException("Trying to listen on loopback address, "
-                    + addr + " but this is forbidden by default "
+                    + bookieSocketAddress + " but this is forbidden by default "
                     + "(see ServerConfiguration#getAllowLoopback()).\n"
                     + "If this happen, you can consider specifying the network interface"
                     + " to listen on (e.g. listeningInterface=eth0) or specifying the"
                     + " advertised address (e.g. advertisedAddress=172.x.y.z)");
         }
-        return addr;
+        return bookieSocketAddress;
     }
 
     public LedgerDirsManager getLedgerDirsManager() {

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java

@@ -1239,6 +1239,16 @@ public ServerConfiguration setBookieId(String bookieId) {
         return this;
     }
 
+    /**
+     * Remove the configured BookieId for the bookie.
+     *
+     * @return server configuration
+     */
+    public ServerConfiguration removeBookieId() {
+        this.setProperty(BOOKIE_ID, null);
+        return this;
+    }
+
     /**
      * Get the configured advertised address for the bookie.
      *

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieSocketAddressTest.java

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bookkeeper.bookie;
+
+import static org.apache.bookkeeper.bookie.BookieImpl.getBookieAddress;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.google.common.net.InetAddresses;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.junit.Test;
+
+public class BookieSocketAddressTest {
+
+    private void testAdvertisedWithLoopbackAddress(String address) throws UnknownHostException {
+        ServerConfiguration conf = new ServerConfiguration();
+        conf.setAdvertisedAddress(address);
+        conf.setAllowLoopback(false);
+        assertThatThrownBy(() -> getBookieAddress(conf)).isExactlyInstanceOf(UnknownHostException.class);
+
+        conf.setAllowLoopback(true);
+        BookieSocketAddress bookieAddress = getBookieAddress(conf);
+        assertThat(bookieAddress.getHostName()).isEqualTo(address);
+    }
+
+    @Test
+    public void testAdvertisedWithLoopbackAddress() throws UnknownHostException {
+        testAdvertisedWithLoopbackAddress("localhost");
+        testAdvertisedWithLoopbackAddress("127.0.0.1");
+    }
+
+    @Test
+    public void testAdvertisedWithNonLoopbackAddress() throws UnknownHostException {
+        String hostAddress = InetAddress.getLocalHost().getHostAddress();
+        if (hostAddress == null) {
+            throw new UnknownHostException("Host address is null");
+        }
+        ServerConfiguration conf = new ServerConfiguration();
+        conf.setAllowLoopback(false);
+        conf.setAdvertisedAddress(hostAddress);
+        BookieSocketAddress bookieAddress = getBookieAddress(conf);
+        assertThat(bookieAddress.getHostName()).isEqualTo(hostAddress);
+    }
+
+    @Test
+    public void testBookieAddressIsIPAddressByDefault() throws UnknownHostException {
+        ServerConfiguration conf = new ServerConfiguration();
+        BookieSocketAddress bookieAddress = getBookieAddress(conf);
+        assertThat(InetAddresses.isInetAddress(bookieAddress.getHostName())).isTrue();
+    }
+
+    @Test
+    public void testBookieAddressIsHostname() throws UnknownHostException {
+        ServerConfiguration conf = new ServerConfiguration();
+        conf.setUseHostNameAsBookieID(true);
+        BookieSocketAddress bookieAddress = getBookieAddress(conf);
+        assertThat(InetAddresses.isInetAddress(bookieAddress.getHostName())).isFalse();
+    }
+}

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieIndexDirTest.java

@@ -829,9 +829,10 @@ public void testRestartWithAdvertisedAddressAsBookieID() throws Exception {
             .setBookiePort(bookiePort)
             .setMetadataServiceUri(zkUtil.getMetadataServiceUri());
         conf.setUseHostNameAsBookieID(false);
+        conf.setAllowLoopback(true);
         validateConfig(conf);
 
-        conf.setAdvertisedAddress("unknown");
+        conf.setAdvertisedAddress("localhost");
         try {
             validateConfig(conf);
             fail("Should not start a bookie with ip if the bookie has been started with an ip");

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/datainteg/CookieValidationTest.java

@@ -75,7 +75,8 @@ private File initializedDir() throws Exception {
     private static ServerConfiguration serverConf(boolean stampMissingCookies) {
         ServerConfiguration conf = new ServerConfiguration();
         conf.setDataIntegrityStampMissingCookiesEnabled(stampMissingCookies);
-        conf.setAdvertisedAddress("foobar");
+        conf.setAdvertisedAddress("localhost");
+        conf.setAllowLoopback(true);
         return conf;
     }
 
@@ -255,7 +256,7 @@ public void testChangingBookieIdRaisesError() throws Exception {
                 conf, regManager, new MockDataIntegrityCheck());
         v1.checkCookies(dirs); // stamp original cookies
 
-        conf.setAdvertisedAddress("barfoo");
+        conf.setBookieId("barfoo");
         DataIntegrityCookieValidation v2 = new DataIntegrityCookieValidation(
                 conf, regManager, new MockDataIntegrityCheck());
         try {
@@ -265,7 +266,7 @@ public void testChangingBookieIdRaisesError() throws Exception {
             // expected
         }
 
-        conf.setAdvertisedAddress("foobar");
+        conf.removeBookieId();
         DataIntegrityCookieValidation v3 = new DataIntegrityCookieValidation(
                 conf, regManager, new MockDataIntegrityCheck());
         v3.checkCookies(dirs); // should succeed as the cookie is same as before

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/datainteg/DataIntegrityCheckTest.java

@@ -105,7 +105,8 @@ public void teardown() throws Exception {
 
     private static ServerConfiguration serverConf() {
         ServerConfiguration conf = new ServerConfiguration();
-        conf.setAdvertisedAddress("foobar");
+        conf.setAdvertisedAddress("localhost");
+        conf.setAllowLoopback(true);
         return conf;
     }