commit a047715a9bd3ff3f65ff3fe286589e25b44bc846
parent b38593b7aaa372270c672165337726646aa94926
Author: MS <ms@taler.net>
Date: Tue, 30 Jun 2020 16:46:00 +0200
rename script
Diffstat:
| M | cli/libeufin-cli | | | 1161 | +++++++++++++++++-------------------------------------------------------------- |
| D | cli/libeufin-cli-new | | | 381 | ------------------------------------------------------------------------------- |
2 files changed, 242 insertions(+), 1300 deletions(-)
diff --git a/cli/libeufin-cli b/cli/libeufin-cli
@@ -1,1058 +1,381 @@
#!/usr/bin/env python3
-import os
import click
import json
import hashlib
import errno
from datetime import datetime
-from requests import post, get
+from requests import post, get, auth
from urllib.parse import urljoin
from getpass import getpass
-@click.group()
+@click.group(help="""
+General utility to invoke HTTP REST services offered by Nexus.
+Consider also invoking the 'nexus' command directly, for example
+to create new users.
+"""
+)
def cli():
pass
@cli.group()
@click.pass_context
-def admin(ctx):
+def bank_connection(ctx):
pass
-@admin.command(help="Instruct the sandbox bank to create a new EBICS host ID.")
-@click.option(
- "--host-id",
- help="EBICS host ID",
- required=True
-)
-@click.option(
- "--ebics-version",
- help="EBICS version to support",
- required=True
-)
-@click.argument(
- "bank-base-url"
-)
-@click.pass_obj
-def add_host(obj, host_id, ebics_version, bank_base_url):
- url = urljoin(bank_base_url, "/ebics/hosts")
- body = dict(
- hostId=host_id,
- ebicsVersion=ebics_version
- )
- try:
- resp = post(url, json=body)
- except Exception:
- print("Could not reach the Bank")
- return
-
- print(resp.content.decode("utf-8"))
-
-@admin.command(help="Instruct the sandbox bank to create a new EBICS Subscriber")
-@click.pass_obj
-@click.option(
- "--user-id",
- help="EBICS user ID",
- required=True
-)
-@click.option(
- "--partner-id",
- help="EBICS partner ID",
- required=True
-)
-@click.option(
- "--host-id",
- help="EBICS host ID",
- required=True
-)
-@click.option(
- "--name",
- help="Name of the person associated with the user ID",
- required=True
-)
-@click.argument(
- "bank-base-url"
-)
-def add_subscriber(obj, user_id, partner_id, host_id, name, bank_base_url):
- body = dict(
- userID=user_id,
- partnerID=partner_id,
- hostID=host_id,
- name=name
- )
-
- url = urljoin(bank_base_url, "/admin/add/subscriber")
- try:
- resp = post(url, json=body)
- except Exception as e:
- print(e)
- return
-
- print(resp.content.decode("utf-8"))
-
@cli.group()
@click.pass_context
-def ebics(ctx):
+def bank_accounts(ctx):
pass
@cli.group()
@click.pass_context
-def taler(ctx):
+def sandbox(ctx):
pass
-@cli.group()
-def native():
- pass
-
-@ebics.command(help="Send the HEV message to the bank.")
+@bank_connection.command(help="export backup")
+@click.option("--connection-name", help="Name of the bank connection to backup", required=True)
+@click.option("--nexus-user-id", help="Nexus user ID", required=True)
+@click.option("--nexus-password", help="Nexus password", required=True)
+@click.option("--passphrase", help="Passphrase for locking the backup", required=True)
+@click.option("--output-file", help="Where to store the backup", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Customer id",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def hev(obj, account_id, nexus_base_url):
- url = urljoin(nexus_base_url, "/ebics/{}/sendHev".format(account_id))
- try:
- resp = get(url)
- except Exception:
- print("Could not reach the bank")
- return
-
- print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Restore private keys backup.")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--backup-file",
- help="File where the backup is stored",
- required=False,
- default="/tmp/backup.json")
-@click.argument(
- "nexus-base-url"
-)
-def restore(obj, account_id, backup_file, nexus_base_url):
- try:
- backup = open(backup_file, "r")
- except Exception:
- print("Could not open the backup at {}".format(backup_file))
- return
-
- backup_json = json.loads(backup.read())
- backup.close()
- passphrase = getpass("Passphrase: ")
- backup_json["passphrase"] = passphrase
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/restoreBackup".format(account_id))
-
- try:
- response = post(url, json=backup_json)
- except Exception:
- print("Could not reach the bank")
- return
-
- print("Status code: {}".format(response.status_code))
- print("Nexus says: {}".format(response.content.decode("utf-8")))
-
-@ebics.command(help="Obtain public keys of a nexus ebics account")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def pubkeys(obj, account_id, nexus_base_url):
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/pubkeys".format(account_id))
-
- try:
- response = get(url)
- except Exception as e:
- print("Could not reach nexus:", e)
- return
-
- print(response.content.decode("utf-8"))
-
-@ebics.command(help="Obtain passphrase-protected private keys")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--output-file",
- help="File that will store the backup",
- required=False,
- default="/tmp/backup.json")
-@click.argument(
- "nexus-base-url"
-)
-def backup(obj, account_id, output_file, nexus_base_url):
- passphrase = getpass("Passphrase: ")
- passphrase_again = getpass("Passphrase (again): ")
-
- if passphrase != passphrase_again:
- print("Passphrase differs, exiting.")
- return
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/backup".format(account_id))
-
+def export_backup(obj, connection_name, nexus_user_id, nexus_password, passphrase, output_file, nexus_base_url):
+ url = urljoin(nexus_base_url, "/bank-connections/{}/export-backup".format(connection_name))
try:
- response = post(url, json=dict(passphrase=passphrase))
+ resp = post(
+ url, json=dict(passphrase=passphrase),
+ auth=auth.HTTPBasicAuth(nexus_user_id, nexus_password)
+ )
except Exception:
- print("Could not reach the bank")
- return
+ print("Could not reach nexus")
+ exit(1)
- if response.status_code != 200:
- print("Received unsuccessful status code: {}".format(response.status_code))
- return
-
output = open(output_file, "w+")
- output.write(response.text)
+ output.write(resp.content.decode("utf-8"))
output.close()
print("Backup stored in {}".format(output_file))
-@ebics.command(help="Ask for list of Subscriber's bank accounts (requires HTD first)")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def bank_accounts(obj, account_id, nexus_base_url):
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/accounts".format(account_id))
- try:
- resp = get(url)
- except Exception:
- print("Could not reach the nexus")
- return
-
- print(resp.content.decode("utf-8"))
-
-
-@ebics.command(help="Send test upload message (TSU)")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def tsu(obj, account_id, nexus_base_url):
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendTSU".format(account_id))
- try:
- resp = post(url)
- except Exception:
- print("Could not reach the nexus")
- return
-
- print(resp.content.decode("utf-8"))
-
-
-@ebics.command(help="Send test download message (TSD)")
+@bank_connection.command(help="restore backup")
+@click.option("--connection-name", help="Name of the bank connection to backup", required=True)
+@click.option("--nexus-user-id", help="Nexus user ID", required=True)
+@click.option("--nexus-password", help="Nexus password", required=True)
+@click.option("--backup-file", help="Back file", required=True)
+@click.option("--passphrase", help="Passphrase for locking the backup", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def tsd(obj, account_id, nexus_base_url):
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendTSD".format(account_id))
+def restore_backup(obj, backup_file, passphrase, nexus_base_url, nexus_user_id, nexus_password, connection_name):
+ url = urljoin(nexus_base_url, "/bank-connections")
try:
- resp = post(url)
+ backup = open(backup_file, "r")
except Exception:
- print("Could not reach the nexus")
+ print("Could not open the backup at {}".format(backup_file))
return
- print(resp.content.decode("utf-8"))
-
+ backup_json = json.loads(backup.read())
+ backup.close()
-@ebics.command(help="Send HAA message")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def haa(obj, account_id, nexus_base_url):
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendHAA".format(account_id))
try:
- resp = post(url)
- except Exception:
- print("Could not reach the bank")
- return
+ resp = post(
+ url,
+ json=dict(
+ name=connection_name,
+ data=backup_json,
+ passphrase=passphrase,
+ source="backup"
+ ),
+ auth=auth.HTTPBasicAuth(nexus_user_id, nexus_password)
- print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Send HVZ message")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def hvz(obj, account_id, nexus_base_url):
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendHVZ".format(account_id))
- try:
- resp = post(url)
+ )
except Exception:
- print("Could not reach the bank")
- return
+ print("Could not reach nexus")
+ exit(1)
print(resp.content.decode("utf-8"))
-@ebics.command(help="Send HVU message")
+@bank_connection.command(help="make new Ebics bank connection")
+@click.option("--connection-name", help="Connection ID", required=True)
+@click.option("--ebics-url", help="EBICS URL", required=True)
+@click.option("--host-id", help="Host ID", required=True)
+@click.option("--partner-id", help="Partner ID", required=True)
+@click.option("--ebics-user-id", help="Ebics user ID", required=True)
+@click.option("--nexus-user-id", help="Nexus user ID", required=True)
+@click.option("--nexus-password", help="Nexus password", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def hvu(obj, account_id, nexus_base_url):
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendHVU".format(account_id))
+def new_ebics_connection(obj, connection_name, ebics_url, host_id, partner_id,
+ nexus_user_id, nexus_password, nexus_base_url, ebics_user_id):
+ url = urljoin(nexus_base_url, "/bank-connections")
+ body = dict(
+ name=connection_name,
+ source="new",
+ type="ebics",
+ data=dict(ebicsURL=ebics_url, hostID=host_id, partnerID=partner_id, userID=ebics_user_id)
+ )
try:
- resp = post(url)
+ resp = post(url, json=body, auth=auth.HTTPBasicAuth(nexus_user_id, nexus_password))
except Exception:
- print("Could not reach the bank")
- return
-
+ print("Could not reach nexus")
+ exit(1)
print(resp.content.decode("utf-8"))
-@ebics.command(help="Send HPD message")
+@bank_connection.command(help="bootstrap the bank connection")
+@click.option("--connection-name", help="Connection ID", required=True)
+@click.option("--nexus-user-id", help="Nexus user ID", required=True)
+@click.option("--nexus-password", help="Nexus password", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def hpd(obj, account_id, nexus_base_url):
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendHPD".format(account_id))
+def bootstrap_bank_connection(obj, connection_name, nexus_user_id, nexus_password, nexus_base_url):
+ url = urljoin(nexus_base_url, "/bank-connections/{}/connect".format(connection_name))
try:
- resp = post(url)
+ resp = post(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
except Exception:
- print("Could not reach the bank")
+ print("Could not reach nexus")
return
-
- print(resp.content.decode("utf-8"))
-
-
-@ebics.command(help="Send C52 message")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--date-range",
- help="Date range for the query",
- nargs=2,
- required=False,
-)
-@click.argument(
- "nexus-base-url"
-)
-def c52(obj, account_id, date_range, nexus_base_url):
- if date_range is not None and len(date_range) == 2:
- req = dict(dateRange=dict(start=date_range[0], end=date_range[1]))
- else:
- req = dict()
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendC52".format(account_id))
- resp = post(url, json=req)
- print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Send CRZ message")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--date-range",
- help="Date range for the query",
- nargs=2,
- required=False,
-)
-@click.argument(
- "nexus-base-url"
-)
-def crz(obj, account_id, date_range, nexus_base_url):
- if date_range is not None and len(date_range) == 2:
- req = dict(dateRange=dict(start=date_range[0], end=date_range[1]))
- else:
- req = dict()
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendCRZ".format(account_id))
- resp = post(url, json=req)
print(resp.content.decode("utf-8"))
-@taler.command(help="Trigger refunds for invalid incoming transactions")
+@bank_connection.command(help="import one bank account, chosen from the downloaded ones")
+@click.option("--connection-name", help="Connection ID", required=True)
+@click.option("--nexus-user-id", help="Nexus user ID", required=True)
+@click.option("--nexus-password", help="Nexus password", required=True)
+@click.option("--offered-account-id", help="Name of the account to import", required=True)
+@click.option("--nexus-bank-account-id", help="Name to give to the imported account", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--bank-account-id",
- help="Token that identifies one bank account belonging to --account-id",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def refund(ctx, account_id, bank_account_id, nexus_base_url):
- nexus_url = urljoin(
- nexus_base_url, "/ebics/taler/{}/accounts/{}/refund-invalid-payments".format(
- account_id, bank_account_id
+def import_bank_account(obj, connection_name, nexus_user_id, nexus_password, nexus_base_url, offered_account_id, nexus_bank_account_id):
+ url = urljoin(nexus_base_url, "/bank-connections/{}/import-account".format(connection_name))
+ try:
+ resp = post(
+ url,
+ json=dict(
+ offeredAccountId=offered_account_id,
+ nexusBankAccountId=nexus_bank_account_id
+ ),
+ auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password)
)
- )
- try:
- resp = post(nexus_url, json=body)
- except Exception:
- print("Could not reach the Nexus")
+ except Exception as ee:
+ print(ee)
+ print("Could not reach nexus")
return
print(resp.content.decode("utf-8"))
-@taler.command(help="Flag Taler-invalid incoming payments.")
+@bank_connection.command(help="download bank accounts in raw format WITHOUT importing them")
+@click.option("--connection-name", help="Connection ID", required=True)
+@click.option("--nexus-user-id", help="Nexus user ID", required=True)
+@click.option("--nexus-password", help="Nexus password", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def crunch_transactions(obj, account_id, nexus_base_url):
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/crunch-incoming-transactions".format(account_id))
- resp = post(url, json=dict())
- print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Show raw transactions from the Nexus database")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def show_collected_c53(obj, account_id, nexus_base_url):
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/show-collected-transactions-c53".format(account_id))
- resp = get(url)
- print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Send C53 message AND instruct the Nexus to persist the result")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--date-range",
- help="Date range for the query",
- nargs=2,
- required=False,
-)
-@click.argument(
- "nexus-base-url"
-)
-def collect_c53(obj, account_id, date_range, nexus_base_url):
- if date_range is not None and len(date_range) == 2:
- req = dict(dateRange=dict(start=date_range[0], end=date_range[1]))
- else:
- req = dict()
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/collect-transactions-c53".format(account_id))
- resp = post(url, json=req)
- print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Send C53 message")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--date-range",
- help="Date range for the query",
- nargs=2,
- required=False,
-)
-@click.argument(
- "nexus-base-url"
-)
-def c53(obj, account_id, date_range, nexus_base_url):
- if date_range is not None and len(date_range) == 2:
- req = dict(dateRange=dict(start=date_range[0], end=date_range[1]))
- else:
- req = dict()
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendC53".format(account_id))
- resp = post(url, json=req)
- print(resp.content.decode("utf-8"))
-
-
-@ebics.command(help="Send C54 message")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--date-range",
- help="Date range for the query",
- nargs=2,
- required=False,
-)
-@click.argument(
- "nexus-base-url"
-)
-def c54(obj, account_id, date_range, nexus_base_url):
- if date_range is not None and len(date_range) == 2:
- req = dict(dateRange=dict(start=date_range[0], end=date_range[1]))
- else:
- req = dict()
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendC54".format(account_id))
- resp = post(url, json=req)
- print(resp.content.decode("utf-8"))
-
-
-@ebics.command(help="Send PTK message")
-@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--date-range",
- help="Date range for the query",
- nargs=2,
- required=False,
-)
-@click.argument(
- "nexus-base-url"
-)
-def ptk(obj, account_id, date_range, nexus_base_url):
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendPTK".format(account_id))
- if date_range is not None and len(date_range) == 2:
- req = dict(dateRange=dict(start=date_range[0], end=date_range[1]))
- else:
- req = dict()
- print("requesting PTK", repr(req))
+def download_bank_accounts(obj, connection_name, nexus_user_id, nexus_password, nexus_base_url):
+ # FIXME/NOTE: the 'ebics' part will soon go away.
+ url = urljoin(nexus_base_url, "/bank-connections/{}/fetch-accounts".format(connection_name))
try:
- resp = post(url, json=req)
+ resp = post(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
except Exception:
- print("Could not reach the bank")
+ print("Could not reach nexus")
return
-
print(resp.content.decode("utf-8"))
-@ebics.command(help="Send HAC message")
+@bank_connection.command(help="list offered (= downloaded) bank accounts")
+@click.option("--connection-name", help="Connection ID", required=True)
+@click.option("--nexus-user-id", help="Nexus user ID", required=True)
+@click.option("--nexus-password", help="Nexus password", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--date-range",
- help="Date range for the query",
- nargs=2,
- required=False,
-)
-@click.argument(
- "nexus-base-url"
-)
-def hac(obj, account_id, date_range, nexus_base_url):
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendHAC".format(account_id))
- if date_range is not None and len(date_range) == 2:
- req = dict(dateRange=dict(start=date_range[0], end=date_range[1]))
- else:
- req = dict()
+def list_offered_bank_accounts(obj, connection_name, nexus_user_id, nexus_password, nexus_base_url):
+ # FIXME/NOTE: the 'ebics' part will soon go away.
+ url = urljoin(nexus_base_url, "/bank-connections/{}/accounts".format(connection_name))
try:
- resp = post(url, json=req)
+ resp = get(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
except Exception:
- print("Could not reach the bank")
+ print("Could not reach nexus")
return
-
print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Send INI message")
+@bank_accounts.command(help="list imported bank accounts")
+@click.option("--nexus-user-id", help="Nexus user ID", required=True)
+@click.option("--nexus-password", help="Nexus password", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def ini(obj, account_id, nexus_base_url):
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendIni".format(account_id))
- try:
- resp = post(url)
- except Exception:
- print("Could not reach the bank")
- return
-
- print(resp.content.decode("utf-8"))
-
-
-@ebics.command(help="Give and get keys.")
-@click.pass_context
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def prepare(ctx, account_id, nexus_base_url):
- ctx.invoke(ini, account_id=account_id, nexus_base_url=nexus_base_url)
- ctx.invoke(hia, account_id=account_id, nexus_base_url=nexus_base_url)
- ctx.invoke(sync, account_id=account_id, nexus_base_url=nexus_base_url)
-
-
-@ebics.command(help="Download PAIN.002 related to a particular customer")
-@click.pass_context
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--date-range",
- help="Date range for the query",
- nargs=2,
- required=False,
-)
-@click.argument(
- "nexus-base-url"
-)
-def fetch_payment_status(ctx, account_id, date_range, nexus_base_url):
- if date_range is not None and len(date_range) == 2:
- req = dict(dateRange=dict(start=date_range[0], end=date_range[1]))
- else:
- req = dict()
- try:
- url = urljoin(
- nexus_base_url, "/ebics/subscribers/{}/fetch-payment-status".format(account_id))
- resp = post(url, json=req)
- except Exception as e:
- print("Could not reach the Nexus", e)
- return
- print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Picks the first unsubmitted payment and send it to the bank via CCC")
-@click.pass_context
-@click.argument(
- "nexus-base-url"
-)
-def execute_payments_ccc(ctx, nexus_base_url):
- try:
- url = urljoin(nexus_base_url, "/ebics/admin/execute-payments-ccc")
- resp = post(url)
- except Exception as e:
- print("Could not reach the Nexus", e)
- return
-
- print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Picks the first unsubmitted payment and send it to the bank")
-@click.pass_context
-@click.argument(
- "nexus-base-url"
-)
-def execute_payments(ctx, nexus_base_url):
- try:
- url = urljoin(nexus_base_url, "/ebics/admin/execute-payments")
- resp = post(url)
- except Exception as e:
- print("Could not reach the Nexus", e)
- return
-
- print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Show status of payments")
-@click.pass_context
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def payment_status(
- ctx, account_id, nexus_base_url):
- nexus_url = urljoin(
- nexus_base_url, "/ebics/subscribers/{}/payments".format(account_id))
+def list_bank_accounts(obj, nexus_user_id, nexus_password, nexus_base_url):
+ url = urljoin(nexus_base_url, "/bank-accounts")
try:
- resp = get(nexus_url)
+ resp = get(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
except Exception:
- print("Could not reach the Nexus")
+ print("Could not reach nexus")
return
-
print(resp.content.decode("utf-8"))
-@ebics.command(help="Prepare a payment")
-@click.pass_context
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--bank-account-id",
- help="Token that identifies one bank account belonging to --account-id",
- required=True
-)
-@click.option(
- "--creditor-iban",
- help="IBAN of the creditor",
- required=True
-)
-@click.option(
- "--creditor-bic",
- help="BIC of the creditor",
- required=True
-)
-@click.option(
- "--creditor-name",
- help="First and last name of the creditor",
- required=True
-)
-@click.option(
- "--subject",
- help="subject line of the payment",
- required=True
-)
-@click.option(
- "--sum",
- help="amount to pay, in the form XY[.UW], no currency needed (for now)",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def prepare_payment(
- ctx, account_id, bank_account_id, creditor_iban,
- creditor_bic, creditor_name, subject, sum, nexus_base_url):
- nexus_url = urljoin(
- nexus_base_url, "/ebics/subscribers/{}/accounts/{}/prepare-payment".format(
- account_id, bank_account_id))
+@bank_accounts.command(help="prepare payment debiting 'account-name'")
+@click.option("--account-name", help="bank account name", required=True)
+@click.option("--credit-iban", help="IBAN that will receive the payment", required=True)
+@click.option("--credit-bic", help="BIC that will receive the payment", required=False)
+@click.option("--credit-name", help="Legal name that will receive the payment", required=True)
+@click.option("--payment-amount", help="Amount to be paid (<currency>:X.Y)", required=True)
+@click.option("--payment-subject", help="Subject of this payment", required=True)
+@click.option("--nexus-user-id", help="Nexus user ID", required=True)
+@click.option("--nexus-password", help="Nexus password", required=True)
+@click.argument("nexus-base-url")
+@click.pass_obj
+def prepare_payment(obj, account_name, credit_iban, credit_bic, credit_name,
+ nexus_user_id, nexus_password, nexus_base_url, payment_amount, payment_subject):
+ url = urljoin(nexus_base_url, "/bank-accounts/{}/prepared-payments".format(account_name))
body = dict(
- debtorAccount = bank_account_id,
- creditorIban = creditor_iban,
- creditorBic = creditor_bic,
- creditorName = creditor_name,
- subject = subject,
- sum = sum
+ iban=credit_iban,
+ bic=credit_bic,
+ name=credit_name,
+ subject=payment_subject,
+ amount=payment_amount
)
- try:
- resp = post(nexus_url, json=body)
- except Exception:
- print("Could not reach the Nexus")
- return
- print(resp.content.decode("utf-8"))
-
-@ebics.command(
- help="Trigger the Nexus to download and store \
- bank accounts information (via a HTD message)"
-)
-@click.pass_context
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.option(
- "--prepare/--no-prepare",
- help="Gets keying done before requesting HTD",
- required=False,
- default=False)
-@click.argument(
- "nexus-base-url"
-)
-def fetch_accounts(ctx, account_id, prepare, nexus_base_url):
- if prepare:
- ctx.invoke(ini)
- ctx.invoke(hia)
- ctx.invoke(sync)
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/fetch-accounts".format(account_id))
try:
- resp = post(url, json=dict())
+ resp = post(url, json=body, auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
except Exception:
- print("Could not reach the Nexus")
+ print("Could not reach nexus")
return
-
print(resp.content.decode("utf-8"))
-@ebics.command(help="Send HTD message")
+@bank_accounts.command(help="submit a prepared payment")
+@click.option("--account-name", help="bank account name", required=True)
+@click.option("--payment-uuid", help="payment unique identifier", required=True)
+@click.option("--nexus-user-id", help="nexus user id", required=True)
+@click.option("--nexus-password", help="nexus user password", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def htd(obj, account_id, nexus_base_url):
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendHTD".format(account_id))
+def submit_payment(obj, account_name, payment_uuid, nexus_user_id, nexus_password, nexus_base_url):
+ url = urljoin(
+ nexus_base_url, "/bank-accounts/{}/prepared-payments/{}/submit".format(account_name, payment_uuid)
+ )
try:
- resp = get(url)
+ resp = post(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
except Exception:
- print("Could not reach the Nexus")
+ print("Could not reach nexus")
return
print(resp.content.decode("utf-8"))
-
-
-@ebics.command(help="Send HKD message")
+@bank_accounts.command(help="fetch transactions from the bank")
+@click.option("--account-name", help="bank account name", required=True)
+@click.option("--nexus-user-id", help="nexus user id", required=True)
+@click.option("--nexus-password", help="nexus user password", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def hkd(obj, account_id, nexus_base_url):
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendHKD".format(account_id))
+def fetch_transactions(obj, account_name, nexus_user_id, nexus_password, nexus_base_url):
+ url = urljoin(
+ nexus_base_url, "/bank-accounts/{}/fetch-transactions".format(account_name)
+ )
try:
- resp = get(url)
+ resp = post(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
except Exception:
- print("Could not reach the Nexus")
+ print("Could not reach nexus")
return
print(resp.content.decode("utf-8"))
-@ebics.command(help="Send HIA message")
+@bank_accounts.command(help="get transactions from the simplified nexus JSON API")
+@click.option("--account-name", help="bank account name", required=True)
+@click.option("--nexus-user-id", help="nexus user id", required=True)
+@click.option("--nexus-password", help="nexus user password", required=True)
+@click.argument("nexus-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def hia(obj, account_id, nexus_base_url):
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sendHia".format(account_id))
+def transactions(obj, account_name, nexus_user_id, nexus_password, nexus_base_url):
+ url = urljoin(nexus_base_url, "/bank-accounts/{}/transactions".format(account_name))
try:
- resp = post(url)
+ resp = get(url, auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
except Exception:
- print("Could not reach the Nexus")
+ print("Could not reach nexus")
return
print(resp.content.decode("utf-8"))
-@ebics.command(help="Send HPB message")
+@sandbox.command(help="activate a Ebics host")
+@click.option("--host-id", help="Ebics host ID", required=True)
+@click.argument("sandbox-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Numerical ID of the customer at the Nexus",
- required=True
-)
-@click.argument(
- "nexus-base-url"
-)
-def sync(obj, account_id, nexus_base_url):
-
- url = urljoin(nexus_base_url, "/ebics/subscribers/{}/sync".format(account_id))
+def make_ebics_host(obj, host_id, sandbox_base_url):
+ url = urljoin(sandbox_base_url, "/admin/ebics/host")
try:
- resp = post(url)
+ resp = post(url, json=dict(hostID=host_id, ebicsVersion="2.5"))
except Exception:
- print("Could not reach the Nexus")
+ print("Could not reach sandbox")
return
-
print(resp.content.decode("utf-8"))
-
-@ebics.command(help="Retrieve all the customers managed by Nexus")
+@sandbox.command(help="activate a Ebics subscriber")
+@click.option("--host-id", help="Ebics host ID", required=True)
+@click.option("--partner-id", help="Ebics partner ID", required=True)
+@click.option("--user-id", help="Ebics user ID", required=True)
+@click.argument("sandbox-base-url")
@click.pass_obj
-@click.argument(
- "nexus-base-url"
-)
-def subscribers(obj, nexus_base_url):
-
- url = urljoin(nexus_base_url, "/ebics/subscribers")
+def activate_ebics_subscriber(obj, host_id, partner_id, user_id, sandbox_base_url):
+ url = urljoin(sandbox_base_url, "/admin/ebics/subscribers")
try:
- resp = get(url)
+ resp = post(url, json=dict(hostID=host_id, partnerID=partner_id, userID=user_id))
except Exception:
- print("Could not reach Nexus at {}".format(url))
+ print("Could not reach sandbox")
return
-
print(resp.content.decode("utf-8"))
-@ebics.command(help="Activate a new subscriber into Nexus")
+@sandbox.command(help="associate a bank account to a Ebics subscriber")
+@click.option("--iban", help="IBAN", required=True)
+@click.option("--bic", help="BIC", required=True)
+@click.option("--person-name", help="bank account owner name", required=True)
+@click.option("--account-name", help="label of this bank account", required=True)
+@click.option("--ebics-user-id", help="user ID of the Ebics subscriber", required=True)
+@click.option("--ebics-host-id", help="host ID of the Ebics subscriber", required=True)
+@click.option("--ebics-partner-id", help="partner ID of the Ebics subscriber", required=True)
+@click.argument("sandbox-base-url")
@click.pass_obj
-@click.option(
- "--account-id",
- help="Alphanumeric ID (at the Nexus) of the new customer",
- required=True
-)
-@click.option(
- "--ebics-url",
- help="URL of the EBICS server",
- required=True
-)
-@click.option(
- "--user-id",
- help="ID of the user to add in the system",
- required=True
-)
-@click.option(
- "--partner-id",
- help="ID of the partner associated with the user" ,
- required=True
-)
-@click.option(
- "--system-id",
- help="ID of the software acting on behalf of this user" ,
- required=False
-)
-@click.option(
- "--host-id",
- help="ID of the EBICS server" ,
- required=True
-)
-@click.option(
- "--password",
- help="password to associate to the subscriber being created" ,
- required=False,
- default=None
-)
-@click.argument(
- "nexus-base-url"
-)
-def new_subscriber(obj, account_id, user_id, partner_id, system_id, host_id, ebics_url, password, nexus_base_url):
- nexus_url = urljoin(nexus_base_url, "/ebics/{}/subscribers".format(account_id))
+def associate_bank_account(obj, iban, bic, person_name, account_name,
+ ebics_user_id, ebics_host_id, ebics_partner_id, sandbox_base_url):
+ url = urljoin(sandbox_base_url, "/admin/ebics/bank-accounts")
body = dict(
- ebicsURL=ebics_url,
- userID=user_id,
- partnerID=partner_id,
- hostID=host_id,
- password=password
+ subscriber=dict(userID=ebics_user_id, partnerID=ebics_partner_id, hostID=ebics_host_id),
+ iban=iban, bic=bic, name=person_name, label=account_name
)
- if system_id:
- body.update(system_id)
- try:
- resp = post(nexus_url, json=body)
- except Exception:
- print("Could not reach the Nexus")
- return
- print(resp.content.decode("utf-8"))
-
-
-@native.command(help="Ask the list of transactions related to one account")
-@click.option(
- "--user-id",
- help="ID of the bank customer (no EBICS correlation implied/needed)" ,
- required=True
-)
-@click.option(
- "--start",
- help="Starting date for history elements (YYYY-MM-DD)" ,
- required=False,
- default=None
-)
-@click.option(
- "--end",
- help="Ending date for history elements (YYYY-MM-DD)" ,
- required=False,
- default=None
-)
-@click.argument(
- "bank-base-url"
-)
-def history(user_id, start, end, bank_base_url):
-
- url = urljoin(bank_base_url, f"/{user_id}/history")
- print(url)
try:
- resp = post(url, json=dict(start=start, end=end))
+ resp = post(url, json=body)
except Exception:
- print("Could not reach the bank")
+ print("Could not reach sandbox")
return
-
print(resp.content.decode("utf-8"))
-
-@native.command(help="Ask the balance for a given customer of the bank")
-@click.option(
- "--user-id",
- help="ID of the bank customer (no EBICS correlation implied/needed)",
- required=True
-)
-@click.argument(
- "bank-base-url"
-)
+@sandbox.command(help="book a payment in the sandbox")
+@click.option("--creditor-iban", help="IBAN receiving the payment")
+@click.option("--creditor-bic", help="BIC receiving the payment")
+@click.option("--creditor-name", help="Name of the person who is receiving the payment")
+@click.option("--debtor-iban", help="IBAN sending the payment")
+@click.option("--debtor-bic", help="BIC sending the payment")
+@click.option("--debtor-name", help="name of the person who is sending the payment")
+@click.option("--amount", help="amount, no currency")
+@click.option("--currency", help="currency")
+@click.option("--subject", help="payment subject")
+@click.argument("sandbox-base-url")
@click.pass_obj
-def balance(obj, user_id, bank_base_url):
-
- url = urljoin(bank_base_url, f"/{user_id}/balance")
- print(url)
+def book_payment(obj, creditor_iban, creditor_bic, creditor_name, debtor_iban,
+ debtor_bic, debtor_name, amount, currency, subject, sandbox_base_url):
+
+ url = urljoin(sandbox_base_url, "/admin/payments")
+ body = dict(
+ creditorIban=creditor_iban,
+ creditorBic=creditor_bic,
+ creditorName=creditor_name,
+ debitorIban=debtor_iban,
+ debitorBic=debtor_bic,
+ debitorName=debtor_name,
+ amount=amount,
+ currency=currency,
+ subject=subject
+ )
try:
- resp = get(url)
+ resp = post(url, json=body)
except Exception:
- print("Could not reach the bank")
+ print("Could not reach sandbox")
return
-
print(resp.content.decode("utf-8"))
cli()
diff --git a/cli/libeufin-cli-new b/cli/libeufin-cli-new
@@ -1,381 +0,0 @@
-#!/usr/bin/env python3
-
-import click
-import json
-import hashlib
-import errno
-from datetime import datetime
-from requests import post, get, auth
-from urllib.parse import urljoin
-from getpass import getpass
-
-@click.group(help="""
-General utility to invoke HTTP REST services offered by Nexus.
-Consider also invoking the 'nexus' command directly, for example
-to create new users.
-"""
-)
-def cli():
- pass
-
-@cli.group()
-@click.pass_context
-def bank_connection(ctx):
- pass
-
-@cli.group()
-@click.pass_context
-def bank_accounts(ctx):
- pass
-
-@cli.group()
-@click.pass_context
-def sandbox(ctx):
- pass
-
-@bank_connection.command(help="export backup")
-@click.option("--connection-name", help="Name of the bank connection to backup", required=True)
-@click.option("--nexus-user-id", help="Nexus user ID", required=True)
-@click.option("--nexus-password", help="Nexus password", required=True)
-@click.option("--passphrase", help="Passphrase for locking the backup", required=True)
-@click.option("--output-file", help="Where to store the backup", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def export_backup(obj, connection_name, nexus_user_id, nexus_password, passphrase, output_file, nexus_base_url):
- url = urljoin(nexus_base_url, "/bank-connections/{}/export-backup".format(connection_name))
- try:
- resp = post(
- url, json=dict(passphrase=passphrase),
- auth=auth.HTTPBasicAuth(nexus_user_id, nexus_password)
- )
- except Exception:
- print("Could not reach nexus")
- exit(1)
-
- output = open(output_file, "w+")
- output.write(resp.content.decode("utf-8"))
- output.close()
-
- print("Backup stored in {}".format(output_file))
-
-
-@bank_connection.command(help="restore backup")
-@click.option("--connection-name", help="Name of the bank connection to backup", required=True)
-@click.option("--nexus-user-id", help="Nexus user ID", required=True)
-@click.option("--nexus-password", help="Nexus password", required=True)
-@click.option("--backup-file", help="Back file", required=True)
-@click.option("--passphrase", help="Passphrase for locking the backup", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def restore_backup(obj, backup_file, passphrase, nexus_base_url, nexus_user_id, nexus_password, connection_name):
- url = urljoin(nexus_base_url, "/bank-connections")
- try:
- backup = open(backup_file, "r")
- except Exception:
- print("Could not open the backup at {}".format(backup_file))
- return
-
- backup_json = json.loads(backup.read())
- backup.close()
-
- try:
- resp = post(
- url,
- json=dict(
- name=connection_name,
- data=backup_json,
- passphrase=passphrase,
- source="backup"
- ),
- auth=auth.HTTPBasicAuth(nexus_user_id, nexus_password)
-
- )
- except Exception:
- print("Could not reach nexus")
- exit(1)
-
- print(resp.content.decode("utf-8"))
-
-
-@bank_connection.command(help="make new Ebics bank connection")
-@click.option("--connection-name", help="Connection ID", required=True)
-@click.option("--ebics-url", help="EBICS URL", required=True)
-@click.option("--host-id", help="Host ID", required=True)
-@click.option("--partner-id", help="Partner ID", required=True)
-@click.option("--ebics-user-id", help="Ebics user ID", required=True)
-@click.option("--nexus-user-id", help="Nexus user ID", required=True)
-@click.option("--nexus-password", help="Nexus password", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def new_ebics_connection(obj, connection_name, ebics_url, host_id, partner_id,
- nexus_user_id, nexus_password, nexus_base_url, ebics_user_id):
- url = urljoin(nexus_base_url, "/bank-connections")
- body = dict(
- name=connection_name,
- source="new",
- type="ebics",
- data=dict(ebicsURL=ebics_url, hostID=host_id, partnerID=partner_id, userID=ebics_user_id)
- )
- try:
- resp = post(url, json=body, auth=auth.HTTPBasicAuth(nexus_user_id, nexus_password))
- except Exception:
- print("Could not reach nexus")
- exit(1)
- print(resp.content.decode("utf-8"))
-
-@bank_connection.command(help="bootstrap the bank connection")
-@click.option("--connection-name", help="Connection ID", required=True)
-@click.option("--nexus-user-id", help="Nexus user ID", required=True)
-@click.option("--nexus-password", help="Nexus password", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def bootstrap_bank_connection(obj, connection_name, nexus_user_id, nexus_password, nexus_base_url):
- url = urljoin(nexus_base_url, "/bank-connections/{}/connect".format(connection_name))
- try:
- resp = post(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
- except Exception:
- print("Could not reach nexus")
- return
- print(resp.content.decode("utf-8"))
-
-@bank_connection.command(help="import one bank account, chosen from the downloaded ones")
-@click.option("--connection-name", help="Connection ID", required=True)
-@click.option("--nexus-user-id", help="Nexus user ID", required=True)
-@click.option("--nexus-password", help="Nexus password", required=True)
-@click.option("--offered-account-id", help="Name of the account to import", required=True)
-@click.option("--nexus-bank-account-id", help="Name to give to the imported account", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def import_bank_account(obj, connection_name, nexus_user_id, nexus_password, nexus_base_url, offered_account_id, nexus_bank_account_id):
- url = urljoin(nexus_base_url, "/bank-connections/{}/import-account".format(connection_name))
- try:
- resp = post(
- url,
- json=dict(
- offeredAccountId=offered_account_id,
- nexusBankAccountId=nexus_bank_account_id
- ),
- auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password)
- )
- except Exception as ee:
- print(ee)
- print("Could not reach nexus")
- return
- print(resp.content.decode("utf-8"))
-
-@bank_connection.command(help="download bank accounts in raw format WITHOUT importing them")
-@click.option("--connection-name", help="Connection ID", required=True)
-@click.option("--nexus-user-id", help="Nexus user ID", required=True)
-@click.option("--nexus-password", help="Nexus password", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def download_bank_accounts(obj, connection_name, nexus_user_id, nexus_password, nexus_base_url):
- # FIXME/NOTE: the 'ebics' part will soon go away.
- url = urljoin(nexus_base_url, "/bank-connections/{}/fetch-accounts".format(connection_name))
- try:
- resp = post(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
- except Exception:
- print("Could not reach nexus")
- return
- print(resp.content.decode("utf-8"))
-
-
-@bank_connection.command(help="list offered (= downloaded) bank accounts")
-@click.option("--connection-name", help="Connection ID", required=True)
-@click.option("--nexus-user-id", help="Nexus user ID", required=True)
-@click.option("--nexus-password", help="Nexus password", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def list_offered_bank_accounts(obj, connection_name, nexus_user_id, nexus_password, nexus_base_url):
- # FIXME/NOTE: the 'ebics' part will soon go away.
- url = urljoin(nexus_base_url, "/bank-connections/{}/accounts".format(connection_name))
- try:
- resp = get(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
- except Exception:
- print("Could not reach nexus")
- return
- print(resp.content.decode("utf-8"))
-
-@bank_accounts.command(help="list imported bank accounts")
-@click.option("--nexus-user-id", help="Nexus user ID", required=True)
-@click.option("--nexus-password", help="Nexus password", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def list_bank_accounts(obj, nexus_user_id, nexus_password, nexus_base_url):
- url = urljoin(nexus_base_url, "/bank-accounts")
- try:
- resp = get(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
- except Exception:
- print("Could not reach nexus")
- return
- print(resp.content.decode("utf-8"))
-
-@bank_accounts.command(help="prepare payment debiting 'account-name'")
-@click.option("--account-name", help="bank account name", required=True)
-@click.option("--credit-iban", help="IBAN that will receive the payment", required=True)
-@click.option("--credit-bic", help="BIC that will receive the payment", required=False)
-@click.option("--credit-name", help="Legal name that will receive the payment", required=True)
-@click.option("--payment-amount", help="Amount to be paid (<currency>:X.Y)", required=True)
-@click.option("--payment-subject", help="Subject of this payment", required=True)
-@click.option("--nexus-user-id", help="Nexus user ID", required=True)
-@click.option("--nexus-password", help="Nexus password", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def prepare_payment(obj, account_name, credit_iban, credit_bic, credit_name,
- nexus_user_id, nexus_password, nexus_base_url, payment_amount, payment_subject):
- url = urljoin(nexus_base_url, "/bank-accounts/{}/prepared-payments".format(account_name))
- body = dict(
- iban=credit_iban,
- bic=credit_bic,
- name=credit_name,
- subject=payment_subject,
- amount=payment_amount
- )
-
- try:
- resp = post(url, json=body, auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
- except Exception:
- print("Could not reach nexus")
- return
- print(resp.content.decode("utf-8"))
-
-
-@bank_accounts.command(help="submit a prepared payment")
-@click.option("--account-name", help="bank account name", required=True)
-@click.option("--payment-uuid", help="payment unique identifier", required=True)
-@click.option("--nexus-user-id", help="nexus user id", required=True)
-@click.option("--nexus-password", help="nexus user password", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def submit_payment(obj, account_name, payment_uuid, nexus_user_id, nexus_password, nexus_base_url):
- url = urljoin(
- nexus_base_url, "/bank-accounts/{}/prepared-payments/{}/submit".format(account_name, payment_uuid)
- )
- try:
- resp = post(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
- except Exception:
- print("Could not reach nexus")
- return
- print(resp.content.decode("utf-8"))
-
-@bank_accounts.command(help="fetch transactions from the bank")
-@click.option("--account-name", help="bank account name", required=True)
-@click.option("--nexus-user-id", help="nexus user id", required=True)
-@click.option("--nexus-password", help="nexus user password", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def fetch_transactions(obj, account_name, nexus_user_id, nexus_password, nexus_base_url):
- url = urljoin(
- nexus_base_url, "/bank-accounts/{}/fetch-transactions".format(account_name)
- )
- try:
- resp = post(url, json=dict(), auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
- except Exception:
- print("Could not reach nexus")
- return
- print(resp.content.decode("utf-8"))
-
-@bank_accounts.command(help="get transactions from the simplified nexus JSON API")
-@click.option("--account-name", help="bank account name", required=True)
-@click.option("--nexus-user-id", help="nexus user id", required=True)
-@click.option("--nexus-password", help="nexus user password", required=True)
-@click.argument("nexus-base-url")
-@click.pass_obj
-def transactions(obj, account_name, nexus_user_id, nexus_password, nexus_base_url):
- url = urljoin(nexus_base_url, "/bank-accounts/{}/transactions".format(account_name))
- try:
- resp = get(url, auth = auth.HTTPBasicAuth(nexus_user_id, nexus_password))
- except Exception:
- print("Could not reach nexus")
- return
- print(resp.content.decode("utf-8"))
-
-
-@sandbox.command(help="activate a Ebics host")
-@click.option("--host-id", help="Ebics host ID", required=True)
-@click.argument("sandbox-base-url")
-@click.pass_obj
-def make_ebics_host(obj, host_id, sandbox_base_url):
- url = urljoin(sandbox_base_url, "/admin/ebics/host")
- try:
- resp = post(url, json=dict(hostID=host_id, ebicsVersion="2.5"))
- except Exception:
- print("Could not reach sandbox")
- return
- print(resp.content.decode("utf-8"))
-
-@sandbox.command(help="activate a Ebics subscriber")
-@click.option("--host-id", help="Ebics host ID", required=True)
-@click.option("--partner-id", help="Ebics partner ID", required=True)
-@click.option("--user-id", help="Ebics user ID", required=True)
-@click.argument("sandbox-base-url")
-@click.pass_obj
-def activate_ebics_subscriber(obj, host_id, partner_id, user_id, sandbox_base_url):
- url = urljoin(sandbox_base_url, "/admin/ebics/subscribers")
- try:
- resp = post(url, json=dict(hostID=host_id, partnerID=partner_id, userID=user_id))
- except Exception:
- print("Could not reach sandbox")
- return
- print(resp.content.decode("utf-8"))
-
-@sandbox.command(help="associate a bank account to a Ebics subscriber")
-@click.option("--iban", help="IBAN", required=True)
-@click.option("--bic", help="BIC", required=True)
-@click.option("--person-name", help="bank account owner name", required=True)
-@click.option("--account-name", help="label of this bank account", required=True)
-@click.option("--ebics-user-id", help="user ID of the Ebics subscriber", required=True)
-@click.option("--ebics-host-id", help="host ID of the Ebics subscriber", required=True)
-@click.option("--ebics-partner-id", help="partner ID of the Ebics subscriber", required=True)
-@click.argument("sandbox-base-url")
-@click.pass_obj
-def associate_bank_account(obj, iban, bic, person_name, account_name,
- ebics_user_id, ebics_host_id, ebics_partner_id, sandbox_base_url):
- url = urljoin(sandbox_base_url, "/admin/ebics/bank-accounts")
- body = dict(
- subscriber=dict(userID=ebics_user_id, partnerID=ebics_partner_id, hostID=ebics_host_id),
- iban=iban, bic=bic, name=person_name, label=account_name
- )
-
- try:
- resp = post(url, json=body)
- except Exception:
- print("Could not reach sandbox")
- return
- print(resp.content.decode("utf-8"))
-
-@sandbox.command(help="book a payment in the sandbox")
-@click.option("--creditor-iban", help="IBAN receiving the payment")
-@click.option("--creditor-bic", help="BIC receiving the payment")
-@click.option("--creditor-name", help="Name of the person who is receiving the payment")
-@click.option("--debtor-iban", help="IBAN sending the payment")
-@click.option("--debtor-bic", help="BIC sending the payment")
-@click.option("--debtor-name", help="name of the person who is sending the payment")
-@click.option("--amount", help="amount, no currency")
-@click.option("--currency", help="currency")
-@click.option("--subject", help="payment subject")
-@click.argument("sandbox-base-url")
-@click.pass_obj
-def book_payment(obj, creditor_iban, creditor_bic, creditor_name, debtor_iban,
- debtor_bic, debtor_name, amount, currency, subject, sandbox_base_url):
-
- url = urljoin(sandbox_base_url, "/admin/payments")
- body = dict(
- creditorIban=creditor_iban,
- creditorBic=creditor_bic,
- creditorName=creditor_name,
- debitorIban=debtor_iban,
- debitorBic=debtor_bic,
- debitorName=debtor_name,
- amount=amount,
- currency=currency,
- subject=subject
- )
- try:
- resp = post(url, json=body)
- except Exception:
- print("Could not reach sandbox")
- return
- print(resp.content.decode("utf-8"))
-
-cli()
