| .. | |||||
| __pycache__ | |||||
| __init__.py | |||||
| c14n.py | |||||
| client.py | |||||
| helpers.py | |||||
| server.py | |||||
| simplexml.py | |||||
| transport.py | |||||
| wsse.py | |||||
| xmlsec.py |
| Server IP : 104.21.46.92 / Your IP : 104.23.243.84 Web Server : Apache/2.4.51 (Unix) OpenSSL/1.1.1n System : Linux ip-172-26-8-243 4.19.0-27-cloud-amd64 #1 SMP Debian 4.19.316-1 (2024-06-25) x86_64 User : daemon ( 1) PHP Version : 7.4.24 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /usr/lib/python3/dist-packages/pysimplesoap/ |
Upload File : |
#!/usr/bin/python
# -*- coding: utf-8 -*-
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 3, or (at your option) any
# later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
# for more details.
"""Pythonic simple SOAP Client plugins for WebService Security extensions"""
from __future__ import unicode_literals
import sys
if sys.version > '3':
basestring = unicode = str
import datetime
from decimal import Decimal
import os
import logging
import hashlib
import warnings
from . import __author__, __copyright__, __license__, __version__
from .simplexml import SimpleXMLElement
import random
import string
from hashlib import sha1
def randombytes(N):
return ''.join(random.SystemRandom().choice(string.ascii_uppercase + string.digits) for _ in range(N))
# Namespaces:
WSSE_URI = 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd'
WSU_URI = "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"
XMLDSIG_URI = "http://www.w3.org/2000/09/xmldsig#"
X509v3_URI = "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3"
Base64Binary_URI = "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-soap-message-security-1.0#Base64Binary"
PasswordDigest_URI = "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-username-token-profile-1.0#PasswordDigest"
class UsernameToken:
"WebService Security extension to add a basic credentials to xml request"
def __init__(self, username="", password=""):
self.token = {
'wsse:UsernameToken': {
'wsse:Username': username,
'wsse:Password': password,
}
}
def preprocess(self, client, request, method, args, kwargs, headers, soap_uri):
"Add basic credentials to outgoing message"
# always extract WS Security header and send it
header = request('Header', ns=soap_uri, )
k = 'wsse:Security'
# for backward compatibility, use header if given:
if k in headers:
self.token = headers[k]
# convert the token to xml
header.marshall(k, self.token, ns=False, add_children_ns=False)
header(k)['xmlns:wsse'] = WSSE_URI
#<wsse:UsernameToken xmlns:wsu='http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd'>
def postprocess(self, client, response, method, args, kwargs, headers, soap_uri):
"Analyze incoming credentials"
# TODO: add some password validation callback?
pass
class UsernameDigestToken(UsernameToken):
"""
WebService Security extension to add a http digest credentials to xml request
drift -> time difference from the server in seconds, needed for 'Created' header
"""
def __init__(self, username="", password="", drift=0):
self.username = username
self.password = password
self.drift = datetime.timedelta(seconds=drift)
def preprocess(self, client, request, method, args, kwargs, headers, soap_uri):
header = request('Header', ns=soap_uri, )
wsse = header.add_child('wsse:Security', ns=False)
wsse['xmlns:wsse'] = WSSE_URI
wsse['xmlns:wsu'] = WSU_URI
usertoken = wsse.add_child('wsse:UsernameToken', ns=False)
usertoken.add_child('wsse:Username', self.username, ns=False)
created = (datetime.datetime.utcnow() + self.drift).isoformat() + 'Z'
usertoken.add_child('wsu:Created', created, ns=False)
nonce = randombytes(16)
wssenonce = usertoken.add_child('wsse:Nonce', nonce.encode('base64')[:-1], ns=False)
wssenonce['EncodingType'] = Base64Binary_URI
sha1obj = sha1()
sha1obj.update(nonce + created + self.password)
digest = sha1obj.digest()
password = usertoken.add_child('wsse:Password', digest.encode('base64')[:-1], ns=False)
password['Type'] = PasswordDigest_URI
BIN_TOKEN_TMPL = """<?xml version="1.0" encoding="UTF-8"?>
<wsse:Security soapenv:mustUnderstand="1" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:wsse="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd">
<wsse:BinarySecurityToken EncodingType="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-soap-message-security-1.0#Base64Binary" ValueType="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3" wsu:Id="CertId-45851B081998E431E8132880700036719" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd">
%(certificate)s</wsse:BinarySecurityToken>
<ds:Signature Id="Signature-13" xmlns:ds="http://www.w3.org/2000/09/xmldsig#">
%(signed_info)s
<ds:SignatureValue>%(signature_value)s</ds:SignatureValue>
<ds:KeyInfo Id="KeyId-45851B081998E431E8132880700036720">
<wsse:SecurityTokenReference wsu:Id="STRId-45851B081998E431E8132880700036821" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd">
<wsse:Reference URI="#CertId-45851B081998E431E8132880700036719" ValueType="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3"/>
</wsse:SecurityTokenReference>
</ds:KeyInfo>
</ds:Signature>
</wsse:Security>
"""
class BinaryTokenSignature:
"WebService Security extension to add a basic signature to xml request"
def __init__(self, certificate="", private_key="", password=None, cacert=None):
# read the X509v3 certificate (PEM)
self.certificate = ''.join([line for line in open(certificate)
if not line.startswith("---")])
self.private_key = private_key
self.password = password
self.cacert = cacert
def preprocess(self, client, request, method, args, kwargs, headers, soap_uri):
"Sign the outgoing SOAP request"
# get xml elements:
body = request('Body', ns=soap_uri, )
header = request('Header', ns=soap_uri, )
# prepare body xml attributes to be signed (reference)
body['wsu:Id'] = "id-14"
body['xmlns:wsu'] = WSU_URI
# workaround: copy namespaces so lxml can parse the xml to be signed
for attr, value in request[:]:
if attr.startswith("xmlns"):
body[attr] = value
# use the internal tag xml representation (not the full xml document)
ref_xml = repr(body)
# sign using RSA-SHA1 (XML Security)
from . import xmlsec
vars = xmlsec.rsa_sign(ref_xml, "#id-14",
self.private_key, self.password)
vars['certificate'] = self.certificate
# generate the xml (filling the placeholders)
wsse = SimpleXMLElement(BIN_TOKEN_TMPL % vars)
header.import_node(wsse)
def postprocess(self, client, response, method, args, kwargs, headers, soap_uri):
"Verify the signature of the incoming response"
from . import xmlsec
# get xml elements:
body = response('Body', ns=soap_uri, )
header = response('Header', ns=soap_uri, )
wsse = header("Security", ns=WSSE_URI)
cert = wsse("BinarySecurityToken", ns=WSSE_URI)
# check that the cert (binary token) is coming in the correct format:
self.__check(cert["EncodingType"], Base64Binary_URI)
self.__check(cert["ValueType"], X509v3_URI)
# extract the certificate (in DER to avoid new line & padding issues!)
cert_der = str(cert).decode("base64")
public_key = xmlsec.x509_extract_rsa_public_key(cert_der, binary=True)
# validate the certificate using the certification authority:
if not self.cacert:
warnings.warn("No CA provided, WSSE not validating certificate")
elif not xmlsec.x509_verify(self.cacert, cert_der, binary=True):
raise RuntimeError("WSSE certificate validation failed")
# check body xml attributes was signed correctly (reference)
self.__check(body['xmlns:wsu'], WSU_URI)
ref_uri = body['wsu:Id']
signature = wsse("Signature", ns=XMLDSIG_URI)
signed_info = signature("SignedInfo", ns=XMLDSIG_URI)
signature_value = signature("SignatureValue", ns=XMLDSIG_URI)
# TODO: these sanity checks should be moved to xmlsec?
self.__check(signed_info("Reference", ns=XMLDSIG_URI)['URI'], "#" + ref_uri)
self.__check(signed_info("SignatureMethod", ns=XMLDSIG_URI)['Algorithm'],
XMLDSIG_URI + "rsa-sha1")
self.__check(signed_info("Reference", ns=XMLDSIG_URI)("DigestMethod", ns=XMLDSIG_URI)['Algorithm'],
XMLDSIG_URI + "sha1")
# TODO: check KeyInfo uses the correct SecurityTokenReference
# workaround: copy namespaces so lxml can parse the xml to be signed
for attr, value in response[:]:
if attr.startswith("xmlns"):
body[attr] = value
# use the internal tag xml representation (not the full xml document)
ref_xml = xmlsec.canonicalize(repr(body))
# verify the signed hash
computed_hash = xmlsec.sha1_hash_digest(ref_xml)
digest_value = str(signed_info("Reference", ns=XMLDSIG_URI)("DigestValue", ns=XMLDSIG_URI))
if computed_hash != digest_value:
raise RuntimeError("WSSE SHA1 hash digests mismatch")
# workaround: prepare the signed info (assure the parent ns is present)
signed_info['xmlns'] = XMLDSIG_URI
xml = repr(signed_info)
# verify the signature using RSA-SHA1 (XML Security)
ok = xmlsec.rsa_verify(xml, str(signature_value), public_key)
if not ok:
raise RuntimeError("WSSE RSA-SHA1 signature verification failed")
# TODO: remove any unsigned part from the xml?
def __check(self, value, expected, msg="WSSE sanity check failed"):
if value != expected:
raise RuntimeError(msg)
| .. | |||||
| __pycache__ | |||||
| __init__.py | |||||
| c14n.py | |||||
| client.py | |||||
| helpers.py | |||||
| server.py | |||||
| simplexml.py | |||||
| transport.py | |||||
| wsse.py | |||||
| xmlsec.py |