Skip to content

TEZ-4753: ShuffleHandler: input handling improvements #536

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
abstractdog wants to merge 1 commit into
base: master
Choose a base branch
from
+304 −6
Open

TEZ-4753: ShuffleHandler: input handling improvements #536

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 82 additions & 4 deletions tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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()
Expand All @@ -970,13 +979,19 @@ public void setPort(int port) {
this.port = port;
}

private List<String> splitMaps(List<String> mapq) {
private List<String> splitMaps(List<String> mapq) throws IOException {
if (null == mapq) {
return null;
}
final List<String> 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;
}
Expand Down Expand Up @@ -1044,7 +1059,13 @@ private void handleRequest(ChannelHandlerContext ctx, HttpRequest request)
keepAliveParam = Boolean.parseBoolean(keepAliveList.get(0));
LOG.debug("KeepAliveParam : {} : {}", keepAliveList, keepAliveParam);
}
final List<String> mapIds = splitMaps(q.get("map"));
final List<String> 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<String> jobQ = q.get("job");
final List<String> dagIdQ = q.get("dag");
Expand All @@ -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),
Expand Down Expand Up @@ -1167,6 +1192,40 @@ private boolean isNullOrEmpty(List<String> 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<String> dagIdQ, List<String> 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<String> entries, String key) {
if (entries == null || entries.isEmpty()) {
return false;
Expand Down Expand Up @@ -1223,6 +1282,13 @@ private boolean deleteTaskAttemptDirectories(Channel channel, List<String> 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();
Expand Down Expand Up @@ -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)) {
Expand All @@ -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/<appId> 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());
Expand Down
224 changes: 222 additions & 2 deletions ...plugins/tez-aux-services/src/test/java/org/apache/tez/auxservices/TestShuffleHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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<File> fileMap = new ArrayList<File>();
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<JobTokenIdentifier> jt =
new Token<JobTokenIdentifier>("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<File> fileMap = new ArrayList<File>();
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<JobTokenIdentifier> jt =
new Token<JobTokenIdentifier>("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();
Expand Down
Loading