path: root/regional-currency/ask_questions.py

#!/usr/bin/env python3
"""Python script to ask questions using an interactive prompt"""

import base64
import os
import subprocess
import uuid
from typing import Callable, Dict, TypeVar

log = open("setup.log", "ab", buffering=0)
CONFIG_FILE = "config/user.conf"


def load_conf() -> Dict[str, str]:
    """Load user configuration file"""
    conf = {}
    with open(CONFIG_FILE, "r") as f:
        for kv in f.read().splitlines():
            if len(kv) != 0:
                [k, v] = [part.strip() for part in kv.split("=", 1)]
                conf[k] = v.strip('"').replace('\\"', '"')
    return conf


conf = load_conf()


def add_conf(name: str, value: str):
    """Update a user configuration value and update the configuration file"""
    conf[name] = value
    content = ""
    for key, value in conf.items():
        escaped = value.replace('"', '\\"')
        content += f'{key}="{escaped}"\n'
    with open(CONFIG_FILE, "w") as f:
        f.write(content)


def run_cmd(
    cmd: list[str], input: str | None = None, env: Dict[str, str] | None = None
) -> int:
    """Run a command in a child process and return its exit code"""
    result = subprocess.run(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        input=input.encode() if input is not None else None,
        stdin=subprocess.DEVNULL if input is None else None,
        env=env,
    )
    log.write(result.stdout)
    if result.returncode != 0:
        print(result.stdout.decode("utf-8"), end="")
    return result.returncode


def try_cmd(
    cmd: list[str], input: str | None = None, env: Dict[str, str] | None = None
) -> bool:
    """Run a command in a child process and return if successful"""
    return run_cmd(cmd, input, env) == 0


A = TypeVar("A")
T = TypeVar("T")


def conf_value(
    name: str | None,
    action: Callable[[], str | None],
    default: T | None = None,
    check: Callable[[str], T | None] = lambda it: it,
    fmt: Callable[[T], str] = lambda it: str(it),
) -> T:
    """
    Logic to configure a value

    :param name: if present will try to fetch the current value and will store the new value
    :param action: how a value will be obtained
    :param default: default value to use if no value is given
    :param check: check and normalize the value
    :param fmt: format value for storage
    :return: the configuration value
    """
    value = None

    # Fetch current value
    if name is not None:
        curr = conf.get(name, None)
        if curr is not None:
            # Check the current value and ask again if invalid
            value = check(curr)

    # Ask for a new value until we get a valid one
    while value is None:
        new = action()
        # Use default if no value was provided else check the new value
        value = check(new) if new is not None else default

    # Store the new value
    if name is not None:
        add_conf(name, fmt(value))

    return value


def ask(
    name: str | None,
    msg: str,
    default: T | None = None,
    check: Callable[[str], T | None] = lambda it: it,
    fmt: Callable[[T], str] = lambda it: str(it),
) -> T:
    """
    Prompt the user to configurea value
    :param name: if present will try to fetch the current value and will store the new value
    :param msg: the message to prompt the user with
    :param default: default value to use if no value is obtained
    :param check: check and normalize the value
    :param fmt: format value for storage
    :return: the configuration value
    """

    def do_ask() -> str | None:
        # Log the prompt
        log.write(msg.encode() + "\n".encode())
        # Actual prompt
        raw = input(msg).strip()
        if raw == "":
            if default is None:
                print("You must enter a value")
            return None
        return raw

    return conf_value(name, do_ask, default, check, fmt)


def ask_str(name: str | None, msg: str, default: str | None = None) -> str:
    "Prompt the user to configure a string"
    return ask(name, msg, default)


def ask_currency(name: str, msg: str, default: str | None = None) -> str:
    "Prompt the user to configure a currency name"

    def check_currency(currency: str) -> str | None:
        currency = currency.upper()
        if not currency.isascii() or not currency.isalpha():
            print("The currency name must be an ASCII alphabetic string")
        elif len(currency) < 3 or 11 < len(currency):
            print("The currency name had to be between 3 and 11 characters long")
        else:
            return currency
        return None

    return ask(name, msg, default, check_currency)


def ask_host(name: str, msg: str, default: str | None = None) -> str:
    "Prompt the user to configure the installation hostname"

    def check_host(host: str) -> str | None:
        success = True
        for subdomain in ["backend", "bank", "exchange"]:
            success = try_cmd(["ping", "-c", "1", f"{subdomain}.{host}"]) and success
        if success:
            return host
        else:
            return None

    return ask(name, msg, default, check_host)


def ask_yes_no(name: str | None, msg: str, default: bool | None = None) -> bool:
    "Prompt the user to configure a boolean"

    def check_yes_no(raw: str) -> bool | None:
        raw = raw.lower()
        if raw == "y" or raw == "yes":
            return True
        elif raw == "n" or raw == "no":
            return False
        else:
            print("Expected 'y' or 'n'")
            return None

    return ask(name, msg, default, check_yes_no, lambda it: "y" if it else "n")


ask_currency(
    "CURRENCY",
    "1. Enter the name of the regional currency (e.g. 'NETZBON'): ",
    "NETZBON",
)
ask_currency(
    "FIAT_CURRENCY", "2. Enter the name of the fiat currency (e.g. 'CHF'): ", "CHF"
)
ask_str(
    "BANK_NAME",
    "3. Enter the human-readable name of the bank (e.g. 'Taler Bank'): ",
    "Taler Bank",
)
ask_host("DOMAIN_NAME", "4. Enter the domain name (e.g. 'example.com'): ")
if ask_yes_no("ENABLE_TLS", "5. Setup TLS using Let's Encrypt? (Y/n): ", True):
    ask_str("TLS_EMAIL", "5.1. Enter an email address for Let's Encrypt: ")

    def ask_tos():
        print(
            "5.2. Please read the Terms of Service at https://letsencrypt.org/documents/LE-SA-v1.3-September-21-2022.pdf."
        )
        if not ask_yes_no(
            None,
            "5.2. You must agree in order to register with the ACME server. Do you agree? (y/n): ",
            False,
        ):
            print("You must agree in order to register with the ACME server")
            return None
        else:
            return "y"

    conf_value("TLS_TOS", ask_tos)
    add_conf("PROTO", "https")
else:
    add_conf("PROTO", "http")

if not ask_yes_no(
    "DO_OFFLINE",
    "6. Do you want Taler Exchange keys on this server (Y) or externally on another server (n): ",
    True,
):
    ask_str("MASTER_PUBLIC_KEY", "6.1. Enter the exchange-offline master public key: ")

if ask_yes_no(
    "DO_TELESIGN",
    "7. Setup SMS two-factor authentication using Telesign https://www.telesign.com? (Y/n): ",
    True,
):

    def ask_telesign():
        customer_id = ask_str(None, "7.1. Enter your Telesign Customer ID: ")
        api_key = ask_str(None, "7.2. Enter your Telesign API Key: ")
        phone_number = ask_str(
            None,
            "7.3. Enter a phone number to test your API key (e.g. '+447911123456'): ",
        )
        auth_token = base64.b64encode(f"{customer_id}:{api_key}".encode()).decode()
        if not try_cmd(
            ["libeufin-tan-sms.sh", phone_number],
            "12345 is your verification code for GNU Taler setup",
            {**os.environ, "AUTH_TOKEN": auth_token},
        ):
            print(
                "Failed to send an SMS using Telesign API, check your credentials and phone number"
            )
            return None
        code = ask_str(None, f"7.4. Enter the code received by {phone_number} : ")
        if code != "12345":
            print(
                f"Wrong code got '{code}' expected '12345', check your credentials and phone number"
            )
            return None
        return auth_token

    conf_value("TELESIGN_AUTH_TOKEN", ask_telesign)
ask_str(
    "BANK_ADMIN_PASSWORD",
    "8. Enter the admin password for the bank (or press enter to autogenerate password): ",
    str(uuid.uuid4()),
)

def ask_tos_file(name: str, msg: str) -> str:
    "Prompt the user to select a ToS/privacy policy"

    # msg = "9.1. Enter the filename of the ToS. Some available options are:\n"
    tos_msg = msg

    # Recollect example ToS files
    tos_path = "/usr/share/taler/terms"
    for f in os.listdir(tos_path):
        tos_file = os.path.join(tos_path, f)
        if os.path.isfile(tos_file) and f.endswith(".rst"):
            tos_msg += f"- {tos_file}\n"

    tos_msg += "=> "

    return ask_str(
        name,
        tos_msg,
        os.path.join(tos_path, "exchange-tos-v0.rst")
    )

if ask_yes_no(
    "DO_EXCHANGE_TERMS",
    "9. Do you wish to configure terms of service (ToS) for the exchange? (Y/n): ",
    False,
):
    ask_tos_file(
        "EXCHANGE_TERMS_FILE",
        "9.1. Enter the filename of the ToS. Some available options are:\n"
    )

if ask_yes_no(
    "DO_EXCHANGE_PRIVACY",
    "10. Do you wish to configure a privacy policy for the exchange? (Y/n): ",
    False,
):
    ask_tos_file(
        "EXCHANGE_PRIVACY_FILE",
        "10.1. Enter the filename of the privacy policy. Some available options are:\n"
    )