# are being used due to lack of functionality in pyOpenSSL and some apparant
# bugs in M2Crypto
+import os
+import tempfile
from OpenSSL import crypto
import M2Crypto
from M2Crypto import X509
self.key = crypto.load_privatekey(crypto.FILETYPE_PEM, string)
self.m2key = M2Crypto.EVP.load_key_string(string)
+ def load_pubkey_from_file(self, filename):
+ # load the m2 public key
+ m2rsakey = M2Crypto.RSA.load_pub_key(filename)
+ self.m2key = M2Crypto.EVP.PKey()
+ self.m2key.assign_rsa(m2rsakey)
+
+ # create an m2 x509 cert
+ m2name = M2Crypto.X509.X509_Name()
+ m2name.add_entry_by_txt(field="CN", type=0x1001, entry="junk", len=-1, loc=-1, set=0)
+ m2x509 = M2Crypto.X509.X509()
+ m2x509.set_pubkey(self.m2key)
+ m2x509.set_serial_number(0)
+ m2x509.set_issuer_name(m2name)
+ m2x509.set_subject_name(m2name)
+ ASN1 = M2Crypto.ASN1.ASN1_UTCTIME()
+ ASN1.set_time(500)
+ m2x509.set_not_before(ASN1)
+ m2x509.set_not_after(ASN1)
+ junk_key = Keypair(create=True)
+ m2x509.sign(pkey=junk_key.get_m2_pkey(), md="sha1")
+
+ # convert the m2 x509 cert to a pyopenssl x509
+ m2pem = m2x509.as_pem()
+ pyx509 = crypto.load_certificate(crypto.FILETYPE_PEM, m2pem)
+
+ # get the pyopenssl pkey from the pyopenssl x509
+ self.key = pyx509.get_pubkey()
+
+ def load_pubkey_from_string(self, string):
+ (f, fn) = tempfile.mkstemp()
+ os.write(f, string)
+ os.close(f)
+ self.load_pubkey_from_file(fn)
+ os.remove(fn)
+
def as_pem(self):
return crypto.dump_privatekey(crypto.FILETYPE_PEM, self.key)
# TODO: verify expiration of trusted_cert ?
if self.is_signed_by_cert(trusted_cert):
#print self.get_subject(), "is signed by a root"
- return True
+ return
# if there is no parent, then no way to verify the chain
if not self.parent:
#print self.get_subject(), "has no parent"
- return False
+ raise MissingParent(self.get_subject())
# if it wasn't signed by the parent...
if not self.is_signed_by_cert(self.parent):
#print self.get_subject(), "is not signed by parent"
- return False
+ return NotSignedByParent(self.get_subject())
# if the parent isn't verified...
- if not self.parent.verify_chain(trusted_certs):
- #print self.get_subject(), "parent does not verify"
- return False
+ self.parent.verify_chain(trusted_certs)
- return True
+ return
# to perform this encoding.
class Credential(Certificate):
- uuid = None
- hrn = None
-
gidCaller = None
gidObject = None
lifeTime = None
def verify_chain(self, trusted_certs = None):
# do the normal certificate verification stuff
- if not Certificate.verify_chain(self, trusted_certs):
- return False
+ Certificate.verify_chain(self, trusted_certs)
if parent:
# make sure the parent delegated rights to the child
if not parent.delegate:
- return False
+ raise MissingDelegateBit(self.get_subject())
# XXX todo: make sure child rights are a subset of parent rights
- return True
+ return
+
+ def dump(self):
+ print "CREDENTIAL", self.get_subject()
+
+ print " privs:", self.get_privileges().save_to_string()
+
+ print " gidCaller:"
+ gidCaller = self.get_gid_caller()
+ if gidCaller:
+ gidCaller.dump(indent=8)
+
+ print " gidObject:"
+ gidObject = self.get_gid_object()
+ if gidObject:
+ gidObject.dump(indent=8)
+
+ print " delegate:", self.get_delegate()
+
self.value = value
def __str__(self):
return repr(self.value)
+
+class PlanetLabRecordDoesNotExist(Exception):
+ def __init__(self, value):
+ self.value = value
+ def __str__(self):
+ return repr(self.value)
+
+class PermissionError(Exception):
+ def __init__(self, value):
+ self.value = value
+ def __str__(self):
+ return repr(self.value)
+
+class InsufficientRights(Exception):
+ def __init__(self, value):
+ self.value = value
+ def __str__(self):
+ return repr(self.value)
+
+class MissingDelegateBit(Exception):
+ def __init__(self, value):
+ self.value = value
+ def __str__(self):
+ return repr(self.value)
+
+class MissingParent(Exception):
+ def __init__(self, value):
+ self.value = value
+ def __str__(self):
+ return repr(self.value)
+
+class NotSignedByParent(Exception):
+ def __init__(self, value):
+ self.value = value
+ def __str__(self):
+ return repr(self.value)
self.uuid = dict.get("uuid", None)
self.hrn = dict.get("hrn", None)
- # XXX: I may have decided not to use this
- def get_cache_key(self):
- return self.get_cache_uuid() + self.get_cache_name()
+ def dump(self, indent=0):
+ # TODO: implement indent
+ print " "*indent, " hrn:", self.get_hrn()
+ print " "*indent, "uuid:", self.get_uuid()
- # XXX: I may have decided not to use this
- def get_cache_uuid(self):
- return self.get_uuid()
- # XXX: I may have decided not to use this
- def get_cache_name(self):
- return self.get_name()
# *.DBINFO - database info
import os
+import report
from cert import *
from gid import *
from misc import *
os.path.exists(privkey_filename) and \
os.path.exists(dbinfo_filename)
- def create_auth(self, hrn):
+ def create_auth(self, hrn, create_parents=False):
+ report.trace("Hierarchy: creating authority: " + hrn)
+
+ # create the parent authority if necessary
+ parent_hrn = get_authority(hrn)
+ if (parent_hrn) and (not self.auth_exists(parent_hrn)) and (create_parents):
+ self.create_auth(parent_hrn, create_parents)
+
(directory, gid_filename, privkey_filename, dbinfo_filename) = \
self.get_auth_filenames(hrn)
dbinfo_file.write(str(dbinfo))\r
dbinfo_file.close()
- def get_auth_info(self, hrn, can_create=True):
- if not self.auth_exists(hrn):
- if not can_create:
- return MissingAuthority(hrn)
+ def get_auth_info(self, hrn):
+ #report.trace("Hierarchy: getting authority: " + hrn)
- self.create_auth(hrn)
+ if not self.auth_exists(hrn):
+ raise MissingAuthority(hrn)
(directory, gid_filename, privkey_filename, dbinfo_filename) = \
self.get_auth_filenames(hrn)
def create_gid(self, hrn, uuid, pkey):
parent_hrn = get_authority(hrn)
- gid = GID(subject=hrn, uuid=uuid)
+ gid = GID(subject=hrn, uuid=uuid, hrn=hrn)
if not parent_hrn:
# if there is no parent hrn, then it must be self-signed. this
gid.set_parent(parent_auth_info.get_gid_object())
gid.set_pubkey(pkey)
+ gid.encode()
gid.sign()
return gid
git://git.onelab.eu / sfa.git / commitdiff