import base64 import hashlib import os from zeep import ns from zeep.wsse import utils class UsernameToken: """UsernameToken Profile 1.1 https://docs.oasis-open.org/wss/v1.1/wss-v1.1-spec-os-UsernameTokenProfile.pdf Example response using PasswordText:: scott password Example using PasswordDigest:: NNK weYI3nXd8LjMNVksCKFV8t3rgHh3Rw== WScqanjCEAC4mQoBE07sAQ== 2003-07-16T01:24:32Z """ username_token_profile_ns = "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-username-token-profile-1.0" # noqa soap_message_secutity_ns = "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-soap-message-security-1.0" # noqa def __init__( self, username, password=None, password_digest=None, use_digest=False, nonce=None, created=None, timestamp_token=None, zulu_timestamp=None, hash_password=None, ): """ Some SOAP services want zulu timestamps with Z in timestamps and in password digests they may want password to be hashed before adding it to nonce and created. """ self.username = username self.password = password self.password_digest = password_digest self.nonce = nonce self.created = created self.use_digest = use_digest self.timestamp_token = timestamp_token self.zulu_timestamp = zulu_timestamp self.hash_password = hash_password def apply(self, envelope, headers): security = utils.get_security_header(envelope) # The token placeholder might already exists since it is specified in # the WSDL. token = security.find("{%s}UsernameToken" % ns.WSSE) if token is None: token = utils.WSSE.UsernameToken() security.append(token) if self.timestamp_token is not None: security.append(self.timestamp_token) # Create the sub elements of the UsernameToken element elements = [utils.WSSE.Username(self.username)] if self.password is not None or self.password_digest is not None: if self.use_digest: elements.extend(self._create_password_digest()) else: elements.extend(self._create_password_text()) token.extend(elements) return envelope, headers def verify(self, envelope): pass def _create_password_text(self): return [ utils.WSSE.Password( self.password, Type="%s#PasswordText" % self.username_token_profile_ns ) ] def _create_password_digest(self): if self.nonce: nonce = self.nonce.encode("utf-8") else: nonce = os.urandom(16) timestamp = utils.get_timestamp(self.created, self.zulu_timestamp) if isinstance(self.password, str): password = self.password.encode("utf-8") else: password = self.password # digest = Base64 ( SHA-1 ( nonce + created + password ) ) if not self.password_digest and self.hash_password: digest = base64.b64encode( hashlib.sha1( nonce + timestamp.encode("utf-8") + hashlib.sha1(password).digest() ).digest() ).decode("ascii") elif not self.password_digest: digest = base64.b64encode( hashlib.sha1(nonce + timestamp.encode("utf-8") + password).digest() ).decode("ascii") else: digest = self.password_digest return [ utils.WSSE.Password( digest, Type="%s#PasswordDigest" % self.username_token_profile_ns ), utils.WSSE.Nonce( base64.b64encode(nonce).decode("utf-8"), EncodingType="%s#Base64Binary" % self.soap_message_secutity_ns, ), utils.WSU.Created(timestamp), ]