s3api: Allow some forms of server-side-encryption

...if and only if encryption is enabled. A few things to note about server-side
encryption:

- We register whether encryption is present and enabled when the proxy server
  starts up.
- This is generally considered an operator feature, not a user-facing one. S3
  API users can now learn more about how your cluster is set up than they
  previously could.
- If encryption is enabled but there are no keymasters in the pipeline, all
  writes will fail with "Unable to retrieve encryption keys."
- There's still a 'swift.crypto.override' env key that keymasters can set to
  skip encryption, so this isn't a full guarantee that things will be
  encrypted. On the other hand, none of the keymasters in Swift ever set that
  override.

Note that this *does not* start including x-amz-server-side-encryption
headers in the response, neither during PUT nor GET. We should only
send that when we know for sure that the data on disk was encrypted.

Change-Id: I4c20bca7fedb839628f1b2f8611807631b8bf430
Related-Bug: 1607116
Related-Change: Icf28dc57e589f9be20937947095800d7ce57b2f7

swift/common/middleware/s3api/s3request.py

@@ -24,7 +24,7 @@ import six
 from six.moves.urllib.parse import quote, unquote, parse_qsl
 import string
 
-from swift.common.utils import split_path, json
+from swift.common.utils import split_path, json, get_swift_info
 from swift.common import swob
 from swift.common.http import HTTP_OK, HTTP_CREATED, HTTP_ACCEPTED, \
     HTTP_NO_CONTENT, HTTP_UNAUTHORIZED, HTTP_FORBIDDEN, HTTP_NOT_FOUND, \
@@ -741,8 +741,17 @@ class S3Request(swob.Request):
         if 'x-amz-mfa' in self.headers:
             raise S3NotImplemented('MFA Delete is not supported.')
 
-        if 'x-amz-server-side-encryption' in self.headers:
-            raise S3NotImplemented('Server-side encryption is not supported.')
+        sse_value = self.headers.get('x-amz-server-side-encryption')
+        if sse_value is not None:
+            if sse_value not in ('aws:kms', 'AES256'):
+                raise InvalidArgument(
+                    'x-amz-server-side-encryption', sse_value,
+                    'The encryption method specified is not supported')
+            encryption_enabled = get_swift_info(admin=True)['admin'].get(
+                'encryption', {}).get('enabled')
+            if not encryption_enabled or sse_value != 'AES256':
+                raise S3NotImplemented(
+                    'Server-side encryption is not supported.')
 
         if 'x-amz-website-redirect-location' in self.headers:
             raise S3NotImplemented('Website redirection is not supported.')

test/unit/common/middleware/s3api/test_s3api.py

@@ -543,6 +543,15 @@ class TestS3ApiMiddleware(S3ApiTestCase):
         status, headers, body = self.call_s3api(req)
         self.assertEqual(self._get_error_code(body), 'InvalidStorageClass')
 
+    def test_invalid_ssc(self):
+        req = Request.blank('/',
+                            environ={'REQUEST_METHOD': 'GET',
+                                     'HTTP_AUTHORIZATION': 'AWS X:Y:Z'},
+                            headers={'x-amz-server-side-encryption': 'invalid',
+                                     'Date': self.get_date_header()})
+        status, headers, body = self.call_s3api(req)
+        self.assertEqual(self._get_error_code(body), 'InvalidArgument')
+
     def _test_unsupported_header(self, header, value=None):
         if value is None:
             value = 'value'
@@ -557,8 +566,27 @@ class TestS3ApiMiddleware(S3ApiTestCase):
     def test_mfa(self):
         self._test_unsupported_header('x-amz-mfa')
 
-    def test_server_side_encryption(self):
-        self._test_unsupported_header('x-amz-server-side-encryption')
+    @mock.patch.object(utils, '_swift_admin_info', new_callable=dict)
+    def test_server_side_encryption(self, mock_info):
+        sse_header = 'x-amz-server-side-encryption'
+        self._test_unsupported_header(sse_header, 'AES256')
+        self._test_unsupported_header(sse_header, 'aws:kms')
+        utils.register_swift_info('encryption', admin=True, enabled=False)
+        self._test_unsupported_header(sse_header, 'AES256')
+        self._test_unsupported_header(sse_header, 'aws:kms')
+        utils.register_swift_info('encryption', admin=True, enabled=True)
+        # AES256 now works
+        self.swift.register('PUT', '/v1/AUTH_X/bucket/object',
+                            swob.HTTPCreated, {}, None)
+        req = Request.blank('/bucket/object',
+                            environ={'REQUEST_METHOD': 'PUT',
+                                     'HTTP_AUTHORIZATION': 'AWS X:Y:Z'},
+                            headers={sse_header: 'AES256',
+                                     'Date': self.get_date_header()})
+        status, headers, body = self.call_s3api(req)
+        self.assertEqual(status, '200 OK')
+        # ...but aws:kms continues to fail
+        self._test_unsupported_header(sse_header, 'aws:kms')
 
     def test_website_redirect_location(self):
         self._test_unsupported_header('x-amz-website-redirect-location')