s3api: Transfer REMOTE_USER when using s3_acl

Some middlewares (notably staticweb) use the absence of a REMOTE_USER to
determine that a request was unauthenticated and as a result should be
handled differently. This could cause problems for S3 requests that
were authenticated via s3api's custom auth logic, including

* server errors when a container listing request gets handled by
  staticweb or
* losing storage policy information because staticweb copied the request
  environment.

Change-Id: Idf29c6866fec7b413c4369dce13c4788666c0934
Closes-Bug: #1833287
Related-Change: I5fe5ab31d6b2d9f7b6ecb3bfa246433a78e54808

swift/common/middleware/s3api/s3request.py

@@ -1554,6 +1554,8 @@ class S3AclRequest(S3Request):
 
         sw_req.environ.get('swift.authorize', lambda req: None)(sw_req)
         self.environ['swift_owner'] = sw_req.environ.get('swift_owner', False)
+        if 'REMOTE_USER' in sw_req.environ:
+            self.environ['REMOTE_USER'] = sw_req.environ['REMOTE_USER']
 
         # Need to skip S3 authorization on subsequent requests to prevent
         # overwriting the account in PATH_INFO

test/functional/s3api/test_bucket.py

@@ -22,6 +22,7 @@ import test.functional as tf
 from swift.common.utils import config_true_value
 from test.functional.s3api import S3ApiBaseBoto3
 from test.functional.s3api.s3_test_client import get_boto3_conn
+from test.functional.swift_test_client import Connection
 
 
 def setUpModule():
@@ -120,6 +121,29 @@ class TestS3ApiBucket(S3ApiBaseBoto3):
         self.assertCommonResponseHeaders(
             resp['ResponseMetadata']['HTTPHeaders'])
 
+    def test_bucket_listing_with_staticweb(self):
+        if 'staticweb' not in tf.cluster_info:
+            raise tf.SkipTest('Staticweb not enabled')
+        bucket = 'bucket'
+
+        resp = self.conn.create_bucket(Bucket=bucket)
+        self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])
+
+        resp = self.conn.list_objects(Bucket=bucket)
+        self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])
+
+        # enable staticweb listings; make publicly-readable
+        conn = Connection(tf.config)
+        conn.authenticate()
+        post_status = conn.make_request('POST', [bucket], hdrs={
+            'X-Container-Read': '.r:*,.rlistings',
+            'X-Container-Meta-Web-Listings': 'true',
+        })
+        self.assertEqual(post_status, 204)
+
+        resp = self.conn.list_objects(Bucket=bucket)
+        self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])
+
     def test_put_bucket_error(self):
         event_system = self.conn.meta.events
         event_system.unregister(

test/functional/test_staticweb.py

@@ -21,7 +21,6 @@ from six.moves.urllib.parse import unquote
 from swift.common.utils import quote
 from swift.common.swob import str_to_wsgi
 import test.functional as tf
-from test.functional import cluster_info
 from test.functional.tests import Utils, Base, Base2, BaseEnv
 from test.functional.swift_test_client import Account, Connection, \
     ResponseError
@@ -38,7 +37,7 @@ def tearDownModule():
 def requires_domain_remap(func):
     @functools.wraps(func)
     def wrapper(*args, **kwargs):
-        if 'domain_remap' not in cluster_info:
+        if 'domain_remap' not in tf.cluster_info:
             raise SkipTest('Domain Remap is not enabled')
         # domain_remap middleware does not advertise its storage_domain values
         # in swift /info responses so a storage_domain must be configured in
@@ -60,7 +59,7 @@ class TestStaticWebEnv(BaseEnv):
         cls.conn.authenticate()
 
         if cls.static_web_enabled is None:
-            cls.static_web_enabled = 'staticweb' in cluster_info
+            cls.static_web_enabled = 'staticweb' in tf.cluster_info
             if not cls.static_web_enabled:
                 return
 

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

@@ -209,6 +209,7 @@ class TestS3ApiS3Acl(S3ApiTestCase):
                                      'x-amz-acl': 'private'})
         status, headers, body = self.call_s3api(req)
         self.assertEqual(status.split()[0], '200')
+        self.assertIn('REMOTE_USER', req.environ)
 
     def test_canned_acl_public_read(self):
         req = Request.blank('/bucket/object?acl',