errors.py 3.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101
  1. # Copyright (c) 2020 Tulir Asokan
  2. #
  3. # This Source Code Form is subject to the terms of the Mozilla Public
  4. # License, v. 2.0. If a copy of the MPL was not distributed with this
  5. # file, You can obtain one at http://mozilla.org/MPL/2.0/.
  6. from typing import Any, Dict, Optional
  7. class RPCError(Exception):
  8. pass
  9. class UnexpectedError(RPCError):
  10. pass
  11. class UnexpectedResponse(RPCError):
  12. def __init__(self, resp_type: str, data: Any) -> None:
  13. super().__init__(f"Got unexpected response type {resp_type}")
  14. self.resp_type = resp_type
  15. self.data = data
  16. class NotConnected(RPCError):
  17. pass
  18. class ResponseError(RPCError):
  19. def __init__(self, data: Dict[str, Any], error_type: Optional[str] = None,
  20. message_override: Optional[str] = None) -> None:
  21. self.data = data
  22. msg = message_override or data["message"]
  23. if error_type:
  24. msg = f"{error_type}: {msg}"
  25. super().__init__(msg)
  26. class TimeoutException(ResponseError):
  27. pass
  28. class UnknownIdentityKey(ResponseError):
  29. pass
  30. class CaptchaRequired(ResponseError):
  31. pass
  32. class AuthorizationFailedException(ResponseError):
  33. pass
  34. class UserAlreadyExistsError(ResponseError):
  35. def __init__(self, data: Dict[str, Any]) -> None:
  36. super().__init__(data, message_override="You're already logged in")
  37. class RequestValidationFailure(ResponseError):
  38. def __init__(self, data: Dict[str, Any]) -> None:
  39. results = data["validationResults"]
  40. result_str = ", ".join(results) if isinstance(results, list) else str(results)
  41. super().__init__(data, message_override=result_str)
  42. class InternalError(ResponseError):
  43. def __init__(self, data: Dict[str, Any]) -> None:
  44. exceptions = data.get("exceptions", [])
  45. self.exceptions = exceptions
  46. message = data.get("message")
  47. super().__init__(data, error_type=", ".join(exceptions), message_override=message)
  48. response_error_types = {
  49. "invalid_request": RequestValidationFailure,
  50. "TimeoutException": TimeoutException,
  51. "UserAlreadyExists": UserAlreadyExistsError,
  52. "RequestValidationFailure": RequestValidationFailure,
  53. "UnknownIdentityKey": UnknownIdentityKey,
  54. "CaptchaRequired": CaptchaRequired,
  55. "AuthorizationFailedException": AuthorizationFailedException,
  56. "InternalError": InternalError,
  57. # TODO add rest from https://gitlab.com/signald/signald/-/tree/main/src/main/java/io/finn/signald/exceptions
  58. }
  59. def make_response_error(data: Dict[str, Any]) -> ResponseError:
  60. error_data = data["error"]
  61. if isinstance(error_data, str):
  62. error_data = {"message": error_data}
  63. elif not isinstance(error_data, dict):
  64. error_data = {"message": str(error_data)}
  65. if "message" not in error_data:
  66. error_data["message"] = "no message, see signald logs"
  67. error_type = data.get("error_type")
  68. try:
  69. error_class = response_error_types[error_type]
  70. except KeyError:
  71. return ResponseError(error_data, error_type=error_type)
  72. else:
  73. return error_class(error_data)