inventory-wizard.py (17602B)
1 #!/usr/bin/env python3 2 """Interactively create an Ansible inventory entry for a regional currency.""" 3 4 import argparse 5 import getpass 6 import os 7 import re 8 import secrets 9 import tempfile 10 from pathlib import Path 11 12 try: 13 import yaml 14 except ImportError as exc: 15 raise SystemExit( 16 "PyYAML is required. Install ansible-core before running this wizard." 17 ) from exc 18 19 20 HOST_PATTERN = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.-]*") 21 CURRENCY_PATTERN = re.compile(r"[A-Z]{3,11}") 22 BIC_PATTERN = re.compile(r"[A-Z0-9]{4}[A-Z]{2}[A-Z0-9]{2}(?:[A-Z0-9]{3})?") 23 IBAN_PATTERN = re.compile(r"[A-Z]{2}[0-9]{2}[A-Z0-9]{1,28}") 24 25 26 def ask(message: str, default: str | None = None, *, secret: bool = False) -> str: 27 hint = f" [{default}]" if default is not None and not secret else "" 28 while True: 29 prompt = f"{message}{hint}: " 30 value = getpass.getpass(prompt) if secret else input(prompt) 31 value = value.strip() 32 if value: 33 return value 34 if default is not None: 35 return default 36 print("A value is required.") 37 38 39 def ask_matching( 40 message: str, 41 pattern: re.Pattern[str], 42 default: str | None = None, 43 *, 44 normalize=lambda value: value, 45 ) -> str: 46 while True: 47 value = normalize(ask(message, default)) 48 if pattern.fullmatch(value): 49 return value 50 print("The value has an invalid format.") 51 52 53 def ask_yes_no(message: str, default: bool) -> bool: 54 hint = "Y/n" if default else "y/N" 55 while True: 56 value = input(f"{message} [{hint}]: ").strip().lower() 57 if not value: 58 return default 59 if value in {"y", "yes"}: 60 return True 61 if value in {"n", "no"}: 62 return False 63 print("Please answer yes or no.") 64 65 66 def ask_int(message: str, default: int, minimum: int = 1, maximum: int = 65535) -> int: 67 while True: 68 raw = ask(message, str(default)) 69 try: 70 value = int(raw) 71 except ValueError: 72 value = 0 73 if minimum <= value <= maximum: 74 return value 75 print(f"Enter a number between {minimum} and {maximum}.") 76 77 78 def bool_value(value: object, default: bool) -> bool: 79 return value if isinstance(value, bool) else default 80 81 82 def string_value(value: object, default: str | None = None) -> str | None: 83 return str(value) if value is not None else default 84 85 86 def load_yaml(path: Path) -> dict: 87 if not path.exists(): 88 return {} 89 content = path.read_text(encoding="utf-8") 90 if content.startswith("$ANSIBLE_VAULT;"): 91 raise SystemExit( 92 f"{path} is encrypted. Decrypt it with ansible-vault before editing it." 93 ) 94 try: 95 data = yaml.safe_load(content) 96 except yaml.YAMLError as exc: 97 raise SystemExit(f"Cannot parse {path}: {exc}") from exc 98 if data is None: 99 return {} 100 if not isinstance(data, dict): 101 raise SystemExit(f"Expected a YAML mapping in {path}.") 102 return data 103 104 105 def atomic_write(path: Path, content: str, mode: int = 0o600) -> None: 106 """Durably replace path without exposing a partially written inventory.""" 107 descriptor, temporary_name = tempfile.mkstemp( 108 prefix=f".{path.name}.", dir=path.parent 109 ) 110 temporary_path = Path(temporary_name) 111 try: 112 os.fchmod(descriptor, mode) 113 with os.fdopen(descriptor, "w", encoding="utf-8") as stream: 114 descriptor = -1 115 stream.write(content) 116 stream.flush() 117 os.fsync(stream.fileno()) 118 os.replace(temporary_path, path) 119 directory_descriptor = os.open(path.parent, os.O_RDONLY) 120 try: 121 os.fsync(directory_descriptor) 122 finally: 123 os.close(directory_descriptor) 124 finally: 125 if descriptor >= 0: 126 os.close(descriptor) 127 temporary_path.unlink(missing_ok=True) 128 129 130 def inventory_hosts(inventory: dict, path: Path) -> dict: 131 try: 132 hosts = inventory["all"]["children"]["regional_currency"]["hosts"] 133 except (KeyError, TypeError) as exc: 134 raise SystemExit(f"{path} is not a regional-currency inventory file.") from exc 135 if not isinstance(hosts, dict): 136 raise SystemExit(f"The regional_currency hosts in {path} are not a mapping.") 137 return hosts 138 139 140 def discover_aliases(directory: Path) -> list[str]: 141 aliases = { 142 path.stem for path in directory.glob("*.yml") if path.name != "hosts.yml" 143 } 144 legacy_file = directory / "hosts.yml" 145 if legacy_file.exists(): 146 aliases.update(inventory_hosts(load_yaml(legacy_file), legacy_file)) 147 return sorted(aliases) 148 149 150 def load_target(directory: Path, alias: str) -> tuple[dict, bool]: 151 target_file = directory / f"{alias}.yml" 152 if target_file.exists(): 153 hosts = inventory_hosts(load_yaml(target_file), target_file) 154 if set(hosts) != {alias} or not isinstance(hosts[alias], dict): 155 raise SystemExit( 156 f"{target_file} must contain exactly the inventory host '{alias}'." 157 ) 158 return dict(hosts[alias]), False 159 160 legacy_file = directory / "hosts.yml" 161 if legacy_file.exists(): 162 hosts = inventory_hosts(load_yaml(legacy_file), legacy_file) 163 if alias in hosts: 164 if not isinstance(hosts[alias], dict): 165 raise SystemExit(f"The inventory entry for {alias} is not a mapping.") 166 values = dict(hosts[alias]) 167 values.update(load_yaml(directory / "host_vars" / f"{alias}.yml")) 168 return values, True 169 return {}, False 170 171 172 def write_target(directory: Path, alias: str, values: dict) -> Path: 173 directory.mkdir(parents=True, exist_ok=True) 174 target_file = directory / f"{alias}.yml" 175 inventory = { 176 "all": { 177 "children": { 178 "regional_currency": { 179 "hosts": {alias: values}, 180 } 181 } 182 } 183 } 184 atomic_write( 185 target_file, 186 "---\n" + yaml.safe_dump(inventory, sort_keys=False), 187 ) 188 return target_file 189 190 191 def remove_legacy_target(directory: Path, alias: str) -> None: 192 legacy_file = directory / "hosts.yml" 193 inventory = load_yaml(legacy_file) 194 hosts = inventory_hosts(inventory, legacy_file) 195 hosts.pop(alias) 196 empty_generated_inventory = { 197 "all": {"children": {"regional_currency": {"hosts": {}}}} 198 } 199 if inventory == empty_generated_inventory: 200 legacy_file.unlink() 201 else: 202 atomic_write( 203 legacy_file, 204 "---\n" + yaml.safe_dump(inventory, sort_keys=False), 205 ) 206 legacy_variables = directory / "host_vars" / f"{alias}.yml" 207 if legacy_variables.exists(): 208 legacy_variables.unlink() 209 210 211 def collect_configuration( 212 directory: Path, 213 ) -> tuple[str, dict, dict, bool, bool, bool]: 214 aliases = discover_aliases(directory) 215 if aliases: 216 print(f"Existing inventory hosts: {', '.join(aliases)}") 217 alias_default = aliases[0] if len(aliases) == 1 else "regional-currency" 218 alias = ask_matching( 219 "Inventory host name to add or update", HOST_PATTERN, alias_default 220 ) 221 updating = alias in aliases 222 if updating: 223 print( 224 f"Updating existing host '{alias}'. Press enter to keep each current value." 225 ) 226 227 existing, legacy = load_target(directory, alias) 228 connection_keys = {"ansible_host", "ansible_user", "ansible_port", "ansible_become"} 229 existing_connection = { 230 key: value for key, value in existing.items() if key in connection_keys 231 } 232 variables = { 233 key: value for key, value in existing.items() if key not in connection_keys 234 } 235 236 connection = { 237 **existing_connection, 238 "ansible_host": ask( 239 "SSH host name or address", 240 string_value(existing_connection.get("ansible_host")), 241 ), 242 "ansible_user": ask( 243 "SSH user", 244 string_value(existing_connection.get("ansible_user"), getpass.getuser()), 245 ), 246 "ansible_port": ask_int( 247 "SSH port", int(existing_connection.get("ansible_port", 22)) 248 ), 249 # The role also becomes dedicated service users when SSH connects as root. 250 "ansible_become": True, 251 } 252 253 currency = ask_matching( 254 "Regional currency code", 255 CURRENCY_PATTERN, 256 string_value(variables.get("regional_currency_currency"), "NETZBON"), 257 normalize=str.upper, 258 ) 259 testing = ask_yes_no( 260 "Configure a testing deployment", 261 bool_value(variables.get("regional_currency_testing_deployment"), False), 262 ) 263 variables.update( 264 { 265 "regional_currency_currency": currency, 266 "regional_currency_domain": ask( 267 "Base domain for the service hosts and optional landing page", 268 string_value(variables.get("regional_currency_domain")), 269 ), 270 "regional_currency_enable_landing_page": ask_yes_no( 271 "Configure a landing page on the base domain", 272 bool_value( 273 variables.get("regional_currency_enable_landing_page"), True 274 ), 275 ), 276 "regional_currency_bank_name": ask( 277 "Human-readable bank name", 278 string_value( 279 variables.get("regional_currency_bank_name"), 280 "Taler Test Bank" if testing else "Taler Bank", 281 ), 282 ), 283 "regional_currency_testing_deployment": testing, 284 "regional_currency_bank_port": ask_int( 285 "Internal LibEuFin bank port", 286 int(variables.get("regional_currency_bank_port", 8080)), 287 ), 288 } 289 ) 290 291 existing_password = string_value( 292 variables.get("regional_currency_bank_admin_password") 293 ) 294 if existing_password: 295 bank_password = getpass.getpass( 296 "Bank administrator password (leave empty to keep current): " 297 ).strip() 298 password_generated = False 299 if bank_password: 300 variables["regional_currency_bank_admin_password"] = bank_password 301 else: 302 bank_password = getpass.getpass( 303 "Bank administrator password (leave empty to generate one): " 304 ).strip() 305 password_generated = not bank_password 306 variables["regional_currency_bank_admin_password"] = ( 307 bank_password or secrets.token_urlsafe(24) 308 ) 309 310 tls_was_enabled = variables.get("regional_currency_enable_tls") is True 311 existing_tls = bool_value(variables.get("regional_currency_enable_tls"), True) 312 tls = ask_yes_no("Obtain TLS certificates using Let's Encrypt", existing_tls) 313 variables["regional_currency_enable_tls"] = tls 314 if tls: 315 variables["regional_currency_tls_email"] = ask( 316 "Let's Encrypt contact email", 317 string_value(variables.get("regional_currency_tls_email")), 318 ) 319 if not tls_was_enabled: 320 print("Read the Let's Encrypt subscriber agreement before continuing:") 321 print("https://letsencrypt.org/repository/") 322 if not ask_yes_no("Do you agree to the Let's Encrypt terms", False): 323 raise SystemExit( 324 "TLS setup requires agreement to the Let's Encrypt terms." 325 ) 326 327 conversion = ask_yes_no( 328 "Configure conversion to a fiat currency", 329 bool_value(variables.get("regional_currency_enable_conversion"), True), 330 ) 331 variables["regional_currency_enable_conversion"] = conversion 332 if conversion: 333 variables.update( 334 { 335 "regional_currency_fiat_currency": ask_matching( 336 "Fiat currency code", 337 CURRENCY_PATTERN, 338 string_value( 339 variables.get("regional_currency_fiat_currency"), "CHF" 340 ), 341 normalize=str.upper, 342 ), 343 "regional_currency_fiat_bank_name": ask( 344 "Fiat bank name", 345 string_value(variables.get("regional_currency_fiat_bank_name")), 346 ), 347 "regional_currency_fiat_account_iban": ask_matching( 348 "Fiat account IBAN", 349 IBAN_PATTERN, 350 string_value(variables.get("regional_currency_fiat_account_iban")), 351 normalize=lambda value: ( 352 value.replace(" ", "").replace("-", "").upper() 353 ), 354 ), 355 "regional_currency_fiat_account_bic": ask_matching( 356 "Fiat account BIC", 357 BIC_PATTERN, 358 string_value(variables.get("regional_currency_fiat_account_bic")), 359 normalize=lambda value: ( 360 value.replace(" ", "").replace("-", "").upper() 361 ), 362 ), 363 "regional_currency_fiat_account_name": ask( 364 "Fiat account legal name", 365 string_value(variables.get("regional_currency_fiat_account_name")), 366 ), 367 } 368 ) 369 conversion_test_mode = testing and ask_yes_no( 370 "Use simulated conversion test mode (no EBICS bank connection)", 371 bool_value(variables.get("regional_currency_conversion_test_mode"), False), 372 ) 373 variables["regional_currency_conversion_test_mode"] = conversion_test_mode 374 if conversion_test_mode: 375 variables["regional_currency_enable_nexus_services"] = False 376 else: 377 variables["regional_currency_enable_nexus_services"] = ask_yes_no( 378 "Enable Nexus services (only after EBICS enrollment is complete)", 379 bool_value( 380 variables.get("regional_currency_enable_nexus_services"), False 381 ), 382 ) 383 else: 384 variables["regional_currency_conversion_test_mode"] = False 385 variables["regional_currency_enable_nexus_services"] = False 386 387 telesign = ask_yes_no( 388 "Configure Telesign SMS authentication", 389 bool_value(variables.get("regional_currency_enable_telesign"), False), 390 ) 391 variables["regional_currency_enable_telesign"] = telesign 392 if telesign: 393 existing_token = string_value( 394 variables.get("regional_currency_telesign_auth_token") 395 ) 396 token_prompt = "Base64-encoded Telesign customer-ID/API-key token" 397 if existing_token: 398 token_prompt += " (leave empty to keep current)" 399 token = getpass.getpass(f"{token_prompt}: ").strip() 400 if token: 401 variables["regional_currency_telesign_auth_token"] = token 402 elif not existing_token: 403 raise SystemExit( 404 "A Telesign token is required when SMS authentication is enabled." 405 ) 406 407 terms = ask_yes_no( 408 "Configure exchange terms of service", 409 bool_value(variables.get("regional_currency_enable_exchange_terms"), True), 410 ) 411 variables["regional_currency_enable_exchange_terms"] = terms 412 if terms: 413 variables["regional_currency_exchange_terms_file"] = ask( 414 "Terms file on the managed host", 415 string_value( 416 variables.get("regional_currency_exchange_terms_file"), 417 "/usr/share/taler-exchange/terms/exchange-tos-v0.en.rst", 418 ), 419 ) 420 421 privacy = ask_yes_no( 422 "Configure an exchange privacy policy", 423 bool_value(variables.get("regional_currency_enable_exchange_privacy"), True), 424 ) 425 variables["regional_currency_enable_exchange_privacy"] = privacy 426 if privacy: 427 variables["regional_currency_exchange_privacy_file"] = ask( 428 "Privacy file on the managed host", 429 string_value( 430 variables.get("regional_currency_exchange_privacy_file"), 431 "/usr/share/taler-exchange/terms/exchange-pp-v0.en.rst", 432 ), 433 ) 434 435 variables["regional_currency_apt_testing"] = ask_yes_no( 436 "Use the GNU Taler testing APT repository (Debian only)", 437 bool_value(variables.get("regional_currency_apt_testing"), False), 438 ) 439 return alias, connection, variables, password_generated, updating, legacy 440 441 442 def main() -> None: 443 parser = argparse.ArgumentParser( 444 description="Create an operator-side Ansible inventory for regional-currency" 445 ) 446 parser.add_argument( 447 "directory", 448 nargs="?", 449 type=Path, 450 default=Path("inventory"), 451 help="inventory directory to use (default: inventory)", 452 ) 453 args = parser.parse_args() 454 directory = args.directory 455 alias, connection, variables, password_generated, updating, legacy = ( 456 collect_configuration(directory) 457 ) 458 target_file = write_target(directory, alias, {**connection, **variables}) 459 if legacy: 460 remove_legacy_target(directory, alias) 461 462 action = "Updated" if updating else "Added" 463 print(f"\n{action} inventory target '{alias}' in {target_file}") 464 if legacy: 465 print("Migrated the target from the legacy shared inventory layout.") 466 if password_generated: 467 print( 468 "A bank administrator password was generated and stored in the target file." 469 ) 470 print("\nCheck connectivity:") 471 print(f" ansible -i {directory} regional_currency -m ping") 472 print("Deploy:") 473 print(f" ansible-playbook -i {directory} site.yml") 474 print("Optional inventory encryption:") 475 print(f" ansible-vault encrypt {target_file}") 476 477 478 if __name__ == "__main__": 479 main()