Copilot commented on code in PR #11251:
URL: https://github.com/apache/ozone/pull/11251#discussion_r4030484868
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneCryptoInputStream.java:
##########
@@ -73,8 +79,18 @@ public int getBufferSize() {
return bufferSize;
}
+ /**
+ * {@link CryptoInputStream} does not synchronize its own methods, so every
method moving the cursor of
+ * this stream is serialized here on the monitor of this stream. Otherwise a
read or a seek could land in
+ * the middle of a positioned read and see (or undo) the cursor move that
read does.
+ * This covers both the sequential-read API ({@link #read(byte[], int,
int)}, {@link #read(ByteBuffer)},
+ * {@link #seek(long)}, {@link #getPos()}, {@link #skip(long)}) and all
positioned-read overloads
+ * ({@link #read(long, ByteBuffer)}, {@link #readFully(long, ByteBuffer)},
+ * {@link #read(long, byte[], int, int)}, {@link #readFully(long, byte[],
int, int)},
+ * {@link #readFully(long, byte[])}).
+ */
@Override
- public int read(byte[] b, int off, int len) throws IOException {
+ public synchronized int read(byte[] b, int off, int len) throws IOException {
Review Comment:
This synchronizes the byte-array overload, but `CryptoInputStream` also has
its own `read()` implementation that advances the crypto buffers without
delegating through this method. A concurrent one-byte `read()` can therefore
interleave with the seek/read/restore sequence in `read(long, ...)` and corrupt
the cursor or decryptor state. Override `read()` under the same monitor as well
(and cover that path in the concurrency tests).
##########
hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestOzoneCryptoInputStream.java:
##########
@@ -0,0 +1,464 @@
+/*
+ * 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.ozone.client.io;
+
+import static
org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.ReadOnlyBufferException;
+import java.security.SecureRandom;
+import java.util.Arrays;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.crypto.CryptoCodec;
+import org.apache.hadoop.crypto.Encryptor;
+import org.apache.hadoop.fs.PositionedReadable;
+import org.apache.hadoop.fs.Seekable;
+import org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+/**
+ * Unit tests for {@link OzoneCryptoInputStream}.
+ */
+public class TestOzoneCryptoInputStream {
+
+ private static final String KEY_NAME = "testKey";
+ private static final int PART_INDEX = 0;
+ // AES-128 key and IV
+ private static final int KEY_LEN = 16;
+ private static final int IV_LEN = 16;
+
+ private static CryptoCodec codec;
+ private static byte[] key;
+ private static byte[] iv;
+
+ @BeforeAll
+ static void setup() throws Exception {
+ Configuration conf = new Configuration();
+ // Force JCE backend to avoid native library dependency in tests
+ conf.set("hadoop.security.crypto.codec.classes.AES/CTR/NoPadding",
+ "org.apache.hadoop.crypto.JceAesCtrCryptoCodec");
+ codec = CryptoCodec.getInstance(conf);
+ key = new byte[KEY_LEN];
+ iv = new byte[IV_LEN];
+ new SecureRandom().nextBytes(key);
+ new SecureRandom().nextBytes(iv);
+ }
+
+ @Test
+ void testPositionedReadAtStart() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(64 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ ByteBuffer buf = ByteBuffer.allocate(1024);
+ int n = s.read(0, buf);
+ buf.flip();
+ assertArrayEquals(Arrays.copyOf(plaintext, n), toArray(buf));
+ }
+ }
+
+ @Test
+ void testPositionedReadInMiddle() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(64 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ int offset = 12345;
+ ByteBuffer buf = ByteBuffer.allocate(2048);
+ int n = s.read(offset, buf);
+ buf.flip();
+ assertArrayEquals(Arrays.copyOfRange(plaintext, offset, offset + n),
toArray(buf));
+ }
+ }
+
+ @Test
+ void testPositionedReadCrossesBufferBoundary() throws Exception {
+ // The crypto buffer is 8 KB by default; read crossing that boundary
exercises
+ // the position/length adjustment logic in
OzoneCryptoInputStream.read(byte[], int, int).
+ byte[] plaintext = RandomUtils.secure().randomBytes(32 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ // Start 100 bytes before the 8 KB boundary
+ int offset = 8 * 1024 - 100;
+ ByteBuffer buf = ByteBuffer.allocate(1024);
+ int n = s.read(offset, buf);
+ buf.flip();
+ assertArrayEquals(Arrays.copyOfRange(plaintext, offset, offset + n),
toArray(buf));
+ }
+ }
+
+ @Test
+ void testPositionedReadDoesNotMoveSequentialCursor() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(16 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ // Advance the sequential cursor
+ s.seek(500);
+ assertEquals(500, s.getPos());
+
+ // Positioned read at a different offset
+ ByteBuffer buf = ByteBuffer.allocate(256);
+ s.read(8000, buf);
+
+ // Sequential cursor must be restored
+ assertEquals(500, s.getPos());
+ }
+ }
+
+ @Test
+ void testPositionedReadAtEof() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ ByteBuffer buf = ByteBuffer.allocate(64);
+ assertEquals(-1, s.read(plaintext.length, buf),
+ "read at position == length should return EOF");
+ assertEquals(-1, s.read(plaintext.length + 1, buf),
+ "read beyond length should return EOF");
+ }
+ }
+
+ @Test
+ void testPositionedReadEmptyBuffer() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ assertEquals(0, s.read(0, ByteBuffer.allocate(0)));
+ }
+ }
+
+ @Test
+ void testReadFullyBasic() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(64 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ int offset = 1000;
+ int len = 8192;
+ ByteBuffer buf = ByteBuffer.allocate(len);
+ s.readFully(offset, buf);
+ buf.flip();
+ assertArrayEquals(Arrays.copyOfRange(plaintext, offset, offset + len),
toArray(buf));
+ }
+ }
+
+ @Test
+ void testReadFullyDoesNotMoveSequentialCursor() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(32 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ s.seek(1234);
+ ByteBuffer buf = ByteBuffer.allocate(512);
+ s.readFully(16000, buf);
+ assertEquals(1234, s.getPos(), "sequential cursor must be restored after
readFully");
+ }
+ }
+
+ @Test
+ void testByteArrayReadNullBufferThrowsIAE() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ // Aligns with AbstractContractSeekTest.testReadNullBuffer: null must
throw IAE, not NPE.
+ assertThrows(IllegalArgumentException.class, () -> s.read(0, (byte[])
null, 0, 16));
+ }
+ }
+
+ @Test
+ void testByteArrayReadNegativePositionThrowsEOF() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ // Aligns with AbstractContractSeekTest.testReadSmallFile: negative
position must throw.
+ assertThrows(EOFException.class, () -> s.read(-1, new byte[16], 0, 16));
+ }
+ }
+
+ @Test
+ void testByteArrayReadFullyNullBufferThrowsIAE() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ assertThrows(IllegalArgumentException.class,
+ () -> s.readFully(0, (byte[]) null, 0, 16));
+ }
+ }
+
+ @Test
+ void testByteArrayReadFullyNegativePositionThrowsEOF() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ // Aligns with AbstractContractSeekTest.testReadFullySmallFile:
readFully must throw at invalid position.
+ assertThrows(EOFException.class, () -> s.readFully(-1, new byte[16], 0,
16));
+ }
+ }
+
+ @Test
+ void testByteArrayReadAtEofReturnsMinusOne() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ // Position exactly at EOF must return -1, not throw.
+ assertEquals(-1, s.read(plaintext.length, new byte[16], 0, 16));
+ }
+ }
+
+ @Test
+ void testByteBufferReadFullyThrowsEofWhenStreamTooShort() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(100);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ // Request 200 bytes starting at offset 50: 50 + 200 > 100 → EOFException
+ ByteBuffer buf = ByteBuffer.allocate(200);
+ assertThrows(EOFException.class, () -> s.readFully(50, buf));
+ }
+ }
+
+ @Test
+ void testByteArrayReadFullyThrowsEofWhenStreamTooShort() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(100);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ // Aligns with AbstractContractSeekTest.testReadFullySmallFile: partial
buffer must throw.
+ byte[] buf = new byte[200];
+ assertThrows(EOFException.class, () -> s.readFully(50, buf, 0,
buf.length));
+ }
+ }
+
+ @Test
+ void testPositionedReadRejectsReadOnlyBuffer() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(4 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ ByteBuffer readOnly = ByteBuffer.wrap(new byte[256]).asReadOnlyBuffer();
+ assertThrows(ReadOnlyBufferException.class, () -> s.read(0, readOnly));
+ }
+ }
+
+ @Test
+ void testSequentialReadAfterReadOnlyRejection() throws Exception {
+ // A ReadOnlyBufferException on read(long, ByteBuffer) must leave the
stream in
+ // a usable state: adjustment fields reset, cursor restored, next
sequential read
+ // returns correct data.
+ byte[] plaintext = RandomUtils.secure().randomBytes(8 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ ByteBuffer readOnly = ByteBuffer.allocate(256).asReadOnlyBuffer();
+ // Seek to a non-boundary offset so adjustReadPosition sets
readPositionAdjustedBy.
+ s.seek(100);
+ assertThrows(ReadOnlyBufferException.class, () -> s.read(100, readOnly));
+ // Cursor must be restored.
+ assertEquals(100, s.getPos());
+ // Adjustment fields must be reset — sequential read must not throw.
+ byte[] buf = new byte[256];
+ assertDoesNotThrow(() -> s.read(buf, 0, buf.length));
+ assertArrayEquals(Arrays.copyOfRange(plaintext, 100, 356), buf);
+ }
+ }
+
+ @Test
+ void testByteArrayPositionedRead() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(32 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ int offset = 5000;
+ byte[] buf = new byte[2048];
+ int n = s.read(offset, buf, 0, buf.length);
+ assertArrayEquals(Arrays.copyOfRange(plaintext, offset, offset + n),
+ Arrays.copyOf(buf, n));
+ }
+ }
+
+ @Test
+ void testByteArrayReadFully() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(32 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ int offset = 3000;
+ byte[] buf = new byte[4096];
+ s.readFully(offset, buf, 0, buf.length);
+ assertArrayEquals(Arrays.copyOfRange(plaintext, offset, offset +
buf.length), buf);
+ }
+ }
+
+ @Test
+ void testByteArrayReadFullyShortForm() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(32 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ int offset = 1024;
+ byte[] buf = new byte[512];
+ s.readFully(offset, buf);
+ assertArrayEquals(Arrays.copyOfRange(plaintext, offset, offset +
buf.length), buf);
+ }
+ }
+
+ @Test
+ void testByteArrayReadFullyDoesNotMoveSequentialCursor() throws Exception {
+ byte[] plaintext = RandomUtils.secure().randomBytes(16 * 1024);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ s.seek(777);
+ s.readFully(8000, new byte[256]);
+ assertEquals(777, s.getPos(), "sequential cursor must be restored after
readFully(byte[])");
+ }
+ }
+
+ @Test
+ @Timeout(value = 30)
+ void testConcurrentSkipAndPositionedRead() throws Exception {
+ // A concurrent skip must not slip between a positioned read's seek and
+ // restore and corrupt the sequential cursor or the read result.
+ byte[] plaintext = RandomUtils.secure().randomBytes(SOURCE_SIZE);
+ try (OzoneCryptoInputStream s = buildStream(plaintext)) {
+ PositionedReadTestHelper.runConcurrentPositionedReads(plaintext,
(offset, buf) -> {
+ if ((offset & 1) == 0) {
+ s.readFully(offset, buf);
+ } else {
+ // Interleave skip(0) — a no-op skip that still acquires the monitor
—
+ // to exercise the happens-before between skip and positioned reads.
+ s.skip(0);
+ s.readFully(offset, buf);
Review Comment:
`skip(0)` cannot mutate the cursor, and it is invoked immediately before the
same thread's positioned read. This therefore does not exercise a
cursor-changing sequential operation interleaving with the seek/read/restore
critical section; the test would pass even if `skip` were unsynchronized. Use a
concurrent worker/barrier with a non-zero skip or sequential read and assert
the cursor and returned data after the interleaving.
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneCryptoInputStream.java:
##########
@@ -136,6 +152,133 @@ keyName, partIndex, getLength(), numBytesToRead,
return numBytesRead;
}
+ @Override
+ public synchronized int read(ByteBuffer buf) throws IOException {
+ return super.read(buf);
+ }
+
+ @Override
+ public synchronized void seek(long pos) throws IOException {
+ super.seek(pos);
+ }
+
+ @Override
+ public synchronized long getPos() throws IOException {
+ return super.getPos();
+ }
+
+ @Override
+ public synchronized long skip(long n) throws IOException {
+ return super.skip(n);
+ }
+
+ @Override
+ public synchronized boolean seekToNewSource(long targetPos) throws
IOException {
+ return super.seekToNewSource(targetPos);
+ }
+
+ /**
+ * Positioned read. Decryption can only happen at Crypto buffer boundaries,
so this stream cannot read
+ * at an arbitrary position without moving its cursor. The read is
serialized against the other reads:
+ * the cursor is moved to {@code position}, data is read via {@link
#read(byte[], int, int)} (which
+ * handles the Crypto boundary adjustment), and the cursor is restored
before the lock is released.
+ *
+ * <p>Returns -1 for {@code position >= length}. A read-only {@code dst} is
rejected before any state
+ * is mutated. If any exception escapes the read loop the crypto-boundary
adjustment fields are reset so
+ * the next sequential read does not fail the precondition in {@code
getNumBytesToRead}.
+ */
+ @Override
+ public synchronized int read(long position, ByteBuffer dst) throws
IOException {
+ if (!dst.hasRemaining()) {
+ return 0;
+ }
+ if (dst.isReadOnly()) {
+ throw new ReadOnlyBufferException();
+ }
+ if (position < 0 || position >= getLength()) {
+ return EOF;
+ }
Review Comment:
`read(long, ByteBuffer)` currently treats a negative position as EOF, while
the byte-array overloads below explicitly reject the same invalid input with
`EOFException`. This makes the single-read ByteBuffer API silently return `-1`
instead of reporting an invalid position. Split the checks so only positions at
or beyond `getLength()` return EOF and negative positions throw `EOFException`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
