Merge pull request #47 from MoritzR/support-tan

TAN authentication

Author: MoritzR

README.md

@@ -108,7 +108,7 @@ The configuration file is located at `~/.config/fints2ledger/config.yml` (XDG co
 | `--files-path PATH` | Directory where fints2ledger stores its config files | — |
 | `-f`, `--journal-file FILE` | Path to the ledger journal file | `ledger.journalFile` |
 | `--date DATE` | Start date for fetching transactions (e.g. `25.01.2023`, `90 days ago`, `last monday`). Default: 90 days ago | — |
-| `--python-command CMD` | Python executable to use. Default: `python3` | — |
+| `--python-command PATH` | Python executable to use. Default: `python3` | — |
 | `--demo` | Run with sample transactions, without calling a FinTS endpoint | — |
 | `--config` | Open the config editor UI | — |
 | `--from-csv-file FILE` | Read transactions from a CSV file instead of a FinTS endpoint | — |

data/pyfints.py

@@ -1,74 +1,238 @@
-from fints.client import FinTS3PinTanClient
+from fints.client import FinTS3PinTanClient, NeedTANResponse
 from mt940.models import Date
-import os
+import base64
+import hashlib
 import json
+import os
+import sys
+import tempfile
 
-def retrieve_transactions(
-    account, blz, password, endpoint, selected_account, start, end
-):
-    client = FinTS3PinTanClient(blz, account, password, endpoint, product_id = "EC449295201FA9BE5040B9154")
-    return TRetriever(client, selected_account).get_hbci_transactions(start, end)
+SIMULATED_TAN_METHODS = [
+    {"id": "900", "name": "Simulated pushTAN"},
+    {"id": "901", "name": "Simulated photoTAN"},
+]
+SIMULATED_PHOTOTAN_PNG = base64.b64decode(
+    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII="
+)
 
 
-class TRetriever:
-    def __init__(self, client, accountnumber):
-        self.client = client
-        self.accountnumber = accountnumber
+def send(message_type, **values):
+    print(json.dumps({"type": message_type, **values}), flush=True)
 
-    def get_hbci_transactions(self, start_date, end_date):
-        accounts = self.client.get_sepa_accounts()
 
-        account = self.__find_matching_account(accounts, self.accountnumber)
+def receive(expected_type):
+    line = sys.stdin.readline()
+    if not line:
+        raise RuntimeError("The fints2ledger process closed the protocol input")
+    message = json.loads(line)
+    if message.get("type") != expected_type:
+        raise RuntimeError(
+            f"Expected protocol message '{expected_type}', got {message.get('type')!r}"
+        )
+    return message
+
+
+def state_path(args):
+    identity = "\0".join((args["endpoint"], args["blz"], args["account"]))
+    digest = hashlib.sha256(identity.encode()).hexdigest()[:24]
+    return os.path.join(args["stateDirectory"], f"fints-state-{digest}.bin")
+
+
+def load_state(path):
+    try:
+        with open(path, "rb") as reader:
+            return reader.read()
+    except FileNotFoundError:
+        return None
+
+
+def store_state(path, data):
+    os.makedirs(os.path.dirname(path), mode=0o700, exist_ok=True)
+    fd, temporary_path = tempfile.mkstemp(dir=os.path.dirname(path))
+    try:
+        os.fchmod(fd, 0o600)
+        with os.fdopen(fd, "wb") as writer:
+            writer.write(data)
+        os.replace(temporary_path, path)
+        os.chmod(path, 0o600)
+    except BaseException:
+        try:
+            os.unlink(temporary_path)
+        except FileNotFoundError:
+            pass
+        raise
+
+
+def choose_tan_settings(client):
+    if os.environ.get("FINTS2LEDGER_SIMULATE_TAN") == "1":
+        send("tan_methods", methods=SIMULATED_TAN_METHODS)
+        selected = receive("tan_method")["id"]
+        if selected not in {method["id"] for method in SIMULATED_TAN_METHODS}:
+            raise RuntimeError(f"Unknown simulated TAN method selected: {selected!r}")
+        if selected == "901":
+            send(
+                "tan_challenge",
+                challenge="Simulated photoTAN challenge",
+                decoupled=False,
+                hhduc=None,
+                matrix={
+                    "mimeType": "image/png",
+                    "data": base64.b16encode(SIMULATED_PHOTOTAN_PNG).decode("ascii"),
+                    "base64Data": base64.b64encode(SIMULATED_PHOTOTAN_PNG).decode(
+                        "ascii"
+                    ),
+                },
+            )
+            receive("tan")
+        # The simulated method is deliberately not passed to pyfints: the bank
+        # did not advertise it and would reject it. Continue as a TAN-less bank.
+        return
+
+    if not client.get_current_tan_mechanism():
+        client.fetch_tan_mechanisms()
+        mechanisms = list(client.get_tan_mechanisms().items())
+        if not mechanisms:
+            # TAN-less banks do not advertise any mechanisms. There is no
+            # mechanism or medium to select, so continue with the dialog.
+            return
+        if len(mechanisms) == 1:
+            client.set_tan_mechanism(mechanisms[0][0])
+        else:
+            send(
+                "tan_methods",
+                methods=[
+                    {"id": identifier, "name": mechanism.name}
+                    for identifier, mechanism in mechanisms
+                ],
+            )
+            selected = receive("tan_method")["id"]
+            if selected not in dict(mechanisms):
+                raise RuntimeError(f"Unknown TAN method selected: {selected!r}")
+            client.set_tan_mechanism(selected)
+
+    if client.selected_tan_medium is None and client.is_tan_media_required():
+        media = client.get_tan_media()[1]
+        if len(media) == 1:
+            client.set_tan_medium(media[0])
+        elif len(media) == 0:
+            # Some banks demand a medium field but return no selectable media.
+            client.selected_tan_medium = ""
+        else:
+            send(
+                "tan_media",
+                media=[
+                    {
+                        "index": index,
+                        "name": medium.tan_medium_name,
+                        "mobile": medium.mobile_number_masked,
+                    }
+                    for index, medium in enumerate(media)
+                ],
+            )
+            selected = receive("tan_medium")["index"]
+            if not isinstance(selected, int) or not 0 <= selected < len(media):
+                raise RuntimeError(f"Unknown TAN medium selected: {selected!r}")
+            client.set_tan_medium(media[selected])
+
+
+def send_tan(client, response):
+    while isinstance(response, NeedTANResponse):
+        matrix = None
+        if response.challenge_matrix:
+            mime_type, image_data = response.challenge_matrix
+            matrix = {
+                "mimeType": mime_type,
+                "data": base64.b16encode(image_data).decode("ascii"),
+                "base64Data": base64.b64encode(image_data).decode("ascii"),
+            }
+
+        send(
+            "tan_challenge",
+            challenge=response.challenge or "A TAN is required",
+            decoupled=bool(response.decoupled),
+            hhduc=response.challenge_hhduc,
+            matrix=matrix,
+        )
+        tan = receive("tan")["value"]
+        if not isinstance(tan, str):
+            raise RuntimeError("The TAN must be a string")
+        response = client.send_tan(response, tan)
+    return response
 
+
+class TransactionRetriever:
+    def __init__(self, client, account_number):
+        self.client = client
+        self.account_number = account_number
+
+    def get_transactions(self, start_date, end_date):
+        accounts = self.client.get_sepa_accounts()
+        account = self.find_matching_account(accounts)
         return self.client.get_transactions(account, start_date, end_date)
 
-    def __find_matching_account(self, accounts, accountnumber):
+    def find_matching_account(self, accounts):
         for account in accounts:
-            if account.accountnumber == accountnumber:
+            if account.accountnumber == self.account_number:
                 return account
-        raise Exception(
-            f"Could not find a matching account for account number '{accountnumber}'. Possible accounts: {accounts}"
+        raise RuntimeError(
+            f"Could not find account '{self.account_number}'. Possible accounts: {accounts}"
         )
 
-date_format="%Y/%m/%d"
+
+DATE_FORMAT = "%Y/%m/%d"
+
+
 def transaction_to_object(transaction):
     hbci_data = transaction.data
-
-    date = hbci_data["date"].strftime(date_format)
-    amount = str(hbci_data["amount"].amount)
-    currency = hbci_data["amount"].currency
-    # GLS bank provides no "posting_text", "AdditionalEntryInformation" seems to be equivalent
-    posting_text = hbci_data.get("posting_text", hbci_data.get("AdditionalEntryInformation", None))
-    applicant_name = hbci_data["applicant_name"]
-    purpose = hbci_data["purpose"]
+    posting_text = hbci_data.get(
+        "posting_text", hbci_data.get("AdditionalEntryInformation")
+    )
     return {
-        "date": date,
-        "amount": amount,
-        "currency": currency,
+        "date": hbci_data["date"].strftime(DATE_FORMAT),
+        "amount": str(hbci_data["amount"].amount),
+        "currency": hbci_data["amount"].currency,
         "posting": (posting_text or "").strip(),
-        "payee": (applicant_name or "").strip(),
-        "purpose": (purpose or "").strip(),
+        "payee": (hbci_data["applicant_name"] or "").strip(),
+        "purpose": (hbci_data["purpose"] or "").strip(),
     }
 
 
-def main():
-    args = json.loads(os.environ["FINTS2LEDGER_ARGS"])
-    transactions = retrieve_transactions(
-        account=args["account"],
-        blz=args["blz"],
-        password=args["password"],
-        endpoint=args["endpoint"],
-        selected_account=args["selectedAccount"],
-        start=date_string_to_mt940_date(args["start"]),
-        end=date_string_to_mt940_date(args["end"]),
+def date_string_to_mt940_date(date_string):
+    year, month, day = date_string.split("/")
+    return Date(year=year, month=month, day=day)
+
+
+def run(args):
+    path = state_path(args)
+    client = FinTS3PinTanClient(
+        args["blz"],
+        args["account"],
+        args["password"],
+        args["endpoint"],
+        product_id="EC449295201FA9BE5040B9154",
+        from_data=load_state(path),
     )
-    converted = json.dumps(list(map(transaction_to_object, transactions)))
-    print(converted)
+    choose_tan_settings(client)
 
+    with client:
+        client.init_tan_response = send_tan(client, client.init_tan_response)
+        result = TransactionRetriever(client, args["selectedAccount"]).get_transactions(
+            date_string_to_mt940_date(args["start"]),
+            date_string_to_mt940_date(args["end"]),
+        )
+        result = send_tan(client, result)
+        transactions = [transaction_to_object(transaction) for transaction in result]
 
-def date_string_to_mt940_date(date_string):
-    parts = date_string.split("/")
-    return Date(year=parts[0], month=parts[1], day=parts[2])
+    store_state(path, client.deconstruct(including_private=True))
+    send("transactions", transactions=transactions)
+
+
+def main():
+    try:
+        run(receive("start")["arguments"])
+    except Exception as exception:
+        send("error", message=str(exception))
+        raise
 
 
 if __name__ == "__main__":