Raise Exceptions during executions, generally prepare for integration with external library users
Be over-cautious with PIN errors: On first sign of wrong PIN, block the PIN from usage (raise Exception when trying to use the PIN). Library user will need to create a new FinTS3PinTanClient instance.
from .types import SegmentSequence
from .utils import MT535_Miniparser, Password, mt940_to_array, compress_datablob, decompress_datablob, SubclassesMixin
from .parser import FinTS3Serializer
+from .exceptions import *
logger = logging.getLogger(__name__)
def _ensure_system_id(self):
raise NotImplemented()
- def _process_response(self, segment, response):
+ def _process_response(self, dialog, segment, response):
pass
- def process_response_message(self, message: FinTSInstituteMessage, internal_send=True):
+ def process_response_message(self, dialog, message: FinTSInstituteMessage, internal_send=True):
bpa = message.find_segment_first(HIBPA3)
if bpa:
self.bpa = bpa
self._call_callbacks(None, response)
- self._process_response(None, response)
+ self._process_response(dialog, None, response)
for seg in message.find_segments(HIRMS2):
for response in seg.responses:
self._call_callbacks(segment, response)
- self._process_response(segment, response)
+ self._process_response(dialog, segment, response)
def _send_with_possible_retry(self, dialog, command_seg, resume_func):
response = dialog._send(command_seg)
if self._standing_dialog:
raise Exception("Cannot double __enter__() {}".format(self))
self._standing_dialog = self._get_dialog()
+ self._standing_dialog.lazy_init = True # FIXME Inelegant
self._standing_dialog.__enter__()
def __exit__(self, exc_type, exc_value, traceback):
resume_func = getattr(self, challenge.resume_method)
return resume_func(challenge.command_seg, response)
- def _process_response(self, segment, response):
+ def _process_response(self, dialog, segment, response):
if response.code == '3920':
self.allowed_security_functions = list(response.parameters)
if self.selected_security_function is None or not self.selected_security_function in self.allowed_security_functions:
# Fall back to onestep
self.set_tan_mechanism('999')
+ if (not dialog.open and response.code.startswith('9')) or response.code in ('9340', '9910', '9930', '9931', '9942'):
+ # Assume all 9xxx errors in a not-yet-open dialog refer to the PIN or authentication
+ # During a dialog also listen for the following codes which may explicitly indicate an
+ # incorrect pin: 9340, 9910, 9930, 9931, 9942
+ # Fail-safe block all further attempts with this PIN
+ if self.pin:
+ self.pin.block()
+ raise FinTSClientPINError("Error during dialog initialization, PIN wrong?")
+
+
def get_tan_mechanisms(self):
"""
Get the available TAN mechanisms.
from fints.utils import Password
from .message import FinTSInstituteMessage, FinTSMessage
+from .exceptions import *
logger = logging.getLogger(__name__)
-class FinTSConnectionError(Exception):
- pass
-
class FinTSHTTPSConnection:
def __init__(self, url):
from .segments.dialog import HKEND1
from .segments.message import HNHBK3, HNHBS1
from .utils import compress_datablob, decompress_datablob
+from .connection import FinTSConnectionError
+from .exceptions import *
logger = logging.getLogger(__name__)
DIALOGUE_ID_UNASSIGNED = '0'
DATA_BLOB_MAGIC = b'python-fints_DIALOG_DATABLOB'
-class FinTSDialogError(Exception):
- pass
-
-
class FinTSDialog:
def __init__(self, client=None, lazy_init=False, enc_mechanism=None, auth_mechanisms=[]):
self.client = client
def init(self, *extra_segments):
if self.paused:
- raise Error("Cannot init() a paused dialog")
+ raise FinTSDialogStateError("Cannot init() a paused dialog")
if self.need_init and not self.open:
segments = [
retval = self.send(*segments, internal_send=True)
self.need_init = False
return retval
- except:
+ except Exception as e:
self.open = False
- raise
+ if isinstance(e, (FinTSConnectionError, FinTSClientError)):
+ raise
+ else:
+ raise FinTSDialogInitError("Couldn't establish dialog with bank, Authentication data wrong?") from e
finally:
self.lazy_init = False
def end(self):
if self.paused:
- raise Error("Cannot end() on a paused dialog")
+ raise FinTSDialogStateError("Cannot end() on a paused dialog")
if self.open:
response = self.send(HKEND1(self.dialogue_id), internal_send=True)
internal_send = kwargs.pop('internal_send', False)
if self.paused:
- raise Error("Cannot send() on a paused dialog")
+ raise FinTSDialogStateError("Cannot send() on a paused dialog")
if not self.open:
if self.lazy_init and self.need_init:
self.init()
if not self.open:
- raise Exception("Cannot send on dialog that is not open")
+ raise FinTSDialogStateError("Cannot send on dialog that is not open")
message = self.new_customer_message()
for s in segments:
if self.dialogue_id == DIALOGUE_ID_UNASSIGNED:
seg = response.find_segment_first(HNHBK3)
if not seg:
- raise ValueError('Could not find dialogue_id')
+ raise FinTSDialogError('Could not find dialogue_id')
self.dialogue_id = seg.dialogue_id
- self.client.process_response_message(response, internal_send=internal_send)
+ self.client.process_response_message(self, response, internal_send=internal_send)
return response
def new_customer_message(self):
if self.paused:
- raise Error("Cannot call new_customer_message() on a paused dialog")
+ raise FinTSDialogStateError("Cannot call new_customer_message() on a paused dialog")
message = FinTSCustomerMessage(self)
message += HNHBK3(0, 300, self.dialogue_id, self.next_message_number[message.DIRECTION])
def finish_message(self, message):
if self.paused:
- raise Error("Cannot call finish_message() on a paused dialog")
+ raise FinTSDialogStateError("Cannot call finish_message() on a paused dialog")
# Create signature(s) in reverse order: from inner to outer
for auth_mech in reversed(self.auth_mechanisms):
def pause(self):
# FIXME Document, test
if self.paused:
- raise Error("Cannot pause a paused dialog")
+ raise FinTSDialogStateError("Cannot pause a paused dialog")
external_dialog = self
external_client = self.client
--- /dev/null
+class FinTSError(Exception):
+ pass
+
+class FinTSClientError(FinTSError):
+ pass
+
+class FinTSClientPINError(FinTSClientError):
+ pass
+
+class FinTSDialogError(FinTSError):
+ pass
+
+class FinTSDialogStateError(FinTSDialogError):
+ pass
+
+class FinTSDialogInitError(FinTSDialogError):
+ pass
+
+class FinTSConnectionError(FinTSError):
+ pass
+
def __init__(self, value):
self.value = value
+ self.blocked = False
@classmethod
@contextmanager
finally:
cls.protected = False
+ def block(self):
+ self.blocked = True
+
def __str__(self):
+ if self.blocked and not self.protected:
+ raise Exception("Refusing to use PIN after block")
return '***' if self.protected else str(self.value)
def __repr__(self):
projects / thirdparty / python-fints.git / commitdiff
