Skip to content

Commit 2767d73

Browse files
OIiveirraOIiveirra
committed
[fix](security) Preserve guarded glob literals and failed Avro open cleanup
1 parent 15cff0f commit 2767d73

6 files changed

Lines changed: 229 additions & 59 deletions

File tree

‎fe/be-java-extensions/avro-scanner/src/main/java/org/apache/doris/avro/AvroJNIScanner.java‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -165,8 +165,21 @@ public void open() throws IOException {
165165
if (!isGetTableSchema) {
166166
initDataReader();
167167
}
168-
this.avroReader.setTvfPolicy(requiredParams.get("fs.doris.tvf.remote.paths"));
169-
this.avroReader.open(avroFileContext, isGetTableSchema);
168+
String tvfPolicy = requiredParams.get("fs.doris.tvf.remote.paths");
169+
this.avroReader.setTvfPolicy(tvfPolicy);
170+
try {
171+
this.avroReader.open(avroFileContext, isGetTableSchema);
172+
} catch (IOException | RuntimeException | Error failure) {
173+
// JNI does not call close when Java open fails.
174+
if (tvfPolicy != null && !tvfPolicy.isEmpty()) {
175+
try {
176+
this.avroReader.close();
177+
} catch (IOException | RuntimeException | Error closeFailure) {
178+
failure.addSuppressed(closeFailure);
179+
}
180+
}
181+
throw failure;
182+
}
170183
}
171184

172185
private void initDataReader() {

‎fe/be-java-extensions/avro-scanner/src/main/java/org/apache/doris/avro/AvroReader.java‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,18 @@ public void close() throws IOException {
8383
protected void openSchemaReader() throws IOException {
8484
TvfGuard.checkJavaRead(fileSystem, path, tvfPolicy);
8585
InputStream inputStream = new BufferedInputStream(fileSystem.open(path));
86-
schemaReader = new DataFileStream<>(inputStream, new GenericDatumReader<>());
86+
try {
87+
schemaReader = new DataFileStream<>(inputStream, new GenericDatumReader<>());
88+
} catch (IOException | RuntimeException | Error failure) {
89+
if (tvfPolicy != null && !tvfPolicy.isEmpty()) {
90+
try {
91+
inputStream.close();
92+
} catch (IOException | RuntimeException | Error closeFailure) {
93+
failure.addSuppressed(closeFailure);
94+
}
95+
}
96+
throw failure;
97+
}
8798
if (LOG.isDebugEnabled()) {
8899
LOG.debug("success open avro schema reader.");
89100
}
@@ -106,7 +117,7 @@ protected void openDataReader(AvroFileContext avroFileContext) throws IOExceptio
106117
} catch (IOException | RuntimeException | Error failure) {
107118
try {
108119
input.close();
109-
} catch (IOException closeFailure) {
120+
} catch (IOException | RuntimeException | Error closeFailure) {
110121
failure.addSuppressed(closeFailure);
111122
}
112123
throw failure;

‎fe/be-java-extensions/avro-scanner/src/test/java/org/apache/doris/avro/AvroViewFsLifecycleTest.java‎

Lines changed: 81 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717

1818
package org.apache.doris.avro;
1919

20+
import org.apache.doris.thrift.TFileType;
21+
2022
import org.apache.avro.Schema;
2123
import org.apache.avro.SchemaBuilder;
2224
import org.apache.avro.file.DataFileWriter;
@@ -43,6 +45,7 @@
4345
import org.mockito.Mockito;
4446

4547
import java.io.File;
48+
import java.io.FileNotFoundException;
4649
import java.io.IOException;
4750
import java.lang.reflect.Field;
4851
import java.lang.reflect.Method;
@@ -53,6 +56,7 @@
5356
import java.nio.file.Files;
5457
import java.util.ArrayList;
5558
import java.util.Collections;
59+
import java.util.HashMap;
5660
import java.util.List;
5761
import java.util.Map;
5862
import java.util.concurrent.atomic.AtomicReference;
@@ -92,13 +96,12 @@ public void failedDataConstructionClosesInputAndOwnedClient() throws Exception {
9296
try (RemoteClients clients = new RemoteClients(Files.readAllBytes(file.toPath()))) {
9397
int baseline = cacheSize();
9498
clients.invalidDataHeader = true;
95-
HDFSFileReader reader = guardedReader(0);
96-
try {
97-
Assert.assertThrows(IOException.class, () -> reader.open(context(file.length(), 0), false));
98-
Assert.assertNull(reader.dataReader);
99-
} finally {
100-
reader.close();
101-
}
99+
AvroJNIScanner scanner = scanner(clients, false, file.length());
100+
Assert.assertThrows(IOException.class, scanner::open);
101+
Field readerField = AvroJNIScanner.class.getDeclaredField("avroReader");
102+
readerField.setAccessible(true);
103+
Assert.assertNull(((AvroReader) readerField.get(scanner)).dataReader);
104+
Assert.assertEquals(2, clients.streams.size());
102105
Assert.assertEquals(1, clients.clients.size());
103106
clients.assertClosed();
104107
Assert.assertEquals(baseline, cacheSize());
@@ -121,6 +124,62 @@ public void closeFailureStillClosesDataReaderAndOwnedClient() throws Exception {
121124
}
122125
}
123126

127+
@Test
128+
public void failedProductionSchemaOpenClosesResourcesWithoutCallerClose() throws Exception {
129+
File file = records();
130+
for (boolean missing : new boolean[] {false, true}) {
131+
try (RemoteClients clients = new RemoteClients(Files.readAllBytes(file.toPath()))) {
132+
int baseline = cacheSize();
133+
clients.missingInput = missing;
134+
clients.truncatedSchema = !missing;
135+
AvroJNIScanner scanner = scanner(clients, true, file.length());
136+
Assert.assertThrows(IOException.class, scanner::open);
137+
Assert.assertEquals(1, clients.clients.size());
138+
Assert.assertEquals(missing ? 0 : 1, clients.streams.size());
139+
// JNI skips Java close after a failed open; no manual scanner.close here.
140+
clients.assertClosed();
141+
Assert.assertEquals(baseline, cacheSize());
142+
}
143+
}
144+
}
145+
146+
@Test
147+
public void productionOpenPreservesFailureAndSuppressesCleanupErrors() throws Exception {
148+
File file = records();
149+
try (RemoteClients clients = new RemoteClients(Files.readAllBytes(file.toPath()))) {
150+
int baseline = cacheSize();
151+
clients.truncatedSchema = true;
152+
clients.failStreamClose = true;
153+
clients.failClientClose = true;
154+
AvroJNIScanner scanner = scanner(clients, true, file.length());
155+
IOException failure = Assert.assertThrows(IOException.class, scanner::open);
156+
Assert.assertFalse(failure.getMessage() != null && failure.getMessage().startsWith("injected"));
157+
Assert.assertEquals(2, failure.getSuppressed().length);
158+
clients.assertClosed();
159+
Assert.assertEquals(baseline, cacheSize());
160+
}
161+
}
162+
163+
private AvroJNIScanner scanner(RemoteClients clients, boolean schemaOnly, long length) throws Exception {
164+
Map<String, String> params = new HashMap<>();
165+
params.put(AvroProperties.IS_GET_TABLE_SCHEMA, Boolean.toString(schemaOnly));
166+
params.put(AvroProperties.FILE_TYPE, Integer.toString(TFileType.FILE_HDFS.getValue()));
167+
params.put(AvroProperties.URI, "viewfs://cluster/data/records.avro");
168+
params.put(TvfGuard.POLICY_KEY, TvfGuard.NATIVE_POLICY);
169+
params.put(AvroProperties.COLUMNS_TYPES, "int");
170+
params.put(AvroProperties.REQUIRED_FIELDS, "id");
171+
params.put(AvroProperties.HIVE_SERDE, "org.apache.hadoop.hive.serde2.avro.AvroSerDe");
172+
params.put(AvroProperties.SPLIT_START_OFFSET, "0");
173+
params.put(AvroProperties.SPLIT_SIZE, Long.toString(length));
174+
params.put(AvroProperties.SPLIT_FILE_SIZE, Long.toString(length));
175+
AvroJNIScanner scanner = new AvroJNIScanner(8, params);
176+
// Give the production scanner its isolated deployment resources, without mocking open/close.
177+
Field loaderField = AvroJNIScanner.class.getDeclaredField("classLoader");
178+
loaderField.setAccessible(true);
179+
loaderField.set(scanner, clients.loader);
180+
return scanner;
181+
}
182+
124183
private HDFSFileReader guardedReader(int index) {
125184
HDFSFileReader reader = new HDFSFileReader("viewfs://cluster/data/records-" + index + ".avro");
126185
reader.setTvfPolicy(TvfGuard.NATIVE_POLICY);
@@ -189,6 +248,10 @@ private final class RemoteClients implements AutoCloseable {
189248
private final Method baseClose;
190249
private final byte[] bytes;
191250
private boolean invalidDataHeader;
251+
private boolean missingInput;
252+
private boolean truncatedSchema;
253+
private boolean failStreamClose;
254+
private boolean failClientClose;
192255

193256
private RemoteClients(byte[] bytes) throws Exception {
194257
this.bytes = bytes;
@@ -230,11 +293,17 @@ private DistributedFileSystem remote(Configuration conf) throws Exception {
230293
Mockito.when(client.getFileStatus(Mockito.any(Path.class))).thenAnswer(call ->
231294
new FileStatus(bytes.length, false, 1, 1, 0, call.getArgument(0)));
232295
Mockito.when(client.open(Mockito.any(Path.class), Mockito.anyInt())).thenAnswer(call -> {
233-
byte[] content = invalidDataHeader && streams.size() % 2 == 1
234-
? new byte[] {0, 0, 0, 0} : bytes;
296+
if (missingInput) {
297+
throw new FileNotFoundException("recording Avro input is missing");
298+
}
299+
byte[] content = truncatedSchema ? new byte[] {'O', 'b', 'j', 1}
300+
: invalidDataHeader && streams.size() % 2 == 1 ? new byte[] {0, 0, 0, 0} : bytes;
235301
SeekableByteArrayInput input = new SeekableByteArrayInput(content);
236302
FSDataInputStream stream = Mockito.mock(FSDataInputStream.class);
237303
streams.add(stream);
304+
if (failStreamClose) {
305+
Mockito.doThrow(new IOException("injected stream close failure")).when(stream).close();
306+
}
238307
Mockito.when(stream.read(Mockito.any(byte[].class), Mockito.anyInt(), Mockito.anyInt()))
239308
.thenAnswer(read -> input.read(read.getArgument(0), read.getArgument(1), read.getArgument(2)));
240309
Mockito.when(stream.read()).thenAnswer(read -> {
@@ -250,6 +319,9 @@ private DistributedFileSystem remote(Configuration conf) throws Exception {
250319
});
251320
Mockito.doAnswer(call -> {
252321
baseClose.invoke(client);
322+
if (failClientClose) {
323+
throw new IOException("injected client close failure");
324+
}
253325
return null;
254326
}).when(client).close();
255327
return client;

‎fe/be-java-extensions/hadoop-deps/src/main/java/org/apache/hadoop/fs/FileSystem.java‎

Lines changed: 48 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -3691,32 +3691,52 @@ public static String encode(List<Rule> rules) {
36913691
}
36923692
}
36933693

3694-
/** Routes touched by Globber, without constructing a filesystem or listing data. */
3694+
/** Literal probes and subtree prefixes touched by Globber, without filesystem IO. */
36953695
public static List<Rule> globRules(URI filesystem, String pattern) throws IOException {
36963696
List<Rule> rules = new ArrayList<>();
36973697
for (String expanded : GlobExpander.expand(pattern)) {
36983698
Path flattened = new Path(expanded.isEmpty() ? Path.CUR_DIR : expanded);
36993699
String path = flattened.toUri().getPath();
3700-
Path candidate = new Path(flattened.isAbsolute() ? "/" : Path.CUR_DIR);
3700+
Path root = new Path(flattened.isAbsolute() ? "/" : Path.CUR_DIR);
3701+
Path candidate = root;
3702+
List<String> components = new ArrayList<>();
3703+
List<Boolean> patterned = new ArrayList<>();
37013704
for (String component : path.split("/")) {
37023705
if (component.isEmpty()) {
37033706
continue;
37043707
}
37053708
GlobFilter filter = new GlobFilter(component);
37063709
if (filter.hasPattern()) {
3707-
// Globber lists this parent before it knows the matching child names. Keep
3708-
// prior wildcard components as a conservative subtree route, including when
3709-
// a later escaped '..' moves the final candidate outside that subtree.
3710-
rules.add(new Rule(filesystem.toString(), new Path(candidate, "*").toUri().getPath(), true));
3710+
// Keep every real listing parent, even if later escaped '..' removes it.
3711+
rules.add(new Rule(filesystem.toString(),
3712+
globLiteralPrefix(root, components, patterned).toUri().getPath(), true));
37113713
}
37123714
component = component.replaceAll("\\\\(.)", "$1");
37133715
candidate = new Path(candidate, component);
3716+
if ("..".equals(component) && !components.isEmpty()
3717+
&& !"..".equals(components.get(components.size() - 1))) {
3718+
components.remove(components.size() - 1);
3719+
patterned.remove(patterned.size() - 1);
3720+
} else if (!".".equals(component)) {
3721+
components.add(component);
3722+
patterned.add(filter.hasPattern());
3723+
}
37143724
}
3715-
rules.add(new Rule(filesystem.toString(), candidate.toUri().getPath(), true));
3725+
boolean subtree = patterned.contains(Boolean.TRUE);
3726+
Path probe = subtree ? globLiteralPrefix(root, components, patterned) : candidate;
3727+
rules.add(new Rule(filesystem.toString(), probe.toUri().getPath(), subtree));
37163728
}
37173729
return rules;
37183730
}
37193731

3732+
private static Path globLiteralPrefix(Path root, List<String> components, List<Boolean> patterned) {
3733+
Path prefix = root;
3734+
for (int i = 0; i < components.size() && !patterned.get(i); i++) {
3735+
prefix = new Path(prefix, components.get(i));
3736+
}
3737+
return prefix;
3738+
}
3739+
37203740
/** Java requests own their mount clients; native connections keep their existing cache. */
37213741
private static final class RequestViewFileSystem extends org.apache.hadoop.fs.viewfs.ViewFileSystem {
37223742
private final Map<List<String>, FileSystem> children = new HashMap<>();
@@ -4040,6 +4060,16 @@ private EffectiveRoute(Map<String, String> properties, boolean descendants, Conf
40404060
this.conf = conf;
40414061
}
40424062

4063+
private EffectiveRoute(EffectiveRoute original, boolean descendants) {
4064+
super(original, descendants);
4065+
conf = original.conf;
4066+
}
4067+
4068+
@Override
4069+
protected Route withDescendants(boolean value) {
4070+
return new EffectiveRoute(this, value);
4071+
}
4072+
40434073
@Override
40444074
protected String property(String key) {
40454075
return conf.get(key);
@@ -4090,6 +4120,14 @@ protected enum Provider {
40904120
private final boolean descendants;
40914121
private final Map<String, String> properties;
40924122

4123+
protected Route(Route original, boolean descendants) {
4124+
this(original.properties, descendants);
4125+
}
4126+
4127+
protected Route withDescendants(boolean value) {
4128+
return new Route(this, value);
4129+
}
4130+
40934131
public Route(Map<String, String> properties, boolean descendants) {
40944132
// Both callers supply a statement-local snapshot that is never mutated afterward.
40954133
this.properties = properties;
@@ -4138,7 +4176,7 @@ public void checkRoute(URI filesystem, String path, Set<String> activeTables) th
41384176
public void checkGlobRoute(URI filesystem, String pattern, Set<String> activeTables)
41394177
throws IOException, URISyntaxException {
41404178
for (Rule rule : globRules(filesystem, pattern)) {
4141-
checkRoute(filesystem, rule.path, activeTables);
4179+
withDescendants(rule.descendants).checkRoute(filesystem, rule.path, activeTables);
41424180
}
41434181
}
41444182

@@ -4190,7 +4228,7 @@ private void checkGlobTargets(String key, String prefix, Set<String> activeTable
41904228
// Listing can touch any descendant of a selected mount. Reuse the same route
41914229
// interpreter for nested tables instead of assuming their root is representative.
41924230
String path = uri.getPath() == null ? "/" : uri.getPath();
4193-
checkRoute(uri, (path.endsWith("/") ? path : path + "/") + "*", activeTables);
4231+
checkRoute(uri, path, activeTables);
41944232
}
41954233
}
41964234

@@ -4271,26 +4309,7 @@ private void checkViewFsEnabled(URI filesystem, String path, Set<String> activeT
42714309
selectedKeys.add(key);
42724310
}
42734311
}
4274-
if (descendants && hasGlob(path)) {
4275-
String literal = path.substring(0, firstGlob(path));
4276-
int slash = literal.lastIndexOf('/');
4277-
literal = slash < 0 ? "/" : literal.substring(0, slash + 1);
4278-
for (String key : properties.keySet()) {
4279-
if (!key.startsWith(prefix)) {
4280-
continue;
4281-
}
4282-
String entry = key.substring(prefix.length());
4283-
int sourceStart = entry.indexOf('/');
4284-
if (sourceStart >= 0 && (entry.startsWith("link.") || entry.startsWith("linkMerge.")
4285-
|| entry.startsWith("linkNfly."))) {
4286-
String mount = entry.substring(sourceStart);
4287-
if (mount.startsWith(literal) || literal.startsWith(mount + "/")) {
4288-
checkGlobTargets(key, prefix, activeTables);
4289-
}
4290-
}
4291-
}
4292-
}
4293-
if (descendants && !hasGlob(path)) {
4312+
if (descendants) {
42944313
String subtree = path.endsWith("/") ? path : path + "/";
42954314
for (String key : properties.keySet()) {
42964315
if (key.startsWith(prefix)) {
@@ -4363,21 +4382,6 @@ private String selector(String value) {
43634382
return value;
43644383
}
43654384

4366-
private static int firstGlob(String path) {
4367-
int first = path.length();
4368-
for (char marker : new char[] {'*', '?', '[', '{'}) {
4369-
int index = path.indexOf(marker);
4370-
if (index >= 0) {
4371-
first = Math.min(first, index);
4372-
}
4373-
}
4374-
return first;
4375-
}
4376-
4377-
private static boolean hasGlob(String path) {
4378-
return firstGlob(path) < path.length();
4379-
}
4380-
43814385
protected void reject() {
43824386
throw new IllegalArgumentException("Hadoop TVF requires a proven remote filesystem route");
43834387
}

‎fe/fe-core/src/test/java/org/apache/doris/tablefunction/HadoopTvfAdmissionTest.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -125,9 +125,12 @@ void readPatternsCheckEveryIntersectingMountBeforeListing() {
125125
Map<String, String> properties = viewFs();
126126
properties.put("fs.viewfs.mounttable.cluster.link./data", "hdfs://remote/base");
127127
properties.put("fs.viewfs.mounttable.cluster.link./data/local", "file:///tmp");
128-
for (String pattern : new String[] {"/data/*", "/data/?/file.csv", "/data/{a,b}/file.csv", "/data"}) {
129-
Assertions.assertThrows(AnalysisException.class, () -> HadoopTvfAdmission.checkReadDiscovery("viewfs://cluster" + pattern, properties));
128+
for (String pattern : new String[] {"/data/*", "/data/?/file.csv", "/data/{a,b}/file.csv"}) {
129+
Assertions.assertThrows(AnalysisException.class,
130+
() -> HadoopTvfAdmission.checkReadDiscovery("viewfs://cluster" + pattern, properties), pattern);
130131
}
132+
// This branch uses Globber: a literal path only probes its own FileStatus.
133+
Assertions.assertDoesNotThrow(() -> HadoopTvfAdmission.checkReadDiscovery("viewfs://cluster/data", properties));
131134
Assertions.assertDoesNotThrow(() -> HadoopTvfAdmission.checkReadDiscovery("viewfs://cluster/data/remote/file.csv", properties));
132135
properties.put("fs.viewfs.mounttable.cluster.link./data/local", "hdfs://remote/other");
133136
Assertions.assertDoesNotThrow(() -> HadoopTvfAdmission.checkReadDiscovery("viewfs://cluster/data/*", properties));

0 commit comments

Comments
 (0)