, org.apache.hadoop.io.serial.RawComparator {
+ @Override
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2);
}
diff --git a/src/java/org/apache/hadoop/io/SecureIOUtils.java b/src/java/org/apache/hadoop/io/SecureIOUtils.java
new file mode 100644
index 0000000000000..321b0e89db66e
--- /dev/null
+++ b/src/java/org/apache/hadoop/io/SecureIOUtils.java
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.io;
+
+import java.io.File;
+import java.io.FileDescriptor;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.nativeio.Errno;
+import org.apache.hadoop.io.nativeio.NativeIO;
+import org.apache.hadoop.io.nativeio.NativeIOException;
+import org.apache.hadoop.io.nativeio.NativeIO.Stat;
+import org.apache.hadoop.security.UserGroupInformation;
+
+/**
+ * This class provides secure APIs for opening and creating files on the local
+ * disk. The main issue this class tries to handle is that of symlink traversal.
+ *
+ * An example of such an attack is:
+ *
+ * - Malicious user removes his task's syslog file, and puts a link to the
+ * jobToken file of a target user.
+ * - Malicious user tries to open the syslog file via the servlet on the
+ * tasktracker.
+ * - The tasktracker is unaware of the symlink, and simply streams the contents
+ * of the jobToken file. The malicious user can now access potentially sensitive
+ * map outputs, etc. of the target user's job.
+ *
+ * A similar attack is possible involving task log truncation, but in that case
+ * due to an insecure write to a file.
+ *
+ */
+public class SecureIOUtils {
+
+ /**
+ * Ensure that we are set up to run with the appropriate native support code.
+ * If security is disabled, and the support code is unavailable, this class
+ * still tries its best to be secure, but is vulnerable to some race condition
+ * attacks.
+ *
+ * If security is enabled but the support code is unavailable, throws a
+ * RuntimeException since we don't want to run insecurely.
+ */
+ static {
+ boolean shouldBeSecure = UserGroupInformation.isSecurityEnabled();
+ boolean canBeSecure = NativeIO.isAvailable();
+
+ if (!canBeSecure && shouldBeSecure) {
+ throw new RuntimeException(
+ "Secure IO is not possible without native code extensions.");
+ }
+
+ // Pre-cache an instance of the raw FileSystem since we sometimes
+ // do secure IO in a shutdown hook, where this call could fail.
+ try {
+ rawFilesystem = FileSystem.getLocal(new Configuration()).getRaw();
+ } catch (IOException ie) {
+ throw new RuntimeException(
+ "Couldn't obtain an instance of RawLocalFileSystem.");
+ }
+
+ // SecureIO just skips security checks in the case that security is
+ // disabled
+ skipSecurity = !canBeSecure;
+ }
+
+ private final static boolean skipSecurity;
+ private final static FileSystem rawFilesystem;
+
+ /**
+ * Open the given File for read access, verifying the expected user/group
+ * constraints.
+ * @param f the file that we are trying to open
+ * @param expectedOwner the expected user owner for the file
+ * @param expectedGroup the expected group owner for the file
+ * @throws IOException if an IO Error occurred, or the user/group does not
+ * match
+ */
+ public static FileInputStream openForRead(File f, String expectedOwner,
+ String expectedGroup) throws IOException {
+ if (skipSecurity) {
+ // Subject to race conditions but this is the best we can do
+ FileStatus status =
+ rawFilesystem.getFileStatus(new Path(f.getAbsolutePath()));
+ checkStat(f, status.getOwner(), status.getGroup(),
+ expectedOwner, expectedGroup);
+ return new FileInputStream(f);
+ }
+
+ FileInputStream fis = new FileInputStream(f);
+ boolean success = false;
+ try {
+ Stat stat = NativeIO.fstat(fis.getFD());
+ checkStat(f, stat.getOwner(), stat.getGroup(), expectedOwner,
+ expectedGroup);
+ success = true;
+ return fis;
+ } finally {
+ if (!success) {
+ fis.close();
+ }
+ }
+ }
+
+ private static FileOutputStream insecureCreateForWrite(File f,
+ int permissions) throws IOException {
+ // If we can't do real security, do a racy exists check followed by an
+ // open and chmod
+ if (f.exists()) {
+ throw new AlreadyExistsException("File " + f + " already exists");
+ }
+ FileOutputStream fos = new FileOutputStream(f);
+ boolean success = false;
+ try {
+ rawFilesystem.setPermission(new Path(f.getAbsolutePath()),
+ new FsPermission((short)permissions));
+ success = true;
+ return fos;
+ } finally {
+ if (!success) {
+ fos.close();
+ }
+ }
+ }
+
+ /**
+ * Open the specified File for write access, ensuring that it does not exist.
+ * @param f the file that we want to create
+ * @param permissions we want to have on the file (if security is enabled)
+ *
+ * @throws AlreadyExistsException if the file already exists
+ * @throws IOException if any other error occurred
+ */
+ public static FileOutputStream createForWrite(File f, int permissions)
+ throws IOException {
+ if (skipSecurity) {
+ return insecureCreateForWrite(f, permissions);
+ } else {
+ // Use the native wrapper around open(2)
+ try {
+ FileDescriptor fd = NativeIO.open(f.getAbsolutePath(),
+ NativeIO.O_WRONLY | NativeIO.O_CREAT | NativeIO.O_EXCL,
+ permissions);
+ return new FileOutputStream(fd);
+ } catch (NativeIOException nioe) {
+ if (nioe.getErrno() == Errno.EEXIST) {
+ throw new AlreadyExistsException(nioe);
+ }
+ throw nioe;
+ }
+ }
+ }
+
+ private static void checkStat(File f, String owner, String group,
+ String expectedOwner,
+ String expectedGroup) throws IOException {
+ if (expectedOwner != null &&
+ !expectedOwner.equals(owner)) {
+ throw new IOException(
+ "Owner '" + owner + "' for path " + f + " did not match " +
+ "expected owner '" + expectedOwner + "'");
+ }
+ if (expectedGroup != null &&
+ !expectedGroup.equals(group)) {
+ throw new IOException(
+ "Group '" + group + "' for path " + f + " did not match " +
+ "expected group '" + expectedGroup + "'");
+ }
+ }
+
+ /**
+ * Signals that an attempt to create a file at a given pathname has failed
+ * because another file already existed at that path.
+ */
+ public static class AlreadyExistsException extends IOException {
+ private static final long serialVersionUID = 1L;
+
+ public AlreadyExistsException(String msg) {
+ super(msg);
+ }
+
+ public AlreadyExistsException(Throwable cause) {
+ super(cause);
+ }
+ }
+}
diff --git a/src/java/org/apache/hadoop/io/SequenceFile.java b/src/java/org/apache/hadoop/io/SequenceFile.java
index b3f1630dd4952..91aed3257d84f 100644
--- a/src/java/org/apache/hadoop/io/SequenceFile.java
+++ b/src/java/org/apache/hadoop/io/SequenceFile.java
@@ -34,12 +34,14 @@
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.io.compress.zlib.ZlibFactory;
-import org.apache.hadoop.io.serializer.Deserializer;
-import org.apache.hadoop.io.serializer.Serializer;
-import org.apache.hadoop.io.serializer.SerializationFactory;
+import org.apache.hadoop.io.serial.RawComparator;
+import org.apache.hadoop.io.serial.Serialization;
+import org.apache.hadoop.io.serial.SerializationFactory;
+import org.apache.hadoop.io.serial.TypedSerialization;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.conf.*;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.Progress;
import org.apache.hadoop.util.ReflectionUtils;
@@ -93,10 +95,16 @@
* version number (e.g. SEQ4 or SEQ6)
*
*
- * keyClassName -key class
+ * key serialization name
*
*
- * valueClassName - value class
+ * key serialization configuration
+ *
+ *
+ * value serialization name
+ *
+ *
+ * value serialization data
*
*
* compression - A boolean which specifies if compression is turned on for
@@ -134,7 +142,7 @@
*
*
*
- * A sync-marker every few 100 bytes or so.
+ * A sync-marker every 2000 bytes or so.
*
*
*
@@ -153,7 +161,7 @@
*
*
*
- * A sync-marker every few 100 bytes or so.
+ * A sync-marker every 2000 bytes or so.
*
*
*
@@ -165,6 +173,7 @@
*
* Record Block
*
+ * - sync-marker
* - Compressed key-lengths block-size
* - Compressed key-lengths block
* - Compressed keys block-size
@@ -175,9 +184,6 @@
* - Compressed values block
*
*
- *
- * A sync-marker every few 100 bytes or so.
- *
*
*
* The compressed blocks of key lengths and value lengths consist of the
@@ -196,8 +202,9 @@ private SequenceFile() {} // no public ctor
private static final byte BLOCK_COMPRESS_VERSION = (byte)4;
private static final byte CUSTOM_COMPRESS_VERSION = (byte)5;
private static final byte VERSION_WITH_METADATA = (byte)6;
+ private static final byte SERIALIZATION_VERSION = (byte) 7;
private static byte[] VERSION = new byte[] {
- (byte)'S', (byte)'E', (byte)'Q', VERSION_WITH_METADATA
+ (byte)'S', (byte)'E', (byte)'Q', SERIALIZATION_VERSION
};
private static final int SYNC_ESCAPE = -1; // "length" of sync entries
@@ -285,6 +292,7 @@ public static Writer createWriter(Configuration conf, Writer.Option... opts
* instead.
*/
@Deprecated
+ @SuppressWarnings("unchecked")
public static Writer
createWriter(FileSystem fs, Configuration conf, Path name,
Class keyClass, Class valClass) throws IOException {
@@ -306,6 +314,7 @@ public static Writer createWriter(Configuration conf, Writer.Option... opts
* instead.
*/
@Deprecated
+ @SuppressWarnings("unchecked")
public static Writer
createWriter(FileSystem fs, Configuration conf, Path name,
Class keyClass, Class valClass,
@@ -330,6 +339,7 @@ public static Writer createWriter(Configuration conf, Writer.Option... opts
* instead.
*/
@Deprecated
+ @SuppressWarnings("unchecked")
public static Writer
createWriter(FileSystem fs, Configuration conf, Path name,
Class keyClass, Class valClass, CompressionType compressionType,
@@ -355,6 +365,7 @@ public static Writer createWriter(Configuration conf, Writer.Option... opts
* instead.
*/
@Deprecated
+ @SuppressWarnings("unchecked")
public static Writer
createWriter(FileSystem fs, Configuration conf, Path name,
Class keyClass, Class valClass, CompressionType compressionType,
@@ -381,6 +392,7 @@ public static Writer createWriter(Configuration conf, Writer.Option... opts
* instead.
*/
@Deprecated
+ @SuppressWarnings("unchecked")
public static Writer
createWriter(FileSystem fs, Configuration conf, Path name,
Class keyClass, Class valClass,
@@ -413,6 +425,7 @@ public static Writer createWriter(Configuration conf, Writer.Option... opts
* instead.
*/
@Deprecated
+ @SuppressWarnings("unchecked")
public static Writer
createWriter(FileSystem fs, Configuration conf, Path name,
Class keyClass, Class valClass, int bufferSize,
@@ -444,6 +457,7 @@ public static Writer createWriter(Configuration conf, Writer.Option... opts
* @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
* instead.
*/
+ @SuppressWarnings("unchecked")
@Deprecated
public static Writer
createWriter(FileSystem fs, Configuration conf, Path name,
@@ -471,6 +485,7 @@ public static Writer createWriter(Configuration conf, Writer.Option... opts
* instead.
*/
@Deprecated
+ @SuppressWarnings("unchecked")
public static Writer
createWriter(Configuration conf, FSDataOutputStream out,
Class keyClass, Class valClass,
@@ -495,6 +510,7 @@ public static Writer createWriter(Configuration conf, Writer.Option... opts
* @deprecated Use {@link #createWriter(Configuration, Writer.Option...)}
* instead.
*/
+ @SuppressWarnings("unchecked")
@Deprecated
public static Writer
createWriter(Configuration conf, FSDataOutputStream out,
@@ -527,18 +543,44 @@ public void writeCompressedBytes(DataOutputStream outStream)
* Size of stored data.
*/
public int getSize();
+
}
+
+ /**
+ * Make an InputStream from a ValueBytes.
+ * @param bytes the bytes to provide as input
+ * @return a new input stream with the bytes
+ * @throws IOException
+ */
+ private static InputStream readUncompressedBytes(ValueBytes bytes
+ ) throws IOException {
+ DataInputBuffer result = new DataInputBuffer();
+ if (bytes instanceof UncompressedBytes) {
+ MutableValueBytes concrete = (MutableValueBytes) bytes;
+ result.reset(concrete.data, concrete.dataSize);
+ } else {
+ DataOutputBuffer outBuf = new DataOutputBuffer();
+ bytes.writeUncompressedBytes(outBuf);
+ result.reset(outBuf.getData(), outBuf.getLength());
+ }
+ return result;
+ }
+
- private static class UncompressedBytes implements ValueBytes {
- private int dataSize;
- private byte[] data;
-
- private UncompressedBytes() {
+ private static abstract class MutableValueBytes implements ValueBytes {
+ protected byte[] data;
+ protected int dataSize;
+
+ MutableValueBytes() {
data = null;
dataSize = 0;
}
+
+ public int getSize() {
+ return dataSize;
+ }
- private void reset(DataInputStream in, int length) throws IOException {
+ void reset(DataInputStream in, int length) throws IOException {
if (data == null) {
data = new byte[length];
} else if (length > data.length) {
@@ -548,10 +590,14 @@ private void reset(DataInputStream in, int length) throws IOException {
in.readFully(data, 0, length);
dataSize = length;
}
-
- public int getSize() {
- return dataSize;
+
+ void set(MutableValueBytes other) {
+ data = other.data;
+ dataSize = other.dataSize;
}
+ }
+
+ private static class UncompressedBytes extends MutableValueBytes {
public void writeUncompressedBytes(DataOutputStream outStream)
throws IOException {
@@ -566,34 +612,15 @@ public void writeCompressedBytes(DataOutputStream outStream)
} // UncompressedBytes
- private static class CompressedBytes implements ValueBytes {
- private int dataSize;
- private byte[] data;
+ private static class CompressedBytes extends MutableValueBytes {
DataInputBuffer rawData = null;
CompressionCodec codec = null;
CompressionInputStream decompressedStream = null;
private CompressedBytes(CompressionCodec codec) {
- data = null;
- dataSize = 0;
this.codec = codec;
}
- private void reset(DataInputStream in, int length) throws IOException {
- if (data == null) {
- data = new byte[length];
- } else if (length > data.length) {
- data = new byte[Math.max(length, data.length * 2)];
- }
- dataSize = -1;
- in.readFully(data, 0, length);
- dataSize = length;
- }
-
- public int getSize() {
- return dataSize;
- }
-
public void writeUncompressedBytes(DataOutputStream outStream)
throws IOException {
if (decompressedStream == null) {
@@ -738,9 +765,6 @@ public static class Writer implements java.io.Closeable {
boolean ownOutputStream = true;
DataOutputBuffer buffer = new DataOutputBuffer();
- Class keyClass;
- Class valClass;
-
private final CompressionType compress;
CompressionCodec codec = null;
CompressionOutputStream deflateFilter = null;
@@ -748,9 +772,8 @@ public static class Writer implements java.io.Closeable {
Metadata metadata = null;
Compressor compressor = null;
- protected Serializer keySerializer;
- protected Serializer uncompressedValSerializer;
- protected Serializer compressedValSerializer;
+ protected Serialization
* @param
+ * @deprecated Use
+ * {@link org.apache.hadoop.io.serial.Serialization}
+ * instead.
*/
@InterfaceAudience.LimitedPrivate({"HDFS", "MapReduce"})
@InterfaceStability.Evolving
+@Deprecated
public interface Serializer {
/**
* Prepare the serializer for writing.
diff --git a/src/java/org/apache/hadoop/io/serializer/WritableSerialization.java b/src/java/org/apache/hadoop/io/serializer/WritableSerialization.java
index 8511d25bcde65..30e773c0e7729 100644
--- a/src/java/org/apache/hadoop/io/serializer/WritableSerialization.java
+++ b/src/java/org/apache/hadoop/io/serializer/WritableSerialization.java
@@ -23,7 +23,6 @@
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.util.Map;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
@@ -36,9 +35,13 @@
* A {@link Serialization} for {@link Writable}s that delegates to
* {@link Writable#write(java.io.DataOutput)} and
* {@link Writable#readFields(java.io.DataInput)}.
+ * @deprecated Use
+ * {@link org.apache.hadoop.io.serial.lib.WritableSerialization}
+ * instead.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
+@Deprecated
public class WritableSerialization extends Configured
implements Serialization {
static class WritableDeserializer extends Configured
diff --git a/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerializable.java b/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerializable.java
index 4a748223e5fc5..49fdf6e2810a9 100644
--- a/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerializable.java
+++ b/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerializable.java
@@ -25,9 +25,13 @@
* Tag interface for Avro 'reflect' serializable classes. Classes implementing
* this interface can be serialized/deserialized using
* {@link AvroReflectSerialization}.
+ * @deprecated Use {@link org.apache.hadoop.io.serial.lib.avro.AvroReflectSerializable}
+ * instead.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
-public interface AvroReflectSerializable {
+@Deprecated
+public interface AvroReflectSerializable
+ extends org.apache.hadoop.io.serial.lib.avro.AvroReflectSerializable {
}
diff --git a/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerialization.java b/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerialization.java
index 5636b597e33a4..2c24144e203d7 100644
--- a/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerialization.java
+++ b/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerialization.java
@@ -35,11 +35,13 @@
* serialization, it must either be in the package list configured via
* avro.reflect.pkgs or implement
* {@link AvroReflectSerializable} interface.
- *
+ * @deprecated Use {@link org.apache.hadoop.io.serial.lib.avro.AvroSerialization}
+ * instead.
*/
@SuppressWarnings("unchecked")
@InterfaceAudience.Public
@InterfaceStability.Evolving
+@Deprecated
public class AvroReflectSerialization extends AvroSerialization