-##
-# Implements SFA Credentials
-#
-# Credentials are layered on top of certificates, and are essentially a
-# certificate that stores a tuple of parameters.
-##
-
-import xmlrpclib
-
-from sfa.util.faults import MissingDelegateBit, ChildRightsNotSubsetOfParent
-from sfa.trust.certificate import Certificate
-from sfa.trust.gid import GID
-
-##
-# Credential is a tuple:
-# (GIDCaller, GIDObject, LifeTime, Privileges, Delegate)
-#
-# These fields are encoded using xmlrpc into the subjectAltName field of the
-# x509 certificate. Note: Call encode() once the fields have been filled in
-# to perform this encoding.
-
-class CredentialLegacy(Certificate):
- gidCaller = None
- gidObject = None
- lifeTime = None
- privileges = None
- delegate = False
-
- ##
- # Create a Credential object
- #
- # @param create If true, create a blank x509 certificate
- # @param subject If subject!=None, create an x509 cert with the subject name
- # @param string If string!=None, load the credential from the string
- # @param filename If filename!=None, load the credential from the file
-
- def __init__(self, create=False, subject=None, string=None, filename=None):
- Certificate.__init__(self, create, subject, string, filename)
-
- ##
- # set the GID of the caller
- #
- # @param gid GID object of the caller
-
- def set_gid_caller(self, gid):
- self.gidCaller = gid
- # gid origin caller is the caller's gid by default
- self.gidOriginCaller = gid
-
- ##
- # get the GID of the object
-
- def get_gid_caller(self):
- if not self.gidCaller:
- self.decode()
- return self.gidCaller
-
- ##
- # set the GID of the object
- #
- # @param gid GID object of the object
-
- def set_gid_object(self, gid):
- self.gidObject = gid
-
- ##
- # get the GID of the object
-
- def get_gid_object(self):
- if not self.gidObject:
- self.decode()
- return self.gidObject
-
- ##
- # set the lifetime of this credential
- #
- # @param lifetime lifetime of credential
-
- def set_lifetime(self, lifeTime):
- self.lifeTime = lifeTime
-
- ##
- # get the lifetime of the credential
-
- def get_lifetime(self):
- if not self.lifeTime:
- self.decode()
- return self.lifeTime
-
- ##
- # set the delegate bit
- #
- # @param delegate boolean (True or False)
-
- def set_delegate(self, delegate):
- self.delegate = delegate
-
- ##
- # get the delegate bit
-
- def get_delegate(self):
- if not self.delegate:
- self.decode()
- return self.delegate
-
- ##
- # set the privileges
- #
- # @param privs either a comma-separated list of privileges of a Rights object
-
- def set_privileges(self, privs):
- if isinstance(privs, str):
- self.privileges = Rights(string = privs)
- else:
- self.privileges = privs
-
- ##
- # return the privileges as a Rights object
-
- def get_privileges(self):
- if not self.privileges:
- self.decode()
- return self.privileges
-
- ##
- # determine whether the credential allows a particular operation to be
- # performed
- #
- # @param op_name string specifying name of operation ("lookup", "update", etc)
-
- def can_perform(self, op_name):
- rights = self.get_privileges()
- if not rights:
- return False
- return rights.can_perform(op_name)
-
- ##
- # Encode the attributes of the credential into a string and store that
- # string in the alt-subject-name field of the X509 object. This should be
- # done immediately before signing the credential.
-
- def encode(self):
- dict = {"gidCaller": None,
- "gidObject": None,
- "lifeTime": self.lifeTime,
- "privileges": None,
- "delegate": self.delegate}
- if self.gidCaller:
- dict["gidCaller"] = self.gidCaller.save_to_string(save_parents=True)
- if self.gidObject:
- dict["gidObject"] = self.gidObject.save_to_string(save_parents=True)
- if self.privileges:
- dict["privileges"] = self.privileges.save_to_string()
- str = xmlrpclib.dumps((dict,), allow_none=True)
- self.set_data('URI:http://' + str)
-
- ##
- # Retrieve the attributes of the credential from the alt-subject-name field
- # of the X509 certificate. This is automatically done by the various
- # get_* methods of this class and should not need to be called explicitly.
-
- def decode(self):
- data = self.get_data().lstrip('URI:http://')
-
- if data:
- dict = xmlrpclib.loads(data)[0][0]
- else:
- dict = {}
-
- self.lifeTime = dict.get("lifeTime", None)
- self.delegate = dict.get("delegate", None)
-
- privStr = dict.get("privileges", None)
- if privStr:
- self.privileges = Rights(string = privStr)
- else:
- self.privileges = None
-
- gidCallerStr = dict.get("gidCaller", None)
- if gidCallerStr:
- self.gidCaller = GID(string=gidCallerStr)
- else:
- self.gidCaller = None
-
- gidObjectStr = dict.get("gidObject", None)
- if gidObjectStr:
- self.gidObject = GID(string=gidObjectStr)
- else:
- self.gidObject = None
-
- ##
- # Verify that a chain of credentials is valid (see cert.py:verify). In
- # addition to the checks for ordinary certificates, verification also
- # ensures that the delegate bit was set by each parent in the chain. If
- # a delegate bit was not set, then an exception is thrown.
- #
- # Each credential must be a subset of the rights of the parent.
-
- def verify_chain(self, trusted_certs = None):
- # do the normal certificate verification stuff
- Certificate.verify_chain(self, trusted_certs)
-
- if self.parent:
- # make sure the parent delegated rights to the child
- if not self.parent.get_delegate():
- raise MissingDelegateBit(self.parent.get_subject())
-
- # make sure the rights given to the child are a subset of the
- # parents rights
- if not self.parent.get_privileges().is_superset(self.get_privileges()):
- raise ChildRightsNotSubsetOfParent(self.get_subject()
- + " " + self.parent.get_privileges().save_to_string()
- + " " + self.get_privileges().save_to_string())
-
- return
-
- ##
- # Dump the contents of a credential to stdout in human-readable format
- #
- # @param dump_parents If true, also dump the parent certificates
-
- def dump(self, *args, **kwargs):
- print self.dump_string(*args,**kwargs)
-
- def dump_string(self, dump_parents=False):
- result=""
- result += "CREDENTIAL %s\n" % self.get_subject()
-
- result += " privs: %s\n" % self.get_privileges().save_to_string()
-
- gidCaller = self.get_gid_caller()
- if gidCaller:
- result += " gidCaller:\n"
- gidCaller.dump(8, dump_parents)
-
- gidObject = self.get_gid_object()
- if gidObject:
- result += " gidObject:\n"
- result += gidObject.dump_string(8, dump_parents)
-
- result += " delegate: %s" % self.get_delegate()
-
- if self.parent and dump_parents:
- result += "PARENT\n"
- result += self.parent.dump_string(dump_parents)
-
- return result
+#----------------------------------------------------------------------\r
+# Copyright (c) 2008 Board of Trustees, Princeton University\r
+#\r
+# Permission is hereby granted, free of charge, to any person obtaining\r
+# a copy of this software and/or hardware specification (the "Work") to\r
+# deal in the Work without restriction, including without limitation the\r
+# rights to use, copy, modify, merge, publish, distribute, sublicense,\r
+# and/or sell copies of the Work, and to permit persons to whom the Work\r
+# is furnished to do so, subject to the following conditions:\r
+#\r
+# The above copyright notice and this permission notice shall be\r
+# included in all copies or substantial portions of the Work.\r
+#\r
+# THE WORK IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS \r
+# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF \r
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND \r
+# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT \r
+# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, \r
+# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, \r
+# OUT OF OR IN CONNECTION WITH THE WORK OR THE USE OR OTHER DEALINGS \r
+# IN THE WORK.\r
+#----------------------------------------------------------------------\r
+##\r
+# Implements SFA Credentials\r
+#\r
+# Credentials are layered on top of certificates, and are essentially a\r
+# certificate that stores a tuple of parameters.\r
+##\r
+\r
+\r
+import xmlrpclib\r
+\r
+from sfa.util.faults import MissingDelegateBit, ChildRightsNotSubsetOfParent\r
+from sfa.trust.certificate import Certificate\r
+from sfa.trust.gid import GID\r
+\r
+##\r
+# Credential is a tuple:\r
+# (GIDCaller, GIDObject, LifeTime, Privileges, Delegate)\r
+#\r
+# These fields are encoded using xmlrpc into the subjectAltName field of the\r
+# x509 certificate. Note: Call encode() once the fields have been filled in\r
+# to perform this encoding.\r
+\r
+class CredentialLegacy(Certificate):\r
+ gidCaller = None\r
+ gidObject = None\r
+ lifeTime = None\r
+ privileges = None\r
+ delegate = False\r
+\r
+ ##\r
+ # Create a Credential object\r
+ #\r
+ # @param create If true, create a blank x509 certificate\r
+ # @param subject If subject!=None, create an x509 cert with the subject name\r
+ # @param string If string!=None, load the credential from the string\r
+ # @param filename If filename!=None, load the credential from the file\r
+\r
+ def __init__(self, create=False, subject=None, string=None, filename=None):\r
+ Certificate.__init__(self, create, subject, string, filename)\r
+\r
+ ##\r
+ # set the GID of the caller\r
+ #\r
+ # @param gid GID object of the caller\r
+\r
+ def set_gid_caller(self, gid):\r
+ self.gidCaller = gid\r
+ # gid origin caller is the caller's gid by default\r
+ self.gidOriginCaller = gid\r
+\r
+ ##\r
+ # get the GID of the object\r
+\r
+ def get_gid_caller(self):\r
+ if not self.gidCaller:\r
+ self.decode()\r
+ return self.gidCaller\r
+\r
+ ##\r
+ # set the GID of the object\r
+ #\r
+ # @param gid GID object of the object\r
+\r
+ def set_gid_object(self, gid):\r
+ self.gidObject = gid\r
+\r
+ ##\r
+ # get the GID of the object\r
+\r
+ def get_gid_object(self):\r
+ if not self.gidObject:\r
+ self.decode()\r
+ return self.gidObject\r
+\r
+ ##\r
+ # set the lifetime of this credential\r
+ #\r
+ # @param lifetime lifetime of credential\r
+\r
+ def set_lifetime(self, lifeTime):\r
+ self.lifeTime = lifeTime\r
+\r
+ ##\r
+ # get the lifetime of the credential\r
+\r
+ def get_lifetime(self):\r
+ if not self.lifeTime:\r
+ self.decode()\r
+ return self.lifeTime\r
+\r
+ ##\r
+ # set the delegate bit\r
+ #\r
+ # @param delegate boolean (True or False)\r
+\r
+ def set_delegate(self, delegate):\r
+ self.delegate = delegate\r
+\r
+ ##\r
+ # get the delegate bit\r
+\r
+ def get_delegate(self):\r
+ if not self.delegate:\r
+ self.decode()\r
+ return self.delegate\r
+\r
+ ##\r
+ # set the privileges\r
+ #\r
+ # @param privs either a comma-separated list of privileges of a Rights object\r
+\r
+ def set_privileges(self, privs):\r
+ if isinstance(privs, str):\r
+ self.privileges = Rights(string = privs)\r
+ else:\r
+ self.privileges = privs\r
+\r
+ ##\r
+ # return the privileges as a Rights object\r
+\r
+ def get_privileges(self):\r
+ if not self.privileges:\r
+ self.decode()\r
+ return self.privileges\r
+\r
+ ##\r
+ # determine whether the credential allows a particular operation to be\r
+ # performed\r
+ #\r
+ # @param op_name string specifying name of operation ("lookup", "update", etc)\r
+\r
+ def can_perform(self, op_name):\r
+ rights = self.get_privileges()\r
+ if not rights:\r
+ return False\r
+ return rights.can_perform(op_name)\r
+\r
+ ##\r
+ # Encode the attributes of the credential into a string and store that\r
+ # string in the alt-subject-name field of the X509 object. This should be\r
+ # done immediately before signing the credential.\r
+\r
+ def encode(self):\r
+ dict = {"gidCaller": None,\r
+ "gidObject": None,\r
+ "lifeTime": self.lifeTime,\r
+ "privileges": None,\r
+ "delegate": self.delegate}\r
+ if self.gidCaller:\r
+ dict["gidCaller"] = self.gidCaller.save_to_string(save_parents=True)\r
+ if self.gidObject:\r
+ dict["gidObject"] = self.gidObject.save_to_string(save_parents=True)\r
+ if self.privileges:\r
+ dict["privileges"] = self.privileges.save_to_string()\r
+ str = xmlrpclib.dumps((dict,), allow_none=True)\r
+ self.set_data('URI:http://' + str)\r
+\r
+ ##\r
+ # Retrieve the attributes of the credential from the alt-subject-name field\r
+ # of the X509 certificate. This is automatically done by the various\r
+ # get_* methods of this class and should not need to be called explicitly.\r
+\r
+ def decode(self):\r
+ data = self.get_data().lstrip('URI:http://')\r
+ \r
+ if data:\r
+ dict = xmlrpclib.loads(data)[0][0]\r
+ else:\r
+ dict = {}\r
+\r
+ self.lifeTime = dict.get("lifeTime", None)\r
+ self.delegate = dict.get("delegate", None)\r
+\r
+ privStr = dict.get("privileges", None)\r
+ if privStr:\r
+ self.privileges = Rights(string = privStr)\r
+ else:\r
+ self.privileges = None\r
+\r
+ gidCallerStr = dict.get("gidCaller", None)\r
+ if gidCallerStr:\r
+ self.gidCaller = GID(string=gidCallerStr)\r
+ else:\r
+ self.gidCaller = None\r
+\r
+ gidObjectStr = dict.get("gidObject", None)\r
+ if gidObjectStr:\r
+ self.gidObject = GID(string=gidObjectStr)\r
+ else:\r
+ self.gidObject = None\r
+\r
+ ##\r
+ # Verify that a chain of credentials is valid (see cert.py:verify). In\r
+ # addition to the checks for ordinary certificates, verification also\r
+ # ensures that the delegate bit was set by each parent in the chain. If\r
+ # a delegate bit was not set, then an exception is thrown.\r
+ #\r
+ # Each credential must be a subset of the rights of the parent.\r
+\r
+ def verify_chain(self, trusted_certs = None):\r
+ # do the normal certificate verification stuff\r
+ Certificate.verify_chain(self, trusted_certs)\r
+\r
+ if self.parent:\r
+ # make sure the parent delegated rights to the child\r
+ if not self.parent.get_delegate():\r
+ raise MissingDelegateBit(self.parent.get_subject())\r
+\r
+ # make sure the rights given to the child are a subset of the\r
+ # parents rights\r
+ if not self.parent.get_privileges().is_superset(self.get_privileges()):\r
+ raise ChildRightsNotSubsetOfParent(self.get_subject() \r
+ + " " + self.parent.get_privileges().save_to_string()\r
+ + " " + self.get_privileges().save_to_string())\r
+\r
+ return\r
+\r
+ ##\r
+ # Dump the contents of a credential to stdout in human-readable format\r
+ #\r
+ # @param dump_parents If true, also dump the parent certificates\r
+\r
+ def dump(self, *args, **kwargs):\r
+ print self.dump_string(*args,**kwargs)\r
+\r
+ def dump_string(self, dump_parents=False):\r
+ result=""\r
+ result += "CREDENTIAL %s\n" % self.get_subject()\r
+\r
+ result += " privs: %s\n" % self.get_privileges().save_to_string()\r
+\r
+ gidCaller = self.get_gid_caller()\r
+ if gidCaller:\r
+ result += " gidCaller:\n"\r
+ gidCaller.dump(8, dump_parents)\r
+\r
+ gidObject = self.get_gid_object()\r
+ if gidObject:\r
+ result += " gidObject:\n"\r
+ result += gidObject.dump_string(8, dump_parents)\r
+\r
+ result += " delegate: %s" % self.get_delegate()\r
+\r
+ if self.parent and dump_parents:\r
+ result += "PARENT\n"\r
+ result += self.parent.dump_string(dump_parents)\r
+\r
+ return result\r
git://git.onelab.eu / sfa.git / commitdiff