openstack/deb-python-pysaml2
Code Issues Proposed changes

Merge remote-tracking branch 'upstream/master'

Browse Source
This commit is contained in:
Hans Hörberg
2014-06-05 12:58:26 +02:00
parent 8286ebbbc4 d837b7b298
commit 4d4cfd44b5
2 changed files with 167 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
src/saml2/sigver.py
View File

@@ -38,7 +38,7 @@ from saml2.saml import EncryptedAssertion
import xmldsig as ds
from saml2 import samlp
from saml2 import samlp, SamlBase
from saml2 import SAMLError
from saml2 import extension_elements_to_elements
from saml2 import class_name
@@ -71,6 +71,10 @@ logger = logging.getLogger(__name__)
SIG = "{%s#}%s" % (ds.NAMESPACE, "Signature")
RSA_SHA1 = "http://www.w3.org/2000/09/xmldsig#rsa-sha1"
RSA_SHA256 = "http://www.w3.org/2001/04/xmldsig-more#rsa-sha256"
RSA_SHA384 = "http://www.w3.org/2001/04/xmldsig-more#rsa-sha384"
RSA_SHA512 = "http://www.w3.org/2001/04/xmldsig-more#rsa-sha512"
RSA_1_5 = "http://www.w3.org/2001/04/xmlenc#rsa-1_5"
TRIPLE_DES_CBC = "http://www.w3.org/2001/04/xmlenc#tripledes-cbc"
XMLTAG = "<?xml version='1.0'?>"
@@ -313,7 +317,7 @@ def signed_instance_factory(instance, seccont, elements_to_sign=None):
signed_xml = "%s" % instance
for (node_name, nodeid) in elements_to_sign:
signed_xml = seccont.sign_statement(
signed_xml, class_name=node_name, node_id=nodeid)
signed_xml, node_name=node_name, node_id=nodeid)
#print "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
#print "%s" % signed_xml
@@ -600,9 +604,9 @@ class RSASigner(Signer):
SIGNER_ALGS = {
RSA_SHA1: RSASigner(SHA),
"http://www.w3.org/2001/04/xmldsig-more#rsa-sha256": RSASigner(SHA256),
"http://www.w3.org/2001/04/xmldsig-more#rsa-sha384": RSASigner(SHA384),
"http://www.w3.org/2001/04/xmldsig-more#rsa-sha512": RSASigner(SHA512),
RSA_SHA256: RSASigner(SHA256),
RSA_SHA384: RSASigner(SHA384),
RSA_SHA512: RSASigner(SHA512),
}
REQ_ORDER = ["SAMLRequest", "RelayState", "SigAlg"]
@@ -698,13 +702,14 @@ class CryptoBackend():
def encrypt(self, text, recv_key, template, key_type):
raise NotImplementedError()
def encrypt_assertion(self, statement, recv_key, key_type, xpath=""):
def encrypt_assertion(self, statement, enc_key, template, key_type,
node_xpath):
raise NotImplementedError()
def decrypt(self, enctext, key_file):
raise NotImplementedError()
def sign_statement(self, statement, class_name, key_file, node_id,
def sign_statement(self, statement, node_name, key_file, node_id,
id_attr):
raise NotImplementedError()
@@ -764,26 +769,29 @@ class CryptoBackendXmlSec1(CryptoBackend):
return output
def encrypt_assertion(self, statement, enc_key, template,
key_type="des-192"):
key_type="des-192", node_xpath=None):
"""
--pubkey-cert-pem ../../example/idp2/pki/mycert.pem \
--session-key des-192 --xml-data pre_saml2_assertion.xml \
--node-xpath '/*[local-name()="Response"]/*[local-name(
)="EncryptedAssertion"]/*[local-name()="Assertion"]' \
enc-element-3des-kt-rsa1_5.tmpl > enc_3des_rsa_assertion.xml
Will encrypt an assertion
:param statement:
:param cert_file:
:param cert_type:
:return:
:param statement: A XML document that contains the assertion to encrypt
:param enc_key: File name of a file containing the encryption key
:param template: A template for the encryption part to be added.
:param key_type: The type of session key to use.
:return: The encrypted text
"""
statement = pre_encrypt_assertion(statement)
if isinstance(statement, SamlBase):
statement = pre_encrypt_assertion(statement)
_, fil = make_temp("%s" % statement, decode=False, delete=False)
_, tmpl = make_temp("%s" % template, decode=False)
if not node_xpath:
node_xpath = ASSERT_XPATH
com_list = [self.xmlsec, "encrypt", "--pubkey-cert-pem", enc_key,
"--session-key", key_type, "--xml-data", fil,
"--node-xpath", ASSERT_XPATH]
"--node-xpath", node_xpath]
(_stdout, _stderr, output) = self._run_xmlsec(
com_list, [tmpl], exception=EncryptError, validate_output=False)
@@ -795,6 +803,13 @@ class CryptoBackendXmlSec1(CryptoBackend):
return output
def decrypt(self, enctext, key_file):
"""
:param enctext: XML document containing an encrypted part
:param key_file: The key to use for the decryption
:return: The decrypted document
"""
logger.debug("Decrypt input len: %d" % len(enctext))
_, fil = make_temp("%s" % enctext, decode=False)
@@ -806,13 +821,13 @@ class CryptoBackendXmlSec1(CryptoBackend):
validate_output=False)
return output
def sign_statement(self, statement, class_name, key_file, node_id,
def sign_statement(self, statement, node_name, key_file, node_id,
id_attr):
"""
Sign an XML statement.
:param statement: The statement to be signed
:param class_name: string like 'urn:oasis:names:...:Assertion'
:param node_name: string like 'urn:oasis:names:...:Assertion'
:param key_file: The file where the key can be found
:param node_id:
:param id_attr: The attribute name for the identifier, normally one of
@@ -824,7 +839,7 @@ class CryptoBackendXmlSec1(CryptoBackend):
com_list = [self.xmlsec, "--sign",
"--privkey-pem", key_file,
"--id-attr:%s" % id_attr, class_name]
"--id-attr:%s" % id_attr, node_name]
if node_id:
com_list.extend(["--node-id", node_id])
@@ -937,7 +952,7 @@ class CryptoBackendXMLSecurity(CryptoBackend):
# better than static 0.0 here.
return "XMLSecurity 0.0"
def sign_statement(self, statement, _class_name, key_file, node_id,
def sign_statement(self, statement, node_name, key_file, node_id,
_id_attr):
"""
Sign an XML statement.
@@ -946,6 +961,7 @@ class CryptoBackendXMLSecurity(CryptoBackend):
implementation are :
:param statement: XML as string
:param node_name: Name of the node to sign
:param key_file: xmlsec key_spec string(), filename,
"pkcs11://" URI or PEM data
:returns: Signed XML as string
@@ -1242,18 +1258,18 @@ class SecurityContext(object):
return self.crypto.encrypt(text, recv_key, template, key_type)
def encrypt_assertion(self, statement, cert_file, cert_type="pem"):
def encrypt_assertion(self, statement, enc_key, template,
key_type="des-192", node_xpath=None):
"""
--pubkey-cert-pem ../../example/idp2/pki/mycert.pem \
--session-key des-192 --xml-data pre_saml2_assertion.xml \
--node-xpath '/*[local-name()="Response"]/*[local-name(
)="EncryptedAssertion"]/*[local-name()="Assertion"]' \
enc-element-3des-kt-rsa1_5.tmpl > enc_3des_rsa_assertion.xml
:param statement:
:param cert_file:
:param cert_type:
:return:
Will encrypt an assertion
:param statement: A XML document that contains the assertion to encrypt
:param enc_key: File name of a file containing the encryption key
:param template: A template for the encryption part to be added.
:param key_type: The type of session key to use.
:return: The encrypted text
"""
raise NotImplemented()
def decrypt(self, enctext, key_file=None):
""" Decrypting an encrypted text by the use of a private key.
@@ -1377,7 +1393,7 @@ class SecurityContext(object):
"""
:param item: Parsed entity
:param node_name: The name of the class that is signed
:param node_name: The name of the node/class/element that is signed
:param origdoc: The original XML string
:param id_attr:
:param must:
@@ -1577,12 +1593,12 @@ class SecurityContext(object):
""" Deprecated function. See sign_statement(). """
return self.sign_statement(statement, **kwargs)
def sign_statement(self, statement, class_name, key=None,
def sign_statement(self, statement, node_name, key=None,
key_file=None, node_id=None, id_attr=""):
"""Sign a SAML statement.
:param statement: The statement to be signed
:param class_name: string like 'urn:oasis:names:...:Assertion'
:param node_name: string like 'urn:oasis:names:...:Assertion'
:param key: The key to be used for the signing, either this or
:param key_file: The file where the key can be found
:param node_id:
@@ -1599,7 +1615,7 @@ class SecurityContext(object):
if not key and not key_file:
key_file = self.key_file
return self.crypto.sign_statement(statement, class_name, key_file,
return self.crypto.sign_statement(statement, node_name, key_file,
node_id, id_attr)
def sign_assertion_using_xmlsec(self, statement, **kwargs):
@@ -1797,3 +1813,16 @@ def response_factory(sign=False, encrypt=False, **kwargs):
setattr(response, key, val)
return response
# ----------------------------------------------------------------------------
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('-s', '--list-sigalgs', dest='listsigalgs',
action='store_true',
help='List implemented signature algorithms')
args = parser.parse_args()
if args.listsigalgs:
print '\n'.join([key for key, value in SIGNER_ALGS.items()])

101
tools/merge_metadata.py Executable file
View File

@@ -0,0 +1,101 @@
#!/usr/bin/env python
from saml2.md import EntitiesDescriptor
from saml2.sigver import _get_xmlsec_cryptobackend, SecurityContext
from saml2.httpbase import HTTPBase
from saml2 import saml
from saml2 import md
from saml2.attribute_converter import ac_factory
from saml2.extension import dri
from saml2.extension import idpdisc
from saml2.extension import mdattr
from saml2.extension import mdrpi
from saml2.extension import mdui
from saml2.extension import shibmd
from saml2.extension import ui
import xmldsig
import xmlenc
import argparse
from saml2.mdstore import MetaDataFile, MetaDataExtern
__author__ = 'rolandh'
"""
A script that imports and verifies metadata.
"""
ONTS = {
saml.NAMESPACE: saml,
mdui.NAMESPACE: mdui,
mdattr.NAMESPACE: mdattr,
mdrpi.NAMESPACE: mdrpi,
dri.NAMESPACE: dri,
ui.NAMESPACE: ui,
idpdisc.NAMESPACE: idpdisc,
md.NAMESPACE: md,
xmldsig.NAMESPACE: xmldsig,
xmlenc.NAMESPACE: xmlenc,
shibmd.NAMESPACE: shibmd
}
parser = argparse.ArgumentParser()
parser.add_argument('-a', dest='attrsmap')
parser.add_argument('-o', dest='output', default="local")
parser.add_argument('-x', dest='xmlsec')
parser.add_argument('-i', dest='ignore_valid', action='store_true')
parser.add_argument(dest="conf")
args = parser.parse_args()
metad = None
output = EntitiesDescriptor()
# config file format
#
# local <local file name>
# remote <url> <local file name for certificate use to verify signature>
#
# for instance
#
#local metadata_sp_1.xml
#local InCommon-metadata.xml
#remote https://kalmar2.org/simplesaml/module.php/aggregator/?id=kalmarcentral2&set=saml2 kalmar2.pem
#
for line in open(args.conf).readlines():
line = line.strip()
if line[0] == "#":
continue
spec = line.split(" ")
if args.ignore_valid:
kwargs = {"check_validity": False}
else:
kwargs = {}
if spec[0] == "local":
metad = MetaDataFile(ONTS.values(), spec[1], spec[1], **kwargs)
elif spec[0] == "remote":
ATTRCONV = ac_factory(args.attrsmap)
httpc = HTTPBase()
crypto = _get_xmlsec_cryptobackend(args.xmlsec)
sc = SecurityContext(crypto, key_type="", cert_type="")
metad = MetaDataExtern(ONTS.values(), ATTRCONV, spec[1],
sc, cert=spec[2], http=httpc, **kwargs)
if metad:
try:
metad.load()
except:
raise
output.entity_descriptor.extend(metad.entities_descr.entity_descriptor)
print output