mirror of
https://github.com/ProtonMail/proton-python-client.git
synced 2026-01-11 19:56:26 +00:00
Add new exceptions
This commit is contained in:
parent
bfe7454fa1
commit
75a910151e
5 changed files with 50 additions and 30 deletions
|
|
@ -1 +1,2 @@
|
|||
from .api import Session, ProtonError # noqa
|
||||
from .api import Session # noqa
|
||||
from .exceptions import ProtonError # noqa
|
||||
|
|
|
|||
|
|
@ -1,26 +1,12 @@
|
|||
import base64
|
||||
import json
|
||||
|
||||
import gnupg
|
||||
import requests
|
||||
|
||||
from .cert_pinning import TLSPinningAdapter
|
||||
from .srp import User as PmsrpUser
|
||||
from .constants import DEFAULT_TIMEOUT, SRP_MODULUS_KEY, SRP_MODULUS_KEY_FINGERPRINT
|
||||
|
||||
|
||||
class ProtonError(Exception):
|
||||
def __init__(self, ret):
|
||||
self.code = ret['Code']
|
||||
self.error = ret['Error']
|
||||
try:
|
||||
self.headers = ret["Headers"]
|
||||
except KeyError:
|
||||
self.headers = ""
|
||||
|
||||
super().__init__("[{}] {} {}".format(
|
||||
self.code, self.error, self.headers
|
||||
))
|
||||
from .exceptions import ProtonError, TLSPinningError, NewConnectionError, UnknownConnectionError
|
||||
|
||||
|
||||
class Session:
|
||||
|
|
@ -106,12 +92,19 @@ class Session:
|
|||
if fct is None:
|
||||
raise ValueError("Unknown method: {}".format(method))
|
||||
|
||||
ret = fct(
|
||||
self.__api_url + endpoint,
|
||||
headers=additional_headers,
|
||||
json=jsondata,
|
||||
timeout=self.__timeout
|
||||
)
|
||||
try:
|
||||
ret = fct(
|
||||
self.__api_url + endpoint,
|
||||
headers=additional_headers,
|
||||
json=jsondata,
|
||||
timeout=self.__timeout
|
||||
)
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
raise NewConnectionError(e)
|
||||
except TLSPinningError as e:
|
||||
raise TLSPinningError(e)
|
||||
except (Exception, requests.exceptions.BaseHTTPError) as e:
|
||||
raise UnknownConnectionError(e)
|
||||
|
||||
try:
|
||||
ret = ret.json()
|
||||
|
|
|
|||
|
|
@ -2,22 +2,15 @@ import base64
|
|||
import hashlib
|
||||
from ssl import DER_cert_to_PEM_cert
|
||||
|
||||
import requests
|
||||
from OpenSSL import crypto
|
||||
from requests.adapters import HTTPAdapter
|
||||
from urllib3.connectionpool import HTTPSConnectionPool
|
||||
from urllib3.poolmanager import PoolManager
|
||||
from urllib3.util.timeout import Timeout
|
||||
|
||||
from .exceptions import TLSPinningError
|
||||
from .constants import PUBKEY_HASH_DICT
|
||||
|
||||
|
||||
class TLSPinningError(requests.exceptions.SSLError):
|
||||
def __init__(self, strerror):
|
||||
self.strerror = strerror
|
||||
super(TLSPinningError, self).__init__(strerror)
|
||||
|
||||
|
||||
class TLSPinningHTTPSConnectionPool(HTTPSConnectionPool):
|
||||
"""Verify the certificate upon each connection"""
|
||||
def __init__(
|
||||
|
|
|
|||
29
proton/exceptions.py
Normal file
29
proton/exceptions.py
Normal file
|
|
@ -0,0 +1,29 @@
|
|||
class ProtonError(Exception):
|
||||
def __init__(self, ret):
|
||||
self.code = ret['Code']
|
||||
self.error = ret['Error']
|
||||
try:
|
||||
self.headers = ret["Headers"]
|
||||
except KeyError:
|
||||
self.headers = ""
|
||||
|
||||
super().__init__("{}".format(self.error))
|
||||
|
||||
|
||||
class ProtonNetworkError(Exception):
|
||||
def __init__(self, message, additional_context=None):
|
||||
self.message = message
|
||||
self.additional_context = additional_context
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class TLSPinningError(ProtonNetworkError):
|
||||
"""TLS Pinning exception"""
|
||||
|
||||
|
||||
class NewConnectionError(ProtonNetworkError):
|
||||
"""Network Error"""
|
||||
|
||||
|
||||
class UnknownConnectionError(ProtonNetworkError):
|
||||
"""UnknownConnectionError"""
|
||||
|
|
@ -1,3 +1,7 @@
|
|||
[flake8]
|
||||
ignore = C901, W503, E402
|
||||
max-line-length = 100
|
||||
|
||||
[metadata]
|
||||
long_description = file: README.md
|
||||
long_description_content_type = text/markdown
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue