Packet-based enc/dec cipher streams (#49896)

This adds a new bare snapshot repository project which contains the classes implementing encryption (and decryption) input stream decorators that support mark and reset.

Relates #48221 , #46170

Author: albertzaharovits

Diff for: x-pack/plugin/repository-encrypted/build.gradle

@@ -0,0 +1,11 @@
+evaluationDependsOn(xpackModule('core'))
+
+apply plugin: 'elasticsearch.esplugin'
+esplugin {
+    name 'repository-encrypted'
+    description 'Elasticsearch Expanded Pack Plugin - client-side encrypted repositories.'
+    classname 'org.elasticsearch.repositories.encrypted.EncryptedRepositoryPlugin'
+    extendedPlugins = ['x-pack-core']
+}
+
+integTest.enabled = false

Diff for: x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/BufferOnMarkInputStream.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.repositories.encrypted;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Objects;
+
+/**
+ * A {@code CountingInputStream} wraps another input stream and counts the number of bytes
+ * that have been read or skipped.
+ * <p>
+ * This input stream does no buffering on its own and only supports {@code mark} and
+ * {@code reset} if the underlying wrapped stream supports it.
+ * <p>
+ * If the stream supports {@code mark} and {@code reset} the byte count is also reset to the
+ * value that it had on the last {@code mark} call, thereby not counting the same bytes twice.
+ * <p>
+ * If the {@code closeSource} constructor argument is {@code true}, closing this
+ * stream will also close the wrapped input stream. Apart from closing the wrapped
+ * stream in this case, the {@code close} method does nothing else.
+ */
+public final class CountingInputStream extends InputStream {
+
+    private final InputStream source;
+    private final boolean closeSource;
+    long count; // package-protected for tests
+    long mark; // package-protected for tests
+    boolean closed; // package-protected for tests
+
+    /**
+     * Wraps another input stream, counting the number of bytes read.
+     *
+     * @param source the input stream to be wrapped
+     * @param closeSource {@code true} if closing this stream will also close the wrapped stream
+     */
+    public CountingInputStream(InputStream source, boolean closeSource) {
+        this.source = Objects.requireNonNull(source);
+        this.closeSource = closeSource;
+        this.count = 0L;
+        this.mark = -1L;
+        this.closed = false;
+    }
+
+    /** Returns the number of bytes read. */
+    public long getCount() {
+        return count;
+    }
+
+    @Override
+    public int read() throws IOException {
+        int result = source.read();
+        if (result != -1) {
+            count++;
+        }
+        return result;
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+        int result = source.read(b, off, len);
+        if (result != -1) {
+            count += result;
+        }
+        return result;
+    }
+
+    @Override
+    public long skip(long n) throws IOException {
+        long result = source.skip(n);
+        count += result;
+        return result;
+    }
+
+    @Override
+    public int available() throws IOException {
+        return source.available();
+    }
+
+    @Override
+    public boolean markSupported() {
+        return source.markSupported();
+    }
+
+    @Override
+    public synchronized void mark(int readlimit) {
+        source.mark(readlimit);
+        mark = count;
+    }
+
+    @Override
+    public synchronized void reset() throws IOException {
+        if (false == source.markSupported()) {
+            throw new IOException("Mark not supported");
+        }
+        if (mark == -1L) {
+            throw new IOException("Mark not set");
+        }
+        count = mark;
+        source.reset();
+    }
+
+    @Override
+    public void close() throws IOException {
+        if (false == closed) {
+            closed = true;
+            if (closeSource) {
+                source.close();
+            }
+        }
+    }
+}

Diff for: x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/ChainingInputStream.java

@@ -0,0 +1,173 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.repositories.encrypted;
+
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+import javax.crypto.NoSuchPaddingException;
+import javax.crypto.SecretKey;
+import javax.crypto.ShortBufferException;
+import javax.crypto.spec.GCMParameterSpec;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.security.InvalidAlgorithmParameterException;
+import java.security.InvalidKeyException;
+import java.security.NoSuchAlgorithmException;
+import java.util.Objects;
+
+import static org.elasticsearch.repositories.encrypted.EncryptedRepository.GCM_IV_LENGTH_IN_BYTES;
+import static org.elasticsearch.repositories.encrypted.EncryptedRepository.GCM_TAG_LENGTH_IN_BYTES;
+
+/**
+ * A {@code DecryptionPacketsInputStream} wraps an encrypted input stream and decrypts
+ * its contents. This is designed (and tested) to decrypt only the encryption format that
+ * {@link EncryptionPacketsInputStream} generates. No decrypted bytes are returned before
+ * they are authenticated.
+ * <p>
+ * The same parameters, namely {@code secretKey}, {@code nonce} and {@code packetLength},
+ * that have been used during encryption must also be used for decryption, otherwise
+ * decryption will fail.
+ * <p>
+ * This implementation buffers the encrypted packet in memory. The maximum packet size it can
+ * accommodate is {@link EncryptedRepository#MAX_PACKET_LENGTH_IN_BYTES}.
+ * <p>
+ * This implementation does not support {@code mark} and {@code reset}.
+ * <p>
+ * The {@code close} call will close the decryption input stream and any subsequent {@code read},
+ * {@code skip}, {@code available} and {@code reset} calls will throw {@code IOException}s.
+ * <p>
+ * This is NOT thread-safe, multiple threads sharing a single instance must synchronize access.
+ *
+ * @see EncryptionPacketsInputStream
+ */
+public final class DecryptionPacketsInputStream extends ChainingInputStream {
+
+    private final InputStream source;
+    private final SecretKey secretKey;
+    private final int nonce;
+    private final int packetLength;
+    private final byte[] packetBuffer;
+
+    private boolean hasNext;
+    private long counter;
+
+    /**
+     * Computes and returns the length of the plaintext given the {@code ciphertextLength} and the {@code packetLength}
+     * used during encryption.
+     * Each ciphertext packet is prepended by the Initilization Vector and has the Authentication Tag appended.
+     * Decryption is 1:1, and the ciphertext is not padded, but stripping away the IV and the AT amounts to a shorter
+     * plaintext compared to the ciphertext.
+     *
+     * @see EncryptionPacketsInputStream#getEncryptionLength(long, int)
+     */
+    public static long getDecryptionLength(long ciphertextLength, int packetLength) {
+        long encryptedPacketLength = packetLength + GCM_TAG_LENGTH_IN_BYTES + GCM_IV_LENGTH_IN_BYTES;
+        long completePackets = ciphertextLength / encryptedPacketLength;
+        long decryptedSize = completePackets * packetLength;
+        if (ciphertextLength % encryptedPacketLength != 0) {
+            decryptedSize += (ciphertextLength % encryptedPacketLength) - GCM_IV_LENGTH_IN_BYTES - GCM_TAG_LENGTH_IN_BYTES;
+        }
+        return decryptedSize;
+    }
+
+    public DecryptionPacketsInputStream(InputStream source, SecretKey secretKey, int nonce, int packetLength) {
+        this.source = Objects.requireNonNull(source);
+        this.secretKey = Objects.requireNonNull(secretKey);
+        this.nonce = nonce;
+        if (packetLength <= 0 || packetLength >= EncryptedRepository.MAX_PACKET_LENGTH_IN_BYTES) {
+            throw new IllegalArgumentException("Invalid packet length [" + packetLength + "]");
+        }
+        this.packetLength = packetLength;
+        this.packetBuffer = new byte[packetLength + GCM_TAG_LENGTH_IN_BYTES];
+        this.hasNext = true;
+        this.counter = EncryptedRepository.PACKET_START_COUNTER;
+    }
+
+    @Override
+    InputStream nextComponent(InputStream currentComponentIn) throws IOException {
+        if (currentComponentIn != null && currentComponentIn.read() != -1) {
+            throw new IllegalStateException("Stream for previous packet has not been fully processed");
+        }
+        if (false == hasNext) {
+            return null;
+        }
+        PrefixInputStream packetInputStream = new PrefixInputStream(source,
+                packetLength + GCM_IV_LENGTH_IN_BYTES + GCM_TAG_LENGTH_IN_BYTES,
+                false);
+        int currentPacketLength = decrypt(packetInputStream);
+        // only the last packet is shorter, so this must be the last packet
+        if (currentPacketLength != packetLength) {
+            hasNext = false;
+        }
+        return new ByteArrayInputStream(packetBuffer, 0, currentPacketLength);
+    }
+
+    @Override
+    public boolean markSupported() {
+        return false;
+    }
+
+    @Override
+    public void mark(int readlimit) {
+    }
+
+    @Override
+    public void reset() throws IOException {
+        throw new IOException("Mark/reset not supported");
+    }
+
+    private int decrypt(PrefixInputStream packetInputStream) throws IOException {
+        // read only the IV prefix into the packet buffer
+        int ivLength = packetInputStream.readNBytes(packetBuffer, 0, GCM_IV_LENGTH_IN_BYTES);
+        if (ivLength != GCM_IV_LENGTH_IN_BYTES) {
+            throw new IOException("Packet heading IV error. Unexpected length [" + ivLength + "].");
+        }
+        // extract the nonce and the counter from the packet IV
+        ByteBuffer ivBuffer = ByteBuffer.wrap(packetBuffer, 0, GCM_IV_LENGTH_IN_BYTES).order(ByteOrder.LITTLE_ENDIAN);
+        int packetIvNonce = ivBuffer.getInt(0);
+        long packetIvCounter = ivBuffer.getLong(Integer.BYTES);
+        if (packetIvNonce != nonce) {
+            throw new IOException("Packet nonce mismatch. Expecting [" + nonce + "], but got [" + packetIvNonce + "].");
+        }
+        if (packetIvCounter != counter) {
+            throw new IOException("Packet counter mismatch. Expecting [" + counter + "], but got [" + packetIvCounter + "].");
+        }
+        // counter increment for the subsequent packet
+        counter++;
+        // counter wrap around
+        if (counter == EncryptedRepository.PACKET_START_COUNTER) {
+            throw new IOException("Maximum packet count limit exceeded");
+        }
+        // cipher used to decrypt only the current packetInputStream
+        Cipher packetCipher = getPacketDecryptionCipher(packetBuffer);
+        // read the rest of the packet, reusing the packetBuffer
+        int packetLength = packetInputStream.readNBytes(packetBuffer, 0, packetBuffer.length);
+        if (packetLength < GCM_TAG_LENGTH_IN_BYTES) {
+            throw new IOException("Encrypted packet is too short");
+        }
+        try {
+            // in-place decryption of the whole packet and return decrypted length
+            return packetCipher.doFinal(packetBuffer, 0, packetLength, packetBuffer);
+        } catch (ShortBufferException | IllegalBlockSizeException | BadPaddingException e) {
+            throw new IOException("Exception during packet decryption", e);
+        }
+    }
+
+    private Cipher getPacketDecryptionCipher(byte[] packet) throws IOException {
+        GCMParameterSpec gcmParameterSpec = new GCMParameterSpec(GCM_TAG_LENGTH_IN_BYTES * Byte.SIZE, packet, 0, GCM_IV_LENGTH_IN_BYTES);
+        try {
+            Cipher packetCipher = Cipher.getInstance(EncryptedRepository.GCM_ENCRYPTION_SCHEME);
+            packetCipher.init(Cipher.DECRYPT_MODE, secretKey, gcmParameterSpec);
+            return packetCipher;
+        } catch (NoSuchAlgorithmException | NoSuchPaddingException | InvalidKeyException | InvalidAlgorithmParameterException e) {
+            throw new IOException("Exception during packet cipher initialisation", e);
+        }
+    }
+}

Diff for: x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/CountingInputStream.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.repositories.encrypted;
+
+public class EncryptedRepository {
+    static final int GCM_TAG_LENGTH_IN_BYTES = 16;
+    static final int GCM_IV_LENGTH_IN_BYTES = 12;
+    static final int AES_BLOCK_SIZE_IN_BYTES = 128;
+    static final String GCM_ENCRYPTION_SCHEME = "AES/GCM/NoPadding";
+    static final long PACKET_START_COUNTER = Long.MIN_VALUE;
+    static final int MAX_PACKET_LENGTH_IN_BYTES = 1 << 30;
+}

Diff for: x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/DecryptionPacketsInputStream.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.repositories.encrypted;
+
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.plugins.ReloadablePlugin;
+import org.elasticsearch.plugins.RepositoryPlugin;
+
+import java.util.List;
+
+public final class EncryptedRepositoryPlugin extends Plugin implements RepositoryPlugin, ReloadablePlugin {
+
+    public EncryptedRepositoryPlugin(final Settings settings) {
+    }
+
+    @Override
+    public List<Setting<?>> getSettings() {
+        return List.of();
+    }
+
+    @Override
+    public void reload(Settings settings) {
+        // Secure settings should be readable inside this method.
+    }
+}

Diff for: x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/EncryptedRepository.java

@@ -0,0 +1,185 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.repositories.encrypted;
+
+import javax.crypto.Cipher;
+import javax.crypto.CipherInputStream;
+import javax.crypto.NoSuchPaddingException;
+import javax.crypto.SecretKey;
+import javax.crypto.spec.GCMParameterSpec;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.security.InvalidAlgorithmParameterException;
+import java.security.InvalidKeyException;
+import java.security.NoSuchAlgorithmException;
+import java.util.Objects;
+
+/**
+ * An {@code EncryptionPacketsInputStream} wraps another input stream and encrypts its contents.
+ * The method of encryption is AES/GCM/NoPadding, which is a type of authenticated encryption.
+ * The encryption works packet wise, i.e. the stream is segmented into fixed-size byte packets
+ * which are separately encrypted using a unique {@link Cipher}. As an exception, only the last
+ * packet will have a different size, possibly zero. Note that the encrypted packets are
+ * larger compared to the plaintext packets, because they contain a 16 byte length trailing
+ * authentication tag. The resulting encrypted and authenticated packets are assembled back into
+ * the resulting stream.
+ * <p>
+ * The packets are encrypted using the same {@link SecretKey} but using a different initialization
+ * vector. The IV of each packet is 12 bytes wide and is comprised of a 4-byte integer {@code nonce},
+ * the same for every packet in the stream, and a monotonically increasing 8-byte integer counter.
+ * The caller must assure that the same {@code nonce} is not reused for other encrypted streams
+ * using the same {@code secretKey}. The counter from the IV identifies the position of the packet
+ * in the encrypted stream, so that packets cannot be reordered without breaking the decryption.
+ * When assembling the encrypted stream, the IV is prepended to the corresponding packet's ciphertext.
+ * <p>
+ * The packet length is preferably a large multiple (typically 128) of the AES block size (128 bytes),
+ * but any positive integer value smaller than {@link EncryptedRepository#MAX_PACKET_LENGTH_IN_BYTES}
+ * is valid. A larger packet length incurs smaller relative size overhead because the 12 byte wide IV
+ * and the 16 byte wide authentication tag are constant no matter the packet length. A larger packet
+ * length also exposes more opportunities for the JIT compilation of the AES encryption loop. But
+ * {@code mark} will buffer up to packet length bytes, and, more importantly, <b>decryption</b> might
+ * need to allocate a memory buffer the size of the packet in order to assure that no un-authenticated
+ * decrypted ciphertext is returned. The decryption procedure is the primary factor that limits the
+ * packet length.
+ * <p>
+ * This input stream supports the {@code mark} and {@code reset} operations, but only if the wrapped
+ * stream supports them as well. A {@code mark} call will trigger the memory buffering of the current
+ * packet and will also trigger a {@code mark} call on the wrapped input stream on the next
+ * packet boundary. Upon a {@code reset} call, the buffered packet will be replayed and new packets
+ * will be generated starting from the marked packet boundary on the wrapped stream.
+ * <p>
+ * The {@code close} call will close the encryption input stream and any subsequent {@code read},
+ * {@code skip}, {@code available} and {@code reset} calls will throw {@code IOException}s.
+ * <p>
+ * This is NOT thread-safe, multiple threads sharing a single instance must synchronize access.
+ *
+ * @see DecryptionPacketsInputStream
+ */
+public final class EncryptionPacketsInputStream extends ChainingInputStream {
+
+    private final SecretKey secretKey;
+    private final int packetLength;
+    private final ByteBuffer packetIv;
+    private final int encryptedPacketLength;
+
+    final InputStream source; // package-protected for tests
+    long counter; // package-protected for tests
+    Long markCounter; // package-protected for tests
+    int markSourceOnNextPacket; // package-protected for tests
+
+    /**
+     * Computes and returns the length of the ciphertext given the {@code plaintextLength} and the {@code packetLength}
+     * used during encryption.
+     * The plaintext is segmented into packets of equal {@code packetLength} length, with the exception of the last
+     * packet which is shorter and can have a length of {@code 0}. Encryption is packet-wise and is 1:1, with no padding.
+     * But each encrypted packet is prepended by the Initilization Vector and appended the Authentication Tag, including
+     * the last packet, so when pieced together will amount to a longer resulting ciphertext.
+     *
+     * @see DecryptionPacketsInputStream#getDecryptionLength(long, int)
+     */
+    public static long getEncryptionLength(long plaintextLength, int packetLength) {
+        return plaintextLength + (plaintextLength / packetLength + 1) * (EncryptedRepository.GCM_TAG_LENGTH_IN_BYTES
+                + EncryptedRepository.GCM_IV_LENGTH_IN_BYTES);
+    }
+
+    public EncryptionPacketsInputStream(InputStream source, SecretKey secretKey, int nonce, int packetLength) {
+        this.source = Objects.requireNonNull(source);
+        this.secretKey = Objects.requireNonNull(secretKey);
+        if (packetLength <= 0 || packetLength >= EncryptedRepository.MAX_PACKET_LENGTH_IN_BYTES) {
+            throw new IllegalArgumentException("Invalid packet length [" + packetLength + "]");
+        }
+        this.packetLength = packetLength;
+        this.packetIv = ByteBuffer.allocate(EncryptedRepository.GCM_IV_LENGTH_IN_BYTES).order(ByteOrder.LITTLE_ENDIAN);
+        // nonce takes the first 4 bytes of the IV
+        this.packetIv.putInt(0, nonce);
+        this.encryptedPacketLength =
+                packetLength + EncryptedRepository.GCM_IV_LENGTH_IN_BYTES + EncryptedRepository.GCM_TAG_LENGTH_IN_BYTES;
+        this.counter = EncryptedRepository.PACKET_START_COUNTER;
+        this.markCounter = null;
+        this.markSourceOnNextPacket = -1;
+    }
+
+    @Override
+    InputStream nextComponent(InputStream currentComponentIn) throws IOException {
+        // the last packet input stream is the only one shorter than encryptedPacketLength
+        if (currentComponentIn != null && ((CountingInputStream) currentComponentIn).getCount() < encryptedPacketLength) {
+            // there are no more packets
+            return null;
+        }
+        // If the enclosing stream has a mark set,
+        // then apply it to the source input stream when we reach a packet boundary
+        if (markSourceOnNextPacket != -1) {
+            source.mark(markSourceOnNextPacket);
+            markSourceOnNextPacket = -1;
+        }
+        // create the new packet
+        InputStream encryptionInputStream = new PrefixInputStream(source, packetLength, false);
+        // the counter takes up the last 8 bytes of the packet IV (12 byte wide)
+        // the first 4 bytes are used by the nonce (which is the same for every packet IV)
+        packetIv.putLong(Integer.BYTES, counter++);
+        // counter wrap around
+        if (counter == EncryptedRepository.PACKET_START_COUNTER) {
+            throw new IOException("Maximum packet count limit exceeded");
+        }
+        Cipher packetCipher = getPacketEncryptionCipher(secretKey, packetIv.array());
+        encryptionInputStream = new CipherInputStream(encryptionInputStream, packetCipher);
+        encryptionInputStream = new SequenceInputStream(new ByteArrayInputStream(packetIv.array()), encryptionInputStream);
+        encryptionInputStream = new BufferOnMarkInputStream(encryptionInputStream, encryptedPacketLength);
+        return new CountingInputStream(encryptionInputStream, false);
+    }
+
+    @Override
+    public boolean markSupported() {
+        return source.markSupported();
+    }
+
+    @Override
+    public void mark(int readlimit) {
+        if (markSupported()) {
+            if (readlimit <= 0) {
+                throw new IllegalArgumentException("Mark readlimit must be a positive integer");
+            }
+            // handles the packet-wise part of the marking operation
+            super.mark(encryptedPacketLength);
+            // saves the counter used to generate packet IVs
+            markCounter = counter;
+            // stores the flag used to mark the source input stream at packet boundary
+            markSourceOnNextPacket = readlimit;
+        }
+    }
+
+    @Override
+    public void reset() throws IOException {
+        if (false == markSupported()) {
+            throw new IOException("Mark/reset not supported");
+        }
+        if (markCounter == null) {
+            throw new IOException("Mark no set");
+        }
+        super.reset();
+        counter = markCounter;
+        if (markSourceOnNextPacket == -1) {
+            source.reset();
+        }
+    }
+
+    private static Cipher getPacketEncryptionCipher(SecretKey secretKey, byte[] packetIv) throws IOException {
+        GCMParameterSpec gcmParameterSpec = new GCMParameterSpec(EncryptedRepository.GCM_TAG_LENGTH_IN_BYTES * Byte.SIZE, packetIv);
+        try {
+            Cipher packetCipher = Cipher.getInstance(EncryptedRepository.GCM_ENCRYPTION_SCHEME);
+            packetCipher.init(Cipher.ENCRYPT_MODE, secretKey, gcmParameterSpec);
+            return packetCipher;
+        } catch (NoSuchAlgorithmException | NoSuchPaddingException | InvalidKeyException | InvalidAlgorithmParameterException e) {
+            throw new IOException(e);
+        }
+    }
+
+}

Diff for: x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/EncryptedRepositoryPlugin.java

@@ -0,0 +1,150 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.repositories.encrypted;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Objects;
+
+/**
+ * A {@code PrefixInputStream} wraps another input stream and exposes
+ * only the first bytes of it. Reading from the wrapping
+ * {@code PrefixInputStream} consumes the underlying stream. The stream
+ * is exhausted when {@code prefixLength} bytes have been read, or the underlying
+ * stream is exhausted before that.
+ * <p>
+ * Only if the {@code closeSource} constructor argument is {@code true}, the
+ * closing of this stream will also close the underlying input stream.
+ * Any subsequent {@code read}, {@code skip} and {@code available} calls
+ * will throw {@code IOException}s.
+ */
+public final class PrefixInputStream extends InputStream {
+
+    /**
+     * The underlying stream of which only a prefix is returned
+     */
+    private final InputStream source;
+    /**
+     * The length in bytes of the prefix.
+     * This is the maximum number of bytes that can be read from this stream,
+     * but fewer bytes can be read if the wrapped source stream itself contains fewer bytes
+     */
+    private final int prefixLength;
+    /**
+     * The current count of bytes read from this stream.
+     * This starts of as {@code 0} and is always smaller or equal to {@code prefixLength}.
+     */
+    private int count;
+    /**
+     * whether closing this stream must also close the underlying stream
+     */
+    private boolean closeSource;
+    /**
+     * flag signalling if this stream has been closed
+     */
+    private boolean closed;
+
+    public PrefixInputStream(InputStream source, int prefixLength, boolean closeSource) {
+        if (prefixLength < 0) {
+            throw new IllegalArgumentException("The prefixLength constructor argument must be a positive integer");
+        }
+        this.source = source;
+        this.prefixLength = prefixLength;
+        this.count = 0;
+        this.closeSource = closeSource;
+        this.closed = false;
+    }
+
+    @Override
+    public int read() throws IOException {
+        ensureOpen();
+        if (remainingPrefixByteCount() <= 0) {
+            return -1;
+        }
+        int byteVal = source.read();
+        if (byteVal == -1) {
+            return -1;
+        }
+        count++;
+        return byteVal;
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+        ensureOpen();
+        Objects.checkFromIndexSize(off, len, b.length);
+        if (len == 0) {
+            return 0;
+        }
+        if (remainingPrefixByteCount() <= 0) {
+            return -1;
+        }
+        int readSize = Math.min(len, remainingPrefixByteCount());
+        int bytesRead = source.read(b, off, readSize);
+        if (bytesRead == -1) {
+            return -1;
+        }
+        count += bytesRead;
+        return bytesRead;
+    }
+
+    @Override
+    public long skip(long n) throws IOException {
+        ensureOpen();
+        if (n <= 0 || remainingPrefixByteCount() <= 0) {
+            return 0;
+        }
+        long bytesToSkip = Math.min(n, remainingPrefixByteCount());
+        assert bytesToSkip > 0;
+        long bytesSkipped = source.skip(bytesToSkip);
+        count += bytesSkipped;
+        return bytesSkipped;
+    }
+
+    @Override
+    public int available() throws IOException {
+        ensureOpen();
+        return Math.min(remainingPrefixByteCount(), source.available());
+    }
+
+    @Override
+    public boolean markSupported() {
+        return false;
+    }
+
+    @Override
+    public void mark(int readlimit) {
+        // mark and reset are not supported
+    }
+
+    @Override
+    public void reset() throws IOException {
+        throw new IOException("mark/reset not supported");
+    }
+
+    @Override
+    public void close() throws IOException {
+        if (closed) {
+            return;
+        }
+        closed = true;
+        if (closeSource) {
+            source.close();
+        }
+    }
+
+    private int remainingPrefixByteCount() {
+        return prefixLength - count;
+    }
+
+    private void ensureOpen() throws IOException {
+        if (closed) {
+            throw new IOException("Stream has been closed");
+        }
+    }
+
+}

Diff for: x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/EncryptionPacketsInputStream.java

@@ -0,0 +1,8 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+grant {
+};

Diff for: x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/PrefixInputStream.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.repositories.encrypted;
+
+import org.elasticsearch.common.Randomness;
+import org.elasticsearch.test.ESTestCase;
+import org.hamcrest.Matchers;
+import org.junit.BeforeClass;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class CountingInputStreamTests extends ESTestCase {
+
+    private static byte[] testArray;
+
+    @BeforeClass
+    static void createTestArray() throws Exception {
+        testArray = new byte[32];
+        for (int i = 0; i < testArray.length; i++) {
+            testArray[i] = (byte) i;
+        }
+    }
+
+    public void testWrappedMarkAndClose() throws Exception {
+        AtomicBoolean isClosed = new AtomicBoolean(false);
+        InputStream mockIn = mock(InputStream.class);
+        doAnswer(new Answer<Void>() {
+            public Void answer(InvocationOnMock invocation) {
+                isClosed.set(true);
+                return null;
+            }
+        }).when(mockIn).close();
+        new CountingInputStream(mockIn, true).close();
+        assertThat(isClosed.get(), Matchers.is(true));
+        isClosed.set(false);
+        new CountingInputStream(mockIn, false).close();
+        assertThat(isClosed.get(), Matchers.is(false));
+        when(mockIn.markSupported()).thenAnswer(invocationOnMock -> {
+            return false;
+        });
+        assertThat(new CountingInputStream(mockIn, randomBoolean()).markSupported(), Matchers.is(false));
+        when(mockIn.markSupported()).thenAnswer(invocationOnMock -> {
+            return true;
+        });
+        assertThat(new CountingInputStream(mockIn, randomBoolean()).markSupported(), Matchers.is(true));
+    }
+
+    public void testSimpleCountForRead() throws Exception {
+        CountingInputStream test = new CountingInputStream(new ByteArrayInputStream(testArray), randomBoolean());
+        assertThat(test.getCount(), Matchers.is(0L));
+        int readLen = Randomness.get().nextInt(testArray.length);
+        test.readNBytes(readLen);
+        assertThat(test.getCount(), Matchers.is((long)readLen));
+        readLen = testArray.length - readLen;
+        test.readNBytes(readLen);
+        assertThat(test.getCount(), Matchers.is((long)testArray.length));
+        test.close();
+        assertThat(test.getCount(), Matchers.is((long)testArray.length));
+    }
+
+    public void testSimpleCountForSkip() throws Exception {
+        CountingInputStream test = new CountingInputStream(new ByteArrayInputStream(testArray), randomBoolean());
+        assertThat(test.getCount(), Matchers.is(0L));
+        int skipLen = Randomness.get().nextInt(testArray.length);
+        test.skip(skipLen);
+        assertThat(test.getCount(), Matchers.is((long)skipLen));
+        skipLen = testArray.length - skipLen;
+        test.readNBytes(skipLen);
+        assertThat(test.getCount(), Matchers.is((long)testArray.length));
+        test.close();
+        assertThat(test.getCount(), Matchers.is((long)testArray.length));
+    }
+
+    public void testCountingForMarkAndReset() throws Exception {
+        CountingInputStream test = new CountingInputStream(new ByteArrayInputStream(testArray), randomBoolean());
+        assertThat(test.getCount(), Matchers.is(0L));
+        assertThat(test.markSupported(), Matchers.is(true));
+        int offset1 = Randomness.get().nextInt(testArray.length - 1);
+        if (randomBoolean()) {
+            test.skip(offset1);
+        } else {
+            test.read(new byte[offset1]);
+        }
+        assertThat(test.getCount(), Matchers.is((long)offset1));
+        test.mark(testArray.length);
+        int offset2 = 1 + Randomness.get().nextInt(testArray.length - offset1 - 1);
+        if (randomBoolean()) {
+            test.skip(offset2);
+        } else {
+            test.read(new byte[offset2]);
+        }
+        assertThat(test.getCount(), Matchers.is((long)offset1 + offset2));
+        test.reset();
+        assertThat(test.getCount(), Matchers.is((long)offset1));
+        int offset3 = Randomness.get().nextInt(offset2);
+        if (randomBoolean()) {
+            test.skip(offset3);
+        } else {
+            test.read(new byte[offset3]);
+        }
+        assertThat(test.getCount(), Matchers.is((long)offset1 + offset3));
+        test.reset();
+        assertThat(test.getCount(), Matchers.is((long)offset1));
+        test.readAllBytes();
+        assertThat(test.getCount(), Matchers.is((long)testArray.length));
+        test.close();
+        assertThat(test.getCount(), Matchers.is((long)testArray.length));
+    }
+
+    public void testCountingForMarkAfterReset() throws Exception {
+        CountingInputStream test = new CountingInputStream(new ByteArrayInputStream(testArray), randomBoolean());
+        assertThat(test.getCount(), Matchers.is(0L));
+        assertThat(test.markSupported(), Matchers.is(true));
+        int offset1 = Randomness.get().nextInt(testArray.length - 1);
+        if (randomBoolean()) {
+            test.skip(offset1);
+        } else {
+            test.read(new byte[offset1]);
+        }
+        assertThat(test.getCount(), Matchers.is((long)offset1));
+        test.mark(testArray.length);
+        int offset2 = 1 + Randomness.get().nextInt(testArray.length - offset1 - 1);
+        if (randomBoolean()) {
+            test.skip(offset2);
+        } else {
+            test.read(new byte[offset2]);
+        }
+        assertThat(test.getCount(), Matchers.is((long)offset1 + offset2));
+        test.reset();
+        assertThat(test.getCount(), Matchers.is((long)offset1));
+        int offset3 = Randomness.get().nextInt(offset2);
+        if (randomBoolean()) {
+            test.skip(offset3);
+        } else {
+            test.read(new byte[offset3]);
+        }
+        test.mark(testArray.length);
+        assertThat(test.getCount(), Matchers.is((long)offset1 + offset3));
+        int offset4 = Randomness.get().nextInt(testArray.length - offset1 - offset3);
+        if (randomBoolean()) {
+            test.skip(offset4);
+        } else {
+            test.read(new byte[offset4]);
+        }
+        assertThat(test.getCount(), Matchers.is((long)offset1 + offset3 + offset4));
+        test.reset();
+        assertThat(test.getCount(), Matchers.is((long)offset1 + offset3));
+        test.readAllBytes();
+        assertThat(test.getCount(), Matchers.is((long)testArray.length));
+        test.close();
+        assertThat(test.getCount(), Matchers.is((long)testArray.length));
+    }
+
+}

Diff for: x-pack/plugin/repository-encrypted/src/main/plugin-metadata/plugin-security.policy

@@ -0,0 +1,219 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.repositories.encrypted;
+
+import org.elasticsearch.common.Randomness;
+import org.elasticsearch.test.ESTestCase;
+import org.hamcrest.Matchers;
+
+import javax.crypto.KeyGenerator;
+import javax.crypto.SecretKey;
+import java.io.ByteArrayInputStream;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.SecureRandom;
+import java.util.Arrays;
+
+public class DecryptionPacketsInputStreamTests extends ESTestCase {
+
+    public void testSuccessEncryptAndDecryptSmallPacketLength() throws Exception {
+        int len = 8 + Randomness.get().nextInt(8);
+        byte[] plainBytes = new byte[len];
+        Randomness.get().nextBytes(plainBytes);
+        SecretKey secretKey = generateSecretKey();
+        int nonce = Randomness.get().nextInt();
+        for (int packetLen : Arrays.asList(1, 2, 3, 4)) {
+            testEncryptAndDecryptSuccess(plainBytes, secretKey, nonce, packetLen);
+        }
+    }
+
+    public void testSuccessEncryptAndDecryptLargePacketLength() throws Exception {
+        int len = 256 + Randomness.get().nextInt(256);
+        byte[] plainBytes = new byte[len];
+        Randomness.get().nextBytes(plainBytes);
+        SecretKey secretKey = generateSecretKey();
+        int nonce = Randomness.get().nextInt();
+        for (int packetLen : Arrays.asList(len - 1, len - 2, len - 3, len - 4)) {
+            testEncryptAndDecryptSuccess(plainBytes, secretKey, nonce, packetLen);
+        }
+    }
+
+    public void testSuccessEncryptAndDecryptTypicalPacketLength() throws Exception {
+        int len = 1024 + Randomness.get().nextInt(512);
+        byte[] plainBytes = new byte[len];
+        Randomness.get().nextBytes(plainBytes);
+        SecretKey secretKey = generateSecretKey();
+        int nonce = Randomness.get().nextInt();
+        for (int packetLen : Arrays.asList(128, 256, 512)) {
+            testEncryptAndDecryptSuccess(plainBytes, secretKey, nonce, packetLen);
+        }
+    }
+
+    public void testFailureEncryptAndDecryptWrongNonce() throws Exception {
+        int len = 256 + Randomness.get().nextInt(256);
+        // 2-3 packets
+        int packetLen = 1 + Randomness.get().nextInt(len / 2);
+        byte[] plainBytes = new byte[len];
+        Randomness.get().nextBytes(plainBytes);
+        SecretKey secretKey = generateSecretKey();
+        int encryptNonce = Randomness.get().nextInt();
+        int decryptNonce = Randomness.get().nextInt();
+        while (decryptNonce == encryptNonce) {
+            decryptNonce = Randomness.get().nextInt();
+        }
+        byte[] encryptedBytes;
+        try (InputStream in = new EncryptionPacketsInputStream(new ByteArrayInputStream(plainBytes, 0, len), secretKey, encryptNonce,
+                packetLen)) {
+            encryptedBytes = in.readAllBytes();
+        }
+        try (InputStream in = new DecryptionPacketsInputStream(new ByteArrayInputStream(encryptedBytes), secretKey, decryptNonce,
+                packetLen)) {
+            IOException e = expectThrows(IOException.class, () -> {
+                in.readAllBytes();
+            });
+            assertThat(e.getMessage(), Matchers.startsWith("Packet nonce mismatch."));
+        }
+    }
+
+    public void testFailureEncryptAndDecryptWrongKey() throws Exception {
+        int len = 256 + Randomness.get().nextInt(256);
+        // 2-3 packets
+        int packetLen = 1 + Randomness.get().nextInt(len / 2);
+        byte[] plainBytes = new byte[len];
+        Randomness.get().nextBytes(plainBytes);
+        SecretKey encryptSecretKey = generateSecretKey();
+        SecretKey decryptSecretKey = generateSecretKey();
+        int nonce = Randomness.get().nextInt();
+        byte[] encryptedBytes;
+        try (InputStream in = new EncryptionPacketsInputStream(new ByteArrayInputStream(plainBytes, 0, len), encryptSecretKey, nonce,
+                packetLen)) {
+            encryptedBytes = in.readAllBytes();
+        }
+        try (InputStream in = new DecryptionPacketsInputStream(new ByteArrayInputStream(encryptedBytes), decryptSecretKey, nonce,
+                packetLen)) {
+            IOException e = expectThrows(IOException.class, () -> {
+                in.readAllBytes();
+            });
+            assertThat(e.getMessage(), Matchers.is("Exception during packet decryption"));
+        }
+    }
+
+    public void testFailureEncryptAndDecryptAlteredCiphertext() throws Exception {
+        int len = 8 + Randomness.get().nextInt(8);
+        // one packet
+        int packetLen = len + Randomness.get().nextInt(8);
+        byte[] plainBytes = new byte[len];
+        Randomness.get().nextBytes(plainBytes);
+        SecretKey secretKey = generateSecretKey();
+        int nonce = Randomness.get().nextInt();
+        byte[] encryptedBytes;
+        try (InputStream in = new EncryptionPacketsInputStream(new ByteArrayInputStream(plainBytes, 0, len), secretKey, nonce,
+                packetLen)) {
+            encryptedBytes = in.readAllBytes();
+        }
+        for (int i = EncryptedRepository.GCM_IV_LENGTH_IN_BYTES; i < EncryptedRepository.GCM_IV_LENGTH_IN_BYTES + len +
+                EncryptedRepository.GCM_TAG_LENGTH_IN_BYTES; i++) {
+            for (int j = 0; j < 8; j++) {
+                // flip bit
+                encryptedBytes[i] ^= (1 << j);
+                // fail decryption
+                try (InputStream in = new DecryptionPacketsInputStream(new ByteArrayInputStream(encryptedBytes), secretKey, nonce,
+                        packetLen)) {
+                    IOException e = expectThrows(IOException.class, () -> {
+                        in.readAllBytes();
+                    });
+                    assertThat(e.getMessage(), Matchers.is("Exception during packet decryption"));
+                }
+                // flip bit back
+                encryptedBytes[i] ^= (1 << j);
+            }
+        }
+    }
+
+    public void testFailureEncryptAndDecryptAlteredCiphertextIV() throws Exception {
+        int len = 8 + Randomness.get().nextInt(8);
+        int packetLen = 4 + Randomness.get().nextInt(4);
+        byte[] plainBytes = new byte[len];
+        Randomness.get().nextBytes(plainBytes);
+        SecretKey secretKey = generateSecretKey();
+        int nonce = Randomness.get().nextInt();
+        byte[] encryptedBytes;
+        try (InputStream in = new EncryptionPacketsInputStream(new ByteArrayInputStream(plainBytes, 0, len), secretKey, nonce,
+                packetLen)) {
+            encryptedBytes = in.readAllBytes();
+        }
+        assertThat(encryptedBytes.length, Matchers.is((int) EncryptionPacketsInputStream.getEncryptionLength(len, packetLen)));
+        int encryptedPacketLen = EncryptedRepository.GCM_IV_LENGTH_IN_BYTES + packetLen + EncryptedRepository.GCM_TAG_LENGTH_IN_BYTES;
+        for (int i = 0; i < encryptedBytes.length; i += encryptedPacketLen) {
+            for (int j = 0; j < EncryptedRepository.GCM_IV_LENGTH_IN_BYTES; j++) {
+                for (int k = 0; k < 8; k++) {
+                    // flip bit
+                    encryptedBytes[i + j] ^= (1 << k);
+                    try (InputStream in = new DecryptionPacketsInputStream(new ByteArrayInputStream(encryptedBytes), secretKey, nonce,
+                            packetLen)) {
+                        IOException e = expectThrows(IOException.class, () -> {
+                            in.readAllBytes();
+                        });
+                        if (j < Integer.BYTES) {
+                            assertThat(e.getMessage(), Matchers.startsWith("Packet nonce mismatch"));
+                        } else {
+                            assertThat(e.getMessage(), Matchers.startsWith("Packet counter mismatch"));
+                        }
+                    }
+                    // flip bit back
+                    encryptedBytes[i + j] ^= (1 << k);
+                }
+            }
+        }
+    }
+
+    private void testEncryptAndDecryptSuccess(byte[] plainBytes, SecretKey secretKey, int nonce, int packetLen) throws Exception {
+        for (int len = 0; len <= plainBytes.length; len++) {
+            byte[] encryptedBytes;
+            try (InputStream in = new EncryptionPacketsInputStream(new ByteArrayInputStream(plainBytes, 0, len), secretKey, nonce,
+                    packetLen)) {
+                encryptedBytes = in.readAllBytes();
+            }
+            assertThat((long) encryptedBytes.length, Matchers.is(EncryptionPacketsInputStream.getEncryptionLength(len, packetLen)));
+            byte[] decryptedBytes;
+            try (InputStream in = new DecryptionPacketsInputStream(new ReadLessFilterInputStream(new ByteArrayInputStream(encryptedBytes)),
+                    secretKey, nonce, packetLen)) {
+                decryptedBytes = in.readAllBytes();
+            }
+            assertThat(decryptedBytes.length, Matchers.is(len));
+            assertThat((long) decryptedBytes.length, Matchers.is(DecryptionPacketsInputStream.getDecryptionLength(encryptedBytes.length,
+                    packetLen)));
+            for (int i = 0; i < len; i++) {
+                assertThat(decryptedBytes[i], Matchers.is(plainBytes[i]));
+            }
+        }
+    }
+
+    // input stream that reads less bytes than asked to, testing that packet-wide reads don't rely on `read` calls for memory buffers which
+    // always return the same number of bytes they are asked to
+    private static class ReadLessFilterInputStream extends FilterInputStream {
+
+        protected ReadLessFilterInputStream(InputStream in) {
+            super(in);
+        }
+
+        @Override
+        public int read(byte[] b, int off, int len) throws IOException {
+            if (len == 0) {
+                return 0;
+            }
+            return super.read(b, off, randomIntBetween(1, len));
+        }
+    }
+
+    private SecretKey generateSecretKey() throws Exception {
+        KeyGenerator keyGen = KeyGenerator.getInstance("AES");
+        keyGen.init(256, new SecureRandom());
+        return keyGen.generateKey();
+    }
+}

Diff for: x-pack/plugin/repository-encrypted/src/test/java/org/elasticsearch/repositories/encrypted/BufferOnMarkInputStreamTests.java

@@ -0,0 +1,231 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.repositories.encrypted;
+
+import org.elasticsearch.common.Randomness;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.test.ESTestCase;
+import org.hamcrest.Matchers;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class PrefixInputStreamTests extends ESTestCase {
+
+    public void testZeroLength() throws Exception {
+        Tuple<AtomicInteger, InputStream> mockTuple = getMockBoundedInputStream(0);
+        PrefixInputStream test = new PrefixInputStream(mockTuple.v2(), 1 + Randomness.get().nextInt(32), randomBoolean());
+        assertThat(test.available(), Matchers.is(0));
+        assertThat(test.read(), Matchers.is(-1));
+        assertThat(test.skip(1 + Randomness.get().nextInt(32)), Matchers.is(0L));
+    }
+
+    public void testClose() throws Exception {
+        int boundedLength = 1 + Randomness.get().nextInt(256);
+        Tuple<AtomicInteger, InputStream> mockTuple = getMockBoundedInputStream(boundedLength);
+        int prefixLength = Randomness.get().nextInt(boundedLength);
+        PrefixInputStream test = new PrefixInputStream(mockTuple.v2(), prefixLength, randomBoolean());
+        test.close();
+        int byteCountBefore = mockTuple.v1().get();
+        IOException e = expectThrows(IOException.class, () -> {
+            test.read();
+        });
+        assertThat(e.getMessage(), Matchers.is("Stream has been closed"));
+        e = expectThrows(IOException.class, () -> {
+            byte[] b = new byte[1 + Randomness.get().nextInt(32)];
+            test.read(b, 0, 1 + Randomness.get().nextInt(b.length));
+        });
+        assertThat(e.getMessage(), Matchers.is("Stream has been closed"));
+        e = expectThrows(IOException.class, () -> {
+            test.skip(1 + Randomness.get().nextInt(32));
+        });
+        assertThat(e.getMessage(), Matchers.is("Stream has been closed"));
+        e = expectThrows(IOException.class, () -> {
+            test.available();
+        });
+        assertThat(e.getMessage(), Matchers.is("Stream has been closed"));
+        int byteCountAfter = mockTuple.v1().get();
+        assertThat(byteCountBefore - byteCountAfter, Matchers.is(0));
+        // test closeSource parameter
+        AtomicBoolean isClosed = new AtomicBoolean(false);
+        InputStream mockIn = mock(InputStream.class);
+        doAnswer(new Answer<Void>() {
+            public Void answer(InvocationOnMock invocation) {
+                isClosed.set(true);
+                return null;
+            }
+        }).when(mockIn).close();
+        new PrefixInputStream(mockIn, 1 + Randomness.get().nextInt(32), true).close();
+        assertThat(isClosed.get(), Matchers.is(true));
+        isClosed.set(false);
+        new PrefixInputStream(mockIn, 1 + Randomness.get().nextInt(32), false).close();
+        assertThat(isClosed.get(), Matchers.is(false));
+    }
+
+    public void testAvailable() throws Exception {
+        AtomicInteger available = new AtomicInteger(0);
+        int boundedLength = 1 + Randomness.get().nextInt(256);
+        InputStream mockIn = mock(InputStream.class);
+        when(mockIn.available()).thenAnswer(invocationOnMock -> {
+            return available.get();
+        });
+        PrefixInputStream test = new PrefixInputStream(mockIn, boundedLength, randomBoolean());
+        assertThat(test.available(), Matchers.is(0));
+        available.set(Randomness.get().nextInt(boundedLength));
+        assertThat(test.available(), Matchers.is(available.get()));
+        available.set(boundedLength + 1 + Randomness.get().nextInt(boundedLength));
+        assertThat(test.available(), Matchers.is(boundedLength));
+    }
+
+    public void testReadPrefixLength() throws Exception {
+        int boundedLength = 1 + Randomness.get().nextInt(256);
+        Tuple<AtomicInteger, InputStream> mockTuple = getMockBoundedInputStream(boundedLength);
+        int prefixLength = Randomness.get().nextInt(boundedLength);
+        PrefixInputStream test = new PrefixInputStream(mockTuple.v2(), prefixLength, randomBoolean());
+        int byteCountBefore = mockTuple.v1().get();
+        byte[] b = test.readAllBytes();
+        int byteCountAfter = mockTuple.v1().get();
+        assertThat(b.length, Matchers.is(prefixLength));
+        assertThat(byteCountBefore - byteCountAfter, Matchers.is(prefixLength));
+        assertThat(test.read(), Matchers.is(-1));
+        assertThat(test.available(), Matchers.is(0));
+        assertThat(mockTuple.v2().read(), Matchers.not(-1));
+    }
+
+    public void testSkipPrefixLength() throws Exception {
+        int boundedLength = 1 + Randomness.get().nextInt(256);
+        Tuple<AtomicInteger, InputStream> mockTuple = getMockBoundedInputStream(boundedLength);
+        int prefixLength = Randomness.get().nextInt(boundedLength);
+        PrefixInputStream test = new PrefixInputStream(mockTuple.v2(), prefixLength, randomBoolean());
+        int byteCountBefore = mockTuple.v1().get();
+        skipNBytes(test, prefixLength);
+        int byteCountAfter = mockTuple.v1().get();
+        assertThat(byteCountBefore - byteCountAfter, Matchers.is(prefixLength));
+        assertThat(test.read(), Matchers.is(-1));
+        assertThat(test.available(), Matchers.is(0));
+        assertThat(mockTuple.v2().read(), Matchers.not(-1));
+    }
+
+    public void testReadShorterWrapped() throws Exception {
+        int boundedLength = 1 + Randomness.get().nextInt(256);
+        Tuple<AtomicInteger, InputStream> mockTuple = getMockBoundedInputStream(boundedLength);
+        int prefixLength = boundedLength;
+        if (randomBoolean()) {
+            prefixLength += 1 + Randomness.get().nextInt(boundedLength);
+        }
+        PrefixInputStream test = new PrefixInputStream(mockTuple.v2(), prefixLength, randomBoolean());
+        int byteCountBefore = mockTuple.v1().get();
+        byte[] b = test.readAllBytes();
+        int byteCountAfter = mockTuple.v1().get();
+        assertThat(b.length, Matchers.is(boundedLength));
+        assertThat(byteCountBefore - byteCountAfter, Matchers.is(boundedLength));
+        assertThat(test.read(), Matchers.is(-1));
+        assertThat(test.available(), Matchers.is(0));
+        assertThat(mockTuple.v2().read(), Matchers.is(-1));
+        assertThat(mockTuple.v2().available(), Matchers.is(0));
+    }
+
+    public void testSkipShorterWrapped() throws Exception {
+        int boundedLength = 1 + Randomness.get().nextInt(256);
+        Tuple<AtomicInteger, InputStream> mockTuple = getMockBoundedInputStream(boundedLength);
+        final int prefixLength;
+        if (randomBoolean()) {
+            prefixLength = boundedLength + 1 + Randomness.get().nextInt(boundedLength);
+        } else {
+            prefixLength = boundedLength;
+        }
+        PrefixInputStream test = new PrefixInputStream(mockTuple.v2(), prefixLength, randomBoolean());
+        int byteCountBefore = mockTuple.v1().get();
+        if (prefixLength == boundedLength) {
+            skipNBytes(test, prefixLength);
+        } else {
+            expectThrows(EOFException.class, () -> {
+                skipNBytes(test, prefixLength);
+            });
+        }
+        int byteCountAfter = mockTuple.v1().get();
+        assertThat(byteCountBefore - byteCountAfter, Matchers.is(boundedLength));
+        assertThat(test.read(), Matchers.is(-1));
+        assertThat(test.available(), Matchers.is(0));
+        assertThat(mockTuple.v2().read(), Matchers.is(-1));
+        assertThat(mockTuple.v2().available(), Matchers.is(0));
+    }
+
+    private Tuple<AtomicInteger, InputStream> getMockBoundedInputStream(int bound) throws IOException {
+        InputStream mockSource = mock(InputStream.class);
+        AtomicInteger bytesRemaining = new AtomicInteger(bound);
+        when(mockSource.read(org.mockito.Matchers.<byte[]>any(), org.mockito.Matchers.anyInt(), org.mockito.Matchers.anyInt())).
+                thenAnswer(invocationOnMock -> {
+                    final byte[] b = (byte[]) invocationOnMock.getArguments()[0];
+                    final int off = (int) invocationOnMock.getArguments()[1];
+                    final int len = (int) invocationOnMock.getArguments()[2];
+                    if (len == 0) {
+                        return 0;
+                    } else {
+                        if (bytesRemaining.get() <= 0) {
+                            return -1;
+                        }
+                        int bytesCount = 1 + Randomness.get().nextInt(Math.min(len, bytesRemaining.get()));
+                        bytesRemaining.addAndGet(-bytesCount);
+                        return bytesCount;
+                    }
+                });
+        when(mockSource.read()).thenAnswer(invocationOnMock -> {
+            if (bytesRemaining.get() <= 0) {
+                return -1;
+            }
+            bytesRemaining.decrementAndGet();
+            return Randomness.get().nextInt(256);
+        });
+        when(mockSource.skip(org.mockito.Matchers.anyLong())).thenAnswer(invocationOnMock -> {
+            final long n = (long) invocationOnMock.getArguments()[0];
+            if (n <= 0 || bytesRemaining.get() <= 0) {
+                return 0;
+            }
+            int bytesSkipped = 1 + Randomness.get().nextInt(Math.min(bytesRemaining.get(), Math.toIntExact(n)));
+            bytesRemaining.addAndGet(-bytesSkipped);
+            return bytesSkipped;
+        });
+        when(mockSource.available()).thenAnswer(invocationOnMock -> {
+            if (bytesRemaining.get() <= 0) {
+                return 0;
+            }
+            return 1 + Randomness.get().nextInt(bytesRemaining.get());
+        });
+        when(mockSource.markSupported()).thenReturn(false);
+        return new Tuple<>(bytesRemaining, mockSource);
+    }
+
+    private static void skipNBytes(InputStream in, long n) throws IOException {
+        if (n > 0) {
+            long ns = in.skip(n);
+            if (ns >= 0 && ns < n) { // skipped too few bytes
+                // adjust number to skip
+                n -= ns;
+                // read until requested number skipped or EOS reached
+                while (n > 0 && in.read() != -1) {
+                    n--;
+                }
+                // if not enough skipped, then EOFE
+                if (n != 0) {
+                    throw new EOFException();
+                }
+            } else if (ns != n) { // skipped negative or too many bytes
+                throw new IOException("Unable to skip exactly");
+            }
+        }
+    }
+}

Diff for: x-pack/plugin/repository-encrypted/src/test/java/org/elasticsearch/repositories/encrypted/ChainingInputStreamTests.java

@@ -0,0 +1,16 @@
+# Integration tests for repository-encrypted
+#
+"Plugin repository-encrypted is loaded":
+    - skip:
+        reason: "contains is a newly added assertion"
+        features: contains
+    - do:
+        cluster.state: {}
+
+    # Get master node id
+    - set: { master_node: master }
+
+    - do:
+        nodes.info: {}
+
+    - contains:  { nodes.$master.plugins: { name: repository-encrypted } }