„Krótko: w tej części portfela Bitcoin SV zamieniam PBKDF2 na Argon2id (odporność na ataki GPU), dodaję obsługę UTXO, checksum HMAC do keystore, bezpieczne czyszczenie mnemonika z pamięci i wysyłanie transakcji z walidacją adresu. To część 3 serii."
W poprzedniej części obiecałem że zajmiemy się bezpieczeństwem i wysyłaniem transakcji. Tym razem pracy było sporo, bo po drodze okazało się że "trochę większe" bezpieczeństwo. to tak naprawdę kilka całkiem konkretnych rzeczy do zmiany. I znowu skończyło się na tym, że siedziałem i dłubałem..
Co się zmieniło względem części 2?
Zanim wrzucę kod, powiem co i dlaczego. W poprzedniej wersji portfela używaliśmy PBKDF2 do wyprowadzania klucza z hasła. To całkiem przyzwoity algorytm, ale ma jedną słabość - jest podatny na ataki GPU. Karta graficzna może liczyć miliony prób na sekundę równolegle a atakujący z porządną kartą graficzną może dość sprawnie przeszukiwać słowniki haseł.
PBKDF2 vs Argon2id: dlaczego to ważne dla bezpieczeństwa portfela
Zamiana jest prosta: Argon2id jest aktualnie złotym standardem. Wymusza nie tylko koszt CPU, ale też koszt pamięci - każda próba odgadnięcia hasła wymaga zajęcia 64 MB RAM. GPU ma setki rdzeni, ale pamięci współdzielonej mu brakuje. Zamiast milionów prób na sekundę, spada do tysięcy. W sumie to na nasze zabawy z portfelem to nawet szkoda zachodu, no ale jak już zmieniłem to zmieniłem :)
Oprócz tego dodałem:
- checksum HMAC do keystore - jeśli ktoś zmodyfikuje plik wallet_keystore.json, program to wykryje zanim spróbuje odszyfrować
- bytearray zamiast str dla mnemonika - możemy go ręcznie wyzerować z pamięci zaraz po użyciu
- uprawnienia do pliku przez icacls - keystore dostępny tylko dla Twojego konta Windows
- UTXO w bazie - portfel pobiera i zapisuje niewydane UTXO, nie tylko salda
- wysyłanie transakcji
- walidację adresu przed wysłaniem - żeby nie stracić satoshi przez literówkę
To co musimy zrobić na początku to zainstalować nową zależność:
pip install argon2-cffi
Funkcje i skrypty z poprzedniego wpisu trzeba będzie podmienić lub dodać nowe, jeśli ich nie masz zapoznaj się w wpisem Cześć 2 - po prostu skopiuj całą zawartość do odpowiednich plików.
crypto.py
To największa zmiana, plik wygląda teraz tak:
import os
import base64
import hmac
import hashlib
import json
import subprocess
from argon2.low_level import hash_secret_raw, Type
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from config import KEYSTORE_FILE
ARGON2_TIME_COST = 3 # liczba iteracji
ARGON2_MEMORY_COST = 65536 # 64 MB pamięci
ARGON2_PARALLELISM = 4 # wątki
ARGON2_HASH_LEN = 32 # długość klucza
def derive_key_from_password(password: str, salt: bytes) -> bytes:
return hash_secret_raw(
secret=password.encode("utf-8"),
salt=salt,
time_cost=ARGON2_TIME_COST,
memory_cost=ARGON2_MEMORY_COST,
parallelism=ARGON2_PARALLELISM,
hash_len=ARGON2_HASH_LEN,
type=Type.ID, # Argon2id
)
def _compute_checksum(keystore_without_checksum: dict, password: str) -> str:
content = json.dumps(keystore_without_checksum, sort_keys=True).encode()
return hmac.new(password.encode(), content, hashlib.sha256).hexdigest()
def encrypt_mnemonic(mnemonic: str, password: str) -> dict:
mnemonic_bytes = bytearray(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, bytes(mnemonic_bytes), None)
# czyścimy mnemonic z pamięci
for i in range(len(mnemonic_bytes)):
mnemonic_bytes[i] = 0
keystore = {
"kdf": "argon2id",
"kdfparams": {
"salt": base64.b64encode(salt).decode(),
"time_cost": ARGON2_TIME_COST,
"memory_cost": ARGON2_MEMORY_COST,
"parallelism": ARGON2_PARALLELISM,
"hash_len": ARGON2_HASH_LEN,
},
"cipher": "aes-256-gcm",
"cipherparams": {
"nonce": base64.b64encode(nonce).decode(),
},
"ciphertext": base64.b64encode(ciphertext).decode(),
}
keystore["checksum"] = _compute_checksum(keystore, password)
return keystore
def decrypt_mnemonic(keystore: dict, password: str) -> str:
stored_checksum = keystore.pop("checksum", None)
expected = _compute_checksum(keystore, password)
keystore["checksum"] = stored_checksum
if not hmac.compare_digest(stored_checksum or "", expected):
raise ValueError("Keystore został zmodyfikowany lub hasło jest błędne.")
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)
mnemonic_bytes = bytearray(aesgcm.decrypt(nonce, ciphertext, None))
mnemonic = mnemonic_bytes.decode("utf-8")
for i in range(len(mnemonic_bytes)):
mnemonic_bytes[i] = 0
return mnemonic
def secure_keystore_permissions():
"""Ogranicza dostęp do pliku keystore tylko dla aktualnego użytkownika."""
username = os.environ.get("USERNAME")
try:
subprocess.run(["icacls", KEYSTORE_FILE, "/inheritance:r"],
check=True, capture_output=True)
subprocess.run(["icacls", KEYSTORE_FILE, "/grant:r", f"{username}:F"],
check=True, capture_output=True)
except Exception as e:
print(f"Uwaga: nie udało się ustalić uprawnień: {e}")
Kilka słów wyjaśnienia bo.. lubię tłumaczyć po co :)
bytearray zamiast str - stringi w Pythonie są niemutowalne. Raz stworzony string może siedzieć w pamięci RAM długo po tym jak przestałeś go używać. bytearray możemy ręcznie wyzerować - i tak robimy zaraz po użyciu mnemonika.
checksum HMAC - to coś czego nie ma ani Exodus, ani MetaMask. Jeśli ktoś ręcznie zmodyfikuje Twój wallet_keystore.json (np. próbując podstawić własny ciphertext), program wykryje to zanim w ogóle spróbuje odszyfrować.
secure_keystore_permissions() - wywołuje windowsowe icacls i ustawia uprawnienia do pliku tak, żeby miał do niego dostęp tylko Twoje konto. Kliknij prawym na wallet_keystore.json → Właściwości → Zabezpieczenia i zobaczysz różnicę.
db.py - dodajemy obsługę UTXO
W poprzedniej wersji,. tabela utxos była w bazie ale nic do niej nie trafiało. Teraz dodajemy funkcje, które naprawdę z niej korzystają:
def get_address_index(address: str, db_path: str = DB_FILE) -> int | None:
"""Zwraca index dla podanego adresu."""
conn = sqlite3.connect(db_path)
cur = conn.cursor()
cur.execute("SELECT derivation_index FROM addresses WHERE address = ?", (address,))
row = cur.fetchone()
conn.close()
return row[0] if row else None
def save_utxos(utxos: list[dict], db_path: str = DB_FILE):
"""Zapisuje UTXO do bazy."""
conn = sqlite3.connect(db_path)
cur = conn.cursor()
cur.execute("DELETE FROM utxos WHERE spent = 0")
for utxo in utxos:
cur.execute("SELECT id FROM addresses WHERE address = ?", (utxo["address"],))
row = cur.fetchone()
if row is None:
continue
cur.execute("""
INSERT OR IGNORE INTO utxos (txid, vout, value_sats, address_index, spent)
VALUES (?, ?, ?, ?, 0)
""", (utxo["txid"], utxo["vout"], utxo["value"], row[0]))
conn.commit()
conn.close()
def get_all_utxos(db_path: str = DB_FILE) -> list:
"""Zwraca wszystkie niewydane UTXO z adresami."""
conn = sqlite3.connect(db_path)
cur = conn.cursor()
cur.execute("""
SELECT u.txid, u.vout, u.value_sats, a.address, a.path
FROM utxos u
JOIN addresses a ON u.address_index = a.id
WHERE u.spent = 0
ORDER BY u.value_sats DESC
""")
rows = cur.fetchall()
conn.close()
return rows
woc.py - pobieranie UTXO
WhatsOnChain ma endpoint dla sald (do 20 adresów naraz), ale dla UTXO trzeba odpytywać każdy adres osobno. Dodajemy dwie rzeczy - pobieranie UTXO i pobieranie hex transakcji (potrzebny do podpisania):
def fetch_tx_hex(txid: str) -> str:
url = f"https://api.whatsonchain.com/v1/bsv/{NETWORK}/tx/{txid}/hex"
response = requests.get(url)
response.raise_for_status()
return response.text.strip()
def fetch_utxos_for_addresses(addresses: list[str]) -> list[dict]:
all_utxos = []
total = len(addresses)
print(Fore.YELLOW + "\n Pobieranie UTXO...")
for i, address in enumerate(addresses):
percent = int(((i + 1) / total) * 100)
bar_length = 30 # pasek postępu :)
filled = int(bar_length * (i + 1) / total)
bar = "█" * filled + "-" * (bar_length - filled)
print(Fore.GREEN + f"\r|{bar}| {percent}% [{i+1}/{total}]",
end="", flush=True)
try:
url = f"https://api.whatsonchain.com/v1/bsv/{NETWORK}/address/{address}/unspent/all"
response = requests.get(url)
response.raise_for_status()
data = response.json()
for utxo in data.get("result", []):
all_utxos.append({
"txid": utxo["tx_hash"],
"vout": utxo["tx_pos"],
"value": utxo["value"],
"address": address,
})
except Exception as e:
print(Fore.RED + f"\nBłąd pobierania UTXO dla {address}: {e}")
time.sleep(0.2)
print(Fore.GREEN + "\n Pobieranie zakończone\n")
return all_utxos
ui.py - UTXO w tej samej tabeli co adresy
Poprzednia wersja pokazywała tylko salda. Teraz przy wyświetlaniu adresów od razu pobieramy UTXO w tle i pokazujemy je w tej samej linii:
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 = {}
# pobieramy UTXO tylko dla adresów z saldem
active = [addr for addr in addresses if balances.get(addr, 0) > 0]
if active:
try:
utxos = fetch_utxos_for_addresses(active)
save_utxos(utxos, db_path)
except Exception as e:
print(f"Błąd zapisywania UTXO: {e}")
utxos_by_address = {}
for txid, vout, value_sats, address, path in get_all_utxos(db_path):
if address not in utxos_by_address:
utxos_by_address[address] = []
utxos_by_address[address].append(f"{txid[:16]}... ({value_sats} sat)")
print(f"\n{'#':<5 aldo="" dres="" style="color: #c9d1d9;" yp="">12} UTXO")
print("-" * 110)
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}"
if sats > 0 else f"{bsv:>12.8f}"
)
utxo_list = utxos_by_address.get(addr, [])
first_utxo = utxo_list[0] if utxo_list else "-"
print(f"{idx:<5 addr:="" balance_str="" ch_str:="" extra_utxo="" f="" first_utxo="" for="" in="" label_str="" ore.cyan="" print="" tyle.reset_all="" utxo_list="">66}{Fore.CYAN}{extra_utxo}{Style.RESET_ALL}")
print("-" * 110)
print(f"{'RAZEM':<54 e8:="" total="">12.8f} BSV")
hd.py - klucz prywatny
Dodajemy funkcję która wyprowadza klucz prywatny dla konkretnego indeksu. Klucz żyje tylko wewnątrz tej funkcji - zaraz po użyciu do podpisania transakcji go nie ma:
from bsv import PrivateKey
def get_private_key_for_index(mnemonic: str, derivation_index: int) -> PrivateKey:
keys = bip44_derive_xprvs_from_mnemonic(
mnemonic,
derivation_index,
derivation_index + 1,
path=BIP44_DERIVATION_PATH,
change=0,
network="testnet",
)
xprv = keys[0]
return PrivateKey(xprv.private_key().wif())
main.py - wysyłanie transakcji i walidacja adresu
No i tu zrobiło się ciekawiej. Dodajemy dwie rzeczy: walidację adresu BSV przed wysłaniem i samą funkcję wysyłającą.
Walidacja adresu:
def validate_address(address: str) -> bool:
try:
from bsv.base58 import base58check_decode
decoded = base58check_decode(address)
if len(decoded) != 21:
return False
prefix = decoded[0]
if NETWORK == "test":
return prefix == 0x6f # adresy m... lub n...
else:
return prefix == 0x00 # adresy 1...
except Exception:
return False
Adresy BSV używają base58check - zakodowany adres zawiera prefix który mówi czy to mainnet czy testnet. 0x6f to testnet (adresy zaczynające się od m lub n), 0x00 to mainnet (adresy od 1). Dzięki temu nie wyślesz przypadkowo testnetowych satoshi na adres mainnetowy i odwrotnie.
cd main.py - Wysyłanie transakcji
async def _broadcast(inputs_data, output_address, change_address, mnemonic):
tx = Transaction()
for item in inputs_data:
tx_hex = fetch_tx_hex(item["txid"])
private_key = get_private_key_for_index(mnemonic, item["derivation_index"])
tx.add_input(TransactionInput(
source_transaction=Transaction.from_hex(tx_hex),
source_output_index=item["vout"],
unlocking_script_template=P2PKH().unlock(private_key),
))
private_key = None # czyścimy zaraz po użyciu
tx.add_output(TransactionOutput(
locking_script=P2PKH().lock(output_address),
satoshis=inputs_data[0]["amount_to_send"],
))
tx.add_output(TransactionOutput(
locking_script=P2PKH().lock(change_address),
change=True, # biblioteka sama wyliczy resztę po odjęciu fee
))
tx.fee()
tx.sign()
response = await tx.broadcast(WhatsOnChainBroadcaster(NETWORK))
print(f"\nStatus: {response.status}")
print(f"TXID: {tx.txid()}")
def send_transaction():
if not wallet_exists():
print("Brak portfela.")
return
utxos = get_all_utxos(DB_FILE)
if not utxos:
print("Brak UTXO w bazie. Użyj opcji 2 żeby odświeżyć.")
return
print("\nDostępne UTXO:")
total_available = sum(u[2] for u in utxos)
for txid, vout, value_sats, address, path in utxos:
print(f" {txid[:20]}... | {value_sats} sat | {address}")
print(f"Łącznie dostępne: {total_available} sat\n")
output_address = input("Adres odbiorcy: ").strip()
if not validate_address(output_address):
print("Nieprawidłowy adres BSV. Sprawdź czy dobrze skopiowałeś.")
return
try:
amount_sats = int(input("Kwota do wysłania (satoshi): ").strip())
except ValueError:
print("Nieprawidłowa kwota.")
return
if amount_sats <= 0 or amount_sats > total_available:
print("Nieprawidłowa kwota lub za mało środków.")
return
# dobieramy UTXO od największego — mniej wejść = mniejsze fee
utxos_sorted = sorted(utxos, key=lambda u: u[2], reverse=True)
selected = []
selected_total = 0
for txid, vout, value_sats, address, path in utxos_sorted:
selected.append({"txid": txid, "vout": vout, "value_sats": value_sats,
"address": address, "amount_to_send": amount_sats})
selected_total += value_sats
if selected_total >= amount_sats:
break
print(f"\nWybrano {len(selected)} UTXO pokrywających {selected_total} sat.")
password = getpass("Hasło do keystore: ")
try:
mnemonic = load_mnemonic_from_keystore(password)
except Exception:
print("Błędne hasło.")
return
finally:
password = ""
inputs_data = []
for utxo in selected:
idx = get_address_index(utxo["address"], DB_FILE)
if idx is None:
print(f"Nie znaleziono indeksu dla {utxo['address']}")
mnemonic = ""
return
inputs_data.append({"txid": utxo["txid"], "vout": utxo["vout"],
"derivation_index": idx, "amount_to_send": amount_sats})
# adres reszty — kolejny wolny adres w naszym portfelu
next_index = get_max_derivation_index(change=0, db_path=DB_FILE) + 1
change_keys = bip44_derive_xprvs_from_mnemonic(
mnemonic, next_index, next_index + 1,
path=BIP44_DERIVATION_PATH, change=0, network="testnet",
)
change_address = change_keys[0].address()
insert_address(next_index, 0, change_address,
f"{BIP44_DERIVATION_PATH}/0/{next_index}", DB_FILE)
print(f"\nPodsumowanie:")
print(f" Odbiorca : {output_address}")
print(f" Kwota : {amount_sats} sat ({amount_sats / 1e8:.8f} BSV)")
print(f" Reszta na: {change_address}")
confirm = input("\nWysłać? (tak/nie): ").strip().lower()
if confirm != "tak":
print("Anulowano.")
mnemonic = ""
return
try:
asyncio.run(_broadcast(inputs_data, output_address, change_address, mnemonic))
except Exception as e:
print(f"Błąd podczas wysyłania: {e}")
finally:
mnemonic = ""
Kilka rzeczy wartych uwagi:
Algorytm dobierania UTXO - sortujemy od największego i bierzemy kolejne aż suma pokryje kwotę. Przykład: chcesz wysłać 150 000 sat, masz UTXO 99900 + 99900 + 50000. Algorytm weźmie pierwsze dwa (199 800 sat) i stop. Mniej wejść = mniejsza transakcja = mniejsze fee.
Reszta na nowy adres - zgodnie z tym co pisał Satoshi: jedna transakcja, jeden adres. Reszta nie wraca na adres z którego wysyłasz, tylko idzie na nowy, świeżo wygenerowany.
cd - Menu
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. Odśwież UTXO")
print("5. Wyślij transakcję")
print("6. Wyjście")
Opcja 4 to ręczne odświeżenie UTXO - przydatne gdy wiesz że właśnie dostałeś przelew i nie chcesz czekać do następnego wyświetlenia adresów. Opcja 2 odświeża UTXO automatycznie przy okazji, ale tylko dla adresów które mają saldo - nie ma sensu odpytywać API o UTXO dla pustych adresów.
Test: na początek utwórz portfel, hasło i doładuj sobie jakieś testowe BSV - tutaj opisałem jak skorzystać z kranu.
2 - odblokowanie portfela i wyświetlenie adresów/sald/utxo
3 - wygenerowanie nowego adresu
5 - wysłanie transakcji/odebranie reszty na nowy adres
Wynik wysyłki:
Działa, ok. Jak to wygląda w porównaniu z popularnymi portfelami?
Pod względem samego szyfrowania mnemonika jest naprawdę blisko - i w kilku miejscach lepiej niż komercyjne rozwiązania. Checksum HMAC to coś czego nie ma żaden z nich.
Gdzie odstajemy? Środowisko uruchomieniowe. Exodus i MetaMask działają w izolowanych procesach z podpisanym kodem. Nasz skrypt .py działa na tym samym Windowsie co przeglądarka. Keylogger na systemie przechwyci hasło zanim trafi do getpass. Tego nie naprawimy kodem - tu pomaga albo dedykowana maszyna wirtualna offline, albo portfel sprzętowy.
Ale na testnet i do nauki? WYSTARCZY. Rozumiesz teraz jak działają HD wallety od środka, wiesz co to UTXO, BIP44, Argon2id i AES-GCM. Większość ludzi używa portfeli nie rozumiejąc nic z tego co się dzieje pod spodem.
Co dalej?
Jest kilka rzeczy, które mógłbym jeszcze dodać - auto-odkrywanie adresów (przy przeniesieniu portfela na nowy komputer automatyczne przeskanowanie ile adresów miałeś), historia transakcji i może blokada portfela po czasie wzorowana na Electrum. Ale.. daje sobie z tym spokój.. Chciałem zrobić portfel to zrobiłem :)
Cały kod umieściłem w repozytorium.
Cały kod umieściłem w repozytorium.
Cześć!
Płacę za czytanie mojego bloga.
Dowiedz się więcej.
**Gwarantuję Ci niezmienność moich treści**
Hash artykułu:
ID transakcji: sprawdź OP_RETURN [tag Relax4000] i porównaj jego hash
Komentarze
Prześlij komentarz