+#----------------------------------------------------------------------
+# 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
#
import os
import datetime
-from xml.dom.minidom import Document, parseString
from tempfile import mkstemp
-
-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 xml.dom.minidom import Document, parseString
from dateutil.parser import parse
-
+from sfa.util.faults import *
+from sfa.util.sfalogging import sfa_logger
+from sfa.trust.certificate import Keypair
+from sfa.trust.credential_legacy import CredentialLegacy
+from sfa.trust.rights import Right, Rights
+from sfa.trust.gid import GID
# Two years, in seconds
DEFAULT_CREDENTIAL_LIFETIME = 60 * 60 * 24 * 365 * 2
# . 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:
##
# set the privileges
#
- # @param privs either a comma-separated list of privileges of a RightList object
+ # @param privs either a comma-separated list of privileges of a Rights object
def set_privileges(self, privs):
if isinstance(privs, str):
- self.privileges = RightList(string = privs)
+ self.privileges = Rights(string = privs)
else:
self.privileges = privs
##
- # return the privileges as a RightList object
+ # return the privileges as a Rights object
def get_privileges(self):
if not self.privileges:
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"))
# Process privileges
privs = cred.getElementsByTagName("privileges")[0]
- rlist = RightList()
+ rlist = Rights()
for priv in privs.getElementsByTagName("privilege"):
kind = getTextNode(priv, "name")
deleg = str2bool(getTextNode(priv, "can_delegate"))
# 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_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)
return list
##
- # Make sure the credential's target gid was signed by (or is the same) as the entity that signed
- # the original credential.
+ # Make sure the credential's target gid was signed by (or is the same) the entity that signed
+ # the original credential or an authority over that namespace.
def verify_issuer(self):
root_cred = self.get_credential_list()[-1]
root_target_gid = root_cred.get_gid_object()
root_cred_signer = root_cred.get_signature().get_issuer_gid()
-
- if root_target_gid.is_signed_by_cert(root_cred_signer) or \
- root_target_gid.save_to_string() == root_cred_signer.save_to_string():
- pass
- else:
- raise CredentialNotVerifiable("Could not verify credential signer")
-
+
+ if root_target_gid.is_signed_by_cert(root_cred_signer):
+ # cred signer matches target signer, return success
+ return
+
+ root_target_gid_str = root_target_gid.save_to_string()
+ root_cred_signer_str = root_cred_signer.save_to_string()
+ if root_target_gid_str == root_cred_signer_str:
+ # cred signer is target, return success
+ return
+
+ # See if it the signer is an authority over the domain of the target
+ # 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_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()
+ 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 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()))
+
##
# -- For Delegates (credentials with parents) verify that:
# . 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
#
git://git.onelab.eu / sfa.git / blobdiff