Tworzenie własnego portfela Bitcoin SV: Bezpieczeństwo, baza danych (Część 2)

W poprzedniej części zbudowaliśmy fundament - wiemy już czym jest entropia, mnemonic i seed. Potrafiliśmy z jednego sekretu wyprowadzić 10 adresów i podejrzeć ich klucze prywatne. Fajne ćwiczenie, ale tamten kod nadawał się tylko do zabawy w konsoli. Klucze prywatne wypisywane wprost na ekran, mnemonic trzymany w zmiennej to.. to nie jest coś, co chciałbyś zobaczyć w prawdziwym portfelu. Przez ostatnie kilka dni (a może i tygodni) jak znalazłem trochę wolnego czasu, to siedziałem i coś tam dłubałem sobie przy portfelu. Na początku wwaliłem wszystko do jednego pliku, później rozdzieliłem, po jakimś czasie znowu w jednym i w sumie ostatecznie rozdzieliłem na moduły. Szkoda, że nie zapisywałem wszystkich moich potknięć (a może to i dobrze) bo można by było zobaczyć mój tok myślowy. Na początek postawiłem na minimum bezpieczeństwa.

Dziś zrobimy to porządniej. Zbudujemy portfel który:

• szyfruje mnemonic hasłem i zapisuje go bezpiecznie na dysku
• trzyma adresy w bazie SQLite (bez kluczy prywatnych!)
• sprawdza salda przez WhatsOnChain API
• ma menu w terminalu

I co ważniejsze, zaczynamy pisać kod jak człowiek, a nie jeden gigantyczny plik mający 500 linijek :) Na początku wszystko wrzucałem do jednego main.py ale po kilku dniach wracałem do projektu i sam nie pamiętałem gdzie co jest. Dlatego od razu dzielimy aplikację na moduły.

Struktura projektu

portfel/
├── main.py       # punkt wejścia, menu
├── config.py     # ścieżki, sieć
├── crypto.py     # szyfrowanie/deszyfrowanie mnemonika
├── db.py         # SQLite
├── hd.py         # generowanie adresów BIP44
├── woc.py        # WhatsOnChain API
└── ui.py         # wyświetlanie menu

Każdy plik odpowiada za jedną konkretną rzecz. Gdy coś przestanie działać, od razu wiesz gdzie szukać problemu.

config.py 

Zacznijmy od najprostszego. Tutaj lądują wartości które możesz chcieć kiedyś zmienić, np. przełączyć się z testnetu na mainnet. Taki nasz mini panel ustawień.

ITERATIONS = 200_000
KEYSTORE_FILE = "wallet_keystore.json"
DB_FILE = "wallet.db"
NETWORK = "test"  # zmień na "main" gdy idziesz na mainnet
WOC_BULK_BALANCE_URL = f"https://api.whatsonchain.com/v1/bsv/{NETWORK}/addresses/confirmed/balance"

ITERATIONS = 200_000 to liczba rund przy wyprowadzaniu klucza z hasła (PBKDF2). Im więcej, tym wolniej atakujący może zgadywać hasło metodą brute force. 200 tysięcy to rozsądny kompromis - na normalnym sprzęcie odblokowanie portfela zajmuje ułamek sekundy, a atakujący musi wykonać 200 000 operacji per każde sprawdzane hasło.

crypto.py

Pamiętasz z części pierwszej jak wypisywałem mnemonic i seed wprost na ekran? Teraz tego nie robimy. Traktujemy go jak sekret i szyfrujemy AES-256-GCM z kluczem wyprowadzonym z hasła użytkownika przez PBKDF2. Funkcja derive_key_from_password bierze hasło użytkownika i zamienia je na prawdziwy klucz kryptograficzny.

import os
import base64
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from config import ITERATIONS


def derive_key_from_password(password: str, salt: bytes) -> bytes:
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt,
        iterations=ITERATIONS,
    )
    return kdf.derive(password.encode("utf-8"))


def encrypt_mnemonic(mnemonic: str, password: str) -> dict:
    secret = mnemonic.encode("utf-8")
    salt = os.urandom(32)
    key = derive_key_from_password(password, salt)
    nonce = os.urandom(12)
    aesgcm = AESGCM(key)
    ciphertext = aesgcm.encrypt(nonce, secret, None)

    return {
        "kdf": "pbkdf2",
        "kdfparams": {
            "salt": base64.b64encode(salt).decode(),
            "iterations": ITERATIONS,
            "dklen": 32,
            "hash": "sha256",
        },
        "cipher": "aes-256-gcm",
        "cipherparams": {
            "nonce": base64.b64encode(nonce).decode(),
        },
        "ciphertext": base64.b64encode(ciphertext).decode(),
    }


def decrypt_mnemonic(keystore: dict, password: str) -> str:
    salt = base64.b64decode(keystore["kdfparams"]["salt"])
    nonce = base64.b64decode(keystore["cipherparams"]["nonce"])
    ciphertext = base64.b64decode(keystore["ciphertext"])
    key = derive_key_from_password(password, salt)
    aesgcm = AESGCM(key)
    secret = aesgcm.decrypt(nonce, ciphertext, None)
    mnemonic = secret.decode("utf-8")
    secret = b"\x00" * len(secret)  # nadpisujemy bajty zerami zaraz po użyciu
    return mnemonic

Tutaj dzieją się w sumie trzy ważne rzeczy:

1. SHA256 - funkcja hashująca używana przez PBKDF2,
2. length=32 - chcemy 32 bajty = 256 bitów,
3. salt.

Salt jest niezwykle ważny. Gdyby dwóch użytkowników miało identyczne hasło, bez salta otrzymaliby identyczny klucz szyfrujący. Salt sprawia, że nawet to samo hasło daje różne wyniki.

db.py

Ważna zasada: klucze prywatne nigdy nie trafiają do bazy. W SQLite trzymamy tylko adresy publiczne i ich indeksy derywacji. Nawet jak ktoś podejrzy plik wallet.db - nie zobaczy nic czułego :)

import sqlite3
from config import DB_FILE


def init_db(db_path: str = DB_FILE):
    conn = sqlite3.connect(db_path)
    cur = conn.cursor()
    cur.execute("""
        CREATE TABLE IF NOT EXISTS addresses (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            derivation_index INTEGER NOT NULL,
            change INTEGER NOT NULL DEFAULT 0,
            address TEXT NOT NULL UNIQUE,
            path TEXT NOT NULL,
            label TEXT
        )
    """)
    cur.execute("""
        CREATE TABLE IF NOT EXISTS utxos (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            txid TEXT NOT NULL,
            vout INTEGER NOT NULL,
            value_sats INTEGER NOT NULL,
            address_index INTEGER NOT NULL,
            spent INTEGER NOT NULL DEFAULT 0,
            FOREIGN KEY(address_index) REFERENCES addresses(derivation_index)
        )
    """)
    conn.commit()
    conn.close()


def get_all_addresses(db_path: str = DB_FILE) -> list:
    conn = sqlite3.connect(db_path)
    cur = conn.cursor()
    cur.execute("""
        SELECT derivation_index, change, address, path, COALESCE(label, '')
        FROM addresses ORDER BY change, derivation_index
    """)
    rows = cur.fetchall()
    conn.close()
    return rows


def get_max_derivation_index(change: int = 0, db_path: str = DB_FILE) -> int:
    conn = sqlite3.connect(db_path)
    cur = conn.cursor()
    cur.execute("SELECT MAX(derivation_index) FROM addresses WHERE change = ?", (change,))
    row = cur.fetchone()
    conn.close()
    return -1 if row[0] is None else row[0]


def insert_address(derivation_index: int, change: int, address: str, path: str, db_path: str = DB_FILE):
    conn = sqlite3.connect(db_path)
    cur = conn.cursor()
    cur.execute("""
        INSERT OR IGNORE INTO addresses (derivation_index, change, address, path)
        VALUES (?, ?, ?, ?)
    """, (derivation_index, change, address, path))
    conn.commit()
    conn.close()

Tabela utxos zostawiam na przyszłość - przyda się przy wysyłaniu transakcji. Na razie jest pusta, ale schemat jest gotowy.

Pole change w tabeli addresses to nie "reszta od zakupów" - to flaga z BIP44. Adresy receive (change=0) to te które dajesz komuś do wpłaty. Adresy change (change=1) to te na które wraca reszta z transakcji. Dziś generujemy tylko receive, ale struktura już to uwzględnia.

hd.py - generowanie adresów

Tu sięgamy do wiedzy z części pierwszej. Z mnemonika wyprowadzamy adresy według ścieżki BIP44 - dokładnie tak samo jak robiliśmy w poprzednim wpisie, tylko tym razem nie wypisujemy kluczy prywatnych na ekran, tylko zapisujemy same adresy do bazy.

from bsv.hd import mnemonic_from_entropy, bip44_derive_xprvs_from_mnemonic
from bsv.constants import BIP44_DERIVATION_PATH
import secrets
from db import get_max_derivation_index, insert_address
from config import DB_FILE


def generate_mnemonic() -> str:
    entropy = secrets.token_bytes(32)  # 256 bitów losowości
    return mnemonic_from_entropy(entropy)


def insert_addresses(mnemonic: str, count: int = 10, db_path: str = DB_FILE):
    keys = bip44_derive_xprvs_from_mnemonic(
        mnemonic,
        0,        # indeks startowy
        count,    # indeks stop
        path=BIP44_DERIVATION_PATH,
        change=0,
        network="testnet",
    )
    for i, key in enumerate(keys):
        insert_address(i, 0, key.address(), f"{BIP44_DERIVATION_PATH}/0/{i}", db_path)


def generate_next_address(mnemonic: str, db_path: str = DB_FILE) -> tuple:
    next_index = get_max_derivation_index(change=0, db_path=db_path) + 1

    keys = bip44_derive_xprvs_from_mnemonic(
        mnemonic,
        next_index,       # indeks startowy
        next_index + 1,   # indeks stop
        path=BIP44_DERIVATION_PATH,
        change=0,
        network="testnet",
    )
    key = keys[0]
    addr = key.address()
    path = f"{BIP44_DERIVATION_PATH}/0/{next_index}"
    insert_address(next_index, 0, addr, path, db_path)
    return next_index, addr, path

Jeden detal który tu widać: bip44_derive_xprvs_from_mnemonic działa jak range() - podajesz indeks startowy i końcowy. Żeby dostać jeden adres na indeksie 7, piszesz (mnemonic, 7, 8, ...). Nie musisz generować wszystkich poprzednich od zera - biblioteka liczy dokładnie ten jeden klucz który cię interesuje. To ważne gdy portfel będzie miał setki adresów. Generowanie wszystkich od początku za każdym razem byłoby marnotrawstwem.

woc.py - saldo

WhatsOnChain udostępnia endpoint, który przyjmuje do 20 adresów naraz w jednym zapytaniu. Przy większej liczbie adresów automatycznie dzielimy je na paczki.

import time
import requests
from config import WOC_BULK_BALANCE_URL


def fetch_balances_from_woc(addresses: list[str]) -> dict[str, int]:
    if not addresses:
        return {}

    chunk_size = 20
    balances = {}

    for i in range(0, len(addresses), chunk_size):
        chunk = addresses[i:i + chunk_size]
        response = requests.post(WOC_BULK_BALANCE_URL, json={"addresses": chunk})
        response.raise_for_status()
        for item in response.json():
            balances[item["address"]] = item["confirmed"]
        if i + chunk_size < len(addresses):
            time.sleep(0.3)  # krótka pauza między paczkami żeby nie trafić w rate limit

    return balances

time.sleep(0.3) aktywuje się tylko między paczkami - przy 20 lub mniej adresach w ogóle go nie poczujesz. Przy 100 adresach zrobi 5 zapytań z pauzami po 0.3s - spokojnie mieści się w limitach API.

ui.py

from colorama import init, Fore, Style
from db import get_all_addresses
from woc import fetch_balances_from_woc
init(autoreset=True)


def show_addresses(db_path: str = "wallet.db"):
    rows = get_all_addresses(db_path)
    if not rows:
        print("Brak adresów w bazie.")
        return

    addresses = [row[2] for row in rows]
    print("\nPobieram salda z WhatsOnChain...")
    try:
        balances = fetch_balances_from_woc(addresses)
    except Exception as e:
        print(f"Błąd pobierania sald: {e}")
        balances = {}

    print(f"\n{'#':<5} {'Typ':<8} {'Adres':<40} {'Saldo (BSV)':>12}  Ścieżka")
    print("-" * 90)

    total = 0
    for idx, change, addr, path, label in rows:
        ch_str = "change" if change else "receive"
        sats = balances.get(addr, 0)
        bsv = sats / 1e8
        total += sats
        label_str = f"  [{label}]" if label else ""

        balance_str = (
            f"{Fore.GREEN}{bsv:>12.8f}{Style.RESET_ALL}" # <-- funkcja koloru
            if sats > 0
            else f"{bsv:>12.8f}"
        )
        print(f"{idx:<5} {ch_str:<8} {addr:<40} {balance_str}  {path}{label_str}")

    print("-" * 90)
    print(f"{'RAZEM':<54} {total/1e8:>12.8f} BSV")

Drobna rzecz, a cieszy - adresy z saldem świecą na zielono. Zero satoshi zostaje szare i nie przyciąga wzroku.

main.py - spinamy wszystko razem

import os
import json
from getpass import getpass
from config import KEYSTORE_FILE, DB_FILE
from crypto import encrypt_mnemonic, decrypt_mnemonic
from db import init_db
from hd import generate_mnemonic, insert_addresses, generate_next_address
from ui import show_addresses


def clear_screen():
    os.system("cls" if os.name == "nt" else "clear")


def wallet_exists() -> bool:
    return os.path.exists(KEYSTORE_FILE)


def load_mnemonic_from_keystore(password: str) -> str:
    with open(KEYSTORE_FILE, "r") as f:
        keystore = json.load(f)
    return decrypt_mnemonic(keystore, password)


def create_new_wallet():
    mnemonic = generate_mnemonic()
    print("NOWY MNEMONIC (ZAPISZ NA KARTCE!):")
    print(mnemonic)

    while True:
        password = getpass("Ustaw hasło: ")
        password_confirm = getpass("Potwierdź hasło: ")
        if password == password_confirm:
            break
        print("Hasła nie są zgodne, spróbuj ponownie.\n")

    keystore = encrypt_mnemonic(mnemonic, password)
    with open(KEYSTORE_FILE, "w") as f:
        json.dump(keystore, f, indent=2)

    init_db(DB_FILE)
    insert_addresses(mnemonic, count=10, db_path=DB_FILE)
    print(f"\nPortfel utworzony. Dodano 10 adresów odbiorczych.")
    mnemonic = password = password_confirm = ""


def unlock_and_show_addresses():
    if not wallet_exists():
        print("Brak keystore. Najpierw utwórz portfel (opcja 1).")
        return
    password = getpass("Hasło do keystore: ")
    try:
        mnemonic = load_mnemonic_from_keystore(password)
    except Exception:
        print("Błędne hasło lub uszkodzony keystore.")
        return
    if not os.path.exists(DB_FILE):
        init_db(DB_FILE)
        insert_addresses(mnemonic, count=10, db_path=DB_FILE)
    show_addresses(DB_FILE)
    mnemonic = password = ""


def unlock_and_generate_address():
    if not wallet_exists():
        print("Brak keystore.")
        return
    password = getpass("Hasło do keystore: ")
    try:
        mnemonic = load_mnemonic_from_keystore(password)
    except Exception:
        print("Błędne hasło lub uszkodzony keystore.")
        return
    idx, addr, path = generate_next_address(mnemonic)
    print(f"\nNowy adres dodany:\nIndex: {idx}\nAdres: {addr}\nPath : {path}")
    mnemonic = password = ""


def main_menu():
    clear_screen()
    while True:
        print("\n=== MENU PORTFELA BSV ===")
        print("1. Utwórz nowy portfel")
        print("2. Odblokuj portfel i wyświetl adresy")
        print("3. Wygeneruj nowy adres")
        print("4. Wyjście")

        choice = input("Wybierz opcję (1-4): ").strip()
        clear_screen()

        if choice == "1":
            if wallet_exists():
                print("Keystore już istnieje – nie tworzę drugiego.")
            else:
                create_new_wallet()
        elif choice == "2":
            unlock_and_show_addresses()
        elif choice == "3":
            unlock_and_generate_address()
        elif choice == "4":
            print("Zamykam program.")
            break
        else:
            print("Nieprawidłowy wybór.")

        input("\nNaciśnij ENTER aby wrócić do menu...")
        clear_screen()


if __name__ == "__main__":
    main_menu()

No i w taki sposób dojeżdżamy do końca. Jak to działa?

Uruchamiasz python main.py i widzisz menu. 

Przy pierwszym uruchomieniu wybierasz opcję 1:

• Program generuje 24-słowny mnemonic
• Ustawiasz hasło
• W katalogu pojawiają się dwa pliki: wallet_keystore.json (zaszyfrowany mnemonic) i wallet.db (baza z 10 adresami).

Opcja 2 pyta o hasło, deszyfruje mnemonic w pamięci, odpytuje WhatsOnChain o salda wszystkich adresów i wyświetla tabelkę.

Opcja 3 generuje kolejny adres.

Co dalej?

Portfel umie już: tworzyć adresy, przechowywać je całkiem bezpiecznie i sprawdzać salda. Tego czego mu brakuje to wysyłanie transakcji - czyli temat na część 3. A tam wrócimy do znajomych z poprzednich wpisów: UTXO, budowanie transakcji, podpisywanie i rozgłaszanie. Tylko tym razem bez wklejania kluczy prywatnych na sztywno w kodzie. Tak pisząc to pomyślałem o trochę większym bezpieczeństwie ale to już temat na wpis nr 3.

No nie powiem, mamy w tym wpisie więcej kodu, niż w części pierwszej ale każdy kawałek robi jedną konkretną rzecz. A jak coś przestanie działać - wiesz dokładnie w którym pliku szukać. 

Do następnego! Cześć! 😊

---

 **Gwarantuję Ci niezmienność moich treści**

Hash artykułu:

ID transakcji: sprawdź OP_RETURN i porównaj jego hash