diff --git a/etc/proxy-server.conf-sample b/etc/proxy-server.conf-sample
index 030d0d93d8..6e4d369f39 100644
--- a/etc/proxy-server.conf-sample
+++ b/etc/proxy-server.conf-sample
@@ -1008,6 +1008,11 @@ use = egg:swift#gatekeeper
 # difficult-to-delete data.
 # shunt_inbound_x_timestamp = true
 #
+# Set this to true if you want to allow clients to access and manipulate the
+# (normally internal-to-swift) null namespace by including a header like
+#    X-Allow-Reserved-Names: true
+# allow_reserved_names_header = false
+#
 # You can override the default log routing for this filter here:
 # set log_name = gatekeeper
 # set log_facility = LOG_LOCAL0
diff --git a/swift/account/backend.py b/swift/account/backend.py
index 7e65fa054e..cb1eeb5863 100644
--- a/swift/account/backend.py
+++ b/swift/account/backend.py
@@ -22,7 +22,7 @@ import sqlite3
 
 import six
 
-from swift.common.utils import Timestamp
+from swift.common.utils import Timestamp, RESERVED_BYTE
 from swift.common.db import DatabaseBroker, utf8encode, zero_like
 
 DATADIR = 'accounts'
@@ -355,7 +355,7 @@ class AccountBroker(DatabaseBroker):
             ''').fetchone())
 
     def list_containers_iter(self, limit, marker, end_marker, prefix,
-                             delimiter, reverse=False):
+                             delimiter, reverse=False, allow_reserved=False):
         """
         Get a list of containers sorted by name starting at marker onward, up
         to limit entries. Entries will begin with the prefix and will not have
@@ -367,6 +367,7 @@ class AccountBroker(DatabaseBroker):
         :param prefix: prefix query
         :param delimiter: delimiter for query
         :param reverse: reverse the result order.
+        :param allow_reserved: exclude names with reserved-byte by default
 
         :returns: list of tuples of (name, object_count, bytes_used,
                   put_timestamp, 0)
@@ -410,6 +411,9 @@ class AccountBroker(DatabaseBroker):
                 elif prefix:
                     query += ' name >= ? AND'
                     query_args.append(prefix)
+                if not allow_reserved:
+                    query += ' name >= ? AND'
+                    query_args.append(chr(ord(RESERVED_BYTE) + 1))
                 if self.get_db_version(conn) < 1:
                     query += ' +deleted = 0'
                 else:
@@ -441,7 +445,7 @@ class AccountBroker(DatabaseBroker):
                         curs.close()
                         return results
                     end = name.find(delimiter, len(prefix))
-                    if end > 0:
+                    if end >= 0:
                         if reverse:
                             end_marker = name[:end + len(delimiter)]
                         else:
diff --git a/swift/account/reaper.py b/swift/account/reaper.py
index 885b0099cf..fc47d776aa 100644
--- a/swift/account/reaper.py
+++ b/swift/account/reaper.py
@@ -262,7 +262,7 @@ class AccountReaper(Daemon):
             container_limit *= len(nodes)
         try:
             containers = list(broker.list_containers_iter(
-                container_limit, '', None, None, None))
+                container_limit, '', None, None, None, allow_reserved=True))
             while containers:
                 try:
                     for (container, _junk, _junk, _junk, _junk) in containers:
@@ -282,7 +282,8 @@ class AccountReaper(Daemon):
                     self.logger.exception(
                         'Exception with containers for account %s', account)
                 containers = list(broker.list_containers_iter(
-                    container_limit, containers[-1][0], None, None, None))
+                    container_limit, containers[-1][0], None, None, None,
+                    allow_reserved=True))
             log_buf = ['Completed pass on account %s' % account]
         except (Exception, Timeout):
             self.logger.exception('Exception with account %s', account)
diff --git a/swift/account/server.py b/swift/account/server.py
index bda89f82e9..2bfbd93166 100644
--- a/swift/account/server.py
+++ b/swift/account/server.py
@@ -26,7 +26,8 @@ from swift.account.backend import AccountBroker, DATADIR
 from swift.account.utils import account_listing_response, get_response_headers
 from swift.common.db import DatabaseConnectionError, DatabaseAlreadyExists
 from swift.common.request_helpers import get_param, \
-    split_and_validate_path
+    split_and_validate_path, validate_internal_account, \
+    validate_internal_container
 from swift.common.utils import get_logger, hash_path, public, \
     Timestamp, storage_directory, config_true_value, \
     timing_stats, replication, get_log_line, \
@@ -44,6 +45,32 @@ from swift.common.swob import HTTPAccepted, HTTPBadRequest, \
 from swift.common.request_helpers import is_sys_or_user_meta
 
 
+def get_account_name_and_placement(req):
+    """
+    Split and validate path for an account.
+
+    :param req: a swob request
+
+    :returns: a tuple of path parts as strings
+    """
+    drive, part, account = split_and_validate_path(req, 3)
+    validate_internal_account(account)
+    return drive, part, account
+
+
+def get_container_name_and_placement(req):
+    """
+    Split and validate path for a container.
+
+    :param req: a swob request
+
+    :returns: a tuple of path parts as strings
+    """
+    drive, part, account, container = split_and_validate_path(req, 3, 4)
+    validate_internal_container(account, container)
+    return drive, part, account, container
+
+
 class AccountController(BaseStorageServer):
     """WSGI controller for the account server."""
 
@@ -96,7 +123,7 @@ class AccountController(BaseStorageServer):
     @timing_stats()
     def DELETE(self, req):
         """Handle HTTP DELETE request."""
-        drive, part, account = split_and_validate_path(req, 3)
+        drive, part, account = get_account_name_and_placement(req)
         try:
             check_drive(self.root, drive, self.mount_check)
         except ValueError:
@@ -120,7 +147,7 @@ class AccountController(BaseStorageServer):
     @timing_stats()
     def PUT(self, req):
         """Handle HTTP PUT request."""
-        drive, part, account, container = split_and_validate_path(req, 3, 4)
+        drive, part, account, container = get_container_name_and_placement(req)
         try:
             check_drive(self.root, drive, self.mount_check)
         except ValueError:
@@ -185,7 +212,7 @@ class AccountController(BaseStorageServer):
     @timing_stats()
     def HEAD(self, req):
         """Handle HTTP HEAD request."""
-        drive, part, account = split_and_validate_path(req, 3)
+        drive, part, account = get_account_name_and_placement(req)
         out_content_type = listing_formats.get_listing_content_type(req)
         try:
             check_drive(self.root, drive, self.mount_check)
@@ -204,7 +231,7 @@ class AccountController(BaseStorageServer):
     @timing_stats()
     def GET(self, req):
         """Handle HTTP GET request."""
-        drive, part, account = split_and_validate_path(req, 3)
+        drive, part, account = get_account_name_and_placement(req)
         prefix = get_param(req, 'prefix')
         delimiter = get_param(req, 'delimiter')
         limit = constraints.ACCOUNT_LISTING_LIMIT
@@ -262,7 +289,7 @@ class AccountController(BaseStorageServer):
     @timing_stats()
     def POST(self, req):
         """Handle HTTP POST request."""
-        drive, part, account = split_and_validate_path(req, 3)
+        drive, part, account = get_account_name_and_placement(req)
         req_timestamp = valid_timestamp(req)
         try:
             check_drive(self.root, drive, self.mount_check)
@@ -280,8 +307,8 @@ class AccountController(BaseStorageServer):
         start_time = time.time()
         req = Request(env)
         self.logger.txn_id = req.headers.get('x-trans-id', None)
-        if not check_utf8(wsgi_to_str(req.path_info)):
-            res = HTTPPreconditionFailed(body='Invalid UTF8 or contains NULL')
+        if not check_utf8(wsgi_to_str(req.path_info), internal=True):
+            res = HTTPPreconditionFailed(body='Invalid UTF8')
         else:
             try:
                 # disallow methods which are not publicly accessible
diff --git a/swift/account/utils.py b/swift/account/utils.py
index 4e5ebe1add..89fb26f841 100644
--- a/swift/account/utils.py
+++ b/swift/account/utils.py
@@ -17,6 +17,7 @@ import json
 
 import six
 
+from swift.common import constraints
 from swift.common.middleware import listing_formats
 from swift.common.swob import HTTPOk, HTTPNoContent, str_to_wsgi
 from swift.common.utils import Timestamp
@@ -71,15 +72,17 @@ def get_response_headers(broker):
 
 
 def account_listing_response(account, req, response_content_type, broker=None,
-                             limit='', marker='', end_marker='', prefix='',
-                             delimiter='', reverse=False):
+                             limit=constraints.ACCOUNT_LISTING_LIMIT,
+                             marker='', end_marker='', prefix='', delimiter='',
+                             reverse=False):
     if broker is None:
         broker = FakeAccountBroker()
 
     resp_headers = get_response_headers(broker)
 
     account_list = broker.list_containers_iter(limit, marker, end_marker,
-                                               prefix, delimiter, reverse)
+                                               prefix, delimiter, reverse,
+                                               req.allow_reserved_names)
     data = []
     for (name, object_count, bytes_used, put_timestamp, is_subdir) \
             in account_list:
diff --git a/swift/common/constraints.py b/swift/common/constraints.py
index c7263f7b90..dd2e75a8f0 100644
--- a/swift/common/constraints.py
+++ b/swift/common/constraints.py
@@ -346,12 +346,13 @@ def check_delete_headers(request):
     return request
 
 
-def check_utf8(string):
+def check_utf8(string, internal=False):
     """
     Validate if a string is valid UTF-8 str or unicode and that it
-    does not contain any null character.
+    does not contain any reserved characters.
 
     :param string: string to be validated
+    :param internal: boolean, allows reserved characters if True
     :returns: True if the string is valid utf-8 str or unicode and
               contains no null characters, False otherwise
     """
@@ -382,7 +383,9 @@ def check_utf8(string):
         if any(0xD800 <= ord(codepoint) <= 0xDFFF
                for codepoint in decoded):
             return False
-        return b'\x00' not in encoded
+        if b'\x00' != utils.RESERVED_BYTE and b'\x00' in encoded:
+            return False
+        return True if internal else utils.RESERVED_BYTE not in encoded
     # If string is unicode, decode() will raise UnicodeEncodeError
     # So, we should catch both UnicodeDecodeError & UnicodeEncodeError
     except UnicodeError:
@@ -413,6 +416,7 @@ def check_name_format(req, name, target_type):
             body='%s name cannot contain slashes' % target_type)
     return name
 
+
 check_account_format = functools.partial(check_name_format,
                                          target_type='Account')
 check_container_format = functools.partial(check_name_format,
diff --git a/swift/common/direct_client.py b/swift/common/direct_client.py
index 174b61601f..3c9f140e67 100644
--- a/swift/common/direct_client.py
+++ b/swift/common/direct_client.py
@@ -100,6 +100,7 @@ def _make_req(node, part, method, path, headers, stype,
         if content_length is None:
             headers['Transfer-Encoding'] = 'chunked'
 
+    headers.setdefault('X-Backend-Allow-Reserved-Names', 'true')
     with Timeout(conn_timeout):
         conn = http_connect(node['ip'], node['port'], node['device'], part,
                             method, path, headers=headers)
@@ -193,6 +194,7 @@ def gen_headers(hdrs_in=None, add_ts=True):
         hdrs_out['X-Timestamp'] = Timestamp.now().internal
     if 'user-agent' not in hdrs_out:
         hdrs_out['User-Agent'] = 'direct-client %s' % os.getpid()
+    hdrs_out.setdefault('X-Backend-Allow-Reserved-Names', 'true')
     return hdrs_out
 
 
diff --git a/swift/common/internal_client.py b/swift/common/internal_client.py
index 0e2c4ab449..851282bd6f 100644
--- a/swift/common/internal_client.py
+++ b/swift/common/internal_client.py
@@ -184,6 +184,7 @@ class InternalClient(object):
 
         headers = dict(headers)
         headers['user-agent'] = self.user_agent
+        headers.setdefault('x-backend-allow-reserved-names', 'true')
         for attempt in range(self.request_tries):
             resp = exc_type = exc_value = exc_traceback = None
             req = Request.blank(
@@ -384,6 +385,19 @@ class InternalClient(object):
         return self._iter_items(path, marker, end_marker, prefix,
                                 acceptable_statuses)
 
+    def create_account(self, account):
+        """
+        Creates an account.
+
+        :param account: Account to create.
+        :raises UnexpectedResponse: Exception raised when requests fail
+                                    to get a response with an acceptable status
+        :raises Exception: Exception is raised when code fails in an
+                           unexpected way.
+        """
+        path = self.make_path(account)
+        self.make_request('PUT', path, {}, (201, 202))
+
     def delete_account(self, account, acceptable_statuses=(2, HTTP_NOT_FOUND)):
         """
         Deletes an account.
@@ -514,7 +528,8 @@ class InternalClient(object):
         self.make_request('PUT', path, headers, acceptable_statuses)
 
     def delete_container(
-            self, account, container, acceptable_statuses=(2, HTTP_NOT_FOUND)):
+            self, account, container, headers=None,
+            acceptable_statuses=(2, HTTP_NOT_FOUND)):
         """
         Deletes a container.
 
@@ -529,8 +544,9 @@ class InternalClient(object):
                            unexpected way.
         """
 
+        headers = headers or {}
         path = self.make_path(account, container)
-        self.make_request('DELETE', path, {}, acceptable_statuses)
+        self.make_request('DELETE', path, headers, acceptable_statuses)
 
     def get_container_metadata(
             self, account, container, metadata_prefix='',
@@ -669,7 +685,7 @@ class InternalClient(object):
         return self._get_metadata(path, metadata_prefix, acceptable_statuses,
                                   headers=headers, params=params)
 
-    def get_object(self, account, container, obj, headers,
+    def get_object(self, account, container, obj, headers=None,
                    acceptable_statuses=(2,), params=None):
         """
         Gets an object.
@@ -771,7 +787,8 @@ class InternalClient(object):
             path, metadata, metadata_prefix, acceptable_statuses)
 
     def upload_object(
-            self, fobj, account, container, obj, headers=None):
+            self, fobj, account, container, obj, headers=None,
+            acceptable_statuses=(2,)):
         """
         :param fobj: File object to read object's content from.
         :param account: The object's account.
@@ -789,7 +806,7 @@ class InternalClient(object):
         if 'Content-Length' not in headers:
             headers['Transfer-Encoding'] = 'chunked'
         path = self.make_path(account, container, obj)
-        self.make_request('PUT', path, headers, (2,), fobj)
+        self.make_request('PUT', path, headers, acceptable_statuses, fobj)
 
 
 def get_auth(url, user, key, auth_version='1.0', **kwargs):
diff --git a/swift/common/middleware/gatekeeper.py b/swift/common/middleware/gatekeeper.py
index 511394691a..0254fef98b 100644
--- a/swift/common/middleware/gatekeeper.py
+++ b/swift/common/middleware/gatekeeper.py
@@ -75,6 +75,8 @@ class GatekeeperMiddleware(object):
         self.outbound_condition = make_exclusion_test(outbound_exclusions)
         self.shunt_x_timestamp = config_true_value(
             conf.get('shunt_inbound_x_timestamp', 'true'))
+        self.allow_reserved_names_header = config_true_value(
+            conf.get('allow_reserved_names_header', 'false'))
 
     def __call__(self, env, start_response):
         req = Request(env)
@@ -89,6 +91,11 @@ class GatekeeperMiddleware(object):
             self.logger.debug('shunted request headers: %s' %
                               [('X-Timestamp', ts)])
 
+        if 'X-Allow-Reserved-Names' in req.headers \
+                and self.allow_reserved_names_header:
+            req.headers['X-Backend-Allow-Reserved-Names'] = \
+                req.headers.pop('X-Allow-Reserved-Names')
+
         def gatekeeper_response(status, response_headers, exc_info=None):
             def fixed_response_headers():
                 def relative_path(value):
diff --git a/swift/common/middleware/listing_formats.py b/swift/common/middleware/listing_formats.py
index 1924facd8f..8c07965af6 100644
--- a/swift/common/middleware/listing_formats.py
+++ b/swift/common/middleware/listing_formats.py
@@ -21,7 +21,8 @@ from swift.common.constraints import valid_api_version
 from swift.common.http import HTTP_NO_CONTENT
 from swift.common.request_helpers import get_param
 from swift.common.swob import HTTPException, HTTPNotAcceptable, Request, \
-    RESPONSE_REASONS, HTTPBadRequest
+    RESPONSE_REASONS, HTTPBadRequest, wsgi_quote, wsgi_to_bytes
+from swift.common.utils import RESERVED, get_logger
 
 
 #: Mapping of query string ``format=`` values to their corresponding
@@ -73,8 +74,6 @@ def to_xml(document_element):
 
 
 def account_to_xml(listing, account_name):
-    if isinstance(account_name, bytes):
-        account_name = account_name.decode('utf-8')
     doc = Element('account', name=account_name)
     doc.text = '\n'
     for record in listing:
@@ -91,8 +90,6 @@ def account_to_xml(listing, account_name):
 
 
 def container_to_xml(listing, base_name):
-    if isinstance(base_name, bytes):
-        base_name = base_name.decode('utf-8')
     doc = Element('container', name=base_name)
     for record in listing:
         if 'subdir' in record:
@@ -119,8 +116,33 @@ def listing_to_text(listing):
 
 
 class ListingFilter(object):
-    def __init__(self, app):
+    def __init__(self, app, conf, logger=None):
         self.app = app
+        self.logger = logger or get_logger(conf, log_route='listing-filter')
+
+    def filter_reserved(self, listing, account, container):
+        new_listing = []
+        for entry in list(listing):
+            for key in ('name', 'subdir'):
+                value = entry.get(key, '')
+                if six.PY2:
+                    value = value.encode('utf-8')
+                if RESERVED in value:
+                    if container:
+                        self.logger.warning(
+                            'Container listing for %s/%s had '
+                            'reserved byte in %s: %r',
+                            wsgi_quote(account), wsgi_quote(container),
+                            key, value)
+                    else:
+                        self.logger.warning(
+                            'Account listing for %s had '
+                            'reserved byte in %s: %r',
+                            wsgi_quote(account), key, value)
+                    break  # out of the *key* loop; check next entry
+            else:
+                new_listing.append(entry)
+        return new_listing
 
     def __call__(self, env, start_response):
         req = Request(env)
@@ -128,10 +150,10 @@ class ListingFilter(object):
             # account and container only
             version, acct, cont = req.split_path(2, 3)
         except ValueError:
-            is_container_req = False
+            is_account_or_container_req = False
         else:
-            is_container_req = True
-        if not is_container_req:
+            is_account_or_container_req = True
+        if not is_account_or_container_req:
             return self.app(env, start_response)
 
         if not valid_api_version(version) or req.method not in ('GET', 'HEAD'):
@@ -201,15 +223,21 @@ class ListingFilter(object):
             start_response(status, headers)
             return [body]
 
+        if not req.allow_reserved_names:
+            listing = self.filter_reserved(listing, acct, cont)
+
         try:
             if out_content_type.endswith('/xml'):
                 if cont:
-                    body = container_to_xml(listing, cont)
+                    body = container_to_xml(
+                        listing, wsgi_to_bytes(cont).decode('utf-8'))
                 else:
-                    body = account_to_xml(listing, acct)
+                    body = account_to_xml(
+                        listing, wsgi_to_bytes(acct).decode('utf-8'))
             elif out_content_type == 'text/plain':
                 body = listing_to_text(listing)
-            # else, json -- we continue down here to be sure we set charset
+            else:
+                body = json.dumps(listing).encode('ascii')
         except KeyError:
             # listing was in a bad format -- funky static web listing??
             start_response(status, headers)
@@ -226,4 +254,9 @@ class ListingFilter(object):
 
 
 def filter_factory(global_conf, **local_conf):
-    return ListingFilter
+    conf = global_conf.copy()
+    conf.update(local_conf)
+
+    def listing_filter(app):
+        return ListingFilter(app, conf)
+    return listing_filter
diff --git a/swift/common/request_helpers.py b/swift/common/request_helpers.py
index 03d571f1f8..bdc234c9ea 100644
--- a/swift/common/request_helpers.py
+++ b/swift/common/request_helpers.py
@@ -38,8 +38,8 @@ from swift.common.swob import HTTPBadRequest, \
 from swift.common.utils import split_path, validate_device_partition, \
     close_if_possible, maybe_multipart_byteranges_to_document_iters, \
     multipart_byteranges_to_document_iters, parse_content_type, \
-    parse_content_range, csv_append, list_from_csv, Spliterator, quote
-
+    parse_content_range, csv_append, list_from_csv, Spliterator, quote, \
+    RESERVED
 from swift.common.wsgi import make_subrequest
 
 
@@ -83,6 +83,54 @@ def get_param(req, name, default=None):
     return value
 
 
+def _validate_internal_name(name, type_='name'):
+    if RESERVED in name and not name.startswith(RESERVED):
+        raise HTTPBadRequest(body='Invalid reserved-namespace %s' % (type_))
+
+
+def validate_internal_account(account):
+    """
+    Validate internal account name.
+
+    :raises: HTTPBadRequest
+    """
+    _validate_internal_name(account, 'account')
+
+
+def validate_internal_container(account, container):
+    """
+    Validate internal account and container names.
+
+    :raises: HTTPBadRequest
+    """
+    if not account:
+        raise ValueError('Account is required')
+    validate_internal_account(account)
+    if container:
+        _validate_internal_name(container, 'container')
+
+
+def validate_internal_obj(account, container, obj):
+    """
+    Validate internal account, container and object names.
+
+    :raises: HTTPBadRequest
+    """
+    if not account:
+        raise ValueError('Account is required')
+    if not container:
+        raise ValueError('Container is required')
+    validate_internal_container(account, container)
+    if obj:
+        _validate_internal_name(obj, 'object')
+        if container.startswith(RESERVED) and not obj.startswith(RESERVED):
+            raise HTTPBadRequest(body='Invalid user-namespace object '
+                                 'in reserved-namespace container')
+        elif obj.startswith(RESERVED) and not container.startswith(RESERVED):
+            raise HTTPBadRequest(body='Invalid reserved-namespace object '
+                                 'in user-namespace container')
+
+
 def get_name_and_placement(request, minsegs=1, maxsegs=None,
                            rest_with_last=False):
     """
@@ -273,6 +321,28 @@ def get_container_update_override_key(key):
     return header.title()
 
 
+def get_reserved_name(*parts):
+    """
+    Generate a valid reserved name that joins the component parts.
+
+    :returns: a string
+    """
+    if any(RESERVED in p for p in parts):
+        raise ValueError('Invalid reserved part in components')
+    return RESERVED + RESERVED.join(parts)
+
+
+def split_reserved_name(name):
+    """
+    Seperate a valid reserved name into the component parts.
+
+    :returns: a list of strings
+    """
+    if not name.startswith(RESERVED):
+        raise ValueError('Invalid reserved name')
+    return name.split(RESERVED)[1:]
+
+
 def remove_items(headers, condition):
     """
     Removes items from a dict whose keys satisfy
diff --git a/swift/common/swob.py b/swift/common/swob.py
index e864a39a5a..c5bb978389 100644
--- a/swift/common/swob.py
+++ b/swift/common/swob.py
@@ -52,7 +52,7 @@ from six.moves import urllib
 
 from swift.common.header_key_dict import HeaderKeyDict
 from swift.common.utils import UTC, reiterate, split_path, Timestamp, pairs, \
-    close_if_possible, closing_if_possible
+    close_if_possible, closing_if_possible, config_true_value
 from swift.common.exceptions import InvalidTimestamp
 
 
@@ -1063,6 +1063,11 @@ class Request(object):
         "Provides the full url of the request"
         return self.host_url + self.path_qs
 
+    @property
+    def allow_reserved_names(self):
+        return config_true_value(self.environ.get(
+            'HTTP_X_BACKEND_ALLOW_RESERVED_NAMES'))
+
     def as_referer(self):
         return self.method + ' ' + self.url
 
diff --git a/swift/common/utils.py b/swift/common/utils.py
index e4bb964bcd..68005f324b 100644
--- a/swift/common/utils.py
+++ b/swift/common/utils.py
@@ -186,6 +186,10 @@ O_TMPFILE = getattr(os, 'O_TMPFILE', 0o20000000 | os.O_DIRECTORY)
 IPV6_RE = re.compile("^\[(?P<address>.*)\](:(?P<port>[0-9]+))?$")
 
 MD5_OF_EMPTY_STRING = 'd41d8cd98f00b204e9800998ecf8427e'
+RESERVED_BYTE = b'\x00'
+RESERVED_STR = u'\x00'
+RESERVED = '\x00'
+
 
 LOG_LINE_DEFAULT_FORMAT = '{remote_addr} - - [{time.d}/{time.b}/{time.Y}' \
                           ':{time.H}:{time.M}:{time.S} +0000] ' \
diff --git a/swift/container/backend.py b/swift/container/backend.py
index 0915901a05..8dee9d243c 100644
--- a/swift/container/backend.py
+++ b/swift/container/backend.py
@@ -30,7 +30,8 @@ from swift.common.exceptions import LockTimeout
 from swift.common.utils import Timestamp, encode_timestamps, \
     decode_timestamps, extract_swift_bytes, storage_directory, hash_path, \
     ShardRange, renamer, find_shard_range, MD5_OF_EMPTY_STRING, mkdirs, \
-    get_db_files, parse_db_filename, make_db_file_path, split_path
+    get_db_files, parse_db_filename, make_db_file_path, split_path, \
+    RESERVED_BYTE
 from swift.common.db import DatabaseBroker, utf8encode, BROKER_TIMEOUT, \
     zero_like, DatabaseAlreadyExists
 
@@ -1028,7 +1029,8 @@ class ContainerBroker(DatabaseBroker):
     def list_objects_iter(self, limit, marker, end_marker, prefix, delimiter,
                           path=None, storage_policy_index=0, reverse=False,
                           include_deleted=False, since_row=None,
-                          transform_func=None, all_policies=False):
+                          transform_func=None, all_policies=False,
+                          allow_reserved=False):
         """
         Get a list of objects sorted by name starting at marker onward, up
         to limit entries.  Entries will begin with the prefix and will not
@@ -1054,6 +1056,8 @@ class ContainerBroker(DatabaseBroker):
             :meth:`~_transform_record`; defaults to :meth:`~_transform_record`.
         :param all_policies: if True, include objects for all storage policies
             ignoring any value given for ``storage_policy_index``
+        :param allow_reserved: exclude names with reserved-byte by default
+
         :returns: list of tuples of (name, created_at, size, content_type,
                   etag, deleted)
         """
@@ -1110,6 +1114,9 @@ class ContainerBroker(DatabaseBroker):
                 elif prefix:
                     query_conditions.append('name >= ?')
                     query_args.append(prefix)
+                if not allow_reserved:
+                    query_conditions.append('name >= ?')
+                    query_args.append(chr(ord(RESERVED_BYTE) + 1))
                 query_conditions.append(deleted_key + deleted_arg)
                 if since_row:
                     query_conditions.append('ROWID > ?')
diff --git a/swift/container/server.py b/swift/container/server.py
index 0e8a443442..5d0915aa07 100644
--- a/swift/container/server.py
+++ b/swift/container/server.py
@@ -32,7 +32,8 @@ from swift.container.replicator import ContainerReplicatorRpc
 from swift.common.db import DatabaseAlreadyExists
 from swift.common.container_sync_realms import ContainerSyncRealms
 from swift.common.request_helpers import get_param, \
-    split_and_validate_path, is_sys_or_user_meta
+    split_and_validate_path, is_sys_or_user_meta, \
+    validate_internal_container, validate_internal_obj
 from swift.common.utils import get_logger, hash_path, public, \
     Timestamp, storage_directory, validate_sync_to, \
     config_true_value, timing_stats, replication, \
@@ -83,6 +84,33 @@ def gen_resp_headers(info, is_deleted=False):
     return headers
 
 
+def get_container_name_and_placement(req):
+    """
+    Split and validate path for a container.
+
+    :param req: a swob request
+
+    :returns: a tuple of path parts as strings
+    """
+    drive, part, account, container = split_and_validate_path(req, 4)
+    validate_internal_container(account, container)
+    return drive, part, account, container
+
+
+def get_obj_name_and_placement(req):
+    """
+    Split and validate path for an object.
+
+    :param req: a swob request
+
+    :returns: a tuple of path parts as strings
+    """
+    drive, part, account, container, obj = split_and_validate_path(
+        req, 4, 5, True)
+    validate_internal_obj(account, container, obj)
+    return drive, part, account, container, obj
+
+
 class ContainerController(BaseStorageServer):
     """WSGI Controller for the container server."""
 
@@ -311,8 +339,7 @@ class ContainerController(BaseStorageServer):
     @timing_stats()
     def DELETE(self, req):
         """Handle HTTP DELETE request."""
-        drive, part, account, container, obj = split_and_validate_path(
-            req, 4, 5, True)
+        drive, part, account, container, obj = get_obj_name_and_placement(req)
         req_timestamp = valid_timestamp(req)
         try:
             check_drive(self.root, drive, self.mount_check)
@@ -433,8 +460,7 @@ class ContainerController(BaseStorageServer):
     @timing_stats()
     def PUT(self, req):
         """Handle HTTP PUT request."""
-        drive, part, account, container, obj = split_and_validate_path(
-            req, 4, 5, True)
+        drive, part, account, container, obj = get_obj_name_and_placement(req)
         req_timestamp = valid_timestamp(req)
         if 'x-container-sync-to' in req.headers:
             err, sync_to, realm, realm_key = validate_sync_to(
@@ -514,8 +540,7 @@ class ContainerController(BaseStorageServer):
     @timing_stats(sample_rate=0.1)
     def HEAD(self, req):
         """Handle HTTP HEAD request."""
-        drive, part, account, container, obj = split_and_validate_path(
-            req, 4, 5, True)
+        drive, part, account, container, obj = get_obj_name_and_placement(req)
         out_content_type = listing_formats.get_listing_content_type(req)
         try:
             check_drive(self.root, drive, self.mount_check)
@@ -632,8 +657,7 @@ class ContainerController(BaseStorageServer):
         :param req: an instance of :class:`swift.common.swob.Request`
         :returns: an instance of :class:`swift.common.swob.Response`
         """
-        drive, part, account, container, obj = split_and_validate_path(
-            req, 4, 5, True)
+        drive, part, account, container, obj = get_obj_name_and_placement(req)
         path = get_param(req, 'path')
         prefix = get_param(req, 'prefix')
         delimiter = get_param(req, 'delimiter')
@@ -696,7 +720,7 @@ class ContainerController(BaseStorageServer):
             container_list = src_broker.list_objects_iter(
                 limit, marker, end_marker, prefix, delimiter, path,
                 storage_policy_index=info['storage_policy_index'],
-                reverse=reverse)
+                reverse=reverse, allow_reserved=req.allow_reserved_names)
         return self.create_listing(req, out_content_type, info, resp_headers,
                                    broker.metadata, container_list, container)
 
@@ -751,7 +775,7 @@ class ContainerController(BaseStorageServer):
         """
         Handle HTTP UPDATE request (merge_items RPCs coming from the proxy.)
         """
-        drive, part, account, container = split_and_validate_path(req, 4)
+        drive, part, account, container = get_container_name_and_placement(req)
         req_timestamp = valid_timestamp(req)
         try:
             check_drive(self.root, drive, self.mount_check)
@@ -775,7 +799,7 @@ class ContainerController(BaseStorageServer):
     @timing_stats()
     def POST(self, req):
         """Handle HTTP POST request."""
-        drive, part, account, container = split_and_validate_path(req, 4)
+        drive, part, account, container = get_container_name_and_placement(req)
         req_timestamp = valid_timestamp(req)
         if 'x-container-sync-to' in req.headers:
             err, sync_to, realm, realm_key = validate_sync_to(
@@ -800,7 +824,7 @@ class ContainerController(BaseStorageServer):
         start_time = time.time()
         req = Request(env)
         self.logger.txn_id = req.headers.get('x-trans-id', None)
-        if not check_utf8(wsgi_to_str(req.path_info)):
+        if not check_utf8(wsgi_to_str(req.path_info), internal=True):
             res = HTTPPreconditionFailed(body='Invalid UTF8 or contains NULL')
         else:
             try:
diff --git a/swift/obj/server.py b/swift/obj/server.py
index 98f6887c50..33bc7ff007 100644
--- a/swift/obj/server.py
+++ b/swift/obj/server.py
@@ -35,7 +35,8 @@ from swift.common.utils import public, get_logger, \
     normalize_delete_at_timestamp, get_log_line, Timestamp, \
     get_expirer_container, parse_mime_headers, \
     iter_multipart_mime_documents, extract_swift_bytes, safe_json_loads, \
-    config_auto_int_value, split_path, get_redirect_data, normalize_timestamp
+    config_auto_int_value, split_path, get_redirect_data, \
+    normalize_timestamp
 from swift.common.bufferedhttp import http_connect
 from swift.common.constraints import check_object_creation, \
     valid_timestamp, check_utf8
@@ -51,7 +52,7 @@ from swift.common.base_storage_server import BaseStorageServer
 from swift.common.header_key_dict import HeaderKeyDict
 from swift.common.request_helpers import get_name_and_placement, \
     is_user_meta, is_sys_or_user_meta, is_object_transient_sysmeta, \
-    resolve_etag_is_at_header, is_sys_meta
+    resolve_etag_is_at_header, is_sys_meta, validate_internal_obj
 from swift.common.swob import HTTPAccepted, HTTPBadRequest, HTTPCreated, \
     HTTPInternalServerError, HTTPNoContent, HTTPNotFound, \
     HTTPPreconditionFailed, HTTPRequestTimeout, HTTPUnprocessableEntity, \
@@ -89,6 +90,20 @@ def drain(file_like, read_size, timeout):
                 break
 
 
+def get_obj_name_and_placement(request):
+    """
+    Split and validate path for an object.
+
+    :param request: a swob request
+
+    :returns: a tuple of path parts and storage policy
+    """
+    device, partition, account, container, obj, policy = \
+        get_name_and_placement(request, 5, 5, True)
+    validate_internal_obj(account, container, obj)
+    return device, partition, account, container, obj, policy
+
+
 def _make_backend_fragments_header(fragments):
     if fragments:
         result = {}
@@ -603,7 +618,8 @@ class ObjectController(BaseStorageServer):
     def POST(self, request):
         """Handle HTTP POST requests for the Swift Object Server."""
         device, partition, account, container, obj, policy = \
-            get_name_and_placement(request, 5, 5, True)
+            get_obj_name_and_placement(request)
+
         req_timestamp = valid_timestamp(request)
         new_delete_at = int(request.headers.get('X-Delete-At') or 0)
         if new_delete_at and new_delete_at < req_timestamp:
@@ -995,7 +1011,7 @@ class ObjectController(BaseStorageServer):
     def PUT(self, request):
         """Handle HTTP PUT requests for the Swift Object Server."""
         device, partition, account, container, obj, policy = \
-            get_name_and_placement(request, 5, 5, True)
+            get_obj_name_and_placement(request)
         disk_file, fsize, orig_metadata = self._pre_create_checks(
             request, device, partition, account, container, obj, policy)
         writer = disk_file.writer(size=fsize)
@@ -1037,7 +1053,7 @@ class ObjectController(BaseStorageServer):
     def GET(self, request):
         """Handle HTTP GET requests for the Swift Object Server."""
         device, partition, account, container, obj, policy = \
-            get_name_and_placement(request, 5, 5, True)
+            get_obj_name_and_placement(request)
         request.headers.setdefault('X-Timestamp',
                                    normalize_timestamp(time.time()))
         req_timestamp = valid_timestamp(request)
@@ -1104,7 +1120,7 @@ class ObjectController(BaseStorageServer):
     def HEAD(self, request):
         """Handle HTTP HEAD requests for the Swift Object Server."""
         device, partition, account, container, obj, policy = \
-            get_name_and_placement(request, 5, 5, True)
+            get_obj_name_and_placement(request)
         request.headers.setdefault('X-Timestamp',
                                    normalize_timestamp(time.time()))
         req_timestamp = valid_timestamp(request)
@@ -1163,7 +1179,7 @@ class ObjectController(BaseStorageServer):
     def DELETE(self, request):
         """Handle HTTP DELETE requests for the Swift Object Server."""
         device, partition, account, container, obj, policy = \
-            get_name_and_placement(request, 5, 5, True)
+            get_obj_name_and_placement(request)
         req_timestamp = valid_timestamp(request)
         next_part_power = request.headers.get('X-Backend-Next-Part-Power')
         try:
@@ -1275,7 +1291,7 @@ class ObjectController(BaseStorageServer):
         req = Request(env)
         self.logger.txn_id = req.headers.get('x-trans-id', None)
 
-        if not check_utf8(wsgi_to_str(req.path_info)):
+        if not check_utf8(wsgi_to_str(req.path_info), internal=True):
             res = HTTPPreconditionFailed(body='Invalid UTF8 or contains NULL')
         else:
             try:
diff --git a/swift/proxy/controllers/base.py b/swift/proxy/controllers/base.py
index b56bcd285f..1d25a95d9f 100644
--- a/swift/proxy/controllers/base.py
+++ b/swift/proxy/controllers/base.py
@@ -356,6 +356,9 @@ def get_container_info(env, app, swift_source=None):
         req = _prepare_pre_auth_info_request(
             env, ("/%s/%s/%s" % (version, wsgi_account, wsgi_container)),
             (swift_source or 'GET_CONTAINER_INFO'))
+        # *Always* allow reserved names for get-info requests -- it's on the
+        # caller to keep the result private-ish
+        req.headers['X-Backend-Allow-Reserved-Names'] = 'true'
         resp = req.get_response(app)
         close_if_possible(resp.app_iter)
         # Check in infocache to see if the proxy (or anyone else) already
@@ -412,6 +415,9 @@ def get_account_info(env, app, swift_source=None):
         req = _prepare_pre_auth_info_request(
             env, "/%s/%s" % (version, wsgi_account),
             (swift_source or 'GET_ACCOUNT_INFO'))
+        # *Always* allow reserved names for get-info requests -- it's on the
+        # caller to keep the result private-ish
+        req.headers['X-Backend-Allow-Reserved-Names'] = 'true'
         resp = req.get_response(app)
         close_if_possible(resp.app_iter)
         # Check in infocache to see if the proxy (or anyone else) already
@@ -739,6 +745,9 @@ def _get_object_info(app, env, account, container, obj, swift_source=None):
     # Not in cache, let's try the object servers
     path = '/v1/%s/%s/%s' % (account, container, obj)
     req = _prepare_pre_auth_info_request(env, path, swift_source)
+    # *Always* allow reserved names for get-info requests -- it's on the
+    # caller to keep the result private-ish
+    req.headers['X-Backend-Allow-Reserved-Names'] = 'true'
     resp = req.get_response(app)
     # Unlike get_account_info() and get_container_info(), we don't save
     # things in memcache, so we can store the info without network traffic,
diff --git a/swift/proxy/controllers/obj.py b/swift/proxy/controllers/obj.py
index e272287ab4..f237d98293 100644
--- a/swift/proxy/controllers/obj.py
+++ b/swift/proxy/controllers/obj.py
@@ -24,7 +24,7 @@
 #   These shenanigans are to ensure all related objects can be garbage
 # collected. We've seen objects hang around forever otherwise.
 
-from six.moves.urllib.parse import unquote
+from six.moves.urllib.parse import quote, unquote
 from six.moves import zip
 
 import collections
@@ -72,7 +72,7 @@ from swift.common.swob import HTTPAccepted, HTTPBadRequest, HTTPNotFound, \
     HTTPUnprocessableEntity, Response, HTTPException, \
     HTTPRequestedRangeNotSatisfiable, Range, HTTPInternalServerError
 from swift.common.request_helpers import update_etag_is_at_header, \
-    resolve_etag_is_at_header
+    resolve_etag_is_at_header, validate_internal_obj
 
 
 def check_content_type(req):
@@ -168,6 +168,8 @@ class BaseObjectController(Controller):
         self.account_name = unquote(account_name)
         self.container_name = unquote(container_name)
         self.object_name = unquote(object_name)
+        validate_internal_obj(
+            self.account_name, self.container_name, self.object_name)
 
     def iter_nodes_local_first(self, ring, partition, policy=None,
                                local_handoffs_first=False):
@@ -637,7 +639,8 @@ class BaseObjectController(Controller):
             except (Exception, Timeout):
                 self.app.exception_occurred(
                     node, _('Object'),
-                    _('Expect: 100-continue on %s') % req.swift_entity_path)
+                    _('Expect: 100-continue on %s') %
+                    quote(req.swift_entity_path))
 
     def _get_put_connections(self, req, nodes, partition, outgoing_headers,
                              policy):
@@ -1399,10 +1402,10 @@ class ECAppIter(object):
             except ChunkReadTimeout:
                 # unable to resume in GetOrHeadHandler
                 self.logger.exception(_("Timeout fetching fragments for %r"),
-                                      self.path)
+                                      quote(self.path))
             except:  # noqa
                 self.logger.exception(_("Exception fetching fragments for"
-                                        " %r"), self.path)
+                                        " %r"), quote(self.path))
             finally:
                 queue.resize(2)  # ensure there's room
                 queue.put(None)
@@ -1431,7 +1434,7 @@ class ECAppIter(object):
                     segment = self.policy.pyeclib_driver.decode(fragments)
                 except ECDriverError:
                     self.logger.exception(_("Error decoding fragments for"
-                                            " %r"), self.path)
+                                            " %r"), quote(self.path))
                     raise
 
                 yield segment
@@ -1670,7 +1673,7 @@ class Putter(object):
                 self.failed = True
                 self.send_exception_handler(self.node, _('Object'),
                                             _('Trying to write to %s')
-                                            % self.path)
+                                            % quote(self.path))
 
     def close(self):
         # release reference to response to ensure connection really does close,
diff --git a/swift/proxy/server.py b/swift/proxy/server.py
index ae7ee8ee26..6b3e3c9bc5 100644
--- a/swift/proxy/server.py
+++ b/swift/proxy/server.py
@@ -480,7 +480,8 @@ class Application(object):
                                       body='Invalid Content-Length')
 
             try:
-                if not check_utf8(wsgi_to_str(req.path_info)):
+                if not check_utf8(wsgi_to_str(req.path_info),
+                                  internal=req.allow_reserved_names):
                     self.logger.increment('errors')
                     return HTTPPreconditionFailed(
                         request=req, body='Invalid UTF8 or contains NULL')
diff --git a/test/probe/common.py b/test/probe/common.py
index 6392b00163..9652c163bc 100644
--- a/test/probe/common.py
+++ b/test/probe/common.py
@@ -487,6 +487,7 @@ class ProbeTest(unittest.TestCase):
 
             [app:proxy-server]
             use = egg:swift#proxy
+            allow_account_management = True
 
             [filter:copy]
             use = egg:swift#copy
diff --git a/test/probe/test_account_reaper.py b/test/probe/test_account_reaper.py
index 6530205faf..367e8a761c 100644
--- a/test/probe/test_account_reaper.py
+++ b/test/probe/test_account_reaper.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from io import BytesIO
 from time import sleep
 import uuid
 import unittest
@@ -23,6 +24,7 @@ from swift.common import utils
 from swift.common.manager import Manager
 from swift.common.direct_client import direct_delete_account, \
     direct_get_object, direct_head_container, ClientException
+from swift.common.request_helpers import get_reserved_name
 from test.probe.common import ReplProbeTest, ENABLED_POLICIES
 
 
@@ -30,8 +32,9 @@ class TestAccountReaper(ReplProbeTest):
     def setUp(self):
         super(TestAccountReaper, self).setUp()
         self.all_objects = []
+        int_client = self.make_internal_client()
         # upload some containers
-        body = 'test-body'
+        body = b'test-body'
         for policy in ENABLED_POLICIES:
             container = 'container-%s-%s' % (policy.name, uuid.uuid4())
             client.put_container(self.url, self.token, container,
@@ -39,6 +42,18 @@ class TestAccountReaper(ReplProbeTest):
             obj = 'object-%s' % uuid.uuid4()
             client.put_object(self.url, self.token, container, obj, body)
             self.all_objects.append((policy, container, obj))
+
+            # Also create some reserved names
+            container = get_reserved_name(
+                'reserved', policy.name, str(uuid.uuid4()))
+            int_client.create_container(
+                self.account, container,
+                headers={'X-Storage-Policy': policy.name})
+            obj = get_reserved_name('object', str(uuid.uuid4()))
+            int_client.upload_object(
+                BytesIO(body), self.account, container, obj)
+            self.all_objects.append((policy, container, obj))
+
             policy.load_ring('/etc/swift')
 
         Manager(['container-updater']).once()
@@ -46,11 +61,11 @@ class TestAccountReaper(ReplProbeTest):
         headers = client.head_account(self.url, self.token)
 
         self.assertEqual(int(headers['x-account-container-count']),
-                         len(ENABLED_POLICIES))
+                         len(self.all_objects))
         self.assertEqual(int(headers['x-account-object-count']),
-                         len(ENABLED_POLICIES))
+                         len(self.all_objects))
         self.assertEqual(int(headers['x-account-bytes-used']),
-                         len(ENABLED_POLICIES) * len(body))
+                         len(self.all_objects) * len(body))
 
         part, nodes = self.account_ring.get_nodes(self.account)
 
diff --git a/test/probe/test_replication_servers_working.py b/test/probe/test_replication_servers_working.py
index 1052d5ba7c..fae77b742f 100644
--- a/test/probe/test_replication_servers_working.py
+++ b/test/probe/test_replication_servers_working.py
@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from io import BytesIO
 from unittest import main
 from uuid import uuid4
 import os
@@ -25,6 +26,7 @@ from swiftclient import client
 from swift.obj.diskfile import get_data_dir
 
 from test.probe.common import ReplProbeTest
+from swift.common.request_helpers import get_reserved_name
 from swift.common.utils import readconf
 
 EXCLUDE_FILES = re.compile('^(hashes\.(pkl|invalid)|lock(-\d+)?)$')
@@ -80,6 +82,15 @@ class TestReplicatorFunctions(ReplProbeTest):
     different port values.
     """
 
+    def put_data(self):
+        container = 'container-%s' % uuid4()
+        client.put_container(self.url, self.token, container,
+                             headers={'X-Storage-Policy':
+                                      self.policy.name})
+
+        obj = 'object-%s' % uuid4()
+        client.put_object(self.url, self.token, container, obj, 'VERIFY')
+
     def test_main(self):
         # Create one account, container and object file.
         # Find node with account, container and object replicas.
@@ -102,13 +113,7 @@ class TestReplicatorFunctions(ReplProbeTest):
             path_list.append(os.path.join(device_path, device))
 
         # Put data to storage nodes
-        container = 'container-%s' % uuid4()
-        client.put_container(self.url, self.token, container,
-                             headers={'X-Storage-Policy':
-                                      self.policy.name})
-
-        obj = 'object-%s' % uuid4()
-        client.put_object(self.url, self.token, container, obj, 'VERIFY')
+        self.put_data()
 
         # Get all data file information
         (files_list, dir_list) = collect_info(path_list)
@@ -200,5 +205,19 @@ class TestReplicatorFunctions(ReplProbeTest):
             self.replicators.stop()
 
 
+class TestReplicatorFunctionsReservedNames(TestReplicatorFunctions):
+    def put_data(self):
+        int_client = self.make_internal_client()
+        int_client.create_account(self.account)
+        container = get_reserved_name('container', str(uuid4()))
+        int_client.create_container(self.account, container,
+                                    headers={'X-Storage-Policy':
+                                             self.policy.name})
+
+        obj = get_reserved_name('object', str(uuid4()))
+        int_client.upload_object(
+            BytesIO(b'VERIFY'), self.account, container, obj)
+
+
 if __name__ == '__main__':
     main()
diff --git a/test/probe/test_reserved_name.py b/test/probe/test_reserved_name.py
new file mode 100644
index 0000000000..bce0095058
--- /dev/null
+++ b/test/probe/test_reserved_name.py
@@ -0,0 +1,131 @@
+#!/usr/bin/python -u
+# Copyright (c) 2019 OpenStack Foundation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import unittest
+
+from io import BytesIO
+from uuid import uuid4
+
+from swift.common.request_helpers import get_reserved_name
+
+from test.probe.common import ReplProbeTest
+
+from swiftclient import client, ClientException
+
+
+class TestReservedNames(ReplProbeTest):
+
+    def test_simple_crud(self):
+        int_client = self.make_internal_client()
+
+        # Create reserve named container
+        user_cont = 'container-%s' % uuid4()
+        reserved_cont = get_reserved_name('container-%s' % uuid4())
+        client.put_container(self.url, self.token, user_cont)
+        int_client.create_container(self.account, reserved_cont)
+
+        # Check that we can list both reserved and non-reserved containers
+        self.assertEqual([reserved_cont, user_cont], [
+            c['name'] for c in int_client.iter_containers(self.account)])
+
+        # sanity, user can't get to reserved name
+        with self.assertRaises(ClientException) as cm:
+            client.head_container(self.url, self.token, reserved_cont)
+        self.assertEqual(412, cm.exception.http_status)
+
+        user_obj = 'obj-%s' % uuid4()
+        reserved_obj = get_reserved_name('obj-%s' % uuid4())
+
+        # InternalClient can write & read reserved names fine
+        int_client.upload_object(
+            BytesIO(b'data'), self.account, reserved_cont, reserved_obj)
+        int_client.get_object_metadata(
+            self.account, reserved_cont, reserved_obj)
+        _, _, app_iter = int_client.get_object(
+            self.account, reserved_cont, reserved_obj)
+        self.assertEqual(b''.join(app_iter), b'data')
+        self.assertEqual([reserved_obj], [
+            o['name']
+            for o in int_client.iter_objects(self.account, reserved_cont)])
+
+        # But reserved objects must be in reserved containers, and
+        # user objects must be in user containers (at least for now)
+        int_client.upload_object(
+            BytesIO(b'data'), self.account, reserved_cont, user_obj,
+            acceptable_statuses=(400,))
+
+        int_client.upload_object(
+            BytesIO(b'data'), self.account, user_cont, reserved_obj,
+            acceptable_statuses=(400,))
+
+        # Make sure we can clean up, too
+        int_client.delete_object(self.account, reserved_cont, reserved_obj)
+        int_client.delete_container(self.account, reserved_cont)
+
+    def test_symlink_target(self):
+        if 'symlink' not in self.cluster_info:
+            raise unittest.SkipTest(
+                "Symlink not enabled in proxy; can't test "
+                "symlink to reserved name")
+        int_client = self.make_internal_client()
+
+        # create link container first, ensure account gets created too
+        client.put_container(self.url, self.token, 'c1')
+
+        # Create reserve named container
+        tgt_cont = get_reserved_name('container-%s' % uuid4())
+        int_client.create_container(self.account, tgt_cont)
+
+        # sanity, user can't get to reserved name
+        with self.assertRaises(ClientException) as cm:
+            client.head_container(self.url, self.token, tgt_cont)
+        self.assertEqual(412, cm.exception.http_status)
+
+        tgt_obj = get_reserved_name('obj-%s' % uuid4())
+        int_client.upload_object(
+            BytesIO(b'target object'), self.account, tgt_cont, tgt_obj)
+        metadata = int_client.get_object_metadata(
+            self.account, tgt_cont, tgt_obj)
+        etag = metadata['etag']
+
+        # users can write a dynamic symlink that targets a reserved
+        # name object
+        client.put_object(
+            self.url, self.token, 'c1', 'symlink',
+            headers={
+                'X-Symlink-Target': '%s/%s' % (tgt_cont, tgt_obj),
+                'Content-Type': 'application/symlink',
+            })
+
+        # but can't read the symlink
+        with self.assertRaises(ClientException) as cm:
+            client.get_object(self.url, self.token, 'c1', 'symlink')
+        self.assertEqual(412, cm.exception.http_status)
+
+        # user's can't create static symlink to reserved name
+        with self.assertRaises(ClientException) as cm:
+            client.put_object(
+                self.url, self.token, 'c1', 'static-symlink',
+                headers={
+                    'X-Symlink-Target': '%s/%s' % (tgt_cont, tgt_obj),
+                    'X-Symlink-Target-Etag': etag,
+                    'Content-Type': 'application/symlink',
+                })
+        self.assertEqual(412, cm.exception.http_status)
+
+        # clean-up
+        client.delete_object(self.url, self.token, 'c1', 'symlink')
+        int_client.delete_object(self.account, tgt_cont, tgt_obj)
+        int_client.delete_container(self.account, tgt_cont)
diff --git a/test/unit/account/test_backend.py b/test/unit/account/test_backend.py
index 82e1a77075..15422bd139 100644
--- a/test/unit/account/test_backend.py
+++ b/test/unit/account/test_backend.py
@@ -38,6 +38,7 @@ from swift.account.backend import AccountBroker
 from swift.common.utils import Timestamp
 from test.unit import patch_policies, with_tempdir, make_timestamp_iter
 from swift.common.db import DatabaseConnectionError
+from swift.common.request_helpers import get_reserved_name
 from swift.common.storage_policy import StoragePolicy, POLICIES
 
 from test.unit.common import test_db
@@ -545,6 +546,35 @@ class TestAccountBroker(unittest.TestCase):
         self.assertEqual([row[0] for row in listing],
                          ['c10', 'c1'])
 
+    def test_list_container_iter_with_reserved_name(self):
+        # Test ContainerBroker.list_objects_iter
+        broker = AccountBroker(':memory:', account='a')
+        broker.initialize(next(self.ts).internal, 0)
+
+        broker.put_container(
+            'foo', next(self.ts).internal, 0, 0, 0, POLICIES.default.idx)
+        broker.put_container(
+            get_reserved_name('foo'), next(self.ts).internal, 0, 0, 0,
+            POLICIES.default.idx)
+
+        listing = broker.list_containers_iter(100, None, None, '', '')
+        self.assertEqual([row[0] for row in listing], ['foo'])
+
+        listing = broker.list_containers_iter(100, None, None, '', '',
+                                              reverse=True)
+        self.assertEqual([row[0] for row in listing], ['foo'])
+
+        listing = broker.list_containers_iter(100, None, None, '', '',
+                                              allow_reserved=True)
+        self.assertEqual([row[0] for row in listing],
+                         [get_reserved_name('foo'), 'foo'])
+
+        listing = broker.list_containers_iter(100, None, None, '', '',
+                                              reverse=True,
+                                              allow_reserved=True)
+        self.assertEqual([row[0] for row in listing],
+                         ['foo', get_reserved_name('foo')])
+
     def test_reverse_prefix_delim(self):
         expectations = [
             {
diff --git a/test/unit/account/test_reaper.py b/test/unit/account/test_reaper.py
index ebb5407f7f..d8b76eb9b2 100644
--- a/test/unit/account/test_reaper.py
+++ b/test/unit/account/test_reaper.py
@@ -50,7 +50,12 @@ class FakeAccountBroker(object):
                 'delete_timestamp': time.time() - 10}
         return info
 
-    def list_containers_iter(self, limit, marker, *args):
+    def list_containers_iter(self, limit, marker, *args, **kwargs):
+        if not kwargs.pop('allow_reserved'):
+            raise RuntimeError('Expected allow_reserved to be True!')
+        if kwargs:
+            raise RuntimeError('Got unexpected keyword arguments: %r' % (
+                kwargs, ))
         for cont in self.containers:
             if cont > marker:
                 yield cont, None, None, None, None
@@ -710,11 +715,16 @@ class TestReaper(unittest.TestCase):
         devices = self.prepare_data_dir()
         self.called_amount = 0
         conf = {'devices': devices}
-        r = self.init_reaper(conf, myips=['10.10.10.2'])
+        r = self.init_reaper(conf, myips=['10.10.10.2'], fakelogger=True)
 
         container_reaped = [0]
 
-        def fake_list_containers_iter(self, *args):
+        def fake_list_containers_iter(self, *args, **kwargs):
+            if not kwargs.pop('allow_reserved'):
+                raise RuntimeError('Expected allow_reserved to be True!')
+            if kwargs:
+                raise RuntimeError('Got unexpected keyword arguments: %r' % (
+                    kwargs, ))
             for container in self.containers:
                 if container in self.containers_yielded:
                     continue
diff --git a/test/unit/account/test_server.py b/test/unit/account/test_server.py
index 9bee9b5c6b..c1168a852f 100644
--- a/test/unit/account/test_server.py
+++ b/test/unit/account/test_server.py
@@ -27,17 +27,19 @@ from io import BytesIO
 
 import json
 from six import StringIO
+from six.moves.urllib.parse import quote
 import xml.dom.minidom
 
 from swift import __version__ as swift_version
 from swift.common.swob import (Request, WsgiBytesIO, HTTPNoContent)
-from swift.common import constraints
+from swift.common.constraints import ACCOUNT_LISTING_LIMIT
 from swift.account.backend import AccountBroker
 from swift.account.server import AccountController
 from swift.common.utils import (normalize_timestamp, replication, public,
                                 mkdirs, storage_directory, Timestamp)
-from swift.common.request_helpers import get_sys_meta_prefix
-from test.unit import patch_policies, debug_logger, mock_check_drive
+from swift.common.request_helpers import get_sys_meta_prefix, get_reserved_name
+from test.unit import patch_policies, debug_logger, mock_check_drive, \
+    make_timestamp_iter
 from swift.common.storage_policy import StoragePolicy, POLICIES
 
 
@@ -52,6 +54,7 @@ class TestAccountController(unittest.TestCase):
         self.controller = AccountController(
             {'devices': self.testdir, 'mount_check': 'false'},
             logger=debug_logger())
+        self.ts = make_timestamp_iter()
 
     def tearDown(self):
         """Tear down for testing swift.account.server.AccountController"""
@@ -501,6 +504,65 @@ class TestAccountController(unittest.TestCase):
         self.assertEqual(resp.body, b'Recently deleted')
         self.assertEqual(resp.headers['X-Account-Status'], 'Deleted')
 
+    def test_create_reserved_namespace_account(self):
+        path = '/sda1/p/%s' % get_reserved_name('a')
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status, '201 Created')
+
+        path = '/sda1/p/%s' % get_reserved_name('foo', 'bar')
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status, '201 Created')
+
+    def test_create_invalid_reserved_namespace_account(self):
+        account_name = get_reserved_name('foo', 'bar')[1:]
+        path = '/sda1/p/%s' % account_name
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status, '400 Bad Request')
+
+    def test_create_reserved_container_in_account(self):
+        # create account
+        path = '/sda1/p/a'
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 201)
+        # put null container in it
+        path += '/%s' % get_reserved_name('c', 'stuff')
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal,
+            'X-Put-Timestamp': next(self.ts).internal,
+            'X-Delete-Timestamp': 0,
+            'X-Object-Count': 0,
+            'X-Bytes-Used': 0,
+        })
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status, '201 Created')
+
+    def test_create_invalid_reserved_container_in_account(self):
+        # create account
+        path = '/sda1/p/a'
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 201)
+        # put invalid container in it
+        path += '/%s' % get_reserved_name('c', 'stuff')[1:]
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal,
+            'X-Put-Timestamp': next(self.ts).internal,
+            'X-Delete-Timestamp': 0,
+            'X-Object-Count': 0,
+            'X-Bytes-Used': 0,
+        })
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status, '400 Bad Request')
+
     def test_PUT_non_utf8_metadata(self):
         # Set metadata header
         req = Request.blank(
@@ -706,7 +768,7 @@ class TestAccountController(unittest.TestCase):
             headers={'X-Timestamp': normalize_timestamp(1),
                      'X-Account-Meta-Test': 'Value'})
         resp = req.get_response(self.controller)
-        self.assertEqual(resp.status_int, 204)
+        self.assertEqual(resp.status_int, 204, resp.body)
         req = Request.blank('/sda1/p/a', environ={'REQUEST_METHOD': 'HEAD'})
         resp = req.get_response(self.controller)
         self.assertEqual(resp.status_int, 204)
@@ -935,7 +997,7 @@ class TestAccountController(unittest.TestCase):
 
     def test_GET_over_limit(self):
         req = Request.blank(
-            '/sda1/p/a?limit=%d' % (constraints.ACCOUNT_LISTING_LIMIT + 1),
+            '/sda1/p/a?limit=%d' % (ACCOUNT_LISTING_LIMIT + 1),
             environ={'REQUEST_METHOD': 'GET'})
         resp = req.get_response(self.controller)
         self.assertEqual(resp.status_int, 412)
@@ -1752,6 +1814,382 @@ class TestAccountController(unittest.TestCase):
              for item in json.loads(resp.body)],
             [{"name": "US~~UT~~~B"}])
 
+    def _expected_listing(self, containers):
+        return [dict(
+            last_modified=c['timestamp'].isoformat, **{
+                k: v for k, v in c.items()
+                if k != 'timestamp'
+            }) for c in sorted(containers, key=lambda c: c['name'])]
+
+    def _report_containers(self, containers, account='a'):
+        req = Request.blank('/sda1/p/%s' % account, method='PUT', headers={
+            'x-timestamp': next(self.ts).internal})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int // 100, 2, resp.body)
+        for container in containers:
+            path = '/sda1/p/%s/%s' % (account, container['name'])
+            req = Request.blank(path, method='PUT', headers={
+                'X-Put-Timestamp': container['timestamp'].internal,
+                'X-Delete-Timestamp': container.get(
+                    'deleted', Timestamp(0)).internal,
+                'X-Object-Count': container['count'],
+                'X-Bytes-Used': container['bytes'],
+            })
+            resp = req.get_response(self.controller)
+            self.assertEqual(resp.status_int // 100, 2, resp.body)
+
+    def test_delimiter_with_reserved_and_no_public(self):
+        containers = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 200,
+            'count': 2,
+            'timestamp': next(self.ts),
+        }]
+        self._report_containers(containers)
+
+        req = Request.blank('/sda1/p/a', headers={
+            'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank('/sda1/p/a', headers={
+            'X-Backend-Allow-Reserved-Names': 'true',
+            'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers))
+
+        req = Request.blank('/sda1/p/a?prefix=%s&delimiter=l' %
+                            get_reserved_name('nul'), headers={
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank('/sda1/p/a?prefix=%s&delimiter=l' %
+                            get_reserved_name('nul'), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [{
+            'subdir': '%s' % get_reserved_name('null')}])
+
+    def test_delimiter_with_reserved_and_public(self):
+        containers = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 200,
+            'count': 2,
+            'timestamp': next(self.ts),
+        }, {
+            'name': 'nullish',
+            'bytes': 10,
+            'count': 10,
+            'timestamp': next(self.ts),
+        }]
+        self._report_containers(containers)
+
+        req = Request.blank('/sda1/p/a?prefix=nul&delimiter=l', headers={
+            'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [{'subdir': 'null'}])
+
+        # allow-reserved header doesn't really make a difference
+        req = Request.blank('/sda1/p/a?prefix=nul&delimiter=l', headers={
+            'X-Backend-Allow-Reserved-Names': 'true',
+            'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [{'subdir': 'null'}])
+
+        req = Request.blank('/sda1/p/a?prefix=%s&delimiter=l' %
+                            get_reserved_name('nul'), headers={
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank('/sda1/p/a?prefix=%s&delimiter=l' %
+                            get_reserved_name('nul'), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [{
+            'subdir': '%s' % get_reserved_name('null')}])
+
+        req = Request.blank('/sda1/p/a?delimiter=%00', headers={
+                            'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers)[1:])
+
+        req = Request.blank('/sda1/p/a?delimiter=%00', headers={
+                            'X-Backend-Allow-Reserved-Names': 'true',
+                            'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         [{'subdir': '\x00'}] +
+                         self._expected_listing(containers)[1:])
+
+    def test_markers_with_reserved(self):
+        containers = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 200,
+            'count': 2,
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'test02'),
+            'bytes': 10,
+            'count': 10,
+            'timestamp': next(self.ts),
+        }]
+        self._report_containers(containers)
+
+        req = Request.blank('/sda1/p/a?marker=%s' %
+                            get_reserved_name('null', ''), headers={
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank('/sda1/p/a?marker=%s' %
+                            get_reserved_name('null', ''), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers))
+
+        req = Request.blank('/sda1/p/a?marker=%s' % quote(
+            self._expected_listing(containers)[0]['name']), headers={
+                'X-Backend-Allow-Reserved-Names': 'true',
+                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers)[1:])
+
+        containers.append({
+            'name': get_reserved_name('null', 'test03'),
+            'bytes': 300,
+            'count': 30,
+            'timestamp': next(self.ts),
+        })
+        self._report_containers(containers)
+
+        req = Request.blank('/sda1/p/a?marker=%s' % quote(
+            self._expected_listing(containers)[0]['name']), headers={
+                'X-Backend-Allow-Reserved-Names': 'true',
+                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers)[1:])
+
+        req = Request.blank('/sda1/p/a?marker=%s' % quote(
+            self._expected_listing(containers)[1]['name']), headers={
+                'X-Backend-Allow-Reserved-Names': 'true',
+                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers)[-1:])
+
+    def test_prefix_with_reserved(self):
+        containers = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 200,
+            'count': 2,
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'test02'),
+            'bytes': 10,
+            'count': 10,
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'foo'),
+            'bytes': 10,
+            'count': 10,
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('nullish'),
+            'bytes': 300,
+            'count': 32,
+            'timestamp': next(self.ts),
+        }]
+        self._report_containers(containers)
+
+        req = Request.blank('/sda1/p/a?prefix=%s' %
+                            get_reserved_name('null', 'test'), headers={
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank('/sda1/p/a?prefix=%s' %
+                            get_reserved_name('null', 'test'), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers[:2]))
+
+    def test_prefix_and_delim_with_reserved(self):
+        containers = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 200,
+            'count': 2,
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'test02'),
+            'bytes': 10,
+            'count': 10,
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'foo'),
+            'bytes': 10,
+            'count': 10,
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('nullish'),
+            'bytes': 300,
+            'count': 32,
+            'timestamp': next(self.ts),
+        }]
+        self._report_containers(containers)
+
+        req = Request.blank('/sda1/p/a?prefix=%s&delimiter=%s' % (
+            get_reserved_name('null'), get_reserved_name()), headers={
+                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank('/sda1/p/a?prefix=%s&delimiter=%s' % (
+            get_reserved_name('null'), get_reserved_name()), headers={
+                'X-Backend-Allow-Reserved-Names': 'true',
+                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        expected = [{'subdir': get_reserved_name('null', '')}] + \
+            self._expected_listing(containers[-1:])
+        self.assertEqual(json.loads(resp.body), expected)
+
+    def test_reserved_markers_with_non_reserved(self):
+        containers = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 200,
+            'count': 2,
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'test02'),
+            'bytes': 10,
+            'count': 10,
+            'timestamp': next(self.ts),
+        }, {
+            'name': 'nullish',
+            'bytes': 300,
+            'count': 32,
+            'timestamp': next(self.ts),
+        }]
+        self._report_containers(containers)
+
+        req = Request.blank('/sda1/p/a?marker=%s' %
+                            get_reserved_name('null', ''), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers))
+
+        req = Request.blank('/sda1/p/a?marker=%s' %
+                            get_reserved_name('null', ''), headers={
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         [c for c in self._expected_listing(containers)
+                          if get_reserved_name() not in c['name']])
+
+        req = Request.blank('/sda1/p/a?marker=%s' %
+                            get_reserved_name('null', ''), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers))
+
+        req = Request.blank('/sda1/p/a?marker=%s' % quote(
+            self._expected_listing(containers)[0]['name']), headers={
+                'X-Backend-Allow-Reserved-Names': 'true',
+                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers)[1:])
+
+    def test_null_markers(self):
+        containers = [{
+            'name': get_reserved_name('null', ''),
+            'bytes': 200,
+            'count': 2,
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 200,
+            'count': 2,
+            'timestamp': next(self.ts),
+        }, {
+            'name': 'null',
+            'bytes': 300,
+            'count': 32,
+            'timestamp': next(self.ts),
+        }]
+        self._report_containers(containers)
+
+        req = Request.blank('/sda1/p/a?marker=%s' % get_reserved_name('null'),
+                            headers={'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers)[-1:])
+
+        req = Request.blank('/sda1/p/a?marker=%s' % get_reserved_name('null'),
+                            headers={'X-Backend-Allow-Reserved-Names': 'true',
+                                     'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers))
+
+        req = Request.blank('/sda1/p/a?marker=%s' %
+                            get_reserved_name('null', ''), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers)[1:])
+
+        req = Request.blank('/sda1/p/a?marker=%s' %
+                            get_reserved_name('null', 'test00'), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(containers)[1:])
+
     def test_through_call(self):
         inbuf = BytesIO()
         errbuf = StringIO()
@@ -1814,7 +2252,7 @@ class TestAccountController(unittest.TestCase):
 
         self.controller.__call__({'REQUEST_METHOD': 'GET',
                                   'SCRIPT_NAME': '',
-                                  'PATH_INFO': '\x00',
+                                  'PATH_INFO': '/sda1/p/a/c\xd8\x3e%20',
                                   'SERVER_NAME': '127.0.0.1',
                                   'SERVER_PORT': '8080',
                                   'SERVER_PROTOCOL': 'HTTP/1.0',
diff --git a/test/unit/account/test_utils.py b/test/unit/account/test_utils.py
index f931ad58bd..eb332395a7 100644
--- a/test/unit/account/test_utils.py
+++ b/test/unit/account/test_utils.py
@@ -14,15 +14,18 @@
 import itertools
 import time
 import unittest
+import json
 
 import mock
 
 from swift.account import utils, backend
-from swift.common.storage_policy import POLICIES
+from swift.common.storage_policy import POLICIES, StoragePolicy
+from swift.common.swob import Request
 from swift.common.utils import Timestamp
 from swift.common.header_key_dict import HeaderKeyDict
+from swift.common.request_helpers import get_reserved_name
 
-from test.unit import patch_policies
+from test.unit import patch_policies, make_timestamp_iter
 
 
 class TestFakeAccountBroker(unittest.TestCase):
@@ -57,6 +60,9 @@ class TestFakeAccountBroker(unittest.TestCase):
 
 class TestAccountUtils(unittest.TestCase):
 
+    def setUp(self):
+        self.ts = make_timestamp_iter()
+
     def test_get_response_headers_fake_broker(self):
         broker = utils.FakeAccountBroker()
         now = time.time()
@@ -187,3 +193,77 @@ class TestAccountUtils(unittest.TestCase):
                              'value for %r was %r not %r' % (
                                  key, value, expected_value))
         self.assertFalse(expected)
+
+    def test_account_listing_response(self):
+        req = Request.blank('')
+        now = time.time()
+        with mock.patch('time.time', new=lambda: now):
+            resp = utils.account_listing_response('a', req, 'text/plain')
+        self.assertEqual(resp.status_int, 204)
+        expected = HeaderKeyDict({
+            'Content-Type': 'text/plain; charset=utf-8',
+            'X-Account-Container-Count': 0,
+            'X-Account-Object-Count': 0,
+            'X-Account-Bytes-Used': 0,
+            'X-Timestamp': Timestamp(now).normal,
+            'X-PUT-Timestamp': Timestamp(now).normal,
+        })
+        self.assertEqual(expected, resp.headers)
+        self.assertEqual(b'', resp.body)
+
+    @patch_policies([StoragePolicy(0, 'zero', is_default=True)])
+    def test_account_listing_reserved_names(self):
+        broker = backend.AccountBroker(':memory:', account='a')
+        put_timestamp = next(self.ts)
+        now = time.time()
+        with mock.patch('time.time', new=lambda: now):
+            broker.initialize(put_timestamp.internal)
+        container_timestamp = next(self.ts)
+        broker.put_container(get_reserved_name('foo'),
+                             container_timestamp.internal, 0, 10, 100, 0)
+
+        req = Request.blank('')
+        resp = utils.account_listing_response(
+            'a', req, 'application/json', broker)
+        self.assertEqual(resp.status_int, 200)
+        expected = HeaderKeyDict({
+            'Content-Type': 'application/json; charset=utf-8',
+            'Content-Length': 2,
+            'X-Account-Container-Count': 1,
+            'X-Account-Object-Count': 10,
+            'X-Account-Bytes-Used': 100,
+            'X-Timestamp': Timestamp(now).normal,
+            'X-PUT-Timestamp': put_timestamp.normal,
+            'X-Account-Storage-Policy-Zero-Container-Count': 1,
+            'X-Account-Storage-Policy-Zero-Object-Count': 10,
+            'X-Account-Storage-Policy-Zero-Bytes-Used': 100,
+        })
+        self.assertEqual(expected, resp.headers)
+        self.assertEqual(b'[]', resp.body)
+
+        req = Request.blank('', headers={
+            'X-Backend-Allow-Reserved-Names': 'true'})
+        resp = utils.account_listing_response(
+            'a', req, 'application/json', broker)
+        self.assertEqual(resp.status_int, 200)
+        expected = HeaderKeyDict({
+            'Content-Type': 'application/json; charset=utf-8',
+            'Content-Length': 97,
+            'X-Account-Container-Count': 1,
+            'X-Account-Object-Count': 10,
+            'X-Account-Bytes-Used': 100,
+            'X-Timestamp': Timestamp(now).normal,
+            'X-PUT-Timestamp': put_timestamp.normal,
+            'X-Account-Storage-Policy-Zero-Container-Count': 1,
+            'X-Account-Storage-Policy-Zero-Object-Count': 10,
+            'X-Account-Storage-Policy-Zero-Bytes-Used': 100,
+        })
+        self.assertEqual(expected, resp.headers)
+        expected = [{
+            "last_modified": container_timestamp.isoformat,
+            "count": 10,
+            "bytes": 100,
+            "name": get_reserved_name('foo'),
+        }]
+        self.assertEqual(sorted(json.dumps(expected).encode('ascii')),
+                         sorted(resp.body))
diff --git a/test/unit/common/middleware/test_gatekeeper.py b/test/unit/common/middleware/test_gatekeeper.py
index 0fac175dbf..4f8cb480b7 100644
--- a/test/unit/common/middleware/test_gatekeeper.py
+++ b/test/unit/common/middleware/test_gatekeeper.py
@@ -245,6 +245,39 @@ class TestGatekeeper(unittest.TestCase):
         self._test_location_header('/v/a/c/o2?query=path#test')
         self._test_location_header('/v/a/c/o2;whatisparam?query=path#test')
 
+    def test_allow_reserved_names(self):
+        fake_app = FakeApp()
+        app = self.get_app(fake_app, {})
+        headers = {
+            'X-Allow-Reserved-Names': 'some-value'
+        }
+
+        req = Request.blank('/v/a/c/o', method='GET', headers=headers)
+        resp = req.get_response(app)
+        self.assertEqual('200 OK', resp.status)
+        self.assertNotIn('X-Backend-Allow-Reserved-Names',
+                         fake_app.req.headers)
+        self.assertIn('X-Allow-Reserved-Names',
+                      fake_app.req.headers)
+        self.assertEqual(
+            'some-value',
+            fake_app.req.headers['X-Allow-Reserved-Names'])
+
+        app.allow_reserved_names_header = True
+        req = Request.blank('/v/a/c/o', method='GET', headers=headers)
+        resp = req.get_response(app)
+        self.assertEqual('200 OK', resp.status)
+        self.assertIn('X-Backend-Allow-Reserved-Names',
+                      fake_app.req.headers)
+        self.assertEqual(
+            'some-value',
+            fake_app.req.headers['X-Backend-Allow-Reserved-Names'])
+        self.assertEqual(
+            'some-value',
+            req.headers['X-Backend-Allow-Reserved-Names'])
+        self.assertNotIn('X-Allow-Reserved-Names', fake_app.req.headers)
+        self.assertNotIn('X-Allow-Reserved-Names', req.headers)
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/test/unit/common/middleware/test_listing_formats.py b/test/unit/common/middleware/test_listing_formats.py
index 6ac6d08735..22cf3e55f7 100644
--- a/test/unit/common/middleware/test_listing_formats.py
+++ b/test/unit/common/middleware/test_listing_formats.py
@@ -18,13 +18,17 @@ import unittest
 
 from swift.common.swob import Request, HTTPOk
 from swift.common.middleware import listing_formats
+from swift.common.request_helpers import get_reserved_name
+from test.unit import debug_logger
 from test.unit.common.middleware.helpers import FakeSwift
 
 
 class TestListingFormats(unittest.TestCase):
     def setUp(self):
         self.fake_swift = FakeSwift()
-        self.app = listing_formats.ListingFilter(self.fake_swift)
+        self.logger = debug_logger('test-listing')
+        self.app = listing_formats.ListingFilter(self.fake_swift, {},
+                                                 logger=self.logger)
         self.fake_account_listing = json.dumps([
             {'name': 'bar', 'bytes': 0, 'count': 0,
              'last_modified': '1970-01-01T00:00:00.000000'},
@@ -37,6 +41,25 @@ class TestListingFormats(unittest.TestCase):
             {'subdir': 'foo/'},
         ]).encode('ascii')
 
+        self.fake_account_listing_with_reserved = json.dumps([
+            {'name': 'bar', 'bytes': 0, 'count': 0,
+             'last_modified': '1970-01-01T00:00:00.000000'},
+            {'name': get_reserved_name('bar', 'versions'), 'bytes': 0,
+             'count': 0, 'last_modified': '1970-01-01T00:00:00.000000'},
+            {'subdir': 'foo_'},
+            {'subdir': get_reserved_name('foo_')},
+        ]).encode('ascii')
+        self.fake_container_listing_with_reserved = json.dumps([
+            {'name': 'bar', 'hash': 'etag', 'bytes': 0,
+             'content_type': 'text/plain',
+             'last_modified': '1970-01-01T00:00:00.000000'},
+            {'name': get_reserved_name('bar', 'extra data'), 'hash': 'etag',
+             'bytes': 0, 'content_type': 'text/plain',
+             'last_modified': '1970-01-01T00:00:00.000000'},
+            {'subdir': 'foo/'},
+            {'subdir': get_reserved_name('foo/')},
+        ]).encode('ascii')
+
     def test_valid_account(self):
         self.fake_swift.register('GET', '/v1/a', HTTPOk, {
             'Content-Length': str(len(self.fake_account_listing)),
@@ -60,7 +83,8 @@ class TestListingFormats(unittest.TestCase):
 
         req = Request.blank('/v1/a?format=json')
         resp = req.get_response(self.app)
-        self.assertEqual(resp.body, self.fake_account_listing)
+        self.assertEqual(json.loads(resp.body),
+                         json.loads(self.fake_account_listing))
         self.assertEqual(resp.headers['Content-Type'],
                          'application/json; charset=utf-8')
         self.assertEqual(self.fake_swift.calls[-1], (
@@ -82,6 +106,119 @@ class TestListingFormats(unittest.TestCase):
         self.assertEqual(self.fake_swift.calls[-1], (
             'GET', '/v1/a?format=json'))
 
+    def test_valid_account_with_reserved(self):
+        body_len = len(self.fake_account_listing_with_reserved)
+        self.fake_swift.register(
+            'GET', '/v1/a\xe2\x98\x83', HTTPOk, {
+                'Content-Length': str(body_len),
+                'Content-Type': 'application/json',
+            }, self.fake_account_listing_with_reserved)
+
+        req = Request.blank('/v1/a\xe2\x98\x83')
+        resp = req.get_response(self.app)
+        self.assertEqual(resp.body, b'bar\nfoo_\n')
+        self.assertEqual(resp.headers['Content-Type'],
+                         'text/plain; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', '/v1/a\xe2\x98\x83?format=json'))
+        self.assertEqual(self.logger.get_lines_for_level('warning'), [
+            "Account listing for a%E2%98%83 had reserved byte in name: "
+            "'\\x00bar\\x00versions'",
+            "Account listing for a%E2%98%83 had reserved byte in subdir: "
+            "'\\x00foo_'",
+        ])
+
+        req = Request.blank('/v1/a\xe2\x98\x83', headers={
+            'X-Backend-Allow-Reserved-Names': 'true'})
+        resp = req.get_response(self.app)
+        self.assertEqual(resp.body, b'bar\n%s\nfoo_\n%s\n' % (
+            get_reserved_name('bar', 'versions').encode('ascii'),
+            get_reserved_name('foo_').encode('ascii'),
+        ))
+        self.assertEqual(resp.headers['Content-Type'],
+                         'text/plain; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', '/v1/a\xe2\x98\x83?format=json'))
+
+        req = Request.blank('/v1/a\xe2\x98\x83?format=txt')
+        resp = req.get_response(self.app)
+        self.assertEqual(resp.body, b'bar\nfoo_\n')
+        self.assertEqual(resp.headers['Content-Type'],
+                         'text/plain; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', '/v1/a\xe2\x98\x83?format=json'))
+
+        req = Request.blank('/v1/a\xe2\x98\x83?format=txt', headers={
+            'X-Backend-Allow-Reserved-Names': 'true'})
+        resp = req.get_response(self.app)
+        self.assertEqual(resp.body, b'bar\n%s\nfoo_\n%s\n' % (
+            get_reserved_name('bar', 'versions').encode('ascii'),
+            get_reserved_name('foo_').encode('ascii'),
+        ))
+        self.assertEqual(resp.headers['Content-Type'],
+                         'text/plain; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', '/v1/a\xe2\x98\x83?format=json'))
+
+        req = Request.blank('/v1/a\xe2\x98\x83?format=json')
+        resp = req.get_response(self.app)
+        self.assertEqual(json.loads(resp.body),
+                         json.loads(self.fake_account_listing))
+        self.assertEqual(resp.headers['Content-Type'],
+                         'application/json; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', '/v1/a\xe2\x98\x83?format=json'))
+
+        req = Request.blank('/v1/a\xe2\x98\x83?format=json', headers={
+            'X-Backend-Allow-Reserved-Names': 'true'})
+        resp = req.get_response(self.app)
+        self.assertEqual(json.loads(resp.body),
+                         json.loads(self.fake_account_listing_with_reserved))
+        self.assertEqual(resp.headers['Content-Type'],
+                         'application/json; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', '/v1/a\xe2\x98\x83?format=json'))
+
+        req = Request.blank('/v1/a\xe2\x98\x83?format=xml')
+        resp = req.get_response(self.app)
+        self.assertEqual(resp.body.split(b'\n'), [
+            b'<?xml version="1.0" encoding="UTF-8"?>',
+            b'<account name="a\xe2\x98\x83">',
+            b'<container><name>bar</name><count>0</count><bytes>0</bytes>'
+            b'<last_modified>1970-01-01T00:00:00.000000</last_modified>'
+            b'</container>',
+            b'<subdir name="foo_" />',
+            b'</account>',
+        ])
+        self.assertEqual(resp.headers['Content-Type'],
+                         'application/xml; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', '/v1/a\xe2\x98\x83?format=json'))
+
+        req = Request.blank('/v1/a\xe2\x98\x83?format=xml', headers={
+            'X-Backend-Allow-Reserved-Names': 'true'})
+        resp = req.get_response(self.app)
+        self.assertEqual(resp.body.split(b'\n'), [
+            b'<?xml version="1.0" encoding="UTF-8"?>',
+            b'<account name="a\xe2\x98\x83">',
+            b'<container><name>bar</name><count>0</count><bytes>0</bytes>'
+            b'<last_modified>1970-01-01T00:00:00.000000</last_modified>'
+            b'</container>',
+            b'<container><name>%s</name>'
+            b'<count>0</count><bytes>0</bytes>'
+            b'<last_modified>1970-01-01T00:00:00.000000</last_modified>'
+            b'</container>' % get_reserved_name(
+                'bar', 'versions').encode('ascii'),
+            b'<subdir name="foo_" />',
+            b'<subdir name="%s" />' % get_reserved_name(
+                'foo_').encode('ascii'),
+            b'</account>',
+        ])
+        self.assertEqual(resp.headers['Content-Type'],
+                         'application/xml; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', '/v1/a\xe2\x98\x83?format=json'))
+
     def test_valid_container(self):
         self.fake_swift.register('GET', '/v1/a/c', HTTPOk, {
             'Content-Length': str(len(self.fake_container_listing)),
@@ -105,7 +242,8 @@ class TestListingFormats(unittest.TestCase):
 
         req = Request.blank('/v1/a/c?format=json')
         resp = req.get_response(self.app)
-        self.assertEqual(resp.body, self.fake_container_listing)
+        self.assertEqual(json.loads(resp.body),
+                         json.loads(self.fake_container_listing))
         self.assertEqual(resp.headers['Content-Type'],
                          'application/json; charset=utf-8')
         self.assertEqual(self.fake_swift.calls[-1], (
@@ -129,6 +267,126 @@ class TestListingFormats(unittest.TestCase):
         self.assertEqual(self.fake_swift.calls[-1], (
             'GET', '/v1/a/c?format=json'))
 
+    def test_valid_container_with_reserved(self):
+        path = '/v1/a\xe2\x98\x83/c\xf0\x9f\x8c\xb4'
+        body_len = len(self.fake_container_listing_with_reserved)
+        self.fake_swift.register(
+            'GET', path, HTTPOk, {
+                'Content-Length': str(body_len),
+                'Content-Type': 'application/json',
+            }, self.fake_container_listing_with_reserved)
+
+        req = Request.blank(path)
+        resp = req.get_response(self.app)
+        self.assertEqual(resp.body, b'bar\nfoo/\n')
+        self.assertEqual(resp.headers['Content-Type'],
+                         'text/plain; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', path + '?format=json'))
+        self.assertEqual(self.logger.get_lines_for_level('warning'), [
+            "Container listing for a%E2%98%83/c%F0%9F%8C%B4 had reserved byte "
+            "in name: '\\x00bar\\x00extra data'",
+            "Container listing for a%E2%98%83/c%F0%9F%8C%B4 had reserved byte "
+            "in subdir: '\\x00foo/'",
+        ])
+
+        req = Request.blank(path, headers={
+            'X-Backend-Allow-Reserved-Names': 'true'})
+        resp = req.get_response(self.app)
+        self.assertEqual(resp.body, b'bar\n%s\nfoo/\n%s\n' % (
+            get_reserved_name('bar', 'extra data').encode('ascii'),
+            get_reserved_name('foo/').encode('ascii'),
+        ))
+        self.assertEqual(resp.headers['Content-Type'],
+                         'text/plain; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', path + '?format=json'))
+
+        req = Request.blank(path + '?format=txt')
+        resp = req.get_response(self.app)
+        self.assertEqual(resp.body, b'bar\nfoo/\n')
+        self.assertEqual(resp.headers['Content-Type'],
+                         'text/plain; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', path + '?format=json'))
+
+        req = Request.blank(path + '?format=txt', headers={
+            'X-Backend-Allow-Reserved-Names': 'true'})
+        resp = req.get_response(self.app)
+        self.assertEqual(resp.body, b'bar\n%s\nfoo/\n%s\n' % (
+            get_reserved_name('bar', 'extra data').encode('ascii'),
+            get_reserved_name('foo/').encode('ascii'),
+        ))
+        self.assertEqual(resp.headers['Content-Type'],
+                         'text/plain; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', path + '?format=json'))
+
+        req = Request.blank(path + '?format=json')
+        resp = req.get_response(self.app)
+        self.assertEqual(json.loads(resp.body),
+                         json.loads(self.fake_container_listing))
+        self.assertEqual(resp.headers['Content-Type'],
+                         'application/json; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', path + '?format=json'))
+
+        req = Request.blank(path + '?format=json', headers={
+            'X-Backend-Allow-Reserved-Names': 'true'})
+        resp = req.get_response(self.app)
+        self.assertEqual(json.loads(resp.body),
+                         json.loads(self.fake_container_listing_with_reserved))
+        self.assertEqual(resp.headers['Content-Type'],
+                         'application/json; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', path + '?format=json'))
+
+        req = Request.blank(path + '?format=xml')
+        resp = req.get_response(self.app)
+        self.assertEqual(
+            resp.body,
+            b'<?xml version="1.0" encoding="UTF-8"?>\n'
+            b'<container name="c\xf0\x9f\x8c\xb4">'
+            b'<object><name>bar</name><hash>etag</hash><bytes>0</bytes>'
+            b'<content_type>text/plain</content_type>'
+            b'<last_modified>1970-01-01T00:00:00.000000</last_modified>'
+            b'</object>'
+            b'<subdir name="foo/"><name>foo/</name></subdir>'
+            b'</container>'
+        )
+        self.assertEqual(resp.headers['Content-Type'],
+                         'application/xml; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', path + '?format=json'))
+
+        req = Request.blank(path + '?format=xml', headers={
+            'X-Backend-Allow-Reserved-Names': 'true'})
+        resp = req.get_response(self.app)
+        self.assertEqual(
+            resp.body,
+            b'<?xml version="1.0" encoding="UTF-8"?>\n'
+            b'<container name="c\xf0\x9f\x8c\xb4">'
+            b'<object><name>bar</name><hash>etag</hash><bytes>0</bytes>'
+            b'<content_type>text/plain</content_type>'
+            b'<last_modified>1970-01-01T00:00:00.000000</last_modified>'
+            b'</object>'
+            b'<object><name>%s</name>'
+            b'<hash>etag</hash><bytes>0</bytes>'
+            b'<content_type>text/plain</content_type>'
+            b'<last_modified>1970-01-01T00:00:00.000000</last_modified>'
+            b'</object>'
+            b'<subdir name="foo/"><name>foo/</name></subdir>'
+            b'<subdir name="%s"><name>%s</name></subdir>'
+            b'</container>' % (
+                get_reserved_name('bar', 'extra data').encode('ascii'),
+                get_reserved_name('foo/').encode('ascii'),
+                get_reserved_name('foo/').encode('ascii'),
+            ))
+        self.assertEqual(resp.headers['Content-Type'],
+                         'application/xml; charset=utf-8')
+        self.assertEqual(self.fake_swift.calls[-1], (
+            'GET', path + '?format=json'))
+
     def test_blank_account(self):
         self.fake_swift.register('GET', '/v1/a', HTTPOk, {
             'Content-Length': '2', 'Content-Type': 'application/json'}, b'[]')
diff --git a/test/unit/common/test_constraints.py b/test/unit/common/test_constraints.py
index aa42f243c8..5090fddeb6 100644
--- a/test/unit/common/test_constraints.py
+++ b/test/unit/common/test_constraints.py
@@ -506,7 +506,7 @@ class TestConstraints(unittest.TestCase):
 
     def test_check_utf8(self):
         unicode_sample = u'\uc77c\uc601'
-        unicode_with_null = u'abc\u0000def'
+        unicode_with_reserved = u'abc%sdef' % utils.RESERVED_STR
 
         # Some false-y values
         self.assertFalse(constraints.check_utf8(None))
@@ -518,15 +518,24 @@ class TestConstraints(unittest.TestCase):
         self.assertFalse(constraints.check_utf8(
             unicode_sample.encode('utf-8')[::-1]))
         # unicode with null
-        self.assertFalse(constraints.check_utf8(unicode_with_null))
+        self.assertFalse(constraints.check_utf8(unicode_with_reserved))
         # utf8 bytes with null
         self.assertFalse(constraints.check_utf8(
-            unicode_with_null.encode('utf8')))
+            unicode_with_reserved.encode('utf8')))
 
         self.assertTrue(constraints.check_utf8('this is ascii and utf-8, too'))
         self.assertTrue(constraints.check_utf8(unicode_sample))
         self.assertTrue(constraints.check_utf8(unicode_sample.encode('utf8')))
 
+    def test_check_utf8_internal(self):
+        unicode_with_reserved = u'abc%sdef' % utils.RESERVED_STR
+        # sanity
+        self.assertFalse(constraints.check_utf8(unicode_with_reserved))
+        self.assertTrue(constraints.check_utf8('foobar', internal=True))
+        # internal allows reserved names
+        self.assertTrue(constraints.check_utf8(unicode_with_reserved,
+                                               internal=True))
+
     def test_check_utf8_non_canonical(self):
         self.assertFalse(constraints.check_utf8(b'\xed\xa0\xbc\xed\xbc\xb8'))
         self.assertTrue(constraints.check_utf8(u'\U0001f338'))
diff --git a/test/unit/common/test_direct_client.py b/test/unit/common/test_direct_client.py
index ad711f378d..a714d16ba9 100644
--- a/test/unit/common/test_direct_client.py
+++ b/test/unit/common/test_direct_client.py
@@ -132,66 +132,76 @@ class TestDirectClient(unittest.TestCase):
         stub_user_agent = 'direct-client %s' % os.getpid()
 
         headers = direct_client.gen_headers(add_ts=False)
-        self.assertEqual(headers['user-agent'], stub_user_agent)
-        self.assertEqual(1, len(headers))
+        self.assertEqual(dict(headers), {
+            'User-Agent': stub_user_agent,
+            'X-Backend-Allow-Reserved-Names': 'true',
+        })
 
-        now = time.time()
-        headers = direct_client.gen_headers()
-        self.assertEqual(headers['user-agent'], stub_user_agent)
-        self.assertTrue(now - 1 < Timestamp(headers['x-timestamp']) < now + 1)
-        self.assertEqual(headers['x-timestamp'],
-                         Timestamp(headers['x-timestamp']).internal)
-        self.assertEqual(2, len(headers))
+        with mock.patch('swift.common.utils.Timestamp.now',
+                        return_value=Timestamp('123.45')):
+            headers = direct_client.gen_headers()
+        self.assertEqual(dict(headers), {
+            'User-Agent': stub_user_agent,
+            'X-Backend-Allow-Reserved-Names': 'true',
+            'X-Timestamp': '0000000123.45000',
+        })
 
         headers = direct_client.gen_headers(hdrs_in={'x-timestamp': '15'})
-        self.assertEqual(headers['x-timestamp'], '15')
-        self.assertEqual(headers['user-agent'], stub_user_agent)
-        self.assertEqual(2, len(headers))
+        self.assertEqual(dict(headers), {
+            'User-Agent': stub_user_agent,
+            'X-Backend-Allow-Reserved-Names': 'true',
+            'X-Timestamp': '15',
+        })
 
-        headers = direct_client.gen_headers(hdrs_in={'foo-bar': '63'})
-        self.assertEqual(headers['user-agent'], stub_user_agent)
-        self.assertEqual(headers['foo-bar'], '63')
-        self.assertTrue(now - 1 < Timestamp(headers['x-timestamp']) < now + 1)
-        self.assertEqual(headers['x-timestamp'],
-                         Timestamp(headers['x-timestamp']).internal)
-        self.assertEqual(3, len(headers))
+        with mock.patch('swift.common.utils.Timestamp.now',
+                        return_value=Timestamp('12345.6789')):
+            headers = direct_client.gen_headers(hdrs_in={'foo-bar': '63'})
+        self.assertEqual(dict(headers), {
+            'User-Agent': stub_user_agent,
+            'Foo-Bar': '63',
+            'X-Backend-Allow-Reserved-Names': 'true',
+            'X-Timestamp': '0000012345.67890',
+        })
 
         hdrs_in = {'foo-bar': '55'}
         headers = direct_client.gen_headers(hdrs_in, add_ts=False)
-        self.assertEqual(headers['user-agent'], stub_user_agent)
-        self.assertEqual(headers['foo-bar'], '55')
-        self.assertEqual(2, len(headers))
+        self.assertEqual(dict(headers), {
+            'User-Agent': stub_user_agent,
+            'Foo-Bar': '55',
+            'X-Backend-Allow-Reserved-Names': 'true',
+        })
 
-        headers = direct_client.gen_headers(hdrs_in={'user-agent': '32'})
-        self.assertEqual(headers['user-agent'], '32')
-        self.assertTrue(now - 1 < Timestamp(headers['x-timestamp']) < now + 1)
-        self.assertEqual(headers['x-timestamp'],
-                         Timestamp(headers['x-timestamp']).internal)
-        self.assertEqual(2, len(headers))
+        with mock.patch('swift.common.utils.Timestamp.now',
+                        return_value=Timestamp('12345')):
+            headers = direct_client.gen_headers(hdrs_in={'user-agent': '32'})
+        self.assertEqual(dict(headers), {
+            'User-Agent': '32',
+            'X-Backend-Allow-Reserved-Names': 'true',
+            'X-Timestamp': '0000012345.00000',
+        })
 
         hdrs_in = {'user-agent': '47'}
         headers = direct_client.gen_headers(hdrs_in, add_ts=False)
-        self.assertEqual(headers['user-agent'], '47')
-        self.assertEqual(1, len(headers))
+        self.assertEqual(dict(headers), {
+            'User-Agent': '47',
+            'X-Backend-Allow-Reserved-Names': 'true',
+        })
 
         for policy in POLICIES:
             for add_ts in (True, False):
-                now = time.time()
-                headers = direct_client.gen_headers(
-                    {'X-Backend-Storage-Policy-Index': policy.idx},
-                    add_ts=add_ts)
-                self.assertEqual(headers['user-agent'], stub_user_agent)
-                self.assertEqual(headers['X-Backend-Storage-Policy-Index'],
-                                 str(policy.idx))
-                expected_header_count = 2
+                with mock.patch('swift.common.utils.Timestamp.now',
+                                return_value=Timestamp('123456789')):
+                    headers = direct_client.gen_headers(
+                        {'X-Backend-Storage-Policy-Index': policy.idx},
+                        add_ts=add_ts)
+                expected = {
+                    'User-Agent': stub_user_agent,
+                    'X-Backend-Storage-Policy-Index': str(policy.idx),
+                    'X-Backend-Allow-Reserved-Names': 'true',
+                }
                 if add_ts:
-                    expected_header_count += 1
-                    self.assertEqual(
-                        headers['x-timestamp'],
-                        Timestamp(headers['x-timestamp']).internal)
-                    self.assertTrue(
-                        now - 1 < Timestamp(headers['x-timestamp']) < now + 1)
-                self.assertEqual(expected_header_count, len(headers))
+                    expected['X-Timestamp'] = '0123456789.00000'
+                self.assertEqual(dict(headers), expected)
 
     def test_direct_get_account(self):
         def do_test(req_params):
diff --git a/test/unit/common/test_internal_client.py b/test/unit/common/test_internal_client.py
index 524be1054d..e957ce894c 100644
--- a/test/unit/common/test_internal_client.py
+++ b/test/unit/common/test_internal_client.py
@@ -1233,7 +1233,8 @@ class TestInternalClient(unittest.TestCase):
         self.assertEqual(app.call_count, 1)
         req_headers.update({
             'host': 'localhost:80',  # from swob.Request.blank
-            'user-agent': 'test',   # from InternalClient.make_request
+            'user-agent': 'test',  # from InternalClient.make_request
+            'x-backend-allow-reserved-names': 'true',  # also from IC
         })
         self.assertEqual(app.calls_with_headers, [(
             'GET', path_info + '?symlink=get', HeaderKeyDict(req_headers))])
diff --git a/test/unit/common/test_request_helpers.py b/test/unit/common/test_request_helpers.py
index 5e07ddb584..bbdc3be7ff 100644
--- a/test/unit/common/test_request_helpers.py
+++ b/test/unit/common/test_request_helpers.py
@@ -181,6 +181,174 @@ class TestRequestHelpers(unittest.TestCase):
         self.assertEqual(policy, POLICIES[1])
         self.assertEqual(policy.policy_type, REPL_POLICY)
 
+    def test_validate_internal_name(self):
+        self.assertIsNone(rh._validate_internal_name('foo'))
+        self.assertIsNone(rh._validate_internal_name(
+            rh.get_reserved_name('foo')))
+        self.assertIsNone(rh._validate_internal_name(
+            rh.get_reserved_name('foo', 'bar')))
+        self.assertIsNone(rh._validate_internal_name(''))
+        self.assertIsNone(rh._validate_internal_name(rh.RESERVED))
+
+    def test_invalid_reserved_name(self):
+        with self.assertRaises(HTTPException) as raised:
+            rh._validate_internal_name('foo' + rh.RESERVED)
+        e = raised.exception
+        self.assertEqual(e.status_int, 400)
+        self.assertEqual(str(e), '400 Bad Request')
+        self.assertEqual(e.body, b"Invalid reserved-namespace name")
+
+    def test_validate_internal_account(self):
+        self.assertIsNone(rh.validate_internal_account('AUTH_foo'))
+        self.assertIsNone(rh.validate_internal_account(
+            rh.get_reserved_name('AUTH_foo')))
+        with self.assertRaises(HTTPException) as raised:
+            rh.validate_internal_account('AUTH_foo' + rh.RESERVED)
+        e = raised.exception
+        self.assertEqual(e.status_int, 400)
+        self.assertEqual(str(e), '400 Bad Request')
+        self.assertEqual(e.body, b"Invalid reserved-namespace account")
+
+    def test_validate_internal_container(self):
+        self.assertIsNone(rh.validate_internal_container('AUTH_foo', 'bar'))
+        self.assertIsNone(rh.validate_internal_container(
+            rh.get_reserved_name('AUTH_foo'), 'bar'))
+        self.assertIsNone(rh.validate_internal_container(
+            'foo', rh.get_reserved_name('bar')))
+        self.assertIsNone(rh.validate_internal_container(
+            rh.get_reserved_name('AUTH_foo'), rh.get_reserved_name('bar')))
+        with self.assertRaises(HTTPException) as raised:
+            rh.validate_internal_container('AUTH_foo' + rh.RESERVED, 'bar')
+        e = raised.exception
+        self.assertEqual(e.status_int, 400)
+        self.assertEqual(str(e), '400 Bad Request')
+        self.assertEqual(e.body, b"Invalid reserved-namespace account")
+        with self.assertRaises(HTTPException) as raised:
+            rh.validate_internal_container('AUTH_foo', 'bar' + rh.RESERVED)
+        e = raised.exception
+        self.assertEqual(e.status_int, 400)
+        self.assertEqual(str(e), '400 Bad Request')
+        self.assertEqual(e.body, b"Invalid reserved-namespace container")
+
+        # These should always be operating on split_path outputs so this
+        # shouldn't really be an issue, but just in case...
+        for acct in ('', None):
+            with self.assertRaises(ValueError) as raised:
+                rh.validate_internal_container(
+                    acct, 'bar')
+            self.assertEqual(raised.exception.args[0], 'Account is required')
+
+    def test_validate_internal_object(self):
+        self.assertIsNone(rh.validate_internal_obj('AUTH_foo', 'bar', 'baz'))
+        self.assertIsNone(rh.validate_internal_obj(
+            rh.get_reserved_name('AUTH_foo'), 'bar', 'baz'))
+        for acct in ('AUTH_foo', rh.get_reserved_name('AUTH_foo')):
+            self.assertIsNone(rh.validate_internal_obj(
+                acct,
+                rh.get_reserved_name('bar'),
+                rh.get_reserved_name('baz')))
+        for acct in ('AUTH_foo', rh.get_reserved_name('AUTH_foo')):
+            with self.assertRaises(HTTPException) as raised:
+                rh.validate_internal_obj(
+                    acct, 'bar', rh.get_reserved_name('baz'))
+            e = raised.exception
+            self.assertEqual(e.status_int, 400)
+            self.assertEqual(str(e), '400 Bad Request')
+            self.assertEqual(e.body, b"Invalid reserved-namespace object "
+                             b"in user-namespace container")
+        for acct in ('AUTH_foo', rh.get_reserved_name('AUTH_foo')):
+            with self.assertRaises(HTTPException) as raised:
+                rh.validate_internal_obj(
+                    acct, rh.get_reserved_name('bar'), 'baz')
+            e = raised.exception
+            self.assertEqual(e.status_int, 400)
+            self.assertEqual(str(e), '400 Bad Request')
+            self.assertEqual(e.body, b"Invalid user-namespace object "
+                             b"in reserved-namespace container")
+
+        # These should always be operating on split_path outputs so this
+        # shouldn't really be an issue, but just in case...
+        for acct in ('', None):
+            with self.assertRaises(ValueError) as raised:
+                rh.validate_internal_obj(
+                    acct, 'bar', 'baz')
+            self.assertEqual(raised.exception.args[0], 'Account is required')
+
+        for cont in ('', None):
+            with self.assertRaises(ValueError) as raised:
+                rh.validate_internal_obj(
+                    'AUTH_foo', cont, 'baz')
+            self.assertEqual(raised.exception.args[0], 'Container is required')
+
+    def test_invalid_reserved_names(self):
+        with self.assertRaises(HTTPException) as raised:
+            rh.validate_internal_obj('AUTH_foo' + rh.RESERVED, 'bar', 'baz')
+        e = raised.exception
+        self.assertEqual(e.status_int, 400)
+        self.assertEqual(str(e), '400 Bad Request')
+        self.assertEqual(e.body, b"Invalid reserved-namespace account")
+        with self.assertRaises(HTTPException) as raised:
+            rh.validate_internal_obj('AUTH_foo', 'bar' + rh.RESERVED, 'baz')
+        e = raised.exception
+        self.assertEqual(e.status_int, 400)
+        self.assertEqual(str(e), '400 Bad Request')
+        self.assertEqual(e.body, b"Invalid reserved-namespace container")
+        with self.assertRaises(HTTPException) as raised:
+            rh.validate_internal_obj('AUTH_foo', 'bar', 'baz' + rh.RESERVED)
+        e = raised.exception
+        self.assertEqual(e.status_int, 400)
+        self.assertEqual(str(e), '400 Bad Request')
+        self.assertEqual(e.body, b"Invalid reserved-namespace object")
+
+    def test_get_reserved_name(self):
+        expectations = {
+            tuple(): rh.RESERVED,
+            ('',): rh.RESERVED,
+            ('foo',): rh.RESERVED + 'foo',
+            ('foo', 'bar'): rh.RESERVED + 'foo' + rh.RESERVED + 'bar',
+            ('foo', ''): rh.RESERVED + 'foo' + rh.RESERVED,
+            ('', ''): rh.RESERVED * 2,
+        }
+        failures = []
+        for parts, expected in expectations.items():
+            name = rh.get_reserved_name(*parts)
+            if name != expected:
+                failures.append('get given %r expected %r != %r' % (
+                    parts, expected, name))
+        if failures:
+            self.fail('Unexpected reults:\n' + '\n'.join(failures))
+
+    def test_invalid_get_reserved_name(self):
+        self.assertRaises(ValueError)
+        with self.assertRaises(ValueError) as ctx:
+            rh.get_reserved_name('foo', rh.RESERVED + 'bar', 'baz')
+        self.assertEqual(str(ctx.exception),
+                         'Invalid reserved part in components')
+
+    def test_split_reserved_name(self):
+        expectations = {
+            rh.RESERVED: ('',),
+            rh.RESERVED + 'foo': ('foo',),
+            rh.RESERVED + 'foo' + rh.RESERVED + 'bar': ('foo', 'bar'),
+            rh.RESERVED + 'foo' + rh.RESERVED: ('foo', ''),
+            rh.RESERVED * 2: ('', ''),
+        }
+        failures = []
+        for name, expected in expectations.items():
+            parts = rh.split_reserved_name(name)
+            if tuple(parts) != expected:
+                failures.append('split given %r expected %r != %r' % (
+                    name, expected, parts))
+        if failures:
+            self.fail('Unexpected reults:\n' + '\n'.join(failures))
+
+    def test_invalid_split_reserved_name(self):
+        self.assertRaises(ValueError)
+        with self.assertRaises(ValueError) as ctx:
+            rh.split_reserved_name('foo')
+        self.assertEqual(str(ctx.exception),
+                         'Invalid reserved name')
+
 
 class TestHTTPResponseToDocumentIters(unittest.TestCase):
     def test_200(self):
diff --git a/test/unit/common/test_swob.py b/test/unit/common/test_swob.py
index cfaaa1fd80..a6dbeed27a 100644
--- a/test/unit/common/test_swob.py
+++ b/test/unit/common/test_swob.py
@@ -1115,6 +1115,19 @@ class TestRequest(unittest.TestCase):
         else:
             self.fail("Expected an AttributeError raised for 'gzip,identity'")
 
+    def test_allow_reserved_names(self):
+        req = swift.common.swob.Request.blank('', headers={})
+        self.assertFalse(req.allow_reserved_names)
+        req = swift.common.swob.Request.blank('', headers={
+            'X-Allow-Reserved-Names': 'true'})
+        self.assertFalse(req.allow_reserved_names)
+        req = swift.common.swob.Request.blank('', headers={
+            'X-Backend-Allow-Reserved-Names': 'false'})
+        self.assertFalse(req.allow_reserved_names)
+        req = swift.common.swob.Request.blank('', headers={
+            'X-Backend-Allow-Reserved-Names': 'true'})
+        self.assertTrue(req.allow_reserved_names)
+
 
 class TestStatusMap(unittest.TestCase):
     def test_status_map(self):
diff --git a/test/unit/container/test_backend.py b/test/unit/container/test_backend.py
index b43f91d055..d53f1cda40 100644
--- a/test/unit/container/test_backend.py
+++ b/test/unit/container/test_backend.py
@@ -36,6 +36,7 @@ from swift.container.backend import ContainerBroker, \
     update_new_item_from_existing, UNSHARDED, SHARDING, SHARDED, \
     COLLAPSED, SHARD_LISTING_STATES, SHARD_UPDATE_STATES
 from swift.common.db import DatabaseAlreadyExists, GreenDBConnection
+from swift.common.request_helpers import get_reserved_name
 from swift.common.utils import Timestamp, encode_timestamps, hash_path, \
     ShardRange, make_db_file_path
 from swift.common.storage_policy import POLICIES
@@ -2365,6 +2366,33 @@ class TestContainerBroker(unittest.TestCase):
         self.assertEqual(len(listing), 2)
         self.assertEqual([row[0] for row in listing], ['3/0000', '3/0001'])
 
+    def test_list_objects_iter_with_reserved_name(self):
+        broker = ContainerBroker(':memory:', account='a', container='c')
+        broker.initialize(next(self.ts).internal, 0)
+
+        broker.put_object(
+            'foo', next(self.ts).internal, 0, 0, 0, POLICIES.default.idx)
+        broker.put_object(
+            get_reserved_name('foo'), next(self.ts).internal, 0, 0, 0,
+            POLICIES.default.idx)
+
+        listing = broker.list_objects_iter(100, None, None, '', '')
+        self.assertEqual([row[0] for row in listing], ['foo'])
+
+        listing = broker.list_objects_iter(100, None, None, '', '',
+                                           reverse=True)
+        self.assertEqual([row[0] for row in listing], ['foo'])
+
+        listing = broker.list_objects_iter(100, None, None, '', '',
+                                           allow_reserved=True)
+        self.assertEqual([row[0] for row in listing],
+                         [get_reserved_name('foo'), 'foo'])
+
+        listing = broker.list_objects_iter(100, None, None, '', '',
+                                           reverse=True, allow_reserved=True)
+        self.assertEqual([row[0] for row in listing],
+                         ['foo', get_reserved_name('foo')])
+
     def test_reverse_prefix_delim(self):
         expectations = [
             {
diff --git a/test/unit/container/test_server.py b/test/unit/container/test_server.py
index 01d26ab8fc..c45ec01c9b 100644
--- a/test/unit/container/test_server.py
+++ b/test/unit/container/test_server.py
@@ -26,13 +26,13 @@ from contextlib import contextmanager
 from io import BytesIO
 from shutil import rmtree
 from tempfile import mkdtemp
-from test.unit import make_timestamp_iter, mock_timestamp_now
 from xml.dom import minidom
 
 from eventlet import spawn, Timeout
 import json
 import six
 from six import StringIO
+from six.moves.urllib.parse import quote
 
 from swift import __version__ as swift_version
 from swift.common.header_key_dict import HeaderKeyDict
@@ -43,13 +43,13 @@ from swift.container import server as container_server
 from swift.common import constraints
 from swift.common.utils import (Timestamp, mkdirs, public, replication,
                                 storage_directory, lock_parent_directory,
-                                ShardRange)
+                                ShardRange, RESERVED_STR)
 from test.unit import fake_http_connect, debug_logger, mock_check_drive
 from swift.common.storage_policy import (POLICIES, StoragePolicy)
-from swift.common.request_helpers import get_sys_meta_prefix
+from swift.common.request_helpers import get_sys_meta_prefix, get_reserved_name
 
 from test import listen_zero, annotate_failure
-from test.unit import patch_policies
+from test.unit import patch_policies, make_timestamp_iter, mock_timestamp_now
 
 
 @contextmanager
@@ -78,6 +78,7 @@ class TestContainerController(unittest.TestCase):
             logger=self.logger)
         # some of the policy tests want at least two policies
         self.assertTrue(len(POLICIES) > 1)
+        self.ts = make_timestamp_iter()
 
     def tearDown(self):
         rmtree(os.path.dirname(self.testdir), ignore_errors=1)
@@ -282,11 +283,9 @@ class TestContainerController(unittest.TestCase):
             self.assertIsNone(resp.headers[header])
 
     def test_deleted_headers(self):
-        ts = (Timestamp(t).internal for t in
-              itertools.count(int(time.time())))
         request_method_times = {
-            'PUT': next(ts),
-            'DELETE': next(ts),
+            'PUT': next(self.ts).internal,
+            'DELETE': next(self.ts).internal,
         }
         # setup a deleted container
         for method in ('PUT', 'DELETE'):
@@ -547,11 +546,10 @@ class TestContainerController(unittest.TestCase):
         self.assertFalse('X-Backend-Storage-Policy-Index' in resp.headers)
 
     def test_PUT_no_policy_change(self):
-        ts = (Timestamp(t).internal for t in itertools.count(time.time()))
         policy = random.choice(list(POLICIES))
         # Set metadata header
         req = Request.blank('/sda1/p/a/c', method='PUT', headers={
-            'X-Timestamp': next(ts),
+            'X-Timestamp': next(self.ts).internal,
             'X-Backend-Storage-Policy-Index': policy.idx})
         resp = req.get_response(self.controller)
         self.assertEqual(resp.status_int, 201)
@@ -565,7 +563,7 @@ class TestContainerController(unittest.TestCase):
         # now try to update w/o changing the policy
         for method in ('POST', 'PUT'):
             req = Request.blank('/sda1/p/a/c', method=method, headers={
-                'X-Timestamp': next(ts),
+                'X-Timestamp': next(self.ts).internal,
                 'X-Backend-Storage-Policy-Index': policy.idx
             })
             resp = req.get_response(self.controller)
@@ -578,11 +576,10 @@ class TestContainerController(unittest.TestCase):
                          str(policy.idx))
 
     def test_PUT_bad_policy_change(self):
-        ts = (Timestamp(t).internal for t in itertools.count(time.time()))
         policy = random.choice(list(POLICIES))
         # Set metadata header
         req = Request.blank('/sda1/p/a/c', method='PUT', headers={
-            'X-Timestamp': next(ts),
+            'X-Timestamp': next(self.ts).internal,
             'X-Backend-Storage-Policy-Index': policy.idx})
         resp = req.get_response(self.controller)
         self.assertEqual(resp.status_int, 201)
@@ -597,7 +594,7 @@ class TestContainerController(unittest.TestCase):
         for other_policy in other_policies:
             # now try to change it and make sure we get a conflict
             req = Request.blank('/sda1/p/a/c', method='PUT', headers={
-                'X-Timestamp': next(ts),
+                'X-Timestamp': next(self.ts).internal,
                 'X-Backend-Storage-Policy-Index': other_policy.idx
             })
             resp = req.get_response(self.controller)
@@ -615,10 +612,9 @@ class TestContainerController(unittest.TestCase):
                          str(policy.idx))
 
     def test_POST_ignores_policy_change(self):
-        ts = (Timestamp(t).internal for t in itertools.count(time.time()))
         policy = random.choice(list(POLICIES))
         req = Request.blank('/sda1/p/a/c', method='PUT', headers={
-            'X-Timestamp': next(ts),
+            'X-Timestamp': next(self.ts).internal,
             'X-Backend-Storage-Policy-Index': policy.idx})
         resp = req.get_response(self.controller)
         self.assertEqual(resp.status_int, 201)
@@ -633,7 +629,7 @@ class TestContainerController(unittest.TestCase):
         for other_policy in other_policies:
             # now try to change it and make sure we get a conflict
             req = Request.blank('/sda1/p/a/c', method='POST', headers={
-                'X-Timestamp': next(ts),
+                'X-Timestamp': next(self.ts).internal,
                 'X-Backend-Storage-Policy-Index': other_policy.idx
             })
             resp = req.get_response(self.controller)
@@ -650,11 +646,9 @@ class TestContainerController(unittest.TestCase):
                              str(policy.idx))
 
     def test_PUT_no_policy_for_existing_default(self):
-        ts = (Timestamp(t).internal for t in
-              itertools.count(int(time.time())))
         # create a container with the default storage policy
         req = Request.blank('/sda1/p/a/c', method='PUT', headers={
-            'X-Timestamp': next(ts),
+            'X-Timestamp': next(self.ts).internal,
         })
         resp = req.get_response(self.controller)
         self.assertEqual(resp.status_int, 201)  # sanity check
@@ -668,7 +662,7 @@ class TestContainerController(unittest.TestCase):
 
         # put again without specifying the storage policy
         req = Request.blank('/sda1/p/a/c', method='PUT', headers={
-            'X-Timestamp': next(ts),
+            'X-Timestamp': next(self.ts).internal,
         })
         resp = req.get_response(self.controller)
         self.assertEqual(resp.status_int, 202)  # sanity check
@@ -685,11 +679,9 @@ class TestContainerController(unittest.TestCase):
         # during a config change restart across a multi node cluster.
         proxy_default = random.choice([p for p in POLICIES if not
                                        p.is_default])
-        ts = (Timestamp(t).internal for t in
-              itertools.count(int(time.time())))
         # create a container with the default storage policy
         req = Request.blank('/sda1/p/a/c', method='PUT', headers={
-            'X-Timestamp': next(ts),
+            'X-Timestamp': next(self.ts).internal,
             'X-Backend-Storage-Policy-Default': int(proxy_default),
         })
         resp = req.get_response(self.controller)
@@ -704,7 +696,7 @@ class TestContainerController(unittest.TestCase):
 
         # put again without proxy specifying the different default
         req = Request.blank('/sda1/p/a/c', method='PUT', headers={
-            'X-Timestamp': next(ts),
+            'X-Timestamp': next(self.ts).internal,
             'X-Backend-Storage-Policy-Default': int(POLICIES.default),
         })
         resp = req.get_response(self.controller)
@@ -718,11 +710,10 @@ class TestContainerController(unittest.TestCase):
                          int(proxy_default))
 
     def test_PUT_no_policy_for_existing_non_default(self):
-        ts = (Timestamp(t).internal for t in itertools.count(time.time()))
         non_default_policy = [p for p in POLICIES if not p.is_default][0]
         # create a container with the non-default storage policy
         req = Request.blank('/sda1/p/a/c', method='PUT', headers={
-            'X-Timestamp': next(ts),
+            'X-Timestamp': next(self.ts).internal,
             'X-Backend-Storage-Policy-Index': non_default_policy.idx,
         })
         resp = req.get_response(self.controller)
@@ -737,7 +728,7 @@ class TestContainerController(unittest.TestCase):
 
         # put again without specifying the storage policy
         req = Request.blank('/sda1/p/a/c', method='PUT', headers={
-            'X-Timestamp': next(ts),
+            'X-Timestamp': next(self.ts).internal,
         })
         resp = req.get_response(self.controller)
         self.assertEqual(resp.status_int, 202)  # sanity check
@@ -749,6 +740,39 @@ class TestContainerController(unittest.TestCase):
         self.assertEqual(resp.headers['X-Backend-Storage-Policy-Index'],
                          str(non_default_policy.idx))
 
+    def test_create_reserved_namespace_container(self):
+        path = '/sda1/p/a/%sc' % RESERVED_STR
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status, '201 Created', resp.body)
+
+        path = '/sda1/p/a/%sc%stest' % (RESERVED_STR, RESERVED_STR)
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status, '201 Created', resp.body)
+
+    def test_create_reserved_object_in_container(self):
+        # create container
+        path = '/sda1/p/a/c/'
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 201)
+        # put null object in it
+        path += '%so' % RESERVED_STR
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal,
+            'X-Size': 0,
+            'X-Content-Type': 'application/x-test',
+            'X-Etag': 'x',
+        })
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status, '400 Bad Request')
+        self.assertEqual(resp.body, b'Invalid reserved-namespace object '
+                         b'in user-namespace container')
+
     def test_PUT_non_utf8_metadata(self):
         # Set metadata header
         req = Request.blank(
@@ -4106,6 +4130,238 @@ class TestContainerController(unittest.TestCase):
              for item in json.loads(resp.body)],
             [{"name": "US~~UT~~~B"}])
 
+    def _report_objects(self, path, objects):
+        req = Request.blank(path, method='PUT', headers={
+            'x-timestamp': next(self.ts).internal})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int // 100, 2, resp.body)
+        for obj in objects:
+            obj_path = path + '/%s' % obj['name']
+            req = Request.blank(obj_path, method='PUT', headers={
+                'X-Timestamp': obj['timestamp'].internal,
+                'X-Size': obj['bytes'],
+                'X-Content-Type': obj['content_type'],
+                'X-Etag': obj['hash'],
+            })
+            self._update_object_put_headers(req)
+            resp = req.get_response(self.controller)
+            self.assertEqual(resp.status_int // 100, 2, resp.body)
+
+    def _expected_listing(self, objects):
+        return [dict(
+            last_modified=o['timestamp'].isoformat, **{
+                k: v for k, v in o.items()
+                if k != 'timestamp'
+            }) for o in sorted(objects, key=lambda o: o['name'])]
+
+    def test_listing_with_reserved(self):
+        objects = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 8,
+            'content_type': 'application/octet-stream',
+            'hash': '70c1db56f301c9e337b0099bd4174b28',
+            'timestamp': next(self.ts),
+        }]
+        path = '/sda1/p/a/%s' % get_reserved_name('null')
+        self._report_objects(path, objects)
+
+        req = Request.blank(path, headers={'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank(path, headers={
+            'X-Backend-Allow-Reserved-Names': 'true',
+            'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(objects))
+
+    def test_delimiter_with_reserved(self):
+        objects = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 8,
+            'content_type': 'application/octet-stream',
+            'hash': '70c1db56f301c9e337b0099bd4174b28',
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'test02'),
+            'bytes': 8,
+            'content_type': 'application/octet-stream',
+            'hash': '70c1db56f301c9e337b0099bd4174b28',
+            'timestamp': next(self.ts),
+        }]
+        path = '/sda1/p/a/%s' % get_reserved_name('null')
+        self._report_objects(path, objects)
+
+        req = Request.blank(path + '?prefix=%s&delimiter=l' %
+                            get_reserved_name('nul'), headers={
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank(path + '?prefix=%s&delimiter=l' %
+                            get_reserved_name('nul'), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [{
+            'subdir': '%s' % get_reserved_name('null')}])
+
+        req = Request.blank(path + '?prefix=%s&delimiter=%s' % (
+                            get_reserved_name('nul'), get_reserved_name('')),
+                            headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [{
+            'subdir': '%s' % get_reserved_name('null', '')}])
+
+    def test_markers_with_reserved(self):
+        objects = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 8,
+            'content_type': 'application/octet-stream',
+            'hash': '70c1db56f301c9e337b0099bd4174b28',
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'test02'),
+            'bytes': 10,
+            'content_type': 'application/octet-stream',
+            'hash': '912ec803b2ce49e4a541068d495ab570',
+            'timestamp': next(self.ts),
+        }]
+        path = '/sda1/p/a/%s' % get_reserved_name('null')
+        self._report_objects(path, objects)
+
+        req = Request.blank(path + '?marker=%s' %
+                            get_reserved_name('null', ''), headers={
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank(path + '?marker=%s' %
+                            get_reserved_name('null', ''), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(objects))
+
+        req = Request.blank(path + '?marker=%s' %
+                            quote(json.loads(resp.body)[0]['name']), headers={
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank(path + '?marker=%s' %
+                            quote(self._expected_listing(objects)[0]['name']),
+                            headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(objects)[1:])
+
+    def test_prefix_with_reserved(self):
+        objects = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 8,
+            'content_type': 'application/octet-stream',
+            'hash': '70c1db56f301c9e337b0099bd4174b28',
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'test02'),
+            'bytes': 10,
+            'content_type': 'application/octet-stream',
+            'hash': '912ec803b2ce49e4a541068d495ab570',
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'foo'),
+            'bytes': 12,
+            'content_type': 'application/octet-stream',
+            'hash': 'acbd18db4cc2f85cedef654fccc4a4d8',
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('nullish'),
+            'bytes': 13,
+            'content_type': 'application/octet-stream',
+            'hash': '37b51d194a7513e45b56f6524f2d51f2',
+            'timestamp': next(self.ts),
+        }]
+        path = '/sda1/p/a/%s' % get_reserved_name('null')
+        self._report_objects(path, objects)
+
+        req = Request.blank(path + '?prefix=%s' %
+                            get_reserved_name('null', 'test'), headers={
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank(path + '?prefix=%s' %
+                            get_reserved_name('null', 'test'), headers={
+                                'X-Backend-Allow-Reserved-Names': 'true',
+                                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body),
+                         self._expected_listing(objects[:2]))
+
+    def test_prefix_and_delim_with_reserved(self):
+        objects = [{
+            'name': get_reserved_name('null', 'test01'),
+            'bytes': 8,
+            'content_type': 'application/octet-stream',
+            'hash': '70c1db56f301c9e337b0099bd4174b28',
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'test02'),
+            'bytes': 10,
+            'content_type': 'application/octet-stream',
+            'hash': '912ec803b2ce49e4a541068d495ab570',
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('null', 'foo'),
+            'bytes': 12,
+            'content_type': 'application/octet-stream',
+            'hash': 'acbd18db4cc2f85cedef654fccc4a4d8',
+            'timestamp': next(self.ts),
+        }, {
+            'name': get_reserved_name('nullish'),
+            'bytes': 13,
+            'content_type': 'application/octet-stream',
+            'hash': '37b51d194a7513e45b56f6524f2d51f2',
+            'timestamp': next(self.ts),
+        }]
+        path = '/sda1/p/a/%s' % get_reserved_name('null')
+        self._report_objects(path, objects)
+
+        req = Request.blank(path + '?prefix=%s&delimiter=%s' % (
+            get_reserved_name('null'), get_reserved_name()), headers={
+                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        self.assertEqual(json.loads(resp.body), [])
+
+        req = Request.blank(path + '?prefix=%s&delimiter=%s' % (
+            get_reserved_name('null'), get_reserved_name()), headers={
+                'X-Backend-Allow-Reserved-Names': 'true',
+                'Accept': 'application/json'})
+        resp = req.get_response(self.controller)
+        self.assertEqual(resp.status_int, 200, resp.body)
+        expected = [{'subdir': get_reserved_name('null', '')}] + \
+            self._expected_listing(objects)[-1:]
+        self.assertEqual(json.loads(resp.body), expected)
+
     def test_GET_delimiter_non_ascii(self):
         req = Request.blank(
             '/sda1/p/a/c', environ={'REQUEST_METHOD': 'PUT',
@@ -4309,7 +4565,7 @@ class TestContainerController(unittest.TestCase):
 
         self.controller.__call__({'REQUEST_METHOD': 'GET',
                                   'SCRIPT_NAME': '',
-                                  'PATH_INFO': '\x00',
+                                  'PATH_INFO': '/sda1/p/a/c\xd8\x3e%20/%',
                                   'SERVER_NAME': '127.0.0.1',
                                   'SERVER_PORT': '8080',
                                   'SERVER_PROTOCOL': 'HTTP/1.0',
diff --git a/test/unit/helpers.py b/test/unit/helpers.py
index 67dac477c5..d14e8a38cd 100644
--- a/test/unit/helpers.py
+++ b/test/unit/helpers.py
@@ -212,7 +212,8 @@ def setup_servers(the_object_server=object_server, extra_conf=None):
          obj4srv, obj5srv, obj6srv)
     nl = NullLogger()
     logging_prosv = proxy_logging.ProxyLoggingMiddleware(
-        listing_formats.ListingFilter(prosrv), conf, logger=prosrv.logger)
+        listing_formats.ListingFilter(prosrv, {}, logger=prosrv.logger),
+        conf, logger=prosrv.logger)
     prospa = spawn(wsgi.server, prolis, logging_prosv, nl,
                    protocol=SwiftHttpProtocol,
                    capitalize_response_headers=False)
diff --git a/test/unit/obj/test_server.py b/test/unit/obj/test_server.py
index 23ae40700e..3e5a5f16f7 100644
--- a/test/unit/obj/test_server.py
+++ b/test/unit/obj/test_server.py
@@ -55,6 +55,7 @@ from swift.common.utils import hash_path, mkdirs, normalize_timestamp, \
     NullLogger, storage_directory, public, replication, encode_timestamps, \
     Timestamp
 from swift.common import constraints
+from swift.common.request_helpers import get_reserved_name
 from swift.common.swob import Request, WsgiBytesIO
 from swift.common.splice import splice
 from swift.common.storage_policy import (StoragePolicy, ECStoragePolicy,
@@ -7147,6 +7148,60 @@ class TestObjectController(unittest.TestCase):
             self.assertEqual(errbuf.getvalue(), '')
             self.assertEqual(outbuf.getvalue()[:4], '405 ')
 
+    def test_create_reserved_namespace_object(self):
+        path = '/sda1/p/a/%sc/%so' % (utils.RESERVED_STR, utils.RESERVED_STR)
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal,
+            'Content-Type': 'application/x-test',
+            'Content-Length': 0,
+        })
+        resp = req.get_response(self.object_controller)
+        self.assertEqual(resp.status, '201 Created')
+
+    def test_create_reserved_namespace_object_in_user_container(self):
+        path = '/sda1/p/a/c/%so' % utils.RESERVED_STR
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal,
+            'Content-Type': 'application/x-test',
+            'Content-Length': 0,
+        })
+        resp = req.get_response(self.object_controller)
+        self.assertEqual(resp.status, '400 Bad Request', resp.body)
+        self.assertEqual(resp.body, b'Invalid reserved-namespace object in '
+                         b'user-namespace container')
+
+    def test_other_methods_reserved_namespace_object(self):
+        container = get_reserved_name('c')
+        obj = get_reserved_name('o', 'v1')
+        path = '/sda1/p/a/%s/%s' % (container, obj)
+        req = Request.blank(path, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal,
+            'Content-Type': 'application/x-test',
+            'Content-Length': 0,
+        })
+        resp = req.get_response(self.object_controller)
+        self.assertEqual(resp.status, '201 Created')
+
+        bad_req = Request.blank('/sda1/p/a/c/%s' % obj, method='PUT', headers={
+            'X-Timestamp': next(self.ts).internal})
+        resp = bad_req.get_response(self.object_controller)
+        self.assertEqual(resp.status, '400 Bad Request')
+        self.assertEqual(resp.body, b'Invalid reserved-namespace object '
+                         b'in user-namespace container')
+
+        for method in ('GET', 'POST', 'DELETE'):
+            req.method = method
+            req.headers['X-Timestamp'] = next(self.ts).internal
+            resp = req.get_response(self.object_controller)
+            self.assertEqual(resp.status_int // 100, 2)
+
+            bad_req.method = method
+            req.headers['X-Timestamp'] = next(self.ts).internal
+            resp = bad_req.get_response(self.object_controller)
+            self.assertEqual(resp.status, '400 Bad Request')
+            self.assertEqual(resp.body, b'Invalid reserved-namespace object '
+                             b'in user-namespace container')
+
     def test_not_utf8_and_not_logging_requests(self):
         inbuf = WsgiBytesIO()
         errbuf = StringIO()
@@ -7164,7 +7219,7 @@ class TestObjectController(unittest.TestCase):
 
         env = {'REQUEST_METHOD': method,
                'SCRIPT_NAME': '',
-               'PATH_INFO': '/sda1/p/a/c/\x00%20/%',
+               'PATH_INFO': '/sda1/p/a/c/\xd8\x3e%20/%',
                'SERVER_NAME': '127.0.0.1',
                'SERVER_PORT': '8080',
                'SERVER_PROTOCOL': 'HTTP/1.0',
diff --git a/test/unit/proxy/controllers/test_obj.py b/test/unit/proxy/controllers/test_obj.py
index 4acc962069..19a7050538 100644
--- a/test/unit/proxy/controllers/test_obj.py
+++ b/test/unit/proxy/controllers/test_obj.py
@@ -31,6 +31,7 @@ from eventlet import Timeout
 import six
 from six import StringIO
 from six.moves import range
+from six.moves.urllib.parse import quote
 if six.PY2:
     from email.parser import FeedParser as EmailFeedParser
 else:
@@ -1258,10 +1259,7 @@ class TestReplicatedObjController(CommonObjectControllerMixin,
         log_lines = self.app.logger.get_lines_for_level('error')
         self.assertFalse(log_lines[1:])
         self.assertIn('ERROR with Object server', log_lines[0])
-        if six.PY3:
-            self.assertIn(req.swift_entity_path, log_lines[0])
-        else:
-            self.assertIn(req.swift_entity_path.decode('utf-8'), log_lines[0])
+        self.assertIn(quote(req.swift_entity_path), log_lines[0])
         self.assertIn('re: Expect: 100-continue', log_lines[0])
 
     def test_PUT_get_expect_errors_with_unicode_path(self):
@@ -1305,11 +1303,7 @@ class TestReplicatedObjController(CommonObjectControllerMixin,
             log_lines = self.app.logger.get_lines_for_level('error')
             self.assertFalse(log_lines[1:])
             self.assertIn('ERROR with Object server', log_lines[0])
-            if six.PY3:
-                self.assertIn(req.swift_entity_path, log_lines[0])
-            else:
-                self.assertIn(req.swift_entity_path.decode('utf-8'),
-                              log_lines[0])
+            self.assertIn(quote(req.swift_entity_path), log_lines[0])
             self.assertIn('Trying to write to', log_lines[0])
 
         do_test(Exception('Exception while sending data on connection'))
diff --git a/test/unit/proxy/test_server.py b/test/unit/proxy/test_server.py
index 7a481509dc..3b509c380d 100644
--- a/test/unit/proxy/test_server.py
+++ b/test/unit/proxy/test_server.py
@@ -83,7 +83,7 @@ from swift.common.swob import Request, Response, HTTPUnauthorized, \
     HTTPException, HTTPBadRequest, wsgi_to_str
 from swift.common.storage_policy import StoragePolicy, POLICIES
 import swift.common.request_helpers
-from swift.common.request_helpers import get_sys_meta_prefix
+from swift.common.request_helpers import get_sys_meta_prefix, get_reserved_name
 
 # mocks
 logging.getLogger().addHandler(logging.StreamHandler(sys.stdout))
@@ -698,6 +698,29 @@ class TestProxyServer(unittest.TestCase):
         self.assertEqual(sorted(resp.headers['Allow'].split(', ')), [
             'DELETE', 'GET', 'HEAD', 'OPTIONS', 'POST', 'PUT', 'UPDATE'])
 
+    def test_internal_reserved_name_request(self):
+        # set account info
+        fake_cache = FakeMemcache()
+        fake_cache.store[get_cache_key('a')] = {'status': 200}
+        app = proxy_server.Application({}, fake_cache,
+                                       container_ring=FakeRing(),
+                                       account_ring=FakeRing())
+        # build internal container request
+        container = get_reserved_name('c')
+        req = Request.blank('/v1/a/%s' % container)
+        app.update_request(req)
+
+        # try client request to reserved name
+        resp = app.handle_request(req)
+        self.assertEqual(resp.status_int, 412)
+        self.assertEqual(resp.body, b'Invalid UTF8 or contains NULL')
+
+        # set backend header
+        req.headers['X-Backend-Allow-Reserved-Names'] = 'true'
+        with mocked_http_conn(200):
+            resp = app.handle_request(req)
+        self.assertEqual(resp.status_int, 200)
+
     def test_calls_authorize_allow(self):
         called = [False]
 
@@ -10405,7 +10428,8 @@ class TestAccountControllerFakeGetResponse(unittest.TestCase):
         self.app = listing_formats.ListingFilter(
             proxy_server.Application(conf, FakeMemcache(),
                                      account_ring=FakeRing(),
-                                     container_ring=FakeRing()))
+                                     container_ring=FakeRing()),
+            {})
         self.app.app.memcache = FakeMemcacheReturnsNone()
 
     def test_GET_autocreate_accept_json(self):
@@ -10823,7 +10847,8 @@ class TestSocketObjectVersions(unittest.TestCase):
                         _test_servers[0], conf,
                         logger=_test_servers[0].logger), {}),
                 {}
-            )
+            ),
+            {}, logger=_test_servers[0].logger
         )
         self.coro = spawn(wsgi.server, prolis, prosrv, NullLogger(),
                           protocol=SwiftHttpProtocol)