Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ dependencies = [
"websockets>=12.0",
"pillow>=10.0",
"zxing-cpp>=2.2",
"qrcode[pil]>=7.0",
]

[project.optional-dependencies]
Expand All @@ -42,7 +43,6 @@ dev = [
"black>=24.0.0",
"ruff>=0.2.0",
"pytest-cov>=7.0.0",
"qrcode[pil]>=7.0",
]

[project.scripts]
Expand Down
53 changes: 53 additions & 0 deletions src/aqua/qr.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,62 @@
import hashlib
import os
import tempfile
from pathlib import Path

import qrcode
import zxingcpp
from PIL import Image


def generate_qr(data: str, output_dir: str | Path, filename: str | None = None) -> str:
"""Generate a PNG QR code for ``data`` and return the saved file path.

Used to render deposit addresses, Lightning invoices and swap deposit
addresses as scannable QR images. The file is written atomically (temp
file + ``os.replace``) with ``0o600`` permissions, mirroring the rest of
the on-disk ``~/.aqua`` layout.

Args:
data: The string to encode (address, BOLT11 invoice, etc.).
output_dir: Directory to write the PNG into (created at ``0o700`` if
missing). Typically ``Storage.qr_dir``.
filename: Optional file name. Defaults to a content-addressed name
``qr_<sha256(data)[:16]>.png`` so identical payloads reuse one file.

Returns:
Absolute path to the written PNG.

Raises:
ValueError: If ``data`` is not a non-empty string.
"""
if not isinstance(data, str) or not data:
raise ValueError("QR data must be a non-empty string")

out_dir = Path(output_dir)
out_dir.mkdir(parents=True, exist_ok=True, mode=0o700)

if filename is None:
digest = hashlib.sha256(data.encode("utf-8")).hexdigest()[:16]
filename = f"qr_{digest}.png"
path = out_dir / filename

img = qrcode.make(data)
fd, tmp_name = tempfile.mkstemp(dir=str(out_dir), suffix=".png.tmp")
tmp_path = Path(tmp_name)
try:
with os.fdopen(fd, "wb") as fh:
img.save(fh, format="PNG")
fh.flush()
os.fsync(fh.fileno())
os.chmod(tmp_path, 0o600)
os.replace(tmp_path, path)
except BaseException:
tmp_path.unlink(missing_ok=True)
raise

return str(path.resolve())


def decode_qr(image_path: str) -> str:
path = Path(image_path)
if not path.is_file():
Expand Down
33 changes: 28 additions & 5 deletions src/aqua/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,10 @@
},
},
"lw_address": {
"description": "Generate a receive address",
"description": (
"Generate a receive address. Also returns qr_code_path: a PNG QR of the "
"address — display it to the user so they can scan it."
),
"inputSchema": {
"type": "object",
"properties": {
Expand Down Expand Up @@ -270,7 +273,10 @@
},
},
"btc_address": {
"description": "Generate a Bitcoin receive address (bc1...)",
"description": (
"Generate a Bitcoin receive address (bc1...). Also returns qr_code_path: a "
"PNG QR of the address — display it to the user so they can scan it."
),
"inputSchema": {
"type": "object",
"properties": {
Expand Down Expand Up @@ -399,7 +405,7 @@
},
},
"lightning_receive": {
"description": "Generate a Lightning invoice to receive L-BTC into a Liquid wallet (~1-2 min after payment). Limits: 100 – 25,000,000 Sats.",
"description": "Generate a Lightning invoice to receive L-BTC into a Liquid wallet (~1-2 min after payment). Limits: 100 – 25,000,000 Sats. Also returns qr_code_path: a PNG QR of the invoice — display it to the user so they can scan it.",
"inputSchema": {
"type": "object",
"properties": {
Expand Down Expand Up @@ -572,7 +578,8 @@
"Receive USDt-Liquid via a Changelly variable-rate swap. Returns a "
"deposit address on the source chain — the external sender pays to it. "
"Settles to the wallet's Liquid address as USDt-Liquid. STRONGLY RECOMMEND "
"passing external_refund_address."
"passing external_refund_address. Also returns qr_code_path: a PNG QR of "
"the deposit address — display it to the user so they can scan it."
),
"inputSchema": {
"type": "object",
Expand Down Expand Up @@ -1004,7 +1011,9 @@
"ethereum/tron/bsc/solana/polygon/liquid, or BTC on bitcoin) — mirrors "
"AQUA Flutter's supported pairs. Set SIDESHIFT_ALLOW_ALL_NETWORKS=1 to bypass. "
"STRONGLY RECOMMEND passing external_refund_address (the deposit-side "
"sender's address) so a stuck shift can refund automatically."
"sender's address) so a stuck shift can refund automatically. Also returns "
"qr_code_path: a PNG QR of the deposit address — display it to the user so "
"they can scan it."
),
"inputSchema": {
"type": "object",
Expand Down Expand Up @@ -1138,6 +1147,20 @@ def create_server(config: Config | None = None) -> Server:
derived keys. The seed alone fully restores the same descriptors on Liquid
and Bitcoin in any BIP39-compliant wallet.

QR CODES (deposit addresses & invoices):
- The receive tools — btc_address, lw_address, lightning_receive, changelly_receive,
sideshift_receive — return a `qr_code_path`: an absolute path to a PNG QR image of
the address/invoice saved on disk.
- ALWAYS surface this to the user so they can scan instead of copy-paste: display the
image inline if your client renders local image paths, otherwise tell the user the
file path where the QR was saved.
- If a result has `qr_error` instead of `qr_code_path`, QR generation failed but the
address/invoice is still valid — show the text and mention the QR was unavailable.
- MEMO WARNING: some deposits (e.g. memo/tag-based chains via sideshift_receive)
return a `deposit_memo` and a `qr_warning`. The QR encodes ONLY the address, never
the memo. Always surface the memo as text and warn the user it must be entered
manually — scanning the QR alone omits it and can cause permanent loss of funds.

LIGHTNING:
- Use lightning_receive to generate an invoice for receiving L-BTC from Lightning
Fees: ~0.1%, Limits: 100 - 25,000,000 Sats, Time: ~1-2 min after payment
Expand Down
3 changes: 3 additions & 0 deletions src/aqua/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,7 @@ def __init__(self, base_dir: Optional[Path] = None):
self.sideshift_shifts_dir = self.base_dir / "sideshift_shifts"
self.sideswap_pegs_dir = self.base_dir / "sideswap_pegs"
self.sideswap_swaps_dir = self.base_dir / "sideswap_swaps"
self.qr_dir = self.base_dir / "qr"
self.config_path = self.base_dir / "config.json"
self._ensure_dirs()

Expand Down Expand Up @@ -151,6 +152,8 @@ def _ensure_dirs(self):
os.chmod(self.sideswap_pegs_dir, 0o700)
self.sideswap_swaps_dir.mkdir(exist_ok=True, mode=0o700)
os.chmod(self.sideswap_swaps_dir, 0o700)
self.qr_dir.mkdir(exist_ok=True, mode=0o700)
os.chmod(self.qr_dir, 0o700)

def _derive_key(self, password: str, salt: bytes) -> bytes:
"""Derive encryption key from password."""
Expand Down
85 changes: 66 additions & 19 deletions src/aqua/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
from .assets import MAINNET_ASSETS, TESTNET_ASSETS, resolve_asset_name, resolve_liquid_asset_id
from .bitcoin import BitcoinWalletManager
from .bolt11 import decode_bolt11_fields
from .qr import decode_qr
from .qr import decode_qr, generate_qr
from .wallet import WalletManager

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -137,6 +137,33 @@ def get_sideswap_swap_manager() -> "SideSwapSwapManager":
# Tool implementations


def _attach_deposit_qr(
result: dict[str, Any], data_key: str, *, encode_transform=None
) -> dict[str, Any]:
"""Generate a PNG QR for ``result[data_key]`` and attach its path.

Adds ``qr_code_path`` (absolute path to the saved PNG) so the agent can
display the QR to the user. QR rendering is auxiliary to the deposit flow:
if generation fails the deposit address/invoice is still valid, so the
error is reported as ``qr_error`` rather than failing the whole tool.

``encode_transform`` optionally maps the stored value to the form encoded
into the QR (the stored field is left untouched). Used to uppercase BOLT11
invoices, which are case-insensitive: an uppercase payload lets the QR use
denser alphanumeric mode, producing a smaller, easier-to-scan code.
"""
value = result.get(data_key)
if not value:
return result
try:
qr_dir = get_manager().storage.qr_dir
payload = encode_transform(value) if encode_transform else value
result["qr_code_path"] = generate_qr(payload, qr_dir)
except Exception as exc: # noqa: BLE001 - auxiliary feature, report don't crash
result["qr_error"] = str(exc)
return result


def lw_generate_mnemonic() -> dict[str, Any]:
"""
Generate a new BIP39 mnemonic phrase (12 words).
Expand Down Expand Up @@ -274,10 +301,11 @@ def lw_address(
Returns:
address: The Liquid address
index: Address index
qr_code_path: Path to a PNG QR of the address (or qr_error on failure)
"""
manager = get_manager()
addr = manager.get_address(wallet_name, index)
return addr.to_dict()
return _attach_deposit_qr(addr.to_dict(), "address")


def lw_transactions(
Expand Down Expand Up @@ -526,10 +554,11 @@ def btc_address(
Returns:
address: The Bitcoin address
index: Address index
qr_code_path: Path to a PNG QR of the address (or qr_error on failure)
"""
btc = get_btc_manager()
addr = btc.get_address(wallet_name, index)
return addr.to_dict()
return _attach_deposit_qr(addr.to_dict(), "address")


def btc_transactions(
Expand Down Expand Up @@ -782,7 +811,8 @@ def lightning_receive(
password: Password to decrypt mnemonic (if encrypted at rest)

Returns:
swap_id, invoice, amount, wallet_name, message
swap_id, invoice, amount, wallet_name, message, qr_code_path
(qr_code_path is a PNG QR of the invoice, or qr_error on failure)
"""
manager = get_lightning_manager()
swap = manager.create_receive_invoice(amount, wallet_name, password)
Expand All @@ -791,17 +821,21 @@ def lightning_receive(
all_wallets = get_manager().storage.list_wallets()
wallet_note = f" in wallet '{wallet_name}'" if len(all_wallets) > 1 else ""

return {
"swap_id": swap.swap_id,
"invoice": swap.invoice,
"amount": amount,
"wallet_name": wallet_name,
"message": (
f"Pay this Lightning invoice to receive {amount} satoshis of L-BTC{wallet_note}. "
f"Usually takes 1–2 minutes to confirm on Liquid after Lightning payment confirms. "
f"You can ask the agent to check status with swap_id: {swap.swap_id}"
),
}
return _attach_deposit_qr(
{
"swap_id": swap.swap_id,
"invoice": swap.invoice,
"amount": amount,
"wallet_name": wallet_name,
"message": (
f"Pay this Lightning invoice to receive {amount} satoshis of L-BTC{wallet_note}. "
f"Usually takes 1–2 minutes to confirm on Liquid after Lightning payment confirms. "
f"You can ask the agent to check status with swap_id: {swap.swap_id}"
),
},
"invoice",
encode_transform=str.upper,
)


def lightning_send(
Expand Down Expand Up @@ -1095,7 +1129,8 @@ def changelly_receive(
Mutually exclusive with amount_from.

Returns:
order_id, deposit_address, settle_address, amount_from, status, track_url
order_id, deposit_address, settle_address, amount_from, status, track_url,
qr_code_path (PNG QR of deposit_address, or qr_error on failure)
"""
if (amount_from is None) == (amount_to is None):
raise ValueError("Provide exactly one of amount_from or amount_to")
Expand All @@ -1110,7 +1145,7 @@ def changelly_receive(
amount_from=amount_from,
amount_to=amount_to,
)
return swap.to_dict()
return _attach_deposit_qr(swap.to_dict(), "deposit_address")


def changelly_status(order_id: str) -> dict[str, Any]:
Expand Down Expand Up @@ -1320,7 +1355,8 @@ def sideshift_receive(

Returns:
shift_id, deposit_address, deposit_min, deposit_max, deposit_memo
(if applicable), settle_address, status, expires_at
(if applicable), settle_address, status, expires_at, qr_code_path
(qr_code_path is a PNG QR of deposit_address, or qr_error on failure)
"""
shift = get_sideshift_manager().receive_shift(
deposit_coin=deposit_coin,
Expand All @@ -1332,7 +1368,18 @@ def sideshift_receive(
external_refund_memo=external_refund_memo,
settle_memo=settle_memo,
)
return shift.to_dict()
result = _attach_deposit_qr(shift.to_dict(), "deposit_address")
# Memo-based chains (e.g. BNB) require a deposit memo/tag alongside the
# address. The QR encodes ONLY the address, so scanning it would silently
# drop the memo — sending without it can permanently lose funds. Flag it.
if result.get("deposit_memo") and "qr_code_path" in result:
result["qr_warning"] = (
"QR encodes the deposit address only. This network also requires a "
f"deposit memo ({result['deposit_memo']}) that the QR does NOT include — "
"it must be entered manually when sending. Sending without the memo can "
"cause permanent loss of funds."
)
return result


def sideshift_status(shift_id: str) -> dict[str, Any]:
Expand Down
Loading
Loading