From 4e5586e0b94e0a522bdbad7148d42b71e20a410e Mon Sep 17 00:00:00 2001 From: root Date: Tue, 10 Mar 2026 13:25:24 -0400 Subject: [PATCH 1/2] Fix firefly experiment-id resolution for SciTag and FQAN fallback --- .../dcache/pool/movers/TransferLifeCycle.java | 16 ++- .../pool/movers/TransferLifeCycleTest.java | 97 +++++++++++++++++++ 2 files changed, 110 insertions(+), 3 deletions(-) create mode 100644 modules/dcache/src/test/java/org/dcache/pool/movers/TransferLifeCycleTest.java diff --git a/modules/dcache/src/main/java/org/dcache/pool/movers/TransferLifeCycle.java b/modules/dcache/src/main/java/org/dcache/pool/movers/TransferLifeCycle.java index d451d696a69..f6f0c578cce 100644 --- a/modules/dcache/src/main/java/org/dcache/pool/movers/TransferLifeCycle.java +++ b/modules/dcache/src/main/java/org/dcache/pool/movers/TransferLifeCycle.java @@ -254,7 +254,7 @@ private OptionalInt getExperimentId(ProtocolInfo protocolInfo, Subject subject) if (protocolInfo.getTransferTag() != null && !protocolInfo.getTransferTag().isEmpty()) { try { int transferTag = Integer.parseInt(protocolInfo.getTransferTag()); - if (transferTag <= 64 || transferTag >= 65536) { + if (transferTag < 64 || transferTag > 65535) { LOGGER.warn("Invalid integer range for transfer tag: {}", protocolInfo.getTransferTag()); return OptionalInt.empty(); } @@ -271,8 +271,18 @@ private OptionalInt getExperimentId(ProtocolInfo protocolInfo, Subject subject) return OptionalInt.empty(); } - return voToExpId.containsKey(vo.getGroup().toLowerCase()) - ? OptionalInt.of(voToExpId.get(vo.getGroup().toLowerCase())) + // FQAN.getGroup() returns paths like "/atlas" or "/atlas/usatlas". + // Strip the leading slash and take only the first path component to + // get the plain VO name (e.g. "atlas") that matches the vo-mapping keys. + String groupPath = vo.getGroup().toLowerCase(); + if (groupPath.startsWith("/")) { + groupPath = groupPath.substring(1); + } + int subgroupSlash = groupPath.indexOf('/'); + String voName = subgroupSlash != -1 ? groupPath.substring(0, subgroupSlash) : groupPath; + + return voToExpId.containsKey(voName) + ? OptionalInt.of(voToExpId.get(voName)) : OptionalInt.empty(); } diff --git a/modules/dcache/src/test/java/org/dcache/pool/movers/TransferLifeCycleTest.java b/modules/dcache/src/test/java/org/dcache/pool/movers/TransferLifeCycleTest.java new file mode 100644 index 00000000000..d030331d007 --- /dev/null +++ b/modules/dcache/src/test/java/org/dcache/pool/movers/TransferLifeCycleTest.java @@ -0,0 +1,97 @@ +package org.dcache.pool.movers; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import diskCacheV111.vehicles.ProtocolInfo; +import java.lang.reflect.Method; +import java.util.OptionalInt; +import javax.security.auth.Subject; +import org.dcache.auth.FQANPrincipal; +import org.junit.Before; +import org.junit.Test; + +public class TransferLifeCycleTest { + + private Method getExperimentId; + private TransferLifeCycle transferLifeCycle; + + @Before + public void setup() throws Exception { + transferLifeCycle = new TransferLifeCycle(); + transferLifeCycle.setVoMapping("atlas:2,cms:3"); + + getExperimentId = TransferLifeCycle.class.getDeclaredMethod( + "getExperimentId", ProtocolInfo.class, Subject.class); + getExperimentId.setAccessible(true); + } + + @Test + public void shouldAcceptMinimumValidSciTagValue() throws Exception { + OptionalInt experimentId = resolveExperimentId("64", new Subject()); + + assertTrue(experimentId.isPresent()); + assertEquals(1, experimentId.getAsInt()); + } + + @Test + public void shouldRejectSciTagValueBelowValidRange() throws Exception { + OptionalInt experimentId = resolveExperimentId("63", new Subject()); + + assertFalse(experimentId.isPresent()); + } + + @Test + public void shouldMapSlashPrefixedFqanToVoName() throws Exception { + Subject subject = new Subject(); + subject.getPrincipals().add(new FQANPrincipal("/atlas/usatlas", true)); + + OptionalInt experimentId = resolveExperimentId("", subject); + + assertTrue(experimentId.isPresent()); + assertEquals(2, experimentId.getAsInt()); + } + + private OptionalInt resolveExperimentId(String transferTag, Subject subject) throws Exception { + return (OptionalInt) getExperimentId.invoke(transferLifeCycle, + new TestProtocolInfo("xrootd", transferTag), subject); + } + + private static class TestProtocolInfo implements ProtocolInfo { + + private static final long serialVersionUID = 1L; + private final String protocol; + private final String transferTag; + + private TestProtocolInfo(String protocol, String transferTag) { + this.protocol = protocol; + this.transferTag = transferTag; + } + + @Override + public String getProtocol() { + return protocol; + } + + @Override + public int getMinorVersion() { + return 0; + } + + @Override + public int getMajorVersion() { + return 0; + } + + @Override + public String getVersionString() { + return "test"; + } + + @Override + public String getTransferTag() { + return transferTag; + } + } +} From 69ccf2d5bf3e92b41c31e10e087c662ecdf09b0b Mon Sep 17 00:00:00 2001 From: root Date: Tue, 10 Mar 2026 13:56:35 -0400 Subject: [PATCH 2/2] Require both endpoints to match firefly excludes --- .../dcache/pool/movers/TransferLifeCycle.java | 13 ++++-- .../pool/movers/TransferLifeCycleTest.java | 46 +++++++++++++++++++ 2 files changed, 54 insertions(+), 5 deletions(-) diff --git a/modules/dcache/src/main/java/org/dcache/pool/movers/TransferLifeCycle.java b/modules/dcache/src/main/java/org/dcache/pool/movers/TransferLifeCycle.java index f6f0c578cce..edef543751d 100644 --- a/modules/dcache/src/main/java/org/dcache/pool/movers/TransferLifeCycle.java +++ b/modules/dcache/src/main/java/org/dcache/pool/movers/TransferLifeCycle.java @@ -83,7 +83,7 @@ public void onStart(InetSocketAddress src, InetSocketAddress dst, ProtocolInfo p return; } - if (isLocalTransfer(src)) { + if (isExcludedTransfer(src, dst)) { return; } @@ -126,7 +126,7 @@ public void onEnd(InetSocketAddress src, InetSocketAddress dst, MoverInfoMessage return; } - if (isLocalTransfer(src)) { + if (isExcludedTransfer(src, dst)) { return; } @@ -286,9 +286,12 @@ private OptionalInt getExperimentId(ProtocolInfo protocolInfo, Subject subject) : OptionalInt.empty(); } - private boolean isLocalTransfer(InetSocketAddress dst) { - InetAddress addr = dst.getAddress(); - return localSubnet.test(addr); + private boolean isExcludedTransfer(InetSocketAddress src, InetSocketAddress dst) { + InetAddress srcAddress = src.getAddress(); + InetAddress dstAddress = dst.getAddress(); + return srcAddress != null && dstAddress != null + && localSubnet.test(srcAddress) + && localSubnet.test(dstAddress); } private int getActivity(ProtocolInfo protocolInfo) { diff --git a/modules/dcache/src/test/java/org/dcache/pool/movers/TransferLifeCycleTest.java b/modules/dcache/src/test/java/org/dcache/pool/movers/TransferLifeCycleTest.java index d030331d007..af1b35adcae 100644 --- a/modules/dcache/src/test/java/org/dcache/pool/movers/TransferLifeCycleTest.java +++ b/modules/dcache/src/test/java/org/dcache/pool/movers/TransferLifeCycleTest.java @@ -5,6 +5,11 @@ import static org.junit.Assert.assertTrue; import diskCacheV111.vehicles.ProtocolInfo; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketTimeoutException; import java.lang.reflect.Method; import java.util.OptionalInt; import javax.security.auth.Subject; @@ -53,11 +58,52 @@ public void shouldMapSlashPrefixedFqanToVoName() throws Exception { assertEquals(2, experimentId.getAsInt()); } + @Test + public void shouldSuppressMarkerWhenBothEndpointsAreExcluded() throws Exception { + assertFalse(sendsStartMarker("10.10.10.10", "10.20.20.20", "10.0.0.0/8")); + } + + @Test + public void shouldNotSuppressMarkerWhenOnlySourceIsExcluded() throws Exception { + assertTrue(sendsStartMarker("10.10.10.10", "203.0.113.20", "10.0.0.0/8")); + } + + @Test + public void shouldNotSuppressMarkerWhenOnlyDestinationIsExcluded() throws Exception { + assertTrue(sendsStartMarker("203.0.113.20", "10.20.20.20", "10.0.0.0/8")); + } + private OptionalInt resolveExperimentId(String transferTag, Subject subject) throws Exception { return (OptionalInt) getExperimentId.invoke(transferLifeCycle, new TestProtocolInfo("xrootd", transferTag), subject); } + private boolean sendsStartMarker(String srcIp, String dstIp, String excludes) throws Exception { + try (DatagramSocket socket = new DatagramSocket(0, InetAddress.getByName("127.0.0.1"))) { + socket.setSoTimeout(700); + + TransferLifeCycle lifecycle = new TransferLifeCycle(); + lifecycle.setEnabled(true); + lifecycle.setVoMapping("atlas:2"); + lifecycle.setExcludes(new String[]{excludes}); + lifecycle.setFireflyDestination("127.0.0.1:" + socket.getLocalPort()); + + lifecycle.onStart( + new InetSocketAddress(srcIp, 40000), + new InetSocketAddress(dstIp, 20066), + new TestProtocolInfo("xrootd", "129"), + new Subject()); + + var packet = new DatagramPacket(new byte[4096], 4096); + try { + socket.receive(packet); + return true; + } catch (SocketTimeoutException ignored) { + return false; + } + } + } + private static class TestProtocolInfo implements ProtocolInfo { private static final long serialVersionUID = 1L;