(self, params)
| 336 | self.delete_subscriptions(self.subscriptions[:]) |
| 337 | |
| 338 | def activate_session(self, params): |
| 339 | self.logger.info("activate session") |
| 340 | result = ua.ActivateSessionResult() |
| 341 | if self.state != SessionState.Created: |
| 342 | raise utils.ServiceError(ua.StatusCodes.BadSessionIdInvalid) |
| 343 | self.nonce = utils.create_nonce(32) |
| 344 | result.ServerNonce = self.nonce |
| 345 | for _ in params.ClientSoftwareCertificates: |
| 346 | result.Results.append(ua.StatusCode()) |
| 347 | self.state = SessionState.Activated |
| 348 | id_token = params.UserIdentityToken |
| 349 | if isinstance(id_token, ua.UserNameIdentityToken): |
| 350 | if self.user_manager.check_user_token(self, id_token) == False: |
| 351 | raise utils.ServiceError(ua.StatusCodes.BadUserAccessDenied) |
| 352 | self.logger.info("Activated internal session %s for user %s", self.name, self.user) |
| 353 | return result |
| 354 | |
| 355 | def read(self, params): |
| 356 | results = self.iserver.attribute_service.read(params) |
no test coverage detected