EAR: port the ear feature from enterprise

Bulk transfer of EAR functionality. Includes all providers etc. Could maybe break up into smaller blocks, but once it gets down to the core of it, would require messing with code instead of just moving. So this is it. Note: KMIP support is disabled unless you happen to have the kmipc SDK in your scylla dir. Adds optional encryption of sstables and commitlog, using block level file encryption. Provides key sourcing from various sources, such as local files or popular KMS systems.
2025-01-08 10:49:08 +00:00
parent e734fc11ec
commit 723518c390
32 changed files with 7731 additions and 1 deletions
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -267,6 +267,7 @@ add_subdirectory(compaction)
 add_subdirectory(cql3)
 add_subdirectory(data_dictionary)
 add_subdirectory(dht)
+add_subdirectory(ent)
 add_subdirectory(gms)
 add_subdirectory(idl)
 add_subdirectory(index)
@@ -308,6 +309,7 @@ set(scylla_libs
    cql3
    data_dictionary
    dht
+    encryption
    gms
    idl
    index
--- a/cmake/Findkmip.cmake
+++ b/cmake/Findkmip.cmake
@@ -0,0 +1,53 @@
+#
+# Copyright 2024-present ScyllaDB
+#
+
+#
+# SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+#
+
+set(kmip_ver "2.1.0t")
+
+cmake_host_system_information(
+  RESULT distrib_id QUERY DISTRIB_ID)
+if(distrib_id MATCHES "centos|fedora|rhel")
+  set(kmip_distrib "rhel84")
+else()
+  message(FATAL_ERROR "Could not locate kmipc library for ${distrib_id}")
+endif()
+
+if(CMAKE_SYSTEM_PROCESSOR MATCHES "aarch64|AARCH64")
+  set(kmip_arch "aarch64")
+elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "amd64|x86_64")
+  set(kmip_arch "64")
+elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "(powerpc|ppc)64le")
+  set(kmip_arch "ppc64le")
+endif()
+
+set(kmip_ROOT "${PROJECT_SOURCE_DIR}/kmipc/kmipc-${kmip_ver}-${kmip_distrib}_${kmip_arch}")
+find_library(kmip_LIBRARY
+  NAMES kmip
+  HINTS ${kmip_ROOT}/lib)
+
+find_path(kmip_INCLUDE_DIR
+  NAMES kmip.h
+  HINTS ${kmip_ROOT}/include)
+
+mark_as_advanced(
+  kmip_LIBRARY
+  kmip_INCLUDE_DIR)
+
+find_package_handle_standard_args(kmip
+  REQUIRED_VARS
+    kmip_LIBRARY
+    kmip_INCLUDE_DIR)
+
+if(kmip_FOUND)
+  if (NOT TARGET KMIP::kmipc)
+    add_library(KMIP::kmipc UNKNOWN IMPORTED)
+    set_target_properties(KMIP::kmipc PROPERTIES
+      INTERFACE_INCLUDE_DIRECTORIES "${kmip_INCLUDE_DIR}"
+      IMPORTED_LINK_INTERFACE_LANGUAGES "C"
+      IMPORTED_LOCATION "${kmip_LIBRARY}")
+  endif()
+endif()
--- a/configure.py
+++ b/configure.py
@@ -1123,6 +1123,19 @@ scylla_core = (['message/messaging_service.cc',
                'utils/arch/powerpc/crc32-vpmsum/crc32_wrapper.cc',
                'querier.cc',
                'mutation_writer/multishard_writer.cc',
+                'ent/encryption/encryption_config.cc',
+                'ent/encryption/encryption.cc',
+                'ent/encryption/symmetric_key.cc',
+                'ent/encryption/local_file_provider.cc',
+                'ent/encryption/replicated_key_provider.cc',
+                'ent/encryption/system_key.cc',
+                'ent/encryption/encrypted_file_impl.cc',
+                'ent/encryption/kmip_host.cc',
+                'ent/encryption/kmip_key_provider.cc',
+                'ent/encryption/kms_host.cc',
+                'ent/encryption/kms_key_provider.cc',
+                'ent/encryption/gcp_host.cc',
+                'ent/encryption/gcp_key_provider.cc',
                'multishard_mutation_query.cc',
                'reader_concurrency_semaphore.cc',
                'sstables_loader.cc',
@@ -2000,7 +2013,7 @@ pkgs = ['libsystemd',
 pkgs.append('lua53' if have_pkg('lua53') else 'lua')


-libs = ' '.join([maybe_static(args.staticyamlcpp, '-lyaml-cpp'), '-latomic', '-lz', '-lsnappy',
+libs = ' '.join([maybe_static(args.staticyamlcpp, '-lyaml-cpp'), '-latomic', '-lz', '-lsnappy', '-lcrypto',
                 ' -lstdc++fs', ' -lcrypt', ' -lcryptopp', ' -lpthread',
                 # Must link with static version of libzstd, since
                 # experimental APIs that we use are only present there.
@@ -2022,6 +2035,32 @@ user_ldflags += ' -fvisibility=hidden'
 if args.staticcxx:
    user_ldflags += " -static-libstdc++"

+kmip_lib_ver = '1.9.2a';
+
+def kmiplib():
+    os_ids = get_os_ids()
+    for id in os_ids:
+        if id in { 'centos', 'fedora', 'rhel' }:
+            return 'rhel84'
+    print('Could not resolve libkmip.a for platform {}'.format(os_ids))
+    sys.exit(1)
+
+def target_cpu():
+    cpu, _, _ = subprocess.check_output([cxx, '-dumpmachine']).decode('utf-8').partition('-')
+    return cpu    
+
+def kmip_arch():
+    arch = target_cpu()
+    if arch == 'x86_64':
+        return '64'
+    return arch 
+
+kmipc_dir = f'kmipc/kmipc-2.1.0t-{kmiplib()}_{kmip_arch()}'
+kmipc_lib = f'{kmipc_dir}/lib/libkmip.a'
+libs += ' -lboost_filesystem'
+if os.path.exists(kmipc_lib):
+    libs += f' {kmipc_lib}'
+    user_cflags += f' -I{kmipc_dir}/include -DHAVE_KMIP'

 def get_extra_cxxflags(mode, mode_config, cxx, debuginfo):
    cxxflags = []
--- a/ent/CMakeLists.txt
+++ b/ent/CMakeLists.txt
@@ -0,0 +1 @@
+add_subdirectory(encryption)
--- a/ent/encryption/CMakeLists.txt
+++ b/ent/encryption/CMakeLists.txt
@@ -0,0 +1,44 @@
+include(add_whole_archive)
+
+find_package(cpp-jwt REQUIRED)
+find_package(kmip)
+
+add_library(scylla_encryption STATIC)
+target_sources(scylla_encryption
+  PRIVATE
+    encrypted_file_impl.cc
+    encryption.cc
+    encryption_config.cc
+    gcp_host.cc
+    gcp_key_provider.cc
+    kmip_host.cc
+    kmip_key_provider.cc
+    kms_host.cc
+    kms_key_provider.cc
+    local_file_provider.cc
+    replicated_key_provider.cc
+    symmetric_key.cc
+    system_key.cc)
+target_include_directories(scylla_encryption
+  PUBLIC
+    ${CMAKE_SOURCE_DIR})
+target_link_libraries(scylla_encryption
+  PUBLIC
+    Seastar::seastar
+  PRIVATE
+    cql3
+    utils
+    cpp-jwt::cpp-jwt)
+if(kmip_FOUND)
+  target_link_libraries(scylla_encryption
+    PRIVATE
+      KMIP::kmipc)
+  target_compile_definitions(scylla_encryption
+    PUBLIC
+      HAVE_KMIP)
+endif()
+
+check_headers(check-headers scylla_encryption
+  GLOB_RECURSE ${CMAKE_CURRENT_SOURCE_DIR}/*.hh)
+
+add_whole_archive(encryption scylla_encryption)
--- a/ent/encryption/encrypted_file_impl.cc
+++ b/ent/encryption/encrypted_file_impl.cc
@@ -0,0 +1,555 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#include <fcntl.h>
+#include <fmt/format.h>
+#include <seastar/core/file.hh>
+#include <seastar/core/byteorder.hh>
+
+#include "symmetric_key.hh"
+#include "encryption.hh"
+#include "utils/serialization.hh"
+#include "encrypted_file_impl.hh"
+
+namespace encryption {
+
+static inline bool is_aligned(size_t n, size_t a) {
+    return (n & (a - 1)) == 0;
+}
+
+/**
+ * Very simple block-encrypting file wrapper.
+ *
+ * Uses user provided symmetric key + ESSIV block IV calculation
+ * to encrypt data.
+ *
+ * The essiv block key is created by generating the SHA256 hash
+ * of the provided data encryption key bytes, truncated to block_key_len/8
+ * and generating an AES/ECB key using this data.
+ *
+ * The file is divided in N blocks of `block_size` size.
+ * Each block is encrypted (unpadded) with the provided key and
+ * block mode, using an IV derived by (essiv):
+ *
+ *  bytes tmp[<key block size, typically 16>] = { 0, ..., uint64_t-little-endian(<block number>) }
+ *  iv = block_key->encrypt(tmp);
+ *
+ * All encryption is done unpadded. To handle file sizes we use
+ * a slightly shaky scheme:
+ *
+ * Since all writes are assumed to be done by us, and must be aligned,
+ * we can assume in turn that any resizing should be made by our truncate
+ * method. If we attept to truncate to a size not a multiple of our
+ * _key_ block size (typically 16), we add the same size to the actual
+ * truncation size.
+ * On read we then check the file size. If we're reading from a file
+ * with unaliged size, we know there are key-block-size junk at the end.
+ * We can align down the last decryption call to match block size, then
+ * discard the excessive bytes from the result.
+ *
+ * If we're in a read/write situation, we need to keep size updated, and
+ * we could possibly race with disk op/continuations.
+ * But we are really only for ro/wo cases.
+ */
+class encrypted_file_impl : public seastar::file_impl {
+    file _file;
+    ::shared_ptr<symmetric_key> _key;
+    ::shared_ptr<symmetric_key> _block_key;
+    bytes _hash_salt;
+
+    std::optional<uint64_t> _file_length;
+
+    // this is somewhat large, but we assume this is for bulky stuff like sstables/commitlog
+    // so large alignment should be preferable to reaclculating block IV too often.
+    static constexpr size_t block_size = 4096;
+    static constexpr size_t block_key_len = 256;
+
+    class my_file_handle_impl;
+    friend class my_file_handle_impl;
+
+    bytes iv_for(uint64_t pos) const;
+
+    using mode = symmetric_key::mode;
+
+    temporary_buffer<uint8_t> transform(uint64_t, const void* buffer, size_t len, mode);
+    size_t transform(uint64_t, const void* buffer, size_t len, void*, mode);
+
+    future<> verify_file_length();
+    void maybe_set_length(uint64_t);
+    void clear_length();
+
+    static ::shared_ptr<symmetric_key> generate_block_key(::shared_ptr<symmetric_key>);
+
+public:
+    encrypted_file_impl(file, ::shared_ptr<symmetric_key>);
+
+    future<size_t> write_dma(uint64_t pos, const void* buffer, size_t len, io_intent*) override;
+    future<size_t> write_dma(uint64_t pos, std::vector<iovec> iov, io_intent*) override;
+    future<size_t> read_dma(uint64_t pos, void* buffer, size_t len, io_intent*) override;
+    future<size_t> read_dma(uint64_t pos, std::vector<iovec> iov, io_intent*) override;
+
+
+    future<> flush() override {
+        return _file.flush();
+    }
+    future<struct stat> stat(void) override;
+    future<> truncate(uint64_t length) override;
+    future<> discard(uint64_t offset, uint64_t length) override {
+        return _file.discard(offset, length);
+    }
+    future<> allocate(uint64_t position, uint64_t length) override {
+        return _file.allocate(position, length);
+    }
+    future<uint64_t> size(void) override;
+    future<> close() override {
+        return _file.close();
+    }
+    std::unique_ptr<file_handle_impl> dup() override;
+
+    subscription<directory_entry> list_directory(std::function<future<> (directory_entry de)> next) override {
+        return _file.list_directory(std::move(next));
+    }
+    future<temporary_buffer<uint8_t>> dma_read_bulk(uint64_t offset, size_t range_size, io_intent*) override;
+};
+
+/**
+ * Note: ESSIV block iv generation implementation.
+ * See: http://securityevaluators.com/knowledge/papers/fde_whitepaper_draft_20170627.pdf
+ *
+ * We generate a key based on the sha256 of the data key, then encrypt each block number
+ * using this to get per-block IV.
+ * The key is AES-256, using ECB (non-iv) encryption
+ *
+ */
+::shared_ptr<symmetric_key> encrypted_file_impl::generate_block_key(::shared_ptr<symmetric_key> key) {
+    auto hash = calculate_sha256(key->key());
+    hash.resize(block_key_len / 8);
+    return ::make_shared<symmetric_key>(key_info{"AES/ECB", block_key_len }, hash);
+}
+
+encrypted_file_impl::encrypted_file_impl(file f, ::shared_ptr<symmetric_key> key)
+    : _file(std::move(f))
+    , _key(std::move(key))
+    , _block_key(generate_block_key(_key))
+{
+    _memory_dma_alignment = std::max<unsigned>(_file.memory_dma_alignment(), block_size);
+    _disk_read_dma_alignment = std::max<unsigned>(_file.disk_read_dma_alignment(), block_size);
+    _disk_write_dma_alignment = std::max<unsigned>(_file.disk_write_dma_alignment(), block_size);
+}
+
+static future<uint64_t> calculate_file_length(const file& f, size_t key_block_size) {
+    return f.size().then([key_block_size](uint64_t s) {
+        if (!is_aligned(s, key_block_size)) {
+            if (s < key_block_size) {
+                throw std::domain_error(fmt::format("file size {}, expected 0 or at least {}", s, key_block_size));
+            }
+            s -= key_block_size;
+        }
+        return s;
+    });
+}
+
+future<> encrypted_file_impl::verify_file_length() {
+    if (_file_length) {
+        return make_ready_future();
+    }
+    return calculate_file_length(_file, _key->block_size()).then([this](uint64_t s) {
+        _file_length = s;
+    });
+}
+
+void encrypted_file_impl::maybe_set_length(uint64_t s) {
+    if (s > _file_length.value_or(0)) {
+        _file_length = s;
+    }
+}
+
+void encrypted_file_impl::clear_length() {
+    _file_length = std::nullopt;
+}
+
+bytes encrypted_file_impl::iv_for(uint64_t pos) const {
+    assert(!(pos & (block_size - 1)));
+
+    // #658. ECB block mode has no IV. Bad for security,
+    // but must handle.
+    size_t iv_len = _key->iv_len();
+    if (iv_len == 0) {
+        return bytes{};
+    }
+
+    assert(iv_len >= _key->block_size());
+    assert(iv_len >= sizeof(uint64_t));
+
+    bytes b(bytes::initialized_later(), std::max(iv_len, _block_key->block_size()));
+    std::fill(b.begin(), b.end() - sizeof(uint64_t), 0);
+
+    // write block pos as little endian IV-len integer
+    auto block = pos / block_size;
+    write_le(reinterpret_cast<char *>(b.end()) - sizeof(uint64_t), block);
+
+    // encrypt the encoded block number to build an IV
+    _block_key->encrypt_unpadded(b.data(), b.size(), b.data());
+
+    b.resize(iv_len);
+
+    return b;
+}
+
+size_t encrypted_file_impl::transform(uint64_t pos, const void* buffer, size_t len, void* dst, mode m) {
+    assert(!(pos & (block_size - 1)));
+    assert(_file_length || m == mode::encrypt);
+
+    auto o = reinterpret_cast<char*>(dst);
+    auto i = reinterpret_cast<const char*>(buffer);
+    auto l = _file_length.value_or(std::numeric_limits<uint64_t>::max());
+    auto b = _key->block_size();
+
+    size_t off = 0;
+
+    for (; off < len; off += block_size) {
+        auto iv = iv_for(pos + off);
+        auto rem = std::min<uint64_t>(block_size, len - off);
+
+        if (rem < block_size || ((pos + off + rem) > l && m == symmetric_key::mode::decrypt)) {
+            // truncated block. should be the last one.
+            if (m != symmetric_key::mode::decrypt) {
+                throw std::invalid_argument("Output data not aligned");
+            }
+            _key->transform_unpadded(m, i + off, align_down(rem, b), o + off, iv.data());
+            return l - pos;
+        }
+        _key->transform_unpadded(m, i + off, block_size, o + off, iv.data());
+    }
+
+    return off;
+}
+
+temporary_buffer<uint8_t> encrypted_file_impl::transform(uint64_t pos, const void* buffer, size_t len, mode m) {
+    assert(!(len & (block_size - 1)));
+    auto tmp = temporary_buffer<uint8_t>::aligned(_file.memory_dma_alignment(), len);
+    auto s = transform(pos, buffer, len, tmp.get_write(), m);
+    tmp.trim(s);
+    return tmp;
+}
+
+future<size_t> encrypted_file_impl::write_dma(uint64_t pos, const void* buffer, size_t len, io_intent* intent) {
+    assert(!(len & (block_size - 1)));
+    auto tmp = transform(pos, buffer, len, mode::encrypt);
+    assert(tmp.size() == len); // writing
+    auto p = tmp.get();
+    return _file.dma_write(pos, p, len, intent).then([this, tmp = std::move(tmp), pos](size_t s) {
+        maybe_set_length(pos + s);
+        return s;
+    });
+}
+
+future<size_t> encrypted_file_impl::write_dma(uint64_t pos, std::vector<iovec> iov, io_intent* intent) {
+    std::vector<temporary_buffer<uint8_t>> tmp;
+    tmp.reserve(iov.size());
+    size_t n = 0;
+    for (auto& i : iov) {
+        assert(!(i.iov_len & (block_size - 1)));
+
+        tmp.emplace_back(transform(pos + n, i.iov_base, i.iov_len, mode::encrypt));
+        assert(tmp.back().size() == i.iov_len); // writing
+        n += i.iov_len;
+        i = iovec{ tmp.back().get_write(), tmp.back().size() };
+    }
+    return _file.dma_write(pos, std::move(iov), intent).then([this, tmp = std::move(tmp), pos](size_t s) {
+        maybe_set_length(pos + s);
+        return s;
+    });
+}
+
+future<size_t> encrypted_file_impl::read_dma(uint64_t pos, void* buffer, size_t len, io_intent* intent) {
+    assert(!(len & (block_size - 1)));
+    return verify_file_length().then([this, pos, buffer, len, intent] {
+        return _file.dma_read(pos, buffer, len, intent).then([this, pos, buffer](size_t len) {
+            return transform(pos, buffer, len, buffer, mode::decrypt);
+        });
+    });
+}
+
+future<size_t> encrypted_file_impl::read_dma(uint64_t pos, std::vector<iovec> iov, io_intent* intent) {
+    return verify_file_length().then([this, pos, iov = std::move(iov), intent]() mutable {
+        auto f = _file.dma_read(pos, iov, intent);
+        return f.then([this, pos, iov = std::move(iov)](size_t len) mutable {
+            size_t off = 0;
+            for (auto& i : iov) {
+                off += transform(pos + off, i.iov_base, i.iov_len, i.iov_base, mode::decrypt);
+            }
+            return off;
+        });
+    });
+}
+
+future<temporary_buffer<uint8_t>> encrypted_file_impl::dma_read_bulk(uint64_t offset, size_t range_size, io_intent* intent) {
+    return verify_file_length().then([this, offset, range_size, intent]() mutable {
+        auto front = offset & (block_size - 1);
+        offset -= front;
+        range_size += front;
+        // enterprise #925
+        // If caller is clever and asks for the last chunk of file
+        // explicitly (as in offset = N, range_size = size() - N),
+        // or any other unaligned size, we need to add enough padding
+        // to get the actual full block to decode.
+        auto block_size = align_up(range_size, _key->block_size());
+        return _file.dma_read_bulk<uint8_t>(offset, block_size, intent).then([this, offset, front, range_size](temporary_buffer<uint8_t> result) {
+            auto s = transform(offset, result.get(), result.size(), result.get_write(), mode::decrypt);
+            // never give back more than asked for.
+            result.trim(std::min(s, range_size));
+            result.trim_front(front);
+            return result;
+        });
+    });
+}
+
+future<> encrypted_file_impl::truncate(uint64_t length) {
+    return size().then([this, length](uint64_t s) {
+        if (s >= length) {
+            auto kb = _key->block_size();
+            auto n = length;
+            if (!is_aligned(length, kb)) {
+                n += kb;
+            }
+            return _file.truncate(n).then([this, length] {
+                _file_length = length;
+            });
+        }
+
+        // crap. we need to pad zeros. But zeros here means
+        // encrypted zeros. So we must do this surprisingly
+        // expensively, by actually writing said zeros block
+        // by block. Anyone hoping for sparse files is now
+        // severely disappointed!
+
+        auto buf_size = align_up(std::min(length, 32 * block_size), block_size);
+        auto aligned_size = align_down(s, block_size);
+
+        temporary_buffer<char> buf(buf_size);
+        std::fill(buf.get_write(), buf.get_write() + buf_size, 0);
+
+        struct trunc {
+            temporary_buffer<char> buf;
+            uint64_t aligned_size;
+            uint64_t size;
+            uint64_t length;
+        };
+
+        return do_with(trunc{std::move(buf), aligned_size, s, length}, [this](trunc & t) {
+            return repeat([this, &t] {
+                if (t.aligned_size >= t.length) {
+                    return make_ready_future<stop_iteration>(stop_iteration::yes);
+                }
+                auto n = std::min(t.buf.size(), align_up(size_t(t.length - t.aligned_size), block_size));
+                if (t.aligned_size < t.size) {
+                    return read_dma(t.aligned_size, t.buf.get_write(), n, nullptr).then([&, n](size_t r) mutable {
+                        auto rem = size_t(t.size - t.aligned_size);
+                        auto ar = align_up(r, block_size);
+                        assert(ar <= t.buf.size());
+                        if (rem < ar) {
+                            std::fill(t.buf.get_write() + rem, t.buf.get_write() + ar, 0);
+                        }
+                        return write_dma(t.aligned_size, t.buf.get(), ar, nullptr).then([&, n](size_t w) {
+                            t.aligned_size += w;
+                            // #1869. On btrfs, we get the buffer potentially clobbered up to "n" (max read amount)
+                            // even when "r" (actual bytes read) is less.
+                            std::fill(t.buf.get_write(), t.buf.get_write() + n, 0);
+                            return make_ready_future<stop_iteration>(stop_iteration::no);
+                        });
+                    });
+                }
+                return write_dma(t.aligned_size, t.buf.get(), n, nullptr).then([&](size_t w) {
+                    t.aligned_size += w;
+                    return make_ready_future<stop_iteration>(stop_iteration::no);
+                });
+            });
+        }).then([this, length] {
+            return truncate(length);
+        });;
+    });
+}
+
+future<struct stat> encrypted_file_impl::stat() {
+    return _file.stat().then([this](struct stat s) {
+        return verify_file_length().then([this, s]() mutable {
+            s.st_size = *_file_length;
+            return s;
+        });
+    });
+}
+
+future<uint64_t> encrypted_file_impl::size() {
+    return  verify_file_length().then([this] {
+        return *_file_length;
+    });
+}
+
+
+std::unique_ptr<file_handle_impl> encrypted_file_impl::dup() {
+    class my_file_handle_impl : public seastar::file_handle_impl {
+        seastar::file_handle _handle;
+        key_info _info;
+        bytes _key;
+    public:
+        my_file_handle_impl(seastar::file_handle h, const key_info& info, const bytes& key)
+            : _handle(std::move(h))
+            , _info(info)
+            , _key(key)
+        {}
+        std::unique_ptr<file_handle_impl> clone() const override {
+            return std::make_unique<my_file_handle_impl>(_handle, _info, _key);
+        }
+        seastar::shared_ptr<file_impl> to_file() && override {
+            return seastar::make_shared<encrypted_file_impl>(_handle.to_file(), ::make_shared<symmetric_key>(_info, _key));
+        }
+    };
+
+    return std::make_unique<my_file_handle_impl>(_file.dup(), _key->info(), _key->key());
+}
+
+shared_ptr<file_impl> make_encrypted_file(file f, ::shared_ptr<symmetric_key> k) {
+    return ::make_shared<encrypted_file_impl>(std::move(f), std::move(k));
+}
+
+class indirect_encrypted_file_impl : public file_impl {
+    ::shared_ptr<file_impl> _impl;
+    file _f;
+    size_t _key_block_size;
+    get_key_func _get;
+
+    future<> get() {
+        if (_impl) {
+           return make_ready_future<>();
+        }
+        return _get().then([this](::shared_ptr<symmetric_key> k) {
+            // #978 could be running the getting more than once.
+            // Only write _impl once though
+            if (!_impl) {
+                _impl = make_encrypted_file(_f, std::move(k));
+            }
+        });
+    }
+public:
+    indirect_encrypted_file_impl(file f, size_t key_block_size, get_key_func get)
+        : _f(f), _key_block_size(key_block_size), _get(std::move(get))
+    {}
+
+    future<size_t> write_dma(uint64_t pos, const void* buffer, size_t len, io_intent* intent) override {
+        return get().then([this, pos, buffer, len, intent]() {
+            return _impl->write_dma(pos, buffer, len, intent);
+        });
+    }
+    future<size_t> write_dma(uint64_t pos, std::vector<iovec> iov, io_intent* intent) override {
+        return get().then([this, pos, iov = std::move(iov), intent]() mutable {
+            return _impl->write_dma(pos, std::move(iov), intent);
+        });
+    }
+    future<size_t> read_dma(uint64_t pos, void* buffer, size_t len, io_intent* intent) override {
+        return get().then([this, pos, buffer, len, intent]() {
+            return _impl->read_dma(pos, buffer, len, intent);
+        });
+    }
+    future<size_t> read_dma(uint64_t pos, std::vector<iovec> iov, io_intent* intent) override {
+        return get().then([this, pos, iov = std::move(iov), intent]() mutable {
+            return _impl->read_dma(pos, std::move(iov), intent);
+        });
+    }
+    future<temporary_buffer<uint8_t>> dma_read_bulk(uint64_t offset, size_t range_size, io_intent* intent) override {
+        return get().then([this, offset, range_size, intent]() {
+            return _impl->dma_read_bulk(offset, range_size, intent);
+        });
+    }
+    future<> flush(void) override {
+        if (_impl) {
+            return _impl->flush();
+        }
+        return _f.flush();
+    }
+    future<struct stat> stat(void) override {
+        if (_impl) {
+            return _impl->stat();
+        }
+        return _f.stat().then([this](struct stat s) {
+           return calculate_file_length(_f, _key_block_size).then([s](uint64_t fs) mutable {
+               s.st_size = fs;
+               return s;
+           });
+        });
+    }
+    future<> truncate(uint64_t length) override {
+        if (_impl) {
+            return _impl->truncate(length);
+        }
+        return _f.truncate(length);
+    }
+    future<> discard(uint64_t offset, uint64_t length) override {
+        if (_impl) {
+            return _impl->discard(offset, length);
+        }
+        return _f.discard(offset, length);
+    }
+    future<> allocate(uint64_t position, uint64_t length) override {
+        if (_impl) {
+            return _impl->allocate(position, length);
+        }
+        return _f.allocate(position, length);
+    }
+    future<uint64_t> size(void) override {
+        if (_impl) {
+            return _impl->size();
+        }
+        return calculate_file_length(_f, _key_block_size);
+    }
+    future<> close() override {
+        if (_impl) {
+            return _impl->close();
+        }
+        return _f.close();
+    }
+    std::unique_ptr<file_handle_impl> dup() override {
+        if (_impl) {
+            return _impl->dup();
+        }
+        class my_file_handle_impl : public seastar::file_handle_impl {
+            seastar::file_handle _handle;
+            size_t _key_block_size;
+            get_key_func _get;
+        public:
+            my_file_handle_impl(seastar::file_handle h, size_t key_block_size, get_key_func get)
+                : _handle(std::move(h))
+                , _key_block_size(key_block_size)
+                , _get(std::move(get))
+            {}
+            std::unique_ptr<file_handle_impl> clone() const override {
+                return std::make_unique<my_file_handle_impl>(_handle, _key_block_size, _get);
+            }
+            seastar::shared_ptr<file_impl> to_file() && override {
+                return make_delayed_encrypted_file(_handle.to_file(), _key_block_size, _get);
+            }
+        };
+        return std::make_unique<my_file_handle_impl>(_f.dup(), _key_block_size, _get);
+    }
+
+    subscription<directory_entry> list_directory(std::function<future<> (directory_entry de)> next) override {
+        if (_impl) {
+            return _impl->list_directory(std::move(next));
+        }
+        return _f.list_directory(std::move(next));
+    }
+};
+
+shared_ptr<seastar::file_impl> make_delayed_encrypted_file(file f, size_t key_block_size, get_key_func get) {
+    return ::make_shared<indirect_encrypted_file_impl>(std::move(f), key_block_size, std::move(get));
+}
+
+
+}
+
--- a/ent/encryption/encrypted_file_impl.hh
+++ b/ent/encryption/encrypted_file_impl.hh
@@ -0,0 +1,24 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#include <seastar/core/file.hh>
+#include <seastar/core/shared_ptr.hh>
+
+#include "symmetric_key.hh"
+
+namespace encryption {
+
+class symmetric_key;
+
+shared_ptr<file_impl> make_encrypted_file(file, ::shared_ptr<symmetric_key>);
+
+using get_key_func = std::function<future<::shared_ptr<symmetric_key>>()>;
+
+shared_ptr<file_impl> make_delayed_encrypted_file(file, size_t, get_key_func);
+}
--- a/ent/encryption/encryption.cc
+++ b/ent/encryption/encryption.cc
--- a/ent/encryption/encryption.hh
+++ b/ent/encryption/encryption.hh
@@ -0,0 +1,196 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include <map>
+
+#include <seastar/core/future.hh>
+#include <seastar/core/sstring.hh>
+#include <seastar/core/shared_ptr.hh>
+#include <seastar/core/distributed.hh>
+
+#include <fmt/core.h>
+#include <fmt/ostream.h>
+
+#include "../../bytes.hh"
+#include "../../compress.hh"
+
+class service_set;
+
+namespace replica {
+class database;
+}
+
+namespace db {
+class config;
+class extensions;
+}
+
+namespace cql3 {
+class query_processor;
+}
+namespace service {
+class storage_service;
+class migration_manager;
+}
+
+namespace sstables {
+    class sstable;
+}
+
+namespace encryption {
+inline const sstring KEY_PROVIDER = "key_provider";
+inline const sstring SECRET_KEY_PROVIDER_FACTORY_CLASS = "secret_key_provider_factory_class";
+inline const sstring SECRET_KEY_FILE = "secret_key_file";
+inline const sstring SYSTEM_KEY_FILE = "system_key_file";
+inline const sstring CIPHER_ALGORITHM = "cipher_algorithm";
+inline const sstring SECRET_KEY_STRENGTH = "secret_key_strength";
+
+inline const sstring HOST_NAME = "kmip_host";
+inline const sstring TEMPLATE_NAME = "template_name";
+inline const sstring KEY_NAMESPACE = "key_namespace";
+
+bytes base64_decode(const sstring&, size_t off = 0, size_t n = sstring::npos);
+sstring base64_encode(const bytes&, size_t off = 0, size_t n = bytes::npos);
+bytes calculate_md5(const bytes&, size_t off = 0, size_t n = bytes::npos);
+bytes calculate_sha256(const bytes&, size_t off = 0, size_t n = bytes::npos);
+bytes calculate_sha256(bytes_view);
+bytes hmac_sha256(bytes_view msg, bytes_view key);
+
+future<temporary_buffer<char>> read_text_file_fully(const sstring&);
+future<> write_text_file_fully(const sstring&, temporary_buffer<char>);
+future<> write_text_file_fully(const sstring&, const sstring&);
+
+std::optional<std::chrono::milliseconds> parse_expiry(std::optional<std::string>);
+
+class symmetric_key;
+struct key_info;
+
+using options = std::map<sstring, sstring>;
+using opt_bytes = std::optional<bytes>;
+using key_ptr = shared_ptr<symmetric_key>;
+
+/**
+ * wrapper for "options" (map) to provide an
+ * interface returning empty optionals for
+ * non-available values. Makes query simpler
+ * and allows .value_or(...)-statements, which
+ * are neat for default values.
+ *
+ * In the long run one could contemplate
+ * using non-std maps with similar built-in
+ * functionality for all our various configs
+ * in the system, but for now we are firmly
+ * entrenched in map<string, string>
+ */
+template<typename Map>
+class map_wrapper {
+    const Map& _options;
+public:
+    using mapped_type = typename Map::mapped_type;
+    using key_type = typename Map::key_type;
+
+    map_wrapper(const Map& opts)
+        : _options(opts)
+    {}
+
+    std::optional<mapped_type> operator()(const key_type& k) const {
+        auto i = _options.find(k);
+        if (i != _options.end()) {
+            return i->second;
+        }
+        return std::nullopt;
+    }
+};
+
+using opt_wrapper = map_wrapper<options>;
+
+key_info get_key_info(const options&);
+
+class encryption_context;
+
+class key_provider {
+public:
+    virtual ~key_provider()
+    {}
+    virtual future<std::tuple<key_ptr, opt_bytes>> key(const key_info&, opt_bytes = {}) = 0;
+    virtual future<> validate() const {
+        return make_ready_future<>();
+    }
+    virtual bool should_delay_read(const opt_bytes&) const {
+        return false;
+    }
+private:
+    friend std::ostream& operator<<(std::ostream&, const key_provider&);
+    virtual void print(std::ostream&) const = 0;
+};
+
+std::ostream& operator<<(std::ostream&, const key_provider&);
+
+}
+
+template <> struct fmt::formatter<encryption::key_provider> : fmt::ostream_formatter {};
+
+namespace encryption {
+
+class key_provider_factory {
+public:
+    virtual ~key_provider_factory()
+    {}
+    virtual shared_ptr<key_provider> get_provider(encryption_context& c, const options&) = 0;
+};
+
+class encryption_config;
+class system_key;
+class kmip_host;
+class kms_host;
+class gcp_host;
+
+/**
+ * Context is a singleton object, shared across shards. I.e. even though there are obvious mutating
+ * calls in it, it guarantees thread/shard safety.
+ *
+ * Why is this not a sharded thingamajing? Because its own instance methods need to send itself
+ * as a shard-safe object forwards, and thus need to know that same shard, which breaks the circle of
+ * ownership and stuff.
+ */
+class encryption_context {
+public:
+    virtual ~encryption_context() = default;
+    virtual shared_ptr<key_provider> get_provider(const options&) = 0;
+    virtual shared_ptr<system_key> get_system_key(const sstring&) = 0;
+    virtual shared_ptr<kmip_host> get_kmip_host(const sstring&) = 0;
+    virtual shared_ptr<kms_host> get_kms_host(const sstring&) = 0;
+    virtual shared_ptr<gcp_host> get_gcp_host(const sstring&) = 0;
+
+    virtual shared_ptr<key_provider> get_cached_provider(const sstring& id) const = 0;
+    virtual void cache_provider(const sstring& id, shared_ptr<key_provider>) = 0;
+
+    virtual const encryption_config& config() const = 0;
+    virtual shared_ptr<symmetric_key> get_config_encryption_key() const = 0;
+
+    virtual distributed<cql3::query_processor>& get_query_processor() const = 0;
+    virtual distributed<service::storage_service>& get_storage_service() const = 0;
+    virtual distributed<replica::database>& get_database() const = 0;
+    virtual distributed<service::migration_manager>& get_migration_manager() const = 0;
+
+    sstring maybe_decrypt_config_value(const sstring&) const;
+
+    virtual future<> start() = 0;
+    virtual future<> stop() = 0;
+};
+
+future<seastar::shared_ptr<encryption_context>>
+register_extensions(const db::config&, std::unique_ptr<encryption_config>, db::extensions&, const ::service_set&);
+
+// for testing
+std::string encryption_provider(const sstables::sstable&);
+}
+
--- a/ent/encryption/encryption_config.cc
+++ b/ent/encryption/encryption_config.cc
@@ -0,0 +1,164 @@
+/*
+ * Copyright (C) 2015 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#include <boost/filesystem.hpp>
+
+#include "db/config.hh"
+#include "utils/config_file_impl.hh"
+
+#include "init.hh"
+#include "encryption_config.hh"
+#include "encryption.hh"
+
+#include <fmt/ranges.h>
+
+encryption::encryption_config::encryption_config()
+                : config_file()
+// BEGIN entry definitions
+
+        , system_key_directory(this, "system_key_directory", value_status::Used, "/etc/scylla/conf/resources/system_keys",
+                                R"foo(The directory where system keys are kept
+
+This directory should have 700 permissions and belong to the scylla user)foo")
+
+		, config_encryption_active(this, "config_encryption_active", value_status::Used, false, "")
+
+        , config_encryption_key_name(this, "config_encryption_key_name", value_status::Used, "system_key",
+                                "Set to the local encryption key filename or KMIP key URL to use for configuration file property value decryption")
+
+        , system_info_encryption(this, "system_info_encryption", value_status::Used,
+                                { { "enabled", "false" }, { CIPHER_ALGORITHM,
+                                                "AES/CBC/PKCS5Padding" }, {
+                                                SECRET_KEY_STRENGTH, "128" },
+                                                },
+                                R"foo(System information encryption settings
+
+If enabled, system tables that may contain sensitive information (system.batchlog,
+system.paxos), hints files and commit logs are encrypted with the
+encryption settings below.
+
+When enabling system table encryption on a node with existing data, run
+`nodetool upgradesstables -a` on the listed tables to encrypt existing data.
+
+When tracing is enabled, sensitive info will be written into the tables in the
+system_traces keyspace. Those tables should be configured to encrypt their data
+on disk.
+
+It is recommended to use remote encryption keys from a KMIP server when using 
+Transparent Data Encryption (TDE) features.
+Local key support is provided when a KMIP server is not available.
+
+See the scylla documentation for available key providers and their properties.
+)foo")
+		, kmip_hosts(this, "kmip_hosts", value_status::Used, { },
+                                R"foo(KMIP host(s). 
+
+The unique name of kmip host/cluster that can be referenced in table schema.
+
+host.yourdomain.com={ hosts=<host1[:port]>[, <host2[:port]>...], keyfile=/path/to/keyfile, truststore=/path/to/truststore.pem, key_cache_millis=<cache ms>, timeout=<timeout ms> }:...
+
+The KMIP connection management only supports failover, so all requests will go through a 
+single KMIP server. There is no load balancing, as no KMIP servers (at the time of this writing)
+support read replication, or other strategies for availability.
+
+Hosts are tried in the order they appear here. Add them in the same sequence they'll fail over in.
+
+KMIP requests will fail over/retry 'max_command_retries' times (default 3)
+
+)foo")
+		  , kms_hosts(this, "kms_hosts", value_status::Used, { },
+                                R"foo(KMS host(s). 
+
+The unique name of kms host that can be referenced in table schema.
+
+host.yourdomain.com={ endpoint=<http(s)://host[:port]>, aws_access_key_id=<AWS access id>, aws_secret_access_key=<AWS secret key>, aws_profile<profile>, aws_region=<AWS region>, aws_use_ec2_credentials<bool>, aws_use_ec2_region=<bool>, aws_assume_role_arn=<AWS role arn>, master_key=<alias or id>, keyfile=/path/to/keyfile, truststore=/path/to/truststore.pem, key_cache_millis=<cache ms>, timeout=<timeout ms> }:...
+
+Actual connection can be either an explicit endpoint (<host>:<port>), or selected automatic via aws_region.
+
+If aws_use_ec2_region is true, regions is instead queried from EC2 metadata.
+
+Authentication can be explicit with aws_access_key_id and aws_secret_access_key. Either secret or both can be ommitted
+in which case the provider will try to read them from AWS credentials in ~/.aws/credentials
+
+If aws_use_ec2_credentials is true, authentication is instead queried from EC2 metadata.
+
+If aws_assume_role_arn is set, scylla will issue an AssumeRole command and use the resulting security token for key operations.
+
+master_key is an AWS KMS key id or alias from which all keys used for actual encryption of scylla data will be derived.
+This key must be pre-created with access policy allowing the above AWS id Encrypt, Decrypt and GenerateDataKey operations.
+
+)foo")
+		  , gcp_hosts(this, "gcp_hosts", value_status::Used, { },
+                                R"foo(Google Compute Engine KMS host(s). 
+
+The unique name of GCP kms host that can be referenced in table schema.
+
+gcp_project_id=<GCP project>, gcp_location=<project location>, master_key=<existing key to use for encrypting data keys>, gcp_credentials_file=<credentials json file>, gcp_impersonate_service_account=<impersonate this account for all KMS operations>,keyfile=/path/to/keyfile, truststore=/path/to/truststore.pem, key_cache_millis=<cache ms>, timeout=<timeout ms> }:...
+
+Authentication can be explicit with auth_file or by resolving default credentials (see google docs). 
+
+If use_gcp_machine_credentials is true, authentication is instead queried from GCP metadata.
+
+auth_file can contain either a user, service or impersonated service account.
+
+master_key is an GCP KMS key name from which all keys used for actual encryption of scylla data will be derived.
+This key must be pre-created with access policy allowing the above credentials Encrypt and Decrypt operations.
+
+)foo")
+        , user_info_encryption(this, "user_info_encryption", value_status::Used,
+                                { { "enabled", "false" }, { CIPHER_ALGORITHM,
+                                                "AES/CBC/PKCS5Padding" }, {
+                                                SECRET_KEY_STRENGTH, "128" },
+                                                },
+                                R"foo(Global user table encryption settings. If enabled, all user tables
+                                will be encrypted using the provided settings, unless overridden
+                                by table scylla_encryption_options.)foo")
+        , allow_per_table_encryption(this, "allow_per_table_encryption", value_status::Used, true,
+            "If 'user_info_encryption` is enabled this controls whether specifying per-table encryption using create/alter table is allowed"
+        )
+
+
+// END entry definitions
+{}
+
+static class : public configurable {
+    std::unordered_map<const db::config*, std::unique_ptr<encryption::encryption_config>> _cfgs;
+
+public:
+    void append_options(db::config& cfg, boost::program_options::options_description_easy_init& init) override {
+        // While it is fine for normal execution to have just one, static (us) encryption config, 
+        // it does not work well with unit testing, where we repeatedly create new cql_test_envs etc,
+        // since new config values will not be overwritten due to the actual named_values being shared here.
+        // Fix this (temporarily) by simply keeping a local map cfg->ecfg and using these.
+        // TODO: improve this by allowing db::config to hold named sub->configs (mapping config file objects).
+        if (_cfgs.count(&cfg)) {
+            throw std::runtime_error("Config already processed");
+        }
+        auto& ccfg = _cfgs.emplace(&cfg, std::make_unique<encryption::encryption_config>()).first->second;
+        // hook into main scylla.yaml.
+        cfg.add(ccfg->values());
+    }
+    future<notify_func> initialize_ex(const boost::program_options::variables_map& opts, const db::config& cfg, db::extensions& exts, const service_set& services) override {
+        auto ccfg = _cfgs.count(&cfg) ? std::move(_cfgs.at(&cfg)) : std::make_unique<encryption::encryption_config>();
+        _cfgs.erase(&cfg);
+        auto ctxt = co_await encryption::register_extensions(cfg, std::move(ccfg), exts, services);
+        co_return [ctxt](system_state e) -> future<> {
+            switch (e) {
+                case system_state::started:
+                    co_await ctxt->start();
+                    break;
+                case system_state::stopped:
+                    co_await ctxt->stop();
+                    break;
+                default:
+                    break;
+            }
+        };
+    }
+} cfg;
--- a/ent/encryption/encryption_config.hh
+++ b/ent/encryption/encryption_config.hh
@@ -0,0 +1,33 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include "../../utils/config_file.hh"
+
+namespace encryption {
+
+class encryption_config : public utils::config_file {
+public:
+    encryption_config();
+
+    typedef std::unordered_map<sstring, string_map> string_string_map;
+
+    named_value<sstring> system_key_directory;
+    named_value<bool> config_encryption_active;
+    named_value<sstring> config_encryption_key_name;
+    named_value<string_map> system_info_encryption;
+    named_value<string_string_map> kmip_hosts;
+    named_value<string_string_map> kms_hosts;
+    named_value<string_string_map> gcp_hosts;
+    named_value<string_map> user_info_encryption;
+    named_value<bool> allow_per_table_encryption;
+};
+
+}
--- a/ent/encryption/encryption_exceptions.hh
+++ b/ent/encryption/encryption_exceptions.hh
@@ -0,0 +1,55 @@
+/*
+ * Copyright (C) 2024 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include "db/extensions.hh"
+
+namespace encryption {
+
+using base_error = db::extension_storage_exception;
+
+class permission_error : public db::extension_storage_permission_error {
+public:
+    using mybase = db::extension_storage_permission_error;
+    using mybase::mybase;
+};
+
+class configuration_error : public db::extension_storage_misconfigured {
+public:
+    using mybase = db::extension_storage_misconfigured;
+    using mybase::mybase;
+};
+
+class service_error : public base_error {
+public:
+    using base_error::base_error;
+};
+
+class missing_resource_error : public db::extension_storage_resource_unavailable {
+public:
+    using mybase = db::extension_storage_resource_unavailable;
+    using mybase::mybase;
+};
+
+// #4970 - not 100% correct, but network errors are 
+// generally intermittent/recoverable. Mark as a non-isolating
+// error.
+class network_error : public missing_resource_error {
+public:
+    using missing_resource_error::missing_resource_error;
+};
+
+class malformed_response_error : public service_error {
+public:
+    using service_error::service_error;
+};
+
+}
+
--- a/ent/encryption/gcp_host.cc
+++ b/ent/encryption/gcp_host.cc
--- a/ent/encryption/gcp_host.hh
+++ b/ent/encryption/gcp_host.hh
@@ -0,0 +1,80 @@
+/*
+ * Copyright (C) 2024 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once 
+
+#include <vector>
+#include <optional>
+#include <chrono>
+#include <iosfwd>
+#include <string>
+
+#include <seastar/core/future.hh>
+#include <seastar/core/shared_ptr.hh>
+
+#include "symmetric_key.hh"
+
+namespace encryption {
+
+class encryption_context;
+struct key_info;
+
+class gcp_host {
+public:
+    class impl;
+
+    template<typename T>
+    struct t_credentials_source {
+        // Path to credentials JSON file (exported from gcloud console)
+        T gcp_credentials_file;
+        // Optional service account (email address) to impersonate
+        T gcp_impersonate_service_account;
+    };
+
+    using credentials_source = t_credentials_source<std::string>;
+
+    struct host_options : public credentials_source {
+        std::string gcp_project_id;
+        std::string gcp_location;
+
+        // GCP KMS Key to encrypt data keys with. Format: <keychain>/<keyname>
+        std::string master_key;
+
+        // tls. if unspeced, use system for https
+        // GCP does not (afaik?) allow certificate auth
+        // but we keep the option available just in case.
+        std::string certfile;
+        std::string keyfile;
+        std::string truststore;
+        std::string priority_string;
+
+        std::optional<std::chrono::milliseconds> key_cache_expiry;
+        std::optional<std::chrono::milliseconds> key_cache_refresh;
+    };
+
+    using id_type = bytes;
+
+    gcp_host(encryption_context&, const std::string& name, const host_options&);
+    gcp_host(encryption_context&, const std::string& name, const std::unordered_map<sstring, sstring>&);
+    ~gcp_host();
+
+    future<> init();
+    const host_options& options() const;
+
+    struct option_override : public t_credentials_source<std::optional<std::string>> {
+        std::optional<std::string> master_key;
+    };
+
+    future<std::tuple<shared_ptr<symmetric_key>, id_type>> get_or_create_key(const key_info&, const option_override* = nullptr);
+    future<shared_ptr<symmetric_key>> get_key_by_id(const id_type&, const key_info&, const option_override* = nullptr);
+private:
+    std::unique_ptr<impl> _impl;
+};
+
+}
--- a/ent/encryption/gcp_key_provider.cc
+++ b/ent/encryption/gcp_key_provider.cc
@@ -0,0 +1,77 @@
+/*
+ * Copyright (C) 2024 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#include <boost/lexical_cast.hpp>
+#include <regex>
+
+#include "gcp_key_provider.hh"
+#include "gcp_host.hh"
+
+namespace encryption {
+
+class gcp_key_provider : public key_provider {
+public:
+    gcp_key_provider(::shared_ptr<gcp_host> gcp_host, std::string name, gcp_host::option_override oov)
+        : _gcp_host(std::move(gcp_host))
+        , _name(std::move(name))
+        , _oov(std::move(oov))
+    {}
+    future<std::tuple<key_ptr, opt_bytes>> key(const key_info& info, opt_bytes id) override {
+        if (id) {
+            return _gcp_host->get_key_by_id(*id, info, &_oov).then([id](key_ptr k) {
+                return make_ready_future<std::tuple<key_ptr, opt_bytes>>(std::tuple(k, id));
+            });
+        }
+        return _gcp_host->get_or_create_key(info, &_oov).then([](std::tuple<key_ptr, opt_bytes> k_id) {
+            return make_ready_future<std::tuple<key_ptr, opt_bytes>>(k_id);
+        });
+    }
+    void print(std::ostream& os) const override {
+        os << _name;
+    }
+private:
+    ::shared_ptr<gcp_host> _gcp_host;
+    std::string _name;
+    gcp_host::option_override _oov;
+};
+
+shared_ptr<key_provider> gcp_key_provider_factory::get_provider(encryption_context& ctxt, const options& map) {
+    opt_wrapper opts(map);
+    auto gcp_host = opts("gcp_host");
+
+
+    gcp_host::option_override oov {
+        .master_key =  opts("master_key"),
+    };
+
+    oov.gcp_credentials_file = opts("gcp_credentials_file");
+    oov.gcp_impersonate_service_account = opts("gcp_impersonate_service_account");
+
+    if (!gcp_host) {
+        throw std::invalid_argument("gcp_host must be provided");
+    }
+
+    auto host = ctxt.get_gcp_host(*gcp_host);
+    auto id = gcp_host.value()
+        + ":" + oov.master_key.value_or(host->options().master_key)
+        + ":" + oov.gcp_credentials_file.value_or(host->options().gcp_credentials_file)
+        + ":" + oov.gcp_impersonate_service_account.value_or(host->options().gcp_impersonate_service_account)
+        ;
+
+    auto provider = ctxt.get_cached_provider(id);
+
+    if (!provider) {
+        provider = ::make_shared<gcp_key_provider>(host, *gcp_host, std::move(oov));
+        ctxt.cache_provider(id, provider);
+    }
+
+    return provider;
+}
+
+}
--- a/ent/encryption/gcp_key_provider.hh
+++ b/ent/encryption/gcp_key_provider.hh
@@ -0,0 +1,25 @@
+/*
+ * Copyright (C) 2024 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include "encryption.hh"
+#include "system_key.hh"
+
+namespace encryption {
+
+class gcp_key_provider_factory : public key_provider_factory {
+public:
+    shared_ptr<key_provider> get_provider(encryption_context&, const options&) override;
+};
+
+/**
+ * See comment for AWS KMS regarding system key support.
+ */
+}
--- a/ent/encryption/kmip_host.cc
+++ b/ent/encryption/kmip_host.cc
--- a/ent/encryption/kmip_host.hh
+++ b/ent/encryption/kmip_host.hh
@@ -0,0 +1,80 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include <vector>
+#include <optional>
+#include <chrono>
+#include <iosfwd>
+
+#include <fmt/core.h>
+#include <fmt/ostream.h>
+
+#include <seastar/core/future.hh>
+#include <seastar/core/sstring.hh>
+#include <seastar/core/shared_ptr.hh>
+
+#include "../../bytes.hh"
+
+#include "symmetric_key.hh"
+
+namespace encryption {
+
+class symmetric_key;
+class encryption_context;
+struct key_info;
+
+class kmip_host {
+public:
+    struct host_options {
+        std::vector<sstring> hosts;
+
+        sstring username;
+        sstring password;
+
+        sstring certfile;
+        sstring keyfile;
+        sstring truststore;
+        sstring priority_string;
+
+        std::optional<std::chrono::milliseconds> key_cache_expiry;
+        std::optional<std::chrono::milliseconds> key_cache_refresh;
+
+        std::optional<size_t> max_pooled_connections_per_host;
+        std::optional<size_t> max_command_retries;
+    };
+    struct key_options {
+        sstring template_name;
+        sstring key_namespace;
+    };
+    using id_type = bytes;
+
+    kmip_host(encryption_context&, const sstring& name, const host_options&);
+    kmip_host(encryption_context&, const sstring& name, const std::unordered_map<sstring, sstring>&);
+    ~kmip_host();
+
+    future<> connect();
+    future<> disconnect();
+    future<std::tuple<shared_ptr<symmetric_key>, id_type>> get_or_create_key(const key_info&, const key_options& = {});
+    future<shared_ptr<symmetric_key>> get_key_by_id(const id_type&, std::optional<key_info> = std::nullopt);
+
+    /** for system key(s) */
+    future<shared_ptr<symmetric_key>> get_key_by_name(const sstring&);
+
+private:
+    class impl;
+    std::unique_ptr<impl> _impl;
+};
+
+std::ostream& operator<<(std::ostream&, const kmip_host::key_options&);
+
+}
+
+template <> struct fmt::formatter<encryption::kmip_host::key_options> : fmt::ostream_formatter {};
--- a/ent/encryption/kmip_key_provider.cc
+++ b/ent/encryption/kmip_key_provider.cc
@@ -0,0 +1,119 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#include <boost/lexical_cast.hpp>
+#include <regex>
+
+#include "utils/UUID.hh"
+#include "utils/UUID_gen.hh"
+
+#include "kmip_key_provider.hh"
+#include "kmip_host.hh"
+
+namespace encryption {
+
+class kmip_key_provider : public key_provider {
+public:
+    kmip_key_provider(::shared_ptr<kmip_host> kmip_host, kmip_host::key_options kopts, sstring name)
+        : _kmip_host(std::move(kmip_host))
+        , _kopts(std::move(kopts))
+        , _name(std::move(name))
+    {}
+    future<std::tuple<key_ptr, opt_bytes>> key(const key_info& info, opt_bytes id) override {
+        if (id) {
+            return _kmip_host->get_key_by_id(*id, info).then([id](key_ptr k) {
+                return make_ready_future<std::tuple<key_ptr, opt_bytes>>(std::tuple(k, id));
+            });
+        }
+        return _kmip_host->get_or_create_key(info, _kopts).then([](std::tuple<key_ptr, opt_bytes> k_id) {
+            return make_ready_future<std::tuple<key_ptr, opt_bytes>>(k_id);
+        });
+    }
+    void print(std::ostream& os) const override {
+        os << _name;
+        if (!_kopts.key_namespace.empty()) {
+            os << ", namespace=" << _kopts.key_namespace;
+        }
+        if (!_kopts.template_name.empty()) {
+            os << ", template=" << _kopts.template_name;
+        }
+    }
+
+private:
+    ::shared_ptr<kmip_host> _kmip_host;
+    kmip_host::key_options _kopts;
+    sstring _name;
+};
+
+
+shared_ptr<key_provider> kmip_key_provider_factory::get_provider(encryption_context& ctxt, const options& map) {
+    opt_wrapper opts(map);
+    auto host = opts(HOST_NAME);
+    if (!host) {
+        throw std::invalid_argument("kmip_host must be provided");
+    }
+    kmip_host::key_options kopts = {
+                    opts(TEMPLATE_NAME).value_or(""),
+                    opts(KEY_NAMESPACE).value_or(""),
+    };
+
+    auto cache_key = *host + ":" + boost::lexical_cast<std::string>(kopts);
+    auto provider = ctxt.get_cached_provider(cache_key);
+
+    if (!provider) {
+        provider = ::make_shared<kmip_key_provider>(ctxt.get_kmip_host(*host), std::move(kopts), *host);
+        ctxt.cache_provider(cache_key, provider);
+    }
+
+    return provider;
+}
+
+static std::optional<std::pair<sstring, sstring>> parse_kmip_host_and_path(const sstring & s) {
+    static const std::regex kmip_ex("kmip://([^/]+)/([\\w/]+)");
+
+    std::match_results<typename sstring::const_iterator> m;
+    if (std::regex_match(s.begin(), s.end(), m, kmip_ex)) {
+        return std::make_pair(sstring(m[1]), sstring(m[2]));
+    }
+    return std::nullopt;
+}
+
+kmip_system_key::kmip_system_key(encryption_context& ctxt, const sstring& s) {
+    auto p = parse_kmip_host_and_path(s);
+    if (!p) {
+        throw std::invalid_argument("Not a kmip path: " + s);
+    }
+
+    _host = ctxt.get_kmip_host(p->first);
+    _name = p->second;
+}
+
+kmip_system_key::~kmip_system_key() = default;
+
+bool kmip_system_key::is_kmip_path(const sstring& s) {
+    return parse_kmip_host_and_path(s) != std::nullopt;
+}
+
+future<shared_ptr<symmetric_key>> kmip_system_key::get_key() {
+    if (_key) {
+        return make_ready_future<shared_ptr<symmetric_key>>(_key);
+    }
+    return _host->get_key_by_name(_name).then([this](shared_ptr<symmetric_key> k) {
+        _key = k;
+        return k;
+    });
+}
+
+const sstring& kmip_system_key::name() const {
+    return _name;
+}
+
+
+}
+
--- a/ent/encryption/kmip_key_provider.hh
+++ b/ent/encryption/kmip_key_provider.hh
@@ -0,0 +1,40 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+#pragma once
+
+#include "encryption.hh"
+#include "system_key.hh"
+
+namespace encryption {
+
+class kmip_key_provider_factory : public key_provider_factory {
+public:
+    shared_ptr<key_provider> get_provider(encryption_context&, const options&) override;
+};
+
+class kmip_host;
+
+class kmip_system_key : public system_key {
+    shared_ptr<symmetric_key> _key;
+    shared_ptr<kmip_host> _host;
+    sstring _name;
+public:
+    kmip_system_key(encryption_context&, const sstring&);
+    ~kmip_system_key();
+
+    static bool is_kmip_path(const sstring&);
+
+    future<shared_ptr<symmetric_key>> get_key() override;
+    const sstring& name() const override;
+    bool is_local() const override {
+        return false;
+    }
+};
+
+}
--- a/ent/encryption/kms_host.cc
+++ b/ent/encryption/kms_host.cc
--- a/ent/encryption/kms_host.hh
+++ b/ent/encryption/kms_host.hh
@@ -0,0 +1,80 @@
+/*
+ * Copyright (C) 2022 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once 
+
+#include <vector>
+#include <optional>
+#include <chrono>
+#include <iosfwd>
+#include <string>
+
+#include <seastar/core/future.hh>
+#include <seastar/core/shared_ptr.hh>
+
+#include "symmetric_key.hh"
+
+namespace encryption {
+
+class encryption_context;
+struct key_info;
+
+class kms_host {
+public:
+    struct host_options {
+        std::string endpoint;
+        // or...
+        std::string host;
+        uint16_t    port;
+        bool        https = true;
+        // auth
+        std::string aws_access_key_id;
+        std::string aws_secret_access_key;
+        std::string aws_region;
+        std::string aws_profile;
+        std::string aws_assume_role_arn;
+
+        bool        aws_use_ec2_credentials;
+        bool        aws_use_ec2_region;
+
+        // key to use for keys
+        std::string master_key;
+        // tls. if unspeced, use system for https
+        // AWS does not (afaik?) allow certificate auth
+        // but we keep the option available just in case.
+        std::string certfile;
+        std::string keyfile;
+        std::string truststore;
+        std::string priority_string;
+
+        std::optional<std::chrono::milliseconds> key_cache_expiry;
+        std::optional<std::chrono::milliseconds> key_cache_refresh;
+    };
+    using id_type = bytes;
+
+    kms_host(encryption_context&, const std::string& name, const host_options&);
+    kms_host(encryption_context&, const std::string& name, const std::unordered_map<sstring, sstring>&);
+    ~kms_host();
+
+    future<> init();
+    const host_options& options() const;
+
+    struct option_override {
+        std::optional<std::string> master_key;
+        std::optional<std::string> aws_assume_role_arn;
+    };
+
+    future<std::tuple<shared_ptr<symmetric_key>, id_type>> get_or_create_key(const key_info&, const option_override* = nullptr);
+    future<shared_ptr<symmetric_key>> get_key_by_id(const id_type&, const key_info&, const option_override* = nullptr);
+private:
+    class impl;
+    std::unique_ptr<impl> _impl;
+};
+
+}
--- a/ent/encryption/kms_key_provider.cc
+++ b/ent/encryption/kms_key_provider.cc
@@ -0,0 +1,71 @@
+/*
+ * Copyright (C) 2022 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#include <boost/lexical_cast.hpp>
+#include <regex>
+
+#include "kms_key_provider.hh"
+#include "kms_host.hh"
+
+namespace encryption {
+
+class kms_key_provider : public key_provider {
+public:
+    kms_key_provider(::shared_ptr<kms_host> kms_host, std::string name, kms_host::option_override oov)
+        : _kms_host(std::move(kms_host))
+        , _name(std::move(name))
+        , _oov(std::move(oov))
+    {}
+    future<std::tuple<key_ptr, opt_bytes>> key(const key_info& info, opt_bytes id) override {
+        if (id) {
+            return _kms_host->get_key_by_id(*id, info, &_oov).then([id](key_ptr k) {
+                return make_ready_future<std::tuple<key_ptr, opt_bytes>>(std::tuple(k, id));
+            });
+        }
+        return _kms_host->get_or_create_key(info, &_oov).then([](std::tuple<key_ptr, opt_bytes> k_id) {
+            return make_ready_future<std::tuple<key_ptr, opt_bytes>>(k_id);
+        });
+    }
+    void print(std::ostream& os) const override {
+        os << _name;
+    }
+private:
+    ::shared_ptr<kms_host> _kms_host;
+    std::string _name;
+    kms_host::option_override _oov;
+};
+
+shared_ptr<key_provider> kms_key_provider_factory::get_provider(encryption_context& ctxt, const options& map) {
+    opt_wrapper opts(map);
+    auto kms_host = opts("kms_host");
+    kms_host::option_override oov {
+        .master_key = opts("master_key"),
+        .aws_assume_role_arn = opts("aws_assume_role_arn"),
+    };
+
+    if (!kms_host) {
+        throw std::invalid_argument("kms_host must be provided");
+    }
+
+    auto host = ctxt.get_kms_host(*kms_host);
+    auto id = kms_host.value() 
+        + ":" + oov.master_key.value_or(host->options().master_key)
+        + ":" + oov.aws_assume_role_arn.value_or(host->options().aws_assume_role_arn)
+        ;
+    auto provider = ctxt.get_cached_provider(id);
+
+    if (!provider) {
+        provider = ::make_shared<kms_key_provider>(host, *kms_host, std::move(oov));
+        ctxt.cache_provider(id, provider);
+    }
+
+    return provider;
+}
+
+}
--- a/ent/encryption/kms_key_provider.hh
+++ b/ent/encryption/kms_key_provider.hh
@@ -0,0 +1,37 @@
+/*
+ * Copyright (C) 2022 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include "encryption.hh"
+#include "system_key.hh"
+
+namespace encryption {
+
+class kms_key_provider_factory : public key_provider_factory {
+public:
+    shared_ptr<key_provider> get_provider(encryption_context&, const options&) override;
+};
+
+/**
+ * As it stands today, given system_key api (gives keys), and
+ * what it is used for (config encryption), we cannot provide
+ * a KMS system key. This is because:
+ * 
+ * a.) KMS does not allow us to store a named object (key) in a secure(ish) way.
+ *     We can encrypt/decrypt and create one-off keys for local usage, which are
+ *     encoded in their own ID (see kms_host), but having a unique key from
+ *     a "path" is not possible. Esp. due to key rotation, encrypted data preamble
+ *     etc. We could keep the encrypted key material in a local file, then decrypt
+ *     it using a named key on startup, but given b.) it is dubious if this is useful.
+ * b.) System keys are only used for config encryption. The authentication config for
+ *     AWS/KMS access is typically one of the things that should be encrypted. Thus 
+ *     we would create a big chicken and egg problem here.
+ */
+}
--- a/ent/encryption/local_file_provider.cc
+++ b/ent/encryption/local_file_provider.cc
@@ -0,0 +1,292 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+#include <unordered_map>
+#include <stdexcept>
+#include <regex>
+
+#include <boost/filesystem.hpp>
+
+#include <openssl/evp.h>
+#include <openssl/rand.h>
+
+#include <seastar/core/semaphore.hh>
+#include <seastar/core/seastar.hh>
+#include <seastar/core/fstream.hh>
+#include <seastar/core/reactor.hh>
+
+#include "local_file_provider.hh"
+#include "symmetric_key.hh"
+#include "encryption.hh"
+#include "encryption_exceptions.hh"
+#include "encryption_config.hh"
+#include "db/config.hh"
+
+namespace encryption {
+
+namespace bfs = std::filesystem;
+
+const sstring default_key_file_path = (bfs::path(db::config::get_conf_dir()) / "data_encryption_keys").string();
+
+static const key_info system_key_info{ "System", 0 };
+
+class local_file_provider : public key_provider {
+public:
+    local_file_provider(encryption_context& ctxt, const bfs::path& path, bool must_exist = false)
+        : local_file_provider(ctxt, sstring(bfs::absolute(path).string()), must_exist)
+    {}
+    local_file_provider(encryption_context& ctxt, const sstring& path, bool must_exist = false)
+        : _ctxt(ctxt)
+        , _path(path)
+        , _sem(1)
+        , _must_exist(must_exist)
+    {}
+    future<std::tuple<key_ptr, opt_bytes>> key(const key_info& info, opt_bytes = {}) override {
+        // TODO: assert options -> my key
+        auto i = _keys.find(info);
+        if (i != _keys.end()) {
+            return make_ready_future<std::tuple<key_ptr, opt_bytes>>(std::tuple(i->second, std::nullopt));
+        }
+        return load_or_create(info).then([](key_ptr k) {
+            return make_ready_future<std::tuple<key_ptr, opt_bytes>>(std::tuple(k, std::nullopt));
+        });
+    }
+    future<> validate() const override {
+        auto f = make_ready_future<>();
+        if (!_must_exist) {
+            return f;
+        }
+        // if we must exist, we don't change. Ok to open from all shards.
+        return f.then([this] {
+            return open_file_dma(_path, open_flags::ro).then([](file f) {
+                return f.close();
+            });
+        }).handle_exception([this](auto ep) {
+            try {
+                std::rethrow_exception(ep);
+            } catch (...) {
+                std::throw_with_nested(missing_resource_error("Could not read '" + _path + "'"));
+            }
+        });
+    }
+
+    const sstring& path() const {
+        return _path;
+    }
+    void print(std::ostream& os) const override {
+        os << "key=" << _path;
+    }
+
+private:
+    future<key_ptr> load_or_create(const key_info&);
+    future<key_ptr> load_or_create_local(const key_info&);
+    future<> read_key_file();
+    future<key_ptr> write_key_file(key_info);
+
+    std::unordered_map<key_info, key_ptr, key_info_hash> _keys;
+    encryption_context& _ctxt;
+    sstring _path;
+    semaphore _sem;
+    bool _read_file = false;
+    bool _must_exist = false;
+};
+
+shared_ptr<key_provider> local_file_provider_factory::find(encryption_context& ctxt, const sstring& path) {
+    auto p = ctxt.get_cached_provider(path);
+    if (!p) {
+        p = make_shared<local_file_provider>(ctxt, path);
+        ctxt.cache_provider(path, p);
+    }
+    return p;
+}
+
+shared_ptr<key_provider> local_file_provider_factory::get_provider(encryption_context& ctxt, const options& map) {
+    opt_wrapper opts(map);
+    return find(ctxt, opts(SECRET_KEY_FILE).value_or(default_key_file_path));
+}
+
+future<key_ptr>
+local_file_provider::load_or_create(const key_info& info) {
+    // if someone uses a system key as a table key, we could still race
+    // here. but that is a user error, so ignore
+    if (this_shard_id() == 0 || &info == &system_key_info) {
+        return load_or_create_local(info);
+    }
+
+    struct data {
+        bytes key;
+        key_info info;
+    };
+
+    /**
+     * Key files are singular. Not sharded. This would be ok if we only read from them.
+     * But in keeping with dse compat, we don't. So rather than dealing with lock files
+     * or whatnot, we simply say that a single file is handled by a single key object,
+     * and only on shard 0. So if we are not shard 0, we call to there, find our
+     * counterpart object (local_file_provider_factory::find), and as him about the
+     * key data instead. He in turn will sync on his semaphore.
+     *
+     * The downside is that we are not resilient against multiple processes messing
+     * with the key file, but neither is dse
+     */
+    return do_with(data{bytes(bytes::initialized_later(), info.len/8), info}, [this](data& i) {
+        return smp::submit_to(0, [this, &i]{
+            auto kp = static_pointer_cast<local_file_provider>(local_file_provider_factory::find(_ctxt, _path));
+            auto f = kp->load_or_create_local(i.info);
+            return f.then([&i, kp](key_ptr k) {
+                auto& kd = k->key();
+                i.key.resize(kd.size());
+                std::copy(kd.begin(), kd.end(), i.key.begin());
+            });
+        }).then([this, &i] {
+            auto k = make_shared<symmetric_key>(i.info, i.key);
+            _keys.emplace(i.info, k);
+            return make_ready_future<key_ptr>(std::move(k));
+        });
+    });
+}
+
+future<key_ptr>
+local_file_provider::load_or_create_local(const key_info& info) {
+    if (_keys.count(info)) {
+        return make_ready_future<key_ptr>(_keys.at(info));
+    }
+    return read_key_file().then([this, info] {
+        if (_keys.count(info)) {
+            return make_ready_future<key_ptr>(_keys.at(info));
+        }
+        if (info == system_key_info) {
+            if (_keys.size() != 1) {
+                _keys.clear();
+                return make_exception_future<key_ptr>(std::invalid_argument("System key must contain exactly one entry"));
+            }
+            auto k = _keys.begin()->second;
+            _keys.clear();
+            _keys.emplace(info, k);
+            return make_ready_future<key_ptr>(k);
+        }
+        // create it.
+        return write_key_file(info);
+    });
+}
+
+future<> local_file_provider::read_key_file() {
+    if (_read_file) {
+        return make_ready_future();
+    }
+
+    // #1923 - a key can have a descriptor string line "AES:128:<data>" iff user relies on
+    // defaults. Must match this as well.
+    static const std::regex key_line_expr(R"foo((\w+(?:\/\w+)?(?:\/\w+)?)\:(\d+)\:(\S+)\s*)foo");
+
+    return with_semaphore(_sem, 1, [this] {
+        // could do this twice, but it is only reading
+        return read_text_file_fully(_path).then([this](temporary_buffer<char> buf) {
+            auto i = std::cregex_iterator(buf.begin(), buf.end(), key_line_expr);
+            auto e = std::cregex_iterator();
+
+            while (i != e) {
+                std::cmatch m = *i;
+                auto alg = m[1].str();
+                auto len = std::stoul(m[2].str());
+                auto key = m[3].str();
+
+                auto info = key_info{alg, unsigned(len)};
+                if (!_keys.count(info)) {
+                    auto kb = base64_decode(key);
+                    auto k = make_shared<symmetric_key>(info, kb);
+                    _keys.emplace(info, std::move(k));
+                }
+                ++i;
+            }
+            _read_file = true;
+        }).handle_exception([this](auto ep) {
+            try {
+                std::rethrow_exception(ep);
+            } catch (std::system_error& e) {
+                if (e.code() == std::error_code(ENOENT, std::system_category())) {
+                    if (!_must_exist) {
+                        return;
+                    }
+                    std::throw_with_nested(configuration_error("Key file '" + _path + "' does not exist"));
+                }
+                std::throw_with_nested(service_error("read_key_file"));
+            } catch (std::invalid_argument& e) {
+                std::throw_with_nested(configuration_error(fmt::format("read_key_file: {}", e.what())));
+            } catch (...) {
+                std::throw_with_nested(service_error(fmt::format("read_key_file: {}", std::current_exception())));
+            }
+        });
+    });
+}
+
+future<key_ptr> local_file_provider::write_key_file(key_info info) {
+    return with_semaphore(_sem, 1, [this, info] {
+        // we can get here more than once if shards race.
+        // however, we only need to use/write the first key matching
+        // the required info.
+        if (_keys.count(info)) {
+            return make_ready_future<key_ptr>(_keys.at(info));
+        }
+
+        auto k = make_shared<symmetric_key>(info);
+
+        std::ostringstream ss;
+        for (auto& p : _keys) {
+            ss << p.first.alg << ":" << p.first.len << ":" << base64_encode(p.second->key()) << std::endl;
+        }
+        ss << info.alg << ":" << info.len << ":" << base64_encode(k->key()) << std::endl;
+        auto s = ss.str();
+        auto tmpnam = _path + ".tmp";
+        auto f = make_ready_future<>();
+        if (!_must_exist) {
+            f = seastar::recursive_touch_directory((bfs::path(tmpnam).remove_filename()).string());
+        }
+        return f.then([this, tmpnam, s] {
+            return write_text_file_fully(tmpnam, s).then([this, tmpnam] {
+                return rename_file(tmpnam, _path);
+            });
+        }).then([this, k, info] {
+            // don't cache until written
+            _keys[info] = k;
+            return make_ready_future<key_ptr>(k);
+        });
+    }).handle_exception([this](auto ep) -> key_ptr{
+        try {
+            std::rethrow_exception(ep);
+        } catch (...) {
+            std::throw_with_nested(service_error("Could not write key file '" + _path + "'"));
+        }
+    });
+}
+
+local_system_key::local_system_key(encryption_context& ctxt, const sstring& path)
+    : _provider(make_shared<local_file_provider>(ctxt, bfs::path(ctxt.config().system_key_directory()) / bfs::path(path), true))
+{}
+
+local_system_key::~local_system_key()
+{}
+
+future<shared_ptr<symmetric_key>> local_system_key::get_key() {
+    return _provider->key(system_key_info).then([](std::tuple<key_ptr, opt_bytes> k_id) {
+       return make_ready_future<shared_ptr<symmetric_key>>(std::get<0>(std::move(k_id)));
+    });
+}
+
+future<> local_system_key::validate() const {
+    // first, just validate the file provider itself
+    co_await _provider->validate();
+    // second, do an early load of the actual key to ensure file contents.
+    co_await _provider->key(system_key_info);
+}
+
+const sstring& local_system_key::name() const {
+    return _provider->path();
+}
+
+}
--- a/ent/encryption/local_file_provider.hh
+++ b/ent/encryption/local_file_provider.hh
@@ -0,0 +1,41 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include "encryption.hh"
+#include "system_key.hh"
+
+namespace encryption {
+
+const extern sstring default_key_file_path;
+
+class local_file_provider;
+
+class local_file_provider_factory : public key_provider_factory {
+public:
+    static shared_ptr<key_provider> find(encryption_context&, const sstring& path);
+    shared_ptr<key_provider> get_provider(encryption_context&, const options&) override;
+};
+
+class local_system_key : public system_key {
+    shared_ptr<local_file_provider> _provider;
+public:
+    local_system_key(encryption_context&, const sstring&);
+    ~local_system_key();
+
+    future<shared_ptr<symmetric_key>> get_key() override;
+    future<> validate() const override;
+    const sstring& name() const override;
+    bool is_local() const override {
+        return true;
+    }
+};
+
+}
--- a/ent/encryption/replicated_key_provider.cc
+++ b/ent/encryption/replicated_key_provider.cc
@@ -0,0 +1,477 @@
+/*
+ * Copyright (C) 2015 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+#include <unordered_map>
+#include <stdexcept>
+#include <regex>
+#include <tuple>
+
+#include <boost/filesystem.hpp>
+
+#include <openssl/evp.h>
+#include <openssl/rand.h>
+
+#include <seastar/core/semaphore.hh>
+#include <seastar/core/seastar.hh>
+#include <seastar/core/fstream.hh>
+#include <seastar/core/reactor.hh>
+#include <seastar/core/coroutine.hh>
+
+#include <fmt/ranges.h>
+#include <fmt/std.h>
+#include "utils/to_string.hh"
+
+#include "replicated_key_provider.hh"
+#include "encryption.hh"
+#include "encryption_exceptions.hh"
+#include "local_file_provider.hh"
+#include "symmetric_key.hh"
+#include "replica/database.hh"
+#include "cql3/query_processor.hh"
+#include "cql3/untyped_result_set.hh"
+#include "utils/UUID.hh"
+#include "utils/UUID_gen.hh"
+#include "utils/hash.hh"
+#include "service/storage_service.hh"
+#include "service/migration_manager.hh"
+#include "compaction/compaction_manager.hh"
+#include "replica/distributed_loader.hh"
+#include "schema/schema_builder.hh"
+#include "db/system_keyspace.hh"
+#include "db/extensions.hh"
+#include "locator/everywhere_replication_strategy.hh"
+
+namespace encryption {
+
+static auto constexpr KSNAME = "system_replicated_keys";
+static auto constexpr TABLENAME = "encrypted_keys";
+
+static logger log("replicated_key_provider");
+
+using utils::UUID;
+
+class replicated_key_provider : public key_provider {
+public:
+    static constexpr int8_t version = 0;
+    /**
+     * Header:
+     * 1 byte version
+     * 16 bytes UUID of key
+     * 16 bytes MD5 of UUID
+     */
+    static const size_t header_size = 33;
+
+    struct key_id {
+        key_info info;
+        opt_bytes id;
+
+        key_id(key_info k, opt_bytes b = {})
+            : info(std::move(k))
+            , id(std::move(b))
+        {}
+        bool operator==(const key_id& v) const {
+            return info == v.info && id == v.id;
+        }
+    };
+
+    struct key_id_hash {
+        size_t operator()(const key_id& id) const {
+            return utils::tuple_hash()(std::tie(id.info.alg, id.info.len, id.id));
+        }
+    };
+
+    replicated_key_provider(encryption_context& ctxt, shared_ptr<system_key> system_key, shared_ptr<key_provider> local_provider)
+        : _ctxt(ctxt)
+        , _system_key(std::move(system_key))
+        , _local_provider(std::move(local_provider))
+    {}
+
+
+    future<std::tuple<key_ptr, opt_bytes>> key(const key_info&, opt_bytes = {}) override;
+    future<> validate() const override;
+    future<> maybe_initialize_tables();
+    static future<> do_initialize_tables(::replica::database& db, service::migration_manager&);
+
+    bool should_delay_read(const opt_bytes& id) const override {
+        if (!id || _initialized) {
+            return false;
+        }
+        if (!_initialized) {
+            return true;
+        }
+        auto& qp = _ctxt.get_query_processor();
+        // This check should be ok, and even somewhat redundant. "Initialized" above
+        // will only be set once we've generated/queried a key not passing through here
+        // (i.e. a key for write _or_ commit log (should we allow this)). This can only be
+        // done if:
+        // a.) Encryption was already set up, thus table existed and we waited
+        //     for distributed_tables in "ensure_populated"
+        // b.) Encryption was added. In which case we are way past bootstrap
+        //     and can receive user commands.
+        // c.) System table/commit log write, with either first use of this provider,
+        //     in which case we're creating the table (here at least) - thus fine,
+        //     or again, we've waited through "ensure_populated", so keys are
+        //     readble. At worst, we create a few extra keys.
+        // Note: currently c.) is not relevant, as we don't support system/commitlog
+        //       encryption using repl_prov.
+        return !qp.local_is_initialized();
+    }
+
+    void print(std::ostream& os) const override {
+        os << "system_key=" << _system_key->name() << ", local=" << *_local_provider;
+    }
+
+private:
+    void store_key(const key_id&, const UUID&, key_ptr);
+
+    static opt_bytes decode_id(const opt_bytes&);
+    static bytes encode_id(const UUID&);
+
+    future<std::tuple<UUID, key_ptr>> get_key(const key_info&, opt_bytes = {});
+
+    future<key_ptr> load_or_create(const key_info&);
+    future<key_ptr> load_or_create_local(const key_info&);
+    future<> read_key_file();
+    future<> write_key_file();
+
+    template<typename... Args>
+    future<::shared_ptr<cql3::untyped_result_set>> query(sstring, Args&& ...);
+
+    future<> force_blocking_flush();
+
+    encryption_context& _ctxt;
+    shared_ptr<system_key> _system_key;
+    shared_ptr<key_provider> _local_provider;
+    std::unordered_map<key_id, std::pair<UUID, key_ptr>, key_id_hash> _keys;
+
+    bool _initialized = false;
+    bool _use_cache = true;
+
+    friend class replicated_key_provider_factory;
+
+    static const utils::UUID local_fallback_uuid;
+    static const bytes local_fallback_id;
+    static const bytes_view local_fallback_bytes;
+};
+
+using namespace std::chrono_literals;
+
+static const timeout_config rkp_db_timeout_config {
+    5s, 5s, 5s, 5s, 5s, 5s, 5s,
+};
+
+static service::query_state& rkp_db_query_state() {
+    static thread_local service::client_state cs(service::client_state::internal_tag{}, rkp_db_timeout_config);
+    static thread_local service::query_state qs(cs, empty_service_permit());
+    return qs;
+}
+
+template<typename... Args>
+future<::shared_ptr<cql3::untyped_result_set>> replicated_key_provider::query(sstring q, Args&& ...params) {
+    auto mode = co_await _ctxt.get_storage_service().local().get_operation_mode();
+    if (mode != service::storage_service::mode::STARTING) {
+        co_return co_await _ctxt.get_query_processor().local().execute_internal(q, { std::forward<Args>(params)...}, cql3::query_processor::cache_internal::no);
+    }
+    co_return co_await _ctxt.get_query_processor().local().execute_internal(q, db::consistency_level::ONE, rkp_db_query_state(), { std::forward<Args>(params)...}, cql3::query_processor::cache_internal::no);
+}
+
+future<> replicated_key_provider::force_blocking_flush() {
+    return _ctxt.get_database().invoke_on_all([](replica::database& db) {
+        // if (!Boolean.getBoolean("cassandra.unsafesystem"))
+        replica::column_family& cf = db.find_column_family(KSNAME, TABLENAME);
+        return cf.flush();
+    });
+}
+
+void replicated_key_provider::store_key(const key_id& id, const UUID& uuid, key_ptr k) {
+    if (!_use_cache) {
+        return;
+    }
+    _keys[id] = std::make_pair(uuid, k);
+    if (!id.id) {
+        _keys[key_id(id.info, uuid.serialize())] = std::make_pair(uuid, k);
+    }
+}
+
+opt_bytes replicated_key_provider::decode_id(const opt_bytes& b) {
+    if (b) {
+        auto i = b->begin();
+        auto v = *i++;
+        if (v == version && b->size() == 33) {
+            bytes id(i + 1, i + 1 + 16);
+            bytes md(i + 1 + 16, b->end());
+            if (calculate_md5(id) == md) {
+                return id;
+            }
+        }
+    }
+    return std::nullopt;
+}
+
+bytes replicated_key_provider::encode_id(const UUID& uuid) {
+    bytes b{bytes::initialized_later(), header_size};
+    auto i = b.begin();
+    *i++ = version;
+    uuid.serialize(i);
+    auto md = calculate_md5(b, 1, 16);
+    std::copy(md.begin(), md.end(), i);
+    return b;
+}
+
+const utils::UUID replicated_key_provider::local_fallback_uuid(0u, 0u); // not valid!
+const bytes replicated_key_provider::local_fallback_id = encode_id(local_fallback_uuid);
+const bytes_view replicated_key_provider::local_fallback_bytes(local_fallback_id.data() + 1, 16);
+
+future<std::tuple<key_ptr, opt_bytes>> replicated_key_provider::key(const key_info& info, opt_bytes input) {
+    opt_bytes id;
+
+    if (input) { //reading header?
+        auto v = *input;
+        if (v[0] == version) {
+            bytes bid(v.begin() + 1, v.begin() + 1 + 16);
+            bytes md(v.begin() + 1 + 16, v.begin() + 1 + 32);
+            if (calculate_md5(bid) == md) {
+                id = bid;
+            }
+        }
+    }
+
+    bool try_local = id == local_fallback_bytes;
+
+    // if the id indicates the key came from local fallback, don't even
+    // try keyspace lookup.
+    if (!try_local) {
+        try {
+            auto [uuid, k] = co_await get_key(info, std::move(id));
+            co_return std::make_tuple(k, encode_id(uuid));
+        } catch (std::invalid_argument& e) {
+            std::throw_with_nested(configuration_error(e.what()));
+        } catch (...) {
+            auto ep = std::current_exception();
+            log.warn("Exception looking up key {}: {}", info, ep);
+            if (_local_provider) {
+                try {
+                    std::rethrow_exception(ep);
+                } catch (replica::no_such_keyspace&) {
+                } catch (exceptions::invalid_request_exception&) {
+                } catch (exceptions::read_failure_exception&) {
+                } catch (...) {
+                    std::throw_with_nested(service_error(fmt::format("key: {}", std::current_exception())));
+                }
+                if (!id) {
+                    try_local = true;
+                }
+            }
+            if (!try_local) {
+                std::throw_with_nested(service_error(fmt::format("key: {}", std::current_exception())));
+            }
+        }
+    }
+
+    log.warn("Falling back to local key {}", info);
+    auto [k, nid] = co_await _local_provider->key(info, id);
+    if (nid && nid != id) {
+        // local provider does not give ids.
+        throw malformed_response_error("Expected null id back from local provider");
+    }
+    co_return std::make_tuple(k, local_fallback_id);
+}
+
+future<std::tuple<UUID, key_ptr>> replicated_key_provider::get_key(const key_info& info, opt_bytes opt_id) {
+    if (!_initialized) {
+        co_await maybe_initialize_tables();
+    }
+
+    key_id id(info, std::move(opt_id));
+    auto i = _keys.find(id);
+    if (i != _keys.end()) {
+        co_return std::tuple(i->second.first, i->second.second);
+    }
+
+    // TODO: origin does non-cql acquire of all available keys from
+    // replicas in the "host_ids" table iff we get here during boot.
+    // For now, ignore this and assume that if we have a sstable with
+    // key X, we should have a local replica of X as well, given
+    // the "everywhere strategy of the keys table.
+
+    auto cipher = info.alg.substr(0, info.alg.find('/')); // e.g. "AES"
+
+    UUID uuid;
+    shared_ptr<cql3::untyped_result_set> res;
+
+    if (id.id) {
+        uuid = utils::UUID_gen::get_UUID(*id.id);
+        log.debug("Finding key {} ({})", uuid, info);
+        auto s = fmt::format("SELECT * FROM {}.{} WHERE key_file=? AND cipher=? AND strength=? AND key_id=?;", KSNAME, TABLENAME);
+        res = co_await query(std::move(s), _system_key->name(), cipher, int32_t(id.info.len), uuid);
+
+        // if we find nothing, and we actually queried a specific key (by uuid), we've failed.
+        if (res->empty()) {
+            log.debug("Could not find key {}", id.id);
+            throw std::runtime_error(fmt::format("Unable to find key for cipher={} strength={} id={}", cipher, id.info.len, uuid));
+        }
+    } else {
+        log.debug("Finding key ({})", info);
+        auto s = fmt::format("SELECT * FROM {}.{} WHERE key_file=? AND cipher=? AND strength=? LIMIT 1;", KSNAME, TABLENAME);
+        res = co_await query(std::move(s), _system_key->name(), cipher, int32_t(id.info.len));
+    }
+
+    // otoh, if we don't need a specific key, we can just create a new one (writing a sstable)
+    if (res->empty()) {
+        uuid = utils::UUID_gen::get_time_UUID();
+
+        log.debug("No key found. Generating {}", uuid);
+
+        auto k = make_shared<symmetric_key>(id.info);
+        store_key(id, uuid, k);
+
+        auto b = co_await _system_key->encrypt(k->key());
+        auto ks = base64_encode(b);
+        log.trace("Inserting generated key {}", uuid);
+        co_await query(fmt::format("INSERT INTO {}.{} (key_file, cipher, strength, key_id, key) VALUES (?, ?, ?, ?, ?)", 
+            KSNAME, TABLENAME), _system_key->name(), cipher, int32_t(id.info.len), uuid, ks
+        );
+        log.trace("Flushing key table");
+        co_await force_blocking_flush();
+
+        co_return std::tuple(uuid, k);
+    }
+
+    // found it
+    auto& row = res->one();
+    uuid = row.get_as<UUID>("key_id");
+    auto ks = row.get_as<sstring>("key");
+    auto kb = base64_decode(ks);
+    auto b = co_await _system_key->decrypt(kb);
+    auto k = make_shared<symmetric_key>(id.info, b);
+    store_key(id, uuid, k);
+
+    co_return std::tuple(uuid, k);
+}
+
+future<> replicated_key_provider::validate() const {
+    try {
+        co_await _system_key->validate();
+    } catch (...) {
+        std::throw_with_nested(std::invalid_argument(fmt::format("Could not validate system key: {}", _system_key->name())));
+    }
+    if (_local_provider){
+        co_await _local_provider->validate();
+    }
+}
+
+schema_ptr encrypted_keys_table() {
+    static thread_local auto schema = [] {
+        auto id = generate_legacy_id(KSNAME, TABLENAME);
+        return schema_builder(KSNAME, TABLENAME, std::make_optional(id))
+                .with_column("key_file", utf8_type, column_kind::partition_key)
+                .with_column("cipher", utf8_type, column_kind::partition_key)
+                .with_column("strength", int32_type, column_kind::clustering_key)
+                .with_column("key_id", timeuuid_type, column_kind::clustering_key)
+                .with_column("key", utf8_type)
+                .with_hash_version()
+                .build();
+    }();
+    return schema;
+}
+
+future<> replicated_key_provider::maybe_initialize_tables() {
+    if (!_initialized) {
+        co_await do_initialize_tables(_ctxt.get_database().local(), _ctxt.get_migration_manager().local());
+        _initialized = true;
+    }
+}
+
+future<> replicated_key_provider::do_initialize_tables(::replica::database& db, service::migration_manager& mm) {
+    if (db.has_schema(KSNAME, TABLENAME)) {
+        co_return;
+    }
+
+    log.debug("Creating keyspace and table");
+    if (!db.has_keyspace(KSNAME)) {
+        auto group0_guard = co_await mm.start_group0_operation();
+        auto ts = group0_guard.write_timestamp();
+        try {
+            auto ksm = keyspace_metadata::new_keyspace(
+                    KSNAME,
+                    "org.apache.cassandra.locator.EverywhereStrategy",
+                    {},
+                    std::nullopt,
+                    true);
+            co_await mm.announce(service::prepare_new_keyspace_announcement(db, ksm, ts), std::move(group0_guard), fmt::format("encryption at rest: create keyspace {}", KSNAME));
+        } catch (exceptions::already_exists_exception&) {
+        }
+    }
+    auto group0_guard = co_await mm.start_group0_operation();
+    auto ts = group0_guard.write_timestamp();
+    try {
+        co_await mm.announce(co_await service::prepare_new_column_family_announcement(mm.get_storage_proxy(), encrypted_keys_table(), ts), std::move(group0_guard),
+                             fmt::format("encryption at rest: create table {}.{}", KSNAME, TABLENAME));
+    } catch (exceptions::already_exists_exception&) {
+    }
+    auto& ks = db.find_keyspace(KSNAME);
+    auto& rs = ks.get_replication_strategy();
+    // should perhaps check name also..
+    if (rs.get_type() != locator::replication_strategy_type::everywhere_topology) {
+        // TODO: reset to everywhere + repair.
+    }
+}
+
+const size_t replicated_key_provider::header_size;
+
+replicated_key_provider_factory::replicated_key_provider_factory()
+{}
+
+replicated_key_provider_factory::~replicated_key_provider_factory()
+{}
+
+namespace bfs = std::filesystem;
+
+shared_ptr<key_provider> replicated_key_provider_factory::get_provider(encryption_context& ctxt, const options& map) {
+    opt_wrapper opts(map);
+    auto system_key_name = opts(SYSTEM_KEY_FILE).value_or("system_key");
+    if (system_key_name.find('/') != sstring::npos) {
+        throw std::invalid_argument("system_key cannot contain '/'");
+    }
+
+    auto system_key = ctxt.get_system_key(system_key_name);
+    auto local_key_file = bfs::absolute(bfs::path(opts(SECRET_KEY_FILE).value_or(default_key_file_path)));
+
+    if (system_key->is_local() && bfs::absolute(bfs::path(system_key->name())) == local_key_file) {
+        throw std::invalid_argument("system key and local key cannot be the same");
+    }
+
+    auto name = system_key->name() + ":" + local_key_file.string();
+    auto debug = opts("DEBUG");
+    if (debug) {
+        name = name + ":" + *debug;
+    }
+    auto p = ctxt.get_cached_provider(name);
+    if (!p) {
+        auto rp = seastar::make_shared<replicated_key_provider>(ctxt, std::move(system_key), local_file_provider_factory::find(ctxt, local_key_file.string()));
+        ctxt.cache_provider(name, rp);
+
+        if (debug && debug->find("nocache") != sstring::npos) {
+            log.debug("Turn off cache");
+            rp->_use_cache = false;
+        }
+        p = std::move(rp);
+    }
+
+    return p;
+}
+
+void replicated_key_provider_factory::init(db::extensions& exts) {
+    exts.add_extension_internal_keyspace(KSNAME);
+}
+
+future<> replicated_key_provider_factory::on_started(::replica::database& db, service::migration_manager& mm) {
+    return replicated_key_provider::do_initialize_tables(db, mm);
+}
+
+}
--- a/ent/encryption/replicated_key_provider.hh
+++ b/ent/encryption/replicated_key_provider.hh
@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2015 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include "encryption.hh"
+
+namespace db {
+class extensions;
+}
+
+namespace replica {
+class database;
+}
+
+namespace service {
+class migration_manager;
+}
+
+namespace encryption {
+
+class replicated_key_provider_factory : public key_provider_factory {
+public:
+    replicated_key_provider_factory();
+    ~replicated_key_provider_factory();
+
+    shared_ptr<key_provider> get_provider(encryption_context&, const options&) override;
+
+    static void init(db::extensions&);
+    static future<> on_started(::replica::database&, service::migration_manager&);
+};
+
+}
--- a/ent/encryption/symmetric_key.cc
+++ b/ent/encryption/symmetric_key.cc
@@ -0,0 +1,396 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+#include <stdexcept>
+#include <regex>
+#include <algorithm>
+
+#include <openssl/evp.h>
+#include <openssl/rand.h>
+#include <openssl/err.h>
+
+#if OPENSSL_VERSION_NUMBER >= (3<<28)
+#   include <openssl/provider.h>
+#endif
+
+#include <seastar/core/align.hh>
+#include <seastar/core/print.hh>
+
+#include "symmetric_key.hh"
+#include "utils/hash.hh"
+
+namespace {
+struct openssl_env {
+    OSSL_PROVIDER* legacy_provider = nullptr;
+    OSSL_PROVIDER* default_provider = nullptr;
+    openssl_env() {
+        OpenSSL_add_all_ciphers();
+#if OPENSSL_VERSION_NUMBER >= (3<<28)
+        legacy_provider = OSSL_PROVIDER_load(NULL, "legacy");
+        default_provider = OSSL_PROVIDER_load(NULL, "default");
+#endif
+    }
+    ~openssl_env() {
+        OSSL_PROVIDER_unload(legacy_provider);
+        OSSL_PROVIDER_unload(default_provider);
+    }
+};
+static const openssl_env ossl_env;
+}
+
+std::ostream& encryption::operator<<(std::ostream& os, const key_info& info) {
+    return os << info.alg << ":" << info.len;
+}
+
+static void throw_evp_error(std::string msg) {
+    auto e = ERR_get_error();
+    if (e != 0) {
+        char buf[512];
+        ERR_error_string_n(e, buf, sizeof(buf));
+        msg += "(" + std::string(buf) + ")";
+    }
+    throw std::runtime_error(msg);
+}
+
+bool encryption::key_info::compatible(const key_info& rhs) const {
+    sstring malg, halg;
+    std::tie(malg, std::ignore, std::ignore) = parse_key_spec(alg);
+    std::tie(halg, std::ignore, std::ignore) = parse_key_spec(rhs.alg);
+    if (malg != halg) {
+        return false;
+    }
+    // If lengths differ we need to actual create keys to
+    // check what the true lengths are. Since openssl and
+    // java designators count different for DES etc.
+    if (len != rhs.len) {
+        symmetric_key k1(*this);
+        symmetric_key k2(rhs);
+        if (k1.key().size() != k2.key().size()) {
+            return false;
+        }
+    }
+    return true;
+}
+
+std::tuple<sstring, sstring, sstring>
+encryption::parse_key_spec(const sstring& alg) {
+    static const std::regex alg_exp(R"foo(^(\w+)(?:\/(\w+))?(?:\/(\w+))?$)foo");
+
+    std::cmatch m;
+    if (!std::regex_match(alg.begin(), alg.end(), m, alg_exp)) {
+        throw std::invalid_argument("Invalid algorithm string: " + alg);
+    }
+
+    auto type = m[1].str();
+    auto mode = m[2].str();
+    auto padd = m[3].str();
+
+    std::transform(type.begin(), type.end(), type.begin(), ::tolower);
+    std::transform(mode.begin(), mode.end(), mode.begin(), ::tolower);
+    std::transform(padd.begin(), padd.end(), padd.begin(), ::tolower);
+
+    static const std::string padding = "padding";
+    if (padd.size() > padding.size() && std::equal(padding.rbegin(), padding.rend(), padd.rbegin())) {
+        padd.resize(padd.size() - padding.size());
+    }
+
+    return std::make_tuple<sstring, sstring, sstring>(type, mode, padd);
+}
+
+std::tuple<sstring, sstring, sstring> encryption::parse_key_spec_and_validate_defaults(const sstring& alg) {
+    auto [type, mode, padd] = parse_key_spec(alg);
+
+    // openssl AND kmip server(s?) does not allow missing block mode. so default one.
+    if (mode.empty()) {
+        mode = "cbc";
+    }
+
+    // OpenSSL only supports one form of padding. We used to just allow
+    // non-empty string -> pkcs5/pcks7. Better to verify
+    // (note: pcks5 is sortof a misnomeanor here, as in the Sun world, it
+    // sort of means "pkcs7 with automatic block size" - which is pretty
+    // much how things are in the OpenSSL universe as well)
+    if (padd == "no") {
+        padd = "";
+    }
+    if (!padd.empty() && padd != "pkcs5" && padd != "pkcs" && padd != "pkcs7") {
+        throw std::invalid_argument("non-supported padding option: " + padd);
+    }
+
+    return { type, mode, padd };
+}
+
+encryption::symmetric_key::symmetric_key(const key_info& info, const bytes& key)
+    : _ctxt(EVP_CIPHER_CTX_new(), &EVP_CIPHER_CTX_free)
+    , _info(info)
+    , _key(key)
+{
+    if (!_ctxt) {
+        throw std::bad_alloc();
+    }
+
+    sstring type, mode, padd;
+    std::tie(type, mode, padd) = parse_key_spec_and_validate_defaults(info.alg);
+
+    // Note: we are using some types here that are explicitly marked as "unsupported - placeholder"
+    // in gnutls.
+
+    // camel case vs. dash
+    if (type == "desede") {
+        type = "des-ede";
+        // and 168-bits desede is ede3 in openssl...
+        if (info.len > 16*8) {
+            type = "des-ede3";
+        }
+    }
+
+    auto str = fmt::format("{}-{}-{}", type, info.len, mode);
+    auto cipher = EVP_get_cipherbyname(str.c_str());
+
+    if (!cipher) {
+        str = fmt::format("{}-{}", type, mode);
+        cipher = EVP_get_cipherbyname(str.c_str());
+    }
+    if (!cipher) {
+        str = fmt::format("{}-{}", type, info.len);
+        cipher = EVP_get_cipherbyname(str.c_str());
+    }
+    if (!cipher) {
+        str = type;
+        cipher = EVP_get_cipherbyname(str.c_str());
+    }
+    if (!cipher) {
+        throw_evp_error("Invalid algorithm: " + info.alg);
+    }
+
+    size_t len = EVP_CIPHER_key_length(cipher);
+
+    if ((_info.len/8) != len) {
+        if (!EVP_CipherInit_ex(*this, cipher, nullptr, nullptr, nullptr, 0)) {
+            throw_evp_error("Could not initialize cipher");
+        }
+        auto dlen = _info.len/8;
+        // Openssl describes des-56 length as 64 (counts parity),
+        // des-ede-112 as 128 etc...
+        // do some special casing...
+        if ((type == "des" || type == "des-ede" || type == "des-ede3") && (dlen & 7) != 0) {
+            dlen = align_up(dlen, 8u);
+        }
+        // if we had to find a cipher without explicit key length (like rc2),
+        // try to set the key length to the desired strength.
+        if (!EVP_CIPHER_CTX_set_key_length(*this, dlen)) {
+            throw_evp_error(fmt::format("Invalid length {} for resolved type {} (wanted {})", len*8, str, _info.len));
+        }
+
+        len = EVP_CIPHER_key_length(cipher);
+    }
+
+
+    if (_key.empty()) {
+        _key.resize(len);
+        if (!RAND_bytes(reinterpret_cast<uint8_t*>(_key.data()), _key.size())) {
+            throw_evp_error(fmt::format("Could not generate key: {}", info.alg));
+        }
+    }
+    if (_key.size() < len) {
+        throw std::invalid_argument(fmt::format("Invalid key data length {} for resolved type {} ({})", _key.size()*8, str, len*8));
+    }
+
+    if (!EVP_CipherInit_ex(*this, cipher, nullptr,
+                    reinterpret_cast<const uint8_t*>(_key.data()), nullptr,
+                    0)) {
+        throw_evp_error("Could not initialize cipher from key materiel");
+    }
+
+    _iv_len = EVP_CIPHER_CTX_iv_length(*this);
+    _block_size = EVP_CIPHER_CTX_block_size(*this);
+    _padding = !padd.empty();
+
+}
+
+std::string encryption::symmetric_key::validate_exact_info_result() const {
+    auto [types, modes, padds] = parse_key_spec(_info.alg);
+
+    auto cipher = EVP_CIPHER_CTX_get0_cipher(*this);
+    auto len = EVP_CIPHER_key_length(cipher);
+    auto mode = EVP_CIPHER_get_mode(cipher);
+
+    std::ostringstream ss;
+
+    if (unsigned(len)*8 != align_up(_info.len, 16u)) {
+        ss << "Length " << len*8 << " differs from requested " << _info.len << std::endl;
+    }
+
+    static std::unordered_map<int, std::string> openssl_modes({
+        { EVP_CIPH_ECB_MODE, "ecb"  },
+        { EVP_CIPH_CBC_MODE, "cbc"  },
+        { EVP_CIPH_CFB_MODE, "cfb"  }, 
+        { EVP_CIPH_OFB_MODE, "ofb"  },
+        { EVP_CIPH_CTR_MODE, "ctr"  },
+        { EVP_CIPH_GCM_MODE, "cgm"  },
+        { EVP_CIPH_CCM_MODE, "ccm"  },
+        { EVP_CIPH_XTS_MODE, "xts"  },
+        { EVP_CIPH_WRAP_MODE, "wrap"},
+        { EVP_CIPH_OCB_MODE, "ocb"  },
+        { EVP_CIPH_SIV_MODE, "siv"  },
+    });
+
+    auto i = openssl_modes.find(mode);
+    if (i != openssl_modes.end() && i->second != modes) {
+        ss << _info << ": " << "Block mode " << i->second << " differers from requested " << modes << std::endl;
+    }
+
+    if ((!padds.empty() && padds != "no") != _padding) {
+        ss << _info << ": " << "Padding (" << bool(_padding) << " differs from requested " << padds << std::endl;
+    }
+
+    return ss.str();
+}
+
+void encryption::symmetric_key::generate_iv_impl(uint8_t* dst, size_t s) const {
+    if (s < _iv_len) {
+        throw std::invalid_argument("Buffer underflow");
+    }
+    if (!RAND_bytes(dst, s)) {
+        throw_evp_error("Could not generate initialization vector");
+    }
+}
+
+void encryption::symmetric_key::transform_unpadded_impl(const uint8_t* input,
+                size_t input_len, uint8_t* output, const uint8_t* iv, mode m) const {
+    if (!EVP_CipherInit_ex(*this, nullptr, nullptr,
+                    reinterpret_cast<const uint8_t*>(_key.data()), iv, int(m))) {
+        throw_evp_error("Could not initialize cipher (transform)");
+    }
+    if (!EVP_CIPHER_CTX_set_padding(*this, 0)) {
+        throw_evp_error("Could not disable padding");
+    }
+
+    if (input_len & (_block_size - 1)) {
+        throw std::invalid_argument("Data must be aligned to 'blocksize'");
+    }
+
+    int outl = 0;
+    auto res = m == mode::decrypt ?
+                    EVP_DecryptUpdate(*this, output, &outl, input,
+                                    int(input_len)) :
+                    EVP_EncryptUpdate(*this, output, &outl, input,
+                                    int(input_len));
+
+    if (!res || outl != int(input_len)) {
+        throw std::runtime_error("transformation failed");
+    }
+}
+
+size_t encryption::symmetric_key::decrypt_impl(const uint8_t* input,
+                size_t input_len, uint8_t* output, size_t output_len,
+                const uint8_t* iv) const {
+    if (!EVP_CipherInit_ex(*this, nullptr, nullptr,
+                    reinterpret_cast<const uint8_t*>(_key.data()), iv, 0)) {
+        throw_evp_error("Could not initialize cipher (decrypt)");
+    }
+    if (!EVP_CIPHER_CTX_set_padding(*this, int(_padding))) {
+        throw_evp_error("Could not initialize padding");
+    }
+
+    // normal case, caller provides output enough to deal with any padding.
+    // in padding case, max out size is input_len - 1.
+    if (input_len <= output_len) {
+        // one go.
+        int outl = 0;
+        int finl = 0;
+        if (!EVP_DecryptUpdate(*this, output, &outl, input, int(input_len))) {
+            throw_evp_error("decryption failed");
+        }
+        if (!EVP_DecryptFinal(*this, output + outl, &finl)) {
+            throw_evp_error("decryption failed");
+        }
+
+        return outl + finl;
+    }
+
+    // meh. must provide block padding.
+    constexpr size_t local_buf_size = 1024;
+
+    static thread_local std::vector<unsigned char> cached_buf;
+
+    if (cached_buf.size() < local_buf_size + _block_size) [[unlikely]] {
+        cached_buf.resize(local_buf_size + _block_size);
+    }
+
+    auto buf = cached_buf.data();
+    size_t res = 0;
+    while (input_len) {
+        auto n = std::min(input_len, local_buf_size);
+        int outl = 0;
+        if (!EVP_DecryptUpdate(*this, buf, &outl, input, int(n))) {
+            throw std::runtime_error("decryption failed");
+        }
+        if (n < local_buf_size) {
+            // last block
+            int finl = 0;
+            if (!EVP_DecryptFinal(*this, buf + outl, &finl)) {
+                throw std::runtime_error("decryption failed");
+            }
+            outl += finl;
+        }
+        if ((res + outl) > output_len) {
+            throw std::invalid_argument("Output buffer too small");
+        }
+        output = std::copy(buf, buf + outl, output);
+        res += outl;
+        input_len -= n;
+        input += n;
+    }
+
+    return res;
+}
+
+size_t encryption::symmetric_key::encrypted_size(size_t n) const {
+    // encryption always adds padding. So if n is multiple of blocksize
+    // the size is n + blocksize. But if its not, things are "better"...
+    return _block_size + align_down<size_t>(n, _block_size);
+}
+
+size_t encryption::symmetric_key::encrypt_impl(const uint8_t* input,
+                size_t input_len, uint8_t* output, size_t output_len,
+                const uint8_t* iv) const {
+    if (output_len < encrypted_size(input_len)) {
+        throw std::invalid_argument("Insufficient buffer");
+    }
+
+    if (!EVP_CipherInit_ex(*this, nullptr, nullptr,
+                    reinterpret_cast<const uint8_t*>(_key.data()), iv, 1)) {
+        throw_evp_error("Could not initialize cipher (encrypt)");
+    }
+    if (!EVP_CIPHER_CTX_set_padding(*this, int(_padding))) {
+        throw_evp_error("Could not initialize padding");
+    }
+
+    int outl = 0;
+    int finl = 0;
+    if (!EVP_EncryptUpdate(*this, output, &outl, input, int(input_len))) {
+        throw_evp_error("encryption failed");
+    }
+    if (!EVP_EncryptFinal(*this, output + outl, &finl)) {
+        throw_evp_error("encryption failed");
+    }
+    return outl + finl;
+}
+
+bool encryption::operator==(const key_info& k1, const key_info& k2) {
+    return k1.alg == k2.alg && k1.len == k2.len;
+}
+
+bool encryption::operator!=(const key_info& k1, const key_info& k2) {
+    return !(k1 == k2);
+}
+
+size_t encryption::key_info_hash::operator()(const key_info& e) const {
+    return utils::tuple_hash()(std::tie(e.alg, e.len));
+}
--- a/ent/encryption/symmetric_key.hh
+++ b/ent/encryption/symmetric_key.hh
@@ -0,0 +1,154 @@
+/*
+ * Copyright (C) 2018 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include <memory>
+#include <tuple>
+#include <iosfwd>
+#include <fmt/core.h>
+#include <fmt/ostream.h>
+
+#include "../../bytes.hh"
+
+// forward declare openssl evp.
+extern "C" {
+struct evp_cipher_ctx_st;
+}
+
+namespace encryption {
+
+struct key_info {
+    sstring alg;
+    unsigned len;
+
+    bool compatible(const key_info&) const;
+};
+
+bool operator==(const key_info& k1, const key_info& k2);
+bool operator!=(const key_info& k1, const key_info& k2);
+std::ostream& operator<<(std::ostream&, const key_info&);
+
+struct key_info_hash {
+    size_t operator()(const key_info& e) const;
+};
+
+std::tuple<sstring, sstring, sstring> parse_key_spec(const sstring&);
+
+// shared between key & kmip
+std::tuple<sstring, sstring, sstring> parse_key_spec_and_validate_defaults(const sstring&);
+
+class symmetric_key {
+    std::unique_ptr<evp_cipher_ctx_st, void (*)(evp_cipher_ctx_st*)> _ctxt;
+    key_info _info;
+    bytes _key;
+    unsigned _iv_len = 0;
+    unsigned _block_size = 0;
+    bool _padding = true;
+
+    operator evp_cipher_ctx_st *() const {
+        return _ctxt.get();
+    }
+
+    void generate_iv_impl(uint8_t* dst, size_t) const;
+    size_t decrypt_impl(const uint8_t* input, size_t input_len, uint8_t* output,
+                    size_t output_len, const uint8_t* iv) const;
+    size_t encrypt_impl(const uint8_t* input, size_t input_len, uint8_t* output,
+                    size_t output_len, const uint8_t* iv) const;
+
+public:
+    symmetric_key(const key_info& info, const bytes& key = { });
+
+    const key_info& info() const {
+        return _info;
+    }
+    const bytes& key() const {
+        return _key;
+    }
+    size_t iv_len() const {
+        return _iv_len;
+    }
+    size_t block_size() const {
+        return _block_size;
+    }
+
+    /**
+     * Evaluates whether or not the key info provided resulted in
+     * the exact same result from openssl, i.e. whether the combination
+     * of alg/block mode/padding etc was actually fully valid (or our 
+     * heuristics have issues)
+    */
+    std::string validate_exact_info_result() const;
+
+    /**
+     * Write a random IV to dst. Must be iv_len() sized or larger
+     */
+    template<typename T>
+    void generate_iv(T* dst, size_t s) const {
+        static_assert(sizeof(T) == sizeof(uint8_t) && std::is_integral_v<T>);
+        generate_iv_impl(reinterpret_cast<uint8_t *>(dst), s);
+    }
+
+    // returns minimal buffer size required to encrypt n bytes. I.e.
+    // block alignment
+    size_t encrypted_size(size_t n) const;
+
+    template<typename T, typename V, typename I = char>
+    size_t decrypt(const T* input, size_t input_len, V* output,
+                    size_t output_len, const I* iv = nullptr) const {
+        static_assert(sizeof(T) == sizeof(uint8_t) && std::is_integral_v<T>);
+        return decrypt_impl(reinterpret_cast<const uint8_t*>(input), input_len,
+                        reinterpret_cast<uint8_t *>(output), output_len,
+                        reinterpret_cast<const uint8_t*>(iv));
+    }
+    template<typename T, typename V, typename I = char>
+    size_t encrypt(const T* input, size_t input_len, V* output,
+                    size_t output_len, const I* iv = nullptr) const {
+        static_assert(sizeof(T) == sizeof(uint8_t) && std::is_integral_v<T>);
+        return encrypt_impl(reinterpret_cast<const uint8_t*>(input), input_len,
+                        reinterpret_cast<uint8_t *>(output), output_len,
+                        reinterpret_cast<const uint8_t*>(iv));
+    }
+
+    enum class mode {
+        decrypt, encrypt,
+    };
+    template<typename T, typename V, typename I = char>
+    void transform_unpadded(mode m, const T* input, size_t input_len, V* output,
+                    const I* iv = nullptr) const {
+        static_assert(sizeof(T) == sizeof(uint8_t) && std::is_integral_v<T>);
+        return transform_unpadded_impl(reinterpret_cast<const uint8_t*>(input),
+                        input_len, reinterpret_cast<uint8_t *>(output),
+                        reinterpret_cast<const uint8_t*>(iv), m);
+    }
+    template<typename T, typename V, typename I = char>
+    void encrypt_unpadded(const T* input, size_t input_len, V* output,
+                    const I* iv = nullptr) const {
+        static_assert(sizeof(T) == sizeof(uint8_t) && std::is_integral_v<T>);
+        return transform_unpadded_impl(reinterpret_cast<const uint8_t*>(input),
+                        input_len, reinterpret_cast<uint8_t *>(output),
+                        reinterpret_cast<const uint8_t*>(iv), mode::encrypt);
+    }
+    template<typename T, typename V, typename I = char>
+    void decrypt_unpadded(const T* input, size_t input_len, V* output,
+                    const I* iv = nullptr) const {
+        static_assert(sizeof(T) == sizeof(uint8_t) && std::is_integral_v<T>);
+        return transform_unpadded_impl(reinterpret_cast<const uint8_t*>(input),
+                        input_len, reinterpret_cast<uint8_t *>(output),
+                        reinterpret_cast<const uint8_t*>(iv), mode::decrypt);
+    }
+
+private:
+    void transform_unpadded_impl(const uint8_t* input, size_t input_len,
+                    uint8_t* output, const uint8_t* iv, mode) const;
+};
+
+}
+
+template <> struct fmt::formatter<encryption::key_info> : fmt::ostream_formatter {};
--- a/ent/encryption/system_key.cc
+++ b/ent/encryption/system_key.cc
@@ -0,0 +1,65 @@
+/*
+ * Copyright (C) 2015 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+#include <stdexcept>
+#include <regex>
+#include <algorithm>
+#include <unordered_map>
+
+#include <openssl/evp.h>
+#include <openssl/rand.h>
+
+#include <seastar/core/align.hh>
+
+#include "symmetric_key.hh"
+#include "system_key.hh"
+
+future<> encryption::system_key::validate() const {
+    return make_ready_future<>();
+}
+
+future<sstring> encryption::system_key::decrypt(const sstring& s) {
+    auto b = base64_decode(s);
+    return decrypt(b).then([](bytes b) {
+        return make_ready_future<sstring>(sstring(b.begin(), b.end()));
+    });
+}
+
+future<sstring> encryption::system_key::encrypt(const sstring& s) {
+    return encrypt(bytes(s.begin(), s.end())).then([](bytes b) {
+        return make_ready_future<sstring>(base64_encode(b));
+    });
+}
+
+future<bytes> encryption::system_key::encrypt(const bytes& b) {
+    return get_key().then([b](shared_ptr<symmetric_key> k) {
+        auto i = k->iv_len();
+        auto n = k->encrypted_size(b.size());
+        bytes res(bytes::initialized_later(), n + i);
+        k->generate_iv(reinterpret_cast<char*>(res.data()), i);
+        n = k->encrypt(reinterpret_cast<const char*>(b.data()), b.size()
+                        , reinterpret_cast<char*>(res.data()) + i, res.size() - i
+                        , reinterpret_cast<const char*>(res.data()));
+        res.resize(n + i);
+        return make_ready_future<bytes>(std::move(res));
+    });
+
+}
+
+future<bytes> encryption::system_key::decrypt(const bytes& b) {
+    return get_key().then([b](shared_ptr<symmetric_key> k) {
+        auto i = k->iv_len();
+        bytes res(bytes::initialized_later(), b.size() - i);
+        auto n = k->decrypt(reinterpret_cast<const char*>(b.data()) + i,
+                        b.size() - i, reinterpret_cast<char*>(res.data()),
+                        res.size(), reinterpret_cast<const char*>(b.data()));
+        res.resize(n);
+        return make_ready_future<bytes>(std::move(res));
+    });
+}
+
--- a/ent/encryption/system_key.hh
+++ b/ent/encryption/system_key.hh
@@ -0,0 +1,34 @@
+/*
+ * Copyright (C) 2015 ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include "encryption.hh"
+#include "../../bytes.hh"
+
+namespace encryption {
+
+class symmetric_key;
+
+class system_key {
+public:
+    virtual ~system_key() {}
+    virtual future<shared_ptr<symmetric_key>> get_key() = 0;
+    virtual const sstring& name() const = 0;
+    virtual bool is_local() const = 0;
+    virtual future<> validate() const;
+
+    future<sstring> encrypt(const sstring&);
+    future<sstring> decrypt(const sstring&);
+    future<bytes> encrypt(const bytes&);
+    future<bytes> decrypt(const bytes&);
+};
+
+}
+