except:
pass
-from sfa.util.faults import *
+from xml.parsers.expat import ExpatError
+
+from sfa.util.faults import CredentialNotVerifiable, ChildRightsNotSubsetOfParent
from sfa.util.sfalogging import logger
from sfa.util.sfatime import utcparse
-from sfa.trust.certificate import Keypair
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
# 2 weeks, in seconds
-DEFAULT_CREDENTIAL_LIFETIME = 86400 * 14
+DEFAULT_CREDENTIAL_LIFETIME = 86400 * 31
# TODO:
# not be changed else the signature is no longer valid. So, once
# you have loaded an existing signed credential, do not call encode() or sign() on it.
-def filter_creds_by_caller(creds, caller_hrn):
+def filter_creds_by_caller(creds, caller_hrn_list):
"""
Returns a list of creds who's gid caller matches the
specified caller hrn
"""
if not isinstance(creds, list): creds = [creds]
+ if not isinstance(caller_hrn_list, list):
+ caller_hrn_list = [caller_hrn_list]
caller_creds = []
for cred in creds:
try:
tmp_cred = Credential(string=cred)
- if tmp_cred.get_gid_caller().get_hrn() == caller_hrn:
+ if tmp_cred.get_gid_caller().get_hrn() in caller_hrn_list:
caller_creds.append(cred)
except: pass
return caller_creds
# @param string If string!=None, load the credential from the string
# @param filename If filename!=None, load the credential from the file
# FIXME: create and subject are ignored!
- def __init__(self, create=False, subject=None, string=None, filename=None):
+ def __init__(self, create=False, subject=None, string=None, filename=None, cred=None):
self.gidCaller = None
self.gidObject = None
self.expiration = None
self.xml = None
self.refid = None
self.legacy = None
+ self.type = None
+ self.version = None
+
+ if cred:
+ if isinstance(cred, StringTypes):
+ string = cred
+ self.type = 'geni_sfa'
+ self.version = '1.0'
+ elif isinstance(cred, dict):
+ string = cred['geni_value']
+ self.type = cred['geni_type']
+ self.version = cred['geni_version']
+
# Check if this is a legacy credential, translate it if so
if string or filename:
break
def get_subject(self):
+ subject = ""
if not self.gidObject:
self.decode()
- return self.gidObject.get_printable_subject()
+ if self.gidObject:
+ subject = self.gidObject.get_printable_subject()
+ return subject
+ # sounds like this should be __repr__ instead ??
def get_summary_tostring(self):
if not self.gidObject:
self.decode()
if not self.gidObject:
self.decode()
return self.gidObject
-
-
##
# Expiration: an absolute UTC time of expiration (as either an int or string or datetime)
if isinstance(privs, str):
self.privileges = Rights(string = privs)
else:
- self.privileges = privs
-
+ self.privileges = privs
##
# return the privileges as a Rights object
# Below throws InUse exception if we forgot to clone the attribute first
oldAttr = signed_cred.setAttributeNode(attr.cloneNode(True))
if oldAttr and oldAttr.value != attr.value:
- msg = "Delegating cred from owner %s to %s over %s replaced attribute %s value '%s' with '%s'" % (self.parent.gidCaller.get_urn(), self.gidCaller.get_urn(), self.gidObject.get_urn(), oldAttr.name, oldAttr.value, attr.value)
+ msg = "Delegating cred from owner %s to %s over %s:\n - Replaced attribute %s value '%s' with '%s'" % (self.parent.gidCaller.get_urn(), self.gidCaller.get_urn(), self.gidObject.get_urn(), oldAttr.name, oldAttr.value, attr.value)
logger.warn(msg)
#raise CredentialNotVerifiable("Can't encode new valid delegated credential: %s" % msg)
def decode(self):
if not self.xml:
return
+
+ doc = None
+ try:
+ doc = parseString(self.xml)
+ except ExpatError,e:
+ raise CredentialNotVerifiable("Malformed credential")
doc = parseString(self.xml)
sigs = []
signed_cred = doc.getElementsByTagName("signed-credential")
trusted_cert_objects.append(GID(filename=f))
ok_trusted_certs.append(f)
except Exception, exc:
- logger.error("Failed to load trusted cert from %s: %r", f, exc)
+ 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
# But we haven't verified that it is _signed by_ an authority
# We also don't know if xmlsec1 requires that cert signers
# are marked as CAs.
- root_cred_signer.verify_chain(trusted_gids)
+
+ # Note that if verify() gave us no trusted_gids then this
+ # call will fail. So skip it if we have no trusted_gids
+ if trusted_gids and len(trusted_gids) > 0:
+ root_cred_signer.verify_chain(trusted_gids)
+ else:
+ logger.debug("No trusted gids. Cannot verify that cred signer is signed by a trusted authority. Skipping that check.")
# See if the signer is an authority over the domain of the target.
# There are multiple types of authority - accept them all here
def get_filename(self):
return getattr(self,'filename',None)
+ # a helper function used by some methods to find out who really is the caller
+ # using a heuristic to identify a delegated credential
+ # this admittedly is a bit of a hack, please USE IN LAST RESORT
+ #
+ def actual_caller_hrn (self):
+ caller_hrn = self.get_gid_caller().get_hrn()
+ issuer_hrn = self.get_signature().get_issuer_gid().get_hrn()
+ subject_hrn = self.get_gid_object().get_hrn()
+ # if we find that the caller_hrn is an immediate descendant of the issuer, then
+ # this seems to be a 'regular' credential
+ if caller_hrn.startswith(issuer_hrn):
+ actual_caller_hrn=caller_hrn
+ # else this looks like a delegated credential, and the real caller is the issuer
+ else:
+ actual_caller_hrn=issuer_hrn
+ logger.info("actual_caller_hrn: caller_hrn=%s, issuer_hrn=%s, returning %s"%(caller_hrn,issuer_hrn,actual_caller_hrn))
+ return actual_caller_hrn
+
##
# Dump the contents of a credential to stdout in human-readable format
#
def dump (self, *args, **kwargs):
print self.dump_string(*args, **kwargs)
-
- def dump_string(self, dump_parents=False):
+ # show_xml is ignored
+ def dump_string(self, dump_parents=False, show_xml=None):
result=""
result += "CREDENTIAL %s\n" % self.get_subject()
filename=self.get_filename()
if filename: result += "Filename %s\n"%filename
- result += " privs: %s\n" % self.get_privileges().save_to_string()
+ privileges = self.get_privileges()
+ if privileges:
+ result += " privs: %s\n" % privileges.save_to_string()
+ else:
+ result += " privs: \n"
gidCaller = self.get_gid_caller()
if gidCaller:
result += " gidCaller:\n"
print " gidIssuer:"
self.get_signature().get_issuer_gid().dump(8, dump_parents)
+ if self.expiration:
+ print " expiration:", self.expiration.isoformat()
+
gidObject = self.get_gid_object()
if gidObject:
result += " gidObject:\n"
git://git.onelab.eu / sfa.git / blobdiff