This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new aedb62928 Serve only worker files that resolve inside the log root
(#9182)
aedb62928 is described below
commit aedb629289026c96edde8c39f16e35bc91c8a198
Author: Richard Zowalla <[email protected]>
AuthorDate: Fri Oct 23 19:43:17 2026 +0200
Serve only worker files that resolve inside the log root (#9182)
* Serve only worker files that resolve inside the log root
The Logviewer no longer follows symbolic links below the worker log root
for downloads, log pages, searches and dump files. Worker files must be
direct children of a port directory; links are answered like missing files
and are left out of file listings. Link-free paths and a linked log root
keep working, and daemon logs are unchanged.
* Leave symbolic links out of the log list and answer a late link like a
missing file
The log list and getAllLogsForRootDir leave out topology, port and file
links.
A worker file that turns into a link between the check and the open is
answered
like a missing file instead of a read error, and the daemon log page takes
its
length from the opened file like the worker log page does.
* Answer a file removed before the open like a missing file and drop unused
helpers
A log or dump file removed between the check and the open is answered with
404 and not counted as an error.
Only numeric ports name a dump directory; the unused file based download
helper is removed and zipFileSize is deprecated.
* Close downloaded files and check port directories in deep search and dump
listing
Deep search and the dump file listing now leave out port directories that
are reached through a symbolic link, like the log listing already does.
Worker files are opened through LogFilePaths.openWorkerFile, which checks
the file before opening it. The download stream is closed once it has been
written. The port of a host:port is the part after the last ':', so IPv6
hosts work. Tests creating symbolic links are skipped where links can not
be created.
---
.../java/org/apache/storm/utils/ServerUtils.java | 2 +
.../logviewer/handler/LogviewerLogPageHandler.java | 115 +++++---
.../handler/LogviewerLogSearchHandler.java | 68 ++++-
.../logviewer/handler/LogviewerProfileHandler.java | 57 +++-
.../daemon/logviewer/utils/LogFileDownloader.java | 37 ++-
.../storm/daemon/logviewer/utils/LogFilePaths.java | 210 ++++++++++++++
.../logviewer/utils/LogviewerResponseBuilder.java | 43 ++-
.../storm/daemon/logviewer/utils/WorkerLogs.java | 12 +-
.../handler/LogviewerLogDownloadHandlerTest.java | 261 +++++++++++++++++
.../handler/LogviewerLogPageHandlerTest.java | 313 +++++++++++++++++++++
.../handler/LogviewerLogSearchHandlerTest.java | 238 ++++++++++++++++
.../handler/LogviewerProfileHandlerTest.java | 250 ++++++++++++++++
.../logviewer/testsupport/SymbolicLinks.java | 56 ++++
.../daemon/logviewer/utils/LogFilePathsTest.java | 253 +++++++++++++++++
.../utils/LogviewerResponseBuilderTest.java | 49 ++++
.../daemon/logviewer/utils/WorkerLogsTest.java | 24 ++
16 files changed, 1891 insertions(+), 97 deletions(-)
diff --git a/storm-server/src/main/java/org/apache/storm/utils/ServerUtils.java
b/storm-server/src/main/java/org/apache/storm/utils/ServerUtils.java
index 066334c83..ea141e2e9 100644
--- a/storm-server/src/main/java/org/apache/storm/utils/ServerUtils.java
+++ b/storm-server/src/main/java/org/apache/storm/utils/ServerUtils.java
@@ -659,7 +659,9 @@ public class ServerUtils {
*
* @param myFile The zip file as input
* @return zip file size as a long
+ * @deprecated no longer used by Storm, it will be removed in a future
release
*/
+ @Deprecated
public static long zipFileSize(File myFile) throws IOException {
try (RandomAccessFile raf = new RandomAccessFile(myFile, "r")) {
raf.seek(raf.length() - 4);
diff --git
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogPageHandler.java
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogPageHandler.java
index b1144ca65..776075715 100644
---
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogPageHandler.java
+++
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogPageHandler.java
@@ -45,11 +45,11 @@ import jakarta.ws.rs.core.Response;
import java.io.ByteArrayOutputStream;
import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
+import java.nio.channels.Channels;
+import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -69,6 +69,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.storm.daemon.logviewer.LogviewerConstant;
import org.apache.storm.daemon.logviewer.utils.DirectoryCleaner;
import org.apache.storm.daemon.logviewer.utils.ExceptionMeterNames;
+import org.apache.storm.daemon.logviewer.utils.LogFilePaths;
import org.apache.storm.daemon.logviewer.utils.LogviewerResponseBuilder;
import org.apache.storm.daemon.logviewer.utils.ResourceAuthorizer;
import org.apache.storm.daemon.logviewer.utils.WorkerLogs;
@@ -78,7 +79,6 @@ import org.apache.storm.daemon.utils.StreamUtil;
import org.apache.storm.daemon.utils.UrlBuilder;
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.utils.ConfigUtils;
-import org.apache.storm.utils.ServerUtils;
public class LogviewerLogPageHandler {
private final Meter numPageRead;
@@ -137,7 +137,8 @@ public class LogviewerLogPageHandler {
File[] topoDirFiles = topoDir.listFiles();
if (topoDirFiles != null) {
for (File portDir : topoDirFiles) {
- if (portDir.getName().equals(port.toString()))
{
+ if (portDir.getName().equals(port.toString())
+ &&
LogFilePaths.isWorkerDirWithoutSymbolicLink(logRoot, portDir.toPath())) {
fileResults.addAll(directoryCleaner.getFilesForDir(portDir.toPath()));
}
}
@@ -157,18 +158,20 @@ public class LogviewerLogPageHandler {
File[] topoDirFiles = topoDir.toFile().listFiles();
if (topoDirFiles != null) {
for (File portDir : topoDirFiles) {
-
fileResults.addAll(directoryCleaner.getFilesForDir(portDir.toPath()));
+ if
(LogFilePaths.isWorkerDirWithoutSymbolicLink(logRoot, portDir.toPath())) {
+
fileResults.addAll(directoryCleaner.getFilesForDir(portDir.toPath()));
+ }
}
}
}
} else {
- File portDir =
ConfigUtils.getWorkerDirFromRoot(logRoot.toString(), topologyId,
port).getCanonicalFile();
- if (!portDir.getPath().startsWith(logRoot.toString())) {
+ Path portDir =
ConfigUtils.getWorkerDirFromRoot(logRoot.toString(), topologyId,
port).toPath().toAbsolutePath().normalize();
+ if (!portDir.startsWith(logRoot)) {
return
LogviewerResponseBuilder.buildSuccessJsonResponse(Collections.emptyList(),
callback, origin);
}
- if (portDir.exists()) {
- fileResults =
directoryCleaner.getFilesForDir(portDir.toPath());
+ if (LogFilePaths.isWorkerDirWithoutSymbolicLink(logRoot,
portDir)) {
+ fileResults = directoryCleaner.getFilesForDir(portDir);
}
}
}
@@ -176,7 +179,10 @@ public class LogviewerLogPageHandler {
List<String> files;
if (fileResults != null) {
Map<String, Boolean> authorizedPortDirs = new HashMap<>();
+ //Symbolic links are left out of the listing, they are not served
either. Only links are left out, the listing is
+ //otherwise unchanged (other entries such as subdirectories are
still listed as before)
files = fileResults.stream()
+ .filter(file -> !Files.isSymbolicLink(file))
.map(WorkerLogs::getTopologyPortWorkerLog)
.filter(fileStr -> isUserAllowedToAccessLog(user, fileStr,
authorizedPortDirs))
.sorted().collect(toList());
@@ -221,13 +227,16 @@ public class LogviewerLogPageHandler {
}
if (resourceAuthorizer.isUserAllowedToAccessFile(user, fileName)) {
- workerLogs.setLogFilePermission(fileName);
+ //A worker log must be a direct child of a port directory and must
not be reached through a symbolic link,
+ //otherwise it is answered like a missing file
+ if (LogFilePaths.isServableWorkerFile(logRoot, absFile)) {
+ workerLogs.setLogFilePermission(fileName);
- Path topoDir = absFile.getParent().getParent();
- if (absFile.toFile().exists()) {
+ Path topoDir = absFile.getParent().getParent();
SortedSet<Path> logFiles;
try {
logFiles = Arrays.stream(topoDir.toFile().listFiles())
+ .filter(portDir ->
LogFilePaths.isDirectoryNotLink(portDir.toPath()))
.flatMap(portDir -> {
try {
return
directoryCleaner.getFilesForDir(portDir.toPath()).stream();
@@ -235,7 +244,7 @@ public class LogviewerLogPageHandler {
throw new RuntimeException(e);
}
})
- .filter(Files::isRegularFile)
+ .filter(LogFilePaths::isRegularFileNotLink)
.collect(toCollection(TreeSet::new));
} catch (UncheckedIOException e) {
throw e.getCause();
@@ -249,13 +258,23 @@ public class LogviewerLogPageHandler {
length = length != null ? Math.min(10485760, length) :
LogviewerConstant.DEFAULT_BYTES_PER_PAGE;
final boolean isZipFile =
absFile.getFileName().toString().endsWith(".gz");
- long fileLength = getFileLength(absFile.toFile(), isZipFile);
- if (start == null) {
- start = Long.valueOf(fileLength - length).intValue();
- }
+ long fileLength;
+ String logString;
+ //The length is taken from the opened file, so it always
belongs to the content that is shown
+ try (SeekableByteChannel channel = openFile(absFile, false)) {
+ if (channel == null) {
+ //The worker log was removed or replaced by a symbolic
link after the check
+ return
LogviewerResponseBuilder.buildResponsePageNotFound();
+ }
+ fileLength = getOpenedFileLength(channel, isZipFile);
+ if (start == null) {
+ start = Long.valueOf(fileLength - length).intValue();
+ }
- String logString = isTxtFile(fileName) ?
escapeHtml4(pageFile(absFile.toString(), isZipFile, fileLength, start, length))
:
- escapeHtml4("This is a binary file and cannot display! You
may download the full file.");
+ logString = isTxtFile(fileName)
+ ?
escapeHtml4(pageFile(Channels.newInputStream(channel), isZipFile, fileLength,
start, length)) :
+ escapeHtml4("This is a binary file and cannot display!
You may download the full file.");
+ }
List<DomContent> bodyContents = new ArrayList<>();
if (StringUtils.isNotEmpty(grep)) {
@@ -331,13 +350,23 @@ public class LogviewerLogPageHandler {
length = length != null ? Math.min(10485760, length) :
LogviewerConstant.DEFAULT_BYTES_PER_PAGE;
final boolean isZipFile =
file.getFileName().toString().endsWith(".gz");
- long fileLength = getFileLength(file.toFile(), isZipFile);
- if (start == null) {
- start = Long.valueOf(fileLength - length).intValue();
- }
+ long fileLength;
+ String logString;
+ //A daemon log may be a symbolic link, so it is opened following
links
+ try (SeekableByteChannel channel = openFile(file, true)) {
+ if (channel == null) {
+ //The daemon log was removed after the check
+ return
LogviewerResponseBuilder.buildResponsePageNotFound();
+ }
+ fileLength = getOpenedFileLength(channel, isZipFile);
+ if (start == null) {
+ start = Long.valueOf(fileLength - length).intValue();
+ }
- String logString = isTxtFile(fileName) ?
escapeHtml4(pageFile(file.toString(), isZipFile, fileLength, start, length)) :
+ logString = isTxtFile(fileName)
+ ? escapeHtml4(pageFile(Channels.newInputStream(channel),
isZipFile, fileLength, start, length)) :
escapeHtml4("This is a binary file and cannot display! You
may download the full file.");
+ }
List<DomContent> bodyContents = new ArrayList<>();
if (StringUtils.isNotEmpty(grep)) {
@@ -370,18 +399,6 @@ public class LogviewerLogPageHandler {
}
}
- private long getFileLength(File file, boolean isZipFile) throws
IOException {
- try {
- return isZipFile ? ServerUtils.zipFileSize(file) : file.length();
- } catch (FileNotFoundException e) {
- numFileOpenExceptions.mark();
- throw e;
- } catch (IOException e) {
- numFileReadExceptions.mark();
- throw e;
- }
- }
-
private DomContent logTemplate(List<DomContent> bodyContents, String
fileName, String user) {
List<DomContent> finalBodyContents = new ArrayList<>();
@@ -485,9 +502,28 @@ public class LogviewerLogPageHandler {
return a(text).withHref(url).withClass("btn btn-default " + (enabled ?
"enabled" : "disabled"));
}
- private String pageFile(String path, boolean isZipFile, long fileLength,
Integer start, Integer readLength)
+ private SeekableByteChannel openFile(Path file, boolean isDaemon) throws
IOException {
+ try {
+ //A daemon log may be a symbolic link, a worker log is opened only
if it may be served
+ return isDaemon ? LogFilePaths.openForReading(file, true) :
LogFilePaths.openWorkerFile(logRoot, file);
+ } catch (IOException e) {
+ numFileOpenExceptions.mark();
+ throw e;
+ }
+ }
+
+ private long getOpenedFileLength(SeekableByteChannel channel, boolean
isZipFile) throws IOException {
+ try {
+ return LogFilePaths.fileLength(channel, isZipFile);
+ } catch (IOException e) {
+ numFileReadExceptions.mark();
+ throw e;
+ }
+ }
+
+ private String pageFile(InputStream fileInput, boolean isZipFile, long
fileLength, Integer start, Integer readLength)
throws IOException, InvalidRequestException {
- try (InputStream input = isZipFile ? new GZIPInputStream(new
FileInputStream(path)) : new FileInputStream(path);
+ try (InputStream input = isZipFile ? new GZIPInputStream(fileInput) :
fileInput;
ByteArrayOutputStream output = new ByteArrayOutputStream()) {
if (start >= fileLength) {
throw new InvalidRequestException("Cannot start past the end
of the file");
@@ -508,9 +544,6 @@ public class LogviewerLogPageHandler {
numPageRead.mark();
return output.toString();
- } catch (FileNotFoundException e) {
- numFileOpenExceptions.mark();
- throw e;
} catch (IOException e) {
numFileReadExceptions.mark();
throw e;
diff --git
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogSearchHandler.java
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogSearchHandler.java
index 99da13fb1..11c847eb4 100644
---
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogSearchHandler.java
+++
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogSearchHandler.java
@@ -40,6 +40,8 @@ import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.SeekableByteChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -66,6 +68,7 @@ import org.apache.storm.daemon.common.JsonResponseBuilder;
import org.apache.storm.daemon.logviewer.LogviewerConstant;
import org.apache.storm.daemon.logviewer.utils.DirectoryCleaner;
import org.apache.storm.daemon.logviewer.utils.ExceptionMeterNames;
+import org.apache.storm.daemon.logviewer.utils.LogFilePaths;
import org.apache.storm.daemon.logviewer.utils.LogviewerResponseBuilder;
import org.apache.storm.daemon.logviewer.utils.ResourceAuthorizer;
import org.apache.storm.daemon.logviewer.utils.WorkerLogs;
@@ -75,7 +78,6 @@ import org.apache.storm.daemon.utils.StreamUtil;
import org.apache.storm.daemon.utils.UrlBuilder;
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.utils.ObjectReader;
-import org.apache.storm.utils.ServerUtils;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
@@ -164,7 +166,10 @@ public class LogviewerLogSearchHandler {
return searchLogFileNotFound(callback);
}
Response response;
- if (absFile.toFile().exists()) {
+ //A worker log must be a direct child of a port directory and must not
be reached through a symbolic link,
+ //otherwise it is answered like a missing file
+ boolean exists = isDaemon ? absFile.toFile().exists() :
LogFilePaths.isServableWorkerFile(rootDir, absFile);
+ if (exists) {
if (isDaemon ?
resourceAuthorizer.isUserAllowedToAccessDaemonFile(user)
: resourceAuthorizer.isUserAllowedToAccessFile(user,
fileName)) {
Integer numMatchesInt = numMatchesStr != null ?
tryParseIntParam("num-matches", numMatchesStr) : null;
@@ -175,10 +180,15 @@ public class LogviewerLogSearchHandler {
Map<String, Object> entity = new HashMap<>();
entity.put("isDaemon", isDaemon ? "yes" : "no");
Map<String, Object> res = substringSearch(absFile,
search, isDaemon, numMatchesInt, offsetInt);
- entity.putAll(res);
- noResult = ((List) res.get("matches")).isEmpty();
-
- response =
LogviewerResponseBuilder.buildSuccessJsonResponse(entity, callback, origin);
+ if (res == null) {
+ //The log file was removed or replaced by a
symbolic link after the check
+ response = searchLogFileNotFound(callback);
+ } else {
+ entity.putAll(res);
+ noResult = ((List) res.get("matches")).isEmpty();
+
+ response =
LogviewerResponseBuilder.buildSuccessJsonResponse(entity, callback, origin);
+ }
} else {
throw new InvalidRequestException("Search substring
must be between 1 and 1024 "
+ "UTF-8 bytes in size (inclusive)");
@@ -242,8 +252,9 @@ public class LogviewerLogSearchHandler {
if (StringUtils.isEmpty(portStr) || portStr.equals("*")) {
try (Stream<Path> topoDir = Files.list(absTopoDir)) {
- // check for all ports
+ // check for all ports, a port directory reached through a
symbolic link is left out
Stream<List<Path>> portsOfLogs = topoDir
+ .filter(portDir ->
LogFilePaths.isWorkerDirWithoutSymbolicLink(rootDir, portDir))
.map(portDir -> logsForPort(user, portDir))
.filter(logs -> logs != null && !logs.isEmpty());
@@ -270,8 +281,8 @@ public class LogviewerLogSearchHandler {
} else {
Path absPortDir =
absTopoDir.resolve(Integer.toString(port)).toAbsolutePath().normalize();
- if (!absPortDir.toFile().exists()
- || !absPortDir.startsWith(absTopoDir)) {
+ if (!absPortDir.startsWith(absTopoDir)
+ ||
!LogFilePaths.isWorkerDirWithoutSymbolicLink(rootDir, absPortDir)) {
returnValue = new ArrayList<>();
} else {
List<Path> filteredLogs = logsForPort(user,
absPortDir);
@@ -320,6 +331,10 @@ public class LogviewerLogSearchHandler {
return substringSearch(file, searchString, false, numMatches,
startByteOffset);
}
+ /**
+ * Search the file for the search string. Returns null if the file has
been removed, or a worker log has been replaced by a
+ * symbolic link, since it was checked.
+ */
private Map<String, Object> substringSearch(Path file, String
searchString, boolean isDaemon, Integer numMatches,
Integer startByteOffset) throws InvalidRequestException {
if (StringUtils.isEmpty(searchString)) {
@@ -331,13 +346,25 @@ public class LogviewerLogSearchHandler {
}
boolean isZipFile = file.toString().endsWith(".gz");
- try (InputStream fis = Files.newInputStream(file)) {
- try (InputStream gzippedInputStream = isZipFile ? new
GZIPInputStream(fis) : fis;
+ //A worker log is opened without following a symbolic link, its length
is taken from the opened file. The callers check that
+ //a worker log may be served: searchLogFile checks the file, deep
search only lists port directories that may be served
+ try (SeekableByteChannel channel = LogFilePaths.openForReading(file,
isDaemon)) {
+ if (channel == null) {
+ //The log file was removed or replaced by a symbolic link
after the check, it is treated like a missing file
+ return null;
+ }
+ //It's more likely to be a file read exception here, so we don't
differentiate
+ int fileLength;
+ try {
+ fileLength = (int) LogFilePaths.fileLength(channel, isZipFile);
+ } catch (IOException e) {
+ numFileReadExceptions.mark();
+ throw new RuntimeException(e);
+ }
+ try (InputStream fis = Channels.newInputStream(channel);
+ InputStream gzippedInputStream = isZipFile ? new
GZIPInputStream(fis) : fis;
BufferedInputStream stream = new
BufferedInputStream(gzippedInputStream)) {
- //It's more likely to be a file read exception here, so we
don't differentiate
- int fileLength = isZipFile ? (int)
ServerUtils.zipFileSize(file.toFile()) : (int) Files.size(file);
-
ByteBuffer buf = ByteBuffer.allocate(GREP_BUF_SIZE);
final byte[] bufArray = buf.array();
final byte[] searchBytes =
searchString.getBytes(StandardCharsets.UTF_8);
@@ -428,6 +455,7 @@ public class LogviewerLogSearchHandler {
List<Path> logsForPort(String user, Path portDir) {
try {
List<Path> workerLogs =
directoryCleaner.getFilesForDir(portDir).stream()
+ .filter(LogFilePaths::isRegularFileNotLink)
.filter(file ->
WORKER_LOG_FILENAME_PATTERN.asPredicate().test(file.getFileName().toString()))
.collect(toList());
@@ -478,7 +506,12 @@ public class LogviewerLogSearchHandler {
try {
LOG.debug("Looking through {}", firstLog);
matchInLog = substringSearch(firstLog, targetStr, numMatches -
matchCount, startByteOffset);
- scannedFiles++;
+ if (matchInLog == null) {
+ //The worker log was removed or replaced by a symbolic
link after it was listed, it is not scanned
+ matchInLog = new HashMap<>();
+ } else {
+ scannedFiles++;
+ }
} catch (InvalidRequestException e) {
LOG.error("Can't search past end of file.", e);
matchInLog = new HashMap<>();
@@ -790,6 +823,11 @@ public class LogviewerLogSearchHandler {
return matches;
}
+ @VisibleForTesting
+ int getOpenedFiles() {
+ return openedFiles;
+ }
+
@Override
public String toJSONString() {
try {
diff --git
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerProfileHandler.java
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerProfileHandler.java
index 566d5dcad..0f9e7c5d7 100644
---
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerProfileHandler.java
+++
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerProfileHandler.java
@@ -35,13 +35,17 @@ import jakarta.ws.rs.core.Response;
import java.io.File;
import java.io.IOException;
+import java.nio.channels.Channels;
+import java.nio.channels.SeekableByteChannel;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
+import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
import org.apache.storm.daemon.logviewer.utils.DirectoryCleaner;
import org.apache.storm.daemon.logviewer.utils.ExceptionMeterNames;
+import org.apache.storm.daemon.logviewer.utils.LogFilePaths;
import org.apache.storm.daemon.logviewer.utils.LogviewerResponseBuilder;
import org.apache.storm.daemon.logviewer.utils.ResourceAuthorizer;
import org.apache.storm.metric.StormMetricsRegistry;
@@ -50,6 +54,8 @@ public class LogviewerProfileHandler {
public static final String WORKER_LOG_FILENAME = "worker.log";
+ private static final Pattern PORT_PATTERN = Pattern.compile("[0-9]+");
+
private final Meter numFileDownloadExceptions;
private final Path logRoot;
@@ -79,7 +85,10 @@ public class LogviewerProfileHandler {
* @return The HTML page representing list page of dump files
*/
public Response listDumpFiles(String topologyId, String hostPort, String
user) throws IOException {
- String portStr = hostPort.split(":")[1];
+ String portStr = getPort(hostPort);
+ if (portStr == null) {
+ return LogviewerResponseBuilder.buildResponsePageNotFound();
+ }
Path rawDir = logRoot.resolve(topologyId).resolve(portStr);
Path absDir = rawDir.toAbsolutePath().normalize();
if (!absDir.startsWith(logRoot) ||
!rawDir.normalize().toString().equals(rawDir.toString())) {
@@ -87,7 +96,8 @@ public class LogviewerProfileHandler {
return LogviewerResponseBuilder.buildResponsePageNotFound();
}
- if (absDir.toFile().exists()) {
+ //A port directory must not be reached through a symbolic link,
otherwise it is answered like a missing directory
+ if (LogFilePaths.isWorkerDirWithoutSymbolicLink(logRoot, absDir)) {
String workerFileRelativePath = String.join(File.separator,
topologyId, portStr, WORKER_LOG_FILENAME);
if (resourceAuthorizer.isUserAllowedToAccessFile(user,
workerFileRelativePath)) {
String content = buildDumpFileListPage(topologyId, hostPort,
absDir.toFile());
@@ -110,9 +120,11 @@ public class LogviewerProfileHandler {
* @return a Response which lets browsers download that file.
*/
public Response downloadDumpFile(String topologyId, String hostPort,
String fileName, String user) throws IOException {
- String[] hostPortSplit = hostPort.split(":");
- String host = hostPortSplit[0];
- String portStr = hostPortSplit[1];
+ String portStr = getPort(hostPort);
+ if (portStr == null) {
+ return LogviewerResponseBuilder.buildResponsePageNotFound();
+ }
+ String host = hostPort.substring(0, hostPort.lastIndexOf(':'));
Path rawFile =
logRoot.resolve(topologyId).resolve(portStr).resolve(fileName);
Path absFile = rawFile.toAbsolutePath().normalize();
if (!absFile.startsWith(logRoot) ||
!rawFile.normalize().toString().equals(rawFile.toString())) {
@@ -120,11 +132,29 @@ public class LogviewerProfileHandler {
return LogviewerResponseBuilder.buildResponsePageNotFound();
}
- if (absFile.toFile().exists()) {
+ if (Paths.get(fileName).getNameCount() != 1) {
+ //A dump file is a direct child of the port directory
+ return LogviewerResponseBuilder.buildResponsePageNotFound();
+ }
+
+ //A dump file must be a direct child of a port directory and must not
be reached through a symbolic link,
+ //otherwise it is answered like a missing file
+ if (LogFilePaths.isServableWorkerFile(logRoot, absFile)) {
String workerFileRelativePath = String.join(File.separator,
topologyId, portStr, WORKER_LOG_FILENAME);
if (resourceAuthorizer.isUserAllowedToAccessFile(user,
workerFileRelativePath)) {
String downloadedFileName = host + "-" + topologyId + "-" +
portStr + "-" + absFile.getFileName();
- return
LogviewerResponseBuilder.buildDownloadFile(downloadedFileName,
absFile.toFile(), numFileDownloadExceptions);
+ SeekableByteChannel channel;
+ try {
+ channel = LogFilePaths.openWorkerFile(logRoot, absFile);
+ } catch (IOException e) {
+ numFileDownloadExceptions.mark();
+ throw e;
+ }
+ if (channel == null) {
+ //The dump file was removed or replaced by a symbolic link
after the check
+ return
LogviewerResponseBuilder.buildResponsePageNotFound();
+ }
+ return
LogviewerResponseBuilder.buildDownloadFile(downloadedFileName,
Channels.newInputStream(channel));
} else {
return
LogviewerResponseBuilder.buildResponseUnauthorizedUser(user);
}
@@ -133,6 +163,18 @@ public class LogviewerProfileHandler {
}
}
+ /**
+ * Get the port part of "host:port", the host may contain ':' itself (e.g.
"[::1]:6700"). Returns null if the port is missing or
+ * not a number, so it can not name anything other than a port directory.
+ */
+ private static String getPort(String hostPort) {
+ String portStr = hostPort.substring(hostPort.lastIndexOf(':') + 1);
+ if (hostPort.indexOf(':') < 0 ||
!PORT_PATTERN.matcher(portStr).matches()) {
+ return null;
+ }
+ return portStr;
+ }
+
private String buildDumpFileListPage(String topologyId, String hostPort,
File dir) throws IOException {
List<DomContent> liTags = getProfilerDumpFiles(dir).stream()
.map(file -> li(a(file).withHref("/api/v1/dumps/" + topologyId +
"/" + hostPort + "/" + file)))
@@ -154,6 +196,7 @@ public class LogviewerProfileHandler {
private List<String> getProfilerDumpFiles(File dir) throws IOException {
List<Path> filesForDir = directoryCleaner.getFilesForDir(dir.toPath());
return filesForDir.stream()
+ .filter(LogFilePaths::isRegularFileNotLink)
.map(path -> path.toFile())
.filter(file -> {
String fileName = file.getName();
diff --git
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogFileDownloader.java
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogFileDownloader.java
index 04e08bfc4..4b5a67bf3 100644
---
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogFileDownloader.java
+++
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogFileDownloader.java
@@ -24,6 +24,8 @@ import com.codahale.metrics.Meter;
import jakarta.ws.rs.core.Response;
import java.io.IOException;
+import java.nio.channels.Channels;
+import java.nio.channels.SeekableByteChannel;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -79,27 +81,50 @@ public class LogFileDownloader {
//Prevent daemon log reads from pathing into worker logs
return LogviewerResponseBuilder.buildResponsePageNotFound();
}
-
- if (file.toFile().exists()) {
+
+ //A worker log must be a direct child of a port directory and must not
be reached through a symbolic link,
+ //otherwise it is answered like a missing file
+ boolean exists = isDaemon ? file.toFile().exists() :
LogFilePaths.isServableWorkerFile(rootDir, file);
+ if (exists) {
if (isDaemon ?
resourceAuthorizer.isUserAllowedToAccessDaemonFile(user)
: resourceAuthorizer.isUserAllowedToAccessFile(user,
fileName)) {
if (!isDaemon) {
//Only widen the permission of a worker log once the
request is known to be served
workerLogs.setLogFilePermission(fileName);
}
- fileDownloadSizeDistMb.update(Math.round((double)
file.toFile().length() / FileUtils.ONE_MB));
String downloadedFileName;
- Path pathRelativeToRootDir = rootDir.relativize(file);
- if (isDaemon || pathRelativeToRootDir.getNameCount() != 3) {
+ if (isDaemon) {
downloadedFileName = host + "-" + rawFile.getFileName();
} else {
//host-topoId-port-fileName
+ Path pathRelativeToRootDir = rootDir.relativize(file);
downloadedFileName = host + "-"
+ pathRelativeToRootDir.getName(0) + "-"
+ pathRelativeToRootDir.getName(1) + "-"
+ pathRelativeToRootDir.getName(2);
}
- return
LogviewerResponseBuilder.buildDownloadFile(downloadedFileName, file.toFile(),
numFileDownloadExceptions);
+ //A daemon log may be a symbolic link, a worker log is opened
without following one
+ SeekableByteChannel channel;
+ try {
+ channel = isDaemon ? LogFilePaths.openForReading(file,
true) : LogFilePaths.openWorkerFile(rootDir, file);
+ } catch (IOException e) {
+ numFileDownloadExceptions.mark();
+ throw e;
+ }
+ if (channel == null) {
+ //The log file was removed or replaced by a symbolic link
after the check
+ return
LogviewerResponseBuilder.buildResponsePageNotFound();
+ }
+ long size;
+ try {
+ size = channel.size();
+ } catch (IOException e) {
+ channel.close();
+ numFileDownloadExceptions.mark();
+ throw e;
+ }
+ fileDownloadSizeDistMb.update(Math.round((double) size /
FileUtils.ONE_MB));
+ return
LogviewerResponseBuilder.buildDownloadFile(downloadedFileName,
Channels.newInputStream(channel));
} else {
return
LogviewerResponseBuilder.buildResponseUnauthorizedUser(user);
}
diff --git
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogFilePaths.java
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogFilePaths.java
new file mode 100644
index 000000000..b6ee14f02
--- /dev/null
+++
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogFilePaths.java
@@ -0,0 +1,210 @@
+/*
+ * 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.storm.daemon.logviewer.utils;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.LinkOption;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+
+/**
+ * Helpers to serve only files that resolve inside the worker log root.
+ *
+ * <p>Files below a worker's port directory are created by the worker itself,
so a symbolic link placed there could point to the
+ * logs of another topology or to any file outside the log root. These helpers
make sure such a link is treated like a missing file.
+ * The log root itself may still be a symbolic link (or lie below one).
+ *
+ * <p>Worker files are limited to direct children of a port directory
("topologyId/port/fileName"). The topology and port
+ * directories are created by the supervisor, and only the port directory is
handed to the topology owner (see
+ * {@code Container.setupWorkerArtifactsDir}); the topology directory stays
owned by the supervisor user. So only the last path
+ * element can change between the check and the open, and the open does not
follow a symbolic link at the last path element.
+ *
+ * <p>Hard links can not be told apart from regular files by resolving the
real path. Creating a hard link to a file of another
+ * user is expected to be prevented by the operating system (e.g. {@code
fs.protected_hardlinks=1} on Linux).
+ *
+ * <p>The worker may also replace a checked file by a named pipe before the
open. Opening a named pipe blocks until a writer
+ * appears, and Java offers no non-blocking open to detect this, so the
request thread waits until the worker writes to or
+ * removes the pipe. This lets a topology owner tie up Logviewer request
threads, a denial of service by a trusted submitter that
+ * is not guarded against here.
+ */
+public final class LogFilePaths {
+
+ private static final int GZIP_TRAILER_SIZE_BYTES = 4;
+
+ private LogFilePaths() {
+ }
+
+ /**
+ * Check that a file is a worker file directly below a port directory,
i.e. "topologyId/port/fileName" below the log root.
+ *
+ * @param logRoot the absolute, normalized worker log root
+ * @param file the absolute, normalized file path
+ * @return true if the file has exactly three path elements below the log
root
+ */
+ public static boolean isWorkerFilePath(Path logRoot, Path file) {
+ return file.startsWith(logRoot) &&
logRoot.relativize(file).getNameCount() == 3;
+ }
+
+ /**
+ * Check that a file is a worker file ("topologyId/port/fileName") that
may be served: a regular file reached without
+ * passing through any symbolic link below the log root.
+ *
+ * @param logRoot the absolute, normalized worker log root; it may itself
be or lie below a symbolic link
+ * @param file the absolute, normalized file path below logRoot
+ * @return true if the file may be served, false otherwise
+ */
+ public static boolean isServableWorkerFile(Path logRoot, Path file) {
+ return isWorkerFilePath(logRoot, file) &&
isRegularFileWithoutSymbolicLink(logRoot, file);
+ }
+
+ /**
+ * Check that a directory is a worker port directory ("topologyId/port"
below the log root) and that neither the port
+ * directory nor the topology directory is a symbolic link.
+ *
+ * @param logRoot the absolute, normalized worker log root; it may itself
be or lie below a symbolic link
+ * @param portDir the absolute, normalized directory path
+ * @return true if the directory is a port directory reached without
passing through a symbolic link below the log root
+ */
+ public static boolean isWorkerDirWithoutSymbolicLink(Path logRoot, Path
portDir) {
+ return portDir.startsWith(logRoot) &&
logRoot.relativize(portDir).getNameCount() == 2
+ && isDirectoryNotLink(portDir.getParent()) &&
isDirectoryNotLink(portDir);
+ }
+
+ /**
+ * Check that a file below the given root directory is a regular file and
is reached without passing through any symbolic
+ * link below the root directory.
+ *
+ * @param rootDir the absolute, normalized root directory; it may itself
be or lie below a symbolic link
+ * @param file the absolute, normalized file path below rootDir
+ * @return true if the file is a regular file and resolves to exactly the
same location below the real root directory,
+ * false otherwise
+ */
+ public static boolean isRegularFileWithoutSymbolicLink(Path rootDir, Path
file) {
+ if (!file.startsWith(rootDir) || !isRegularFileNotLink(file)) {
+ return false;
+ }
+ try {
+ Path expected =
rootDir.toRealPath().resolve(rootDir.relativize(file));
+ return file.toRealPath().equals(expected);
+ } catch (IOException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Check whether the path is a regular file which is not a symbolic link.
+ *
+ * @param path the path to check
+ * @return true if the path is a regular file and not a symbolic link
+ */
+ public static boolean isRegularFileNotLink(Path path) {
+ return Files.isRegularFile(path, LinkOption.NOFOLLOW_LINKS);
+ }
+
+ /**
+ * Check whether the path is a directory which is not a symbolic link.
+ *
+ * @param path the path to check
+ * @return true if the path is a directory and not a symbolic link
+ */
+ public static boolean isDirectoryNotLink(Path path) {
+ return Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS);
+ }
+
+ /**
+ * Open a file for reading.
+ *
+ * <p>A file may be removed (e.g. by the log cleaner or by log rotation)
after it was checked. When symbolic links are not
+ * followed, the file may also have been replaced by a symbolic link since
it was checked, which makes the open fail. Both
+ * cases are reported by returning null, so callers can answer them like a
missing file instead of a read error.
+ *
+ * @param file the file to open
+ * @param followLinks whether a symbolic link at the last path element is
followed; only the last path element is affected
+ * @return a channel for reading the file, or null if the file does not
exist or if followLinks is false and the file is a
+ * symbolic link
+ * @throws IOException if the file can not be opened for any other reason
+ */
+ public static SeekableByteChannel openForReading(Path file, boolean
followLinks) throws IOException {
+ try {
+ if (followLinks) {
+ return Files.newByteChannel(file, StandardOpenOption.READ);
+ }
+ return Files.newByteChannel(file, StandardOpenOption.READ,
LinkOption.NOFOLLOW_LINKS);
+ } catch (NoSuchFileException e) {
+ return null;
+ } catch (IOException e) {
+ if (!followLinks && Files.isSymbolicLink(file)) {
+ return null;
+ }
+ throw e;
+ }
+ }
+
+ /**
+ * Open a worker file ("topologyId/port/fileName") for reading, if it may
be served. This is the only way a worker file should
+ * be opened, so the checks of {@link #isServableWorkerFile(Path, Path)}
can not be left out.
+ *
+ * @param logRoot the absolute, normalized worker log root; it may itself
be or lie below a symbolic link
+ * @param file the absolute, normalized file path below logRoot
+ * @return a channel for reading the file, or null if the file may not be
served, does not exist (any more) or has been
+ * replaced by a symbolic link since it was checked
+ * @throws IOException if the file can not be opened for any other reason
+ */
+ public static SeekableByteChannel openWorkerFile(Path logRoot, Path file)
throws IOException {
+ if (!isServableWorkerFile(logRoot, file)) {
+ return null;
+ }
+ return openForReading(file, false);
+ }
+
+ /**
+ * Get the size of the opened file, or the uncompressed size stored in its
gzip trailer. The channel is positioned at the start
+ * of the file afterwards.
+ *
+ * @param channel the opened file
+ * @param isZipFile whether the file is gzip compressed
+ * @return the (uncompressed) size of the file
+ * @throws IOException if the size can not be read
+ */
+ public static long fileLength(SeekableByteChannel channel, boolean
isZipFile) throws IOException {
+ long size = channel.size();
+ if (!isZipFile) {
+ return size;
+ }
+ if (size < GZIP_TRAILER_SIZE_BYTES) {
+ throw new EOFException("File is too short to be gzip compressed");
+ }
+ ByteBuffer trailer =
ByteBuffer.allocate(GZIP_TRAILER_SIZE_BYTES).order(ByteOrder.LITTLE_ENDIAN);
+ channel.position(size - GZIP_TRAILER_SIZE_BYTES);
+ while (trailer.hasRemaining()) {
+ if (channel.read(trailer) < 0) {
+ throw new EOFException("Unexpected end of gzip trailer");
+ }
+ }
+ channel.position(0);
+ trailer.flip();
+ return Integer.toUnsignedLong(trailer.getInt());
+ }
+}
diff --git
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogviewerResponseBuilder.java
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogviewerResponseBuilder.java
index d1ddabd8e..21f3f5a40 100644
---
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogviewerResponseBuilder.java
+++
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogviewerResponseBuilder.java
@@ -22,7 +22,6 @@ import static j2html.TagCreator.body;
import static j2html.TagCreator.h2;
import static org.apache.commons.text.StringEscapeUtils.escapeHtml4;
-import com.codahale.metrics.Meter;
import com.google.common.io.ByteStreams;
import jakarta.ws.rs.core.MediaType;
@@ -30,11 +29,8 @@ import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.StreamingOutput;
import java.io.BufferedOutputStream;
-import java.io.File;
-import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.nio.file.Files;
import org.apache.storm.daemon.common.JsonResponseBuilder;
import org.apache.storm.daemon.ui.UIHelpers;
@@ -69,25 +65,17 @@ public class LogviewerResponseBuilder {
}
/**
- * Build a Response object representing download a file.
+ * Build a Response object representing download of an already opened file.
*
* @param contentDispositionName The name to set in the
Content-Disposition header
- * @param file file to download
+ * @param is the opened file; it is read and then closed when the response
is written
*/
- public static Response buildDownloadFile(String contentDispositionName,
- File file, Meter numFileDownloadExceptions) throws IOException {
- try {
- // do not close this InputStream in method: it will be used from
jetty server
- InputStream is = Files.newInputStream(file.toPath());
- return Response.status(jakarta.ws.rs.core.Response.Status.OK)
- .entity(wrapWithStreamingOutput(is))
- .type(MediaType.APPLICATION_OCTET_STREAM_TYPE)
- .header("Content-Disposition", "attachment; filename=\"" +
contentDispositionName + "\"")
- .build();
- } catch (IOException e) {
- numFileDownloadExceptions.mark();
- throw e;
- }
+ public static Response buildDownloadFile(String contentDispositionName,
InputStream is) {
+ return Response.status(jakarta.ws.rs.core.Response.Status.OK)
+ .entity(wrapWithStreamingOutput(is))
+ .type(MediaType.APPLICATION_OCTET_STREAM_TYPE)
+ .header("Content-Disposition", "attachment; filename=\"" +
contentDispositionName + "\"")
+ .build();
}
/**
@@ -142,15 +130,18 @@ public class LogviewerResponseBuilder {
}
private static StreamingOutput wrapWithStreamingOutput(final InputStream
inputStream) {
+ //The input is closed once it has been written (or the write failed),
Jersey does not close it for a StreamingOutput
return os -> {
- OutputStream wrappedOutputStream = os;
- if (!(os instanceof BufferedOutputStream)) {
- wrappedOutputStream = new BufferedOutputStream(os);
- }
+ try (InputStream is = inputStream) {
+ OutputStream wrappedOutputStream = os;
+ if (!(os instanceof BufferedOutputStream)) {
+ wrappedOutputStream = new BufferedOutputStream(os);
+ }
- ByteStreams.copy(inputStream, wrappedOutputStream);
+ ByteStreams.copy(is, wrappedOutputStream);
- wrappedOutputStream.flush();
+ wrappedOutputStream.flush();
+ }
};
}
diff --git
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/WorkerLogs.java
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/WorkerLogs.java
index 7a9af3647..fb288a37a 100644
---
a/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/WorkerLogs.java
+++
b/storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/WorkerLogs.java
@@ -116,14 +116,22 @@ public class WorkerLogs {
}
/**
- * Return a list of all log files from worker directories in root log
directory.
+ * Return a list of all log files from worker directories in root log
directory. Symbolic links are left out, both as worker
+ * directories and as files. Only links are left out, the listing is
otherwise unchanged.
*/
public List<Path> getAllLogsForRootDir() throws IOException {
List<Path> files = new ArrayList<>();
Set<Path> topoDirFiles = getAllWorkerDirs();
if (topoDirFiles != null) {
for (Path portDir : topoDirFiles) {
- files.addAll(directoryCleaner.getFilesForDir(portDir));
+ if (!LogFilePaths.isWorkerDirWithoutSymbolicLink(logRootDir,
portDir)) {
+ continue;
+ }
+ for (Path file : directoryCleaner.getFilesForDir(portDir)) {
+ if (!Files.isSymbolicLink(file)) {
+ files.add(file);
+ }
+ }
}
}
diff --git
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogDownloadHandlerTest.java
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogDownloadHandlerTest.java
index 0d40fce8d..c17cd1af0 100644
---
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogDownloadHandlerTest.java
+++
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogDownloadHandlerTest.java
@@ -25,13 +25,16 @@ import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.startsWith;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.google.common.net.HttpHeaders;
import java.io.ByteArrayOutputStream;
+import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
@@ -39,12 +42,16 @@ import java.nio.file.Path;
import java.util.Map;
import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.StreamingOutput;
+import org.apache.storm.daemon.logviewer.testsupport.SymbolicLinks;
+import org.apache.storm.daemon.logviewer.utils.ExceptionMeterNames;
+import org.apache.storm.daemon.logviewer.utils.LogFilePaths;
import org.apache.storm.daemon.logviewer.utils.ResourceAuthorizer;
import org.apache.storm.daemon.logviewer.utils.WorkerLogs;
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.testing.TmpPath;
import org.apache.storm.utils.Utils;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
public class LogviewerLogDownloadHandlerTest {
@@ -242,6 +249,260 @@ public class LogviewerLogDownloadHandlerTest {
}
}
+ @Test
+ public void
testDownloadLogFileDoesNotFollowSymbolicLinkToAnotherTopology() throws
IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("logs"));
+ Path link =
workerLogRoot.resolve("topoA").resolve("1111").resolve("other.log");
+ SymbolicLinks.create(link,
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log"));
+
+ WorkerLogs workerLogs = mock(WorkerLogs.class);
+ LogviewerLogDownloadHandler handler =
createHandlerAllowingOnlyTopoA(rootPath.getFile().toPath().resolve("logs"),
workerLogs);
+
+ Response response = handler.downloadLogFile("host",
"topoA/1111/other.log", "user");
+ int status = response.getStatus();
+ String content = status == Response.Status.OK.getStatusCode() ?
readEntity(response) : null;
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat("served content: " + content, status,
is(Response.Status.NOT_FOUND.getStatusCode()));
+ verify(workerLogs, never()).setLogFilePermission(anyString());
+ }
+ }
+
+ @Test
+ public void testDownloadLogFileDoesNotFollowSymbolicLinkOutsideLogRoot()
throws IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("logs"));
+ Path outside = rootPath.getFile().toPath().resolve("outside.txt");
+ Files.writeString(outside, "content outside the log root");
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("outside.log"),
outside);
+
+ WorkerLogs workerLogs = mock(WorkerLogs.class);
+ LogviewerLogDownloadHandler handler =
createHandlerAllowingOnlyTopoA(rootPath.getFile().toPath().resolve("logs"),
workerLogs);
+
+ Response response = handler.downloadLogFile("host",
"topoA/1111/outside.log", "user");
+ int status = response.getStatus();
+ String content = status == Response.Status.OK.getStatusCode() ?
readEntity(response) : null;
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat("served content: " + content, status,
is(Response.Status.NOT_FOUND.getStatusCode()));
+ verify(workerLogs, never()).setLogFilePermission(anyString());
+ }
+ }
+
+ @Test
+ public void testDownloadLogFileUnderSymbolicLinkRoot() throws IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path realDaemonLogRoot =
rootPath.getFile().toPath().resolve("real-logs");
+ createWorkerLogTree(realDaemonLogRoot);
+ Path linkedDaemonLogRoot =
rootPath.getFile().toPath().resolve("logs");
+ SymbolicLinks.create(linkedDaemonLogRoot, realDaemonLogRoot);
+
+ WorkerLogs workerLogs = mock(WorkerLogs.class);
+ LogviewerLogDownloadHandler handler =
createHandlerAllowingOnlyTopoA(linkedDaemonLogRoot, workerLogs);
+
+ Response response = handler.downloadLogFile("host",
"topoA/1111/worker.log", "user");
+ int status = response.getStatus();
+ String content = status == Response.Status.OK.getStatusCode() ?
readEntity(response) : null;
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(status, is(Response.Status.OK.getStatusCode()));
+ assertThat(content, is("topoA content"));
+ verify(workerLogs).setLogFilePermission("topoA/1111/worker.log");
+ }
+ }
+
+ @Test
+ public void testDownloadLogFileRejectsFileBelowSubdirectoryOfPortDir()
throws IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("logs"));
+ Path nested =
workerLogRoot.resolve("topoA").resolve("1111").resolve("sub").resolve("worker.log");
+ Files.createDirectories(nested.getParent());
+ Files.writeString(nested, "nested content");
+
+ WorkerLogs workerLogs = mock(WorkerLogs.class);
+ LogviewerLogDownloadHandler handler =
createHandlerAllowingOnlyTopoA(rootPath.getFile().toPath().resolve("logs"),
workerLogs);
+
+ Response response = handler.downloadLogFile("host",
"topoA/1111/sub/worker.log", "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(response.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ verify(workerLogs, never()).setLogFilePermission(anyString());
+ }
+ }
+
+ @Test
+ public void testDownloadDaemonLogFileFollowsSymbolicLink() throws
IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ createWorkerLogTree(daemonLogRoot);
+ Path realLog =
Files.writeString(daemonLogRoot.resolve("nimbus.log.real"), "nimbus content");
+ SymbolicLinks.create(daemonLogRoot.resolve("nimbus.log"), realLog);
+
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+
when(resourceAuthorizer.isUserAllowedToAccessDaemonFile(anyString())).thenReturn(true);
+ LogviewerLogDownloadHandler handler = new
LogviewerLogDownloadHandler(daemonLogRoot.resolve("workers-artifacts").toString(),
+ daemonLogRoot.toString(), mock(WorkerLogs.class),
resourceAuthorizer, new StormMetricsRegistry());
+
+ Response response = handler.downloadDaemonLogFile("host",
"nimbus.log", "user");
+ int status = response.getStatus();
+ String content = status == Response.Status.OK.getStatusCode() ?
readEntity(response) : null;
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(status, is(Response.Status.OK.getStatusCode()));
+ assertThat(content, is("nimbus content"));
+ }
+ }
+
+ @Test
+ public void
testDownloadLogFileReplacedBySymbolicLinkAfterCheckIsNotFound() throws
IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ Path workerLogRoot = createWorkerLogTree(daemonLogRoot);
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("other.log"),
+
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log"));
+
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerLogDownloadHandler handler = new
LogviewerLogDownloadHandler(workerLogRoot.toString(),
+ daemonLogRoot.toString(), mock(WorkerLogs.class),
resourceAuthorizer, metricsRegistry);
+
+ Response response;
+ try (MockedStatic<LogFilePaths> ignored = passWorkerFileCheck()) {
+ response = handler.downloadLogFile("host",
"topoA/1111/other.log", "user");
+ }
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(response.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+
assertThat(metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_DOWNLOAD_EXCEPTIONS).getCount(),
is(0L));
+ }
+ }
+
+ @Test
+ public void testDownloadLogFileRemovedAfterCheckIsNotFound() throws
IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ Path workerLogRoot = createWorkerLogTree(daemonLogRoot);
+
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerLogDownloadHandler handler = new
LogviewerLogDownloadHandler(workerLogRoot.toString(),
+ daemonLogRoot.toString(), mock(WorkerLogs.class),
resourceAuthorizer, metricsRegistry);
+
+ Response response;
+ try (MockedStatic<LogFilePaths> ignored = removeFileBeforeOpen()) {
+ response = handler.downloadLogFile("host",
"topoA/1111/worker.log", "user");
+ }
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(response.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+
assertThat(metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_DOWNLOAD_EXCEPTIONS).getCount(),
is(0L));
+
assertThat(metricsRegistry.registerHistogram("logviewer:download-file-size-rounded-MB").getCount(),
is(0L));
+ }
+ }
+
+ @Test
+ public void testDownloadDaemonLogFileRemovedAfterCheckIsNotFound() throws
IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ Path workerLogRoot = createWorkerLogTree(daemonLogRoot);
+ Files.writeString(daemonLogRoot.resolve("nimbus.log"), "nimbus
content");
+
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+
when(resourceAuthorizer.isUserAllowedToAccessDaemonFile(anyString())).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerLogDownloadHandler handler = new
LogviewerLogDownloadHandler(workerLogRoot.toString(),
+ daemonLogRoot.toString(), mock(WorkerLogs.class),
resourceAuthorizer, metricsRegistry);
+
+ Response response;
+ try (MockedStatic<LogFilePaths> ignored = removeFileBeforeOpen()) {
+ response = handler.downloadDaemonLogFile("host", "nimbus.log",
"user");
+ }
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(response.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+
assertThat(metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_DOWNLOAD_EXCEPTIONS).getCount(),
is(0L));
+ }
+ }
+
+ @Test
+ public void testDownloadLogFileRecordsSize() throws IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ Path workerLogRoot = createWorkerLogTree(daemonLogRoot);
+
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerLogDownloadHandler handler = new
LogviewerLogDownloadHandler(workerLogRoot.toString(),
+ daemonLogRoot.toString(), mock(WorkerLogs.class),
resourceAuthorizer, metricsRegistry);
+
+ Response response = handler.downloadLogFile("host",
"topoA/1111/worker.log", "user");
+ String content = readEntity(response);
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(response.getStatus(),
is(Response.Status.OK.getStatusCode()));
+ assertThat(content, is("topoA content"));
+
assertThat(metricsRegistry.registerHistogram("logviewer:download-file-size-rounded-MB").getCount(),
is(1L));
+ }
+ }
+
+ /**
+ * Lets every worker file pass the check, so the open sees a file that was
removed or replaced by a symbolic link after the
+ * check.
+ */
+ static MockedStatic<LogFilePaths> passWorkerFileCheck() {
+ return mockStatic(LogFilePaths.class, invocation ->
"isServableWorkerFile".equals(invocation.getMethod().getName())
+ ? Boolean.TRUE : invocation.callRealMethod());
+ }
+
+ /**
+ * Removes each file right before it is opened, as the log cleaner or log
rotation may do between the check and the open.
+ */
+ static MockedStatic<LogFilePaths> removeFileBeforeOpen() {
+ return mockStatic(LogFilePaths.class, invocation -> {
+ if ("openForReading".equals(invocation.getMethod().getName())) {
+ Files.deleteIfExists(invocation.getArgument(0));
+ }
+ return invocation.callRealMethod();
+ });
+ }
+
+ /**
+ * Creates topoA/1111/worker.log and topoB/1111/worker.log with distinct
content below the worker log root.
+ *
+ * @return the worker log root
+ */
+ private Path createWorkerLogTree(Path daemonLogRoot) throws IOException {
+ Path workerLogRoot = daemonLogRoot.resolve("workers-artifacts");
+ Path topoAFile =
workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log");
+ Path topoBFile =
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log");
+ Files.createDirectories(topoAFile.getParent());
+ Files.createDirectories(topoBFile.getParent());
+ Files.writeString(topoAFile, "topoA content");
+ Files.writeString(topoBFile, "topoB content");
+ return workerLogRoot;
+ }
+
+ private LogviewerLogDownloadHandler createHandlerAllowingOnlyTopoA(Path
daemonLogRoot, WorkerLogs workerLogs) {
+ //Models per topology access: the user may only read the logs of topoA.
+ ResourceAuthorizer resourceAuthorizer = mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ return new
LogviewerLogDownloadHandler(daemonLogRoot.resolve("workers-artifacts").toString(),
+ daemonLogRoot.toString(), workerLogs, resourceAuthorizer, new
StormMetricsRegistry());
+ }
+
private String readEntity(Response response) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
((StreamingOutput) response.getEntity()).write(out);
diff --git
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogPageHandlerTest.java
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogPageHandlerTest.java
index 0d277cf8a..d1aa2073b 100644
---
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogPageHandlerTest.java
+++
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogPageHandlerTest.java
@@ -20,12 +20,14 @@ package org.apache.storm.daemon.logviewer.handler;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.startsWith;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -34,14 +36,20 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.File;
import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
+import java.util.zip.GZIPOutputStream;
import jakarta.ws.rs.core.Response;
+import org.apache.storm.daemon.logviewer.testsupport.SymbolicLinks;
+import org.apache.storm.daemon.logviewer.utils.ExceptionMeterNames;
+import org.apache.storm.daemon.logviewer.utils.LogFilePaths;
import org.apache.storm.daemon.logviewer.utils.LogviewerResponseBuilder;
import org.apache.storm.daemon.logviewer.utils.ResourceAuthorizer;
import org.apache.storm.daemon.logviewer.utils.WorkerLogs;
@@ -49,6 +57,7 @@ import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.testing.TmpPath;
import org.apache.storm.utils.Utils;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
public class LogviewerLogPageHandlerTest {
@@ -250,6 +259,310 @@ public class LogviewerLogPageHandlerTest {
}
}
+ @Test
+ public void testDaemonLogPageWithoutStartShowsLastPage() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+
when(resourceAuthorizer.isUserAllowedToAccessDaemonFile(anyString())).thenReturn(true);
+ LogviewerLogPageHandler handler =
createHandlerForTraversalTests(rootPath.getFile().toPath(), resourceAuthorizer);
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ String content = "first part " + "x".repeat(200) + " last part";
+ Files.writeString(daemonLogRoot.resolve("nimbus.log"), content);
+ try (OutputStream out = new
GZIPOutputStream(Files.newOutputStream(daemonLogRoot.resolve("nimbus.log.gz"))))
{
+ out.write(content.getBytes(StandardCharsets.UTF_8));
+ }
+
+ final Response plain = handler.daemonLogPage("nimbus.log", null,
50, null, "user");
+ final Response compressed = handler.daemonLogPage("nimbus.log.gz",
null, 50, null, "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ for (Response returned : List.of(plain, compressed)) {
+ assertThat(returned.getStatus(),
is(Response.Status.OK.getStatusCode()));
+ String page = (String) returned.getEntity();
+ assertThat(page, containsString("last part"));
+ assertThat(page, not(containsString("first part")));
+ assertThat(page, containsString("Prev"));
+ assertThat(page, containsString("First"));
+ }
+ }
+ }
+
+ @Test
+ public void testLogPageDoesNotFollowSymbolicLinkToAnotherTopology() throws
Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("logs"));
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log.1"),
+
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log"));
+
+ WorkerLogs workerLogs = mock(WorkerLogs.class);
+ LogviewerLogPageHandler handler =
createHandlerAllowingOnlyTopoA(rootPath.getFile().toPath().resolve("logs"),
workerLogs);
+
+ final Response returned =
handler.logPage("topoA/1111/worker.log.1", 0, 100, null, "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat("rendered page: " + returned.getEntity(),
returned.getStatus(), is(Response.Status.NOT_FOUND.getStatusCode()));
+ verify(workerLogs, never()).setLogFilePermission(anyString());
+ }
+ }
+
+ @Test
+ public void testLogPageDoesNotListSymbolicLinks() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("logs"));
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log.1"),
+
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log"));
+
+ LogviewerLogPageHandler handler =
createHandlerAllowingOnlyTopoA(rootPath.getFile().toPath().resolve("logs"),
+ mock(WorkerLogs.class));
+
+ final Response returned = handler.logPage("topoA/1111/worker.log",
0, 100, null, "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(returned.getStatus(),
is(Response.Status.OK.getStatusCode()));
+ String page = (String) returned.getEntity();
+ assertThat(page, containsString("topoA content"));
+ assertThat(page, not(containsString("worker.log.1")));
+ }
+ }
+
+ @Test
+ public void testLogPageUnderSymbolicLinkRoot() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path realDaemonLogRoot =
rootPath.getFile().toPath().resolve("real-logs");
+ createWorkerLogTree(realDaemonLogRoot);
+ Path linkedDaemonLogRoot =
rootPath.getFile().toPath().resolve("logs");
+ SymbolicLinks.create(linkedDaemonLogRoot, realDaemonLogRoot);
+
+ WorkerLogs workerLogs = mock(WorkerLogs.class);
+ LogviewerLogPageHandler handler =
createHandlerAllowingOnlyTopoA(linkedDaemonLogRoot, workerLogs);
+
+ final Response returned = handler.logPage("topoA/1111/worker.log",
0, 100, null, "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(returned.getStatus(),
is(Response.Status.OK.getStatusCode()));
+ assertThat((String) returned.getEntity(), containsString("topoA
content"));
+ verify(workerLogs).setLogFilePermission("topoA/1111/worker.log");
+ }
+ }
+
+ @Test
+ public void testLogPageRejectsFileBelowSubdirectoryOfPortDir() throws
Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("logs"));
+ Path nested =
workerLogRoot.resolve("topoA").resolve("1111").resolve("sub").resolve("worker.log");
+ Files.createDirectories(nested.getParent());
+ Files.writeString(nested, "nested content");
+
+ WorkerLogs workerLogs = mock(WorkerLogs.class);
+ LogviewerLogPageHandler handler =
createHandlerAllowingOnlyTopoA(rootPath.getFile().toPath().resolve("logs"),
workerLogs);
+
+ final Response returned =
handler.logPage("topoA/1111/sub/worker.log", 0, 100, null, "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(returned.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ verify(workerLogs, never()).setLogFilePermission(anyString());
+ }
+ }
+
+ @Test
+ public void testLogPageServesRotatedAndCompressedWorkerLogs() throws
Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("logs"));
+ Path portDir = workerLogRoot.resolve("topoA").resolve("1111");
+ Files.writeString(portDir.resolve("worker.log.1"), "rotated
content");
+ try (OutputStream out = new
GZIPOutputStream(Files.newOutputStream(portDir.resolve("worker.log.2.gz")))) {
+ out.write("compressed
content".getBytes(StandardCharsets.UTF_8));
+ }
+
+ LogviewerLogPageHandler handler =
createHandlerAllowingOnlyTopoA(rootPath.getFile().toPath().resolve("logs"),
+ mock(WorkerLogs.class));
+
+ final Response rotated =
handler.logPage("topoA/1111/worker.log.1", null, 100, null, "user");
+ final Response compressed =
handler.logPage("topoA/1111/worker.log.2.gz", null, 100, null, "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(rotated.getStatus(),
is(Response.Status.OK.getStatusCode()));
+ assertThat((String) rotated.getEntity(), containsString("rotated
content"));
+ assertThat(compressed.getStatus(),
is(Response.Status.OK.getStatusCode()));
+ assertThat((String) compressed.getEntity(),
containsString("compressed content"));
+ }
+ }
+
+ @Test
+ public void testDaemonLogPageFollowsSymbolicLink() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ createWorkerLogTree(daemonLogRoot);
+ Path realLog =
Files.writeString(daemonLogRoot.resolve("nimbus.log.real"), "nimbus content");
+ SymbolicLinks.create(daemonLogRoot.resolve("nimbus.log"), realLog);
+
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+
when(resourceAuthorizer.isUserAllowedToAccessDaemonFile(anyString())).thenReturn(true);
+ LogviewerLogPageHandler handler = new
LogviewerLogPageHandler(daemonLogRoot.resolve("workers-artifacts").toString(),
+ daemonLogRoot.toString(), mock(WorkerLogs.class),
resourceAuthorizer, new StormMetricsRegistry());
+
+ final Response returned = handler.daemonLogPage("nimbus.log", 0,
100, null, "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(returned.getStatus(),
is(Response.Status.OK.getStatusCode()));
+ assertThat((String) returned.getEntity(), containsString("nimbus
content"));
+ }
+ }
+
+ @Test
+ public void testLogPageReplacedBySymbolicLinkAfterCheckIsNotFound() throws
Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ Path workerLogRoot = createWorkerLogTree(daemonLogRoot);
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log.1"),
+
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log"));
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerLogPageHandler handler = new
LogviewerLogPageHandler(workerLogRoot.toString(), daemonLogRoot.toString(),
+ mock(WorkerLogs.class), resourceAuthorizer, metricsRegistry);
+
+ final Response returned;
+ try (MockedStatic<LogFilePaths> ignored =
LogviewerLogDownloadHandlerTest.passWorkerFileCheck()) {
+ returned = handler.logPage("topoA/1111/worker.log.1", 0, 100,
null, "user");
+ }
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(returned.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_OPEN_EXCEPTIONS).getCount());
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_READ_EXCEPTIONS).getCount());
+ }
+ }
+
+ @Test
+ public void testLogPageRemovedAfterCheckIsNotFound() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ Path workerLogRoot = createWorkerLogTree(daemonLogRoot);
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerLogPageHandler handler = new
LogviewerLogPageHandler(workerLogRoot.toString(), daemonLogRoot.toString(),
+ mock(WorkerLogs.class), resourceAuthorizer, metricsRegistry);
+
+ final Response returned;
+ try (MockedStatic<LogFilePaths> ignored =
LogviewerLogDownloadHandlerTest.removeFileBeforeOpen()) {
+ returned = handler.logPage("topoA/1111/worker.log", 0, 100,
null, "user");
+ }
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(returned.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_OPEN_EXCEPTIONS).getCount());
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_READ_EXCEPTIONS).getCount());
+ }
+ }
+
+ @Test
+ public void testDaemonLogPageRemovedAfterCheckIsNotFound() throws
Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ createWorkerLogTree(daemonLogRoot);
+ Files.writeString(daemonLogRoot.resolve("nimbus.log"), "nimbus
content");
+
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+
when(resourceAuthorizer.isUserAllowedToAccessDaemonFile(anyString())).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerLogPageHandler handler = new
LogviewerLogPageHandler(daemonLogRoot.resolve("workers-artifacts").toString(),
+ daemonLogRoot.toString(), mock(WorkerLogs.class),
resourceAuthorizer, metricsRegistry);
+
+ final Response returned;
+ try (MockedStatic<LogFilePaths> ignored =
LogviewerLogDownloadHandlerTest.removeFileBeforeOpen()) {
+ returned = handler.daemonLogPage("nimbus.log", 0, 100, null,
"user");
+ }
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(returned.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_OPEN_EXCEPTIONS).getCount());
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_READ_EXCEPTIONS).getCount());
+ }
+ }
+
+ @Test
+ public void testListLogFilesLeavesOutSymbolicLinks() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ Path workerLogRoot = createWorkerLogTree(daemonLogRoot);
+ Path topoBFile =
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log");
+ //A file link, a port directory link and a topology directory link
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log.1"),
topoBFile);
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("2222"),
topoBFile.getParent());
+ SymbolicLinks.create(workerLogRoot.resolve("topoC"),
workerLogRoot.resolve("topoB"));
+
+ Map<String, Object> stormConf = Utils.readStormConfig();
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
anyString())).thenReturn(true);
+ LogviewerLogPageHandler handler = new
LogviewerLogPageHandler(workerLogRoot.toString(), daemonLogRoot.toString(),
+ new WorkerLogs(stormConf, workerLogRoot, metricsRegistry),
resourceAuthorizer, metricsRegistry);
+
+ ObjectMapper objectMapper = new ObjectMapper();
+ List<?> all = objectMapper.readValue((String)
handler.listLogFiles("user", null, null, null, null).getEntity(), List.class);
+ List<?> byPort = objectMapper.readValue((String)
handler.listLogFiles("user", 1111, null, null, null).getEntity(),
+ List.class);
+ List<?> byTopology = objectMapper.readValue((String)
handler.listLogFiles("user", null, "topoA", null, null).getEntity(),
+ List.class);
+ List<?> byLinkedTopology = objectMapper.readValue(
+ (String) handler.listLogFiles("user", null, "topoC", null,
null).getEntity(), List.class);
+ List<?> byLinkedPort = objectMapper.readValue((String)
handler.listLogFiles("user", 2222, "topoA", null, null).getEntity(),
+ List.class);
+ List<?> byLinkedPortOnly = objectMapper.readValue((String)
handler.listLogFiles("user", 2222, null, null, null).getEntity(),
+ List.class);
+ List<?> byTopologyAndPort = objectMapper.readValue(
+ (String) handler.listLogFiles("user", 1111, "topoA", null,
null).getEntity(), List.class);
+
+ Utils.forceDelete(rootPath.toString());
+
+ String topoALog = String.join(File.separator, "topoA", "1111",
"worker.log");
+ String topoBLog = String.join(File.separator, "topoB", "1111",
"worker.log");
+ assertEquals(List.of(topoALog, topoBLog), all);
+ assertEquals(List.of(topoALog, topoBLog), byPort);
+ assertEquals(List.of(topoALog), byTopology);
+ assertEquals(List.of(), byLinkedTopology);
+ assertEquals(List.of(), byLinkedPort);
+ assertEquals(List.of(), byLinkedPortOnly);
+ assertEquals(List.of(topoALog), byTopologyAndPort);
+ }
+ }
+
+ /**
+ * Creates topoA/1111/worker.log and topoB/1111/worker.log with distinct
content below the worker log root.
+ *
+ * @return the worker log root
+ */
+ private Path createWorkerLogTree(Path daemonLogRoot) throws IOException {
+ Path workerLogRoot = daemonLogRoot.resolve("workers-artifacts");
+ Path topoAFile =
workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log");
+ Path topoBFile =
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log");
+ Files.createDirectories(topoAFile.getParent());
+ Files.createDirectories(topoBFile.getParent());
+ Files.writeString(topoAFile, "topoA content");
+ Files.writeString(topoBFile, "topoB content");
+ return workerLogRoot;
+ }
+
+ private LogviewerLogPageHandler createHandlerAllowingOnlyTopoA(Path
daemonLogRoot, WorkerLogs workerLogs) {
+ //Models per topology access: the user may only read the logs of topoA.
+ ResourceAuthorizer resourceAuthorizer = mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ return new
LogviewerLogPageHandler(daemonLogRoot.resolve("workers-artifacts").toString(),
daemonLogRoot.toString(),
+ workerLogs, resourceAuthorizer, new StormMetricsRegistry());
+ }
+
private LogviewerLogPageHandler createHandlerForTraversalTests(Path
rootPath) throws IOException {
return createHandlerForTraversalTests(rootPath, new
ResourceAuthorizer(Utils.readStormConfig()));
}
diff --git
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogSearchHandlerTest.java
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogSearchHandlerTest.java
index 90c15c861..9c3e7f30a 100644
---
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogSearchHandlerTest.java
+++
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogSearchHandlerTest.java
@@ -19,12 +19,15 @@
package org.apache.storm.daemon.logviewer.handler;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.ArgumentMatchers.startsWith;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
@@ -53,6 +56,9 @@ import jakarta.ws.rs.core.Response;
import org.apache.storm.DaemonConfig;
import org.apache.storm.daemon.logviewer.LogviewerConstant;
+import org.apache.storm.daemon.logviewer.testsupport.SymbolicLinks;
+import org.apache.storm.daemon.logviewer.utils.ExceptionMeterNames;
+import org.apache.storm.daemon.logviewer.utils.LogFilePaths;
import org.apache.storm.daemon.logviewer.utils.ResourceAuthorizer;
import org.apache.storm.daemon.ui.InvalidRequestException;
import org.apache.storm.metric.StormMetricsRegistry;
@@ -64,6 +70,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
public class LogviewerLogSearchHandlerTest {
@@ -914,6 +921,237 @@ public class LogviewerLogSearchHandlerTest {
}
}
+ @Test
+ public void testSearchLogFileDoesNotFollowSymbolicLink() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log.1"),
+
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log"));
+ LogviewerLogSearchHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response response =
handler.searchLogFile("topoA/1111/worker.log.1", "user", false, "needle", null,
null, null, null);
+
+ assertEquals(404, response.getStatus(), "response: " +
response.getEntity());
+ }
+ }
+
+ @Test
+ public void testSearchLogFileRejectsFileBelowSubdirectoryOfPortDir()
throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ Path nested =
workerLogRoot.resolve("topoA").resolve("1111").resolve("sub").resolve("worker.log");
+ Files.createDirectories(nested.getParent());
+ Files.writeString(nested, "needle from nested\n");
+ LogviewerLogSearchHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response response =
handler.searchLogFile("topoA/1111/sub/worker.log", "user", false, "needle",
null, null, null, null);
+
+ assertEquals(404, response.getStatus(), "response: " +
response.getEntity());
+ }
+ }
+
+ @Test
+ public void testSearchLogFileUnderSymbolicLinkRoot() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path realWorkerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("real-workers-artifacts"));
+ Path linkedWorkerLogRoot =
rootPath.getFile().toPath().resolve("workers-artifacts");
+ SymbolicLinks.create(linkedWorkerLogRoot, realWorkerLogRoot);
+ LogviewerLogSearchHandler handler =
createHandlerAllowingOnlyTopoA(linkedWorkerLogRoot);
+
+ Response response = handler.searchLogFile("topoA/1111/worker.log",
"user", false, "needle", null, null, null, null);
+
+ assertEquals(200, response.getStatus());
+ Map<?, ?> entity = new ObjectMapper().readValue((String)
response.getEntity(), Map.class);
+ List<?> matches = (List<?>) entity.get("matches");
+ assertEquals(1, matches.size());
+ String afterString = (String) ((Map<?, ?>)
matches.get(0)).get("afterString");
+ assertTrue(afterString.startsWith(" from topoA"), afterString);
+ }
+ }
+
+ @Test
+ public void testLogsForPortSkipsSymbolicLinks() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ Path portDir = workerLogRoot.resolve("topoA").resolve("1111");
+ SymbolicLinks.create(portDir.resolve("worker.log.1"),
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log"));
+ LogviewerLogSearchHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ List<Path> logs = handler.logsForPort("user", portDir);
+
+ assertEquals(List.of(portDir.resolve("worker.log")), logs);
+ }
+ }
+
+ @Test
+ public void testDeepSearchLogsForTopologyDoesNotFollowSymbolicLinks()
throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log.1"),
+
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log"));
+ LogviewerLogSearchHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response response = handler.deepSearchLogsForTopology("topoA",
"user", "needle", "10", "*", "0", "0", true, null, null);
+
+ assertEquals(200, response.getStatus());
+ String entity = (String) response.getEntity();
+ assertTrue(entity.contains("from topoA"), entity);
+ assertFalse(entity.contains("from topoB"), entity);
+ }
+ }
+
+ @Test
+ public void testDeepSearchLogsForTopologyDoesNotFollowSymbolicPortDir()
throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ //6700 is one of the default supervisor slot ports, so the single
port search looks at it
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("6700"),
workerLogRoot.resolve("topoB").resolve("1111"));
+ LogviewerLogSearchHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response allPorts = handler.deepSearchLogsForTopology("topoA",
"user", "needle", "10", "*", "0", "0", true, null, null);
+ Response singlePort = handler.deepSearchLogsForTopology("topoA",
"user", "needle", "10", "6700", "0", "0", true, null,
+ null);
+
+ assertEquals(200, allPorts.getStatus());
+ String allPortsEntity = (String) allPorts.getEntity();
+ assertTrue(allPortsEntity.contains("from topoA"), allPortsEntity);
+ assertFalse(allPortsEntity.contains("from topoB"), allPortsEntity);
+ assertEquals(200, singlePort.getStatus());
+ String singlePortEntity = (String) singlePort.getEntity();
+ assertFalse(singlePortEntity.contains("from topoB"),
singlePortEntity);
+ }
+ }
+
+ @Test
+ public void testSearchLogFileReplacedBySymbolicLinkAfterCheckIsNotFound()
throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log.1"),
+
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log"));
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerLogSearchHandler handler = new
LogviewerLogSearchHandler(Utils.readStormConfig(), workerLogRoot, Paths.get(""),
+ resourceAuthorizer, metricsRegistry);
+
+ Response response;
+ try (MockedStatic<LogFilePaths> ignored =
LogviewerLogDownloadHandlerTest.passWorkerFileCheck()) {
+ response = handler.searchLogFile("topoA/1111/worker.log.1",
"user", false, "needle", null, null, null, null);
+ }
+
+ assertEquals(404, response.getStatus(), "response: " +
response.getEntity());
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_OPEN_EXCEPTIONS).getCount());
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_READ_EXCEPTIONS).getCount());
+ }
+ }
+
+ @Test
+ public void testFindNMatchesSkipsLogReplacedBySymbolicLink() throws
Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ Path linkedLog =
workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log.1");
+ SymbolicLinks.create(linkedLog,
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log"));
+ List<Path> files = List.of(linkedLog,
workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log"));
+
+ final LogviewerLogSearchHandler.Matched matched =
getSearchHandler().findNMatches(files, 20, 0, 0, "needle");
+ final List<Map<String, Object>> matches = matched.getMatches();
+
+ //Only the regular file is counted as scanned
+ assertEquals(1, matched.getOpenedFiles());
+ assertEquals(1, matches.size());
+ assertEquals(String.join(File.separator, "topoA", "1111",
"worker.log"), matches.get(0).get("fileName"));
+ List<?> fileMatches = (List<?>) matches.get(0).get("matches");
+ assertEquals(1, fileMatches.size());
+ assertTrue(((Map<?, ?>)
fileMatches.get(0)).get("afterString").toString().contains("from topoA"));
+ }
+ }
+
+ @Test
+ public void testSearchLogFileRemovedAfterCheckIsNotFound() throws
Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerLogSearchHandler handler = new
LogviewerLogSearchHandler(Utils.readStormConfig(), workerLogRoot, Paths.get(""),
+ resourceAuthorizer, metricsRegistry);
+
+ Response response;
+ try (MockedStatic<LogFilePaths> ignored =
LogviewerLogDownloadHandlerTest.removeFileBeforeOpen()) {
+ response = handler.searchLogFile("topoA/1111/worker.log",
"user", false, "needle", null, null, null, null);
+ }
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertEquals(404, response.getStatus(), "response: " +
response.getEntity());
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_OPEN_EXCEPTIONS).getCount());
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_READ_EXCEPTIONS).getCount());
+ }
+ }
+
+ @Test
+ public void testSearchDaemonLogFileRemovedAfterCheckIsNotFound() throws
Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+ Files.createDirectories(daemonLogRoot);
+ Files.writeString(daemonLogRoot.resolve("nimbus.log"), "a needle
in the daemon log\n");
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+
when(resourceAuthorizer.isUserAllowedToAccessDaemonFile(anyString())).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerLogSearchHandler handler = new
LogviewerLogSearchHandler(Utils.readStormConfig(), Paths.get(""), daemonLogRoot,
+ resourceAuthorizer, metricsRegistry);
+
+ Response response;
+ try (MockedStatic<LogFilePaths> ignored =
LogviewerLogDownloadHandlerTest.removeFileBeforeOpen()) {
+ response = handler.searchLogFile("nimbus.log", "user", true,
"needle", null, null, null, null);
+ }
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertEquals(404, response.getStatus(), "response: " +
response.getEntity());
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_OPEN_EXCEPTIONS).getCount());
+ assertEquals(0L,
metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_READ_EXCEPTIONS).getCount());
+ }
+ }
+
+ @Test
+ public void testFindNMatchesSkipsRemovedLog() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createWorkerLogTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ Path removedLog =
workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log.1");
+ List<Path> files = List.of(removedLog,
workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log"));
+
+ final LogviewerLogSearchHandler.Matched matched =
getSearchHandler().findNMatches(files, 20, 0, 0, "needle");
+
+ assertEquals(1, matched.getOpenedFiles());
+ assertEquals(1, matched.getMatches().size());
+ assertEquals(String.join(File.separator, "topoA", "1111",
"worker.log"), matched.getMatches().get(0).get("fileName"));
+ }
+ }
+
+ /**
+ * Creates topoA/1111/worker.log and topoB/1111/worker.log, each with one
distinct match for "needle".
+ *
+ * @return the worker log root
+ */
+ private static Path createWorkerLogTree(Path workerLogRoot) throws
IOException {
+ Path topoAFile =
workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log");
+ Path topoBFile =
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.log");
+ Files.createDirectories(topoAFile.getParent());
+ Files.createDirectories(topoBFile.getParent());
+ Files.writeString(topoAFile, "needle from topoA\n");
+ Files.writeString(topoBFile, "needle from topoB\n");
+ return workerLogRoot;
+ }
+
+ private static LogviewerLogSearchHandler
createHandlerAllowingOnlyTopoA(Path workerLogRoot) {
+ //Models per topology access: the user may only read the logs of topoA.
+ ResourceAuthorizer resourceAuthorizer = mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ return new LogviewerLogSearchHandler(Utils.readStormConfig(),
workerLogRoot, Paths.get(""),
+ resourceAuthorizer, new StormMetricsRegistry());
+ }
+
private static LogviewerLogSearchHandler getSearchHandler() {
Map<String, Object> stormConf = Utils.readStormConfig();
return new LogviewerLogSearchHandler(stormConf, Paths.get(""),
Paths.get(""),
diff --git
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerProfileHandlerTest.java
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerProfileHandlerTest.java
index 136390175..f6afb13bc 100644
---
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerProfileHandlerTest.java
+++
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerProfileHandlerTest.java
@@ -24,19 +24,30 @@ import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.startsWith;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import com.google.common.net.HttpHeaders;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Map;
import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.StreamingOutput;
+import org.apache.storm.daemon.logviewer.testsupport.SymbolicLinks;
+import org.apache.storm.daemon.logviewer.utils.ExceptionMeterNames;
+import org.apache.storm.daemon.logviewer.utils.LogFilePaths;
import org.apache.storm.daemon.logviewer.utils.ResourceAuthorizer;
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.testing.TmpPath;
import org.apache.storm.utils.Utils;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
public class LogviewerProfileHandlerTest {
@@ -141,6 +152,245 @@ public class LogviewerProfileHandlerTest {
}
}
+ @Test
+ public void testDownloadDumpFileDoesNotFollowSymbolicLink() throws
IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createDumpFileTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("other.jfr"),
+
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.jfr"));
+ LogviewerProfileHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response response = handler.downloadDumpFile("topoA",
"localhost:1111", "other.jfr", "user");
+ int status = response.getStatus();
+ String content = status == Response.Status.OK.getStatusCode() ?
readEntity(response) : null;
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat("served content: " + content, status,
is(Response.Status.NOT_FOUND.getStatusCode()));
+ }
+ }
+
+ @Test
+ public void testListDumpFilesDoesNotListSymbolicLinks() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createDumpFileTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("other.jfr"),
+
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.jfr"));
+ LogviewerProfileHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response response = handler.listDumpFiles("topoA",
"localhost:1111", "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(response.getStatus(),
is(Response.Status.OK.getStatusCode()));
+ String content = (String) response.getEntity();
+ assertThat(content, containsString("worker.jfr"));
+ assertThat(content, not(containsString("other.jfr")));
+ }
+ }
+
+ @Test
+ public void testListDumpFilesDoesNotFollowSymbolicPortDir() throws
Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createDumpFileTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ Path outside = rootPath.getFile().toPath().resolve("outside");
+ Files.createDirectories(outside);
+ Files.writeString(outside.resolve("secret.jfr"), "outside jfr");
+ //A topology id containing a separator would name a directory
below the port directory, which the worker controls
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("123"),
outside);
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("2222"),
workerLogRoot.resolve("topoB").resolve("1111"));
+ LogviewerProfileHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response belowPortDir = handler.listDumpFiles("topoA" +
File.separator + "1111", "localhost:123", "user");
+ Response linkedPortDir = handler.listDumpFiles("topoA",
"localhost:2222", "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat((String) belowPortDir.getEntity(),
belowPortDir.getStatus(), is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertThat((String) linkedPortDir.getEntity(),
linkedPortDir.getStatus(), is(Response.Status.NOT_FOUND.getStatusCode()));
+ }
+ }
+
+ @Test
+ public void testDownloadDumpFileWithIpv6Host() throws IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createDumpFileTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ LogviewerProfileHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response list = handler.listDumpFiles("topoA", "[::1]:1111",
"user");
+ Response response = handler.downloadDumpFile("topoA",
"[::1]:1111", "worker.jfr", "user");
+ String content = readEntity(response);
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(list.getStatus(),
is(Response.Status.OK.getStatusCode()));
+ assertThat((String) list.getEntity(),
containsString("worker.jfr"));
+ assertThat(response.getStatus(),
is(Response.Status.OK.getStatusCode()));
+
assertThat(response.getHeaderString(HttpHeaders.CONTENT_DISPOSITION),
containsString("[::1]-topoA-1111-worker.jfr"));
+ assertThat(content, is("TopoA jfr"));
+ }
+ }
+
+ @Test
+ public void testDownloadDumpFileUnderSymbolicLinkRoot() throws IOException
{
+ try (TmpPath rootPath = new TmpPath()) {
+ Path realWorkerLogRoot =
createDumpFileTree(rootPath.getFile().toPath().resolve("real-workers-artifacts"));
+ Path linkedWorkerLogRoot =
rootPath.getFile().toPath().resolve("workers-artifacts");
+ SymbolicLinks.create(linkedWorkerLogRoot, realWorkerLogRoot);
+ LogviewerProfileHandler handler =
createHandlerAllowingOnlyTopoA(linkedWorkerLogRoot);
+
+ Response response = handler.downloadDumpFile("topoA",
"localhost:1111", "worker.jfr", "user");
+ int status = response.getStatus();
+ String content = status == Response.Status.OK.getStatusCode() ?
readEntity(response) : null;
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(status, is(Response.Status.OK.getStatusCode()));
+ assertThat(content, is("TopoA jfr"));
+ }
+ }
+
+ @Test
+ public void
testDownloadDumpFileReplacedBySymbolicLinkAfterCheckIsNotFound() throws
IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createDumpFileTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+
SymbolicLinks.create(workerLogRoot.resolve("topoA").resolve("1111").resolve("other.jfr"),
+
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.jfr"));
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerProfileHandler handler = new
LogviewerProfileHandler(workerLogRoot.toString(), resourceAuthorizer,
metricsRegistry);
+
+ Response response;
+ try (MockedStatic<LogFilePaths> ignored =
LogviewerLogDownloadHandlerTest.passWorkerFileCheck()) {
+ response = handler.downloadDumpFile("topoA", "localhost:1111",
"other.jfr", "user");
+ }
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(response.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+
assertThat(metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_DOWNLOAD_EXCEPTIONS).getCount(),
is(0L));
+ }
+ }
+
+ @Test
+ public void testDownloadDumpFileRejectsFileNameWithSeveralElements()
throws IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createDumpFileTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ Path nested =
workerLogRoot.resolve("topoA").resolve("1111").resolve("sub").resolve("worker.jfr");
+ Files.createDirectories(nested.getParent());
+ Files.writeString(nested, "nested jfr");
+ LogviewerProfileHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response nestedResponse = handler.downloadDumpFile("topoA",
"localhost:1111", "sub/worker.jfr", "user");
+ //With an empty port, a two element file name would point at a
direct child of another port directory
+ Response emptyPortResponse = handler.downloadDumpFile("topoA",
"localhost::1111", "1111/worker.jfr", "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(nestedResponse.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertThat(emptyPortResponse.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ }
+ }
+
+ @Test
+ public void testDownloadDumpFileRemovedAfterCheckIsNotFound() throws
IOException {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createDumpFileTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ ResourceAuthorizer resourceAuthorizer =
mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+ LogviewerProfileHandler handler = new
LogviewerProfileHandler(workerLogRoot.toString(), resourceAuthorizer,
metricsRegistry);
+
+ Response response;
+ try (MockedStatic<LogFilePaths> ignored =
LogviewerLogDownloadHandlerTest.removeFileBeforeOpen()) {
+ response = handler.downloadDumpFile("topoA", "localhost:1111",
"worker.jfr", "user");
+ }
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(response.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+
assertThat(metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_DOWNLOAD_EXCEPTIONS).getCount(),
is(0L));
+ }
+ }
+
+ @Test
+ public void testDumpEndpointsRejectPortThatIsNotANumber() throws Exception
{
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createDumpFileTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+
Files.createDirectories(workerLogRoot.resolve("topoA").resolve("abc"));
+
Files.writeString(workerLogRoot.resolve("topoA").resolve("abc").resolve("worker.jfr"),
"abc jfr");
+
Files.createDirectories(workerLogRoot.resolve("topoA").resolve("-1111"));
+
Files.writeString(workerLogRoot.resolve("topoA").resolve("-1111").resolve("worker.jfr"),
"negative jfr");
+ LogviewerProfileHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response listNonNumeric = handler.listDumpFiles("topoA",
"localhost:abc", "user");
+ Response downloadNonNumeric = handler.downloadDumpFile("topoA",
"localhost:abc", "worker.jfr", "user");
+ Response listNegative = handler.listDumpFiles("topoA",
"localhost:-1111", "user");
+ Response downloadNegative = handler.downloadDumpFile("topoA",
"localhost:-1111", "worker.jfr", "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(listNonNumeric.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertThat(downloadNonNumeric.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertThat(listNegative.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertThat(downloadNegative.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ }
+ }
+
+ @Test
+ public void testDumpEndpointsRejectEmptyPort() throws Exception {
+ try (TmpPath rootPath = new TmpPath()) {
+ Path workerLogRoot =
createDumpFileTree(rootPath.getFile().toPath().resolve("workers-artifacts"));
+ LogviewerProfileHandler handler =
createHandlerAllowingOnlyTopoA(workerLogRoot);
+
+ Response listEmpty = handler.listDumpFiles("topoA", "localhost:",
"user");
+ Response downloadEmpty = handler.downloadDumpFile("topoA",
"localhost:", "worker.jfr", "user");
+ //The port is the part after the last ':', so an IPv6 host without
a port has an empty port
+ Response listEmptyWithSuffix = handler.listDumpFiles("topoA",
"[::1]:", "user");
+ Response downloadEmptyWithSuffix =
handler.downloadDumpFile("topoA", "[::1]:", "worker.jfr", "user");
+ Response listNoPort = handler.listDumpFiles("topoA", "localhost",
"user");
+ Response downloadNoPort = handler.downloadDumpFile("topoA",
"localhost", "worker.jfr", "user");
+
+ Utils.forceDelete(rootPath.toString());
+
+ assertThat(listEmpty.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertThat(downloadEmpty.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertThat(listEmptyWithSuffix.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertThat(downloadEmptyWithSuffix.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertThat(listNoPort.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ assertThat(downloadNoPort.getStatus(),
is(Response.Status.NOT_FOUND.getStatusCode()));
+ }
+ }
+
+ /**
+ * Creates topoA/1111/worker.jfr and topoB/1111/worker.jfr with distinct
content below the worker log root.
+ *
+ * @return the worker log root
+ */
+ private Path createDumpFileTree(Path workerLogRoot) throws IOException {
+ Path topoAFile =
workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.jfr");
+ Path topoBFile =
workerLogRoot.resolve("topoB").resolve("1111").resolve("worker.jfr");
+ Files.createDirectories(topoAFile.getParent());
+ Files.createDirectories(topoBFile.getParent());
+ Files.writeString(topoAFile, "TopoA jfr");
+ Files.writeString(topoBFile, "TopoB jfr");
+ return workerLogRoot;
+ }
+
+ private LogviewerProfileHandler createHandlerAllowingOnlyTopoA(Path
workerLogRoot) {
+ //Models per topology access: the user may only read the files of
topoA.
+ ResourceAuthorizer resourceAuthorizer = mock(ResourceAuthorizer.class);
+ when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(),
startsWith("topoA" + File.separator))).thenReturn(true);
+ return new LogviewerProfileHandler(workerLogRoot.toString(),
resourceAuthorizer, new StormMetricsRegistry());
+ }
+
+ private String readEntity(Response response) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ((StreamingOutput) response.getEntity()).write(out);
+ return out.toString(StandardCharsets.UTF_8);
+ }
+
private LogviewerProfileHandler createHandlerTraversalTests(Path rootPath)
throws IOException {
Path daemonLogRoot = rootPath.resolve("logs");
Path fileOutsideDaemonRoot = rootPath.resolve("evil.bin");
diff --git
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/testsupport/SymbolicLinks.java
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/testsupport/SymbolicLinks.java
new file mode 100644
index 000000000..1b69d3192
--- /dev/null
+++
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/testsupport/SymbolicLinks.java
@@ -0,0 +1,56 @@
+/*
+ * 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.storm.daemon.logviewer.testsupport;
+
+import java.io.IOException;
+import java.nio.file.FileSystemException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.apache.commons.lang3.SystemUtils;
+import org.junit.jupiter.api.Assumptions;
+
+/**
+ * Create symbolic links in tests. The test is skipped where symbolic links
can not be created, e.g. on Windows without the
+ * privilege to create them.
+ */
+public final class SymbolicLinks {
+
+ private SymbolicLinks() {
+ }
+
+ /**
+ * Create a symbolic link, or skip the test if symbolic links can not be
created.
+ *
+ * @param link the link to create
+ * @param target the target of the link
+ * @return the link
+ */
+ public static Path create(Path link, Path target) throws IOException {
+ try {
+ return Files.createSymbolicLink(link, target);
+ } catch (UnsupportedOperationException e) {
+ return Assumptions.abort("Symbolic links are not supported: " + e);
+ } catch (FileSystemException e) {
+ if (SystemUtils.IS_OS_WINDOWS) {
+ return Assumptions.abort("Symbolic links can not be created: "
+ e);
+ }
+ throw e;
+ }
+ }
+}
diff --git
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/LogFilePathsTest.java
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/LogFilePathsTest.java
new file mode 100644
index 000000000..d2a22203c
--- /dev/null
+++
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/LogFilePathsTest.java
@@ -0,0 +1,253 @@
+/*
+ * 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.storm.daemon.logviewer.utils;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeFalse;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.Channels;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+
+import org.apache.storm.daemon.logviewer.testsupport.SymbolicLinks;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class LogFilePathsTest {
+
+ @TempDir
+ Path tempDir;
+
+ private Path createWorkerLog(Path root, String topology, String content)
throws IOException {
+ Path portDir =
Files.createDirectories(root.resolve(topology).resolve("6700"));
+ return Files.write(portDir.resolve("worker.log"),
content.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static Path normalized(Path path) {
+ return path.toAbsolutePath().normalize();
+ }
+
+ private static String readAll(SeekableByteChannel channel) throws
IOException {
+ return new String(Channels.newInputStream(channel).readAllBytes(),
StandardCharsets.UTF_8);
+ }
+
+ @Test
+ public void testRegularFileIsAccepted() throws IOException {
+ Path root =
normalized(Files.createDirectories(tempDir.resolve("logs")));
+ Path file = normalized(createWorkerLog(root, "topoA", "content"));
+
+ assertTrue(LogFilePaths.isRegularFileWithoutSymbolicLink(root, file));
+ assertTrue(LogFilePaths.isServableWorkerFile(root, file));
+ assertTrue(LogFilePaths.isRegularFileNotLink(file));
+ try (SeekableByteChannel channel = LogFilePaths.openForReading(file,
false)) {
+ assertEquals("content", readAll(channel));
+ }
+ }
+
+ @Test
+ public void testMissingFileIsRejected() throws IOException {
+ Path root =
normalized(Files.createDirectories(tempDir.resolve("logs")));
+ createWorkerLog(root, "topoA", "content");
+
+ assertFalse(LogFilePaths.isServableWorkerFile(root,
root.resolve("topoA").resolve("6700").resolve("missing.log")));
+ assertFalse(LogFilePaths.isServableWorkerFile(root,
root.resolve("topoB").resolve("6700").resolve("worker.log")));
+ }
+
+ @Test
+ public void testDirectoryIsRejected() throws IOException {
+ Path root =
normalized(Files.createDirectories(tempDir.resolve("logs")));
+ Path dir =
Files.createDirectories(root.resolve("topoA").resolve("6700").resolve("worker.log"));
+
+ assertFalse(LogFilePaths.isServableWorkerFile(root, dir));
+ }
+
+ @Test
+ public void testFileOutsideRootIsRejected() throws IOException {
+ Path root =
normalized(Files.createDirectories(tempDir.resolve("logs")));
+ Path outside = Files.write(tempDir.resolve("outside.txt"),
"outside".getBytes(StandardCharsets.UTF_8));
+
+ assertFalse(LogFilePaths.isRegularFileWithoutSymbolicLink(root,
normalized(outside)));
+ assertFalse(LogFilePaths.isServableWorkerFile(root,
normalized(outside)));
+ }
+
+ @Test
+ public void testOnlyDirectChildrenOfPortDirAreWorkerFiles() throws
IOException {
+ Path root =
normalized(Files.createDirectories(tempDir.resolve("logs")));
+ Path nested =
root.resolve("topoA").resolve("6700").resolve("sub").resolve("worker.log");
+ Files.createDirectories(nested.getParent());
+ Files.write(nested, "nested".getBytes(StandardCharsets.UTF_8));
+ Path shallow =
Files.write(root.resolve("topoA").resolve("worker.log"),
"shallow".getBytes(StandardCharsets.UTF_8));
+
+ assertTrue(LogFilePaths.isWorkerFilePath(root,
root.resolve("topoA").resolve("6700").resolve("worker.log")));
+ assertFalse(LogFilePaths.isWorkerFilePath(root, nested));
+ assertFalse(LogFilePaths.isServableWorkerFile(root, nested));
+ assertFalse(LogFilePaths.isServableWorkerFile(root, shallow));
+ }
+
+ @Test
+ public void testSymbolicLinkBelowRootIsRejected() throws IOException {
+ Path root =
normalized(Files.createDirectories(tempDir.resolve("logs")));
+ Path target = createWorkerLog(root, "topoB", "topoB content");
+ Path portDir =
Files.createDirectories(root.resolve("topoA").resolve("6700"));
+ Path link = SymbolicLinks.create(portDir.resolve("worker.log"),
target);
+ Path outside = Files.write(tempDir.resolve("outside.txt"),
"outside".getBytes(StandardCharsets.UTF_8));
+ Path outsideLink =
SymbolicLinks.create(portDir.resolve("outside.log"), outside);
+
+ assertFalse(LogFilePaths.isServableWorkerFile(root, normalized(link)));
+ assertFalse(LogFilePaths.isServableWorkerFile(root,
normalized(outsideLink)));
+ assertFalse(LogFilePaths.isRegularFileNotLink(link));
+ assertNull(LogFilePaths.openForReading(link, false));
+ try (SeekableByteChannel channel = LogFilePaths.openForReading(link,
true)) {
+ assertEquals("topoB content", readAll(channel));
+ }
+ }
+
+ @Test
+ public void testOpenReportsFileReplacedBySymbolicLinkAfterCheck() throws
IOException {
+ Path root =
normalized(Files.createDirectories(tempDir.resolve("logs")));
+ Path file = createWorkerLog(root, "topoA", "topoA content");
+ Path target = createWorkerLog(root, "topoB", "topoB content");
+ assertTrue(LogFilePaths.isServableWorkerFile(root, normalized(file)));
+
+ //The file is replaced by a symbolic link between the check and the
open
+ Files.delete(file);
+ SymbolicLinks.create(file, target);
+
+ assertNull(LogFilePaths.openForReading(file, false));
+ }
+
+ @Test
+ public void testOpenOfMissingFileReturnsNull() throws IOException {
+ Path missing = tempDir.resolve("missing.log");
+
+ assertNull(LogFilePaths.openForReading(missing, false));
+ assertNull(LogFilePaths.openForReading(missing, true));
+ }
+
+ @Test
+ public void testOpenReportsFileRemovedAfterCheck() throws IOException {
+ Path root =
normalized(Files.createDirectories(tempDir.resolve("logs")));
+ Path file = createWorkerLog(root, "topoA", "topoA content");
+ assertTrue(LogFilePaths.isServableWorkerFile(root, normalized(file)));
+
+ //The file is removed (e.g. by the log cleaner) between the check and
the open
+ Files.delete(file);
+
+ assertNull(LogFilePaths.openForReading(file, false));
+ }
+
+ @Test
+ public void testOpenOfDanglingSymbolicLinkReturnsNull() throws IOException
{
+ Path link = SymbolicLinks.create(tempDir.resolve("nimbus.log"),
tempDir.resolve("missing.log"));
+
+ assertNull(LogFilePaths.openForReading(link, true));
+ assertNull(LogFilePaths.openForReading(link, false));
+ }
+
+ @Test
+ public void testOpenOfUnreadableFileFails() throws IOException {
+
assumeTrue(FileSystems.getDefault().supportedFileAttributeViews().contains("posix"));
+ assumeFalse("root".equals(System.getProperty("user.name")));
+ Path file = Files.writeString(tempDir.resolve("unreadable.log"),
"content");
+ Files.setPosixFilePermissions(file, Collections.emptySet());
+
+ //Errors other than a missing file or a symbolic link are still thrown
+ assertThrows(AccessDeniedException.class, () ->
LogFilePaths.openForReading(file, false));
+ assertThrows(AccessDeniedException.class, () ->
LogFilePaths.openForReading(file, true));
+ }
+
+ @Test
+ public void testSymbolicLinkDirectoryBelowRootIsRejected() throws
IOException {
+ Path root =
normalized(Files.createDirectories(tempDir.resolve("logs")));
+ createWorkerLog(root, "topoB", "topoB content");
+ Path topoDir = Files.createDirectories(root.resolve("topoA"));
+ SymbolicLinks.create(topoDir.resolve("6700"),
root.resolve("topoB").resolve("6700"));
+
+ assertFalse(LogFilePaths.isServableWorkerFile(root,
topoDir.resolve("6700").resolve("worker.log")));
+ assertFalse(LogFilePaths.isDirectoryNotLink(topoDir.resolve("6700")));
+
assertTrue(LogFilePaths.isDirectoryNotLink(root.resolve("topoB").resolve("6700")));
+ }
+
+ @Test
+ public void testOpenWorkerFileOpensOnlyServableFiles() throws IOException {
+ Path root =
normalized(Files.createDirectories(tempDir.resolve("logs")));
+ Path file = normalized(createWorkerLog(root, "topoA", "topoA
content"));
+ createWorkerLog(root, "topoB", "topoB content");
+ SymbolicLinks.create(root.resolve("topoA").resolve("6701"),
root.resolve("topoB").resolve("6700"));
+ Path nested =
Files.createDirectories(root.resolve("topoA").resolve("6700").resolve("sub")).resolve("worker.log");
+ Files.writeString(nested, "nested content");
+
+ try (SeekableByteChannel channel = LogFilePaths.openWorkerFile(root,
file)) {
+ assertEquals("topoA content", readAll(channel));
+ }
+ assertNull(LogFilePaths.openWorkerFile(root,
root.resolve("topoA").resolve("6701").resolve("worker.log")));
+ assertNull(LogFilePaths.openWorkerFile(root, nested));
+ assertNull(LogFilePaths.openWorkerFile(root,
root.resolve("topoA").resolve("6700").resolve("missing.log")));
+ }
+
+ @Test
+ public void testSymbolicLinkAsRootIsAccepted() throws IOException {
+ Path realRoot = Files.createDirectories(tempDir.resolve("real-logs"));
+ createWorkerLog(realRoot, "topoA", "content");
+ Path linkRoot =
normalized(SymbolicLinks.create(tempDir.resolve("logs"), realRoot));
+
+ Path file =
linkRoot.resolve("topoA").resolve("6700").resolve("worker.log");
+ assertTrue(LogFilePaths.isServableWorkerFile(linkRoot, file));
+ assertTrue(LogFilePaths.isRegularFileNotLink(file));
+ try (SeekableByteChannel channel = LogFilePaths.openForReading(file,
false)) {
+ assertEquals("content", readAll(channel));
+ }
+ }
+
+ @Test
+ public void testFileLengthOfPlainAndCompressedFile() throws IOException {
+ Path plain = Files.write(tempDir.resolve("worker.log"), "plain
content".getBytes(StandardCharsets.UTF_8));
+ Path compressed = tempDir.resolve("worker.log.1.gz");
+ byte[] content = "compressed content".getBytes(StandardCharsets.UTF_8);
+ try (OutputStream out = new
GZIPOutputStream(Files.newOutputStream(compressed))) {
+ out.write(content);
+ }
+
+ try (SeekableByteChannel channel = LogFilePaths.openForReading(plain,
false)) {
+ assertEquals(13, LogFilePaths.fileLength(channel, false));
+ }
+ try (SeekableByteChannel channel =
LogFilePaths.openForReading(compressed, false)) {
+ assertEquals(content.length, LogFilePaths.fileLength(channel,
true));
+ //The channel is positioned at the start again, so the content can
be read afterwards
+ try (InputStream in = new
GZIPInputStream(Channels.newInputStream(channel))) {
+ assertEquals("compressed content", new
String(in.readAllBytes(), StandardCharsets.UTF_8));
+ }
+ }
+ }
+}
diff --git
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/LogviewerResponseBuilderTest.java
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/LogviewerResponseBuilderTest.java
index 4a06c3fc2..b516b1b09 100644
---
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/LogviewerResponseBuilderTest.java
+++
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/LogviewerResponseBuilderTest.java
@@ -21,10 +21,19 @@ package org.apache.storm.daemon.logviewer.utils;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.nullValue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.StreamingOutput;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
import java.util.Collections;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.jupiter.api.Test;
@@ -43,4 +52,44 @@ public class LogviewerResponseBuilderTest {
assertThat(response.getHeaderString("Access-Control-Allow-Origin"),
is("*"));
assertThat(response.getHeaderString("Access-Control-Allow-Credentials"),
is(nullValue()));
}
+
+ @Test
+ public void testDownloadFileClosesInputAfterWrite() throws IOException {
+ AtomicBoolean closed = new AtomicBoolean();
+ InputStream is = new
ByteArrayInputStream("content".getBytes(StandardCharsets.UTF_8)) {
+ @Override
+ public void close() {
+ closed.set(true);
+ }
+ };
+ Response response =
LogviewerResponseBuilder.buildDownloadFile("file.log", is);
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+
+ ((StreamingOutput) response.getEntity()).write(out);
+
+ assertThat(out.toString(StandardCharsets.UTF_8), is("content"));
+ assertThat(closed.get(), is(true));
+ }
+
+ @Test
+ public void testDownloadFileClosesInputWhenWriteFails() {
+ AtomicBoolean closed = new AtomicBoolean();
+ InputStream is = new
ByteArrayInputStream("content".getBytes(StandardCharsets.UTF_8)) {
+ @Override
+ public void close() {
+ closed.set(true);
+ }
+ };
+ Response response =
LogviewerResponseBuilder.buildDownloadFile("file.log", is);
+ //Models a client that aborts the download
+ OutputStream failing = new OutputStream() {
+ @Override
+ public void write(int b) throws IOException {
+ throw new IOException("aborted");
+ }
+ };
+
+ assertThrows(IOException.class, () -> ((StreamingOutput)
response.getEntity()).write(failing));
+ assertThat(closed.get(), is(true));
+ }
}
diff --git
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/WorkerLogsTest.java
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/WorkerLogsTest.java
index 6a5ae7dcc..b14ae4d3d 100644
---
a/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/WorkerLogsTest.java
+++
b/storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/WorkerLogsTest.java
@@ -27,11 +27,13 @@ import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.SortedSet;
import java.util.TreeSet;
+import org.apache.storm.daemon.logviewer.testsupport.SymbolicLinks;
import org.apache.storm.daemon.supervisor.SupervisorUtils;
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.testing.TmpPath;
@@ -75,4 +77,26 @@ public class WorkerLogsTest {
}
}
+ @Test
+ public void testGetAllLogsForRootDirLeavesOutSymbolicLinks() throws
Exception {
+ try (TmpPath testDir = new TmpPath()) {
+ Path logRoot =
testDir.getFile().toPath().resolve("workers-artifacts");
+ Path topoAPortDir =
Files.createDirectories(logRoot.resolve("topoA").resolve("6700"));
+ Path topoBPortDir =
Files.createDirectories(logRoot.resolve("topoB").resolve("6700"));
+ Path topoALog =
Files.writeString(topoAPortDir.resolve("worker.log"), "topoA");
+ Path topoBLog =
Files.writeString(topoBPortDir.resolve("worker.log"), "topoB");
+ //A file link, a port directory link and a topology directory link
+ SymbolicLinks.create(topoAPortDir.resolve("worker.log.1"),
topoBLog);
+ SymbolicLinks.create(logRoot.resolve("topoA").resolve("6701"),
topoBPortDir);
+ SymbolicLinks.create(logRoot.resolve("topoC"),
logRoot.resolve("topoB"));
+
+ WorkerLogs workerLogs = new WorkerLogs(Utils.readStormConfig(),
logRoot, new StormMetricsRegistry());
+
+ List<Path> logs = workerLogs.getAllLogsForRootDir();
+ Collections.sort(logs);
+
+ assertEquals(List.of(topoALog, topoBLog), logs);
+ }
+ }
+
}
