1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
|
import os.path
from base64 import b64encode
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15
from lxml import etree
from lxml.etree import Element, SubElement
from signxml import (
InvalidInput,
XMLSignatureProcessor,
_remove_sig,
ds_tag,
dsig11_tag,
ensure_str,
iterate_pem,
long_to_bytes,
namespaces,
strip_pem_header,
)
class XMLEnvelopedEnvelopingSigner(XMLSignatureProcessor):
"""
A class that signs multiple data by putting the signature enveloped in the
first data item, and enveloping the others.
"""
def __init__(
self,
signature_algorithm="rsa-sha256",
digest_algorithm="sha256",
c14n_algorithm=XMLSignatureProcessor.default_c14n_algorithm,
):
self.sign_alg = signature_algorithm
assert self.sign_alg in self.known_signature_digest_tags or self.sign_alg in self.known_hmac_digest_tags
assert digest_algorithm in self.known_digest_tags
self.digest_alg = digest_algorithm
assert c14n_algorithm in self.known_c14n_algorithms
self.c14n_alg = c14n_algorithm
self.namespaces = dict(ds=namespaces.ds)
self._parser = None
def sign(self, data, key=None, passphrase=None, cert=None, key_name=None, key_info=None, id_attribute=None):
"""
Sign the data and return the root element of the resulting XML tree.
:param data: Data to sign - must be a sequence of items to sign.
:type data_others: Sequence of String, file-like object, or XML ElementTree Element API compatible object
:param key:
Key to be used for signing. When signing with a certificate or RSA/DSA/ECDSA key, this can be a string
containing a PEM-formatted key, or a :py:class:`cryptography.hazmat.primitives.interfaces.RSAPublicKey`,
:py:class:`cryptography.hazmat.primitives.interfaces.DSAPublicKey`, or
:py:class:`cryptography.hazmat.primitives.interfaces.EllipticCurvePublicKey` object. When signing with a
HMAC, this should be a string containing the shared secret.
:type key:
string, :py:class:`cryptography.hazmat.primitives.interfaces.RSAPublicKey`,
:py:class:`cryptography.hazmat.primitives.interfaces.DSAPublicKey`, or
:py:class:`cryptography.hazmat.primitives.interfaces.EllipticCurvePublicKey` object
:param passphrase: Passphrase to use to decrypt the key, if any.
:type passphrase: string
:param cert:
X.509 certificate to use for signing. This should be a string containing a PEM-formatted certificate, or an
array of strings or OpenSSL.crypto.X509 objects containing the certificate and a chain of intermediate
certificates.
:type cert: string, array of strings, or array of OpenSSL.crypto.X509 objects
:param key_name: Add a KeyName element in the KeyInfo element that may be used by the signer to communicate a
key identifier to the recipient. Typically, KeyName contains an identifier related to the key pair used to
sign the message.
:type key_name: string
:param key_info: A custom KeyInfo element to insert in the signature. Use this to supply
``<wsse:SecurityTokenReference>`` or other custom key references.
:type key_info: :py:class:`lxml.etree.Element`
:param id_attribute:
Name of the attribute whose value ``URI`` refers to. By default, SignXML will search for "Id", then "ID".
:type id_attribute: string
:returns:
A :py:class:`lxml.etree.Element` object representing the root of the XML tree containing the signature and
the payload data.
To specify the location of an enveloped signature within **data**, insert a
``<ds:Signature Id="placeholder"></ds:Signature>`` element in **data** (where
"ds" is the "http://www.w3.org/2000/09/xmldsig#" namespace). This element will
be replaced by the generated signature, and excised when generating the digest.
"""
if id_attribute is not None:
self.id_attributes = (id_attribute,)
if isinstance(cert, (str, bytes)):
cert_chain = list(iterate_pem(cert))
else:
cert_chain = cert
sig_root, doc_root, c14n_inputs, reference_uris = self._unpack(data)
signed_info_element, signature_value_element = self._build_sig(sig_root, reference_uris, c14n_inputs)
if key is None:
raise InvalidInput('Parameter "key" is required')
signed_info_c14n = self._c14n(signed_info_element, algorithm=self.c14n_alg)
if self.sign_alg.startswith("hmac-"):
from cryptography.hazmat.primitives.hmac import HMAC
signer = HMAC(
key=key, algorithm=self._get_hmac_digest_method_by_tag(self.sign_alg), backend=default_backend()
)
signer.update(signed_info_c14n)
signature_value_element.text = ensure_str(b64encode(signer.finalize()))
sig_root.append(signature_value_element)
elif any(self.sign_alg.startswith(i) for i in ["dsa-", "rsa-", "ecdsa-"]):
if isinstance(key, (str, bytes)):
from cryptography.hazmat.primitives.serialization import load_pem_private_key
key = load_pem_private_key(key, password=passphrase, backend=default_backend())
hash_alg = self._get_signature_digest_method_by_tag(self.sign_alg)
if self.sign_alg.startswith("dsa-"):
signature = key.sign(signed_info_c14n, algorithm=hash_alg)
elif self.sign_alg.startswith("ecdsa-"):
signature = key.sign(signed_info_c14n, signature_algorithm=ec.ECDSA(algorithm=hash_alg))
elif self.sign_alg.startswith("rsa-"):
signature = key.sign(signed_info_c14n, padding=PKCS1v15(), algorithm=hash_alg)
else:
raise NotImplementedError()
if self.sign_alg.startswith("dsa-"):
# Note: The output of the DSA signer is a DER-encoded ASN.1 sequence of two DER integers.
from asn1crypto.algos import DSASignature
decoded_signature = DSASignature.load(signature).native
r = decoded_signature["r"]
s = decoded_signature["s"]
signature = long_to_bytes(r).rjust(32, b"\0") + long_to_bytes(s).rjust(32, b"\0")
signature_value_element.text = ensure_str(b64encode(signature))
if key_info is None:
key_info = SubElement(sig_root, ds_tag("KeyInfo"))
if key_name is not None:
keyname = SubElement(key_info, ds_tag("KeyName"))
keyname.text = key_name
if cert_chain is None:
self._serialize_key_value(key, key_info)
else:
x509_data = SubElement(key_info, ds_tag("X509Data"))
for cert in cert_chain:
x509_certificate = SubElement(x509_data, ds_tag("X509Certificate"))
if isinstance(cert, (str, bytes)):
x509_certificate.text = strip_pem_header(cert)
else:
from OpenSSL.crypto import FILETYPE_PEM, dump_certificate
x509_certificate.text = strip_pem_header(dump_certificate(FILETYPE_PEM, cert))
else:
sig_root.append(key_info)
else:
raise NotImplementedError()
for c14n_input in c14n_inputs[1:]:
sig_root.append(c14n_input)
return doc_root
def _unpack(self, data):
sig_root = Element(ds_tag("Signature"), nsmap=self.namespaces)
if isinstance(data[0], (str, bytes)):
raise InvalidInput("First data item **must** be an XML element")
doc_root = self.get_root(data[0])
c14n_inputs = [self.get_root(data[0])]
signature_placeholders = self._findall(doc_root, "Signature[@Id='placeholder']", anywhere=True)
if len(signature_placeholders) == 0:
doc_root.append(sig_root)
elif len(signature_placeholders) == 1:
sig_root = signature_placeholders[0]
del sig_root.attrib["Id"]
for c14n_input in c14n_inputs:
placeholders = self._findall(c14n_input, "Signature[@Id='placeholder']", anywhere=True)
if placeholders:
assert len(placeholders) == 1
_remove_sig(placeholders[0])
else:
raise InvalidInput("Enveloped signature input contains more than one placeholder")
reference_uris = []
for c14n_input in c14n_inputs:
payload_id = None
for id_attribute in self.id_attributes:
payload_id = c14n_input.get(id_attribute)
if payload_id:
break
reference_uris.append("#{}".format(payload_id) if payload_id else "")
index = 1
for enveloped_data in data[1:]:
c14n_inputs.append(Element(ds_tag("Object"), nsmap=self.namespaces, Id="object-{}".format(index)))
if isinstance(enveloped_data, (str, bytes)):
c14n_inputs[index].text = enveloped_data
else:
c14n_inputs[index].append(self.get_root(enveloped_data))
reference_uris.append("#object-{}".format(index))
index += 1
return sig_root, doc_root, c14n_inputs, reference_uris
def _build_sig(self, sig_root, reference_uris, c14n_inputs):
signed_info = SubElement(sig_root, ds_tag("SignedInfo"), nsmap=self.namespaces)
c14n_method = SubElement(signed_info, ds_tag("CanonicalizationMethod"), Algorithm=self.c14n_alg) # noqa:F841
if self.sign_alg.startswith("hmac-"):
algorithm_id = self.known_hmac_digest_tags[self.sign_alg]
else:
algorithm_id = self.known_signature_digest_tags[self.sign_alg]
signature_method = SubElement(signed_info, ds_tag("SignatureMethod"), Algorithm=algorithm_id) # noqa:F841
for i, reference_uri in enumerate(reference_uris):
reference = SubElement(signed_info, ds_tag("Reference"), URI=reference_uri)
if i == 0:
transforms = SubElement(reference, ds_tag("Transforms"))
SubElement(transforms, ds_tag("Transform"), Algorithm=namespaces.ds + "enveloped-signature")
SubElement(transforms, ds_tag("Transform"), Algorithm=self.c14n_alg)
digest_method = SubElement( # noqa:F841
reference, ds_tag("DigestMethod"), Algorithm=self.known_digest_tags[self.digest_alg]
)
digest_value = SubElement(reference, ds_tag("DigestValue"))
payload_c14n = self._c14n(c14n_inputs[i], algorithm=self.c14n_alg)
digest = self._get_digest(payload_c14n, self._get_digest_method_by_tag(self.digest_alg))
digest_value.text = digest
signature_value = SubElement(sig_root, ds_tag("SignatureValue"))
return signed_info, signature_value
def _serialize_key_value(self, key, key_info_element):
key_value = SubElement(key_info_element, ds_tag("KeyValue"))
if self.sign_alg.startswith("rsa-"):
rsa_key_value = SubElement(key_value, ds_tag("RSAKeyValue"))
modulus = SubElement(rsa_key_value, ds_tag("Modulus"))
modulus.text = ensure_str(b64encode(long_to_bytes(key.public_key().public_numbers().n)))
exponent = SubElement(rsa_key_value, ds_tag("Exponent"))
exponent.text = ensure_str(b64encode(long_to_bytes(key.public_key().public_numbers().e)))
elif self.sign_alg.startswith("dsa-"):
dsa_key_value = SubElement(key_value, ds_tag("DSAKeyValue"))
for field in "p", "q", "g", "y":
e = SubElement(dsa_key_value, ds_tag(field.upper()))
if field == "y":
key_params = key.public_key().public_numbers()
else:
key_params = key.parameters().parameter_numbers()
e.text = ensure_str(b64encode(long_to_bytes(getattr(key_params, field))))
elif self.sign_alg.startswith("ecdsa-"):
ec_key_value = SubElement(key_value, dsig11_tag("ECKeyValue"), nsmap=dict(dsig11=namespaces.dsig11))
named_curve = SubElement( # noqa:F841
ec_key_value, dsig11_tag("NamedCurve"), URI=self.known_ecdsa_curve_oids[key.curve.name]
)
public_key = SubElement(ec_key_value, dsig11_tag("PublicKey"))
x = key.public_key().public_numbers().x
y = key.public_key().public_numbers().y
public_key.text = ensure_str(b64encode(long_to_bytes(4) + long_to_bytes(x) + long_to_bytes(y)))
if __name__ == "__main__":
signer = XMLEnvelopedEnvelopingSigner()
with open(os.path.join(os.path.dirname(__file__), "test", "example.pem"), "rb") as fh:
crt = fh.read()
with open(os.path.join(os.path.dirname(__file__), "test", "example.key"), "rb") as fh:
key = fh.read()
data = [
etree.parse(os.path.join(os.path.dirname(__file__), "test", "example2.xml")),
etree.parse(os.path.join(os.path.dirname(__file__), "test", "example.xml")),
]
signed_data = signer.sign(data, key=key, cert=crt)
signed_data_str = etree.tostring(signed_data)
with open(os.path.join(os.path.dirname(__file__), "test", "example-125.xml"), "wb") as fh:
fh.write(signed_data_str)
|
