From e5cb36fd462360de8cb85459373b6d95a53efde9 Mon Sep 17 00:00:00 2001 From: Laszlo Bodor Date: Mon, 21 Sep 2026 11:47:41 +0200 Subject: [PATCH] TEZ-4753: ShuffleHandler: input handling improvements Tighten validation of the dag, vertex and map query parameters in the shuffle handler so unexpected shapes are rejected up front rather than carried into filesystem path construction. Also reject requests where dag or vertex is present more than once, since downstream code only reads the first occurrence and a permissive validator would let a submitter pair a benign first value with a hostile second one. Adds regression tests. Co-Authored-By: Claude Code --- .../tez/auxservices/ShuffleHandler.java | 86 ++++++- .../tez/auxservices/TestShuffleHandler.java | 224 +++++++++++++++++- 2 files changed, 304 insertions(+), 6 deletions(-) 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();