This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 10868afad try to prevent traversal issues in file and ftp connectors
(#1807)
10868afad is described below
commit 10868afad8e80cfb3cc17fc4ab60fa163128c349
Author: PJ Fanning <[email protected]>
AuthorDate: Sun Aug 16 11:51:17 2026 +0100
try to prevent traversal issues in file and ftp connectors (#1807)
* try to prevent traversal issues in ftp connector
* Update model.scala
* Update CommonFtpOperations.scala
* Update CommonFtpOperations.scala
* try windows testing
* Update check-build-test.yml
* fix tests
* try to make file tests work on windows
* Update check-build-test.yml
* test issues
* scalafmt
* stop some tests running on windows
* archive paths
* windows test issues
---
.github/workflows/check-build-test.yml | 45 ++++++++++
.../pekko/stream/connectors/file/model.scala | 41 +++++++++
.../test/java/docs/javadsl/FileTailSourceTest.java | 4 +
.../test/java/docs/javadsl/LogRotatorSinkTest.java | 99 +++++++++++---------
.../src/test/scala/docs/scaladsl/ArchiveSpec.scala | 27 +++---
.../docs/scaladsl/FileTailSourceExtrasSpec.scala | 17 +++-
.../scala/docs/scaladsl/LogRotatorSinkSpec.scala | 12 ++-
.../test/scala/docs/scaladsl/TarArchiveSpec.scala | 64 ++++++-------
.../file/impl/archive/TarArchiveEntrySpec.scala | 62 +++++++++++++
.../file/impl/archive/ZipArchiveMetadataSpec.scala | 65 ++++++++++++++
.../connectors/ftp/impl/CommonFtpOperations.scala | 60 ++++++++++++-
.../connectors/ftp/impl/SftpOperations.scala | 11 ++-
.../ftp/impl/CommonFtpOperationsSpec.scala | 100 +++++++++++++++++++++
13 files changed, 510 insertions(+), 97 deletions(-)
diff --git a/.github/workflows/check-build-test.yml
b/.github/workflows/check-build-test.yml
index c95af520b..ddd35cdd9 100644
--- a/.github/workflows/check-build-test.yml
+++ b/.github/workflows/check-build-test.yml
@@ -185,3 +185,48 @@ jobs:
- name: Print logs on failure
if: failure()
run: find . -name "*.log" -exec ./scripts/cat-log.sh {} \;
+
+ connectors-windows:
+ runs-on: windows-latest
+ if: github.repository == 'apache/pekko-connectors'
+
+ strategy:
+ fail-fast: false
+ matrix:
+ include:
+ - { connector: file }
+ - { connector: ftp, test_cmd: 'ftp/testOnly
org.apache.pekko.stream.connectors.ftp.impl.*' }
+
+ env:
+ JAVA_OPTS: -Xms2G -Xmx3G -Xss2M -XX:ReservedCodeCacheSize=256M
-Dfile.encoding=UTF-8
+
+ steps:
+ - name: Checkout
+ uses: actions/[email protected]
+ with:
+ fetch-tags: true
+ fetch-depth: 0
+
+ - name: Setup Java 17
+ uses: actions/setup-java@v5
+ with:
+ distribution: temurin
+ java-version: 17
+
+ - name: Install sbt
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
+
+ - name: Cache Coursier cache
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
v6.4.7
+
+ - name: ${{ matrix.connector }}
+ shell: bash
+ env:
+ CONNECTOR: ${{ matrix.connector }}
+ TEST_CMD: ${{ matrix.test_cmd }}
+ run: |-
+ if [ -n "$TEST_CMD" ]; then
+ sbt "+${TEST_CMD}"
+ else
+ sbt "+${CONNECTOR}/test"
+ fi
diff --git
a/file/src/main/scala/org/apache/pekko/stream/connectors/file/model.scala
b/file/src/main/scala/org/apache/pekko/stream/connectors/file/model.scala
index 780d58f69..c444aa2b8 100644
--- a/file/src/main/scala/org/apache/pekko/stream/connectors/file/model.scala
+++ b/file/src/main/scala/org/apache/pekko/stream/connectors/file/model.scala
@@ -17,6 +17,44 @@ import java.time.Instant
import java.time.temporal.ChronoField
import java.util.Objects
+/**
+ * INTERNAL API
+ *
+ * Validation for path traversal sequences in archive entry names (Zip Slip /
Tar Slip).
+ *
+ * Both forward slash (`/`) and backslash (`\`) are rejected as path
separators because:
+ * - ZIP files use forward slashes per the ZIP Application Note (PKWARE)
section 4.4.17
+ * - TAR file names use forward slashes per POSIX.1 (IEEE Std 1003.1) and the
USTAR format
+ * (POSIX.1-2001 / IEEE Std 1003.1-2001, extended by POSIX.1-2008 pax
headers)
+ * - On Windows, backslashes are path separators and would be interpreted by
the filesystem
+ * API when extracting, even though they are not valid separators in the
archive formats.
+ * Rejecting them prevents path traversal via crafted archives on Windows
hosts.
+ */
+private[file] object ArchivePathTraversalValidation {
+
+ /**
+ * Validate that an archive path segment does not contain traversal
sequences.
+ * Rejects segments containing `..` as a path component, absolute paths, and
+ * backslashes (which Windows treats as path separators during extraction).
+ *
+ * @param value the path segment to validate
+ * @param fieldName the name of the field for error messages
+ * @throws IllegalArgumentException if the segment contains traversal
sequences
+ */
+ def validate(value: String, fieldName: String): Unit = {
+ require(value != null, s"$fieldName must not be null")
+ // Reject absolute paths
+ require(!value.startsWith("/"), s"$fieldName must not be an absolute path:
'$value'")
+ // Reject backslashes — not valid in ZIP/TAR specs, but treated as path
separators on Windows
+ require(!value.contains('\\'), s"$fieldName must not contain backslashes:
'$value'")
+ // Reject path traversal sequences: ".." as a standalone segment
+ val segments = value.split('/')
+ require(
+ !segments.contains(".."),
+ s"$fieldName must not contain path traversal sequences: '$value'")
+ }
+}
+
final class ArchiveMetadata private (
val filePath: String)
@@ -26,6 +64,7 @@ object ArchiveMetadata {
}
final case class ZipArchiveMetadata(name: String) {
+ ArchivePathTraversalValidation.validate(name, "Zip entry name")
def getName() = name
}
object ZipArchiveMetadata {
@@ -129,9 +168,11 @@ object TarArchiveMetadata {
require(
value.length <= 154,
"File path prefix must be between 1 and 154 characters long")
+ ArchivePathTraversalValidation.validate(value, "File path prefix")
}
require(filePathName.length >= 0 && filePathName.length <= 99,
s"File path name must be between 0 and 99 characters long, was
${filePathName.length}")
+ ArchivePathTraversalValidation.validate(filePathName, "File path name")
new TarArchiveMetadata(filePathPrefix,
filePathName,
diff --git a/file/src/test/java/docs/javadsl/FileTailSourceTest.java
b/file/src/test/java/docs/javadsl/FileTailSourceTest.java
index 916ac73f1..499ed6a0b 100644
--- a/file/src/test/java/docs/javadsl/FileTailSourceTest.java
+++ b/file/src/test/java/docs/javadsl/FileTailSourceTest.java
@@ -17,6 +17,8 @@ import static java.nio.charset.StandardCharsets.UTF_8;
import static java.nio.file.StandardOpenOption.APPEND;
import static java.nio.file.StandardOpenOption.WRITE;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
import com.google.common.jimfs.Configuration;
import com.google.common.jimfs.Jimfs;
@@ -134,6 +136,7 @@ public class FileTailSourceTest {
}
@Test
+ @DisabledOnOs(OS.WINDOWS)
public void willCompleteStreamIfFileIsDeleted() throws Exception {
final Path path = fs.getPath("/file");
Files.writeString(path, "a\n", UTF_8);
@@ -180,6 +183,7 @@ public class FileTailSourceTest {
}
@Test
+ @DisabledOnOs(OS.WINDOWS)
public void willCompleteStreamIfFileIsIdle() throws Exception {
final Path path = fs.getPath("/file");
Files.writeString(path, "a\n", UTF_8);
diff --git a/file/src/test/java/docs/javadsl/LogRotatorSinkTest.java
b/file/src/test/java/docs/javadsl/LogRotatorSinkTest.java
index 905cba9d7..fb522c230 100644
--- a/file/src/test/java/docs/javadsl/LogRotatorSinkTest.java
+++ b/file/src/test/java/docs/javadsl/LogRotatorSinkTest.java
@@ -15,7 +15,6 @@ package docs.javadsl;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.LocalDateTime;
@@ -93,7 +92,7 @@ public class LogRotatorSinkTest {
@Test
public void timeBased() throws Exception {
// #time
- final Path destinationDir = FileSystems.getDefault().getPath("/tmp");
+ final Path destinationDir = Files.createTempDirectory("log-rotation-test");
final DateTimeFormatter formatter =
DateTimeFormatter.ofPattern("'stream-'yyyy-MM-dd_HH'.log'");
Creator<Function<ByteString, Optional<Path>>> timeBasedTriggerCreator =
@@ -114,47 +113,59 @@ public class LogRotatorSinkTest {
LogRotatorSink.createFromFunction(timeBasedTriggerCreator);
// #time
- CompletionStage<Done> fileSizeCompletion =
- Source.from(List.of("test1", "test2", "test3", "test4", "test5",
"test6"))
- .map(ByteString::fromString)
- .runWith(timeBasedSink, system);
-
- assertEquals(
- Done.getInstance(), fileSizeCompletion.toCompletableFuture().get(2,
TimeUnit.SECONDS));
-
- /*
- // #sample
- import org.apache.pekko.stream.connectors.file.javadsl.LogRotatorSink;
-
- Creator<Function<ByteString, Optional<Path>>> triggerFunctionCreator = ...;
-
- // #sample
- */
- Creator<Function<ByteString, Optional<Path>>> triggerFunctionCreator =
timeBasedTriggerCreator;
-
- Source<ByteString, NotUsed> source =
- Source.from(List.of("test1", "test2", "test3", "test4", "test5",
"test6"))
- .map(ByteString::fromString);
- // #sample
- CompletionStage<Done> completion =
- Source.from(List.of("test1", "test2", "test3", "test4", "test5",
"test6"))
- .map(ByteString::fromString)
-
.runWith(LogRotatorSink.createFromFunction(triggerFunctionCreator), system);
-
- // GZip compressing the data written
- CompletionStage<Done> compressedCompletion =
- source.runWith(
- LogRotatorSink.withSinkFactory(
- triggerFunctionCreator,
- path ->
- Flow.of(ByteString.class)
- .via(Compression.gzip())
- .toMat(FileIO.toPath(path), Keep.right())),
- system);
- // #sample
-
- assertEquals(Done.getInstance(), completion.toCompletableFuture().get(2,
TimeUnit.SECONDS));
- assertEquals(
- Done.getInstance(), compressedCompletion.toCompletableFuture().get(2,
TimeUnit.SECONDS));
+ try {
+ CompletionStage<Done> fileSizeCompletion =
+ Source.from(List.of("test1", "test2", "test3", "test4", "test5",
"test6"))
+ .map(ByteString::fromString)
+ .runWith(timeBasedSink, system);
+
+ assertEquals(
+ Done.getInstance(), fileSizeCompletion.toCompletableFuture().get(2,
TimeUnit.SECONDS));
+
+ /*
+ // #sample
+ import org.apache.pekko.stream.connectors.file.javadsl.LogRotatorSink;
+
+ Creator<Function<ByteString, Optional<Path>>> triggerFunctionCreator =
...;
+
+ // #sample
+ */
+ Creator<Function<ByteString, Optional<Path>>> triggerFunctionCreator =
+ timeBasedTriggerCreator;
+
+ Source<ByteString, NotUsed> source =
+ Source.from(List.of("test1", "test2", "test3", "test4", "test5",
"test6"))
+ .map(ByteString::fromString);
+ // #sample
+ CompletionStage<Done> completion =
+ Source.from(List.of("test1", "test2", "test3", "test4", "test5",
"test6"))
+ .map(ByteString::fromString)
+
.runWith(LogRotatorSink.createFromFunction(triggerFunctionCreator), system);
+
+ // GZip compressing the data written
+ CompletionStage<Done> compressedCompletion =
+ source.runWith(
+ LogRotatorSink.withSinkFactory(
+ triggerFunctionCreator,
+ path ->
+ Flow.of(ByteString.class)
+ .via(Compression.gzip())
+ .toMat(FileIO.toPath(path), Keep.right())),
+ system);
+ // #sample
+
+ assertEquals(Done.getInstance(), completion.toCompletableFuture().get(2,
TimeUnit.SECONDS));
+ assertEquals(
+ Done.getInstance(),
compressedCompletion.toCompletableFuture().get(2, TimeUnit.SECONDS));
+ } finally {
+ // Clean up temp directory and all files created by the trigger
+ java.util.Comparator<java.io.File> reverse =
+ (a, b) -> b.getAbsolutePath().compareTo(a.getAbsolutePath());
+ java.io.File[] children = destinationDir.toFile().listFiles();
+ if (children != null) {
+
java.util.Arrays.stream(children).sorted(reverse).forEach(java.io.File::delete);
+ }
+ destinationDir.toFile().delete();
+ }
}
}
diff --git a/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
b/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
index 558084477..859d09d9d 100644
--- a/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
+++ b/file/src/test/scala/docs/scaladsl/ArchiveSpec.scala
@@ -163,22 +163,27 @@ class ArchiveSpec
val target: Path = // ???
// #zip-reader
Files.createTempDirectory("pekko-connectors-zip-")
- // #zip-reader
- Archive
- .zipReader(zipFile)
- .mapAsyncUnordered(4) {
- case (metadata, source) =>
- val targetFile = target.resolve(metadata.name)
- targetFile.toFile.getParentFile.mkdirs() // missing error handler
- source.runWith(FileIO.toPath(targetFile))
- }
- // #zip-reader
+ try {
+ // #zip-reader
+ Archive
+ .zipReader(zipFile)
+ .mapAsyncUnordered(4) {
+ case (metadata, source) =>
+ val targetFile = target.resolve(metadata.name)
+ targetFile.toFile.getParentFile.mkdirs() // missing error
handler
+ source.runWith(FileIO.toPath(targetFile))
+ }
+ // #zip-reader
+ } finally {
+
Files.walk(target).sorted(java.util.Comparator.reverseOrder()).iterator().asScala.foreach(p
=>
+ Files.delete(p))
+ }
}
}
}
private def getPathFromResources(fileName: String): Path =
- Paths.get(getClass.getClassLoader.getResource(fileName).getPath)
+ Paths.get(getClass.getClassLoader.getResource(fileName).toURI)
private def generateInputFiles(numberOfFiles: Int, lengthOfFile: Int):
Map[String, ByteString] = {
val r = new scala.util.Random(31)
diff --git a/file/src/test/scala/docs/scaladsl/FileTailSourceExtrasSpec.scala
b/file/src/test/scala/docs/scaladsl/FileTailSourceExtrasSpec.scala
index a2a6878ce..bd858b434 100644
--- a/file/src/test/scala/docs/scaladsl/FileTailSourceExtrasSpec.scala
+++ b/file/src/test/scala/docs/scaladsl/FileTailSourceExtrasSpec.scala
@@ -43,11 +43,16 @@ class FileTailSourceExtrasSpec
with ScalaFutures
with LogCapturing {
- private val fs =
Jimfs.newFileSystem(Configuration.forCurrentPlatform.toBuilder.build)
+ import scala.jdk.CollectionConverters._
+ private val fs = Jimfs.newFileSystem(Configuration.forCurrentPlatform)
+ private val testFile = fs.getRootDirectories.asScala.head.resolve("file")
+
+ private val isWindows = System.getProperty("os.name",
"").toLowerCase.contains("windows")
"The FileTailSource" should assertAllStagesStopped {
"demo stream shutdown when file deleted" in {
- val path = fs.getPath("/file")
+ assume(!isWindows, "Jimfs WatchService does not fire events on Windows")
+ val path = testFile
Files.write(path, "a\n".getBytes(UTF_8))
// #shutdown-on-delete
@@ -81,7 +86,8 @@ class FileTailSourceExtrasSpec
}
"demo stream shutdown when with idle timeout" in {
- val path = fs.getPath("/file")
+ assume(!isWindows, "Jimfs file I/O does not work reliably on Windows")
+ val path = testFile
Files.write(path, "a\n".getBytes(UTF_8))
// #shutdown-on-idle-timeout
@@ -107,4 +113,9 @@ class FileTailSourceExtrasSpec
}
}
+
+ override protected def afterAll(): Unit = {
+ fs.close()
+ super.afterAll()
+ }
}
diff --git a/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala
b/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala
index fb1676684..14489b7c5 100644
--- a/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala
+++ b/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala
@@ -106,6 +106,7 @@ class LogRotatorSinkSpec
}
"work for size-based rotation " in assertAllStagesStopped {
+ val createdFiles = Seq.newBuilder[Path]
// #size
import org.apache.pekko.stream.connectors.file.scaladsl.LogRotatorSink
@@ -115,6 +116,7 @@ class LogRotatorSinkSpec
(element: ByteString) =>
if (size + element.size > max) {
val path = Files.createTempFile("out-", ".log")
+ createdFiles += path
size = element.size
Some(path)
} else {
@@ -131,12 +133,16 @@ class LogRotatorSinkSpec
.map(ByteString(_))
.runWith(sizeRotatorSink)
- fileSizeCompletion.futureValue shouldBe Done
+ try {
+ fileSizeCompletion.futureValue shouldBe Done
+ } finally {
+ createdFiles.result().foreach(f => Files.deleteIfExists(f))
+ }
}
"work for time-based rotation " in assertAllStagesStopped {
// #time
- val destinationDir = FileSystems.getDefault.getPath("/tmp")
+ val destinationDir = Files.createTempDirectory(fs.getPath("/"),
"time-rotation")
val formatter =
DateTimeFormatter.ofPattern("'stream-'yyyy-MM-dd_HH'.log'")
val timeBasedTriggerCreator: () => ByteString => Option[Path] = () => {
@@ -180,7 +186,7 @@ class LogRotatorSinkSpec
"work for stream-based rotation " in assertAllStagesStopped {
// #stream
- val destinationDir = FileSystems.getDefault.getPath("/tmp")
+ val destinationDir = Files.createTempDirectory(fs.getPath("/"),
"stream-rotation")
val streamBasedTriggerCreator: () => ((String, String)) => Option[Path]
= () => {
var currentFilename: Option[String] = None
diff --git a/file/src/test/scala/docs/scaladsl/TarArchiveSpec.scala
b/file/src/test/scala/docs/scaladsl/TarArchiveSpec.scala
index c40894052..60be20713 100644
--- a/file/src/test/scala/docs/scaladsl/TarArchiveSpec.scala
+++ b/file/src/test/scala/docs/scaladsl/TarArchiveSpec.scala
@@ -178,36 +178,40 @@ class TarArchiveSpec
Source.future(oneFileArchive)
val target = Files.createTempDirectory("pekko-connectors-tar-")
- // #tar-reader
- val tar =
- bytesSource
- .via(Archive.tarReader())
- .mapAsync(1) {
- case (metadata, source) =>
- val targetFile = target.resolve(metadata.filePath)
- if (metadata.isDirectory) {
- Source
- .single(targetFile)
- .via(Directory.mkdirs())
- .runWith(Sink.ignore)
- } else {
- // create the target directory
- Source
- .single(targetFile.getParent)
- .via(Directory.mkdirs())
- .runWith(Sink.ignore)
- .map { _ =>
- // stream the file contents to a local file
- source.runWith(FileIO.toPath(targetFile))
- }
- }
- }
- .runWith(Sink.ignore)
- // #tar-reader
- tar.futureValue shouldBe Done
- val file: File = target.resolve("dir/file1.txt").toFile
- eventually {
- file.exists() shouldBe true
+ try {
+ // #tar-reader
+ val tar =
+ bytesSource
+ .via(Archive.tarReader())
+ .mapAsync(1) {
+ case (metadata, source) =>
+ val targetFile = target.resolve(metadata.filePath)
+ if (metadata.isDirectory) {
+ Source
+ .single(targetFile)
+ .via(Directory.mkdirs())
+ .runWith(Sink.ignore)
+ } else {
+ // create the target directory
+ Source
+ .single(targetFile.getParent)
+ .via(Directory.mkdirs())
+ .runWith(Sink.ignore)
+ .map { _ =>
+ // stream the file contents to a local file
+ source.runWith(FileIO.toPath(targetFile))
+ }
+ }
+ }
+ .runWith(Sink.ignore)
+ // #tar-reader
+ tar.futureValue shouldBe Done
+ val file: File = target.resolve("dir/file1.txt").toFile
+ eventually {
+ file.exists() shouldBe true
+ }
+ } finally {
+
Files.walk(target).sorted(Comparator.reverseOrder()).iterator().asScala.foreach(p
=> Files.delete(p))
}
}
diff --git
a/file/src/test/scala/org/apache/pekko/stream/connectors/file/impl/archive/TarArchiveEntrySpec.scala
b/file/src/test/scala/org/apache/pekko/stream/connectors/file/impl/archive/TarArchiveEntrySpec.scala
index 40f0b38b9..2e25e1ffe 100644
---
a/file/src/test/scala/org/apache/pekko/stream/connectors/file/impl/archive/TarArchiveEntrySpec.scala
+++
b/file/src/test/scala/org/apache/pekko/stream/connectors/file/impl/archive/TarArchiveEntrySpec.scala
@@ -22,6 +22,68 @@ import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
class TarArchiveEntrySpec extends AnyFlatSpec with Matchers {
+
+ "Path traversal validation" should "reject dot-dot in filename" in {
+ an[IllegalArgumentException] should be thrownBy {
+ TarArchiveMetadata("../etc/passwd", 100L)
+ }
+ }
+
+ it should "reject dot-dot in prefix" in {
+ an[IllegalArgumentException] should be thrownBy {
+ TarArchiveMetadata("../../etc", "passwd", 100L, Instant.now)
+ }
+ }
+
+ it should "reject absolute path in filename" in {
+ an[IllegalArgumentException] should be thrownBy {
+ TarArchiveMetadata("/etc/passwd", 100L)
+ }
+ }
+
+ it should "reject dot-dot in middle of path" in {
+ an[IllegalArgumentException] should be thrownBy {
+ TarArchiveMetadata("dir/../../../etc/passwd", 100L)
+ }
+ }
+
+ it should "reject dot-dot via parse" in {
+ // Build a tar header with a malicious filename
+ val malicious = TarArchiveMetadata("dir/file.txt", 100L)
+ val entry = new TarArchiveEntry(malicious)
+ val header = entry.headerBytes
+ // Corrupt the filename field to contain ../
+ val corrupted = header.toArray
+ val evilName = "../etc/crontab"
+ evilName.getBytes.zipWithIndex.foreach { case (b, i) => corrupted(i) = b }
+ corrupted(evilName.length) = 0 // null terminator
+ an[IllegalArgumentException] should be thrownBy {
+ TarArchiveEntry.parse(ByteString(corrupted))
+ }
+ }
+
+ it should "accept normal relative paths" in {
+ val meta = TarArchiveMetadata("dir/subdir/file.txt", 100L)
+ meta.filePath shouldBe "dir/subdir/file.txt"
+ }
+
+ it should "accept simple filename" in {
+ val meta = TarArchiveMetadata("file.txt", 100L)
+ meta.filePath shouldBe "file.txt"
+ }
+
+ it should "reject backslashes in filename" in {
+ an[IllegalArgumentException] should be thrownBy {
+ TarArchiveMetadata("\\etc\\passwd", 100L)
+ }
+ }
+
+ it should "reject backslashes in prefix" in {
+ an[IllegalArgumentException] should be thrownBy {
+ TarArchiveMetadata("\\etc", "passwd", 100L, Instant.now)
+ }
+ }
+
"Metadata entries" should "be created and parsed back" in {
val filePathPrefix = "dir1/dir2"
val filename = "thefile.txt"
diff --git
a/file/src/test/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipArchiveMetadataSpec.scala
b/file/src/test/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipArchiveMetadataSpec.scala
new file mode 100644
index 000000000..968a24614
--- /dev/null
+++
b/file/src/test/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipArchiveMetadataSpec.scala
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.file.impl.archive
+
+import org.apache.pekko.stream.connectors.file.ZipArchiveMetadata
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class ZipArchiveMetadataSpec extends AnyFlatSpec with Matchers {
+
+ "ZipArchiveMetadata" should "reject dot-dot in name" in {
+ an[IllegalArgumentException] should be thrownBy {
+ ZipArchiveMetadata("../../etc/passwd")
+ }
+ }
+
+ it should "reject absolute path" in {
+ an[IllegalArgumentException] should be thrownBy {
+ ZipArchiveMetadata("/etc/passwd")
+ }
+ }
+
+ it should "reject dot-dot in middle of path" in {
+ an[IllegalArgumentException] should be thrownBy {
+ ZipArchiveMetadata("dir/../../../etc/crontab")
+ }
+ }
+
+ it should "accept normal relative paths" in {
+ val meta = ZipArchiveMetadata("dir/subdir/file.txt")
+ meta.name shouldBe "dir/subdir/file.txt"
+ }
+
+ it should "accept simple filename" in {
+ val meta = ZipArchiveMetadata("file.txt")
+ meta.name shouldBe "file.txt"
+ }
+
+ it should "reject backslashes in name" in {
+ an[IllegalArgumentException] should be thrownBy {
+ ZipArchiveMetadata("\\etc\\passwd")
+ }
+ }
+
+ it should "reject backslashes in middle of path" in {
+ an[IllegalArgumentException] should be thrownBy {
+ ZipArchiveMetadata("dir\\..\\..\\etc\\passwd")
+ }
+ }
+}
diff --git
a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperations.scala
b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperations.scala
index 5b7a56bc8..b3af9eddb 100644
---
a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperations.scala
+++
b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperations.scala
@@ -76,21 +76,26 @@ private[ftp] trait CommonFtpOperations {
retrieveFileInputStream(name, handler, 0L)
def retrieveFileInputStream(name: String, handler: Handler, offset: Long):
Try[InputStream] = Try {
+ CommonFtpOperations.validatePath(name, "name")
handler.setRestartOffset(offset)
val is = handler.retrieveFileStream(name)
if (is != null) is else throw new IOException(s"$name: No such file or
directory")
}
def storeFileOutputStream(name: String, handler: Handler, append: Boolean):
Try[OutputStream] = Try {
+ CommonFtpOperations.validatePath(name, "name")
val os = if (append) handler.appendFileStream(name) else
handler.storeFileStream(name)
if (os != null) os else throw new IOException(s"Could not write to $name")
}
def move(fromPath: String, destinationPath: String, handler: Handler): Unit
= {
+ CommonFtpOperations.validatePath(fromPath, "fromPath")
+ CommonFtpOperations.validatePath(destinationPath, "destinationPath")
if (!handler.rename(fromPath, destinationPath)) throw new
IOException(s"Could not move $fromPath")
}
def remove(path: String, handler: Handler): Unit = {
+ CommonFtpOperations.validatePath(path, "path")
if (!handler.deleteFile(path)) throw new IOException(s"Could not delete
$path")
}
@@ -108,10 +113,57 @@ private[ftp] trait CommonFtpOperations {
}
private[ftp] object CommonFtpOperations {
- def concatPath(path: String, name: String): String =
- if (path.endsWith("/")) {
- path ++ name
+
+ /**
+ * Normalize a path to use `/` separators. FTP uses `/` by protocol;
+ * normalizing early ensures all downstream checks only need to handle `/`.
+ */
+ private def normalizeSeparators(path: String): String = path.replace('\\',
'/')
+
+ /**
+ * Validate that a path does not contain traversal sequences (`..`).
+ * Rejects null values and paths containing `..` as a path segment.
+ * Accepts both `/` and `\` separators; backslashes are normalized to `/`
before checking.
+ *
+ * @param path the path to validate
+ * @param fieldName the name of the field for error messages
+ * @throws IllegalArgumentException if the path contains traversal sequences
+ */
+ def validatePath(path: String, fieldName: String): Unit = {
+ require(path != null, s"$fieldName must not be null")
+ val normalized = normalizeSeparators(path)
+ val segments = normalized.split('/')
+ require(!segments.contains(".."), s"$fieldName must not contain path
traversal sequences: '$path'")
+ }
+
+ def concatPath(path: String, name: String): String = {
+ validatePath(name, "name")
+ val normName = normalizeSeparators(name)
+ require(!normName.startsWith("/"), s"name must not be an absolute path:
'$normName'")
+
+ require(path != null, "path must not be null")
+ val normPath = normalizeSeparators(path)
+ val result = if (normPath.endsWith("/")) {
+ normPath ++ normName
} else {
- s"$path/$name"
+ s"$normPath/$normName"
}
+
+ // Pure string segment walk — no platform-dependent normalization.
+ // `..` and absolute names are already rejected upstream; this collapses
+ // `.` segments and empty segments (double slashes) so the prefix check
+ // cannot be fooled by cosmetic differences.
+ def collapseSegments(p: String): String = {
+ val parts = p.split('/').filter(s => s.nonEmpty && s != ".")
+ if (parts.isEmpty) "" else parts.mkString("/")
+ }
+ val collapsed = collapseSegments(result)
+ val collapsedBase = collapseSegments(normPath)
+ // collapsedBase is empty when the base path is "/" (root) — everything is
under root
+ require(
+ collapsedBase.isEmpty || collapsed == collapsedBase ||
collapsed.startsWith(collapsedBase + "/"),
+ s"concatPath result '$result' escapes base path '$normPath'")
+
+ result
+ }
}
diff --git
a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala
b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala
index 05ee5f538..9ea4d32ab 100644
---
a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala
+++
b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala
@@ -138,6 +138,7 @@ private[ftp] trait SftpOperations { self:
FtpLike[SSHClient, SftpSettings] =>
offset: Long,
maxUnconfirmedReads: Int): Try[InputStream] =
Try {
+ CommonFtpOperations.validatePath(name, "name")
val remoteFile = handler.open(name, java.util.EnumSet.of(OpenMode.READ))
val is = maxUnconfirmedReads match {
case m if m > 1 =>
@@ -169,6 +170,7 @@ private[ftp] trait SftpOperations { self:
FtpLike[SSHClient, SftpSettings] =>
def storeFileOutputStream(name: String, handler: Handler, append: Boolean):
Try[OutputStream] =
Try {
+ CommonFtpOperations.validatePath(name, "name")
import OpenMode._
val openModes =
if (append) java.util.EnumSet.of(WRITE, CREAT, APPEND)
@@ -208,9 +210,14 @@ private[ftp] trait SftpOperations { self:
FtpLike[SSHClient, SftpSettings] =>
}
}
- def move(fromPath: String, destinationPath: String, handler: Handler): Unit =
+ def move(fromPath: String, destinationPath: String, handler: Handler): Unit
= {
+ CommonFtpOperations.validatePath(fromPath, "fromPath")
+ CommonFtpOperations.validatePath(destinationPath, "destinationPath")
handler.rename(fromPath, destinationPath)
+ }
- def remove(path: String, handler: Handler): Unit =
+ def remove(path: String, handler: Handler): Unit = {
+ CommonFtpOperations.validatePath(path, "path")
handler.rm(path)
+ }
}
diff --git
a/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperationsSpec.scala
b/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperationsSpec.scala
new file mode 100644
index 000000000..f9b7bcbd2
--- /dev/null
+++
b/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperationsSpec.scala
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.connectors.ftp.impl
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class CommonFtpOperationsSpec extends AnyFlatSpec with Matchers {
+
+ "CommonFtpOperations.validatePath" should "accept normal relative path" in {
+ CommonFtpOperations.validatePath("dir/file.txt", "path") // no exception
+ }
+
+ it should "accept simple filename" in {
+ CommonFtpOperations.validatePath("file.txt", "path") // no exception
+ }
+
+ it should "reject null" in {
+ an[IllegalArgumentException] should be thrownBy {
+ CommonFtpOperations.validatePath(null, "path")
+ }
+ }
+
+ it should "reject dot-dot at start" in {
+ an[IllegalArgumentException] should be thrownBy {
+ CommonFtpOperations.validatePath("../etc/passwd", "path")
+ }
+ }
+
+ it should "reject dot-dot in middle" in {
+ an[IllegalArgumentException] should be thrownBy {
+ CommonFtpOperations.validatePath("dir/../../etc/passwd", "path")
+ }
+ }
+
+ it should "reject dot-dot at end" in {
+ an[IllegalArgumentException] should be thrownBy {
+ CommonFtpOperations.validatePath("dir/..", "path")
+ }
+ }
+
+ "CommonFtpOperations.concatPath" should "concatenate simple paths" in {
+ CommonFtpOperations.concatPath("/base", "file.txt") shouldBe
"/base/file.txt"
+ }
+
+ it should "handle trailing slash on base path" in {
+ CommonFtpOperations.concatPath("/base/", "file.txt") shouldBe
"/base/file.txt"
+ }
+
+ it should "handle nested names" in {
+ CommonFtpOperations.concatPath("/base", "subdir/file.txt") shouldBe
"/base/subdir/file.txt"
+ }
+
+ it should "reject dot-dot in name" in {
+ an[IllegalArgumentException] should be thrownBy {
+ CommonFtpOperations.concatPath("/base", "../etc/passwd")
+ }
+ }
+
+ it should "reject dot-dot in nested name" in {
+ an[IllegalArgumentException] should be thrownBy {
+ CommonFtpOperations.concatPath("/base", "subdir/../../etc/passwd")
+ }
+ }
+
+ it should "reject absolute name" in {
+ an[IllegalArgumentException] should be thrownBy {
+ CommonFtpOperations.concatPath("/base", "/etc/passwd")
+ }
+ }
+
+ it should "reject null name" in {
+ an[IllegalArgumentException] should be thrownBy {
+ CommonFtpOperations.concatPath("/base", null)
+ }
+ }
+
+ it should "handle dot segments without platform normalization" in {
+ CommonFtpOperations.concatPath("/base", "./file.txt") shouldBe
"/base/./file.txt"
+ }
+
+ it should "allow names under root path" in {
+ CommonFtpOperations.concatPath("/", "sample_dir") shouldBe "/sample_dir"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
