""" Alby Hub manager — shared backend for Wallet Connections API and recovery CLI. Interfaces with the local Alby Hub instance at http://127.0.0.1:8080. All sensitive values (passwords, bearer tokens, pairing URIs, macaroon contents, Nostr private keys) are redacted from any exception messages or log output. """ from __future__ import annotations import json import logging import os import re import threading import time import urllib.error import urllib.parse import urllib.request from typing import Any logger = logging.getLogger(__name__) # ── Constants ────────────────────────────────────────────────────── DEFAULT_API_BASE = "http://127.0.0.1:8080" DEFAULT_UNLOCK_PASSWORD_FILE = "/var/lib/albyhub/unlock-password" DEFAULT_MACAROON_FILE = os.environ.get( "NWC_LND_MACAROON_FILE", "/run/lnd/albyhub.macaroon" ) DEFAULT_LND_ADDRESS = os.environ.get("NWC_LND_ADDRESS", "127.0.0.1:10009") DEFAULT_LND_CERT_FILE = os.environ.get("NWC_LND_CERT_FILE", "/var/lib/lnd/tls.cert") DEFAULT_LND_SOCKET = "/run/lnd/lnd.socket" LNURL_DESCRIPTION_DEFAULT = "Pay via Lightning" NWC_MIN_SENDABLE_MSAT = 1000 NWC_MAX_SENDABLE_MSAT = 1_000_000_000 # Metadata key used to mark managed isolated wallets _MANAGED_APP_STORE_ID = "uncle-jim" _MANAGED_META_KEY = "app_store_app_id" RECEIVE_ONLY_SCOPES = [ "get_info", "get_balance", "make_invoice", "lookup_invoice", "list_transactions", "notifications", ] LIMITED_SEND_SCOPES = RECEIVE_ONLY_SCOPES + ["pay_invoice"] # ── Exceptions ───────────────────────────────────────────────────── class AlbyHubError(Exception): """Base error from the Alby Hub manager. The message string is safe to surface to the user — it never contains raw secret material. """ def __init__(self, code: str, message: str) -> None: super().__init__(message) self.code = code def __str__(self) -> str: return f"[{self.code}] {self.args[0]}" class AlbyHubHttpError(AlbyHubError): def __init__(self, status_code: int, message: str) -> None: super().__init__(f"http_{status_code}", message) self.status_code = status_code # ── Manager class ────────────────────────────────────────────────── class AlbyHubManager: """Thread-safe manager for Alby Hub API operations.""" def __init__( self, api_base: str = DEFAULT_API_BASE, unlock_password_file: str = DEFAULT_UNLOCK_PASSWORD_FILE, macaroon_file: str = DEFAULT_MACAROON_FILE, lnd_address: str = DEFAULT_LND_ADDRESS, lnd_cert_file: str = DEFAULT_LND_CERT_FILE, ) -> None: self.api_base = api_base.rstrip("/") self.unlock_password_file = unlock_password_file self.macaroon_file = macaroon_file self.lnd_address = lnd_address self.lnd_cert_file = lnd_cert_file self._lock = threading.Lock() self._token: str | None = None # ── Low-level HTTP ───────────────────────────────────────────── def _request( self, method: str, path: str, body: dict | None = None, token: str | None = None, timeout: int = 30, ) -> dict: """Make a raw HTTP request to the local Alby Hub API. Returns the parsed JSON response body. Raises AlbyHubHttpError on non-2xx responses. Secrets in response bodies are never included in raised exceptions. """ url = f"{self.api_base}{path}" data = json.dumps(body).encode("utf-8") if body is not None else None headers: dict[str, str] = { "Content-Type": "application/json", "Accept": "application/json", } if token: headers["Authorization"] = "Bearer " + token req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read() if not raw: return {} return json.loads(raw) except urllib.error.HTTPError as exc: code = exc.code # Read and discard the body — we do NOT include it in the exception try: exc.read() except Exception: pass raise AlbyHubHttpError(code, f"Hub API {method} {path} returned HTTP {code}") from None except (urllib.error.URLError, TimeoutError, OSError): raise AlbyHubError( "hub_unreachable", f"Hub API {method} {path} is unreachable", ) from None def _authenticated_request( self, method: str, path: str, body: dict | None = None, timeout: int = 30, ) -> dict: """Make an authenticated request; retry once with a fresh token on 401/403.""" token = self.ensure_ready() try: return self._request(method, path, body=body, token=token, timeout=timeout) except AlbyHubHttpError as exc: if exc.status_code in (401, 403): with self._lock: self._token = None token = self.ensure_ready() return self._request(method, path, body=body, token=token, timeout=timeout) raise def _paginate(self, path_template: str, page_size: int = 100) -> list[dict]: """Paginate a list API completely, collecting all items. ``path_template`` must contain ``{limit}`` and ``{offset}`` placeholders. """ token = self.ensure_ready() offset = 0 results: list[dict] = [] while True: path = path_template.format(limit=page_size, offset=offset) page = self._request("GET", path, token=token) # Alby Hub returns apps at the top level or under "apps"/"transactions" total_count: int | None = None if isinstance(page, list): items = page elif isinstance(page, dict): items = page.get("apps") or page.get("transactions") or [] if page.get("totalCount") is not None: total_count = int(page.get("totalCount")) else: items = [] if not isinstance(items, list): break results.extend(items) if total_count is not None: if len(results) >= total_count: break elif len(items) < page_size: break offset += page_size return results # ── Startup / Auth ───────────────────────────────────────────── def _read_unlock_password(self) -> str: try: with open(self.unlock_password_file, "r") as fh: return fh.read().strip() except OSError as exc: raise AlbyHubError( "unlock_password_unavailable", "Cannot read Alby Hub unlock password", ) from exc def _wait_for_file(self, path: str, timeout: int = 120) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: if os.path.exists(path): return time.sleep(2) raise AlbyHubError( "dependency_unavailable", f"Timed out waiting for required file: {path}", ) def _wait_for_hub_api(self, timeout: int = 120) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: self._request("GET", "/api/info", timeout=5) return except AlbyHubError: pass time.sleep(3) raise AlbyHubError("hub_unavailable", "Timed out waiting for Alby Hub API") def _hub_setup(self, password: str) -> None: """Perform /api/setup idempotently.""" try: info = self._request("GET", "/api/info", timeout=10) if info.get("setupCompleted"): return except AlbyHubError: pass setup_body = { "backendType": "LND", "unlockPassword": password, "lndAddress": self.lnd_address, "lndCertFile": self.lnd_cert_file, "lndMacaroonFile": self.macaroon_file, } try: self._request("POST", "/api/setup", body=setup_body, timeout=30) except AlbyHubHttpError as exc: if exc.status_code == 409: return # already setup raise def _obtain_token(self, password: str) -> str: info = self._request("GET", "/api/info", timeout=10) if info.get("running"): resp = self._request( "POST", "/api/unlock", body={ "unlockPassword": password, "permission": "full", }, timeout=30, ) else: resp = self._request( "POST", "/api/start", body={"unlockPassword": password}, timeout=30, ) token = ( resp.get("token") or resp.get("accessToken") or resp.get("access_token") ) if not token or not isinstance(token, str): raise AlbyHubError("auth_failed", "Alby Hub auth response missing token") return token def _wait_for_node_ready(self, token: str, timeout: int = 120) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: status = self._request( "GET", "/api/node/status", token=token, timeout=10 ) if status.get("isReady") or status.get("running") or status.get("online"): return except AlbyHubError: pass time.sleep(3) raise AlbyHubError("node_not_ready", "Timed out waiting for Alby Hub node to be ready") def ensure_ready(self) -> str: """Ensure Alby Hub is set up, unlocked, and authenticated. Returns a valid bearer token. Caches it and uses a lock to prevent concurrent setup races. """ with self._lock: if self._token: return self._token password = self._read_unlock_password() self._wait_for_file(self.macaroon_file, timeout=120) self._wait_for_hub_api(timeout=120) self._hub_setup(password) token = self._obtain_token(password) self._wait_for_node_ready(token, timeout=120) self._token = token return token # ── App isolation helpers ────────────────────────────────────── @staticmethod def _parse_metadata(raw: Any) -> dict: if isinstance(raw, dict): return raw if isinstance(raw, str): try: result = json.loads(raw) if isinstance(result, dict): return result except Exception: pass return {} def _is_managed_app(self, app: dict) -> bool: meta = self._parse_metadata(app.get("metadata")) alias = str(meta.get("lnurl_alias", "")).strip().lower() return ( meta.get(_MANAGED_META_KEY) == _MANAGED_APP_STORE_ID and bool(alias) ) def _app_to_wallet_meta(self, app: dict, domain: str | None) -> dict: meta = self._parse_metadata(app.get("metadata")) alias = meta.get("lnurl_alias", "") address = f"{alias}@{domain}" if alias and domain else None scopes = app.get("scopes") or [] access_preset = ( "send_receive_limited" if "pay_invoice" in scopes else "receive_only" ) balance_msat = int(app.get("balanceMsat", 0) or 0) balance_sats = balance_msat // 1000 dust_msat = balance_msat % 1000 spending_limit_sats: int | None = None max_amount = app.get("maxAmountSat") or 0 if max_amount: spending_limit_sats = int(max_amount) # Count pending transactions from the budget or transactions list pending_txs = int(app.get("pendingTransactionsCount", 0) or 0) return { "id": str(app.get("id", "")), "pubkey": app.get("appPubkey") or app.get("nostrPubkey") or app.get("pubkey") or "", "name": app.get("name", ""), "alias": alias, "lightning_address": address, "access_preset": access_preset, "spending_limit_sats": spending_limit_sats, "balance_sats": balance_sats, "dust_msat": dust_msat, "pending_transactions": pending_txs, "created_at": app.get("createdAt") or app.get("created_at"), "min_sendable_msat": int( meta.get("lnurl_min_sendable_msat", NWC_MIN_SENDABLE_MSAT) ), "max_sendable_msat": int( meta.get("lnurl_max_sendable_msat", NWC_MAX_SENDABLE_MSAT) ), } def _all_managed_apps(self) -> list[dict]: apps = self._paginate("/api/apps?limit={limit}&offset={offset}&order_by=created_at") return [a for a in apps if a.get("isolated") and self._is_managed_app(a)] def _find_managed_app(self, identifier: str) -> dict | None: needle = identifier.strip().lower() for app in self._all_managed_apps(): if str(app.get("id", "")).lower() == needle: return app pubkey = ( app.get("appPubkey") or app.get("nostrPubkey") or app.get("pubkey") or "" ).lower() if pubkey == needle: return app return None # ── Public API ───────────────────────────────────────────────── def list_wallets(self, domain: str | None = None) -> list[dict]: """Return all managed isolated app wallets (no secrets).""" wallets = [] for app in self._all_managed_apps(): app_copy = dict(app) app_copy["pendingTransactionsCount"] = len(self._get_app_pending_txs(int(app["id"]))) wallets.append(self._app_to_wallet_meta(app_copy, domain)) return wallets def create_wallet( self, name: str, alias: str, access_preset: str, spending_limit_sats: int | None, domain: str | None = None, ) -> dict: """Create a new isolated Alby Hub app (wallet connection). Returns a dict containing: ``wallet`` — safe metadata (no secrets) ``pairing_uri`` — real Alby Hub pairingUri (returned ONCE) ``result`` — creation status report """ # Validate uniqueness managed = self._all_managed_apps() for a in managed: meta = self._parse_metadata(a.get("metadata")) if meta.get("lnurl_alias", "").lower() == alias.lower(): raise AlbyHubError( "alias_exists", "That Lightning Address alias is already in use." ) if (a.get("name") or "").lower() == name.lower(): raise AlbyHubError( "wallet_name_exists", "That Wallet Connection name already exists.", ) scopes = ( LIMITED_SEND_SCOPES if access_preset == "send_receive_limited" else RECEIVE_ONLY_SCOPES ) max_amount = ( spending_limit_sats if access_preset == "send_receive_limited" and spending_limit_sats else 0 ) create_body: dict = { "name": name, "scopes": scopes, "isolated": True, "budgetRenewal": "never", "maxAmountSat": max_amount, "metadata": { _MANAGED_META_KEY: _MANAGED_APP_STORE_ID, "lnurl_alias": alias, "lnurl_description": LNURL_DESCRIPTION_DEFAULT, "lnurl_min_sendable_msat": NWC_MIN_SENDABLE_MSAT, "lnurl_max_sendable_msat": NWC_MAX_SENDABLE_MSAT, }, } resp = self._authenticated_request("POST", "/api/apps", body=create_body) pairing_uri: str = resp.get("pairingUri") or resp.get("pairing_uri") or "" app_id = resp.get("id") # Fetch full app details for accurate metadata app_detail: dict | None = None if app_id is not None: try: app_detail = self._authenticated_request( "GET", f"/api/v2/apps/{app_id}" ) except AlbyHubError: pass if app_detail is None: # Fallback: search recent apps for the one we just created updated = self._all_managed_apps() for a in updated: if str(a.get("id", "")) == str(app_id): app_detail = a break wallet_meta = self._app_to_wallet_meta(app_detail or resp, domain) # Initial internal transfer for limited wallets funding_result: dict = {"attempted": False, "success": False} if ( access_preset == "send_receive_limited" and spending_limit_sats and app_id is not None ): funding_result["attempted"] = True try: self._authenticated_request( "POST", "/api/transfers", body={ "toAppId": int(app_id), "amountSat": spending_limit_sats, "description": f"Initial funding for {name}", }, ) funding_result["success"] = True except AlbyHubError as exc: funding_result["error"] = exc.code funding_result["message"] = ( "The wallet was created successfully and the NWC connection secret is shown " "above, but initial funding failed. Save the NWC secret now. " "Do not recreate this wallet." ) return { "wallet": wallet_meta, "pairing_uri": pairing_uri, # returned once on create only "result": { "wallet_created": True, "secret_created": bool(pairing_uri), "lightning_address_registered": bool(alias and domain), "funding": funding_result, }, } def _get_app_balance_msat(self, app: dict) -> int: return int(app.get("balanceMsat", 0) or 0) def _get_app_pending_txs(self, app_id: int) -> list[dict]: txs = self._paginate( f"/api/transactions?appId={app_id}&limit={{limit}}&offset={{offset}}" ) return [ t for t in txs if str(t.get("state", "")).lower() == "pending" ] def drain_wallet(self, identifier: str) -> dict: """Drain all whole-satoshi funds from an isolated app to the primary wallet. Returns ``{"ok": True, "drained_sats": N, "dust_msat": M}``. Raises AlbyHubError on rejection or failure. """ app = self._find_managed_app(identifier) if app is None: raise AlbyHubError("wallet_not_found", "Wallet connection not found.") app_id = int(app["id"]) balance_msat = self._get_app_balance_msat(app) if balance_msat < 0: raise AlbyHubError("negative_balance", "Wallet has a negative balance.") pending = self._get_app_pending_txs(app_id) if pending: raise AlbyHubError( "pending_transactions", "Wallet has pending transactions and cannot be drained.", ) transferable_msat = (balance_msat // 1000) * 1000 expected_dust_msat = balance_msat - transferable_msat if transferable_msat == 0: return {"ok": True, "drained_sats": 0, "dust_msat": expected_dust_msat} # Save original permissions original_scopes = list(app.get("scopes") or []) original_max = app.get("maxAmountSat") or 0 original_renewal = app.get("budgetRenewal") or "never" # Temporarily grant pay_invoice scope with sufficient budget app_pubkey = app.get("appPubkey") or app.get("nostrPubkey") or app.get("pubkey") or "" if not app_pubkey: raise AlbyHubError( "app_pubkey_missing", "Cannot drain app: app public key not available.", ) patch_body = { "scopes": sorted(set(original_scopes) | {"pay_invoice"}), "maxAmountSat": 0, "budgetRenewal": "never", } self._authenticated_request("PATCH", f"/api/apps/{app_pubkey}", body=patch_body) drain_error: AlbyHubError | None = None drained_sats = 0 try: self._authenticated_request( "POST", "/api/transfers", body={ "fromAppId": app_id, "amountMsat": transferable_msat, "description": f"Drain isolated subwallet {app.get('name', '')}", }, ) drained_sats = transferable_msat // 1000 except AlbyHubError as exc: drain_error = exc finally: # Restore original permissions whether drain succeeded or not restore_body = { "scopes": original_scopes, "maxAmountSat": original_max, "budgetRenewal": original_renewal, } try: self._authenticated_request( "PATCH", f"/api/apps/{app_pubkey}", body=restore_body ) except AlbyHubError: pass # best-effort restore; don't mask the original error if drain_error is not None: raise drain_error # Verify remaining balance equals expected dust refreshed = self._authenticated_request("GET", f"/api/v2/apps/{app_id}") remaining_msat = self._get_app_balance_msat(refreshed) if remaining_msat != expected_dust_msat: raise AlbyHubError( "drain_incomplete", "Drain verification failed: final balance does not match expected dust.", ) return { "ok": True, "drained_sats": drained_sats, "dust_msat": expected_dust_msat, "remaining_msat": remaining_msat, } def delete_wallet(self, identifier: str) -> dict: """Safely drain and delete an isolated app. Returns ``{"ok": True, "drained_sats": N}``. """ app = self._find_managed_app(identifier) if app is None: raise AlbyHubError("wallet_not_found", "Wallet connection not found.") app_id = int(app["id"]) pending = self._get_app_pending_txs(app_id) if pending: raise AlbyHubError( "pending_transactions", "Wallet has pending transactions and cannot be deleted.", ) drain_result = self.drain_wallet(identifier) # Verify no transferable balance remains refreshed = self._authenticated_request("GET", f"/api/v2/apps/{app_id}") remaining_msat = self._get_app_balance_msat(refreshed) if remaining_msat < 0: raise AlbyHubError( "negative_balance", "Wallet has a negative final balance and cannot be deleted.", ) if remaining_msat >= 1000: raise AlbyHubError( "drain_incomplete", f"Drain verification failed: funds still remain.", ) # Delete by app pubkey pubkey = app.get("appPubkey") or app.get("nostrPubkey") or app.get("pubkey") or "" if not pubkey: raise AlbyHubError( "app_pubkey_missing", "Cannot delete app: nostr pubkey not available.", ) self._authenticated_request( "DELETE", f"/api/apps/{urllib.parse.quote(pubkey, safe='')}", ) return { "ok": True, "drained_sats": drain_result.get("drained_sats", 0), "dust_msat": remaining_msat, } def issue_invoice( self, app_id: int, amount_msat: int, description: str = "" ) -> str: """Create an LND invoice attributed to a specific isolated app. Returns a valid BOLT11 invoice string. Raises AlbyHubError if the Hub returns an invalid or misattributed invoice. """ resp = self._authenticated_request( "POST", "/api/invoices", body={ "amountMsat": amount_msat, "description": description or LNURL_DESCRIPTION_DEFAULT, "appId": app_id, }, ) invoice: str = resp.get("invoice") or "" returned_app_id = resp.get("appId") if not invoice: raise AlbyHubError("invoice_creation_failed", "Hub returned empty invoice.") # Require a valid BOLT11 prefix (mainnet, testnet, signet, regtest) if not re.match(r"^ln", invoice, re.IGNORECASE): raise AlbyHubError( "invalid_invoice", "Hub returned a non-BOLT11 invoice string." ) if returned_app_id is None or int(returned_app_id) != app_id: raise AlbyHubError( "invoice_attribution_failed", "Invoice attribution mismatch: returned appId does not match.", ) return invoice def find_app_by_alias(self, alias: str) -> dict | None: """Find a managed isolated app by its ``lnurl_alias`` metadata field.""" alias_lower = alias.strip().lower() for a in self._all_managed_apps(): meta = self._parse_metadata(a.get("metadata")) if meta.get("lnurl_alias", "").lower() == alias_lower: return a return None def health(self) -> dict: """Return a basic health summary.""" try: token = self.ensure_ready() status = self._request( "GET", "/api/node/status", token=token, timeout=10 ) return { "ok": True, "hub_ready": bool( status.get("isReady") or status.get("running") ), } except AlbyHubError as exc: return {"ok": False, "error": exc.code, "message": str(exc)} # ── Module-level singleton ────────────────────────────────────────── _manager: AlbyHubManager | None = None _manager_lock = threading.Lock() def get_manager() -> AlbyHubManager: """Return the module-level singleton AlbyHubManager.""" global _manager if _manager is None: with _manager_lock: if _manager is None: _manager = AlbyHubManager() return _manager