(self, params, username, password)
| 485 | params.UserTokenSignature.Signature = sig |
| 486 | |
| 487 | def _add_user_auth(self, params, username, password): |
| 488 | params.UserIdentityToken = ua.UserNameIdentityToken() |
| 489 | params.UserIdentityToken.UserName = username |
| 490 | policy_uri = self.server_policy_uri(ua.UserTokenType.UserName) |
| 491 | if not policy_uri or policy_uri == security_policies.POLICY_NONE_URI: |
| 492 | # see specs part 4, 7.36.3: if the token is NOT encrypted, |
| 493 | # then the password only contains UTF-8 encoded password |
| 494 | # and EncryptionAlgorithm is null |
| 495 | if self._password: |
| 496 | _logger.warning("Sending plain-text password") |
| 497 | params.UserIdentityToken.Password = password.encode('utf8') |
| 498 | params.UserIdentityToken.EncryptionAlgorithm = None |
| 499 | elif self._password: |
| 500 | data, uri = self._encrypt_password(password, policy_uri) |
| 501 | params.UserIdentityToken.Password = data |
| 502 | params.UserIdentityToken.EncryptionAlgorithm = uri |
| 503 | params.UserIdentityToken.PolicyId = self.server_policy_id(ua.UserTokenType.UserName, "username_basic256") |
| 504 | |
| 505 | def _encrypt_password(self, password, policy_uri): |
| 506 | pubkey = uacrypto.x509_from_der(self.security_policy.server_certificate).public_key() |
no test coverage detected