import os
import datetime
-from xml.dom.minidom import Document, parseString
from tempfile import mkstemp
+from xml.dom.minidom import Document, parseString
+from dateutil.parser import parse
+import sfa.util.sfalogging
+from sfa.trust.certificate import Keypair
from sfa.trust.credential_legacy import CredentialLegacy
from sfa.trust.rights import *
from sfa.trust.gid import *
from sfa.util.faults import *
-from sfa.util.sfalogging import logger
-from dateutil.parser import parse
# . Need to add support for other types of credentials, e.g. tickets
-
signature_template = \
'''
<Signature xml:id="Sig_%s" xmlns="http://www.w3.org/2000/09/xmldsig#">
return False
-
##
# Utility function to get the text of an XML element
#
class Signature(object):
-
-
+
def __init__(self, string=None):
self.refid = None
self.issuer_gid = None
self.decode()
-
def get_refid(self):
if not self.refid:
self.decode()
# 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):
+ """
+ Returns a list of creds who's gid caller matches the
+ specified caller hrn
+ """
+ if not isinstance(creds, list): creds = [creds]
+ caller_creds = []
+ for cred in creds:
+ try:
+ tmp_cred = Credential(string=cred)
+ if tmp_cred.get_gid_caller().get_hrn() == caller_hrn:
+ caller_creds.append(cred)
+ except: pass
+ return caller_creds
class Credential(object):
-
##
# Create a Credential object
#
# @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
-
+ # FIXME: create and subject are ignored!
def __init__(self, create=False, subject=None, string=None, filename=None):
self.gidCaller = None
self.gidObject = None
self.refid = None
self.legacy = None
-
-
-
# Check if this is a legacy credential, translate it if so
if string or filename:
if string:
self.xmlsec_path = path + '/' + 'xmlsec1'
break
+ def get_subject(self):
+ if not self.gidObject:
+ self.decode()
+ return self.gidObject.get_subject()
def get_signature(self):
if not self.signature:
append_sub(doc, cred, "target_gid", self.gidObject.save_to_string())
append_sub(doc, cred, "target_urn", self.gidObject.get_urn())
append_sub(doc, cred, "uuid", "")
- if not self.expiration:
+ if not self.expiration:
self.set_lifetime(DEFAULT_CREDENTIAL_LIFETIME)
self.expiration = self.expiration.replace(microsecond=0)
append_sub(doc, cred, "expires", self.expiration.isoformat())
self.legacy = None
# Update signatures
- self.decode()
-
-
+ self.decode()
##
cred = doc.getElementsByTagName("credential")[0]
-
self.set_refid(cred.getAttribute("xml:id"))
self.set_lifetime(parse(getTextNode(cred, "expires")))
self.gidCaller = GID(string=getTextNode(cred, "owner_gid"))
# must be done elsewhere
#
# @param trusted_certs: The certificates of trusted CA certificates
-
def verify(self, trusted_certs):
if not self.xml:
self.decode()
- trusted_cert_objects = [GID(filename=f) for f in trusted_certs]
+
+# trusted_cert_objects = [GID(filename=f) for f in trusted_certs]
+ trusted_cert_objects = []
+ ok_trusted_certs = []
+ for f in trusted_certs:
+ try:
+ # Failures here include unreadable files
+ # or non PEM files
+ trusted_cert_objects.append(GID(filename=f))
+ ok_trusted_certs.append(f)
+ except Exception, exc:
+ sfa.util.sfalogging.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:
# make sure it is not expired
if self.get_lifetime() < datetime.datetime.utcnow():
- raise CredentialNotVerifiable("credential is expired")
+ raise CredentialNotVerifiable("Credential expired at %s" % self.expiration.isoformat())
# Verify the signatures
filename = self.save_to_random_tmp_file()
cert_args = " ".join(['--trusted-pem %s' % x for x in trusted_certs])
# Verify the gids of this cred and of its parents
-
-
-
for cur_cred in self.get_credential_list():
cur_cred.get_gid_object().verify_chain(trusted_cert_objects)
- cur_cred.get_gid_caller().verify_chain(trusted_cert_objects)
-
+ cur_cred.get_gid_caller().verify_chain(trusted_cert_objects)
refs = []
refs.append("Sig_%s" % self.get_refid())
verified = os.popen('%s --verify --node-id "%s" %s %s 2>&1' \
% (self.xmlsec_path, ref, cert_args, filename)).read()
if not verified.strip().startswith("OK"):
- raise CredentialNotVerifiable("xmlsec1 error: " + verified)
+ raise CredentialNotVerifiable("xmlsec1 error verifying cert: " + verified)
os.remove(filename)
# Verify the parents (delegation)
# Maybe should be (hrn, type) = urn_to_hrn(root_cred_signer.get_urn())
root_cred_signer_type = root_cred_signer.get_type()
if (root_cred_signer_type == 'authority'):
+ #sfa.util.sfalogging.logger.debug('Cred signer is an authority')
# signer is an authority, see if target is in authority's domain
hrn = root_cred_signer.get_hrn()
- domain = hrn[:hrn.rindex('.')]
- if root_target_gid.get_hrn().startswith(domain):
- # target is in domain of signer's authority
+ if root_target_gid.get_hrn().startswith(hrn):
return
+ # We've required that the credential be signed by an authority
+ # for that domain. Reasonable and probably correct.
+ # A looser model would also allow the signer to be an authority
+ # in my control framework - eg My CA or CH. Even if it is not
+ # the CH that issued these, eg, user credentials.
+
# Give up, credential does not pass issuer verification
- raise CredentialNotVerifiable("Could not verify credential signer")
+
+ raise CredentialNotVerifiable("Could not verify credential owned by %s for object %s. Cred signer %s not the trusted authority for Cred target %s" % (self.gidCaller.get_urn(), self.gidObject.get_urn(), root_cred_signer.get_hrn(), root_target_gid.get_hrn()))
##
# . The privileges must have "can_delegate" set for each delegated privilege
# . The target gid must be the same between child and parents
# . The expiry time on the child must be no later than the parent
- # . The signer of the child must be the owner of the parent
-
+ # . The signer of the child must be the owner of the parent
def verify_parent(self, parent_cred):
# make sure the rights given to the child are a subset of the
# parents rights (and check delegate bits)
# make sure my target gid is the same as the parent's
if not parent_cred.get_gid_object().save_to_string() == \
self.get_gid_object().save_to_string():
- raise CredentialNotVerifiable("target gid not equal between parent and child")
+ raise CredentialNotVerifiable("Target gid not equal between parent and child")
# make sure my expiry time is <= my parent's
if not parent_cred.get_lifetime() >= self.get_lifetime():
- raise CredentialNotVerifiable("delegated credential expires after parent")
+ raise CredentialNotVerifiable("Delegated credential expires after parent")
# make sure my signer is the parent's caller
if not parent_cred.get_gid_caller().save_to_string(False) == \
self.get_signature().get_issuer_gid().save_to_string(False):
- raise CredentialNotVerifiable("delegated credential not signed by parent caller")
+ raise CredentialNotVerifiable("Delegated credential not signed by parent caller")
+ # Recurse
if parent_cred.parent:
parent_cred.verify_parent(parent_cred.parent)
+
+ def delegate(self, delegee_gidfile, caller_keyfile, caller_gidfile):
+ """
+ Return a delegated copy of this credential, delegated to the
+ specified gid's user.
+ """
+ # get the gid of the object we are delegating
+ object_gid = self.get_gid_object()
+ object_hrn = object_gid.get_hrn()
+
+ # the hrn of the user who will be delegated to
+ delegee_gid = GID(filename=delegee_gidfile)
+ delegee_hrn = delegee_gid.get_hrn()
+
+ #user_key = Keypair(filename=keyfile)
+ #user_hrn = self.get_gid_caller().get_hrn()
+ subject_string = "%s delegated to %s" % (object_hrn, delegee_hrn)
+ dcred = Credential(subject=subject_string)
+ dcred.set_gid_caller(delegee_gid)
+ dcred.set_gid_object(object_gid)
+ dcred.set_parent(self)
+ dcred.set_lifetime(self.get_lifetime())
+ dcred.set_privileges(self.get_privileges())
+ dcred.get_privileges().delegate_all_privileges(True)
+ #dcred.set_issuer_keys(keyfile, delegee_gidfile)
+ dcred.set_issuer_keys(caller_keyfile, caller_gidfile)
+ dcred.encode()
+ dcred.sign()
+
+ return dcred
##
# Dump the contents of a credential to stdout in human-readable format
#