+#----------------------------------------------------------------------
+# 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.
+#----------------------------------------------------------------------
+
##
# SFA uses two crypto libraries: pyOpenSSL and M2Crypto to implement
# the necessary crypto functionality. Ideally just one of these libraries
from sfa.util.faults import *
def convert_public_key(key):
- keyconvert_path = "/usr/bin/keyconvert.py"
+ keyconvert_path = "/usr/bin/keyconvert"
if not os.path.isfile(keyconvert_path):
raise IOError, "Could not find keyconvert in %s" % keyconvert_path
# pyOpenSSL only allows us to add extensions, so if we try to set the
# same extension more than once, it will not work
if self.data.has_key(field):
- raise "cannot set ", field, " more than once"
+ raise "Cannot set ", field, " more than once"
self.data[field] = str
self.add_extension(field, 0, str)
# Verify a chain of certificates. Each certificate must be signed by
# the public key contained in it's parent. The chain is recursed
# until a certificate is found that is signed by a trusted root.
-
# TODO: verify expiration time
#print "====Verify Chain====="
# if this cert is signed by a trusted_cert, then we are set
#print "TRUSTED CERT", trusted_cert.dump()
#print "Client is signed by Trusted?", self.is_signed_by_cert(trusted_cert)
if self.is_signed_by_cert(trusted_cert):
+ logger.debug("Cert %s signed by trusted cert %s", self.get_subject(), trusted_cert.get_subject())
return trusted_cert
# if there is no parent, then no way to verify the chain
# . 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.
-
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:
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:
+ 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'):
+ #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)
# @param dump_parents If true, also dump the parent certificates
def dump(self, dump_parents=False):
- print "CREDENTIAL", self.get_subject()
+# FIXME: get_subject doesnt exist
+# print "CREDENTIAL", self.get_subject()
+ print "CREDENTIAL"
print " privs:", self.get_privileges().save_to_string()
### $Id$
### $URL$
-
import xmlrpclib
import uuid
from sfa.trust.certificate import Certificate
Certificate.__init__(self, create, subject, string, filename)
if subject:
- logger.info("subject: %s" % subject)
+ logger.debug("Creating GID for subject: %s" % subject)
if uuid:
self.uuid = int(uuid)
if hrn:
if self.parent:
# make sure the parent's hrn is a prefix of the child's hrn
if not self.get_hrn().startswith(self.parent.get_hrn()):
- raise GidParentHrn(self.parent.get_subject())
+ #print self.get_hrn(), " ", self.parent.get_hrn()
+ raise GidParentHrn("This cert %s HRN doesnt start with parent HRN %s" % (self.get_hrn(), self.parent.get_hrn()))
else:
# make sure that the trusted root's hrn is a prefix of the child's
trusted_gid = GID(string=trusted_root.save_to_string())
trusted_type = trusted_gid.get_type()
trusted_hrn = trusted_gid.get_hrn()
- if trusted_type == 'authority':
- # Could add a check for type == 'authority'
- trusted_hrn = trusted_hrn[:trusted_hrn.rindex('.')]
+ #if trusted_type == 'authority':
+ # trusted_hrn = trusted_hrn[:trusted_hrn.rindex('.')]
cur_hrn = self.get_hrn()
if not self.get_hrn().startswith(trusted_hrn):
- raise GidParentHrn(trusted_hrn + " " + self.get_hrn())
+ raise GidParentHrn("Trusted roots HRN %s isnt start of this cert %s" % (trusted_hrn, cur_hrn))
return
-
-
-
-
-
git://git.onelab.eu / sfa.git / commitdiff