From 3617fa71f1ba3aebfc26c07fc9efee0fea84baf2 Mon Sep 17 00:00:00 2001 From: Matthias Pohl Date: Mon, 23 Nov 2020 11:38:41 +0100 Subject: [PATCH] [FLINK-20267][runtime] The JaasModule didn't support symbolic links. This is fixed now. A test was added to verify the change. --- .../runtime/security/modules/JaasModule.java | 13 ++++++-- .../security/modules/JaasModuleTest.java | 30 ++++++++++++++++--- 2 files changed, 37 insertions(+), 6 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/security/modules/JaasModule.java b/flink-runtime/src/main/java/org/apache/flink/runtime/security/modules/JaasModule.java index 16c176b1b331bd..0c626e746a6f13 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/security/modules/JaasModule.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/security/modules/JaasModule.java @@ -23,6 +23,7 @@ import org.apache.flink.runtime.security.DynamicConfiguration; import org.apache.flink.runtime.security.KerberosUtils; import org.apache.flink.runtime.security.SecurityConfiguration; +import org.apache.flink.util.FileUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -159,12 +160,20 @@ private static File generateDefaultConfigFile(String workingDir) { checkArgument(workingDir != null, "working directory should not be null."); final File jaasConfFile; try { - Path path = Files.createDirectories(Paths.get(workingDir)); + Path path = Paths.get(workingDir); + if (Files.notExists(Paths.get(workingDir))) { + // We intentionally favored Path.toRealPath over Files.readSymbolicLinks as the latter one might return a + // relative path if the symbolic link refers to it. Path.toRealPath resolves the relative path instead. + Path parent = path.getParent().toRealPath(); + Path resolvedPath = Paths.get(parent.toString(), path.getFileName().toString()); + + path = Files.createDirectories(resolvedPath); + } Path jaasConfPath = Files.createTempFile(path, "jaas-", ".conf"); try (InputStream resourceStream = JaasModule.class.getClassLoader().getResourceAsStream(JAAS_CONF_RESOURCE_NAME)) { Files.copy(resourceStream, jaasConfPath, StandardCopyOption.REPLACE_EXISTING); } - jaasConfFile = jaasConfPath.toFile(); + jaasConfFile = new File(workingDir, jaasConfPath.getFileName().toString()); jaasConfFile.deleteOnExit(); } catch (IOException e) { throw new RuntimeException("unable to generate a JAAS configuration file", e); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/security/modules/JaasModuleTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/security/modules/JaasModuleTest.java index 010a65780c1959..3806e7864c9aba 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/security/modules/JaasModuleTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/security/modules/JaasModuleTest.java @@ -21,7 +21,9 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.CoreOptions; import org.apache.flink.runtime.security.SecurityConfiguration; +import org.apache.flink.util.StringUtils; +import org.hamcrest.core.StringStartsWith; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -29,8 +31,13 @@ import java.io.File; import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Random; import static org.apache.flink.runtime.security.modules.JaasModule.JAVA_SECURITY_AUTH_LOGIN_CONFIG; +import static org.hamcrest.core.StringStartsWith.*; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; /** @@ -59,10 +66,21 @@ public void testJaasModuleFilePathIfWorkingDirNotPresent() throws IOException { testJaasModuleFilePath(file.toPath().toString() + "/tmp"); } + @Test + public void testJaasModuleFilePathIfWorkingDirIsSymLink() throws IOException { + File baseFolder = folder.newFolder(); + File actualFolder = new File(baseFolder, "actual_folder"); + assertTrue(actualFolder.mkdir()); + + Path symlink = new File(baseFolder, "symlink").toPath(); + Files.createSymbolicLink(symlink, actualFolder.toPath()); + testJaasModuleFilePath(symlink.toString()); + } + /** * Test that the jaas config file is created in the working directory. */ - private void testJaasModuleFilePath(String workingDir) { + private void testJaasModuleFilePath(String workingDir) throws IOException { Configuration configuration = new Configuration(); // set the string for CoreOptions.TMP_DIRS to mock the working directory. configuration.setString(CoreOptions.TMP_DIRS, workingDir); @@ -79,7 +97,7 @@ private void testJaasModuleFilePath(String workingDir) { * if we do not manually specify it. */ @Test - public void testCreateJaasModuleFileInTemporary() { + public void testCreateJaasModuleFileInTemporary() throws IOException { Configuration configuration = new Configuration(); SecurityConfiguration sc = new SecurityConfiguration(configuration); JaasModule module = new JaasModule(sc); @@ -89,8 +107,12 @@ public void testCreateJaasModuleFileInTemporary() { assertJaasFileLocateInRightDirectory(CoreOptions.TMP_DIRS.defaultValue()); } - private void assertJaasFileLocateInRightDirectory(String directory) { - assertTrue(System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG).startsWith(directory)); + private void assertJaasFileLocateInRightDirectory(String directory) throws IOException { + String resolvedExpectedPath = new File(directory).toPath().toRealPath().toString(); + String resolvedActualPathWithFile = new File(System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG)).toPath().toRealPath().toString(); + assertThat("The resolved configured directory does not match the expected resolved one.", resolvedActualPathWithFile, startsWith(resolvedExpectedPath)); + + assertThat("The configured directory does not match the expected one.", System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG), startsWith(directory)); } }