rgw lc debug interval: 10
rgw:
storage classes:
- LUKEWARM:
- FROZEN:
+ LUKEWARM: [--compression, zstd]
+ FROZEN: [--compression, zlib]
s3tests:
accounts:
iam root: RGW88888888888888888
--- /dev/null
+../.qa/
\ No newline at end of file
--- /dev/null
+overrides:
+ rgw:
+ enable features: [compress-encrypted]
+tasks:
+- workunit:
+ clients:
+ client.0:
+ - rgw/run-lc-encrypt-recompress.sh
--- /dev/null
+tasks:
+- workunit:
+ clients:
+ client.0:
+ - rgw/run-lc-recompress.sh
--- /dev/null
+overrides:
+ ceph:
+ conf:
+ client:
+ rgw crypt sse algorithm: aes-256-gcm
+ rgw:
+ enable features: [compress-encrypted]
+tasks:
+- workunit:
+ clients:
+ client.0:
+ - rgw/run-lc-encrypt-recompress.sh
--- /dev/null
+#!/usr/bin/env bash
+set -ex
+
+mydir=$(dirname $0)
+
+python3 -m venv $mydir
+source $mydir/bin/activate
+pip install pip --upgrade
+pip install boto3
+
+$mydir/bin/python3 $mydir/test_rgw_lc_recompress.py --encrypt
+
+deactivate
+echo OK.
--- /dev/null
+#!/usr/bin/env bash
+set -ex
+
+mydir=$(dirname $0)
+
+python3 -m venv $mydir
+source $mydir/bin/activate
+pip install pip --upgrade
+pip install boto3
+
+$mydir/bin/python3 $mydir/test_rgw_lc_recompress.py
+
+deactivate
+echo OK.
--- /dev/null
+#!/usr/bin/env python3
+
+import io
+import logging as log
+import json
+import sys
+import time
+import boto3.s3.transfer
+from common import exec_cmd, create_user, boto_connect
+
+"""
+Tests that RGW lifecycle transitions correctly recompress objects
+when moving between storage classes with different compression configs.
+
+The qa suite configures:
+ STANDARD - no compression
+ LUKEWARM - zstd compression
+ FROZEN - zlib compression
+
+Test plan:
+ 1. Upload object to STANDARD (uncompressed)
+ 2. Transition STANDARD -> LUKEWARM (compress with zstd)
+ 3. Transition LUKEWARM -> FROZEN (recompress with zlib)
+ 4. Transition FROZEN -> STANDARD (decompress)
+ 5. GET object and verify data integrity
+
+With --encrypt, objects are uploaded with SSE-KMS and the test
+verifies encryption is preserved across each transition. This
+requires the compress-encrypted zonegroup feature.
+
+Accepts --size-kb <N> to set the object size in KB (default: runs
+both 4KB and 32MB). Sizes above 8MB use multipart upload to
+exercise the multipart compressed+encrypted transition path.
+"""
+
+USER = 'lc-recompress-tester'
+DISPLAY_NAME = 'LC Recompress Testing'
+ACCESS_KEY = 'LCRECOMP0123456789AB'
+SECRET_KEY = 'lcrecompsecretkey0123456789abcdefghijklmn'
+BUCKET_NAME = 'lc-recompress-bucket'
+KMS_KEY_ID = 'testkey-1'
+
+LC_POLL_INTERVAL = 10
+LC_TIMEOUT = 120
+
+
+def make_compressible_body(size_bytes):
+ """Generate compressible data of the requested size."""
+ pattern = b'The quick brown fox jumps over the lazy dog. '
+ repeats = (size_bytes // len(pattern)) + 1
+ return (pattern * repeats)[:size_bytes]
+
+
+def object_stat(bucket_name, object_key):
+ """Run radosgw-admin object stat and return parsed JSON."""
+ out = exec_cmd(
+ f'radosgw-admin object stat --bucket={bucket_name} --object={object_key}'
+ )
+ # some attrs (e.g. crypt.keysel) contain raw binary that isn't valid UTF-8
+ if isinstance(out, bytes):
+ out = out.decode('utf-8', errors='replace')
+ return json.loads(out)
+
+
+def get_compression_type(stat):
+ """
+ Extract the compression_type from object stat output.
+ Returns None if the object is not compressed.
+ """
+ compression = stat.get('compression')
+ if compression is None:
+ return None
+ ct = compression.get('compression_type', 'none')
+ if ct.lower() == 'none':
+ return None
+ return ct.lower()
+
+
+def get_storage_class(stat):
+ """
+ Extract the storage class from object stat output.
+ The storage class attr lives in attrs['user.rgw.storage_class'].
+ If absent, the object is in the STANDARD storage class.
+ """
+ attrs = stat.get('attrs', {})
+ sc = attrs.get('user.rgw.storage_class', '')
+ # The value may be a raw string possibly with trailing null bytes
+ sc = sc.strip().strip('\x00')
+ if not sc:
+ return 'STANDARD'
+ return sc
+
+
+def get_crypt_mode(stat):
+ """
+ Extract the encryption mode from object stat output.
+ Returns None if the object is not encrypted.
+ """
+ attrs = stat.get('attrs', {})
+ mode = attrs.get('user.rgw.crypt.mode', '')
+ mode = mode.strip().strip('\x00')
+ return mode if mode else None
+
+
+def get_crypt_salt(stat):
+ """
+ Extract the raw crypt salt attr for rotation comparisons.
+ Returns None if absent. The value may contain non-printable bytes
+ (decoded with errors='replace') but two distinct 32-byte random
+ salts are overwhelmingly unlikely to collide under that encoding.
+ """
+ attrs = stat.get('attrs', {})
+ salt = attrs.get('user.rgw.crypt.salt', '')
+ return salt if salt else None
+
+
+def is_aead_crypt_mode(mode):
+ """
+ True for GCM-family crypt modes that derive per-object keys from
+ a stored salt. CBC modes don't write crypt.salt at all.
+ """
+ return mode is not None and mode.endswith('-GCM')
+
+
+def put_lifecycle_rule(client, bucket_name, rule_id, target_class):
+ """
+ Set a lifecycle configuration with a single transition rule.
+ Uses Days=0 for immediate transition.
+ """
+ client.put_bucket_lifecycle_configuration(
+ Bucket=bucket_name,
+ LifecycleConfiguration={
+ 'Rules': [
+ {
+ 'ID': rule_id,
+ 'Filter': {'Prefix': ''},
+ 'Status': 'Enabled',
+ 'Transitions': [
+ {
+ 'Days': 0,
+ 'StorageClass': target_class,
+ }
+ ],
+ }
+ ]
+ }
+ )
+ log.info(f'Set lifecycle rule {rule_id}: transition to {target_class}')
+
+
+def wait_for_transition(bucket_name, object_key, expected_class, timeout=LC_TIMEOUT):
+ """
+ Poll radosgw-admin lc process + object stat until the object
+ reaches the expected storage class, or timeout.
+ """
+ deadline = time.time() + timeout
+ while time.time() < deadline:
+ exec_cmd(f'radosgw-admin lc process --bucket={bucket_name}'
+ ' --rgw-lc-debug-interval=10')
+ time.sleep(LC_POLL_INTERVAL)
+
+ stat = object_stat(bucket_name, object_key)
+ sc = get_storage_class(stat)
+ log.info(f' current storage class: {sc} (waiting for {expected_class})')
+ if sc == expected_class:
+ return stat
+
+ raise AssertionError(
+ f'Timed out waiting for object to transition to {expected_class}'
+ )
+
+
+def verify_transition(stat, expected_class, expected_compression, encrypt=False):
+ """
+ Verify that the object stat matches the expected storage class
+ and compression type after a transition. When encrypt=True,
+ also verify that encryption mode is still present.
+ """
+ sc = get_storage_class(stat)
+ ct = get_compression_type(stat)
+ on_disk = stat.get('size', 0)
+ comp = stat.get('compression', {})
+ orig = comp.get('orig_size', on_disk)
+
+ log.info(f' storage_class={sc}, compression_type={ct},'
+ f' on_disk={on_disk}, orig_size={orig}')
+ assert sc == expected_class, \
+ f'Expected storage class {expected_class}, got {sc}'
+ assert ct == expected_compression, \
+ f'Expected compression type {expected_compression}, got {ct}'
+
+ if encrypt:
+ mode = get_crypt_mode(stat)
+ log.info(f' crypt_mode={mode}')
+ assert mode is not None, \
+ 'Expected object to be encrypted, but crypt mode is missing'
+
+
+def run_test(size_kb, encrypt):
+ """Run the full transition test with the given object size."""
+ size_bytes = size_kb * 1024
+ object_key = f'test-object-{size_kb}kb'
+ object_body = make_compressible_body(size_bytes)
+ label = f'{size_kb}KB'
+ if size_kb >= 1024:
+ label = f'{size_kb // 1024}MB'
+
+ log.info(f'=== Testing {label} object (encrypt={encrypt}) ===')
+
+ conn = boto_connect(ACCESS_KEY, SECRET_KEY)
+ client = conn.meta.client
+
+ # Clean up any previous run
+ try:
+ bucket = conn.Bucket(BUCKET_NAME)
+ bucket.objects.all().delete()
+ bucket.delete()
+ except Exception:
+ pass
+
+ bucket = conn.create_bucket(Bucket=BUCKET_NAME)
+
+ # Upload object — use multipart for sizes above 8MB to exercise
+ # the multipart compressed+encrypted transition path
+ MULTIPART_THRESHOLD = 8 * 1024 * 1024
+
+ extra_args = {}
+ if encrypt:
+ extra_args['ServerSideEncryption'] = 'aws:kms'
+ extra_args['SSEKMSKeyId'] = KMS_KEY_ID
+
+ if size_bytes > MULTIPART_THRESHOLD:
+ log.info(f'Uploading {len(object_body)} byte object as {object_key} (multipart)')
+ transfer_config = boto3.s3.transfer.TransferConfig(
+ multipart_threshold=MULTIPART_THRESHOLD,
+ multipart_chunksize=MULTIPART_THRESHOLD,
+ )
+ client.upload_fileobj(
+ io.BytesIO(object_body), BUCKET_NAME, object_key,
+ ExtraArgs=extra_args,
+ Config=transfer_config,
+ )
+ else:
+ log.info(f'Uploading {len(object_body)} byte object as {object_key}')
+ bucket.put_object(Key=object_key, Body=object_body, **extra_args)
+
+ stat = object_stat(BUCKET_NAME, object_key)
+ verify_transition(stat, 'STANDARD', None, encrypt)
+ log.info('Initial upload verified: STANDARD, no compression')
+ # Salt rotation only applies to AEAD modes (CBC has no salt attr).
+ prev_salt = get_crypt_salt(stat) if encrypt else None
+
+ def assert_salt_rotated(new_stat, prev):
+ if not encrypt or not is_aead_crypt_mode(get_crypt_mode(new_stat)):
+ return None
+ new_salt = get_crypt_salt(new_stat)
+ assert new_salt is not None, 'AEAD object missing crypt.salt'
+ assert new_salt != prev, \
+ 'crypt.salt did not rotate across re-encryption'
+ return new_salt
+
+ # Transition 1: STANDARD (none) -> LUKEWARM (zstd)
+ log.info(f'--- Transition 1: STANDARD -> LUKEWARM (zstd) [{label}] ---')
+ put_lifecycle_rule(client, BUCKET_NAME, 'to-lukewarm', 'LUKEWARM')
+ stat = wait_for_transition(BUCKET_NAME, object_key, 'LUKEWARM')
+ verify_transition(stat, 'LUKEWARM', 'zstd', encrypt)
+ orig_size = stat['compression']['orig_size']
+ assert orig_size == len(object_body), \
+ f'Expected orig_size={len(object_body)}, got {orig_size}'
+ body = bucket.Object(object_key).get()['Body'].read()
+ assert body == object_body, 'Data mismatch after transition 1'
+ prev_salt = assert_salt_rotated(stat, prev_salt)
+ log.info('Transition 1 verified')
+
+ # Transition 2: LUKEWARM (zstd) -> FROZEN (zlib)
+ log.info(f'--- Transition 2: LUKEWARM -> FROZEN (zlib) [{label}] ---')
+ put_lifecycle_rule(client, BUCKET_NAME, 'to-frozen', 'FROZEN')
+ stat = wait_for_transition(BUCKET_NAME, object_key, 'FROZEN')
+ verify_transition(stat, 'FROZEN', 'zlib', encrypt)
+ orig_size = stat['compression']['orig_size']
+ assert orig_size == len(object_body), \
+ f'Expected orig_size={len(object_body)}, got {orig_size}'
+ body = bucket.Object(object_key).get()['Body'].read()
+ assert body == object_body, 'Data mismatch after transition 2'
+ prev_salt = assert_salt_rotated(stat, prev_salt)
+ log.info('Transition 2 verified')
+
+ # Same-codec CopyObject (compressed+encrypted only) — exercises the
+ # passthrough path where set_writer skips decompression. Verifies
+ # that CRYPT_ORIGINAL_SIZE is the plaintext size (not the post-
+ # decrypt compressed bytes), which drives bucket-index quota.
+ if encrypt:
+ copy_key = f'{object_key}-same-codec-copy'
+ log.info(f'--- Same-codec CopyObject passthrough [{label}] ---')
+ client.copy_object(
+ Bucket=BUCKET_NAME, Key=copy_key,
+ CopySource={'Bucket': BUCKET_NAME, 'Key': object_key},
+ StorageClass='FROZEN',
+ ServerSideEncryption='aws:kms', SSEKMSKeyId=KMS_KEY_ID,
+ MetadataDirective='COPY',
+ )
+ copy_stat = object_stat(BUCKET_NAME, copy_key)
+ verify_transition(copy_stat, 'FROZEN', 'zlib', encrypt)
+ copy_orig = copy_stat['compression']['orig_size']
+ assert copy_orig == len(object_body), \
+ f'Same-codec copy orig_size {copy_orig} != plaintext {len(object_body)}'
+ if is_aead_crypt_mode(get_crypt_mode(copy_stat)):
+ copy_salt = get_crypt_salt(copy_stat)
+ assert copy_salt is not None, 'AEAD copy missing crypt.salt'
+ assert copy_salt != prev_salt, \
+ 'Copy salt did not rotate vs source'
+ body = bucket.Object(copy_key).get()['Body'].read()
+ assert body == object_body, 'Data mismatch on same-codec copy'
+ bucket.Object(copy_key).delete()
+ log.info('Same-codec CopyObject passthrough verified')
+
+ # Transition 3: FROZEN (zlib) -> STANDARD (none)
+ log.info(f'--- Transition 3: FROZEN -> STANDARD (none) [{label}] ---')
+ put_lifecycle_rule(client, BUCKET_NAME, 'to-standard', 'STANDARD')
+ stat = wait_for_transition(BUCKET_NAME, object_key, 'STANDARD')
+ verify_transition(stat, 'STANDARD', None, encrypt)
+ body = bucket.Object(object_key).get()['Body'].read()
+ assert body == object_body, 'Data mismatch after transition 3'
+ prev_salt = assert_salt_rotated(stat, prev_salt)
+ log.info('Transition 3 verified')
+
+ # Clean up
+ bucket.objects.all().delete()
+ bucket.delete()
+ log.info(f'{label} test passed')
+
+
+def main():
+ encrypt = '--encrypt' in sys.argv
+
+ # parse --size-kb <N> for a single size, otherwise run both 4KB and 32MB
+ sizes_kb = [4, 32 * 1024]
+ for i, arg in enumerate(sys.argv):
+ if arg == '--size-kb' and i + 1 < len(sys.argv):
+ sizes_kb = [int(sys.argv[i + 1])]
+ break
+
+ if encrypt:
+ log.info('Running with encryption enabled (SSE-KMS)')
+
+ log.info('Creating test user')
+ create_user(USER, DISPLAY_NAME, ACCESS_KEY, SECRET_KEY)
+
+ for size_kb in sizes_kb:
+ run_test(size_kb, encrypt)
+
+ suffix = ' (with encryption)' if encrypt else ''
+ log.info(f'All lifecycle recompression tests passed{suffix}')
+
+
+main()
#include "rgw_cr_rest.h"
#include "rgw_crypt.h"
#include "rgw_datalog.h"
+#include "rgw_op.h"
#include "rgw_putobj_processor.h"
#include "rgw_lc_tier.h"
#include "rgw_restore.h"
accounted_size = compressed ? cs_info.orig_size : ofs;
}
+ if (dp_factory) {
+ accounted_size = dp_factory->get_accounted_size(accounted_size);
+ }
+
const req_context rctx{dpp, y, nullptr};
return aoproc.complete(accounted_size, etag, mtime, set_mtime, attrs,
rgw::cksum::no_cksum, delete_at,
return 0;
}
+/*
+ * DataProcessorFactory for lifecycle transitions.
+ *
+ * Thin subclass of RGWRecompressDPF that supplies a pre-fetched
+ * decrypt key and zone compression config. The encrypt key is
+ * built fresh at set_writer() time so AEAD modes get a new salt
+ * (avoids GCM nonce reuse on re-encrypted plaintext).
+ */
+class RGWTransitionDPF : public RGWRecompressDPF {
+ RGWObjectCtx& obj_ctx;
+ rgw_obj& obj;
+ std::string dest_compression;
+ bool compress_encrypted_enabled;
+
+ /* pre-fetched by transition_obj() and passed to constructor */
+ std::unique_ptr<BlockCrypt> prefetched_decrypt;
+
+protected:
+ int get_decrypt_crypt(const DoutPrefixProvider* dpp,
+ optional_yield y,
+ const rgw::sal::Attrs& src_attrs,
+ std::unique_ptr<BlockCrypt>* crypt) override
+ {
+ *crypt = std::move(prefetched_decrypt);
+ return 0;
+ }
+
+ int get_encrypt_crypt(const DoutPrefixProvider* dpp,
+ optional_yield y,
+ rgw::sal::Attrs& dest_attrs,
+ std::unique_ptr<BlockCrypt>* crypt) override
+ {
+ return rgw_prepare_reencrypt_object(dpp, cct, dest_attrs,
+ obj.bucket.bucket_id,
+ obj.key.name, y, crypt);
+ }
+
+ const std::string& get_dest_compression() override { return dest_compression; }
+
+ bool supports_compress_encrypted() override {
+ return compress_encrypted_enabled;
+ }
+
+ void mark_compressed() override { obj_ctx.set_compressed(obj); }
+
+public:
+ RGWTransitionDPF(CephContext* cct_,
+ uint64_t& obj_size,
+ const rgw::sal::Attrs& src_attrs,
+ RGWObjectCtx& obj_ctx_,
+ rgw_obj& obj_,
+ const std::string& dest_compression_,
+ bool compress_encrypted,
+ std::unique_ptr<BlockCrypt> decrypt)
+ : RGWRecompressDPF(cct_, obj_size, src_attrs),
+ obj_ctx(obj_ctx_), obj(obj_),
+ dest_compression(dest_compression_),
+ compress_encrypted_enabled(compress_encrypted),
+ prefetched_decrypt(std::move(decrypt))
+ {}
+};
+
int RGWRados::transition_obj(RGWObjectCtx& obj_ctx,
RGWBucketInfo& bucket_info,
rgw_obj obj,
(void) decode_policy(dpp, i->second, &owner);
}
+ rgw::sal::DataProcessorFactory* dp_factory = nullptr;
+ std::optional<RGWTransitionDPF> transition_dpf;
+
+ const auto& compression_type =
+ svc.zone->get_zone_params().get_compression_type(placement_rule);
+
+ bool src_compressed = false;
+ RGWCompressionInfo cs_info;
+ ret = rgw_compression_info_from_attrset(attrs, src_compressed, cs_info);
+ if (ret < 0)
+ return ret;
+
+ /*
+ * Determine the effective destination compression. If the source
+ * is encrypted and the zonegroup doesn't support compress_encrypted,
+ * force "none" to avoid incompatible layouts.
+ */
+ std::string dest_compression = compression_type;
+ bool is_encrypted = attrs.count(RGW_ATTR_CRYPT_MODE);
+ if (is_encrypted &&
+ !svc.zone->get_zonegroup().supports(
+ rgw::zone_features::compress_encrypted)) {
+ dest_compression = "none";
+ }
+
+ /*
+ * Skip when the source already matches the destination config.
+ * "random" never matches because the stored codec is concrete
+ * (e.g. "zlib") while the config string stays "random".
+ */
+ bool already_matches =
+ dest_compression != "random" &&
+ ((!src_compressed && dest_compression == "none") ||
+ (src_compressed && cs_info.compression_type == dest_compression));
+
+ bool need_recompress = !already_matches;
+
+ /*
+ * Retrieve the decryption key when the source is encrypted
+ * and compression needs to change. The re-encryption path will
+ * fetch the same key again from the backend (after regenerating
+ * the GCM salt for AEAD modes) — a second KMS/SSE-S3 round-trip
+ * per transitioned object.
+ */
+ std::unique_ptr<BlockCrypt> decrypt_crypt;
+
+ if (obj_size == 0) {
+ ldpp_dout(dpp, 20) << __func__ << " " << obj
+ << " is empty, skipping recompression" << dendl;
+ need_recompress = false;
+ }
+
+ if (need_recompress && is_encrypted) {
+ ret = rgw_prepare_decrypt_object(
+ dpp, cct, attrs,
+ bucket_info.bucket.bucket_id, obj.key.name,
+ y, &decrypt_crypt);
+ if (ret == -ENOTSUP) {
+ ldpp_dout(dpp, 10) << __func__ << " cannot decrypt "
+ << obj << ", copying as-is" << dendl;
+ } else if (ret < 0) {
+ return ret;
+ }
+
+ if (!decrypt_crypt) {
+ if (src_compressed && !svc.zone->get_zonegroup().supports(
+ rgw::zone_features::compress_encrypted)) {
+ ldpp_dout(dpp, 0) << "ERROR: " << obj
+ << " is encrypted+compressed but cannot decrypt;"
+ " refusing to copy incompatible layout when"
+ " compress_encrypted is disabled" << dendl;
+ return -ENOTSUP;
+ }
+ ldpp_dout(dpp, 10) << __func__ << " " << obj
+ << " encrypted but cannot decrypt, copying as-is" << dendl;
+ need_recompress = false;
+ } else {
+ ldpp_dout(dpp, 10) << __func__ << " re-encrypting "
+ << obj << " for recompression" << dendl;
+ }
+ }
+
+ bool compress_encrypted = svc.zone->get_zonegroup().supports(
+ rgw::zone_features::compress_encrypted);
+
+ if (need_recompress) {
+ transition_dpf.emplace(cct, obj_size, attrs,
+ obj_ctx, obj,
+ dest_compression,
+ compress_encrypted,
+ std::move(decrypt_crypt));
+ dp_factory = &*transition_dpf;
+ } else if (!is_encrypted) {
+ ldpp_dout(dpp, 20) << __func__
+ << " compression already matches dest config ("
+ << dest_compression << "), skipping" << dendl;
+ } else {
+ ldpp_dout(dpp, 20) << __func__
+ << " encrypted, compression already matches, copying as-is"
+ << dendl;
+ }
+
ret = copy_obj_data(obj_ctx,
owner,
bucket_info,
olh_epoch,
real_time(),
nullptr /* petag */,
- nullptr, /* dp_factory */
+ dp_factory,
dpp,
y,
log_op);
#pragma once
+#include <memory>
+#include <optional>
+#include <string>
#include <vector>
#include "compressor/Compressor.h"
#include "rgw_putobj.h"
#include "rgw_op.h"
#include "rgw_compression_types.h"
+#include "rgw_crypt.h"
int rgw_compression_info_from_attr(const bufferlist& attr,
bool& need_decompress,
std::optional<int32_t> get_compressor_message() { return compressor_message; }
}; /* RGWPutObj_Compress */
+
+/*
+ * Base class for data-processor factories that implement the
+ * decrypt -> decompress -> recompress -> re-encrypt pipeline.
+ */
+class RGWRecompressDPF : public rgw::sal::DataProcessorFactory {
+protected:
+ CephContext* cct;
+ uint64_t& orig_size;
+
+ /*
+ * Source attrs are COPIED into the DPF. In the transition path the
+ * same map is used for both source metadata and destination attrs
+ * passed to copy_obj_data(). Storing by reference would destroy
+ * source metadata when set_writer() strips crypt attrs.
+ */
+ rgw::sal::Attrs src_attrs;
+
+ DataProcessorFilter cb;
+ RGWGetObj_Filter* filter{&cb};
+
+ RGWCompressionInfo decompress_info;
+ std::optional<RGWGetObj_Decompress> decomp;
+ std::unique_ptr<RGWGetObj_BlockDecrypt> decrypt_filter;
+
+ std::unique_ptr<RGWPutObj_BlockEncrypt> encrypt_filter;
+ CompressorRef compressor_plugin;
+ std::optional<RGWPutObj_Compress> compressor;
+
+ virtual int get_decrypt_crypt(const DoutPrefixProvider* dpp,
+ optional_yield y,
+ const rgw::sal::Attrs& src_attrs,
+ std::unique_ptr<BlockCrypt>* crypt) = 0;
+
+ virtual int get_encrypt_crypt(const DoutPrefixProvider* dpp,
+ optional_yield y,
+ rgw::sal::Attrs& dest_attrs,
+ std::unique_ptr<BlockCrypt>* crypt) = 0;
+
+ virtual const std::string& get_dest_compression() = 0;
+ virtual bool supports_compress_encrypted() = 0;
+
+ virtual CompressorRef create_compressor(const std::string& type) {
+ return Compressor::create(cct, type);
+ }
+
+ virtual void mark_compressed() = 0;
+ virtual void on_obj_size_changed(uint64_t new_size) {}
+
+public:
+ RGWRecompressDPF(CephContext* cct_,
+ uint64_t& obj_size,
+ const rgw::sal::Attrs& src_attrs_)
+ : cct(cct_), orig_size(obj_size),
+ src_attrs(src_attrs_) {}
+
+ ~RGWRecompressDPF() override = default;
+
+ int set_writer(rgw::sal::DataProcessor* writer,
+ rgw::sal::Attrs& attrs,
+ const DoutPrefixProvider* dpp,
+ optional_yield y) override
+ {
+ bool src_compressed = false;
+ RGWCompressionInfo cs_info;
+ int ret = rgw_compression_info_from_attrset(src_attrs, src_compressed,
+ cs_info);
+ if (ret < 0) return ret;
+
+ uint64_t on_disk_size = orig_size;
+ bool src_encrypted = src_attrs.count(RGW_ATTR_CRYPT_MODE);
+ const std::string src_mode = get_str_attribute(src_attrs, RGW_ATTR_CRYPT_MODE);
+ const bool aead_src = src_encrypted && is_aead_mode(src_mode);
+
+ // capture codec before potential cs_info move
+ std::string src_comp_type;
+ if (src_compressed) {
+ src_comp_type = cs_info.compression_type;
+ }
+
+ // determine decrypt availability first — needed for decompress decision
+ std::unique_ptr<BlockCrypt> decrypt_crypt;
+ ret = get_decrypt_crypt(dpp, y, src_attrs, &decrypt_crypt);
+ if (ret < 0) return ret;
+
+ /*
+ * Read multipart part lengths (shared logic — both DPFs need this).
+ * Hard-fail on decode errors to avoid misdecrypting multipart objects.
+ */
+ std::vector<size_t> decrypt_parts;
+ std::vector<std::pair<uint32_t, std::string>> decrypt_part_keys;
+ if (decrypt_crypt) {
+ auto parts_iter = src_attrs.find(RGW_ATTR_CRYPT_PARTS);
+ if (parts_iter != src_attrs.end()) {
+ /*
+ * CRYPT_PARTS present — decode it. An empty vector is the
+ * sentinel for single-stream re-encrypted objects; do NOT
+ * fall back to the manifest in that case.
+ */
+ auto bl = parts_iter->second;
+ auto iter = bl.cbegin();
+ try {
+ decode(decrypt_parts, iter);
+ } catch (buffer::error&) {
+ ldpp_dout(dpp, 0) << "ERROR: malformed "
+ RGW_ATTR_CRYPT_PARTS << dendl;
+ return -EIO;
+ }
+ } else {
+ auto manifest_iter = src_attrs.find(RGW_ATTR_MANIFEST);
+ if (manifest_iter != src_attrs.end()) {
+ int r = RGWGetObj_BlockDecrypt::read_manifest_parts(
+ dpp, manifest_iter->second, decrypt_parts);
+ if (r < 0) {
+ ldpp_dout(dpp, 0) << "ERROR: failed to read "
+ "manifest parts" << dendl;
+ return r;
+ }
+ }
+ }
+
+ /*
+ * AEAD multipart objects also store the (part_num, salt) pairs
+ * for per-part IV/key derivation. Empty → single-part.
+ */
+ if (aead_src) {
+ if (auto it = src_attrs.find(RGW_ATTR_CRYPT_PART_NUMS);
+ it != src_attrs.end()) {
+ auto bl = it->second;
+ auto iter = bl.cbegin();
+ try {
+ decode(decrypt_part_keys, iter);
+ } catch (buffer::error&) {
+ ldpp_dout(dpp, 0) << "ERROR: malformed "
+ RGW_ATTR_CRYPT_PART_NUMS << dendl;
+ return -EIO;
+ }
+ }
+ }
+ }
+
+ // determine compression match — needed for decompress decision
+ std::string dest_comp = get_dest_compression();
+ bool compression_matches =
+ dest_comp != "random" &&
+ ((!src_compressed && dest_comp == "none") ||
+ (src_compressed && src_comp_type == dest_comp));
+
+ /*
+ * Decompress decision — must happen after decrypt and
+ * compression_matches are known. Decompress when source is
+ * compressed AND:
+ * - source is not encrypted (always safe to decompress), OR
+ * - source is encrypted AND decrypt is available AND
+ * compression needs to change (different codec or "none")
+ *
+ * When encrypted + same codec, skip decompress — compressed
+ * data passes through after decrypt/re-encrypt.
+ */
+ bool need_decompress = src_compressed &&
+ !compression_matches &&
+ (!src_encrypted || decrypt_crypt);
+
+ if (need_decompress) {
+ decompress_info = std::move(cs_info);
+ decomp.emplace(cct, &decompress_info, false, filter);
+ filter = &*decomp;
+ orig_size = decompress_info.orig_size;
+ on_obj_size_changed(orig_size);
+ } else if (src_compressed) {
+ /*
+ * Compressed-passthrough (same codec): data stays compressed on
+ * the wire, but bucket logging and event notifications expect
+ * orig_size to reflect the S3-logical (plaintext) size recorded
+ * in the source's compression metadata.
+ */
+ orig_size = cs_info.orig_size;
+ on_obj_size_changed(orig_size);
+ } else if (aead_src && decrypt_crypt) {
+ /*
+ * AEAD without compression: prefer the stored ORIGINAL_SIZE attr
+ * (authoritative plaintext) and fall back to the chunk/tag
+ * derivation formula. Both equal plaintext for non-compressed
+ * sources; the attr matches even if a chunk size ever changes.
+ */
+ uint64_t pt_size = 0;
+ if (rgw_get_aead_original_size(dpp, src_attrs, &pt_size) ||
+ rgw_get_aead_decrypted_size(dpp, src_attrs, on_disk_size, &pt_size)) {
+ orig_size = pt_size;
+ on_obj_size_changed(orig_size);
+ }
+ }
+
+ // read: decrypt wraps decompress (outer)
+ if (decrypt_crypt) {
+ if (aead_src) {
+ decrypt_filter = std::make_unique<RGWGetObj_BlockDecrypt>(
+ dpp, cct, filter, std::move(decrypt_crypt),
+ std::move(decrypt_parts), std::move(decrypt_part_keys),
+ static_cast<off_t>(on_disk_size),
+ static_cast<bool>(decomp), y);
+ } else {
+ decrypt_filter = std::make_unique<RGWGetObj_BlockDecrypt>(
+ dpp, cct, filter, std::move(decrypt_crypt),
+ std::move(decrypt_parts), y);
+ }
+ filter = decrypt_filter.get();
+ }
+
+ /*
+ * Crypt attr handling for re-encryption. Two modes:
+ *
+ * CopyObject: get_encrypt_crypt() strips all crypt attrs itself
+ * then writes fresh ones (may change key, mode, etc.)
+ *
+ * LC transition: same mode and same upstream key are preserved.
+ * CRYPT_PARTS / CRYPT_PART_NUMS are always cleared (re-encryption
+ * produces a single stream). For AEAD modes, get_encrypt_crypt()
+ * also regenerates CRYPT_SALT so the re-derived per-object key
+ * differs from the source — required to avoid GCM nonce reuse
+ * when the plaintext changes (e.g., after recompression).
+ */
+ if (decrypt_filter) {
+ attrs.erase(RGW_ATTR_CRYPT_PARTS);
+ attrs.erase(RGW_ATTR_CRYPT_PART_NUMS);
+ }
+
+ // write: encrypt (inner) then compress (outer)
+ rgw::sal::DataProcessor* processor = writer;
+
+ std::unique_ptr<BlockCrypt> encrypt_crypt;
+ ret = get_encrypt_crypt(dpp, y, attrs, &encrypt_crypt);
+ if (ret < 0) return ret;
+
+ if (encrypt_crypt) {
+ encrypt_filter = std::make_unique<RGWPutObj_BlockEncrypt>(
+ dpp, cct, processor, std::move(encrypt_crypt), y);
+ processor = encrypt_filter.get();
+
+ // write empty CRYPT_PARTS sentinel for re-encrypted objects
+ if (decrypt_filter) {
+ std::vector<size_t> empty_parts;
+ bufferlist parts_bl;
+ encode(empty_parts, parts_bl);
+ attrs[RGW_ATTR_CRYPT_PARTS] = std::move(parts_bl);
+ }
+ } else if (decrypt_filter) {
+ // decrypt without re-encrypt (SSE-C -> plaintext) — strip all crypt attrs
+ auto it = attrs.lower_bound(RGW_ATTR_CRYPT_PREFIX);
+ while (it != attrs.end() &&
+ it->first.compare(0, sizeof(RGW_ATTR_CRYPT_PREFIX) - 1,
+ RGW_ATTR_CRYPT_PREFIX) == 0) {
+ it = attrs.erase(it);
+ }
+ }
+
+ if (!compression_matches && dest_comp != "none" &&
+ (encrypt_filter == nullptr || supports_compress_encrypted())) {
+ compressor_plugin = create_compressor(dest_comp);
+ if (compressor_plugin) {
+ compressor.emplace(cct, compressor_plugin, processor);
+ processor = &*compressor;
+ mark_compressed();
+ }
+ } else if (src_compressed && compression_matches) {
+ /*
+ * Source stays compressed through the pipeline (same codec) —
+ * hint the backend that this data is already incompressible.
+ */
+ mark_compressed();
+ }
+ if (decomp || compressor) {
+ attrs.erase(RGW_ATTR_COMPRESSION);
+ }
+
+ cb.set_processor(processor);
+
+ // skip fixup for empty objects — size-1 underflows to max uint
+ if (on_disk_size == 0) {
+ return 0;
+ }
+
+ /*
+ * fixup_range is called with the encrypted on-disk size as the
+ * "plaintext range" argument. RGWGetObj_BlockDecrypt::fixup_range
+ * expects plaintext-domain input, but project_encrypt_range()
+ * clamps the projected encrypted range to encrypted_total_size,
+ * so the final read range is correct regardless. When a decomp
+ * filter is downstream, the decrypt call cascades into decomp
+ * with that intermediate (encrypted-as-plaintext) range, which
+ * is wrong for the decomp side — the second decomp->fixup_range()
+ * below re-initializes it with the proper plaintext range.
+ */
+ if (decrypt_filter) {
+ off_t ofs = 0;
+ off_t end = static_cast<off_t>(on_disk_size) - 1;
+ filter->fixup_range(ofs, end);
+ }
+ if (decomp) {
+ off_t ofs = 0;
+ off_t end = orig_size - 1;
+ decomp->fixup_range(ofs, end);
+ }
+
+ return 0;
+ }
+
+ void finalize_attrs(rgw::sal::Attrs& attrs) override {
+ if (compressor && compressor->is_compressed()) {
+ bufferlist tmp;
+ RGWCompressionInfo cs_info;
+ cs_info.compression_type = compressor_plugin->get_type_name();
+ cs_info.orig_size = orig_size;
+ cs_info.compressor_message = compressor->get_compressor_message();
+ cs_info.blocks = std::move(compressor->get_compression_blocks());
+ encode(cs_info, tmp);
+ attrs[RGW_ATTR_COMPRESSION] = tmp;
+ }
+ /*
+ * AEAD bookkeeping for the re-encrypted dest: ORIGINAL_SIZE must
+ * match the plaintext payload size. set_writer() guarantees
+ * orig_size is plaintext on every path (decomp output, compressed-
+ * passthrough cs_info.orig_size, AEAD ORIGINAL_SIZE/formula, or
+ * the source's logical size). PART_NUMS is cleared because re-
+ * encryption always produces a single stream. SALT and
+ * PREFETCH_ALIGN are written by get_encrypt_crypt().
+ */
+ if (encrypt_filter &&
+ is_aead_mode(get_str_attribute(attrs, RGW_ATTR_CRYPT_MODE))) {
+ bufferlist sz_bl;
+ sz_bl.append(std::to_string(orig_size));
+ attrs[RGW_ATTR_CRYPT_ORIGINAL_SIZE] = std::move(sz_bl);
+ attrs.erase(RGW_ATTR_CRYPT_PART_NUMS);
+ }
+ }
+
+ uint64_t get_accounted_size(uint64_t default_size) override {
+ if (decomp)
+ return orig_size;
+ return default_size;
+ }
+
+ // LC only constructs DPF when transformation is decided
+ bool need_copy_data() override {
+ return true;
+ }
+
+ RGWGetObj_Filter* get_filter() override { return filter; }
+};
* Removing those comments makes that work harder.
* February 25, 2021
*********************************************************************/
+
+/*
+ * Build an AES_256_GCM decrypt crypt from the salt in attrs and the
+ * caller-supplied user key. Zeroizes user_key on every path.
+ */
+static int build_aead_decrypt_crypt(
+ const DoutPrefixProvider* dpp, CephContext* cct,
+ const rgw::sal::Attrs& attrs,
+ std::string& user_key,
+ const std::string& bucket_id,
+ const std::string& object_name,
+ const std::string& domain,
+ std::unique_ptr<BlockCrypt>* block_crypt)
+{
+ auto cleanup = [&]() {
+ ::ceph::crypto::zeroize_for_security(user_key.data(), user_key.length());
+ };
+ if (user_key.size() != AES_256_KEYSIZE) {
+ cleanup();
+ ldpp_dout(dpp, 0) << "ERROR: " << domain << " key not 256 bits" << dendl;
+ return -EINVAL;
+ }
+ std::string salt = get_str_attribute(attrs, RGW_ATTR_CRYPT_SALT);
+ if (salt.size() != AES_256_GCM_SALT_SIZE) {
+ cleanup();
+ ldpp_dout(dpp, 0) << "ERROR: " << domain
+ << " missing/invalid " RGW_ATTR_CRYPT_SALT << dendl;
+ return -EIO;
+ }
+ auto aes = AES_256_GCM_create(dpp, cct,
+ reinterpret_cast<const uint8_t*>(user_key.c_str()), AES_256_KEYSIZE,
+ reinterpret_cast<const uint8_t*>(salt.c_str()), salt.size(), 0);
+ auto* gcm = dynamic_cast<AES_256_GCM*>(aes.get());
+ if (!gcm || !gcm->derive_object_key(
+ reinterpret_cast<const uint8_t*>(user_key.c_str()), AES_256_KEYSIZE,
+ bucket_id, object_name, 0, domain)) {
+ cleanup();
+ ldpp_dout(dpp, 0) << "ERROR: " << domain << " key derivation failed" << dendl;
+ return -EIO;
+ }
+ cleanup();
+ *block_crypt = std::move(aes);
+ return 0;
+}
+
+int rgw_prepare_decrypt_object(const DoutPrefixProvider* dpp,
+ CephContext* cct,
+ const rgw::sal::Attrs& attrs,
+ const std::string& bucket_id,
+ const std::string& object_name,
+ optional_yield y,
+ std::unique_ptr<BlockCrypt>* block_crypt)
+{
+ *block_crypt = nullptr;
+
+ auto mode_iter = attrs.find(RGW_ATTR_CRYPT_MODE);
+ if (mode_iter == attrs.end()) {
+ return 0;
+ }
+ std::string mode = mode_iter->second.to_str();
+
+ if (mode == "SSE-C-AES256" || mode == "SSE-C-AES256-GCM") {
+ return -ENOTSUP;
+ }
+
+ if (mode == "SSE-KMS") {
+ std::string actual_key;
+ auto mutable_attrs = attrs;
+ int r = reconstitute_actual_key_from_kms(dpp, mutable_attrs, nullptr, y, actual_key);
+ if (r != 0) {
+ ldpp_dout(dpp, 10) << "ERROR: failed to retrieve KMS key" << dendl;
+ return r;
+ }
+ if (actual_key.size() != AES_256_KEYSIZE) {
+ ldpp_dout(dpp, 0) << "ERROR: KMS key is not 256 bits" << dendl;
+ actual_key.replace(0, actual_key.length(), actual_key.length(), '\000');
+ return -EINVAL;
+ }
+ auto aes = std::unique_ptr<AES_256_CBC>(new AES_256_CBC(dpp, cct));
+ aes->set_key(reinterpret_cast<const uint8_t*>(actual_key.c_str()),
+ AES_256_KEYSIZE);
+ actual_key.replace(0, actual_key.length(), actual_key.length(), '\000');
+ *block_crypt = std::move(aes);
+ return 0;
+ }
+
+ if (mode == "RGW-AUTO") {
+ std::string master_encryption_key;
+ auto wipe_master = [&]() {
+ ::ceph::crypto::zeroize_for_security(
+ master_encryption_key.data(), master_encryption_key.length());
+ };
+ try {
+ master_encryption_key = from_base64(
+ std::string(cct->_conf->rgw_crypt_default_encryption_key));
+ } catch (...) {
+ wipe_master();
+ ldpp_dout(dpp, 5) << "ERROR: invalid default encryption key" << dendl;
+ return -EINVAL;
+ }
+ if (master_encryption_key.size() != 256 / 8) {
+ wipe_master();
+ ldpp_dout(dpp, 0) << "ERROR: default encryption key not 256 bits"
+ << dendl;
+ return -EINVAL;
+ }
+ std::string attr_key_selector =
+ get_str_attribute(attrs, RGW_ATTR_CRYPT_KEYSEL);
+ if (attr_key_selector.size() != AES_256_CBC::AES_256_KEYSIZE) {
+ wipe_master();
+ ldpp_dout(dpp, 0) << "ERROR: missing or invalid "
+ RGW_ATTR_CRYPT_KEYSEL << dendl;
+ return -EIO;
+ }
+ uint8_t actual_key[AES_256_KEYSIZE];
+ if (!AES_256_ECB_encrypt(dpp, cct,
+ reinterpret_cast<const uint8_t*>(
+ master_encryption_key.c_str()),
+ AES_256_KEYSIZE,
+ reinterpret_cast<const uint8_t*>(
+ attr_key_selector.c_str()),
+ actual_key, AES_256_KEYSIZE)) {
+ ::ceph::crypto::zeroize_for_security(actual_key, sizeof(actual_key));
+ wipe_master();
+ return -EIO;
+ }
+ auto aes = std::unique_ptr<AES_256_CBC>(new AES_256_CBC(dpp, cct));
+ aes->set_key(actual_key, AES_256_KEYSIZE);
+ ::ceph::crypto::zeroize_for_security(actual_key, sizeof(actual_key));
+ wipe_master();
+ *block_crypt = std::move(aes);
+ return 0;
+ }
+
+ if (mode == "AES256") {
+ std::string actual_key;
+ auto mutable_attrs = attrs;
+ int r = reconstitute_actual_key_from_sse_s3(dpp, mutable_attrs, y, actual_key);
+ if (r != 0) {
+ ldpp_dout(dpp, 10) << "ERROR: failed to retrieve SSE-S3 key" << dendl;
+ return r;
+ }
+ if (actual_key.size() != AES_256_KEYSIZE) {
+ ldpp_dout(dpp, 0) << "ERROR: SSE-S3 key is not 256 bits" << dendl;
+ actual_key.replace(0, actual_key.length(), actual_key.length(), '\000');
+ return -EINVAL;
+ }
+ auto aes = std::unique_ptr<AES_256_CBC>(new AES_256_CBC(dpp, cct));
+ aes->set_key(reinterpret_cast<const uint8_t*>(actual_key.c_str()),
+ AES_256_KEYSIZE);
+ actual_key.replace(0, actual_key.length(), actual_key.length(), '\000');
+ *block_crypt = std::move(aes);
+ return 0;
+ }
+
+ if (mode == "SSE-KMS-GCM") {
+ std::string actual_key;
+ auto mutable_attrs = attrs;
+ int r = reconstitute_actual_key_from_kms(dpp, mutable_attrs, nullptr, y, actual_key);
+ if (r != 0) {
+ ldpp_dout(dpp, 10) << "ERROR: failed to retrieve KMS key" << dendl;
+ return r;
+ }
+ return build_aead_decrypt_crypt(dpp, cct, attrs, actual_key,
+ bucket_id, object_name,
+ "SSE-KMS-GCM", block_crypt);
+ }
+
+ if (mode == "AES256-GCM") {
+ std::string actual_key;
+ auto mutable_attrs = attrs;
+ int r = reconstitute_actual_key_from_sse_s3(dpp, mutable_attrs, y, actual_key);
+ if (r != 0) {
+ ldpp_dout(dpp, 10) << "ERROR: failed to retrieve SSE-S3 key" << dendl;
+ return r;
+ }
+ return build_aead_decrypt_crypt(dpp, cct, attrs, actual_key,
+ bucket_id, object_name,
+ "AES256-GCM", block_crypt);
+ }
+
+ if (mode == "RGW-AUTO-GCM") {
+ std::string master_encryption_key;
+ try {
+ master_encryption_key = from_base64(
+ std::string(cct->_conf->rgw_crypt_default_encryption_key));
+ } catch (...) {
+ ::ceph::crypto::zeroize_for_security(
+ master_encryption_key.data(), master_encryption_key.length());
+ ldpp_dout(dpp, 5) << "ERROR: invalid default encryption key" << dendl;
+ return -EINVAL;
+ }
+ return build_aead_decrypt_crypt(dpp, cct, attrs, master_encryption_key,
+ bucket_id, object_name,
+ "RGW-AUTO-GCM", block_crypt);
+ }
+
+ return -ENOTSUP;
+}
+
+int rgw_prepare_reencrypt_object(const DoutPrefixProvider* dpp,
+ CephContext* cct,
+ rgw::sal::Attrs& dest_attrs,
+ const std::string& bucket_id,
+ const std::string& object_name,
+ optional_yield y,
+ std::unique_ptr<BlockCrypt>* block_crypt)
+{
+ *block_crypt = nullptr;
+
+ const std::string mode = get_str_attribute(dest_attrs, RGW_ATTR_CRYPT_MODE);
+ if (mode.empty()) {
+ return 0;
+ }
+
+ if (!is_aead_mode(mode)) {
+ return rgw_prepare_decrypt_object(dpp, cct, dest_attrs,
+ bucket_id, object_name,
+ y, block_crypt);
+ }
+
+ /*
+ * For AEAD, stage a fresh salt so that key derivation produces a
+ * different per-object key on each re-encrypt. Without this, the
+ * same key+salt would reuse GCM IVs across distinct plaintexts
+ * (e.g., after recompression). Roll back on failure so dest_attrs
+ * stays consistent with the on-disk state.
+ */
+ auto saved_iter = dest_attrs.find(RGW_ATTR_CRYPT_SALT);
+ const bool had_salt = (saved_iter != dest_attrs.end());
+ bufferlist saved_bl;
+ if (had_salt) {
+ saved_bl = saved_iter->second;
+ }
+
+ std::string salt(AES_256_GCM_SALT_SIZE, '\0');
+ cct->random()->get_bytes(salt.data(), AES_256_GCM_SALT_SIZE);
+ bufferlist new_salt_bl;
+ new_salt_bl.append(salt);
+ dest_attrs[RGW_ATTR_CRYPT_SALT] = std::move(new_salt_bl);
+
+ int r = rgw_prepare_decrypt_object(dpp, cct, dest_attrs,
+ bucket_id, object_name,
+ y, block_crypt);
+ if (r < 0) {
+ if (had_salt) {
+ dest_attrs[RGW_ATTR_CRYPT_SALT] = std::move(saved_bl);
+ } else {
+ dest_attrs.erase(RGW_ATTR_CRYPT_SALT);
+ }
+ }
+ return r;
+}
+
int process(bufferlist&& data, uint64_t logical_offset) override;
}; /* RGWPutObj_BlockEncrypt */
+/*
+ * Build a decrypt crypt from stored attrs (LC: no req_state).
+ * bucket_id/object_name are used only for AEAD key derivation.
+ * Returns -ENOTSUP for SSE-C; 0 + nullptr for unencrypted.
+ */
+int rgw_prepare_decrypt_object(const DoutPrefixProvider* dpp,
+ CephContext* cct,
+ const rgw::sal::Attrs& attrs,
+ const std::string& bucket_id,
+ const std::string& object_name,
+ optional_yield y,
+ std::unique_ptr<BlockCrypt>* block_crypt);
+
+/*
+ * Build an encrypt crypt for LC in-place re-encryption. For AEAD,
+ * regenerates dest_attrs[CRYPT_SALT] so the re-derived per-object
+ * key differs from the source key (avoids GCM nonce reuse).
+ */
+int rgw_prepare_reencrypt_object(const DoutPrefixProvider* dpp,
+ CephContext* cct,
+ rgw::sal::Attrs& dest_attrs,
+ const std::string& bucket_id,
+ const std::string& object_name,
+ optional_yield y,
+ std::unique_ptr<BlockCrypt>* block_crypt);
+
int rgw_s3_prepare_encrypt(req_state* s, optional_yield y,
std::map<std::string, ceph::bufferlist>& attrs,
}
}
-class RGWCopyObjDPF : public rgw::sal::DataProcessorFactory {
- rgw::sal::Driver* driver;
+static bool copy_dest_requests_encryption(req_state* s)
+{
+ for (const auto& attr : {"x-amz-server-side-encryption-customer-algorithm",
+ "x-amz-server-side-encryption"}) {
+ if (s->info.crypt_attribute_map.count(attr)) {
+ return true;
+ }
+ }
+ /*
+ * rgw_crypt_default_encryption_key produces an RGW-AUTO encrypt
+ * filter without any explicit SSE header (bucket-level encryption
+ * already synthesizes the header upstream and is covered above).
+ */
+ if (!s->cct->_conf->rgw_crypt_default_encryption_key.empty()) {
+ return true;
+ }
+ return false;
+}
+
+class RGWCopyObjDPF : public RGWRecompressDPF {
req_state* s;
- uint64_t &obj_size;
+ std::string normalized_dest_compression;
std::map<std::string, std::string>& crypt_http_responses;
- DataProcessorFilter cb;
- RGWGetObj_Filter* filter{&cb};
- bool need_decompress{false};
- RGWCompressionInfo decompress_info;
- boost::optional<RGWGetObj_Decompress> decompress;
- std::unique_ptr<RGWGetObj_Filter> decrypt;
- std::unique_ptr<rgw::sal::DataProcessor> encrypt;
- std::optional<RGWPutObj_Compress> compressor;
- CompressorRef compressor_plugin;
- off_t ofs_x{0};
- off_t end_x = obj_size;
-
-public:
- RGWCopyObjDPF(rgw::sal::Driver* _driver,
- req_state* _s,
- uint64_t& _obj_size,
- std::map<std::string, std::string>& _crypt_http_responses)
- : driver(_driver),
- s(_s),
- obj_size(_obj_size),
- crypt_http_responses(_crypt_http_responses)
- {}
- ~RGWCopyObjDPF() override {}
- int get_encrypt_filter(std::unique_ptr<rgw::sal::DataProcessor> *filter,
- rgw::sal::DataProcessor *cb,
- rgw::sal::Attrs& attrs)
+protected:
+ int get_decrypt_crypt(const DoutPrefixProvider* dpp,
+ optional_yield y,
+ const rgw::sal::Attrs& src_attrs,
+ std::unique_ptr<BlockCrypt>* crypt) override
{
- std::unique_ptr<BlockCrypt> block_crypt;
- int res = rgw_s3_prepare_encrypt(s, s->yield, attrs, &block_crypt,
- crypt_http_responses);
- if (res == 0 && block_crypt != nullptr) {
- filter->reset(new RGWPutObj_BlockEncrypt(s, s->cct, cb, std::move(block_crypt), s->yield));
- }
- return res;
+ // copy: rgw_s3_prepare_decrypt takes non-const attrs
+ auto mutable_attrs = src_attrs;
+ return rgw_s3_prepare_decrypt(s, y, mutable_attrs, crypt,
+ nullptr, true /* copy_source */);
}
- int set_writer(rgw::sal::DataProcessor* writer,
- rgw::sal::Attrs& attrs,
- const DoutPrefixProvider *dpp,
- optional_yield y) override
+ int get_encrypt_crypt(const DoutPrefixProvider* dpp,
+ optional_yield y,
+ rgw::sal::Attrs& dest_attrs,
+ std::unique_ptr<BlockCrypt>* crypt) override
{
- /* RGWGetObj_Filter */
- // decompress
- int ret = rgw_compression_info_from_attrset(s->src_object->get_attrs(), need_decompress, decompress_info);
- if (ret < 0) {
- return ret;
- }
-
- bool src_encrypted = s->src_object->get_attrs().count(RGW_ATTR_CRYPT_MODE);
- // Preserve encrypted size for decrypt range clamping before any plaintext conversion.
- off_t encrypted_total_size = obj_size;
-
- // Create decompress filter if source is compressed.
- // Must be created BEFORE decrypt so the chain is: decrypt → decompress → cb
- if (need_decompress) {
- static constexpr bool partial_content = false;
- decompress.emplace(s->cct, &decompress_info, partial_content, filter);
- filter = &*decompress;
- }
-
- // decrypt
- if (src_encrypted) {
- auto attr_iter = s->src_object->get_attrs().find(RGW_ATTR_MANIFEST);
- static constexpr bool copy_source = true;
-
- // part_num=0 for copy source (full object read)
- ret = get_decrypt_filter(&decrypt, filter, s, s->src_object->get_attrs(),
- attr_iter != s->src_object->get_attrs().end() ? &attr_iter->second : nullptr,
- nullptr, copy_source, 0, encrypted_total_size);
- if (ret < 0) {
- return ret;
- }
- if (decrypt != nullptr) {
- filter = decrypt.get();
- }
- }
-
- // Set obj_size to the final output size (plaintext) for range handling.
- if (need_decompress) {
- obj_size = decompress_info.orig_size;
- s->src_object->set_obj_size(obj_size);
- end_x = obj_size;
- } else if (src_encrypted) {
- uint64_t decrypted_size = 0;
- const auto& src_attrs = s->src_object->get_attrs();
- if (rgw_calc_aead_obj_size(dpp, src_attrs, encrypted_total_size,
- src_attrs.count(RGW_ATTR_COMPRESSION), &decrypted_size)) {
- obj_size = decrypted_size;
- s->src_object->set_obj_size(obj_size);
- end_x = obj_size;
- }
- }
-
- filter->fixup_range(ofs_x, end_x);
-
- /* rgw::sal::DataProcessor */
- // encrypt
- rgw::sal::DataProcessor* processor = writer;
-
- ret = get_encrypt_filter(&encrypt, processor, attrs);
- if (ret < 0) {
- return ret;
- }
- if (encrypt != nullptr) {
- processor = &*encrypt;
- }
-
- // compression
- attrs.erase(RGW_ATTR_COMPRESSION); // remove any existing compression info from source object
- // a zonegroup feature is required to combine compression and encryption
- const RGWZoneGroup& zonegroup = s->penv.site->get_zonegroup();
- const bool compress_encrypted = zonegroup.supports(rgw::zone_features::compress_encrypted);
- const auto& compression_type = driver->get_compression_type(s->dest_placement);
- if (compression_type != "none" &&
- (encrypt == nullptr || compress_encrypted)) {
- compressor_plugin = get_compressor_plugin(s, compression_type);
- if (!compressor_plugin) {
- ldpp_dout(s, 1) << "Cannot load plugin for compression type "
- << compression_type << dendl;
- } else {
- compressor.emplace(s->cct, compressor_plugin, processor);
- processor = &*compressor;
- // always send incompressible hint when rgw is itself doing compression
- s->object->set_compressed();
- }
+ // strip all old crypt attrs — CopyObj may change key/mode
+ auto it = dest_attrs.lower_bound(RGW_ATTR_CRYPT_PREFIX);
+ while (it != dest_attrs.end() &&
+ it->first.compare(0, sizeof(RGW_ATTR_CRYPT_PREFIX) - 1,
+ RGW_ATTR_CRYPT_PREFIX) == 0) {
+ it = dest_attrs.erase(it);
}
+ std::unique_ptr<BlockCrypt> block_crypt;
+ int ret = rgw_s3_prepare_encrypt(s, y, dest_attrs, &block_crypt,
+ crypt_http_responses);
+ if (ret < 0) return ret;
+ *crypt = std::move(block_crypt);
+ return 0;
+ }
- cb.set_processor(processor);
+ const std::string& get_dest_compression() override {
+ return normalized_dest_compression;
+ }
- return 0;
+ bool supports_compress_encrypted() override {
+ return s->penv.site->get_zonegroup().supports(
+ rgw::zone_features::compress_encrypted);
}
- bool need_copy_data() override {
- // if source object is encrypted, we need to copy data
- if (s->src_object->get_attrs().count(RGW_ATTR_CRYPT_MODE)) {
- return true;
- }
+ // stable codec for multipart copies
+ CompressorRef create_compressor(const std::string& type) override {
+ return get_compressor_plugin(s, type);
+ }
- // check if it's requested to be encrypted
- static const string crypt_attrs[] = {
- "x-amz-server-side-encryption-customer-algorithm", // SSE-C
- "x-amz-server-side-encryption", // SSE-S3, SSE-KMS
- };
- for (const auto& attr : crypt_attrs) {
- if (s->info.crypt_attribute_map.find(attr) !=
- s->info.crypt_attribute_map.end()) {
- return true;
- }
- }
+ void mark_compressed() override {
+ s->object->set_compressed();
+ }
- return false;
+ void on_obj_size_changed(uint64_t new_size) override {
+ // The src_object's in-memory state is what downstream consumers
+ // in RGWCopyObj::execute (bucket logging, publish_commit, op
+ // counters) read via the obj_size reference held by the base.
+ s->src_object->set_obj_size(new_size);
}
- RGWGetObj_Filter* get_filter() override {
- return filter;
+ bool dest_requests_encryption() const {
+ return copy_dest_requests_encryption(s);
}
- void finalize_attrs(rgw::sal::Attrs& attrs) override {
- if (compressor && compressor->is_compressed()) {
- bufferlist tmp;
- RGWCompressionInfo cs_info;
- assert(compressor_plugin != nullptr);
- // plugin exists when the compressor does
- // coverity[dereference:SUPPRESS]
- cs_info.compression_type = compressor_plugin->get_type_name();
- cs_info.orig_size = obj_size;
- cs_info.compressor_message = compressor->get_compressor_message();
- cs_info.blocks = std::move(compressor->get_compression_blocks());
+public:
+ RGWCopyObjDPF(req_state* s_,
+ uint64_t& obj_size,
+ const rgw::sal::Attrs& src_attrs,
+ const std::string& dest_compression,
+ std::map<std::string, std::string>& crypt_http_responses_)
+ : RGWRecompressDPF(s_->cct, obj_size, src_attrs),
+ s(s_),
+ normalized_dest_compression(dest_compression),
+ crypt_http_responses(crypt_http_responses_)
+ {}
- encode(cs_info, tmp);
- attrs[RGW_ATTR_COMPRESSION] = tmp;
+ bool need_copy_data() override {
+ // source is encrypted
+ if (src_attrs.count(RGW_ATTR_CRYPT_MODE)) return true;
- ldpp_dout(s, 20) << "storing " << RGW_ATTR_COMPRESSION
- << " with type=" << cs_info.compression_type
- << ", orig_size=" << cs_info.orig_size
- << ", compressor_message=" << cs_info.compressor_message
- << ", blocks=" << cs_info.blocks.size() << dendl;
- }
+ // destination encryption requires slow path
+ if (dest_requests_encryption()) return true;
- // Copy path: ensure AEAD ORIGINAL_SIZE matches copied payload size.
- // If it doesn't, stale CRYPT_PARTS and CRYPT_PART_NUMS must be dropped.
- const auto mode = get_str_attribute(attrs, RGW_ATTR_CRYPT_MODE);
- if (is_aead_mode(mode)) {
- uint64_t existing = 0;
- if (!rgw_get_aead_original_size(s, attrs, &existing) ||
- existing != obj_size) {
- set_attr(attrs, RGW_ATTR_CRYPT_ORIGINAL_SIZE, std::to_string(obj_size));
- attrs.erase(RGW_ATTR_CRYPT_PARTS);
- attrs.erase(RGW_ATTR_CRYPT_PART_NUMS);
- }
+ // compression change: preserve fast path for same-codec copies
+ bool src_compressed = false;
+ RGWCompressionInfo cs_info;
+ if (rgw_compression_info_from_attrset(src_attrs, src_compressed,
+ cs_info) < 0) {
+ return true;
}
+ bool compression_matches =
+ normalized_dest_compression != "random" &&
+ ((!src_compressed && normalized_dest_compression == "none") ||
+ (src_compressed &&
+ cs_info.compression_type == normalized_dest_compression));
+ return !compression_matches;
}
};
return;
}
- RGWCopyObjDPF copy_obj_dpf(driver, s, obj_size, crypt_http_responses);
+ /*
+ * Compute the normalized destination compression for CopyObject.
+ * When the effective destination will be encrypted and the zonegroup
+ * doesn't support compress_encrypted, force "none" to prevent
+ * writing an incompatible compressed+encrypted layout.
+ */
+ const auto& raw_dest_compression =
+ driver->get_compression_type(s->dest_placement);
+ std::string dest_compression = raw_dest_compression;
+
+ if (copy_dest_requests_encryption(s) &&
+ !s->penv.site->get_zonegroup().supports(
+ rgw::zone_features::compress_encrypted)) {
+ dest_compression = "none";
+ }
+
+ RGWCopyObjDPF copy_obj_dpf(s, obj_size, s->src_object->get_attrs(),
+ dest_compression, crypt_http_responses);
op_ret = s->src_object->copy_object(s->owner,
s->user->get_id(),
virtual RGWGetObj_Filter* get_filter() = 0;
virtual bool need_copy_data() = 0;
virtual void finalize_attrs(Attrs& attrs) { /* default implementation does nothing */ }
+
+ /*
+ * Override the accounted size for the bucket index. Called after
+ * finalize_attrs(). Returns the logical object size when the
+ * factory knows better than the default heuristic, e.g. after
+ * decompressing data whose recompression was a no-op.
+ */
+ virtual uint64_t get_accounted_size(uint64_t default_size) {
+ return default_size;
+ }
};
/**
}
+/*
+ * Verify the encrypt → decrypt filter pipeline produces
+ * identical plaintext. Full helper create() tests require
+ * SAL mocks and are deferred to integration testing.
+ */
+TEST(TestRGWCrypto, verify_transition_encrypt_decrypt_roundtrip)
+{
+ const NoDoutPrefix no_dpp(g_ceph_context, dout_subsys);
+
+ // test both block-aligned and non-aligned sizes
+ for (size_t test_size : {8192, 7777}) {
+ uint8_t key[32];
+ for (size_t i = 0; i < sizeof(key); i++)
+ key[i] = i * 3 + 7;
+
+ std::vector<uint8_t> plaintext(test_size);
+ for (size_t i = 0; i < test_size; i++)
+ plaintext[i] = (i * 7 + 13) & 0xff;
+
+ // encrypt
+ ut_put_sink put_sink;
+ {
+ auto cbc = AES_256_CBC_create(&no_dpp, g_ceph_context, key, 32);
+ ASSERT_NE(cbc.get(), nullptr);
+ RGWPutObj_BlockEncrypt encrypt(&no_dpp, g_ceph_context, &put_sink,
+ std::move(cbc), null_yield);
+ bufferlist bl;
+ bl.append((char*)plaintext.data(), test_size);
+ ASSERT_EQ(encrypt.process(std::move(bl), 0), 0);
+ ASSERT_EQ(encrypt.process({}, test_size), 0);
+ }
+ std::string ciphertext = put_sink.get_sink();
+
+ // decrypt with a different key instance (same bytes)
+ ut_get_sink get_sink;
+ {
+ auto cbc = AES_256_CBC_create(&no_dpp, g_ceph_context, key, 32);
+ ASSERT_NE(cbc.get(), nullptr);
+ RGWGetObj_Filter* filter = &get_sink;
+ RGWGetObj_BlockDecrypt decrypt(&no_dpp, g_ceph_context, filter,
+ std::move(cbc),
+ std::vector<size_t>{}, null_yield);
+ filter = &decrypt;
+
+ off_t ofs = 0;
+ off_t end = test_size - 1;
+ filter->fixup_range(ofs, end);
+
+ bufferlist enc_bl;
+ enc_bl.append(ciphertext.data(), ciphertext.size());
+ ASSERT_EQ(filter->handle_data(enc_bl, ofs, enc_bl.length()), 0);
+ ASSERT_EQ(filter->flush(), 0);
+ }
+
+ std::string result = get_sink.get_sink();
+ ASSERT_EQ(result.size(), test_size);
+ ASSERT_EQ(memcmp(result.data(), plaintext.data(), test_size), 0);
+ }
+}
+
int main(int argc, char **argv) {
auto args = argv_to_vec(argc, argv);
auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT,
projects / ceph.git / commitdiff