fix: forward tower auth for dataset http reads

Signed-off-by: Edmund Miller <edmund.miller@seqera.io>

Author: edmundmiller

plugins/nf-tower/src/main/io/seqera/tower/plugin/dataset/DatasetFileSystemProvider.java

@@ -19,6 +19,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
+import java.net.URLConnection;
 import java.nio.channels.SeekableByteChannel;
 import java.nio.file.AccessMode;
 import java.nio.file.CopyOption;
@@ -36,6 +37,10 @@
 import java.nio.file.spi.FileSystemProvider;
 import java.util.Map;
 import java.util.Set;
+import java.util.regex.Pattern;
+
+import nextflow.file.http.XAuthProvider;
+import nextflow.file.http.XAuthRegistry;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -55,6 +60,7 @@
 public class DatasetFileSystemProvider extends FileSystemProvider {
 
     private static final Logger log = LoggerFactory.getLogger(DatasetFileSystemProvider.class);
+    private static final Pattern PLATFORM_DATASET_PATH = Pattern.compile(".*?/workspaces/[^/]+/datasets/[^/]+/v/[^/]+/n/.+");
 
     private volatile DatasetFileSystem fileSystem;
 
@@ -99,13 +105,13 @@ public Path getPath(URI uri) {
     public InputStream newInputStream(Path path, OpenOption... options) throws IOException {
         final Path resolved = resolvedPath(path);
         log.debug("newInputStream for dataset path {} -> {}", path, resolved);
-        return resolved.getFileSystem().provider().newInputStream(resolved, options);
+        return withPlatformDatasetAuth(resolved, () -> resolved.getFileSystem().provider().newInputStream(resolved, options));
     }
 
     @Override
     public SeekableByteChannel newByteChannel(Path path, Set<? extends OpenOption> options, FileAttribute<?>... attrs) throws IOException {
         final Path resolved = resolvedPath(path);
-        return resolved.getFileSystem().provider().newByteChannel(resolved, options, attrs);
+        return withPlatformDatasetAuth(resolved, () -> resolved.getFileSystem().provider().newByteChannel(resolved, options, attrs));
     }
 
     @Override
@@ -117,20 +123,20 @@ public DirectoryStream<Path> newDirectoryStream(Path dir, DirectoryStream.Filter
     @SuppressWarnings("unchecked")
     public <A extends BasicFileAttributes> A readAttributes(Path path, Class<A> type, LinkOption... options) throws IOException {
         final Path resolved = resolvedPath(path);
-        return resolved.getFileSystem().provider().readAttributes(resolved, type, options);
+        return withPlatformDatasetAuth(resolved, () -> resolved.getFileSystem().provider().readAttributes(resolved, type, options));
     }
 
     @Override
     public Map<String, Object> readAttributes(Path path, String attributes, LinkOption... options) throws IOException {
         final Path resolved = resolvedPath(path);
-        return resolved.getFileSystem().provider().readAttributes(resolved, attributes, options);
+        return withPlatformDatasetAuth(resolved, () -> resolved.getFileSystem().provider().readAttributes(resolved, attributes, options));
     }
 
     @Override
     public <V extends FileAttributeView> V getFileAttributeView(Path path, Class<V> type, LinkOption... options) {
         try {
             final Path resolved = resolvedPath(path);
-            return resolved.getFileSystem().provider().getFileAttributeView(resolved, type, options);
+            return withPlatformDatasetAuth(resolved, () -> resolved.getFileSystem().provider().getFileAttributeView(resolved, type, options));
         }
         catch (IOException e) {
             throw new RuntimeException("Failed to resolve dataset path: " + path, e);
@@ -145,7 +151,10 @@ public void checkAccess(Path path, AccessMode... modes) throws IOException {
             }
         }
         final Path resolved = resolvedPath(path);
-        resolved.getFileSystem().provider().checkAccess(resolved, modes);
+        withPlatformDatasetAuth(resolved, () -> {
+            resolved.getFileSystem().provider().checkAccess(resolved, modes);
+            return null;
+        });
     }
 
     // -- write operations: read-only in phase 1 --
@@ -206,6 +215,86 @@ private DatasetFileSystem getOrCreateFileSystem() {
         return fileSystem;
     }
 
+    private <T> T withPlatformDatasetAuth(Path resolved, IoCallable<T> action) throws IOException {
+        final XAuthProvider authProvider = authProviderFor(resolved);
+        if (authProvider == null) {
+            return action.call();
+        }
+
+        final XAuthRegistry registry = XAuthRegistry.getInstance();
+        registry.register(authProvider);
+        try {
+            return action.call();
+        }
+        finally {
+            registry.unregister(authProvider);
+        }
+    }
+
+    private XAuthProvider authProviderFor(Path resolved) {
+        final URI targetUri = resolved.toUri();
+        if (targetUri == null || !isHttpScheme(targetUri.getScheme())) {
+            return null;
+        }
+        if (!isPlatformDatasetDownloadPath(targetUri.getPath())) {
+            return null;
+        }
+
+        final String endpoint = DatasetResolver.towerEndpoint();
+        final String accessToken = DatasetResolver.towerAccessToken();
+        if (isBlank(endpoint) || isBlank(accessToken)) {
+            return null;
+        }
+
+        final URI endpointUri = URI.create(endpoint);
+        if (!isHttpScheme(endpointUri.getScheme()) || !isSameOrigin(endpointUri, targetUri)) {
+            return null;
+        }
+
+        return new DatasetBearerAuthProvider(endpointUri, accessToken);
+    }
+
+    private boolean isHttpScheme(String scheme) {
+        return "http".equalsIgnoreCase(scheme) || "https".equalsIgnoreCase(scheme);
+    }
+
+    private boolean isPlatformDatasetDownloadPath(String path) {
+        return path != null && PLATFORM_DATASET_PATH.matcher(path).matches();
+    }
+
+    private static boolean isSameOrigin(URI left, URI right) {
+        if (left.getScheme() == null || right.getScheme() == null) {
+            return false;
+        }
+        if (!left.getScheme().equalsIgnoreCase(right.getScheme())) {
+            return false;
+        }
+        if (left.getHost() == null || right.getHost() == null) {
+            return false;
+        }
+        if (!left.getHost().equalsIgnoreCase(right.getHost())) {
+            return false;
+        }
+        return defaultPort(left) == defaultPort(right);
+    }
+
+    private static int defaultPort(URI uri) {
+        if (uri.getPort() != -1) {
+            return uri.getPort();
+        }
+        if ("https".equalsIgnoreCase(uri.getScheme())) {
+            return 443;
+        }
+        if ("http".equalsIgnoreCase(uri.getScheme())) {
+            return 80;
+        }
+        return -1;
+    }
+
+    private boolean isBlank(String value) {
+        return value == null || value.trim().isEmpty();
+    }
+
     /**
      * Resolve a dataset path to its backing cloud storage path.
      * Caches the resolved path on the DatasetPath instance.
@@ -216,4 +305,41 @@ private Path resolvedPath(Path path) throws IOException {
         }
         return ((DatasetPath) path).getResolvedPath();
     }
+
+    @FunctionalInterface
+    private interface IoCallable<T> {
+        T call() throws IOException;
+    }
+
+    private static final class DatasetBearerAuthProvider implements XAuthProvider {
+        private final URI endpoint;
+        private final String accessToken;
+
+        private DatasetBearerAuthProvider(URI endpoint, String accessToken) {
+            this.endpoint = endpoint;
+            this.accessToken = accessToken;
+        }
+
+        @Override
+        public boolean authorize(URLConnection connection) {
+            final URI target = URI.create(connection.getURL().toString());
+            if (!isSameOrigin(endpoint, target)) {
+                return false;
+            }
+            if (!PLATFORM_DATASET_PATH.matcher(target.getPath()).matches()) {
+                return false;
+            }
+            if (connection.getRequestProperty("Authorization") != null) {
+                return false;
+            }
+
+            connection.setRequestProperty("Authorization", "Bearer " + accessToken);
+            return true;
+        }
+
+        @Override
+        public boolean refreshToken(URLConnection connection) {
+            return false;
+        }
+    }
 }

plugins/nf-tower/src/main/io/seqera/tower/plugin/dataset/DatasetResolver.groovy

@@ -44,6 +44,14 @@ import nextflow.platform.PlatformHelper
 @CompileStatic
 class DatasetResolver {
 
+    static String towerEndpoint() {
+        return getEndpoint()
+    }
+
+    static String towerAccessToken() {
+        return getAccessToken()
+    }
+
     /**
      * Resolve a dataset name (and optional version) to the backing cloud storage Path.
      *

plugins/nf-tower/src/test/io/seqera/tower/plugin/dataset/DatasetIntegrationTest.groovy

@@ -20,6 +20,7 @@ import java.nio.file.Files
 import java.nio.file.LinkOption
 import java.nio.file.attribute.BasicFileAttributes
 
+import com.github.tomakehurst.wiremock.WireMockServer
 import spock.lang.TempDir
 import spock.lang.Unroll
 
@@ -114,6 +115,69 @@ class DatasetIntegrationTest extends DatasetWireMockSpec {
         'readAttributes'  | 'hello'             | { DatasetFileSystemProvider providerRef, dsPath -> providerRef.readAttributes(dsPath, BasicFileAttributes, new LinkOption[0]) }      | { attrs -> attrs.size() == 5 && !attrs.isDirectory() && attrs.isRegularFile() }
     }
 
+    def 'should forward bearer auth when reading platform dataset download URLs'() {
+        given:
+        mockSession(workspaceId: '100')
+        stubDatasets([[id: '7', name: 'secure-ds']], '100')
+
+        and:
+        def downloadPath = '/workspaces/100/datasets/7/v/1/n/data.csv'
+        def downloadUrl = "http://localhost:${wireMock.port()}${downloadPath}"
+        stubDatasetVersions('7', [[version: 1, url: downloadUrl, fileName: 'data.csv']], '100')
+        wireMock.stubFor(get(urlPathEqualTo(downloadPath))
+            .withHeader('Authorization', equalTo('Bearer test-token'))
+            .willReturn(ok('secure,data\na,b\n')))
+
+        and:
+        def provider = new DatasetFileSystemProvider()
+        def path = provider.getPath(new URI('dataset://secure-ds'))
+
+        when:
+        def content = provider.newInputStream(path).text
+
+        then:
+        content == 'secure,data\na,b\n'
+
+        and:
+        wireMock.verify(1, getRequestedFor(urlPathEqualTo(downloadPath))
+            .withHeader('Authorization', equalTo('Bearer test-token')))
+    }
+
+    def 'should not forward bearer auth to non-platform hosts'() {
+        given:
+        def externalHost = new WireMockServer(0)
+        externalHost.start()
+
+        and:
+        mockSession(workspaceId: '100')
+        stubDatasets([[id: '7', name: 'external-ds']], '100')
+
+        and:
+        def downloadPath = '/workspaces/100/datasets/7/v/1/n/data.csv'
+        def downloadUrl = "http://localhost:${externalHost.port()}${downloadPath}"
+        stubDatasetVersions('7', [[version: 1, url: downloadUrl, fileName: 'data.csv']], '100')
+        externalHost.stubFor(get(urlPathEqualTo(downloadPath))
+            .withHeader('Authorization', matching('.+'))
+            .atPriority(1)
+            .willReturn(unauthorized()))
+        externalHost.stubFor(get(urlPathEqualTo(downloadPath))
+            .atPriority(10)
+            .willReturn(ok('public,data\nx,y\n')))
+
+        and:
+        def provider = new DatasetFileSystemProvider()
+        def path = provider.getPath(new URI('dataset://external-ds'))
+
+        when:
+        def content = provider.newInputStream(path).text
+
+        then:
+        content == 'public,data\nx,y\n'
+
+        cleanup:
+        externalHost.stop()
+    }
+
     def 'should cache resolved path across multiple reads'() {
         given:
         def dataFile = makeFile('data.csv', 'cached')

plugins/nf-tower/src/test/io/seqera/tower/plugin/dataset/DatasetLiveAuthRegressionTest.groovy

@@ -18,7 +18,6 @@ package io.seqera.tower.plugin.dataset
 
 import nextflow.Global
 import nextflow.Session
-import spock.lang.PendingFeature
 import spock.lang.Requires
 import spock.lang.Specification
 import spock.lang.Unroll
@@ -38,7 +37,6 @@ class DatasetLiveAuthRegressionTest extends Specification {
         Global.session = null
     }
 
-    @PendingFeature(reason = 'Dataset provider does not yet forward Tower bearer auth to resolved HTTP dataset URLs')
     @Unroll
     def 'should read live dataset via provider using bearer auth - #datasetName'() {
         given: