Restrict COPY FROM using local files to superuser

Fixing a security issue where any user could read/import content
of any file on the host system, the CrateDB process user has read
access to.

(cherry picked from commit 4e857d6)

docs/appendices/release-notes/5.5.4.rst

@@ -44,6 +44,14 @@ Version 5.5.4 - Unreleased
 See the :ref:`version_5.5.0` release notes for a full list of changes in the
 5.5 series.
 
+Security Fixes
+==============
+
+- Fixed a security issue where any CrateDB user could read/import the content of
+  any file on the host system, the CrateDB process user has read access to, by
+  using the ``COPY FROM`` command with a file URI. This access is now restricted
+  to the ``crate`` superuser only.
+
 Fixes
 =====
 

docs/sql/statements/copy-from.rst

@@ -207,6 +207,8 @@ For example:
 
 The files must be accessible on at least one node and the system user running
 the ``crate`` process must have read access to every file specified.
+Additionally, only the ``crate`` superuser is allowed to use the ``file://``
+scheme.
 
 By default, every node will attempt to import every file. If the file is
 accessible on multiple nodes, you can set the `shared`_ option to true in order

plugins/cr8-copy-s3/src/test/java/io/crate/copy/s3/S3FileReadingCollectorTest.java

@@ -106,7 +106,7 @@ public void testCollectWithOneSocketTimeout() throws Throwable {
     private FileReadingIterator createBatchIterator(S3ObjectInputStream inputStream, String ... fileUris) {
         String compression = null;
         return new FileReadingIterator(
-            Arrays.asList(fileUris),
+            Arrays.stream(fileUris).map(FileReadingIterator::toURI).toList(),
             compression,
             Map.of(
                 S3FileInputFactory.NAME,

server/src/main/java/io/crate/exceptions/UnauthorizedException.java

@@ -21,10 +21,18 @@
 
 package io.crate.exceptions;
 
-public class UnauthorizedException extends RuntimeException implements UnscopedException {
+import java.io.IOException;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.common.io.stream.StreamInput;
+
+public class UnauthorizedException extends ElasticsearchException implements UnscopedException {
 
     public UnauthorizedException(String message) {
         super(message);
     }
 
+    public UnauthorizedException(StreamInput in) throws IOException {
+        super(in);
+    }
 }

server/src/main/java/io/crate/execution/engine/collect/files/FileReadingIterator.java

@@ -175,7 +175,7 @@ public boolean equals(Object obj) {
         }
     }
 
-    public FileReadingIterator(Collection<String> fileUris,
+    public FileReadingIterator(Collection<URI> fileUris,
                                String compression,
                                Map<String, FileInputFactory> fileInputFactories,
                                Boolean shared,
@@ -398,8 +398,7 @@ public static URI toURI(String fileUri) {
     }
 
     @Nullable
-    private FileInput toFileInput(String fileUri, Settings withClauseOptions) {
-        URI uri = toURI(fileUri);
+    private FileInput toFileInput(URI uri, Settings withClauseOptions) {
         FileInputFactory fileInputFactory = fileInputFactories.get(uri.getScheme());
         if (fileInputFactory != null) {
             try {

server/src/main/java/io/crate/execution/engine/collect/sources/FileCollectSource.java

@@ -21,6 +21,9 @@
 
 package io.crate.execution.engine.collect.sources;
 
+import static java.util.Objects.requireNonNull;
+
+import java.net.URI;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -40,6 +43,7 @@
 import io.crate.data.BatchIterator;
 import io.crate.data.Row;
 import io.crate.data.SkippingBatchIterator;
+import io.crate.exceptions.UnauthorizedException;
 import io.crate.execution.dsl.phases.CollectPhase;
 import io.crate.execution.dsl.phases.FileUriCollectPhase;
 import io.crate.execution.engine.collect.CollectTask;
@@ -55,6 +59,7 @@
 import io.crate.planner.operators.SubQueryResults;
 import io.crate.types.ArrayType;
 import io.crate.types.DataTypes;
+import io.crate.user.UserLookup;
 
 @Singleton
 public class FileCollectSource implements CollectSource {
@@ -64,17 +69,20 @@ public class FileCollectSource implements CollectSource {
     private final InputFactory inputFactory;
     private final NodeContext nodeCtx;
     private final ThreadPool threadPool;
+    private final UserLookup userLookup;
 
     @Inject
     public FileCollectSource(NodeContext nodeCtx,
                              ClusterService clusterService,
                              Map<String, FileInputFactory> fileInputFactoryMap,
-                             ThreadPool threadPool) {
+                             ThreadPool threadPool,
+                             UserLookup userLookup) {
         this.fileInputFactoryMap = fileInputFactoryMap;
         this.nodeCtx = nodeCtx;
         this.inputFactory = new InputFactory(nodeCtx);
         this.clusterService = clusterService;
         this.threadPool = threadPool;
+        this.userLookup = userLookup;
     }
 
     @Override
@@ -87,7 +95,16 @@ public CompletableFuture<BatchIterator<Row>> getIterator(TransactionContext txnC
             inputFactory.ctxForRefs(txnCtx, FileLineReferenceResolver::getImplementation);
         ctx.add(collectPhase.toCollect());
 
-        List<String> fileUris = targetUriToStringList(txnCtx, nodeCtx, fileUriCollectPhase.targetUri());
+        var user = requireNonNull(userLookup.findUser(txnCtx.sessionSettings().userName()), "User who invoked a statement must exist");
+        List<URI> fileUris = targetUriToStringList(txnCtx, nodeCtx, fileUriCollectPhase.targetUri()).stream()
+            .map(s -> {
+                var uri = FileReadingIterator.toURI(s);
+                if (uri.getScheme().equals("file") && user.isSuperUser() == false) {
+                    throw new UnauthorizedException("Only a superuser can read from the local file system");
+                }
+                return uri;
+            })
+            .toList();
         FileReadingIterator fileReadingIterator = new FileReadingIterator(
             fileUris,
             fileUriCollectPhase.compression(),

server/src/main/java/org/elasticsearch/ElasticsearchException.java

@@ -968,7 +968,12 @@ private enum ElasticsearchExceptionHandle {
             org.elasticsearch.cluster.coordination.NodeHealthCheckFailureException.class,
             org.elasticsearch.cluster.coordination.NodeHealthCheckFailureException::new,
             175,
-            Version.V_5_2_0);
+            Version.V_5_2_0),
+        UNAUTHORIZED_EXCEPTION(
+            io.crate.exceptions.UnauthorizedException.class,
+            io.crate.exceptions.UnauthorizedException::new,
+            177,
+            Version.V_5_5_4);
 
         final Class<? extends ElasticsearchException> exceptionClass;
         final CheckedFunction<StreamInput, ? extends ElasticsearchException, IOException> constructor;

server/src/test/java/io/crate/execution/engine/collect/MapSideDataCollectOperationTest.java

@@ -56,6 +56,7 @@
 import io.crate.metadata.CoordinatorTxnCtx;
 import io.crate.test.integration.CrateDummyClusterServiceUnitTest;
 import io.crate.types.DataTypes;
+import io.crate.user.User;
 
 
 public class MapSideDataCollectOperationTest extends CrateDummyClusterServiceUnitTest {
@@ -69,7 +70,8 @@ public void testFileUriCollect() throws Exception {
             createNodeContext(),
             clusterService,
             Collections.emptyMap(),
-            THREAD_POOL
+            THREAD_POOL,
+            () -> List.of(User.CRATE_USER)
             );
 
         File tmpFile = temporaryFolder.newFile("fileUriCollectOperation.json");

server/src/test/java/io/crate/execution/engine/collect/files/FileReadingCollectorTest.java

@@ -200,7 +200,7 @@ private static FileReadingIterator it(String ... fileUris) {
 
     private static FileReadingIterator it(Collection<String> fileUris, String compression) {
         return new FileReadingIterator(
-            fileUris,
+            fileUris.stream().map(FileReadingIterator::toURI).toList(),
             compression,
             Map.of(LocalFsFileInputFactory.NAME, new LocalFsFileInputFactory()),
             false,

server/src/test/java/io/crate/execution/engine/collect/files/FileReadingIteratorTest.java

@@ -34,6 +34,7 @@
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.net.SocketTimeoutException;
+import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -43,6 +44,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
+import java.util.stream.Stream;
 
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
@@ -87,7 +89,9 @@ public void test_iterator_closes_current_reader_on_io_error() throws Exception {
         Path tempFile2 = createTempFile("tempfile2", ".csv");
         List<String> lines2 = List.of("name,id,age", "Trillian,5,33");
         Files.write(tempFile2, lines2);
-        List<String> fileUris = List.of(tempFile1.toUri().toString(), tempFile2.toUri().toString());
+        List<URI> fileUris = Stream.of(tempFile1.toUri().toString(), tempFile2.toUri().toString())
+            .map(FileReadingIterator::toURI)
+            .toList();
 
         Supplier<BatchIterator<LineCursor>> batchIteratorSupplier =
             () -> new FileReadingIterator(
@@ -139,7 +143,8 @@ public void test_consecutive_retries_will_not_result_in_duplicate_reads() throws
         Path tempFile = createTempFile("tempfile1", ".csv");
         List<String> lines = List.of("id", "1", "2", "3", "4", "5");
         Files.write(tempFile, lines);
-        List<String> fileUris = List.of(tempFile.toUri().toString());
+        List<URI> fileUris = Stream.of(tempFile.toUri().toString())
+            .map(FileReadingIterator::toURI).toList();
 
         Supplier<BatchIterator<LineCursor>> batchIteratorSupplier =
             () -> new FileReadingIterator(
@@ -213,7 +218,8 @@ public void test_retry_from_one_uri_does_not_affect_reading_next_uri() throws Ex
         Files.write(tempFile, List.of("1", "2", "3"));
         Path tempFile2 = createTempFile("tempfile2", ".csv");
         Files.write(tempFile2, List.of("4", "5", "6"));
-        List<String> fileUris = List.of(tempFile.toUri().toString(), tempFile2.toUri().toString());
+        List<URI> fileUris = Stream.of(tempFile.toUri().toString(), tempFile2.toUri().toString())
+            .map(FileReadingIterator::toURI).toList();
 
         var fi = new FileReadingIterator(
             fileUris,

server/src/test/java/io/crate/execution/engine/collect/sources/FileCollectSourceTest.java

@@ -92,7 +92,8 @@ public void test_file_collect_source_returns_iterator_that_can_skip_lines() thro
             new NodeContext(new Functions(Map.of()), userLookup),
             clusterService,
             Map.of(),
-            THREAD_POOL
+            THREAD_POOL,
+            () -> List.of(User.CRATE_USER)
         );
 
         CompletableFuture<BatchIterator<Row>> iterator = fileCollectSource.getIterator(

server/src/test/java/io/crate/integrationtests/CopyIntegrationTest.java

@@ -26,6 +26,7 @@
 import static io.crate.testing.Asserts.assertThat;
 import static io.crate.testing.TestingHelpers.printedTable;
 import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.io.File;
 import java.io.FileOutputStream;
@@ -53,10 +54,14 @@
 
 import com.carrotsearch.randomizedtesting.LifecycleScope;
 
+import io.crate.action.sql.Sessions;
+import io.crate.exceptions.UnauthorizedException;
 import io.crate.testing.Asserts;
 import io.crate.testing.SQLResponse;
 import io.crate.testing.UseJdbc;
 import io.crate.testing.UseNewCluster;
+import io.crate.testing.UseRandomizedSchema;
+import io.crate.user.UserLookup;
 
 @IntegTestCase.ClusterScope(numDataNodes = 2)
 public class CopyIntegrationTest extends SQLHttpIntegrationTest {
@@ -1196,4 +1201,22 @@ primary key (id)
             "2| 31123| apple safari| 23073"
         );
     }
+
+    @UseRandomizedSchema(random = false)
+    @Test
+    public void test_copy_from_local_file_is_only_allowed_for_superusers() {
+        execute("CREATE TABLE quotes (id INT PRIMARY KEY, " +
+            "quote STRING INDEX USING FULLTEXT) WITH (number_of_replicas = 0)");
+        execute("CREATE USER test_user");
+        execute("GRANT ALL TO test_user");
+
+        var roles = cluster().getInstance(UserLookup.class);
+        var user = roles.findUser("test_user");
+        Sessions sqlOperations = cluster().getInstance(Sessions.class);
+        try (var session = sqlOperations.newSession(null, user)) {
+            assertThatThrownBy(() -> execute("COPY quotes FROM ?", new Object[]{copyFilePath + "test_copy_from.json"}, session))
+                .isExactlyInstanceOf(UnauthorizedException.class)
+                .hasMessage("Only a superuser can read from the local file system");
+        }
+    }
 }