openstack/swift
Code Issues Proposed changes

crypto - wrap random body key

Browse Source 
use a random key to encrypt the object body, wrap that body key
with keymaster provided object key and persist wrapped key in
object sysmeta header for crypto meta.

Wrap using AES CTR mode with IV, not aes_wrap_key algorithm,
for performance.

Change-Id: I4c2ba12dd35bb3f4951dce65bc7f488a332e0c18
This commit is contained in:
Alistair Coles 2016-05-06 17:29:42 +01:00
parent 3c8388bb8d
commit c5cf7a22f4
10 changed files with 368 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
swift/common/crypto_utils.py
View File

@ -79,18 +79,20 @@ def dump_crypto_meta(crypto_meta):
"""
Serialize crypto meta to a form suitable for including in a header value.
The IV value is random bytes and as a result needs to be encoded before
sending over the wire. Do this by wrapping the crypto meta in a json object
and encode the iv value. Base64 encoding returns a bytes object in py3, to
future proof the code, decode this data to produce a string, which is what
the json.dumps function expects.
The crypto-meta is serialized as a json object. The iv and key values are
random bytes and as a result need to be base64 encoded before sending over
the wire. Base64 encoding returns a bytes object in py3, to future proof
the code, decode this data to produce a string, which is what the
json.dumps function expects.
:param crypto_meta: a dict containing crypto meta items
:returns: a string serialization of a crypto meta dict
"""
# use sort_keys=True to make serialized form predictable for testing
return urllib.quote_plus(json.dumps({
name: (base64.b64encode(value).decode() if name == 'iv' else value)
for name, value in crypto_meta.items()}))
name: (base64.b64encode(value).decode() if name in ('iv', 'key')
else value)
for name, value in crypto_meta.items()}, sort_keys=True))
def load_crypto_meta(value):
@ -98,9 +100,10 @@ def load_crypto_meta(value):
Build the crypto_meta from the json object.
Note that json.loads always produces unicode strings, to ensure the
resultant crypto_meta matches the original object cast all key and
value data (other then the iv) to a str. This will work in py3 as well
where all strings are unicode implying the cast is effectively a no-op.
resultant crypto_meta matches the original object cast all key and value
data to a str except the key and iv which are base64 decoded. This will
work in py3 as well where all strings are unicode implying the cast is
effectively a no-op.
:param value: a string serialization of a crypto meta dict
:returns: a dict containing crypto meta items
@ -110,7 +113,7 @@ def load_crypto_meta(value):
try:
value = urllib.unquote_plus(value)
crypto_meta = {str(name): (base64.b64decode(value)
if name == 'iv' else str(value))
if name in ('iv', 'key') else str(value))
for name, value in json.loads(value).items()}
return crypto_meta
except (KeyError, ValueError, TypeError) as err:

31
swift/common/middleware/crypto.py
View File

@ -23,6 +23,9 @@ from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from swift.common.exceptions import EncryptionException
from swift.common.utils import get_logger
# AES will accept several key sizes - we are using 256 bits i.e. 32 bytes
KEY_LENGTH = 32
class Crypto(object):
"""
@ -32,10 +35,6 @@ class Crypto(object):
conf = {} if conf is None else conf
self.logger = get_logger(conf, log_route="crypto")
def check_key(self, key):
if len(key) != 32:
raise ValueError("Key must be length 32 bytes")
def create_encryption_ctxt(self, key, iv):
"""
Creates a crypto context for encrypting
@ -128,6 +127,30 @@ class Crypto(object):
raise EncryptionException(
'Bad crypto meta: Missing %s' % err)
def create_random_key(self):
# helper method to create random key of correct length
return os.urandom(KEY_LENGTH)
def wrap_key(self, wrapping_key, key_to_wrap, iv):
# we don't use an RFC 3394 key wrap algorithm such as cryptography's
# aes_wrap_key because it's slower and we have iv material readily
# available so don't need a deterministic algorithm
encryptor = Cipher(algorithms.AES(wrapping_key), modes.CTR(iv),
backend=default_backend()).encryptor()
return encryptor.update(key_to_wrap)
def unwrap_key(self, wrapping_key, wrapped_key, iv):
# check the key length early - unwrapping won't change the length
self.check_key(wrapped_key)
decryptor = Cipher(algorithms.AES(wrapping_key), modes.CTR(iv),
backend=default_backend()).decryptor()
key = decryptor.update(wrapped_key)
return key
def check_key(self, key):
if len(key) != KEY_LENGTH:
raise ValueError("Key must be length %s bytes" % KEY_LENGTH)
class CryptoContext(object):
"""

39
swift/common/middleware/decrypter.py
View File

@ -57,6 +57,29 @@ class BaseDecrypterContext(CryptoWSGIContext):
self.crypto.check_crypto_meta(crypto_meta)
return crypto_meta
def get_unwrapped_key(self, crypto_meta, wrapping_key):
"""
Get a wrapped key from crypto-meta and unwrap it using the provided
wrapping key.
:param crypto_meta: a dict of crypto-meta
:param wrapping_key: key to be used to decrypt the wrapped key
:return: an unwrapped key
:raises EncryptionException: if the crypto-meta has no wrapped key or
the unwrapped key is invalid
"""
try:
return self.crypto.unwrap_key(wrapping_key,
crypto_meta['key'],
crypto_meta['iv'])
except KeyError as err:
err = 'Missing %s' % err
except ValueError as err:
pass
msg = 'Error decrypting %s' % self.server_type
self.logger.error('%s: %s', msg, err)
raise HTTPInternalServerError(body=msg, content_type='text/plain')
def decrypt_value_with_meta(self, value, key):
"""
Decrypt a value if suitable crypto_meta can be extracted from the
@ -190,7 +213,7 @@ class DecrypterObjContext(BaseDecrypterContext):
return mod_resp_headers
def multipart_response_iter(self, resp, boundary, keys, crypto_meta):
def multipart_response_iter(self, resp, boundary, body_key, crypto_meta):
"""
Decrypts a multipart mime doc response body.
@ -212,7 +235,7 @@ class DecrypterObjContext(BaseDecrypterContext):
yield "\r\n"
decrypt_ctxt = self.crypto.create_decryption_ctxt(
keys['object'], crypto_meta['iv'], first_byte)
body_key, crypto_meta['iv'], first_byte)
for chunk in iter(lambda: body.read(DECRYPT_CHUNK_SIZE), ''):
chunk = decrypt_ctxt.update(chunk)
yield chunk
@ -221,7 +244,7 @@ class DecrypterObjContext(BaseDecrypterContext):
yield "--" + boundary + "--"
def response_iter(self, resp, keys, crypto_meta, offset):
def response_iter(self, resp, body_key, crypto_meta, offset):
"""
Decrypts a response body.
@ -232,7 +255,7 @@ class DecrypterObjContext(BaseDecrypterContext):
:return: generator for decrypted response body
"""
decrypt_ctxt = self.crypto.create_decryption_ctxt(
keys['object'], crypto_meta['iv'], offset)
body_key, crypto_meta['iv'], offset)
with closing_if_possible(resp):
for chunk in resp:
yield decrypt_ctxt.update(chunk)
@ -259,7 +282,7 @@ class DecrypterObjContext(BaseDecrypterContext):
'X-Object-Sysmeta-Crypto-Meta')
except EncryptionException as err:
msg = 'Error decrypting object'
self.logger.error('%s: %s', msg, err)
self.logger.error('%s: %s', msg, err)
raise HTTPInternalServerError(
body=msg, content_type='text/plain')
@ -268,17 +291,19 @@ class DecrypterObjContext(BaseDecrypterContext):
resp_iter = app_resp
elif (self._get_status_int() == 206 and
content_type == 'multipart/byteranges'):
body_key = self.get_unwrapped_key(crypto_meta, keys['object'])
boundary = dict(content_type_attrs)["boundary"]
resp_iter = self.multipart_response_iter(
app_resp, boundary, keys, crypto_meta)
app_resp, boundary, body_key, crypto_meta)
else:
body_key = self.get_unwrapped_key(crypto_meta, keys['object'])
offset = 0
content_range = self._response_header_value('Content-Range')
if content_range:
# Determine offset within the whole object if ranged GET
offset, end, total = parse_content_range(content_range)
resp_iter = self.response_iter(
app_resp, keys, crypto_meta, offset)
app_resp, body_key, crypto_meta, offset)
else:
# don't decrypt body of non-2xx responses
resp_iter = app_resp

11
swift/common/middleware/encrypter.py
View File

@ -55,7 +55,6 @@ class EncInputWrapper(object):
self.wsgi_input = req.environ['wsgi.input']
self.path = req.path
self.crypto = crypto
self.body_crypto_meta = self.crypto.create_crypto_meta()
self.body_crypto_ctxt = None
self.keys = keys
# remove any Etag from headers, it won't be valid for ciphertext and
@ -69,8 +68,16 @@ class EncInputWrapper(object):
def _init_encryption_context(self):
# do this once when body is first read
if self.body_crypto_ctxt is None:
self.body_crypto_meta = self.crypto.create_crypto_meta()
self.body_key = self.crypto.create_random_key()
# wrap the body key with object key re-using body iv
self.body_crypto_meta['key'] = self.crypto.wrap_key(
self.keys['object'],
self.body_key,
self.body_crypto_meta['iv']
)
self.body_crypto_ctxt = self.crypto.create_encryption_ctxt(
self.keys['object'], self.body_crypto_meta.get('iv'))
self.body_key, self.body_crypto_meta.get('iv'))
self.plaintext_md5 = md5()
self.ciphertext_md5 = md5()

6
test/unit/common/middleware/crypto_helpers.py
View File

@ -42,5 +42,7 @@ def decrypt(key, iv, enc_val):
FAKE_IV = "This is an IV123"
def fake_get_crypto_meta():
return {'iv': FAKE_IV, 'cipher': Crypto({}).get_cipher()}
def fake_get_crypto_meta(**kwargs):
meta = {'iv': FAKE_IV, 'cipher': Crypto({}).get_cipher()}
meta.update(kwargs)
return meta

34
test/unit/common/middleware/test_crypto.py
View File

@ -181,6 +181,40 @@ class TestCrypto(unittest.TestCase):
meta2 = self.crypto.create_crypto_meta()
self.assertNotEqual(meta['iv'], meta2['iv']) # crude sanity check
def test_create_random_key(self):
# crude check that we get unique keys on each call
keys = set()
for i in range(10):
key = self.crypto.create_random_key()
self.assertEqual(32, len(key))
keys.add(key)
self.assertEqual(10, len(keys))
def test_wrap_unwrap_key(self):
wrapping_key = os.urandom(32)
key_to_wrap = os.urandom(32)
iv = os.urandom(16)
wrapped = self.crypto.wrap_key(wrapping_key, key_to_wrap, iv)
cipher = Cipher(algorithms.AES(wrapping_key), modes.CTR(iv),
backend=default_backend())
expected = cipher.encryptor().update(key_to_wrap)
self.assertEqual(expected, wrapped)
unwrapped = self.crypto.unwrap_key(wrapping_key, wrapped, iv)
self.assertEqual(key_to_wrap, unwrapped)
def test_unwrap_bad_key(self):
# verify that ValueError is raised if unwrapped key is invalid
wrapping_key = os.urandom(32)
iv = os.urandom(16)
for length in (0, 16, 24, 31, 33):
key_to_wrap = os.urandom(length)
wrapped = self.crypto.wrap_key(wrapping_key, key_to_wrap, iv)
with self.assertRaises(ValueError) as cm:
self.crypto.unwrap_key(wrapping_key, wrapped, iv)
self.assertEqual(
cm.exception.message, 'Key must be length 32 bytes')
if __name__ == '__main__':
unittest.main()

71
test/unit/common/middleware/test_crypto_utils.py
View File

@ -144,32 +144,71 @@ class TestCryptoWsgiContext(unittest.TestCase):
class TestModuleMethods(unittest.TestCase):
meta = {'iv': '0123456789abcdef', 'cipher': 'AES_CTR_256'}
serialized_meta = '%7B%22cipher%22%3A+%22AES_CTR_256%22%2C+%22' \
'iv%22%3A+%22MDEyMzQ1Njc4OWFiY2RlZg%3D%3D%22%7D'
meta_with_key = {'iv': '0123456789abcdef', 'cipher': 'AES_CTR_256',
'key': '0123456789abcdef0123456789abcdef'}
serialized_meta_with_key = '%7B%22cipher%22%3A+%22AES_CTR_256%22%2C+%22' \
'iv%22%3A+%22MDEyMzQ1Njc4OWFiY2RlZg%3D%3D%22' \
'%2C+%22key%22%3A+%22MDEyMzQ1Njc4OWFiY2RlZjA' \
'xMjM0NTY3ODlhYmNkZWY%3D%22%7D'
def test_dump_crypto_meta(self):
actual = crypto_utils.dump_crypto_meta(self.meta)
self.assertEqual(self.serialized_meta, actual)
actual = crypto_utils.dump_crypto_meta(self.meta_with_key)
self.assertEqual(self.serialized_meta_with_key, actual)
def test_load_crypto_meta(self):
actual = crypto_utils.load_crypto_meta(self.serialized_meta)
self.assertEqual(self.meta, actual)
actual = crypto_utils.load_crypto_meta(self.serialized_meta_with_key)
self.assertEqual(self.meta_with_key, actual)
def test_dump_then_load_crypto_meta(self):
actual = crypto_utils.load_crypto_meta(
crypto_utils.dump_crypto_meta(self.meta))
self.assertEqual(self.meta, actual)
actual = crypto_utils.load_crypto_meta(
crypto_utils.dump_crypto_meta(self.meta_with_key))
self.assertEqual(self.meta_with_key, actual)
def test_append_crypto_meta(self):
actual = crypto_utils.append_crypto_meta(
'abc', {'iv': '0123456789abcdef', 'cipher': 'AES_CTR_256'})
# the order of the dict serialization is unpredictable
expected = [
'abc; meta=%7B%22cipher%22%3A+%22AES_CTR_256%22%2C+%22'
'iv%22%3A+%22MDEyMzQ1Njc4OWFiY2RlZg%3D%3D%22%7D',
'abc; meta=%7B%22iv%22%3A+%22MDEyMzQ1Njc4OWFiY2RlZg%3D%3D%22%2C'
'+%22cipher%22%3A+%22AES_CTR_256%22%7D']
self.assertIn(actual, expected)
actual = crypto_utils.append_crypto_meta('abc', self.meta)
expected = 'abc; meta=%s' % self.serialized_meta
self.assertEqual(actual, expected)
actual = crypto_utils.append_crypto_meta('abc', self.meta_with_key)
expected = 'abc; meta=%s' % self.serialized_meta_with_key
self.assertEqual(actual, expected)
def test_extract_crypto_meta(self):
val, meta = crypto_utils.extract_crypto_meta(
'abc; meta=%7B%22cipher%22%3A+%22AES_CTR_256%22%2C+%22'
'iv%22%3A+%22MDEyMzQ1Njc4OWFiY2RlZg%3D%3D%22%7D')
'abc; meta=%s' % self.serialized_meta)
self.assertEqual('abc', val)
self.assertDictEqual(
{'iv': '0123456789abcdef', 'cipher': 'AES_CTR_256'}, meta)
self.assertDictEqual(self.meta, meta)
val, meta = crypto_utils.extract_crypto_meta(
'abc; meta=%s' % self.serialized_meta_with_key)
self.assertEqual('abc', val)
self.assertDictEqual(self.meta_with_key, meta)
val, meta = crypto_utils.extract_crypto_meta('abc')
self.assertEqual('abc', val)
self.assertIsNone(meta)
# other param names will be ignored
val, meta = crypto_utils.extract_crypto_meta('abc; foo=bar')
self.assertEqual('abc', val)
self.assertIsNone(meta)
def test_append_then_extract_crypto_meta(self):
val = 'abc'
meta = {'iv': '0123456789abcdef', 'cipher': 'AES_CTR_256'}
actual = crypto_utils.extract_crypto_meta(
crypto_utils.append_crypto_meta(val, meta))
self.assertEqual((val, meta), actual)
crypto_utils.append_crypto_meta(val, self.meta))
self.assertEqual((val, self.meta), actual)

210
test/unit/common/middleware/test_decrypter.py
View File

@ -12,16 +12,15 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import unittest
from xml.dom import minidom
import mock
import base64
import json
import urllib
from swift.common.middleware.crypto import Crypto
from swift.common.crypto_utils import CRYPTO_KEY_CALLBACK
from swift.common.crypto_utils import CRYPTO_KEY_CALLBACK, dump_crypto_meta
from swift.common.middleware import decrypter
from swift.common.swob import Request, HTTPException, HTTPOk, \
HTTPPreconditionFailed, HTTPNotFound, HTTPPartialContent
@ -35,10 +34,7 @@ from test.unit.common.middleware.helpers import FakeSwift, FakeAppThatExcepts
def get_crypto_meta_header(crypto_meta=None):
if crypto_meta is None:
crypto_meta = fake_get_crypto_meta()
return urllib.quote_plus(
json.dumps({key: (base64.b64encode(value).decode()
if key == 'iv' else value)
for key, value in crypto_meta.items()}))
return dump_crypto_meta(crypto_meta)
def encrypt_and_append_meta(value, key, crypto_meta=None):
@ -60,26 +56,32 @@ class TestDecrypterObjectRequests(unittest.TestCase):
CRYPTO_KEY_CALLBACK: fetch_crypto_keys}
req = Request.blank('/v1/a/c/o', environ=env)
body = 'FAKE APP'
key = fetch_crypto_keys()['object']
enc_body = encrypt(body, key, FAKE_IV)
hdrs = {'Etag': 'hashOfCiphertext',
'content-type': 'text/plain',
'content-length': len(enc_body),
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(md5hex(body), key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta': get_crypto_meta_header(),
'x-object-meta-test':
base64.b64encode(encrypt('encrypt me', key, FAKE_IV)),
'x-object-transient-sysmeta-crypto-meta-test':
get_crypto_meta_header(),
'x-object-sysmeta-test': 'do not encrypt me'}
plaintext_etag = md5hex(body)
object_key = fetch_crypto_keys()['object']
body_key = os.urandom(32)
enc_body = encrypt(body, body_key, FAKE_IV)
body_crypto_meta = fake_get_crypto_meta(
key=encrypt(body_key, object_key, FAKE_IV))
hdrs = {
'Etag': 'hashOfCiphertext',
'content-type': 'text/plain',
'content-length': len(enc_body),
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(plaintext_etag, object_key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta':
get_crypto_meta_header(body_crypto_meta),
'x-object-meta-test':
base64.b64encode(encrypt('encrypt me', object_key, FAKE_IV)),
'x-object-transient-sysmeta-crypto-meta-test':
get_crypto_meta_header(),
'x-object-sysmeta-test': 'do not encrypt me'}
self.app.register(
'GET', '/v1/a/c/o', HTTPOk, body=enc_body, headers=hdrs)
resp = req.get_response(self.decrypter)
self.assertEqual(body, resp.body)
self.assertEqual('200 OK', resp.status)
self.assertEqual(md5hex(body), resp.headers['Etag'])
self.assertEqual(plaintext_etag, resp.headers['Etag'])
self.assertEqual('text/plain', resp.headers['Content-Type'])
self.assertEqual('encrypt me', resp.headers['x-object-meta-test'])
self.assertEqual('do not encrypt me',
@ -266,7 +268,7 @@ class TestDecrypterObjectRequests(unittest.TestCase):
self.assertIn(
'iv', self.decrypter.logger.get_lines_for_level('error')[0])
def _test_GET_with_bad_iv_for_object_body(self, bad_crypto_meta):
def _test_GET_with_bad_crypto_meta_for_object_body(self, bad_crypto_meta):
# use bad iv for object body
env = {'REQUEST_METHOD': 'GET',
CRYPTO_KEY_CALLBACK: fetch_crypto_keys}
@ -290,18 +292,30 @@ class TestDecrypterObjectRequests(unittest.TestCase):
self.decrypter.logger.get_lines_for_level('error')[0])
def test_GET_with_bad_iv_for_object_body(self):
bad_crypto_meta = fake_get_crypto_meta()
bad_crypto_meta = fake_get_crypto_meta(key=os.urandom(32))
bad_crypto_meta['iv'] = 'bad_iv'
self._test_GET_with_bad_iv_for_object_body(bad_crypto_meta)
self._test_GET_with_bad_crypto_meta_for_object_body(bad_crypto_meta)
self.assertIn('IV must be length 16',
self.decrypter.logger.get_lines_for_level('error')[0])
def test_GET_with_missing_iv_for_object_body(self):
bad_crypto_meta = fake_get_crypto_meta()
bad_crypto_meta = fake_get_crypto_meta(key=os.urandom(32))
bad_crypto_meta.pop('iv')
self._test_GET_with_bad_iv_for_object_body(bad_crypto_meta)
self.assertIn(
'iv', self.decrypter.logger.get_lines_for_level('error')[0])
self._test_GET_with_bad_crypto_meta_for_object_body(bad_crypto_meta)
self.assertIn("Missing 'iv'",
self.decrypter.logger.get_lines_for_level('error')[0])
def test_GET_with_bad_body_key_for_object_body(self):
bad_crypto_meta = fake_get_crypto_meta(key='wrapped too short key')
self._test_GET_with_bad_crypto_meta_for_object_body(bad_crypto_meta)
self.assertIn('Key must be length 32',
self.decrypter.logger.get_lines_for_level('error')[0])
def test_GET_with_missing_body_key_for_object_body(self):
bad_crypto_meta = fake_get_crypto_meta() # no key by default
self._test_GET_with_bad_crypto_meta_for_object_body(bad_crypto_meta)
self.assertIn("Missing 'key'",
self.decrypter.logger.get_lines_for_level('error')[0])
def test_HEAD_success(self):
env = {'REQUEST_METHOD': 'HEAD',
@ -341,21 +355,27 @@ class TestDecrypterObjectRequests(unittest.TestCase):
CRYPTO_KEY_CALLBACK: fetch_crypto_keys}
req = Request.blank('/v1/a/c/o', environ=env)
body = 'FAKE APP'
key = fetch_crypto_keys()['object']
enc_body = encrypt(body, key, FAKE_IV)
hdrs = {'Etag': 'hashOfCiphertext',
'etag': 'hashOfCiphertext',
'content-type': 'text/plain',
'content-length': len(enc_body),
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(md5hex(body), key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta': get_crypto_meta_header()}
plaintext_etag = md5hex(body)
object_key = fetch_crypto_keys()['object']
body_key = os.urandom(32)
enc_body = encrypt(body, body_key, FAKE_IV)
body_crypto_meta = fake_get_crypto_meta(
key=encrypt(body_key, object_key, FAKE_IV))
hdrs = {
'Etag': 'hashOfCiphertext',
'etag': 'hashOfCiphertext',
'content-type': 'text/plain',
'content-length': len(enc_body),
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(plaintext_etag, object_key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta':
get_crypto_meta_header(body_crypto_meta)}
self.app.register(
method, '/v1/a/c/o', HTTPOk, body=enc_body, headers=hdrs)
resp = req.get_response(self.decrypter)
self.assertEqual('200 OK', resp.status)
self.assertEqual(md5hex(body), resp.headers['Etag'])
self.assertEqual(plaintext_etag, resp.headers['Etag'])
self.assertEqual('text/plain', resp.headers['Content-Type'])
def test_HEAD_content_type_not_encrypted(self):
@ -371,22 +391,28 @@ class TestDecrypterObjectRequests(unittest.TestCase):
CRYPTO_KEY_CALLBACK: fetch_crypto_keys}
req = Request.blank('/v1/a/c/o', environ=env)
body = 'FAKE APP'
key = fetch_crypto_keys()['object']
enc_body = encrypt(body, key, FAKE_IV)
hdrs = {'Etag': 'hashOfCiphertext',
'etag': 'hashOfCiphertext',
'content-type': 'text/plain',
'content-length': len(enc_body),
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(md5hex(body), key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta': get_crypto_meta_header(),
'x-object-meta-test': 'plaintext'}
plaintext_etag = md5hex(body)
object_key = fetch_crypto_keys()['object']
body_key = os.urandom(32)
enc_body = encrypt(body, body_key, FAKE_IV)
body_crypto_meta = fake_get_crypto_meta(
key=encrypt(body_key, object_key, FAKE_IV))
hdrs = {
'Etag': 'hashOfCiphertext',
'etag': 'hashOfCiphertext',
'content-type': 'text/plain',
'content-length': len(enc_body),
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(plaintext_etag, object_key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta':
get_crypto_meta_header(body_crypto_meta),
'x-object-meta-test': 'plaintext'}
self.app.register(
method, '/v1/a/c/o', HTTPOk, body=enc_body, headers=hdrs)
resp = req.get_response(self.decrypter)
self.assertEqual('200 OK', resp.status)
self.assertEqual(md5hex(body), resp.headers['Etag'])
self.assertEqual(plaintext_etag, resp.headers['Etag'])
self.assertEqual('text/plain', resp.headers['Content-Type'])
self.assertEqual('plaintext', resp.headers['x-object-meta-test'])
@ -431,22 +457,28 @@ class TestDecrypterObjectRequests(unittest.TestCase):
req = Request.blank('/v1/a/c/o', environ=env)
chunks = ['some', 'chunks', 'of data']
body = ''.join(chunks)
key = fetch_crypto_keys()['object']
ctxt = Crypto().create_encryption_ctxt(key, FAKE_IV)
plaintext_etag = md5hex(body)
object_key = fetch_crypto_keys()['object']
body_key = os.urandom(32)
body_crypto_meta = fake_get_crypto_meta(
key=encrypt(body_key, object_key, FAKE_IV))
ctxt = Crypto().create_encryption_ctxt(body_key, FAKE_IV)
enc_body = [encrypt(chunk, ctxt=ctxt) for chunk in chunks]
hdrs = {'Etag': 'hashOfCiphertext',
'content-type': 'text/plain',
'content-length': sum(map(len, enc_body)),
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(md5hex(body), key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta': get_crypto_meta_header()}
hdrs = {
'Etag': 'hashOfCiphertext',
'content-type': 'text/plain',
'content-length': sum(map(len, enc_body)),
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(plaintext_etag, object_key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta':
get_crypto_meta_header(body_crypto_meta)}
self.app.register(
'GET', '/v1/a/c/o', HTTPOk, body=enc_body, headers=hdrs)
resp = req.get_response(self.decrypter)
self.assertEqual(body, resp.body)
self.assertEqual('200 OK', resp.status)
self.assertEqual(md5hex(body), resp.headers['Etag'])
self.assertEqual(plaintext_etag, resp.headers['Etag'])
self.assertEqual('text/plain', resp.headers['Content-Type'])
def test_GET_multiseg_with_range(self):
@ -456,18 +488,24 @@ class TestDecrypterObjectRequests(unittest.TestCase):
req.headers['Content-Range'] = 'bytes 3-10/17'
chunks = ['0123', '45678', '9abcdef']
body = ''.join(chunks)
key = fetch_crypto_keys()['object']
ctxt = Crypto().create_encryption_ctxt(key, FAKE_IV)
plaintext_etag = md5hex(body)
object_key = fetch_crypto_keys()['object']
body_key = os.urandom(32)
body_crypto_meta = fake_get_crypto_meta(
key=encrypt(body_key, object_key, FAKE_IV))
ctxt = Crypto().create_encryption_ctxt(body_key, FAKE_IV)
enc_body = [encrypt(chunk, ctxt=ctxt) for chunk in chunks]
enc_body = [enc_body[0][3:], enc_body[1], enc_body[2][:2]]
hdrs = {'Etag': 'hashOfCiphertext',
'content-type': 'text/plain',
'content-length': sum(map(len, enc_body)),
'content-range': req.headers['Content-Range'],
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(md5hex(body), key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta': get_crypto_meta_header()}
hdrs = {
'Etag': 'hashOfCiphertext',
'content-type': 'text/plain',
'content-length': sum(map(len, enc_body)),
'content-range': req.headers['Content-Range'],
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(plaintext_etag, object_key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta':
get_crypto_meta_header(body_crypto_meta)}
self.app.register(
'GET', '/v1/a/c/o', HTTPOk, body=enc_body, headers=hdrs)
resp = req.get_response(self.decrypter)
@ -475,7 +513,7 @@ class TestDecrypterObjectRequests(unittest.TestCase):
self.assertEqual('200 OK', resp.status)
# TODO - how do we validate the range body if etag is for whole? Is
# the test actually faking the correct Etag in response?
self.assertEqual(md5hex(body), resp.headers['Etag'])
self.assertEqual(plaintext_etag, resp.headers['Etag'])
self.assertEqual('text/plain', resp.headers['Content-Type'])
# Force the decrypter context updates to be less than one of our range
@ -486,11 +524,13 @@ class TestDecrypterObjectRequests(unittest.TestCase):
@mock.patch.object(decrypter, 'DECRYPT_CHUNK_SIZE', 4)
def test_GET_multipart_ciphertext(self):
# build fake multipart response body
key = fetch_crypto_keys()['object']
ctxt = Crypto().create_encryption_ctxt(key, FAKE_IV)
object_key = fetch_crypto_keys()['object']
body_key = os.urandom(32)
body_crypto_meta = fake_get_crypto_meta(
key=encrypt(body_key, object_key, FAKE_IV))
plaintext = 'Cwm fjord veg balks nth pyx quiz'
plaintext_etag = md5hex(plaintext)
ciphertext = encrypt(plaintext, ctxt=ctxt)
ciphertext = encrypt(plaintext, body_key, FAKE_IV)
parts = ((0, 3, 'text/plain'),
(4, 9, 'text/plain; charset=us-ascii'),
(24, 32, 'text/plain'))
@ -509,9 +549,10 @@ class TestDecrypterObjectRequests(unittest.TestCase):
'content-type': 'multipart/byteranges;boundary=multipartboundary',
'content-length': len(body),
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(plaintext_etag, key, FAKE_IV)),
base64.b64encode(encrypt(plaintext_etag, object_key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta': get_crypto_meta_header()}
'X-Object-Sysmeta-Crypto-Meta':
get_crypto_meta_header(body_crypto_meta)}
self.app.register('GET', '/v1/a/c/o', HTTPPartialContent, body=body,
headers=hdrs)
@ -548,11 +589,13 @@ class TestDecrypterObjectRequests(unittest.TestCase):
def test_GET_multipart_content_type(self):
# *just* having multipart content type shouldn't trigger the mime doc
# code path
key = fetch_crypto_keys()['object']
ctxt = Crypto({}).create_encryption_ctxt(key, FAKE_IV)
object_key = fetch_crypto_keys()['object']
body_key = os.urandom(32)
body_crypto_meta = fake_get_crypto_meta(
key=encrypt(body_key, object_key, FAKE_IV))
plaintext = 'Cwm fjord veg balks nth pyx quiz'
plaintext_etag = md5hex(plaintext)
ciphertext = encrypt(plaintext, ctxt=ctxt)
ciphertext = encrypt(plaintext, body_key, FAKE_IV)
# register request with fake swift
hdrs = {
@ -560,9 +603,10 @@ class TestDecrypterObjectRequests(unittest.TestCase):
'content-type': 'multipart/byteranges;boundary=multipartboundary',
'content-length': len(ciphertext),
'X-Object-Sysmeta-Crypto-Etag':
base64.b64encode(encrypt(plaintext_etag, key, FAKE_IV)),
base64.b64encode(encrypt(plaintext_etag, object_key, FAKE_IV)),
'X-Object-Sysmeta-Crypto-Meta-Etag': get_crypto_meta_header(),
'X-Object-Sysmeta-Crypto-Meta': get_crypto_meta_header()}
'X-Object-Sysmeta-Crypto-Meta':
get_crypto_meta_header(body_crypto_meta)}
self.app.register('GET', '/v1/a/c/o', HTTPOk, body=ciphertext,
headers=hdrs)

71
test/unit/common/middleware/test_encrypter.py
View File

@ -14,6 +14,7 @@
# limitations under the License.
import base64
import json
import os
import urllib
import unittest
@ -41,10 +42,11 @@ class TestEncrypter(unittest.TestCase):
self.encrypter.logger = FakeLogger()
def test_PUT_req(self):
key = fetch_crypto_keys()['object']
body_key = os.urandom(32)
object_key = fetch_crypto_keys()['object']
plaintext = 'FAKE APP'
plaintext_etag = md5hex(plaintext)
ciphertext = encrypt(plaintext, key, FAKE_IV)
ciphertext = encrypt(plaintext, body_key, FAKE_IV)
ciphertext_etag = md5hex(ciphertext)
env = {'REQUEST_METHOD': 'PUT',
@ -57,7 +59,10 @@ class TestEncrypter(unittest.TestCase):
req = Request.blank(
'/v1/a/c/o', environ=env, body=plaintext, headers=hdrs)
self.app.register('PUT', '/v1/a/c/o', HTTPCreated, {})
resp = req.get_response(self.encrypter)
with mock.patch(
'swift.common.middleware.crypto.Crypto.create_random_key',
return_value=body_key):
resp = req.get_response(self.encrypter)
self.assertEqual('201 Created', resp.status)
self.assertEqual(plaintext_etag, resp.headers['Etag'])
@ -66,12 +71,21 @@ class TestEncrypter(unittest.TestCase):
self.assertEqual('PUT', self.app.calls[0][0])
req_hdrs = self.app.headers[0]
# verify body crypto meta
actual = req_hdrs['X-Object-Sysmeta-Crypto-Meta']
actual = json.loads(urllib.unquote_plus(actual))
expected_wrapped_key = encrypt(body_key, object_key, FAKE_IV)
self.assertEqual(Crypto().get_cipher(), actual['cipher'])
self.assertEqual(FAKE_IV, base64.b64decode(actual['iv']))
self.assertEqual(expected_wrapped_key, base64.b64decode(actual['key']))
# verify etag
self.assertEqual(ciphertext_etag, req_hdrs['Etag'])
# verify encrypted version of plaintext etag
actual = base64.b64decode(req_hdrs['X-Object-Sysmeta-Crypto-Etag'])
etag_iv = Crypto().create_iv(iv_base=req.path)
enc_etag = encrypt(plaintext_etag, key, etag_iv)
enc_etag = encrypt(plaintext_etag, object_key, etag_iv)
self.assertEqual(enc_etag, actual)
# verify crypto_meta was not appended to this etag
parts = req_hdrs['X-Object-Sysmeta-Crypto-Etag'].rsplit(';', 1)
@ -103,15 +117,16 @@ class TestEncrypter(unittest.TestCase):
self.assertEqual('text/plain', req_hdrs['Content-Type'])
# user meta is encrypted
self.assertEqual(base64.b64encode(encrypt('encrypt me', key, FAKE_IV)),
req_hdrs['X-Object-Meta-Test'])
self.assertEqual(
base64.b64encode(encrypt('encrypt me', object_key, FAKE_IV)),
req_hdrs['X-Object-Meta-Test'])
actual = req_hdrs['X-Object-Transient-Sysmeta-Crypto-Meta-Test']
actual = json.loads(urllib.unquote_plus(actual))
self.assertEqual(Crypto().get_cipher(), actual['cipher'])
self.assertEqual(FAKE_IV, base64.b64decode(actual['iv']))
self.assertEqual(
base64.b64encode(encrypt('not to be confused with the Etag!',
key, FAKE_IV)),
object_key, FAKE_IV)),
req_hdrs['X-Object-Meta-Etag'])
actual = req_hdrs['X-Object-Transient-Sysmeta-Crypto-Meta-Etag']
actual = json.loads(urllib.unquote_plus(actual))
@ -130,10 +145,11 @@ class TestEncrypter(unittest.TestCase):
def test_PUT_with_other_footers(self):
# verify handling of another middleware's footer callback
key = fetch_crypto_keys()['object']
body_key = os.urandom(32)
object_key = fetch_crypto_keys()['object']
plaintext = 'FAKE APP'
plaintext_etag = md5hex(plaintext)
ciphertext = encrypt(plaintext, key, FAKE_IV)
ciphertext = encrypt(plaintext, body_key, FAKE_IV)
ciphertext_etag = md5hex(ciphertext)
other_footers = {
'Etag': plaintext_etag,
@ -151,7 +167,12 @@ class TestEncrypter(unittest.TestCase):
req = Request.blank(
'/v1/a/c/o', environ=env, body=plaintext, headers=hdrs)
self.app.register('PUT', '/v1/a/c/o', HTTPCreated, {})
resp = req.get_response(self.encrypter)
with mock.patch(
'swift.common.middleware.crypto.Crypto.create_random_key',
lambda *args: body_key):
resp = req.get_response(self.encrypter)
self.assertEqual('201 Created', resp.status)
self.assertEqual(plaintext_etag, resp.headers['Etag'])
@ -170,12 +191,20 @@ class TestEncrypter(unittest.TestCase):
self.assertEqual(ciphertext_etag, req_hdrs['Etag'])
actual = base64.b64decode(req_hdrs['X-Object-Sysmeta-Crypto-Etag'])
etag_iv = Crypto().create_iv(iv_base=req.path)
self.assertEqual(encrypt(plaintext_etag, key, etag_iv), actual)
self.assertEqual(encrypt(plaintext_etag, object_key, etag_iv), actual)
actual = json.loads(urllib.unquote_plus(
req_hdrs['X-Object-Sysmeta-Crypto-Meta-Etag']))
self.assertEqual(Crypto().get_cipher(), actual['cipher'])
self.assertEqual(etag_iv, base64.b64decode(actual['iv']))
# verify body crypto meta
actual = req_hdrs['X-Object-Sysmeta-Crypto-Meta']
actual = json.loads(urllib.unquote_plus(actual))
expected_wrapped_key = encrypt(body_key, object_key, FAKE_IV)
self.assertEqual(Crypto().get_cipher(), actual['cipher'])
self.assertEqual(FAKE_IV, base64.b64decode(actual['iv']))
self.assertEqual(expected_wrapped_key, base64.b64decode(actual['key']))
def test_PUT_with_bad_etag_in_other_footers(self):
# verify that etag supplied in footers from other middleware overrides
# header etag when validating inbound plaintext etags
@ -414,6 +443,7 @@ class TestEncrypter(unittest.TestCase):
self.assertEqual(md5hex(body), resp.headers['Etag'])
def test_PUT_multiseg_no_client_etag(self):
body_key = os.urandom(32)
chunks = ['some', 'chunks', 'of data']
body = ''.join(chunks)
env = {'REQUEST_METHOD': 'PUT',
@ -424,14 +454,20 @@ class TestEncrypter(unittest.TestCase):
req = Request.blank(
'/v1/a/c/o', environ=env, headers=hdrs)
self.app.register('PUT', '/v1/a/c/o', HTTPCreated, {})
resp = req.get_response(self.encrypter)
with mock.patch(
'swift.common.middleware.crypto.Crypto.create_random_key',
lambda *args: body_key):
resp = req.get_response(self.encrypter)
self.assertEqual('201 Created', resp.status)
# verify object is encrypted by getting direct from the app
get_req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'GET'})
self.assertEqual(encrypt(body, fetch_crypto_keys()['object'], FAKE_IV),
self.assertEqual(encrypt(body, body_key, FAKE_IV),
get_req.get_response(self.app).body)
def test_PUT_multiseg_good_client_etag(self):
body_key = os.urandom(32)
chunks = ['some', 'chunks', 'of data']
body = ''.join(chunks)
env = {'REQUEST_METHOD': 'PUT',
@ -443,11 +479,16 @@ class TestEncrypter(unittest.TestCase):
req = Request.blank(
'/v1/a/c/o', environ=env, headers=hdrs)
self.app.register('PUT', '/v1/a/c/o', HTTPCreated, {})
resp = req.get_response(self.encrypter)
with mock.patch(
'swift.common.middleware.crypto.Crypto.create_random_key',
lambda *args: body_key):
resp = req.get_response(self.encrypter)
self.assertEqual('201 Created', resp.status)
# verify object is encrypted by getting direct from the app
get_req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'GET'})
self.assertEqual(encrypt(body, fetch_crypto_keys()['object'], FAKE_IV),
self.assertEqual(encrypt(body, body_key, FAKE_IV),
get_req.get_response(self.app).body)
def test_PUT_multiseg_bad_client_etag(self):

13
test/unit/common/middleware/test_encrypter_decrypter.py
View File

@ -14,6 +14,8 @@
# limitations under the License.
import base64
import json
import mock
import os
import unittest
import urllib
@ -35,7 +37,8 @@ class TestEncrypterDecrypter(unittest.TestCase):
"""
def test_basic_put_get_req(self):
# Setup pass the PUT request through the encrypter.
# pass the PUT request through the encrypter.
body_key = os.urandom(32)
body = 'FAKE APP'
env = {'REQUEST_METHOD': 'PUT',
CRYPTO_KEY_CALLBACK: fetch_crypto_keys}
@ -47,7 +50,11 @@ class TestEncrypterDecrypter(unittest.TestCase):
'/v1/a/c/o', environ=env, body=body, headers=hdrs)
app = FakeSwift()
app.register('PUT', '/v1/a/c/o', HTTPCreated, {})
req.get_response(encrypter.Encrypter(app, {}))
with mock.patch(
'swift.common.middleware.crypto.Crypto.create_random_key',
return_value=body_key):
req.get_response(encrypter.Encrypter(app, {}))
# Verify that at least the request body was indeed encrypted.
# Otherwise, checking that input matches output after decryption is
@ -61,7 +68,7 @@ class TestEncrypterDecrypter(unittest.TestCase):
encrypt_get_resp.headers['X-Object-Sysmeta-Crypto-Meta']))
crypto_meta['iv'] = base64.b64decode(crypto_meta['iv'])
exp_enc_body = encrypt(
body, fetch_crypto_keys()['object'], crypto_meta['iv'])
body, body_key, crypto_meta['iv'])
self.assertEqual(exp_enc_body, encrypt_get_resp.body)
self.assertNotEqual(body, encrypt_get_resp.body) # sanity check