Packet-based enc/dec cipher streams (elastic#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 elastic#48221 , elastic#46170

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

x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/CountingInputStream.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();
+            }
+        }
+    }
+}

x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/DecryptionPacketsInputStream.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);
+        }
+    }
+}

x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/EncryptedRepository.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;
+}

x-pack/plugin/repository-encrypted/src/main/java/org/elasticsearch/repositories/encrypted/EncryptedRepositoryPlugin.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.
+    }
+}