send a CreateSessionRequest to server with reasonable parameters. If you want o modify settings look at code of this methods and make your own
(self)
| 365 | return self.uaclient.find_servers_on_network(params) |
| 366 | |
| 367 | def create_session(self): |
| 368 | """ |
| 369 | send a CreateSessionRequest to server with reasonable parameters. |
| 370 | If you want o modify settings look at code of this methods |
| 371 | and make your own |
| 372 | """ |
| 373 | desc = ua.ApplicationDescription() |
| 374 | desc.ApplicationUri = self.application_uri |
| 375 | desc.ProductUri = self.product_uri |
| 376 | desc.ApplicationName = ua.LocalizedText(self.name) |
| 377 | desc.ApplicationType = ua.ApplicationType.Client |
| 378 | |
| 379 | params = ua.CreateSessionParameters() |
| 380 | # at least 32 random bytes for server to prove possession of private key (specs part 4, 5.6.2.2) |
| 381 | nonce = utils.create_nonce(32) |
| 382 | params.ClientNonce = nonce |
| 383 | params.ClientCertificate = self.security_policy.client_certificate |
| 384 | params.ClientDescription = desc |
| 385 | params.EndpointUrl = self.server_url.geturl() |
| 386 | params.SessionName = self.description + " Session" + str(self._session_counter) |
| 387 | params.RequestedSessionTimeout = self.session_timeout |
| 388 | params.MaxResponseMessageSize = 0 # means no max size |
| 389 | response = self.uaclient.create_session(params) |
| 390 | if self.security_policy.client_certificate is None: |
| 391 | data = nonce |
| 392 | else: |
| 393 | data = self.security_policy.client_certificate + nonce |
| 394 | self.security_policy.asymmetric_cryptography.verify(data, response.ServerSignature.Signature) |
| 395 | self._server_nonce = response.ServerNonce |
| 396 | if not self.security_policy.server_certificate: |
| 397 | self.security_policy.server_certificate = response.ServerCertificate |
| 398 | elif self.security_policy.server_certificate != response.ServerCertificate: |
| 399 | raise ua.UaError("Server certificate mismatch") |
| 400 | # remember PolicyId's: we will use them in activate_session() |
| 401 | ep = Client.find_endpoint(response.ServerEndpoints, self.security_policy.Mode, self.security_policy.URI) |
| 402 | self._policy_ids = ep.UserIdentityTokens |
| 403 | if self.session_timeout != response.RevisedSessionTimeout: |
| 404 | _logger.warning("Requested session timeout to be %dms, got %dms instead", |
| 405 | self.secure_channel_timeout, |
| 406 | response.RevisedSessionTimeout) |
| 407 | self.session_timeout = response.RevisedSessionTimeout |
| 408 | self.keepalive = KeepAlive( |
| 409 | self, min(self.session_timeout, self.secure_channel_timeout) * 0.7) # 0.7 is from spec |
| 410 | self.keepalive.start() |
| 411 | return response |
| 412 | |
| 413 | def server_policy_id(self, token_type, default): |
| 414 | """ |
no test coverage detected