diff --git a/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java b/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java index 30cb71f0ad..f0f65f7897 100644 --- a/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java +++ b/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java @@ -42,7 +42,6 @@ import java.nio.channels.ClosedChannelException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -183,6 +182,11 @@ public class ShuffleHandler extends AuxiliaryService { private static final String DATA_FILE_NAME = "file.out"; private static final String INDEX_FILE_NAME = "file.out.index"; + // Whitelist patterns for query-param ids concatenated into filesystem paths. + private static final Pattern DAG_ID_PATTERN = Pattern.compile("[0-9]+"); + private static final Pattern VERTEX_ID_PATTERN = Pattern.compile("[0-9]+"); + private static final Pattern ATTEMPT_ID_PATTERN = Pattern.compile("attempt_[A-Za-z0-9_]+"); + private int port; private NioEventLoopGroup bossGroup; private NioEventLoopGroup workerGroup; @@ -948,6 +952,11 @@ public int weigh(AttemptPathIdentifier key, @Override public AttemptPathInfo load(AttemptPathIdentifier key) throws Exception { + // Backstop against traversal via the "map" param. + if (key.attemptId == null + || !ATTEMPT_ID_PATTERN.matcher(key.attemptId).matches()) { + throw new IOException("Invalid attempt id: " + key.attemptId); + } String base = getBaseLocation(key.jobId, key.dagId, key.user); String attemptBase = base + key.attemptId; Path indexFileName = getAuxiliaryLocalPathHandler() @@ -970,13 +979,19 @@ public void setPort(int port) { this.port = port; } - private List splitMaps(List mapq) { + private List splitMaps(List mapq) throws IOException { if (null == mapq) { return null; } final List ret = new ArrayList<>(); for (String s : mapq) { - Collections.addAll(ret, s.split(",")); + for (String mapId : s.split(",")) { + // Self-defending sink: reject values that could escape the output dir. + if (mapId == null || !ATTEMPT_ID_PATTERN.matcher(mapId).matches()) { + throw new IOException("Invalid mapId: " + mapId); + } + ret.add(mapId); + } } return ret; } @@ -1044,7 +1059,13 @@ private void handleRequest(ChannelHandlerContext ctx, HttpRequest request) keepAliveParam = Boolean.parseBoolean(keepAliveList.get(0)); LOG.debug("KeepAliveParam : {} : {}", keepAliveList, keepAliveParam); } - final List mapIds = splitMaps(q.get("map")); + final List mapIds; + try { + mapIds = splitMaps(q.get("map")); + } catch (IOException e) { + sendError(ctx, e.getMessage(), BAD_REQUEST); + return; + } final Range reduceRange = splitReduces(q.get("reduce")); final List jobQ = q.get("job"); final List dagIdQ = q.get("dag"); @@ -1068,6 +1089,10 @@ private void handleRequest(ChannelHandlerContext ctx, HttpRequest request) sendError(ctx, "Too many job/reduce parameters", BAD_REQUEST); return; } + // Reject traversal-shaped params before any file access. + if (!validateShufflePathParams(ctx, dagIdQ, vertexIdQ)) { + return; + } if (isDeleteRequest) { try { verifyRequest(jobQ.get(0), ctx, request, new DefaultHttpResponse(HTTP_1_1, OK), @@ -1167,6 +1192,40 @@ private boolean isNullOrEmpty(List entries) { return entries == null || entries.isEmpty(); } + /** + * Validate the {@code dag} and {@code vertex} params: each must be a + * plain integer and appear at most once. Duplicates are rejected because + * only the first value is read downstream. Returns false and closes the + * request with a 400 on failure. The {@code map} param is validated in + * {@link #splitMaps(List)}. + */ + private boolean validateShufflePathParams(ChannelHandlerContext ctx, + List dagIdQ, List vertexIdQ) { + if (dagIdQ != null && !dagIdQ.isEmpty()) { + if (dagIdQ.size() > 1) { + sendError(ctx, "Duplicate dag parameter", BAD_REQUEST); + return false; + } + String dagId = dagIdQ.get(0); + if (dagId == null || !DAG_ID_PATTERN.matcher(dagId).matches()) { + sendError(ctx, "Bad dag parameter", BAD_REQUEST); + return false; + } + } + if (vertexIdQ != null && !vertexIdQ.isEmpty()) { + if (vertexIdQ.size() > 1) { + sendError(ctx, "Duplicate vertex parameter", BAD_REQUEST); + return false; + } + String vertexId = vertexIdQ.get(0); + if (vertexId == null || !VERTEX_ID_PATTERN.matcher(vertexId).matches()) { + sendError(ctx, "Bad vertex parameter", BAD_REQUEST); + return false; + } + } + return true; + } + private boolean notEmptyAndContains(List entries, String key) { if (entries == null || entries.isEmpty()) { return false; @@ -1223,6 +1282,13 @@ private boolean deleteTaskAttemptDirectories(Channel channel, List taskA } if (notEmptyAndContains(taskAttemptFailedQ,"delete") && !isNullOrEmpty(taskAttemptIdQ)) { for (String taskAttemptId : taskAttemptIdQ) { + // taskAttemptId is used as a startsWith prefix over the DAG output + // directory listing; keep the accepted form strict. + if (taskAttemptId == null + || !ATTEMPT_ID_PATTERN.matcher(taskAttemptId).matches()) { + LOG.warn("Ignoring taskAttempt delete for invalid attempt id: {}", taskAttemptId); + continue; + } String baseStr = getBaseLocation(jobQ.get(0), dagIdQ.get(0), userRsrc.get(jobQ.get(0))); try { FileSystem fs = FileSystem.getLocal(conf).getRaw(); @@ -1321,6 +1387,12 @@ private String getBaseLocation(String jobId, String dagId, String user) { * Delete shuffle data in task directories belonging to a vertex. */ private void deleteTaskDirsOfVertex(String jobId, String dagId, String vertexId, String user) throws IOException { + // vertexId is spliced into the file-name prefix used to select which + // task attempt directories to delete. Constrain it to digits so the + // prefix cannot expand into an unrelated match. + if (vertexId == null || !VERTEX_ID_PATTERN.matcher(vertexId).matches()) { + throw new IllegalArgumentException("Invalid vertexId: " + vertexId); + } String baseStr = getBaseLocation(jobId, dagId, user); FileContext lfc = FileContext.getLocalFSFileContext(); for(Path dagPath : getAuxiliaryLocalPathHandler().getAllLocalPathsForRead(baseStr)) { @@ -1342,6 +1414,12 @@ private void deleteTaskDirsOfVertex(String jobId, String dagId, String vertexId, private String getDagLocation(String jobId, String dagId, String user) { final JobID jobID = JobID.forName(jobId); + // dagId comes straight from an HTTP query parameter. Reject anything + // that is not a plain integer so a value like "1/../../other" cannot + // escape the appcache/ directory below. + if (dagId == null || !DAG_ID_PATTERN.matcher(dagId).matches()) { + throw new IllegalArgumentException("Invalid dagId: " + dagId); + } final ApplicationId appID = ApplicationId.newInstance(Long.parseLong(jobID.getJtIdentifier()), jobID.getId()); diff --git a/tez-plugins/tez-aux-services/src/test/java/org/apache/tez/auxservices/TestShuffleHandler.java b/tez-plugins/tez-aux-services/src/test/java/org/apache/tez/auxservices/TestShuffleHandler.java index bcb47f9d79..4d7cd23f32 100644 --- a/tez-plugins/tez-aux-services/src/test/java/org/apache/tez/auxservices/TestShuffleHandler.java +++ b/tez-plugins/tez-aux-services/src/test/java/org/apache/tez/auxservices/TestShuffleHandler.java @@ -1751,8 +1751,8 @@ public void testShuffleHandlerSendsDiskError() throws Exception { String shuffleBaseURL = "http://127.0.0.1:" + shuffleHandler.getConfig().get(ShuffleHandler.SHUFFLE_PORT_CONFIG_KEY); URL url = URI.create( - shuffleBaseURL + "/mapOutput?job=job_12345_1&dag=1&reduce=1&map=attempt_12345_1_m_1_0").toURL(); - shuffleHandler.secretManager.addTokenForJob("job_12345_1", + shuffleBaseURL + "/mapOutput?job=job_12345_0001&dag=1&reduce=1&map=attempt_12345_1_m_1_0").toURL(); + shuffleHandler.secretManager.addTokenForJob("job_12345_0001", new Token<>("id".getBytes(), shuffleHandler.getSecret().getBytes(), null, null)); HttpConnectionParams httpConnectionParams = ShuffleUtils.getHttpConnectionParams(conf); @@ -1810,6 +1810,226 @@ public FullHttpRequest createHttpRequest() { return new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, uri); } + /** Traversal-shaped dag/vertex/map params must be rejected. */ + @Test + @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS) + public void testTraversalInDagVertexMapIsRejected() throws Exception { + Configuration conf = getInitialConf(); + conf.setInt(ShuffleHandler.MAX_SHUFFLE_CONNECTIONS, 3); + conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION, + "simple"); + UserGroupInformation.setConfiguration(conf); + conf.set(YarnConfiguration.NM_LOCAL_DIRS, TEST_DIR.getAbsolutePath()); + ApplicationId appId = ApplicationId.newInstance(12345, 1); + String appAttemptId = "attempt_12345_1_m_1_0"; + String user = "randomUser"; + List fileMap = new ArrayList(); + createShuffleHandlerFiles(TEST_DIR, user, appId.toString(), appAttemptId, + conf, fileMap); + ShuffleHandler shuffleHandler = new ShuffleHandler() { + private AuxiliaryLocalPathHandler pathHandler = new TestAuxiliaryLocalPathHandler(); + @Override + protected Shuffle getShuffle(Configuration conf) { + return new Shuffle(conf) { + @Override + protected void verifyRequest(String appid, ChannelHandlerContext ctx, + HttpRequest request, HttpResponse response, URL requestUri) + throws IOException { + // Reject before auth. + } + }; + } + @Override + public AuxiliaryLocalPathHandler getAuxiliaryLocalPathHandler() { + return pathHandler; + } + }; + shuffleHandler.init(conf); + try { + shuffleHandler.start(); + DataOutputBuffer outputBuffer = new DataOutputBuffer(); + outputBuffer.reset(); + Token jt = + new Token("identifier".getBytes(), + "password".getBytes(), new Text(user), new Text("shuffleService")); + jt.write(outputBuffer); + shuffleHandler + .initializeApplication(new ApplicationInitializationContext(user, + appId, ByteBuffer.wrap(outputBuffer.getData(), 0, + outputBuffer.getLength()))); + String base = "http://127.0.0.1:" + + shuffleHandler.getConfig().get(ShuffleHandler.SHUFFLE_PORT_CONFIG_KEY); + + // Must survive every attempt below. + File outside = new File(TEST_DIR, "outside.txt"); + try (FileOutputStream out = new FileOutputStream(outside)) { + out.write("keep me\n".getBytes()); + } + assertTrue(outside.exists()); + + // Traversing dag: must 4xx, must not delete. + String badDag = URI.create("http:///a").resolve( + "?dagAction=delete&job=job_12345_0001&dag=1/../../..").getRawQuery(); + HttpURLConnection conn = (HttpURLConnection) URI.create( + base + "/mapOutput?" + badDag).toURL().openConnection(); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_NAME, + ShuffleHeader.DEFAULT_HTTP_HEADER_NAME); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_VERSION, + ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION); + conn.connect(); + int code = conn.getResponseCode(); + assertTrue(code >= 400 && code < 600, + "Expected an error response for traversing dag, got " + code); + assertTrue(outside.exists(), + "outside.txt must not be deleted by a traversing dag delete"); + + // Traversing vertex: must 4xx. + conn = (HttpURLConnection) URI.create( + base + "/mapOutput?vertexAction=delete&job=job_12345_0001&dag=1&vertex=00/../" + ).toURL().openConnection(); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_NAME, + ShuffleHeader.DEFAULT_HTTP_HEADER_NAME); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_VERSION, + ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION); + conn.connect(); + code = conn.getResponseCode(); + assertTrue(code >= 400 && code < 600, + "Expected an error response for traversing vertex, got " + code); + assertTrue(outside.exists(), + "outside.txt must not be deleted by a traversing vertex delete"); + + // Traversing map: must 4xx. + conn = (HttpURLConnection) URI.create( + base + "/mapOutput?job=job_12345_1&dag=1&reduce=1&map=" + + "attempt_12345_1_m_1_0/../../../etc" + ).toURL().openConnection(); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_NAME, + ShuffleHeader.DEFAULT_HTTP_HEADER_NAME); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_VERSION, + ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION); + conn.connect(); + code = conn.getResponseCode(); + assertTrue(code >= 400 && code < 600, + "Expected an error response for traversing map, got " + code); + + // Reporter's cross-tenant PoC: map= is a traversal path. + conn = (HttpURLConnection) URI.create( + base + "/mapOutput?job=job_12345_1&dag=1&reduce=1&map=" + + "../../../../../../usercache/victim/appcache/" + + "application_9999_0001/dag_1/output/attempt_victim_0001" + ).toURL().openConnection(); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_NAME, + ShuffleHeader.DEFAULT_HTTP_HEADER_NAME); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_VERSION, + ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION); + conn.connect(); + code = conn.getResponseCode(); + assertTrue(code >= 400 && code < 600, + "Expected an error response for reporter's cross-tenant PoC, " + + "got " + code); + + // Comma-joined benign+traversal: whole request must be rejected. + conn = (HttpURLConnection) URI.create( + base + "/mapOutput?job=job_12345_1&dag=1&reduce=1&map=" + + "attempt_12345_1_m_1_0,../../../etc/passwd" + ).toURL().openConnection(); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_NAME, + ShuffleHeader.DEFAULT_HTTP_HEADER_NAME); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_VERSION, + ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION); + conn.connect(); + code = conn.getResponseCode(); + assertTrue(code >= 400 && code < 600, + "Expected an error response for comma-joined benign+traversal " + + "map, got " + code); + } finally { + shuffleHandler.close(); + FileUtil.fullyDelete(TEST_DIR); + } + } + + /** Repeated dag/vertex is ambiguous (only the first is read) — reject it. */ + @Test + @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS) + public void testDuplicateDagOrVertexParamIsRejected() throws Exception { + Configuration conf = getInitialConf(); + conf.setInt(ShuffleHandler.MAX_SHUFFLE_CONNECTIONS, 3); + conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION, + "simple"); + UserGroupInformation.setConfiguration(conf); + conf.set(YarnConfiguration.NM_LOCAL_DIRS, TEST_DIR.getAbsolutePath()); + ApplicationId appId = ApplicationId.newInstance(12345, 1); + String appAttemptId = "attempt_12345_1_m_1_0"; + String user = "randomUser"; + List fileMap = new ArrayList(); + createShuffleHandlerFiles(TEST_DIR, user, appId.toString(), appAttemptId, + conf, fileMap); + ShuffleHandler shuffleHandler = new ShuffleHandler() { + private AuxiliaryLocalPathHandler pathHandler = new TestAuxiliaryLocalPathHandler(); + @Override + protected Shuffle getShuffle(Configuration conf) { + return new Shuffle(conf) { + @Override + protected void verifyRequest(String appid, ChannelHandlerContext ctx, + HttpRequest request, HttpResponse response, URL requestUri) + throws IOException { + // Reject before verifyRequest runs. + } + }; + } + @Override + public AuxiliaryLocalPathHandler getAuxiliaryLocalPathHandler() { + return pathHandler; + } + }; + shuffleHandler.init(conf); + try { + shuffleHandler.start(); + DataOutputBuffer outputBuffer = new DataOutputBuffer(); + outputBuffer.reset(); + Token jt = + new Token("identifier".getBytes(), + "password".getBytes(), new Text(user), new Text("shuffleService")); + jt.write(outputBuffer); + shuffleHandler + .initializeApplication(new ApplicationInitializationContext(user, + appId, ByteBuffer.wrap(outputBuffer.getData(), 0, + outputBuffer.getLength()))); + String base = "http://127.0.0.1:" + + shuffleHandler.getConfig().get(ShuffleHandler.SHUFFLE_PORT_CONFIG_KEY); + + // Duplicate dag: benign first, hostile second. + HttpURLConnection conn = (HttpURLConnection) URI.create( + base + "/mapOutput?dagAction=delete&job=job_12345_0001&dag=1&dag=../evil" + ).toURL().openConnection(); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_NAME, + ShuffleHeader.DEFAULT_HTTP_HEADER_NAME); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_VERSION, + ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION); + conn.connect(); + int code = conn.getResponseCode(); + assertTrue(code >= 400 && code < 600, + "Expected an error response for duplicate dag, got " + code); + + // Duplicate vertex. + conn = (HttpURLConnection) URI.create( + base + "/mapOutput?vertexAction=delete&job=job_12345_0001&dag=1" + + "&vertex=1&vertex=../evil" + ).toURL().openConnection(); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_NAME, + ShuffleHeader.DEFAULT_HTTP_HEADER_NAME); + conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_VERSION, + ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION); + conn.connect(); + code = conn.getResponseCode(); + assertTrue(code >= 400 && code < 600, + "Expected an error response for duplicate vertex, got " + code); + } finally { + shuffleHandler.close(); + FileUtil.fullyDelete(TEST_DIR); + } + } + @Test public void testConfigPortStatic() throws Exception { Random rand = new Random();