squash-7

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ## [Unreleased]
 
 ### TODO for 0.4.0
+- Cache local address, see AbstractChannel:600
+  srcAddress = getOrCreateService().getExternalIP(path.getFirstHopAddress());
+- Cache paths
+- Server should get firstHop from topofile or daemon, not from packet IP!
+  -> BRs may use different ports for in/outgoing traffic.
+  -> Thios would also solve the server/SHIM problem, see TODO.md 
 - Fix @Disabled tests
 - Support topofile port range
   - Upgrade all JUnit topo files to post 0.11 with new format (including port range)
@@ -21,6 +27,14 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
   - Inherit DatagramChannel 
 - Consider using https://github.com/ascopes/protobuf-maven-plugin (more up to date) 
 
+### Added
+- Support for `dispatched_ports` in topo files
+  [#130](https://github.com/scionproto-contrib/jpan/pull/130)
+
+TODO: 
+- Test demos with JUNIT topo and with scioproto topo
+- FIX: if no ephemeral ports are available, just use ANY port but enforce 30041 as return port.
+
 ## [0.3.1] - 2024-10-11
 
 ### Added

TODO.md

@@ -33,9 +33,6 @@
   - Implement interfaces from nio.DatagramChannel
   - Look into Selectors:  https://www.baeldung.com/java-nio-selector
 - Consider subclassing DatagramChannel directly.
-- DISPATCHER migration:
-  - Daemon supposedly provides information about dispatcher. Double check updated proto files
-  - Parse topofiles with port range information -> indicates DISPATCHER presence
 - Consider SHIM support. SHIM is a compatibility component that supports
   old border-router software (requiring a fixed port on the client, unless
   the client is listening on this very port).  When SHIM is used, we cannot 

src/main/java/org/scion/jpan/AbstractDatagramChannel.java

@@ -25,10 +25,7 @@
 import java.util.Objects;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
-import org.scion.jpan.internal.ExtensionHeader;
-import org.scion.jpan.internal.InternalConstants;
-import org.scion.jpan.internal.ScionHeaderParser;
-import org.scion.jpan.internal.ScmpParser;
+import org.scion.jpan.internal.*;
 
 abstract class AbstractDatagramChannel<C extends AbstractDatagramChannel<?>> implements Closeable {
 
@@ -150,10 +147,26 @@ public C bind(InetSocketAddress address) throws IOException {
     }
   }
 
-  private void ensureBound() throws IOException {
+  protected void ensureBound() throws IOException {
     synchronized (stateLock) {
       if (localAddress == null) {
-        bind(null);
+        LocalTopology.DispatcherPortRange ports = getOrCreateService().getLocalPortRange();
+        if (ports != null && ports.hasPortRange()) {
+          // This is a bit ugly, we iterate through all ports to find a free one.
+          int min = ports.getPortMin();
+          int max = ports.getPortMax();
+          for (int port = min; port <= max; port++) {
+            try {
+              bind(new InetSocketAddress(port));
+              return;
+            } catch (IOException e) {
+              // ignore and try next port
+            }
+          }
+          throw new IOException("No free port found in SCION port range: " + min + "-" + max);
+        } else {
+          bind(null);
+        }
       }
     }
   }
@@ -304,6 +317,14 @@ public C connect(Path path) throws IOException {
     synchronized (stateLock) {
       checkConnected(false);
       ensureBound();
+      if (localAddress.isAnyLocalAddress()) {
+        // Do we really need this?
+        // - It ensures that after connect we have a proper local address for getLocalAddress(),
+        //   this is what connect() should do.
+        // - It allows us to have an ANY address underneath which could help with interface
+        //   switching.
+        localAddress = getOrCreateService().getExternalIP(path.getFirstHopAddress());
+      }
       updateConnection((RequestPath) path, false);
       return (C) this;
     }

src/main/java/org/scion/jpan/ScionService.java

@@ -30,6 +30,7 @@
 import static org.scion.jpan.Constants.PROPERTY_DNS_SEARCH_DOMAINS;
 import static org.scion.jpan.Constants.PROPERTY_USE_OS_SEARCH_DOMAINS;
 
+import com.google.protobuf.Empty;
 import io.grpc.*;
 import java.io.IOException;
 import java.net.InetAddress;
@@ -75,6 +76,7 @@ public class ScionService {
   private final ScionBootstrapper bootstrapper;
   private final DaemonServiceGrpc.DaemonServiceBlockingStub daemonStub;
   private final SegmentLookupServiceGrpc.SegmentLookupServiceBlockingStub segmentStub;
+  private LocalTopology.DispatcherPortRange portRange;
 
   private final boolean minimizeRequests;
   private final ManagedChannel channel;
@@ -653,4 +655,28 @@ List<String> getBorderRouterStrings() {
       return bootstrapper.getLocalTopology().getBorderRouterAddresses();
     }
   }
+
+  LocalTopology.DispatcherPortRange getLocalPortRange() {
+    if (portRange == null) {
+      if (bootstrapper != null) {
+        portRange = bootstrapper.getLocalTopology().getPortRange();
+      } else if (daemonStub != null) {
+        // try daemon
+        Daemon.PortRangeResponse response;
+        try {
+          response = daemonStub.portRange(Empty.getDefaultInstance());
+          portRange =
+              LocalTopology.DispatcherPortRange.create(
+                  response.getDispatchedPortStart(), response.getDispatchedPortEnd());
+        } catch (StatusRuntimeException e) {
+          LOG.warn("ERROR getting port range from daemon: {}", e.getMessage());
+          // Daemon doesn't support port range.
+          portRange = LocalTopology.DispatcherPortRange.createEmpty();
+        }
+      } else {
+        portRange = LocalTopology.DispatcherPortRange.createAll();
+      }
+    }
+    return portRange;
+  }
 }

src/main/java/org/scion/jpan/ScmpSenderAsync.java

@@ -33,6 +33,7 @@
 
 public class ScmpSenderAsync implements AutoCloseable {
   private static final Logger log = LoggerFactory.getLogger(ScmpSenderAsync.class);
+  private static final int PORT_NOT_SET = -1;
   private int timeOutMs = 1000;
   private final InternalChannel channel;
   private final AtomicInteger sequenceIDs = new AtomicInteger(0);
@@ -215,12 +216,8 @@ protected InternalChannel(
       super.channel().configureBlocking(false);
       super.channel().register(selector, SelectionKey.OP_READ);
 
-      if (port == 0) {
-        if (Util.getJavaMajorVersion() >= 17) {
-          super.bind(null);
-        } else {
-          throw new IllegalArgumentException("With Java < 17, Please assign a local port >= 0");
-        }
+      if (port == PORT_NOT_SET) {
+        ensureBound();
       } else {
         // listen on ANY interface: 0.0.0.0 / [::]
         super.bind(new InetSocketAddress(port));
@@ -406,7 +403,7 @@ public void run() {
 
   public static class Builder {
     private ScionService service;
-    private int port = -1;
+    private int port = PORT_NOT_SET;
     private final ResponseHandler handler;
     private java.nio.channels.DatagramChannel channel = null;
     private Selector selector = null;
@@ -417,9 +414,6 @@ private Builder(ResponseHandler handler) {
 
     public Builder setLocalPort(int localPort) {
       this.port = localPort;
-      if (port == 0 && Util.getJavaMajorVersion() < 17) {
-        log.warn("Using port 0 does likely not work with Java < 17");
-      }
       return this;
     }
 
@@ -443,13 +437,6 @@ public ScmpSenderAsync build() {
       try {
         channel = channel == null ? java.nio.channels.DatagramChannel.open() : channel;
         selector = selector == null ? Selector.open() : selector;
-        if (port == -1) {
-          if (Util.getJavaMajorVersion() >= 17) {
-            port = 0;
-          } else {
-            port = 51315; // Some random port
-          }
-        }
         return new ScmpSenderAsync(service, port, handler, channel, selector);
       } catch (IOException e) {
         throw new ScionRuntimeException(e);

src/main/java/org/scion/jpan/internal/LocalTopology.java

@@ -21,6 +21,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import org.scion.jpan.Constants;
 import org.scion.jpan.ScionRuntimeException;
 import org.scion.jpan.ScionUtil;
 
@@ -33,6 +34,44 @@ public class LocalTopology {
   private String localIsdAs;
   private boolean isCoreAs;
   private int localMtu;
+  private DispatcherPortRange portRange;
+
+  public static class DispatcherPortRange {
+    private final int portMin;
+    private final int portMax;
+
+    private DispatcherPortRange(int min, int max) {
+      portMin = min;
+      portMax = max;
+    }
+
+    public static DispatcherPortRange create(int min, int max) {
+      return new DispatcherPortRange(min, max);
+    }
+
+    public static DispatcherPortRange createAll() {
+      return new DispatcherPortRange(-1, -1);
+    }
+
+    public static DispatcherPortRange createEmpty() {
+      // For the empty range we allow only port 30041.
+      // If the port is used by the SHIM, than we cannot connect. However, if there is no SHIM,
+      // then we can safely use 30041.
+      return new DispatcherPortRange(Constants.SCMP_PORT, Constants.SCMP_PORT);
+    }
+
+    public boolean hasPortRange() {
+      return portMin >= 1 && portMax <= 65535;
+    }
+
+    public int getPortMin() {
+      return portMin;
+    }
+
+    public int getPortMax() {
+      return portMax;
+    }
+  }
 
   public static synchronized LocalTopology create(String topologyFile) {
     LocalTopology topo = new LocalTopology();
@@ -123,6 +162,16 @@ private void parseTopologyFile(String topologyFile) {
           discoveryServices.add(new ServiceNode(e.getKey(), ds.get("addr").getAsString()));
         }
       }
+      JsonElement underlay = o.get("underlay");
+      JsonElement dispatchedPorts = o.get("dispatched_ports");
+      if (underlay == null && dispatchedPorts == null) {
+        portRange = DispatcherPortRange.createEmpty();
+      } else if (dispatchedPorts != null) {
+        portRange = parsePortRange(dispatchedPorts.getAsString());
+      } else {
+        JsonObject u = underlay.getAsJsonObject();
+        portRange = parsePortRange(u.get("dispatched_ports").getAsString());
+      }
       JsonArray attr = safeGet(o, "attributes").getAsJsonArray();
       for (int i = 0; i < attr.size(); i++) {
         if ("core".equals(attr.get(i).getAsString())) {
@@ -132,6 +181,32 @@ private void parseTopologyFile(String topologyFile) {
     }
   }
 
+  private static DispatcherPortRange parsePortRange(String v) {
+    if (v.startsWith("\"") && v.endsWith("\"")) {
+      v = v.substring(1, v.length() - 2);
+    }
+    if ("-".equals(v)) {
+      return DispatcherPortRange.createEmpty();
+    } else if ("all".equalsIgnoreCase(v)) {
+      return DispatcherPortRange.createAll();
+    } else {
+      String[] sa = v.split("-");
+      if (sa.length != 2) {
+        throw new ScionRuntimeException("Illegal expression in topo file dispatched_ports: " + v);
+      }
+      int portMin = Integer.parseInt(sa[0]);
+      int portMax = Integer.parseInt(sa[1]);
+      if (portMin < 1 || portMax < 1 || portMax > 65535 || portMin > portMax) {
+        throw new ScionRuntimeException("Illegal port values in topo file dispatched_ports: " + v);
+      }
+      return new DispatcherPortRange(portMin, portMax);
+    }
+  }
+
+  public DispatcherPortRange getPortRange() {
+    return portRange;
+  }
+
   @Override
   public String toString() {
     StringBuilder sb = new StringBuilder();

src/main/proto/scion/protobuf/daemon/v1/daemon.proto

@@ -21,6 +21,7 @@ package proto.daemon.v1;
 
 import "google/protobuf/timestamp.proto";
 import "google/protobuf/duration.proto";
+import "google/protobuf/empty.proto";
 import "scion/protobuf/drkey/v1/drkey.proto";
 
 service DaemonService {
@@ -36,6 +37,8 @@ service DaemonService {
     rpc Services(ServicesRequest) returns (ServicesResponse) {}
     // Inform the SCION Daemon of a revocation.
     rpc NotifyInterfaceDown(NotifyInterfaceDownRequest) returns (NotifyInterfaceDownResponse) {}
+    // Returns the endhost portRange defined in the local AS.
+    rpc PortRange(google.protobuf.Empty) returns (PortRangeResponse) {}
     // DRKeyASHost returns a key that matches the request.
     rpc DRKeyASHost (DRKeyASHostRequest) returns (DRKeyASHostResponse) {}
     // DRKeyHostAS returns a key that matches the request.
@@ -75,7 +78,8 @@ message Path {
     // Latency lists the latencies between any two consecutive interfaces.
     // Entry i describes the latency between interface i and i+1.
     // Consequently, there are N-1 entries for N interfaces.
-    // A 0-value indicates that the AS did not announce a latency for this hop.
+    // A negative value indicates that the AS did not announce a latency for
+    // this hop.
     repeated google.protobuf.Duration latency = 6;
     // Bandwidth lists the bandwidth between any two consecutive interfaces, in
     // Kbit/s.
@@ -200,6 +204,13 @@ message NotifyInterfaceDownRequest {
 
 message NotifyInterfaceDownResponse {};
 
+message PortRangeResponse {
+    // The lowest port in the SCION/UDP dispatched port range.
+    uint32 dispatched_port_start = 1;
+    // The highest port in the SCION/UDP dispatched port range.
+    uint32 dispatched_port_end = 2;
+}
+
 message DRKeyHostASRequest{
     // Point in time where requested key is valid.
     google.protobuf.Timestamp val_time = 1;

src/test/java/org/scion/jpan/api/DatagramChannelApiServerTest.java

@@ -74,6 +74,7 @@ void bind_withoutService() throws IOException {
   void send_withoutService() throws IOException {
     // check that send(Path) does not internally require a ScionService.
     try (ScionDatagramChannel channel = ScionDatagramChannel.open()) {
+      channel.bind(new InetSocketAddress("127.0.0.1", 12345));
       assertNull(channel.getService());
       ResponsePath path =
           PackageVisibilityHelper.createDummyResponsePath(
@@ -101,6 +102,7 @@ void receive_withoutService() throws IOException {
           return addr;
         });
     try (ScionDatagramChannel channel = ScionDatagramChannel.open(null, mock)) {
+      channel.bind(new InetSocketAddress("127.0.0.1", 12345));
       assertNull(channel.getService());
       ByteBuffer buffer = ByteBuffer.allocate(100);
       ScionSocketAddress responseAddress = channel.receive(buffer);

src/test/java/org/scion/jpan/api/DatagramChannelApiTest.java

@@ -101,6 +101,7 @@ void getLocalAddress_withConnect() throws IOException {
     try (ScionDatagramChannel channel = ScionDatagramChannel.open()) {
       channel.connect(dummyAddress);
       InetSocketAddress local = channel.getLocalAddress();
+      assertNotNull(local);
       assertFalse(local.getAddress().isAnyLocalAddress());
     }
   }