Viewing file:  pAclObject.py (7.08 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
########################################################################
# $Header: /var/local/cvsroot/4Suite/Ft/Server/Server/Drivers/pAclObject.py,v 1.12 2005/03/28 11:31:13 mbrown Exp $
"""
ACL implementation
Copyright 2005 Fourthought, Inc. (USA).
Detailed license and copyright information: http://4suite.org/COPYRIGHT
Project home, documentation, distributions: http://4suite.org/
"""
from Ft.Server import FTSERVER_NAMESPACE
from Ft.Server.Common import AclConstants
from Ft.Server.Server import FtServerServerException, Error
from Ft.Server.Server.Drivers import PathImp
from Ft.Xml import EMPTY_NAMESPACE
from Ft.Xml import XPath
class AclObject:
def __init__(self,driver, isAnonymous, aclIdents):
self.__isAnonymous = isAnonymous
self.__aclIdents = aclIdents
self.__driver = driver
self.__aclCache = {}
return
def clearCache(self):
self.__aclCache = {}
return
def verifyFetch(self,path):
"""Verify that a resource can be fetched.
1. Verify that the user has traverse privileges to all of the parents
2. Verify that the user can read the last container"""
self.__verifyAclSteps(path.absolutePaths, AclConstants.READ_ACCESS, 1)
def verifyCreate(self, path):
"""Verify that a resource can be created.
1. Verify that the user has traverse privileges to all of the grand parents
2. Verify that the user can write the parent"""
self.__verifyAclSteps(path.absolutePaths[:-1], AclConstants.WRITE_ACCESS, 1)
def verifyWrite(self, path):
"""Verify that a resource can be written to.
1. Verify that the user has traverse privileges to all of the parents
2. Verify that the user can write the object"""
self.__verifyAclSteps(path.absolutePaths, AclConstants.WRITE_ACCESS, 1)
def verifyDelete(self, path):
"""Verify that a resource can be deleted.
1. Verify that the user has traverse privileges to all of the parents
2. Verify that the user can delete the object"""
self.__verifyAclSteps(path.absolutePaths, AclConstants.DELETE_ACCESS, 1)
def verifyChangePermissions(self, path):
"""Verify that there are change permission on the object"""
self.__verifyAclSteps(path.absolutePaths, AclConstants.CHANGE_PERMISSIONS_ACCESS, 0)
def verifyAcl(self, path, access, verifyTraverse):
self.__verifyAclSteps(path.absolutePaths, access, verifyTraverse)
def verifyAddMember(self, group, user):
"""Verify that the user can be added to the group"""
self.verifyWrite(group)
self.verifyFetch(user)
def getAclIdentifiers(self):
return self.__aclIdents
def getAcl(self, path, access):
acl,owner = self.__getResourceAcl(path)
if not access:
return acl
if access in acl:
return acl[access]
if path.absolutePath == '/':
return {}
if path.displayPath == '/':
#Need to by pass the doc root
p = PathImp.CreateInitialPath('/', path._driver)
path = p.normalize(path._documentRoot.absolutePath)
return self.getAcl(path.getParentPath(), access)
def __verifyAclSteps(self, steps, finalAccess, verifyTraverse):
"""Verify a specific access type on a resourc's md"""
if steps[-1] in self.__aclCache:
t = self.__aclCache[steps[-1]].get(finalAccess)
if t == AclConstants.ALLOWED: return
if t == AclConstants.DENIED:
raise FtServerServerException(Error.PERMISSION_DENIED,
level=finalAccess,
path=steps[-1])
#Sorry, not cached
cur = None
if verifyTraverse:
for s in steps[:-1]:
cur = self.__verifyAcl(s, AclConstants.EXECUTE_ACCESS, cur)
self.__verifyAcl(steps[-1], finalAccess, cur)
def __verifyAcl(self, path, access, parentAcl):
"""Verify one step in the ACL chain"""
if AclConstants.SUPER_USER_GROUP_NAME in self.__aclIdents: return 1
if path in self.__aclCache:
t = self.__aclCache[path].get(access)
if t == AclConstants.ALLOWED: return
if t == AclConstants.DENIED:
raise FtServerServerException(Error.PERMISSION_DENIED,
level=access,
path=path)
acl,owner = self.__getResourceAcl(PathImp.QuickCreate(path))
if parentAcl is not None:
parentAcl.update(acl)
acl = parentAcl
if access not in acl and parentAcl is None:
testPath = path
while 1:
if testPath == '/':
break
#Calculate the parent path
testPath = '/'.join(testPath.split('/')[:-1])
if not testPath:
testPath = '/'
parentAcl,o = self.__getResourceAcl(PathImp.QuickCreate(testPath))
parentAcl.update(acl)
acl = parentAcl
if access in acl:
break
cur = acl.get(access,{})
#Resolve any owner ids
if AclConstants.OWNER in cur and owner not in cur:
cur[owner] = cur[AclConstants.OWNER]
#See if we are explicitly disallowed
for a in self.__aclIdents:
if not cur.get(a, 1):
self.__setCache(path, access, AclConstants.DENIED)
raise FtServerServerException(Error.PERMISSION_DENIED,
level=access,
path=path)
#See if we are allowed
for a in self.__aclIdents:
if cur.get(a, 0):
self.__setCache(path, access, AclConstants.ALLOWED)
return acl
self.__setCache(path, access, AclConstants.DENIED)
raise FtServerServerException(Error.PERMISSION_DENIED,
level=access,
path=path)
def __setCache(self,path,access,allowed):
if path not in self.__aclCache:
self.__aclCache[path] = {}
self.__aclCache[path][access] = allowed
aclExpression = XPath.Compile('/ftss:MetaData/ftss:Acl/ftss:Access')
ownerExpression = XPath.Compile('string(/ftss:MetaData/ftss:Owner)')
def __getResourceAcl(self,path):
con = self.__driver.getContext(path.normalize('.;metadata;no-traverse'))
acl = {}
for access in self.aclExpression.evaluate(con):
typ = access.getAttributeNS(EMPTY_NAMESPACE,'type')
allowed = access.getAttributeNS(EMPTY_NAMESPACE,'allowed')
ident = access.getAttributeNS(EMPTY_NAMESPACE,'ident')
if typ not in acl:
acl[typ] = {}
if allowed and int(allowed):
acl[typ][ident] = AclConstants.ALLOWED
else:
acl[typ][ident] = AclConstants.DENIED
owner = self.ownerExpression.evaluate(con)
return acl,owner
|