Add option to configure local infile path

mirromutth · mirromutth · commit 19d33f5d0f53 · 2024-01-29T14:10:34.000+09:00
diff --git a/README.md b/README.md
@@ -141,6 +141,7 @@ ConnectionFactoryOptions options = ConnectionFactoryOptions.builder()
     .option(Option.valueOf("sslContextBuilderCustomizer"), "com.example.demo.MyCustomizer") // optional, default is no-op customizer
     .option(Option.valueOf("zeroDate"), "use_null") // optional, default "use_null"
     .option(Option.valueOf("useServerPrepareStatement"), true) // optional, default false
+    .option(Option.valueOf("allowLoadLocalInfileInPath"), "/opt") // optional, default null, null means LOCAL INFILE not be allowed
     .option(Option.valueOf("tcpKeepAlive"), true) // optional, default false
     .option(Option.valueOf("tcpNoDelay"), true) // optional, default false
     .option(Option.valueOf("autodetectExtensions"), false) // optional, default false
@@ -189,6 +190,7 @@ MySqlConnectionConfiguration configuration = MySqlConnectionConfiguration.builde
     .sslContextBuilderCustomizer(MyCustomizer.INSTANCE) // optional, default is no-op customizer
     .zeroDateOption(ZeroDateOption.USE_NULL) // optional, default ZeroDateOption.USE_NULL
     .useServerPrepareStatement() // Use server-preparing statements, default use client-preparing statements
+    .allowLoadLocalInfileInPath("/opt") // optional, default null, null means LOCAL INFILE not be allowed
     .tcpKeepAlive(true) // optional, controls TCP Keep Alive, default is false
     .tcpNoDelay(true) // optional, controls TCP No Delay, default is false
     .autodetectExtensions(false) // optional, controls extension auto-detect, default is true
@@ -242,6 +244,7 @@ Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
 | zeroDateOption | Any value of `ZeroDateOption` | Optional, default `USE_NULL` | The option indicates "zero date" handling, see following notice |
 | autodetectExtensions | `true` or `false` | Optional, default is `true` | Controls auto-detect `Extension`s |
 | useServerPrepareStatement | `true`, `false` or `Predicate<String>` | Optional, default is `false` | See following notice |
+| allowLoadLocalInfileInPath | A path | Optional, default is `null` | The path that allows `LOAD DATA LOCAL INFILE` to load file data |
 | passwordPublisher | A `Publisher<String>` | Optional, default is `null` | The password publisher, see following notice |
 
 - `SslMode` Considers security level and verification for SSL, make sure the database server supports SSL before you want change SSL mode to `REQUIRED` or higher. **The Unix Domain Socket only offers "DISABLED" available**
@@ -584,6 +587,7 @@ If you want to raise an issue, please follow the recommendations below:
 - The MySQL may be not support well for searching rows by a binary field, like `BIT` and `JSON`
   - `BIT`: cannot select 'BIT(64)' with value greater than 'Long.MAX_VALUE' (or equivalent in binary)
   - `JSON`: different MySQL may have different serialization formats, e.g. MariaDB and MySQL
+- MySQL 8.0+ disables `@@global.local_infile` by default, make sure `@@local_infile` is `ON` before enable `allowLoadLocalInfileInPath` of the driver. e.g. run `SET GLOBAL local_infile=ON`, or set it in `mysql.cnf`.
 
 ## License
 
diff --git a/src/main/java/io/asyncer/r2dbc/mysql/Capability.java b/src/main/java/io/asyncer/r2dbc/mysql/Capability.java
@@ -361,7 +361,7 @@ void disableCompression() {
             this.bitmap &= ~COMPRESS;
         }
 
-        void disableLoadDataInfile() {
+        void disableLoadDataLocalInfile() {
             this.bitmap &= ~LOCAL_FILES;
         }
 
diff --git a/src/main/java/io/asyncer/r2dbc/mysql/ConnectionContext.java b/src/main/java/io/asyncer/r2dbc/mysql/ConnectionContext.java
@@ -22,6 +22,7 @@
 import io.asyncer.r2dbc.mysql.constant.ZeroDateOption;
 import org.jetbrains.annotations.Nullable;
 
+import java.nio.file.Path;
 import java.time.ZoneId;
 
 import static io.asyncer.r2dbc.mysql.internal.util.AssertUtils.requireNonNull;
@@ -42,6 +43,11 @@ public final class ConnectionContext implements CodecContext {
 
     private final ZeroDateOption zeroDateOption;
 
+    @Nullable
+    private final Path localInfilePath;
+
+    private final int localInfileBufferSize;
+
     @Nullable
     private ZoneId serverZoneId;
 
@@ -53,8 +59,11 @@ public final class ConnectionContext implements CodecContext {
 
     private volatile Capability capability = null;
 
-    ConnectionContext(ZeroDateOption zeroDateOption, @Nullable ZoneId serverZoneId) {
+    ConnectionContext(ZeroDateOption zeroDateOption, @Nullable Path localInfilePath,
+        int localInfileBufferSize, @Nullable ZoneId serverZoneId) {
         this.zeroDateOption = requireNonNull(zeroDateOption, "zeroDateOption must not be null");
+        this.localInfilePath = localInfilePath;
+        this.localInfileBufferSize = localInfileBufferSize;
         this.serverZoneId = serverZoneId;
     }
 
@@ -119,8 +128,23 @@ public ZeroDateOption getZeroDateOption() {
         return zeroDateOption;
     }
 
+    /**
+     * Gets the allowed local infile path.
+     *
+     * @return the path.
+     */
+    @Nullable
+    public Path getLocalInfilePath() {
+        return localInfilePath;
+    }
+
+    /**
+     * Gets the local infile buffer size.
+     *
+     * @return the buffer size.
+     */
     public int getLocalInfileBufferSize() {
-        return 64 * 1024;
+        return localInfileBufferSize;
     }
 
     /**
diff --git a/src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionConfiguration.java b/src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionConfiguration.java
@@ -25,6 +25,8 @@
 
 import javax.net.ssl.HostnameVerifier;
 import java.net.Socket;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.time.Duration;
 import java.time.ZoneId;
 import java.util.ArrayList;
@@ -89,6 +91,11 @@ public final class MySqlConnectionConfiguration {
     @Nullable
     private final Predicate<String> preferPrepareStatement;
 
+    @Nullable
+    private final Path loadLocalInfilePath;
+
+    private final int localInfileBufferSize;
+
     private final int queryCacheSize;
 
     private final int prepareCacheSize;
@@ -104,6 +111,7 @@ private MySqlConnectionConfiguration(
         @Nullable Duration socketTimeout, ZeroDateOption zeroDateOption, @Nullable ZoneId serverZoneId,
         String user, @Nullable CharSequence password, @Nullable String database,
         boolean createDatabaseIfNotExist, @Nullable Predicate<String> preferPrepareStatement,
+        @Nullable Path loadLocalInfilePath, int localInfileBufferSize,
         int queryCacheSize, int prepareCacheSize, Extensions extensions,
         @Nullable Publisher<String> passwordPublisher
     ) {
@@ -122,6 +130,8 @@ private MySqlConnectionConfiguration(
         this.database = database == null || database.isEmpty() ? "" : database;
         this.createDatabaseIfNotExist = createDatabaseIfNotExist;
         this.preferPrepareStatement = preferPrepareStatement;
+        this.loadLocalInfilePath = loadLocalInfilePath;
+        this.localInfileBufferSize = localInfileBufferSize;
         this.queryCacheSize = queryCacheSize;
         this.prepareCacheSize = prepareCacheSize;
         this.extensions = extensions;
@@ -207,6 +217,15 @@ Predicate<String> getPreferPrepareStatement() {
         return preferPrepareStatement;
     }
 
+    @Nullable
+    Path getLoadLocalInfilePath() {
+        return loadLocalInfilePath;
+    }
+
+    int getLocalInfileBufferSize() {
+        return localInfileBufferSize;
+    }
+
     int getQueryCacheSize() {
         return queryCacheSize;
     }
@@ -248,6 +267,8 @@ public boolean equals(Object o) {
             database.equals(that.database) &&
             createDatabaseIfNotExist == that.createDatabaseIfNotExist &&
             Objects.equals(preferPrepareStatement, that.preferPrepareStatement) &&
+            Objects.equals(loadLocalInfilePath, that.loadLocalInfilePath) &&
+            localInfileBufferSize == that.localInfileBufferSize &&
             queryCacheSize == that.queryCacheSize &&
             prepareCacheSize == that.prepareCacheSize &&
             extensions.equals(that.extensions) &&
@@ -258,7 +279,8 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Objects.hash(isHost, domain, port, ssl, tcpKeepAlive, tcpNoDelay, connectTimeout,
             socketTimeout, serverZoneId, zeroDateOption, user, password, database, createDatabaseIfNotExist,
-            preferPrepareStatement, queryCacheSize, prepareCacheSize, extensions, passwordPublisher);
+            preferPrepareStatement, loadLocalInfilePath, localInfileBufferSize, queryCacheSize,
+            prepareCacheSize, extensions, passwordPublisher);
     }
 
     @Override
@@ -269,16 +291,21 @@ public String toString() {
                 connectTimeout + ", socketTimeout=" + socketTimeout + ", serverZoneId=" + serverZoneId +
                 ", zeroDateOption=" + zeroDateOption + ", user='" + user + "', password=" + password +
                 ", database='" + database + "', createDatabaseIfNotExist=" + createDatabaseIfNotExist +
-                ", preferPrepareStatement=" + preferPrepareStatement + ", queryCacheSize=" + queryCacheSize +
-                ", prepareCacheSize=" + prepareCacheSize + ", extensions=" + extensions +
-                ", passwordPublisher=" + passwordPublisher + '}';
+                ", preferPrepareStatement=" + preferPrepareStatement +
+                ", loadLocalInfilePath=" + loadLocalInfilePath +
+                ", localInfileBufferSize=" + localInfileBufferSize +
+                ", queryCacheSize=" + queryCacheSize + ", prepareCacheSize=" + prepareCacheSize +
+                ", extensions=" + extensions + ", passwordPublisher=" + passwordPublisher + '}';
         }
 
         return "MySqlConnectionConfiguration{unixSocket='" + domain + "', connectTimeout=" +
             connectTimeout + ", socketTimeout=" + socketTimeout + ", serverZoneId=" + serverZoneId +
             ", zeroDateOption=" + zeroDateOption + ", user='" + user + "', password=" + password +
             ", database='" + database + "', createDatabaseIfNotExist=" + createDatabaseIfNotExist +
-            ", preferPrepareStatement=" + preferPrepareStatement + ", queryCacheSize=" + queryCacheSize +
+            ", preferPrepareStatement=" + preferPrepareStatement +
+            ", loadLocalInfilePath=" + loadLocalInfilePath +
+            ", localInfileBufferSize=" + localInfileBufferSize +
+            ", queryCacheSize=" + queryCacheSize +
             ", prepareCacheSize=" + prepareCacheSize + ", extensions=" + extensions +
             ", passwordPublisher=" + passwordPublisher + '}';
     }
@@ -345,6 +372,11 @@ public static final class Builder {
         @Nullable
         private Predicate<String> preferPrepareStatement;
 
+        @Nullable
+        private Path loadLocalInfilePath;
+
+        private int localInfileBufferSize = 8192;
+
         private int queryCacheSize = 0;
 
         private int prepareCacheSize = 256;
@@ -379,7 +411,8 @@ public MySqlConnectionConfiguration build() {
                 sslCa, sslKey, sslKeyPassword, sslCert, sslContextBuilderCustomizer);
             return new MySqlConnectionConfiguration(isHost, domain, port, ssl, tcpKeepAlive, tcpNoDelay,
                 connectTimeout, socketTimeout, zeroDateOption, serverZoneId, user, password, database,
-                createDatabaseIfNotExist, preferPrepareStatement, queryCacheSize, prepareCacheSize,
+                createDatabaseIfNotExist, preferPrepareStatement, loadLocalInfilePath,
+                localInfileBufferSize, queryCacheSize, prepareCacheSize,
                 Extensions.from(extensions, autodetectExtensions), passwordPublisher);
         }
 
@@ -754,6 +787,38 @@ public Builder useServerPrepareStatement(Predicate<String> preferPrepareStatemen
             return this;
         }
 
+        /**
+         * Configures to allow the {@code LOAD DATA LOCAL INFILE} statement in the given {@code path} or
+         * disallow the statement.  Default to {@code null} which means not allow the statement.
+         *
+         * @param path which parent path are allowed to load file data, {@code null} means not be allowed.
+         * @return {@link Builder this}.
+         * @throws java.nio.file.InvalidPathException if the string cannot be converted to a {@link Path}.
+         * @since 1.1.0
+         */
+        public Builder allowLoadLocalInfileInPath(@Nullable String path) {
+            this.loadLocalInfilePath = path == null ? null : Paths.get(path);
+
+            return this;
+        }
+
+        /**
+         * Configures the buffer size for {@code LOAD DATA LOCAL INFILE} statement.  Default to {@code 8192}.
+         * <p>
+         * It is used only if {@link #allowLoadLocalInfileInPath(String)} is set.
+         *
+         * @param localInfileBufferSize the buffer size.
+         * @return {@link Builder this}.
+         * @throws IllegalArgumentException if {@code localInfileBufferSize} is not positive.
+         * @since 1.1.0
+         */
+        public Builder localInfileBufferSize(int localInfileBufferSize) {
+            require(localInfileBufferSize > 0, "localInfileBufferSize must be positive");
+
+            this.localInfileBufferSize = localInfileBufferSize;
+            return this;
+        }
+
         /**
          * Configures the maximum size of the {@link Query} parsing cache. Usually it should be power of two.
          * Default to {@code 0}. Driver will use unbounded cache if size is less than {@code 0}.
@@ -823,6 +888,7 @@ public Builder extendWith(Extension extension) {
 
         /**
          * Registers a password publisher function.
+         *
          * @param passwordPublisher function to retrieve password before making connection.
          * @return this {@link Builder}.
          */
diff --git a/src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionFactory.java b/src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionFactory.java
@@ -90,8 +90,12 @@ public static MySqlConnectionFactory from(MySqlConnectionConfiguration configura
             String user = configuration.getUser();
             CharSequence password = configuration.getPassword();
             SslMode sslMode = ssl.getSslMode();
-            ConnectionContext context = new ConnectionContext(configuration.getZeroDateOption(),
-                configuration.getServerZoneId());
+            ConnectionContext context = new ConnectionContext(
+                configuration.getZeroDateOption(),
+                configuration.getLoadLocalInfilePath(),
+                configuration.getLocalInfileBufferSize(),
+                configuration.getServerZoneId()
+            );
             Extensions extensions = configuration.getExtensions();
             Predicate<String> prepare = configuration.getPreferPrepareStatement();
             int prepareCacheSize = configuration.getPrepareCacheSize();
diff --git a/src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionFactoryProvider.java b/src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionFactoryProvider.java
@@ -191,6 +191,9 @@ public final class MySqlConnectionFactoryProvider implements ConnectionFactoryPr
     public static final Option<Object> USE_SERVER_PREPARE_STATEMENT =
         Option.valueOf("useServerPrepareStatement");
 
+    public static final Option<String> ALLOW_LOAD_LOCAL_INFILE_IN_PATH =
+        Option.valueOf("allowLoadLocalInfileInPath");
+
     /**
      * Option to set the maximum size of the {@link Query} parsing cache.  Default to {@code 256}.
      *
@@ -266,6 +269,8 @@ static MySqlConnectionConfiguration setup(ConnectionFactoryOptions options) {
             .to(builder::zeroDateOption);
         mapper.optional(USE_SERVER_PREPARE_STATEMENT).prepare(builder::useClientPrepareStatement,
             builder::useServerPrepareStatement, builder::useServerPrepareStatement);
+        mapper.optional(ALLOW_LOAD_LOCAL_INFILE_IN_PATH).asString()
+            .to(builder::allowLoadLocalInfileInPath);
         mapper.optional(QUERY_CACHE_SIZE).asInt()
             .to(builder::queryCacheSize);
         mapper.optional(PREPARE_CACHE_SIZE).asInt()
diff --git a/src/main/java/io/asyncer/r2dbc/mysql/QueryFlow.java b/src/main/java/io/asyncer/r2dbc/mysql/QueryFlow.java
@@ -973,6 +973,10 @@ private Capability clientCapability(Capability serverCapability) {
             builder.disableConnectWithDatabase();
         }
 
+        if (context.getLocalInfilePath() == null) {
+            builder.disableLoadDataLocalInfile();
+        }
+
         if (ATTRIBUTES.isEmpty()) {
             builder.disableConnectAttributes();
         }
diff --git a/src/main/java/io/asyncer/r2dbc/mysql/QueryLogger.java b/src/main/java/io/asyncer/r2dbc/mysql/QueryLogger.java
@@ -16,7 +16,6 @@
 
 package io.asyncer.r2dbc.mysql;
 
-
 import io.asyncer.r2dbc.mysql.internal.util.StringUtils;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
diff --git a/src/main/java/io/asyncer/r2dbc/mysql/message/client/LocalInfileResponse.java b/src/main/java/io/asyncer/r2dbc/mysql/message/client/LocalInfileResponse.java
@@ -20,10 +20,13 @@
 import io.asyncer.r2dbc.mysql.internal.util.NettyBufferUtils;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.r2dbc.spi.R2dbcNonTransientResourceException;
+import io.r2dbc.spi.R2dbcPermissionDeniedException;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import reactor.core.publisher.SynchronousSink;
 
-import java.io.IOException;
+import java.nio.file.InvalidPathException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.concurrent.atomic.AtomicReference;
@@ -52,22 +55,37 @@ public LocalInfileResponse(int envelopeId, String path, SynchronousSink<?> error
     @Override
     public Flux<ByteBuf> encode(ByteBufAllocator allocator, ConnectionContext context) {
         return Flux.defer(() -> {
+            int bufferSize = context.getLocalInfileBufferSize();
             AtomicReference<Throwable> error = new AtomicReference<>();
-            Path path = Paths.get(this.path);
-
-            return NettyBufferUtils.readFile(path, allocator, context.getLocalInfileBufferSize())
-                .onErrorComplete(e -> {
-                    error.set(e);
-                    return true;
-                })
-                .concatWith(Flux.just(allocator.buffer(0, 0)))
-                .doAfterTerminate(() -> {
-                    Throwable e = error.getAndSet(null);
-
-                    if (e != null) {
-                        errorSink.error(e);
+
+            return Mono.<Path>create(sink -> {
+                try {
+                    Path safePath = context.getLocalInfilePath();
+                    Path file = Paths.get(this.path);
+
+                    if (safePath == null || file.startsWith(safePath)) {
+                        sink.success(file);
+                    } else {
+                        String message = String.format("The file '%s' is not under the safe path '%s'",
+                            file, safePath);
+                        sink.error(new R2dbcPermissionDeniedException(message));
                     }
-                });
+                } catch (InvalidPathException e) {
+                    sink.error(new R2dbcNonTransientResourceException("Invalid path: " + this.path, e));
+                } catch (Throwable e) {
+                    sink.error(e);
+                }
+            }).flatMapMany(p -> NettyBufferUtils.readFile(p, allocator, bufferSize)).onErrorComplete(e -> {
+                // Server needs an empty buffer, so emit error to upstream instead of encoding stream.
+                error.set(e);
+                return true;
+            }).concatWith(Flux.just(allocator.buffer(0, 0))).doAfterTerminate(() -> {
+                Throwable e = error.getAndSet(null);
+
+                if (e != null) {
+                    errorSink.error(e);
+                }
+            });
         });
     }
 
diff --git a/src/test/java/io/asyncer/r2dbc/mysql/ConnectionContextTest.java b/src/test/java/io/asyncer/r2dbc/mysql/ConnectionContextTest.java
diff --git a/src/test/java/io/asyncer/r2dbc/mysql/ConnectionIntegrationTest.java b/src/test/java/io/asyncer/r2dbc/mysql/ConnectionIntegrationTest.java
diff --git a/src/test/java/io/asyncer/r2dbc/mysql/IntegrationTestSupport.java b/src/test/java/io/asyncer/r2dbc/mysql/IntegrationTestSupport.java
diff --git a/src/test/java/io/asyncer/r2dbc/mysql/StatementTestSupport.java b/src/test/java/io/asyncer/r2dbc/mysql/StatementTestSupport.java

Original file line number	Diff line number	Diff line change
`@@ -361,7 +361,7 @@ void disableCompression() {`
`361`	`361`	`this.bitmap &= ~COMPRESS;`
`362`	`362`	`}`
`363`	`363`
`364`		`- void disableLoadDataInfile() {`
	`364`	`+ void disableLoadDataLocalInfile() {`
`365`	`365`	`this.bitmap &= ~LOCAL_FILES;`
`366`	`366`	`}`
`367`	`367`
Original file line number	Diff line number	Diff line change
`@@ -973,6 +973,10 @@ private Capability clientCapability(Capability serverCapability) {`
`973`	`973`	`builder.disableConnectWithDatabase();`
`974`	`974`	`}`
`975`	`975`
	`976`	`+ if (context.getLocalInfilePath() == null) {`
	`977`	`+ builder.disableLoadDataLocalInfile();`
	`978`	`+ }`
	`979`	`+`
`976`	`980`	`if (ATTRIBUTES.isEmpty()) {`
`977`	`981`	`builder.disableConnectAttributes();`
`978`	`982`	`}`