mirror of
https://github.com/ceph/ceph
synced 2026-08-01 22:45:39 +00:00
Merge pull request #67725 from mheler/wip-rgw-transition-recompress
rgw: compress/decompress objects during lifecycle transitions Reviewed-by: Casey Bodley <cbodley@redhat.com>
This commit is contained in:
Adam Emerson
GitHub
commit
423acf868d
@ -14,8 +14,8 @@ overrides:
|
||||
rgw lc debug interval: 10
|
||||
rgw:
|
||||
storage classes:
|
||||
LUKEWARM:
|
||||
FROZEN:
|
||||
LUKEWARM: [--compression, zstd]
|
||||
FROZEN: [--compression, zlib]
|
||||
s3tests:
|
||||
accounts:
|
||||
iam root: RGW88888888888888888
|
||||
|
||||
1
qa/suites/rgw/lifecycle/transition$/.qa
Symbolic link
1
qa/suites/rgw/lifecycle/transition$/.qa
Symbolic link
@ -0,0 +1 @@
|
||||
../.qa/
|
||||
@ -0,0 +1,8 @@
|
||||
overrides:
|
||||
rgw:
|
||||
enable features: [compress-encrypted]
|
||||
tasks:
|
||||
- workunit:
|
||||
clients:
|
||||
client.0:
|
||||
- rgw/run-lc-encrypt-recompress.sh
|
||||
5
qa/suites/rgw/lifecycle/transition$/compressed.yaml
Normal file
5
qa/suites/rgw/lifecycle/transition$/compressed.yaml
Normal file
@ -0,0 +1,5 @@
|
||||
tasks:
|
||||
- workunit:
|
||||
clients:
|
||||
client.0:
|
||||
- rgw/run-lc-recompress.sh
|
||||
@ -0,0 +1,12 @@
|
||||
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
|
||||
14
qa/workunits/rgw/run-lc-encrypt-recompress.sh
Executable file
14
qa/workunits/rgw/run-lc-encrypt-recompress.sh
Executable file
@ -0,0 +1,14 @@
|
||||
#!/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.
|
||||
14
qa/workunits/rgw/run-lc-recompress.sh
Executable file
14
qa/workunits/rgw/run-lc-recompress.sh
Executable file
@ -0,0 +1,14 @@
|
||||
#!/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.
|
||||
356
qa/workunits/rgw/test_rgw_lc_recompress.py
Executable file
356
qa/workunits/rgw/test_rgw_lc_recompress.py
Executable file
@ -0,0 +1,356 @@
|
||||
#!/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()
|
||||
@ -38,6 +38,7 @@
|
||||
#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"
|
||||
@ -4002,6 +4003,8 @@ int RGWRados::rewrite_obj(RGWBucketInfo& dest_bucket_info, const rgw_obj& obj, c
|
||||
attrset.erase(RGW_ATTR_ID_TAG);
|
||||
attrset.erase(RGW_ATTR_TAIL_TAG);
|
||||
attrset.erase(RGW_ATTR_STORAGE_CLASS);
|
||||
attrset.erase(RGW_ATTR_SHARE_MANIFEST);
|
||||
attrset.erase(RGW_ATTR_BLAKE3);
|
||||
|
||||
ACLOwner owner;
|
||||
if (auto i = attrset.find(RGW_ATTR_ACL); i != attrset.end()) {
|
||||
@ -4893,6 +4896,8 @@ int RGWRados::fetch_remote_obj(RGWObjectCtx& dest_obj_ctx,
|
||||
if (!keep_tags) {
|
||||
attrs.erase(RGW_ATTR_TAGS);
|
||||
}
|
||||
attrs.erase(RGW_ATTR_SHARE_MANIFEST);
|
||||
attrs.erase(RGW_ATTR_BLAKE3);
|
||||
|
||||
if (copy_if_newer) {
|
||||
uint64_t pg_ver = 0;
|
||||
@ -5303,6 +5308,8 @@ int RGWRados::copy_obj(RGWObjectCtx& src_obj_ctx,
|
||||
|
||||
if (copy_data) { /* refcounting tail wouldn't work here, just copy the data */
|
||||
attrs.erase(RGW_ATTR_TAIL_TAG);
|
||||
attrs.erase(RGW_ATTR_SHARE_MANIFEST);
|
||||
attrs.erase(RGW_ATTR_BLAKE3);
|
||||
// Data is rewritten as a single stream; drop stale multipart boundaries
|
||||
attrs.erase(RGW_ATTR_CRYPT_PARTS);
|
||||
attrs.erase(RGW_ATTR_CRYPT_PART_NUMS);
|
||||
@ -5344,6 +5351,8 @@ int RGWRados::copy_obj(RGWObjectCtx& src_obj_ctx,
|
||||
if (!copy_itself) {
|
||||
aio = rgw::make_throttle(cct->_conf->rgw_max_copy_obj_concurrent_io, y);
|
||||
attrs.erase(RGW_ATTR_TAIL_TAG);
|
||||
attrs.erase(RGW_ATTR_SHARE_MANIFEST);
|
||||
attrs.erase(RGW_ATTR_BLAKE3);
|
||||
manifest = *amanifest;
|
||||
const rgw_bucket_placement& tail_placement = manifest.get_tail_placement();
|
||||
if (tail_placement.bucket.name.empty()) {
|
||||
@ -5558,6 +5567,10 @@ int RGWRados::copy_obj_data(RGWObjectCtx& obj_ctx,
|
||||
: (src_accounted_size ? src_accounted_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,
|
||||
@ -5594,6 +5607,68 @@ int fixup_manifest_to_parts_len(const DoutPrefixProvider *dpp, rgw::sal::Attrs &
|
||||
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,
|
||||
@ -5639,12 +5714,116 @@ int RGWRados::transition_obj(RGWObjectCtx& obj_ctx,
|
||||
}
|
||||
attrs.erase(RGW_ATTR_ID_TAG);
|
||||
attrs.erase(RGW_ATTR_TAIL_TAG);
|
||||
attrs.erase(RGW_ATTR_SHARE_MANIFEST);
|
||||
attrs.erase(RGW_ATTR_BLAKE3);
|
||||
|
||||
ACLOwner owner;
|
||||
if (auto i = attrs.find(RGW_ATTR_ACL); i != attrs.end()) {
|
||||
(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,
|
||||
@ -5658,7 +5837,7 @@ int RGWRados::transition_obj(RGWObjectCtx& obj_ctx,
|
||||
olh_epoch,
|
||||
real_time(),
|
||||
nullptr /* petag */,
|
||||
nullptr, /* dp_factory */
|
||||
dp_factory,
|
||||
dpp,
|
||||
y,
|
||||
0, /* src_accounted_size: unknown here, keep ofs-based size */
|
||||
|
||||
@ -3,12 +3,16 @@
|
||||
|
||||
#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,
|
||||
@ -60,3 +64,352 @@ public:
|
||||
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; }
|
||||
};
|
||||
|
||||
@ -3041,3 +3041,257 @@ int rgw_remove_sse_s3_bucket_key(req_state *s, optional_yield y)
|
||||
* 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;
|
||||
}
|
||||
|
||||
|
||||
@ -357,6 +357,32 @@ public:
|
||||
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,
|
||||
|
||||
@ -6,6 +6,7 @@
|
||||
*/
|
||||
|
||||
#include <sys/stat.h>
|
||||
#include <optional>
|
||||
#include "include/str_map.h"
|
||||
#include "common/safe_io.h"
|
||||
#include "rgw/rgw_crypt.h"
|
||||
@ -1191,10 +1192,12 @@ static int maybe_cache_kms_fetch(
|
||||
if (kms_cache == nullptr ||
|
||||
!dpp->get_cct()->_conf->rgw_crypt_s3_kms_cache_enabled) {
|
||||
const auto ret = fetch(actual_key);
|
||||
if (ret == -ENOENT) {
|
||||
perfcounter->inc(l_rgw_kms_error_permanent);
|
||||
} else if (ret < 0) {
|
||||
perfcounter->inc(l_rgw_kms_error_transient);
|
||||
if (perfcounter) {
|
||||
if (ret == -ENOENT) {
|
||||
perfcounter->inc(l_rgw_kms_error_permanent);
|
||||
} else if (ret < 0) {
|
||||
perfcounter->inc(l_rgw_kms_error_transient);
|
||||
}
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
@ -1235,7 +1238,10 @@ int reconstitute_actual_key_from_kms(
|
||||
}
|
||||
|
||||
const auto fetch = [&](std::string& out_secret) -> int {
|
||||
PerfGuard perf(perfcounter, l_rgw_kms_fetch_lat);
|
||||
std::optional<PerfGuard> perf;
|
||||
if (perfcounter) {
|
||||
perf.emplace(perfcounter, l_rgw_kms_fetch_lat);
|
||||
}
|
||||
if (RGW_SSE_KMS_BACKEND_BARBICAN == kms_backend) {
|
||||
return get_actual_key_from_barbican(dpp, key_id, y, out_secret);
|
||||
}
|
||||
|
||||
@ -6064,203 +6064,123 @@ void RGWDeleteObj::execute(optional_yield y)
|
||||
}
|
||||
}
|
||||
|
||||
class RGWCopyObjDPF : public rgw::sal::DataProcessorFactory {
|
||||
rgw::sal::Driver* driver;
|
||||
req_state* s;
|
||||
uint64_t &obj_size;
|
||||
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)
|
||||
{
|
||||
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));
|
||||
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;
|
||||
}
|
||||
return res;
|
||||
}
|
||||
/*
|
||||
* 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;
|
||||
std::string normalized_dest_compression;
|
||||
std::map<std::string, std::string>& crypt_http_responses;
|
||||
|
||||
protected:
|
||||
int get_decrypt_crypt(const DoutPrefixProvider* dpp,
|
||||
optional_yield y,
|
||||
const rgw::sal::Attrs& src_attrs,
|
||||
std::unique_ptr<BlockCrypt>* crypt) override
|
||||
{
|
||||
// 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;
|
||||
// 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);
|
||||
}
|
||||
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
cb.set_processor(processor);
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
const std::string& get_dest_compression() override {
|
||||
return normalized_dest_compression;
|
||||
}
|
||||
|
||||
bool supports_compress_encrypted() override {
|
||||
return s->penv.site->get_zonegroup().supports(
|
||||
rgw::zone_features::compress_encrypted);
|
||||
}
|
||||
|
||||
// stable codec for multipart copies
|
||||
CompressorRef create_compressor(const std::string& type) override {
|
||||
return get_compressor_plugin(s, type);
|
||||
}
|
||||
|
||||
void mark_compressed() override {
|
||||
s->object->set_compressed();
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
bool dest_requests_encryption() const {
|
||||
return copy_dest_requests_encryption(s);
|
||||
}
|
||||
|
||||
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_)
|
||||
{}
|
||||
|
||||
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)) {
|
||||
// source is encrypted
|
||||
if (src_attrs.count(RGW_ATTR_CRYPT_MODE)) return true;
|
||||
|
||||
// destination encryption requires slow path
|
||||
if (dest_requests_encryption()) return true;
|
||||
|
||||
// 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;
|
||||
}
|
||||
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
RGWGetObj_Filter* get_filter() override {
|
||||
return filter;
|
||||
}
|
||||
|
||||
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());
|
||||
|
||||
encode(cs_info, tmp);
|
||||
attrs[RGW_ATTR_COMPRESSION] = tmp;
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
// 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);
|
||||
}
|
||||
}
|
||||
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;
|
||||
}
|
||||
};
|
||||
|
||||
@ -6611,7 +6531,24 @@ void RGWCopyObj::execute(optional_yield y)
|
||||
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(),
|
||||
|
||||
@ -290,6 +290,16 @@ class DataProcessorFactory {
|
||||
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;
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
|
||||
@ -1523,6 +1523,66 @@ TEST(TestRGWCrypto, verify_PartLocation_no_manifest)
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
* 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,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
