Merge lp:~gaurav-gangalwar/swift/object_acl into lp:~hudson-openstack/swift/trunk

[Gaurav Gangalwar]

Having container level acl will enforce those acl on all the objects in that container.
I think we should have acl support at object level, to provide more fine grain acl scheme.
Here is what i am proposing.

Permission When granted on a Container When granted on an object

READ Allows to list the objects in the Allows to read the object
         container and read container data and its metadata
  metadata.

WRITE Allows to list create, overwrite, Allows to write the object
  delete any object in the container metadata.
  and write the container metadata.

[Gaurav Gangalwar (gaurav-gangalwar)]

When granted on a Container:->
READ Allows to list the objects in the
         container and read container
  metadata.

WRITE Allows to list, create, overwrite,
  delete any object in the container
  and write the container metadata.

When granted on an object:->
READ Allows to read the object data and its metadata

WRITE Allows to write the object metadata.

[John Dickinson (notmyname)]

I haven't reviewed all of the code yet, but I notice that there aren't any tests provided. This code will need unit tests before it can be accepted.

review: Needs Fixing

[Gaurav Gangalwar (gaurav-gangalwar)]

Yes i am working on unit test cases, will submit it soon.

Regards,
Gaurav

On Mon, Aug 1, 2011 at 7:13 PM, John Dickinson <<email address hidden>
> wrote:

> Review: Needs Fixing
> I haven't reviewed all of the code yet, but I notice that there aren't any
> tests provided. This code will need unit tests before it can be accepted.
> --
> https://code.launchpad.net/~gaurav-gangalwar/swift/object_acl/+merge/70003<https://code.launchpad.net/%7Egaurav-gangalwar/swift/object_acl/+merge/70003>
> You are the owner of lp:~gaurav-gangalwar/swift/object_acl.
>

[lp:~gaurav-gangalwar/swift/object_acl]

336. By <email address hidden> on 2011-08-05

Object acl support fixes.

[Gaurav Gangalwar (gaurav-gangalwar)]

I made some fixes and updated functional tests.
There are some failures in existing unit tests, still figuring them out, will update soon.

[John Dickinson (notmyname)]

In large clusters, it seems that checking the object info could blow through memcache pretty quickly. Can you add the ability to turn this feature on or off with a config variable?

[Gaurav Gangalwar (gaurav-gangalwar)]

Ok, i am thinking about these options,

==>Give object-acl as option.
==>Give memcache for object-acl as option.
==>Reduce the expiry time for objects acl.

Which one is more preferrable?

Also i have one doubt, do we recommend mutiple proxy servers in large
clusters for load balancing?
If that is the case, do we recommend to use a common memcahe server for all
the proxies?

Regards,
Gaurav

On Fri, Aug 5, 2011 at 6:57 PM, John Dickinson <email address hidden> wrote:

> In large clusters, it seems that checking the object info could blow
> through memcache pretty quickly. Can you add the ability to turn this
> feature on or off with a config variable?
> --
> https://code.launchpad.net/~gaurav-gangalwar/swift/object_acl/+merge/70003<https://code.launchpad.net/%7Egaurav-gangalwar/swift/object_acl/+merge/70003>
> You are the owner of lp:~gaurav-gangalwar/swift/object_acl.
>

[John Dickinson (notmyname)]

Yes, large clusters have many proxy servers and all share the same memcached pool.

In large clusters (with billions of objects), checking for the object info for every object as its requested fills up the cache. What should probably happen is that large deployments should be able to turn this off so that the object info is never even looked at (and therefore never cached).

[lp:~gaurav-gangalwar/swift/object_acl]

337. By <email address hidden> on 2011-08-07

Make object_acl feature configurable.

[Gaurav Gangalwar (gaurav-gangalwar)]

Change to make object_acl configurable submitted.

[gholt (gholt)]

The object acl seems to override the container acl; so I could grant Susan write access to my container and then she could create objects that I could not delete or overwrite, even though I have write access to the container. Also, I could seemingly take away Susan's access to a container, but she might be able to access objects in the container still.

This feature would need to be off by default. If a large cluster operator were to upgrade to this code and not realize they had to add "object_acl = false" to their configuration, they'd have some difficulties. Not only can this have a dramatic impact on memcache usage, but it also will up the number of backend requests.

Why is the "object_size" memcached?

In the object GETorHEAD the acls are retrieved even if 'swift.authorize' not in req.environ. This isn't often used, but was added to allow short-circuiting such extra queries to lower memcache and backend requests when they aren't needed.

The logic for setting req.acl is different than before. It used to be that write access could be granted to a container without granting read access (blind drop box type-thing). In the object GETorHEAD, now write access also grants read access.

With segmented objects, why the change to giving the end user all headers from the backend server?

In the container GETorHEAD, the change to calling container_info before performing the backend container request causes extra requests that aren't needed. The way it was coded before, it just always did the container backend request and returned the response depending on the acl values in the backend response.

The code now requires write access to a container to delete or post to that container, whereas before it just required the swift.authorize method to allow it. This change means a user that has full access to an account but isn't explicitly in the write list for a container can't do anything about it.

Whereas there are some new functional tests, there aren't any new unit tests and many existing unit tests are failing, but I think you were already aware of that part and working on it.

[Gaurav Gangalwar (gaurav-gangalwar)]

On Fri, Aug 12, 2011 at 7:40 PM, gholt <email address hidden> wrote:

> The object acl seems to override the container acl; so I could grant Susan
> write access to my container and then she could create objects that I could
> not delete or overwrite, even though I have write access to the container.
> Also, I could seemingly take away Susan's access to a container, but she
> might be able to access objects in the container still.
>

Write object-acl will only control POST and GETorHEAD on the object, You can
DELETE or PUT objects inside the container if you have write access or owner
of the account.

>
> This feature would need to be off by default. If a large cluster operator
> were to upgrade to this code and not realize they had to add "object_acl =
> false" to their configuration, they'd have some difficulties. Not only can
> this have a dramatic impact on memcache usage, but it also will up the
> number of backend requests.
>

Ok, in large clusters if its enabled then it might lead to excess memcache
usage, but it will not affect any functionality if object-acl is not set.

>
> Why is the "object_size" memcached?
>

object_size is not used for now, will remove it.

>
> In the object GETorHEAD the acls are retrieved even if 'swift.authorize'
> not in req.environ. This isn't often used, but was added to allow
> short-circuiting such extra queries to lower memcache and backend requests
> when they aren't needed.
>

In which case swift.authorize will not be in the req.environ? I thought auth
subsystem will set it for each req.

>
> The logic for setting req.acl is different than before. It used to be that
> write access could be granted to a container without granting read access
> (blind drop box type-thing). In the object GETorHEAD, now write access also
> grants read access.
>

Yes, allowing write and not allowing read is not supposed to be the expected
behaviour, so write acl will have read/write access to the object.

>
> With segmented objects, why the change to giving the end user all headers
> from the backend server?
>

This filtering is done by the backend servers, so it was redundant here,
thats why.

>
> In the container GETorHEAD, the change to calling container_info before
> performing the backend container request causes extra requests that aren't
> needed. The way it was coded before, it just always did the container
> backend request and returned the response depending on the acl values in the
> backend response.
>

In case of unauthorized it will be only one request. I thought of keeping a
single interface for memcache, could we remove memcache updation code from
Container ops?

>
> The code now requires write access to a container to delete or post to that
> container, whereas before it just required the swift.authorize method to
> allow it. This change means a user that has full access to an account but
> isn't explicitly in the write list for a container can't do anything about
> it.
>

No, if you are the owner of account, authorize will not check for any access
list, it will give you all access to the containers and objects.

>
> Whereas there are some new functional tests, there aren't any new unit
> tests and many ex...

[gholt (gholt)]

> Write object-acl will only control POST and GETorHEAD on the
> object, You can DELETE or PUT objects inside the container if you
> have write access or owner of the account.

Okay, that's a little weird that POST acts differently; in fact,
POST-as-COPY would act like PUT, so only POST-not-as-COPY would have
the new effect. We've done our best to this point to make
POST-as-COPY and POST-not-as-COPY act the same way. If I have write
access to a container, shouldn't I be able to POST to any object in
that container?

Also, is it true that if my access to a container is taken away I'd
still be able to access any objects I explicitly set ACLs for myself
on within that container?

> Ok, in large clusters if its enabled then it might lead to excess
> memcache usage, but it will not affect any functionality if
> object-acl is not set.

It should not affect functionality, true (it does currently, but
that's discussed elsewhere). But extra memcache queries and extra
object HEADs for no reason is bad when we're already doing thousands
of requests per second.

Along the same vein, with this code the first GET or HEAD of an
object will do a cache_get + HEAD + cache_set + GET|HEAD and a
second request would do a cache_get + GET|HEAD. That should be able
to be reduced to a GET|HEAD + cache_set and the ACL checked on that
response and thrown away if access is denied.

> In which case swift.authorize will not be in the req.environ? I
> thought auth subsystem will set it for each req.

The auth subsystem does not need to set swift.authorize to authorize
a request, it only needs to set swift.authorize if it wants the
proxy code to execute that callback to check access. If
swift.authorize does not exist, access is granted.

As an extreme example, you can actually run with no auth middleware
and all requests will be granted. As a less extreme example, auth
middleware may clear swift.authorize when it has enough information
up front to know the request should be granted.

Honestly, I'm not too worried about this since no one uses it, but
it might become important to someone someday and there didn't seem
to be a good reason to drop the behavior.

> Yes, allowing write and not allowing read is not supposed to be
> the expected behaviour, so write acl will have read/write access
> to the object.

I'm not sure what you mean by "not supposed to be the expected
behaviour". It was the expected behavior before.

> This filtering is done by the backend servers, so it was redundant
> here, thats why.

It is redundant, but is a safety mechanism to ensure we don't
accidentally return headers we didn't mean to. It wastes very little
CPU time, so I'm not sure why we'd want to change what was already
there.

>> In the container GETorHEAD, the change to calling container_info
>> before performing the backend container request causes extra
>> requests that aren't needed.

> In case of unauthorized it will be only one request. I thought of
> keeping a single interface for memcache, could we remove memcache
> updation code from Container ops?

What I meant was that the previous code for container GETorHEAD did
the container request, sent the ACL part to memcache, checked the
ACL a...

[Gaurav Gangalwar (gaurav-gangalwar)]

On Tue, Aug 16, 2011 at 7:49 PM, gholt <email address hidden> wrote:

> > Write object-acl will only control POST and GETorHEAD on the
> > object, You can DELETE or PUT objects inside the container if you
> > have write access or owner of the account.
>
> Okay, that's a little weird that POST acts differently; in fact,
> POST-as-COPY would act like PUT, so only POST-not-as-COPY would have
> the new effect. We've done our best to this point to make
> POST-as-COPY and POST-not-as-COPY act the same way. If I have write
> access to a container, shouldn't I be able to POST to any object in
> that container?
>
Write access to container will allow POST-as-COPY but not POST-not-as-COPY.
The reason for this is,
In general PUT operation on objects affect the container metadata, so it
should have write access to the container, but POST operations on object
doesn't affect the container, so it should require only write access to the
object since it will only affect object metadata. In POST-as-COPY its
equivalent to PUT.
We can allow PUT which will only overwrite the existing the object (not
creating new object inside container) just by checking write permission on
object, no need to check container write acl for this, so this way
POST-as-COPY and POST-not-as-COPY will work the same way.

>
> Also, is it true that if my access to a container is taken away I'd
> still be able to access any objects I explicitly set ACLs for myself
> on within that container?
>

Yes you can do GETorHEAD/POST but won't be able to PUT/DELETE on those
objects.

>
> > Ok, in large clusters if its enabled then it might lead to excess
> > memcache usage, but it will not affect any functionality if
> > object-acl is not set.
>
> It should not affect functionality, true (it does currently, but
> that's discussed elsewhere). But extra memcache queries and extra
> object HEADs for no reason is bad when we're already doing thousands
> of requests per second.
>

If object_acl is disabled, object_info will return the acl for container, so
it will work the same way as it was earlier.

>
> Along the same vein, with this code the first GET or HEAD of an
> object will do a cache_get + HEAD + cache_set + GET|HEAD and a
> second request would do a cache_get + GET|HEAD. That should be able
> to be reduced to a GET|HEAD + cache_set and the ACL checked on that
> response and thrown away if access is denied.
>

Ok i will optimize this code to do minimal requests.

>
> > In which case swift.authorize will not be in the req.environ? I
> > thought auth subsystem will set it for each req.
>
> The auth subsystem does not need to set swift.authorize to authorize
> a request, it only needs to set swift.authorize if it wants the
> proxy code to execute that callback to check access. If
> swift.authorize does not exist, access is granted.
>
> As an extreme example, you can actually run with no auth middleware
> and all requests will be granted. As a less extreme example, auth
> middleware may clear swift.authorize when it has enough information
> up front to know the request should be granted.
>
> Honestly, I'm not too worried about this since no one uses it, but
> it might become important to someone some...

[gholt (gholt)]

The write and no read thing was from someone wanting to use a container as an "upload here" type thing. Where others could drop files that would then be approved and moved to official locations later. In that use case, you don't want the uploaders to be able to see the unapproved files.

The more I think about it, such decisions like "does write give read?" "does container access override object access?" etc. should be made by the auth middleware, not the proxy. The proxy should just provide the information. In other words, the proxy should send up the read and write acls for objects and containers and let the auth middleware figure out what to do.

[John Dickinson (notmyname)]

(Just today) swift has moved to using github for code hosting and gerrit for code reviews. Instructions on setting it up are at http://wiki.openstack.org/GerritWorkflow. Please resubmit your patch there.

I'm disapproving this merge only because it needs to be made through gerrit now.

review: Disapprove

Unmerged revisions

337. By <email address hidden> on 2011-08-07

Make object_acl feature configurable.

336. By <email address hidden> on 2011-08-05

Object acl support fixes.

335. By <email address hidden> on 2011-08-01

Object acl support1.

334. By <email address hidden> on 2011-07-30

Object acl support.

=== modified file 'etc/proxy-server.conf-sample'
--- etc/proxy-server.conf-sample	2011-06-16 22:05:03 +0000
+++ etc/proxy-server.conf-sample	2011-08-07 06:38:25 +0000
@@ -48,6 +48,9 @@
 # If set to 'true' authorized accounts that do not yet exist within the Swift
 # cluster will be automatically created.
 # account_autocreate = false
+# If set to false this will disable the object_acl feature, in large cluster
+# enviornment with limited memcache, it is recommended to disable object_acl
+# object_acl = true
 
 [filter:tempauth]
 use = egg:swift#tempauth

=== modified file 'swift/obj/server.py'
--- swift/obj/server.py	2011-08-01 16:08:55 +0000
+++ swift/obj/server.py	2011-08-07 06:38:25 +0000
@@ -363,7 +363,7 @@
         self.slow = int(conf.get('slow', 0))
         self.bytes_per_sync = int(conf.get('mb_per_sync', 512)) * 1024 * 1024
         default_allowed_headers = 'content-encoding, x-object-manifest, ' \
-                                  'content-disposition'
+                                  'content-disposition, x-object-read, x-object-write'
         self.allowed_headers = set(i.strip().lower() for i in \
                 conf.get('allowed_headers', \
                 default_allowed_headers).split(',') if i.strip() and \

=== modified file 'swift/proxy/server.py'
--- swift/proxy/server.py	2011-07-14 18:43:08 +0000
+++ swift/proxy/server.py	2011-08-07 06:38:25 +0000
@@ -105,6 +105,8 @@
 def get_container_memcache_key(account, container):
     return 'container/%s/%s' % (account, container)
 
+def get_object_memcache_key(account, container, obj):
+    return 'object/%s/%s/%s' % (account, container, obj)
 
 class SegmentedIterable(object):
     """
@@ -493,6 +495,97 @@
             return partition, nodes, read_acl, write_acl, sync_key
         return None, None, None, None, None
 
+    def object_info(self, account, container, obj):
+        """
+        Get object information and thusly verify object existance.
+        This will also make a call to container_info to verify that the
+        container exists.
+
+        :param account: account name for the container
+        :param container: container name for the object
+        :param object: object name to look up.
+        :returns: tuple of (object partition, object nodes, object
+                  read acl, object write acl) or (None, None, None, None) if
+                  the object does not exist
+        """
+        partition, nodes = self.app.object_ring.get_nodes(
+                                            account, container, obj)
+
+        _junk, cnodes, read_acl, write_acl, _junk = \
+               self.container_info(account, container)
+        if not cnodes:
+            return None, None, None, None
+
+        if not self.app.object_acl:
+            return partition, nodes, read_acl, write_acl
+        
+        path = '/%s/%s/%s' % (account, container, obj)
+        if self.app.memcache:
+            cache_key = get_object_memcache_key(account, container, obj)
+            cache_value = self.app.memcache.get(cache_key)
+            
+            if isinstance(cache_value, dict):
+                status = cache_value['status']
+                read_acl = cache_value['read_acl']
+                write_acl = cache_value['write_acl']
+                if status == 200:
+                    return partition, nodes, read_acl, write_acl
+                elif status == 404:
+                    return None, None, None, None
+        
+        result_code = 0
+        read_acl = None
+        write_acl = None
+        obj_size = None
+        attempts_left = self.app.object_ring.replica_count
+        headers = {'x-trans-id': self.trans_id}
+        for node in nodes:
+            try:
+                with ConnectionTimeout(self.app.conn_timeout):
+                    conn = http_connect(node['ip'], node['port'],
+                            node['device'], partition, 'HEAD', path, headers)
+                with Timeout(self.app.node_timeout):
+                    resp = conn.getresponse()
+                    body = resp.read()
+                    if 200 <= resp.status <= 299:
+                        result_code = 200
+                        read_acl = resp.getheader('x-object-read')
+                        write_acl = resp.getheader('x-object-write')
+                        obj_size = \
+                            resp.getheader('Content-Length')
+                        break
+                    elif resp.status == 404:
+                        if result_code == 0:
+                            result_code = 404
+                        elif result_code != 404:
+                            result_code = -1
+                    elif resp.status == 507:
+                        self.error_limit(node)
+                        continue
+                    else:
+                        result_code = -1
+                        attempts_left -= 1
+                        if attempts_left <= 0:
+                            break
+            except (Exception, TimeoutError):
+                self.exception_occurred(node, _('Object'),
+                    _('Trying to get object info for %s') % path)
+        if self.app.memcache and result_code in (200, 404):
+            if result_code == 200:
+                cache_timeout = self.app.recheck_container_existence
+            else:
+                cache_timeout = self.app.recheck_container_existence * 0.1
+            self.app.memcache.set(cache_key,
+                                  {'status': result_code,
+                                   'read_acl': read_acl,
+                                   'write_acl': write_acl,
+                                   'object_size': obj_size},
+                                  timeout=cache_timeout)
+        if result_code == 200:
+            return partition, nodes, read_acl, write_acl
+        return None, None, None, None    
+
+
     def iter_nodes(self, partition, nodes, ring):
         """
         Node iterator that will first iterate over the normal nodes for a
@@ -584,6 +677,19 @@
         resp.status = '503 Internal Server Error'
         return resp
 
+    def clean_acls(self, req):
+        if 'swift.clean_acl' in req.environ:
+            for header in ('x-container-read', 'x-container-write',
+            'x-object-read', 'x-object-write'):
+                if header in req.headers:
+                    try:
+                        req.headers[header] = \
+                            req.environ['swift.clean_acl'](header,
+                                                           req.headers[header])
+                    except ValueError, err:
+                        return HTTPBadRequest(request=req, body=str(err))
+        return None
+
     @public
     def GET(self, req):
         """Handler for HTTP GET requests."""
@@ -728,18 +834,26 @@
 
     def GETorHEAD(self, req):
         """Handle HTTP GET or HEAD requests."""
+        partition, nodes, read_acl, write_acl = \
+                  self.object_info(self.account_name, self.container_name, \
+                                 self.object_name)
+        if not nodes:
+            return HTTPNotFound(request=req)
         if 'swift.authorize' in req.environ:
-            req.acl = \
-                self.container_info(self.account_name, self.container_name)[2]
+            if read_acl and write_acl:
+                req.acl = read_acl + ',' + write_acl
+            else:
+                req.acl = read_acl or write_acl
+                
             aresp = req.environ['swift.authorize'](req)
             if aresp:
                 return aresp
-        partition, nodes = self.app.object_ring.get_nodes(
-            self.account_name, self.container_name, self.object_name)
+        
         shuffle(nodes)
         resp = self.GETorHEAD_base(req, _('Object'), partition,
                 self.iter_nodes(partition, nodes, self.app.object_ring),
                 req.path_info, self.app.object_ring.replica_count)
+        
         # If we get a 416 Requested Range Not Satisfiable we have to check if
         # we were actually requesting a manifest object and then redo the range
         # request on the whole object.
@@ -754,11 +868,24 @@
             resp = resp2
             req.range = req_range
 
+        if not req.environ.get('swift_owner', False):
+            for key in ('x-object-read', 'x-object-write'):
+                if key in resp.headers:
+                    del resp.headers[key]
+
         if 'x-object-manifest' in resp.headers:
             lcontainer, lprefix = \
                 resp.headers['x-object-manifest'].split('/', 1)
-            lpartition, lnodes = self.app.container_ring.get_nodes(
-                self.account_name, lcontainer)
+            lpartition, lnodes, lread_acl, lwrite_acl, _junk = \
+                        self.container_info(self.account_name, lcontainer)
+            if lread_acl and lwrite_acl:
+                req.acl = lread_acl + ',' + lwrite_acl
+            else:
+                req.acl = lread_acl or lwrite_acl
+            if 'swift.authorize' in req.environ:
+                    aresp = req.environ['swift.authorize'](req)
+                    if aresp:
+                        return aresp
             marker = ''
             listing = []
             while True:
@@ -774,11 +901,6 @@
                     lresp.headers['X-Object-Manifest'] = \
                         resp.headers['x-object-manifest']
                     return lresp
-                if 'swift.authorize' in req.environ:
-                    req.acl = lresp.headers.get('x-container-read')
-                    aresp = req.environ['swift.authorize'](req)
-                    if aresp:
-                        return aresp
                 sublisting = json.loads(lresp.body)
                 if not sublisting:
                     break
@@ -805,13 +927,6 @@
                             raise Exception(_('Object manifest GET could not '
                                 'continue listing: %s %s') %
                                 (req.path, lreq.path))
-                        if 'swift.authorize' in req.environ:
-                            req.acl = lresp.headers.get('x-container-read')
-                            aresp = req.environ['swift.authorize'](req)
-                            if aresp:
-                                raise Exception(_('Object manifest GET could '
-                                    'not continue listing: %s %s') %
-                                    (req.path, aresp))
                         sublisting = json.loads(lresp.body)
                         if not sublisting:
                             break
@@ -823,8 +938,7 @@
                     'X-Object-Manifest': resp.headers['x-object-manifest'],
                     'Content-Type': resp.content_type}
                 for key, value in resp.headers.iteritems():
-                    if key.lower().startswith('x-object-meta-'):
-                        headers[key] = value
+                    headers[key] = value
                 resp = Response(headers=headers, request=req,
                                 conditional_response=True)
                 if req.method == 'HEAD':
@@ -865,8 +979,7 @@
                     'Content-Length': content_length,
                     'ETag': etag}
                 for key, value in resp.headers.iteritems():
-                    if key.lower().startswith('x-object-meta-'):
-                        headers[key] = value
+                    headers[key] = value
                 resp = Response(headers=headers, request=req,
                                 conditional_response=True)
                 resp.app_iter = SegmentedIterable(self, lcontainer, listing,
@@ -874,7 +987,7 @@
                 resp.content_length = content_length
                 resp.last_modified = last_modified
             resp.headers['accept-ranges'] = 'bytes'
-
+            
         return resp
 
     @public
@@ -893,6 +1006,10 @@
     @delay_denial
     def POST(self, req):
         """HTTP POST request handler."""
+        error_response = \
+            self.clean_acls(req) or check_metadata(req, 'object')
+        if error_response:
+            return error_response
         if self.app.object_post_as_copy:
             req.method = 'PUT'
             req.path_info = '/%s/%s/%s' % (self.account_name,
@@ -912,17 +1029,21 @@
             error_response = check_metadata(req, 'object')
             if error_response:
                 return error_response
-            container_partition, containers, _junk, req.acl, _junk = \
+            container_partition, containers, _junk, _junk, _junk = \
                 self.container_info(self.account_name, self.container_name,
                     account_autocreate=self.app.account_autocreate)
+            partition, nodes, _junk, req.acl = \
+                       self.object_info(self.account_name, self.container_name,\
+                                        self.object_name)
+            if not nodes:
+                return HTTPNotFound(request=req)
             if 'swift.authorize' in req.environ:
                 aresp = req.environ['swift.authorize'](req)
                 if aresp:
                     return aresp
             if not containers:
                 return HTTPNotFound(request=req)
-            partition, nodes = self.app.object_ring.get_nodes(
-                self.account_name, self.container_name, self.object_name)
+            
             req.headers['X-Timestamp'] = normalize_timestamp(time.time())
             headers = []
             for container in containers:
@@ -931,6 +1052,11 @@
                 nheaders['X-Container-Partition'] = container_partition
                 nheaders['X-Container-Device'] = container['device']
                 headers.append(nheaders)
+            if self.app.memcache:
+                cache_key = get_object_memcache_key(self.account_name,
+                                                self.container_name,
+                                                self.object_name)
+                self.app.memcache.delete(cache_key)
             return self.make_requests(req, self.app.object_ring,
                     partition, 'POST', req.path_info, headers)
 
@@ -970,6 +1096,10 @@
     @delay_denial
     def PUT(self, req):
         """HTTP PUT request handler."""
+        error_response = \
+            self.clean_acls(req) or check_metadata(req, 'object')
+        if error_response:
+            return error_response
         (container_partition, containers, _junk, req.acl,
          req.environ['swift_sync_key']) = \
             self.container_info(self.account_name, self.container_name,
@@ -982,6 +1112,7 @@
             return HTTPNotFound(request=req)
         partition, nodes = self.app.object_ring.get_nodes(
             self.account_name, self.container_name, self.object_name)
+        
         # Used by container sync feature
         if 'x-timestamp' in req.headers:
             try:
@@ -1176,6 +1307,11 @@
             # reset the bytes, since the user didn't actually send anything
             req.bytes_transferred = 0
         resp.last_modified = float(req.headers['X-Timestamp'])
+        if self.app.memcache:
+                cache_key = get_object_memcache_key(self.account_name,
+                                                self.container_name,
+                                                self.object_name)
+                self.app.memcache.delete(cache_key)
         return resp
 
     @public
@@ -1211,6 +1347,11 @@
             nheaders['X-Container-Partition'] = container_partition
             nheaders['X-Container-Device'] = container['device']
             headers.append(nheaders)
+        if self.app.memcache:
+                cache_key = get_object_memcache_key(self.account_name,
+                                                self.container_name,
+                                                self.object_name)
+                self.app.memcache.delete(cache_key)
         return self.make_requests(req, self.app.object_ring,
                 partition, 'DELETE', req.path_info, headers)
 
@@ -1257,24 +1398,22 @@
         self.account_name = unquote(account_name)
         self.container_name = unquote(container_name)
 
-    def clean_acls(self, req):
-        if 'swift.clean_acl' in req.environ:
-            for header in ('x-container-read', 'x-container-write'):
-                if header in req.headers:
-                    try:
-                        req.headers[header] = \
-                            req.environ['swift.clean_acl'](header,
-                                                           req.headers[header])
-                    except ValueError, err:
-                        return HTTPBadRequest(request=req, body=str(err))
-        return None
-
     def GETorHEAD(self, req):
         """Handler for HTTP GET/HEAD requests."""
-        if not self.account_info(self.account_name)[1]:
+        part, nodes, read_acl, write_acl, _junk= \
+              self.container_info(self.account_name, self.container_name)
+        if 'swift.authorize' in req.environ:
+            if read_acl and write_acl:
+                req.acl = read_acl + ',' + write_acl
+            else:
+                req.acl = read_acl or write_acl
+                
+            aresp = req.environ['swift.authorize'](req)
+            if aresp:
+                return aresp
+        if not nodes:
             return HTTPNotFound(request=req)
-        part, nodes = self.app.container_ring.get_nodes(
-                        self.account_name, self.container_name)
+        
         shuffle(nodes)
         resp = self.GETorHEAD_base(req, _('Container'), part, nodes,
                 req.path_info, self.app.container_ring.replica_count)
@@ -1291,11 +1430,6 @@
                'container_size': resp.headers.get('x-container-object-count')},
                                   timeout=self.app.recheck_container_existence)
 
-        if 'swift.authorize' in req.environ:
-            req.acl = resp.headers.get('x-container-read')
-            aresp = req.environ['swift.authorize'](req)
-            if aresp:
-                return aresp
         if not req.environ.get('swift_owner', False):
             for key in ('x-container-read', 'x-container-write',
                         'x-container-sync-key', 'x-container-sync-to'):
@@ -1352,6 +1486,7 @@
                 container_partition, 'PUT', req.path_info, headers)
 
     @public
+    @delay_denial
     def POST(self, req):
         """HTTP POST request handler."""
         error_response = \
@@ -1362,8 +1497,14 @@
             autocreate=self.app.account_autocreate)
         if not accounts:
             return HTTPNotFound(request=req)
-        container_partition, containers = self.app.container_ring.get_nodes(
-            self.account_name, self.container_name)
+        container_partition, containers, _junk, req.acl, _junk =\
+                             self.container_info(self.account_name, self.container_name)
+        if not containers:
+            return HTTPNotFound(request=req)
+        if 'swift.authorize' in req.environ:
+            aresp = req.environ['swift.authorize'](req)
+            if aresp:
+                return aresp
         headers = {'X-Timestamp': normalize_timestamp(time.time()),
                    'x-trans-id': self.trans_id}
         headers.update(value for value in req.headers.iteritems()
@@ -1534,6 +1675,7 @@
         self.put_queue_depth = int(conf.get('put_queue_depth', 10))
         self.object_chunk_size = int(conf.get('object_chunk_size', 65536))
         self.client_chunk_size = int(conf.get('client_chunk_size', 65536))
+        self.object_acl = conf.get('object_acl', 'true').lower() in TRUE_VALUES
         self.log_headers = conf.get('log_headers', 'no').lower() in TRUE_VALUES
         self.error_suppression_interval = \
             int(conf.get('error_suppression_interval', 60))

=== modified file 'test/functional/sample.conf'
--- test/functional/sample.conf	2011-03-14 02:56:37 +0000
+++ test/functional/sample.conf	2011-08-07 06:38:25 +0000
@@ -4,6 +4,7 @@
 auth_port = 8080
 auth_ssl = no
 auth_prefix = /auth/
+object_acl = true
 
 # Primary functional test account (needs admin access to the account)
 account = test

=== modified file 'test/functionalnosetests/swift_testing.py'
--- test/functionalnosetests/swift_testing.py	2011-02-24 22:21:14 +0000
+++ test/functionalnosetests/swift_testing.py	2011-08-07 06:38:25 +0000
@@ -17,6 +17,8 @@
 swift_test_auth = os.environ.get('SWIFT_TEST_AUTH')
 swift_test_user = [os.environ.get('SWIFT_TEST_USER'), None, None]
 swift_test_key = [os.environ.get('SWIFT_TEST_KEY'), None, None]
+object_acl = conf.get('object_acl', 'true').lower() in ('yes', 'true', 'on',
+'1')
 
 if conf:
     swift_test_auth = 'http'

=== modified file 'test/functionalnosetests/test_object.py'
--- test/functionalnosetests/test_object.py	2011-07-22 19:58:22 +0000
+++ test/functionalnosetests/test_object.py	2011-08-07 06:38:25 +0000
@@ -7,7 +7,8 @@
 from swift.common.constraints import MAX_META_COUNT, MAX_META_NAME_LENGTH, \
     MAX_META_OVERALL_SIZE, MAX_META_VALUE_LENGTH
 
-from swift_testing import check_response, retry, skip, skip3, swift_test_user
+from swift_testing import check_response, retry, skip, skip3, swift_test_user, \
+object_acl
 
 
 class TestObject(unittest.TestCase):
@@ -158,24 +159,39 @@
             self.assert_(str(err).startswith('No result after '))
 
         def post(url, token, parsed, conn):
-            conn.request('POST', parsed.path + '/' + self.container, '',
+            if object_acl:
+                conn.request('POST', parsed.path + '/' + self.container + '/' + self.obj, '',
+                             {'X-Auth-Token': token,
+                              'X-Object-Read': '.r:*'})
+            else:
+                conn.request('POST', parsed.path + '/' + self.container, '',
                          {'X-Auth-Token': token,
                           'X-Container-Read': '.r:*'})
             return check_response(conn)
         resp = retry(post)
         resp.read()
-        self.assertEquals(resp.status, 204)
+        if object_acl:
+            self.assertEquals(resp.status, 202)
+        else:
+            self.assertEquals(resp.status, 204)
         resp = retry(get)
         resp.read()
         self.assertEquals(resp.status, 200)
 
         def post(url, token, parsed, conn):
-            conn.request('POST', parsed.path + '/' + self.container, '',
-                         {'X-Auth-Token': token, 'X-Container-Read': ''})
+            if object_acl:
+                conn.request('POST', parsed.path + '/' + self.container + '/' + self.obj, '',
+                             {'X-Auth-Token': token, 'X-Object-Read': ''})
+            else:
+                conn.request('POST', parsed.path + '/' + self.container, '',
+                             {'X-Auth-Token': token, 'X-Container-Read': ''})
             return check_response(conn)
         resp = retry(post)
         resp.read()
-        self.assertEquals(resp.status, 204)
+        if object_acl:
+            self.assertEquals(resp.status, 202)
+        else:
+            self.assertEquals(resp.status, 204)
         try:
             resp = retry(get)
             raise Exception('Should not have been able to GET')
@@ -227,7 +243,9 @@
         # verify third account can write "obj1" to shared container
         def put(url, token, parsed, conn):
             conn.request('PUT', '%s/%s/%s' % (parsed.path, shared_container,
-                'obj1'), 'test', {'X-Auth-Token': token})
+                    'obj1'), 'test', {'X-Auth-Token': token,
+                            'X-Object-Read': swift_test_user[2],
+                            'X-Object-Write': swift_test_user[2]})
             return check_response(conn)
         resp = retry(put, use_account=3)
         resp.read()
@@ -401,13 +419,21 @@
 
             # Grant access to the third account
             def post(url, token, parsed, conn):
-                conn.request('POST', '%s/%s' % (parsed.path, self.container),
-                    '', {'X-Auth-Token': token, 'X-Container-Read':
-                    swift_test_user[2]})
+                if object_acl:
+                    conn.request('POST', '%s/%s/manifest' % (parsed.path, self.container),
+                        '', {'X-Auth-Token': token, 'X-Object-Read':
+                        swift_test_user[2]})
+                else:
+                    conn.request('POST', '%s/%s' % (parsed.path, self.container),
+                        '', {'X-Auth-Token': token, 'X-Container-Read':
+                        swift_test_user[2]})
                 return check_response(conn)
             resp = retry(post)
             resp.read()
-            self.assertEquals(resp.status, 204)
+            if object_acl:
+                self.assertEquals(resp.status, 202)
+            else:
+                self.assertEquals(resp.status, 204)
 
             # The third account should be able to get the manifest now
             def get(url, token, parsed, conn):
@@ -475,14 +501,25 @@
             self.assertEquals(resp.status, 403)
 
             # Grant access to the third account
+            # If its be post as copy for object, then it will copy all the
+            # segments in one single object, and also remove the 'manifest'
+            # header, so it will no more be a manifest object.
             def post(url, token, parsed, conn):
-                conn.request('POST', '%s/%s' % (parsed.path, acontainer),
-                    '', {'X-Auth-Token': token, 'X-Container-Read':
-                    swift_test_user[2]})
+                if object_acl:
+                    conn.request('POST', '%s/%s/manifest' % (parsed.path, self.container),
+                        '', {'X-Auth-Token': token, 'X-Object-Read':
+                        swift_test_user[2]})
+                else:
+                    conn.request('POST', '%s/%s' % (parsed.path, acontainer),
+                        '', {'X-Auth-Token': token, 'X-Container-Read':
+                        swift_test_user[2]})
                 return check_response(conn)
             resp = retry(post)
             resp.read()
-            self.assertEquals(resp.status, 204)
+            if object_acl:
+                self.assertEquals(resp.status, 202)
+            else:
+                self.assertEquals(resp.status, 204)
 
             # The third account should be able to get the manifest now
             def get(url, token, parsed, conn):