from sfa.util.faults import CredentialNotVerifiable, ChildRightsNotSubsetOfParent
from sfa.util.sfalogging import logger
from sfa.util.sfatime import utcparse
-from sfa.trust.credential_legacy import CredentialLegacy
from sfa.trust.rights import Right, Rights, determine_rights
from sfa.trust.gid import GID
from sfa.util.xrn import urn_to_hrn, hrn_authfor_hrn
# A credential provides a caller gid with privileges to an object gid.
# A signed credential is signed by the object's authority.
#
-# Credentials are encoded in one of two ways. The legacy style places
-# it in the subjectAltName of an X509 certificate. The new credentials
-# are placed in signed XML.
+# Credentials are encoded in one of two ways.
+# The legacy style (now unsupported) places it in the subjectAltName of an X509 certificate.
+# The new credentials are placed in signed XML.
#
# WARNING:
# In general, a signed credential obtained externally should
self.signature = None
self.xml = None
self.refid = None
- self.legacy = None
self.type = None
self.version = None
self.version = cred['geni_version']
- # Check if this is a legacy credential, translate it if so
if string or filename:
if string:
str = string
elif filename:
str = file(filename).read()
+ # if this is a legacy credential, write error and bail out
if str.strip().startswith("-----"):
- self.legacy = CredentialLegacy(False,string=str)
- self.translate_legacy(str)
+ logger.error("Legacy credentials not supported any more - giving up with %s..."%str[:10])
+ break
else:
self.xml = str
self.decode()
self.signature = sig
- ##
- # Translate a legacy credential into a new one
- #
- # @param String of the legacy credential
-
- def translate_legacy(self, str):
- legacy = CredentialLegacy(False,string=str)
- self.gidCaller = legacy.get_gid_caller()
- self.gidObject = legacy.get_gid_object()
- lifetime = legacy.get_lifetime()
- if not lifetime:
- self.set_expiration(datetime.datetime.utcnow() + datetime.timedelta(seconds=DEFAULT_CREDENTIAL_LIFETIME))
- else:
- self.set_expiration(int(lifetime))
- self.lifeTime = legacy.get_lifetime()
- self.set_privileges(legacy.get_privileges())
- self.get_privileges().delegate_all_privileges(legacy.get_delegate())
-
##
# Need the issuer's private key and name
# @param key Keypair object containing the private key of the issuer
# at this point self.expiration is normalized as a datetime - DON'T call utcparse again
return self.expiration
- ##
- # For legacy sake
- def get_lifetime(self):
- return self.get_expiration()
-
##
# set the privileges
#
self.xml = signed
- # This is no longer a legacy credential
- if self.legacy:
- self.legacy = None
-
# Update signatures
self.decode()
self.decode()
# validate against RelaxNG schema
- if HAVELXML and not self.legacy:
+ if HAVELXML:
if schema and os.path.exists(schema):
tree = etree.parse(StringIO(self.xml))
schema_doc = etree.parse(schema)
logger.error("Failed to load trusted cert from %s: %r"%( f, exc))
trusted_certs = ok_trusted_certs
- # Use legacy verification if this is a legacy credential
- if self.legacy:
- self.legacy.verify_chain(trusted_cert_objects)
- if self.legacy.client_gid:
- self.legacy.client_gid.verify_chain(trusted_cert_objects)
- if self.legacy.object_gid:
- self.legacy.object_gid.verify_chain(trusted_cert_objects)
- return True
-
# make sure it is not expired
if self.get_expiration() < datetime.datetime.utcnow():
raise CredentialNotVerifiable("Credential %s expired at %s" % (self.get_summary_tostring(), self.expiration.isoformat()))
+++ /dev/null
-#----------------------------------------------------------------------
-# Copyright (c) 2008 Board of Trustees, Princeton University
-#
-# Permission is hereby granted, free of charge, to any person obtaining
-# a copy of this software and/or hardware specification (the "Work") to
-# deal in the Work without restriction, including without limitation the
-# rights to use, copy, modify, merge, publish, distribute, sublicense,
-# and/or sell copies of the Work, and to permit persons to whom the Work
-# is furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be
-# included in all copies or substantial portions of the Work.
-#
-# THE WORK IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
-# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
-# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
-# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
-# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
-# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE WORK OR THE USE OR OTHER DEALINGS
-# IN THE WORK.
-#----------------------------------------------------------------------
-##
-# 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
git://git.onelab.eu / sfa.git / commitdiff