1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
|
#!/usr/bin/env python3
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]:
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):
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:
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:
return run_cmd(cmd, input, env) == 0
A = TypeVar("A")
T = TypeVar("T")
def custom(
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),
):
if name is not None:
prev = conf.get(name, None)
if prev is not None: # If got previous value
checked = check(prev)
if checked is not None: # And previous value is valid
add_conf(name, fmt(checked)) # Add again in case normalization changed
return checked
while True:
raw = action()
if raw is None:
if default is not None:
if name is not None:
add_conf(name, fmt(default))
return default
continue
else:
checked = check(raw)
if checked is not None:
if name is not None:
add_conf(name, fmt(checked))
return checked
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:
def do_ask():
log.write(msg.encode())
log.write("\n".encode())
raw = input(msg).strip()
if raw == "":
if default is None:
print("You must enter a value")
return None
return raw
return custom(name, do_ask, default, check, fmt)
def ask_str(name: str | None, msg: str, default: str | None = None) -> str:
return ask(name, msg, default)
def ask_currency(name: str, msg: str, default: str | None = None) -> str:
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:
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:
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"
custom("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",
{**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
custom("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()),
)
|
