aljaz/nutshell
1
0
Fork 0
mirror of https://github.com/aljazceru/nutshell.git synced 2025-12-21 19:14:19 +01:00
Code Issues Packages Projects Releases Wiki Activity

isort

Browse Source
This commit is contained in:
callebtc
2022-10-02 19:27:14 +02:00
parent a6e655e80a
commit af54161cb3
10 changed files with 179 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
cashu/core/base.py
View File

@@ -4,11 +4,16 @@ from typing import List
from pydantic import BaseModel
class P2SHScript(BaseModel):
script: str
signature: str
class Proof(BaseModel):
amount: int
secret: str = ""
C: str
script: str = ""
script: P2SHScript = None
reserved: bool = False # whether this proof is reserved for sending
send_id: str = "" # unique ID of send attempt
time_created: str = ""

3
cashu/core/helpers.py
View File

@@ -1,7 +1,8 @@
import asyncio
from functools import partial, wraps
from cashu.core.settings import LIGHTNING_FEE_PERCENT, LIGHTNING_RESERVE_FEE_MIN
from cashu.core.settings import (LIGHTNING_FEE_PERCENT,
LIGHTNING_RESERVE_FEE_MIN)
def async_wrap(func):

120
cashu/core/script.py
View File

@@ -1,49 +1,38 @@
import hashlib
import base64
import hashlib
import random
COIN = 100_000_000
TXID = "bff785da9f8169f49be92fa95e31f0890c385bfb1bd24d6b94d7900057c617ae"
SEED = b"__not__used"
from bitcoin.core import (
lx,
COutPoint,
CMutableTxOut,
CMutableTxIn,
CMutableTransaction,
)
from bitcoin.core import (CMutableTxIn, CMutableTxOut, COutPoint, CTransaction,
lx)
from bitcoin.core.script import *
from bitcoin.core.scripteval import VerifyScript
from bitcoin.wallet import CBitcoinAddress, CBitcoinSecret
from bitcoin.core.scripteval import (SCRIPT_VERIFY_P2SH, EvalScriptError,
VerifyScript, VerifyScriptError)
from bitcoin.wallet import CBitcoinSecret, P2SHBitcoinAddress
def step0_carol_privkey():
"""Private key"""
h = hashlib.sha256(b"correct horse battery staple").digest()
# h = hashlib.sha256(SEED).digest()
h = hashlib.sha256(str(random.getrandbits(256)).encode()).digest()
seckey = CBitcoinSecret.from_secret_bytes(h)
return seckey
def step0_carolt_checksig_redeemscrip(carol_pubkey):
def step0_carol_checksig_redeemscrip(carol_pubkey):
"""Create script"""
txin_redeemScript = CScript([carol_pubkey, OP_CHECKSIG])
# txin_redeemScript = CScript(
# [
# 3,
# 3,
# OP_LESSTHANOREQUAL,
# OP_VERIFY,
# ]
# )
# txin_redeemScript = CScript([-123, OP_CHECKLOCKTIMEVERIFY])
# txin_redeemScript = CScript([3, 3, OP_LESSTHAN, OP_VERIFY])
return txin_redeemScript
def step1_carol_create_p2sh_address(txin_redeemScript):
"""Create address (serialized scriptPubKey) to share with Alice"""
# print("Script:", b2x(txin_redeemScript))
# returns [OP_HASH160, bitcointx.core.Hash160(self), OP_EQUAL]
txin_scriptPubKey = txin_redeemScript.to_p2sh_scriptPubKey()
txin_p2sh_address = CBitcoinAddress.from_scriptPubKey(txin_scriptPubKey)
# print("Pay to:", str(txin_p2sh_address))
txin_p2sh_address = P2SHBitcoinAddress.from_redeemScript(txin_redeemScript)
return txin_p2sh_address
@@ -54,49 +43,81 @@ def step1_bob_carol_create_tx(txin_p2sh_address):
txin = CMutableTxIn(COutPoint(txid, vout))
txout = CMutableTxOut(
int(0.0005 * COIN),
CBitcoinAddress(str(txin_p2sh_address)).to_scriptPubKey(),
P2SHBitcoinAddress(str(txin_p2sh_address)).to_scriptPubKey(),
)
tx = CMutableTransaction([txin], [txout])
tx = CTransaction([txin], [txout])
return tx, txin
def step2_carol_sign_tx(txin_redeemScript):
def step2_carol_sign_tx(txin_redeemScript, privatekey):
"""Sign transaction with private key"""
seckey = step0_carol_privkey()
txin_p2sh_address = step1_carol_create_p2sh_address(txin_redeemScript)
tx, txin = step1_bob_carol_create_tx(txin_p2sh_address)
sighash = SignatureHash(txin_redeemScript, tx, 0, SIGHASH_ALL)
sig = seckey.sign(sighash) + bytes([SIGHASH_ALL])
sig = privatekey.sign(sighash) + bytes([SIGHASH_ALL])
txin.scriptSig = CScript([sig, txin_redeemScript])
return txin
def step3_bob_verify_script(txin_signature, txin_redeemScript):
def step3_bob_verify_script(txin_signature, txin_redeemScript, tx):
txin_scriptPubKey = txin_redeemScript.to_p2sh_scriptPubKey()
try:
VerifyScript(txin_signature, txin_scriptPubKey, tx, 0)
VerifyScript(
txin_signature, txin_scriptPubKey, tx, 0, flags=[SCRIPT_VERIFY_P2SH]
)
return True
except VerifyScriptError as e:
print("Could not verify script:", e)
except EvalScriptError as e:
print("Script did not evaluate:", e)
print(f"Script: {txin_scriptPubKey.__repr__()}")
except Exception as e:
print(e)
return False
return False
def verify_script(txin_redeemScript_b64, txin_signature_b64):
txin_redeemScript = CScript(base64.urlsafe_b64decode(txin_redeemScript_b64))
print("Redeem script:", txin_redeemScript.__repr__())
# txin_redeemScript = CScript([2, 3, OP_LESSTHAN, OP_VERIFY])
txin_signature = CScript(value=base64.urlsafe_b64decode(txin_signature_b64))
txin_p2sh_address = step1_carol_create_p2sh_address(txin_redeemScript)
print(f"Bob recreates secret: P2SH:{txin_p2sh_address}")
# MINT checks that P2SH:txin_p2sh_address has not been spent yet
# ...
tx, _ = step1_bob_carol_create_tx(txin_p2sh_address)
print(
f"Bob verifies:\nscript: {txin_redeemScript_b64}\nsignature: {txin_signature_b64}\n"
)
script_valid = step3_bob_verify_script(txin_signature, txin_redeemScript, tx)
# MINT redeems tokens and stores P2SH:txin_p2sh_address
# ...
if script_valid:
print("Successfull.")
else:
print("Error.")
return txin_p2sh_address, script_valid
# simple test case
if __name__ == "__main__":
# https://github.com/romanz/python-bitcointx/blob/master/examples/spend-p2sh-txout.py
# CAROL shares txin_p2sh_address with ALICE:
# ---------
# CAROL defines scripthash and ALICE mints them
txin_redeemScript = step0_carolt_checksig_redeemscrip(step0_carol_privkey().pub)
alice_privkey = step0_carol_privkey()
txin_redeemScript = step0_carol_checksig_redeemscrip(alice_privkey.pub)
print("Script:", txin_redeemScript.__repr__())
txin_p2sh_address = step1_carol_create_p2sh_address(txin_redeemScript)
print(f"Carol sends Alice secret = P2SH:{txin_p2sh_address}")
print("")
# ---------
# ALICE: mint tokens with secret SCRIPT:txin_p2sh_address
# ALICE: mint tokens with secret P2SH:txin_p2sh_address
print(f"Alice mints tokens with secret = P2SH:{txin_p2sh_address}")
print("")
# ...
@@ -105,13 +126,13 @@ if __name__ == "__main__":
# CAROL redeems with MINT
# CAROL PRODUCES txin_redeemScript and txin_signature to send to MINT
txin_redeemScript = step0_carolt_checksig_redeemscrip(step0_carol_privkey().pub)
txin_signature = step2_carol_sign_tx(txin_redeemScript).scriptSig
txin_redeemScript = step0_carol_checksig_redeemscrip(alice_privkey.pub)
txin_signature = step2_carol_sign_tx(txin_redeemScript, alice_privkey).scriptSig
txin_redeemScript_b64 = base64.urlsafe_b64encode(txin_redeemScript).decode()
txin_signature_b64 = base64.urlsafe_b64encode(txin_signature).decode()
print(
f"Carol to Bob:\nscript: {txin_redeemScript_b64}\nsignature: {txin_signature_b64}\n"
f"Carol to Bob:\nscript: {txin_redeemScript.__repr__()}\nscript: {txin_redeemScript_b64}\nsignature: {txin_signature_b64}\n"
)
print("")
# ---------
@@ -119,20 +140,25 @@ if __name__ == "__main__":
# MINT receives txin_redeemScript_b64 and txin_signature_b64 fom CAROL:
txin_redeemScript = CScript(base64.urlsafe_b64decode(txin_redeemScript_b64))
txin_signature = CScript(base64.urlsafe_b64decode(txin_signature_b64))
txin_redeemScript_p2sh = txin_p2sh_address.to_redeemScript()
print("Redeem script:", txin_redeemScript.__repr__())
print("P2SH:", txin_redeemScript_p2sh.__repr__())
# txin_redeemScript = CScript([2, 3, OP_LESSTHAN, OP_VERIFY])
txin_signature = CScript(value=base64.urlsafe_b64decode(txin_signature_b64))
txin_p2sh_address = step1_carol_create_p2sh_address(txin_redeemScript)
print(f"Bob recreates secret: P2SH:{txin_p2sh_address}")
# MINT checks that SCRIPT:txin_p2sh_address has not been spent yet
# MINT checks that P2SH:txin_p2sh_address has not been spent yet
# ...
tx, _ = step1_bob_carol_create_tx(txin_p2sh_address)
print(
f"Bob verifies:\nscript: {txin_redeemScript_b64}\nsignature: {txin_signature_b64}\n"
)
script_valid = step3_bob_verify_script(txin_signature, txin_redeemScript)
# MINT redeems tokens and stores SCRIPT:txin_p2sh_address
script_valid = step3_bob_verify_script(txin_signature, txin_redeemScript, tx)
# MINT redeems tokens and stores P2SH:txin_p2sh_address
# ...
print("Successfull.")
# print("Transaction:", b2x(tx.serialize()))
if script_valid:
print("Successfull.")
else:
print("Error.")

9
cashu/lightning/lnbits.py
View File

@@ -8,13 +8,8 @@ import requests
from cashu.core.settings import LNBITS_ENDPOINT, LNBITS_KEY
from .base import (
InvoiceResponse,
PaymentResponse,
PaymentStatus,
StatusResponse,
Wallet,
)
from .base import (InvoiceResponse, PaymentResponse, PaymentStatus,
StatusResponse, Wallet)
class LNbitsWallet(Wallet):

2
cashu/mint/app.py
View File

@@ -5,7 +5,7 @@ import sys
from fastapi import FastAPI
from loguru import logger
from cashu.core.settings import VERSION, DEBUG
from cashu.core.settings import DEBUG, VERSION
from cashu.lightning import WALLET
from cashu.mint.migrations import m001_initial

69
cashu/mint/ledger.py
View File

@@ -3,24 +3,22 @@ Implementation of https://gist.github.com/phyro/935badc682057f418842c72961cf096c
"""
import hashlib
from inspect import signature
from signal import signal
from typing import List, Set
import cashu.core.b_dhke as b_dhke
from cashu.core.base import BlindedMessage, BlindedSignature, Invoice, Proof
from cashu.core.db import Database
from cashu.core.helpers import fee_reserve
from cashu.core.script import verify_script
from cashu.core.secp import PrivateKey, PublicKey
from cashu.core.settings import LIGHTNING, MAX_ORDER
from cashu.core.split import amount_split
from cashu.lightning import WALLET
from cashu.mint.crud import (
get_lightning_invoice,
get_proofs_used,
invalidate_proof,
store_lightning_invoice,
store_promise,
update_lightning_invoice,
)
from cashu.mint.crud import (get_lightning_invoice, get_proofs_used,
invalidate_proof, store_lightning_invoice,
store_promise, update_lightning_invoice)
class Ledger:
@@ -73,6 +71,11 @@ class Ledger:
"""Checks whether the proof was already spent."""
return not proof.secret in self.proofs_used
def _verify_secret_or_script(self, proof: Proof):
if proof.secret and proof.script:
raise Exception("secret and script present at the same time.")
return True
def _verify_secret_criteria(self, proof: Proof):
if proof.secret is None or proof.secret == "":
raise Exception("no secret in proof.")
@@ -86,23 +89,35 @@ class Ledger:
C = PublicKey(bytes.fromhex(proof.C), raw=True)
return b_dhke.verify(secret_key, C, proof.secret)
def _verify_script(self, proof: Proof):
print(f"secret: {proof.secret}")
print(f"script: {proof.script}")
print(
f"script_hash: {hashlib.sha256(proof.script.encode('utf-8')).hexdigest()}"
)
if len(proof.secret.split("SCRIPT:")) != 2:
return True
if len(proof.script) < 16:
raise Exception("Script error: not long enough.")
def _verify_script(self, idx: int, proof: Proof):
if (
hashlib.sha256(proof.script.encode("utf-8")).hexdigest()
!= proof.secret.split("SCRIPT:")[1]
proof.script is None
or proof.script.script is None
or proof.script.signature is None
):
raise Exception("Script error: script hash not valid.")
print(f"Script {proof.script} valid.")
return True
if len(proof.secret.split("P2SH:")) == 2:
# secret indicates a script but no script is present
return False
else:
# secret indicates no script, so treat script as valid
return True
txin_p2sh_address, valid = verify_script(
proof.script.script, proof.script.signature
)
# if len(proof.script) < 16:
# raise Exception("Script error: not long enough.")
# if (
# hashlib.sha256(proof.script.encode("utf-8")).hexdigest()
# != proof.secret.split("P2SH:")[1]
# ):
# raise Exception("Script error: script hash not valid.")
print(
f"Script {proof.script.script.__repr__()} {'valid' if valid else 'invalid'}."
)
if valid:
print(f"{idx}:P2SH:{txin_p2sh_address}")
proof.secret = f"{idx}:P2SH:{txin_p2sh_address}"
return valid
def _verify_outputs(
self, total: int, amount: int, output_data: List[BlindedMessage]
@@ -255,6 +270,11 @@ class Ledger:
"""Consumes proofs and prepares new promises based on the amount split."""
total = sum([p.amount for p in proofs])
# if not all([self._verify_secret_or_script(p) for p in proofs]):
# raise Exception("can't use secret and script at the same time.")
# Verify scripts
if not all([self._verify_script(i, p) for i, p in enumerate(proofs)]):
raise Exception("could not verify scripts.")
# verify that amount is kosher
self._verify_split_amount(amount)
# verify overspending attempt
@@ -272,9 +292,6 @@ class Ledger:
# Verify proofs
if not all([self._verify_proof(p) for p in proofs]):
raise Exception("could not verify proofs.")
# Verify scripts
if not all([self._verify_script(p) for p in proofs]):
raise Exception("could not verify scripts.")
# Mark proofs as used and prepare new promises
await self._invalidate_proofs(proofs)

3
cashu/mint/router.py
View File

@@ -3,7 +3,8 @@ from typing import Union
from fastapi import APIRouter
from secp256k1 import PublicKey
from cashu.core.base import CheckPayload, MeltPayload, MintPayloads, SplitPayload
from cashu.core.base import (CheckPayload, MeltPayload, MintPayloads,
SplitPayload)
from cashu.mint import ledger
router: APIRouter = APIRouter()

35
cashu/wallet/cli.py
View File

@@ -5,7 +5,6 @@ import base64
import json
import math
import sys
from loguru import logger
from datetime import datetime
from functools import wraps
from itertools import groupby
@@ -13,12 +12,14 @@ from operator import itemgetter
import click
from bech32 import bech32_decode, bech32_encode, convertbits
from loguru import logger
import cashu.core.bolt11 as bolt11
from cashu.core.base import Proof
from cashu.core.bolt11 import Invoice
from cashu.core.helpers import fee_reserve
from cashu.core.migrations import migrate_databases
from cashu.core.script import *
from cashu.core.settings import CASHU_DIR, DEBUG, LIGHTNING, MINT_URL, VERSION
from cashu.wallet import migrations
from cashu.wallet.crud import get_reserved_proofs
@@ -123,19 +124,43 @@ async def send(ctx, amount: int, secret: str):
@cli.command("receive", help="Receive tokens.")
@click.argument("token", type=str)
@click.option("--secret", "-s", default="", help="Token spending condition.", type=str)
@click.option("--script", default=None, help="Token unlock script.", type=str)
@click.option("--secret", "-s", default=None, help="Token secret.", type=str)
@click.option("--script", default=None, help="Unlock script.", type=str)
@click.option("--signature", default=None, help="Script signature.", type=str)
@click.pass_context
@coro
async def receive(ctx, token: str, secret: str, script: str):
async def receive(ctx, token: str, secret: str, script: str, signature: str):
wallet: Wallet = ctx.obj["WALLET"]
wallet.load_mint()
wallet.status()
proofs = [Proof.from_dict(p) for p in json.loads(base64.urlsafe_b64decode(token))]
_, _ = await wallet.redeem(proofs, secret, script)
_, _ = await wallet.redeem(
proofs, snd_secret=secret, snd_script=script, snd_siganture=signature
)
wallet.status()
@cli.command("address", help="Generate receiving address.")
@click.pass_context
@coro
async def address(ctx):
alice_privkey = step0_carol_privkey()
txin_redeemScript = step0_carol_checksig_redeemscrip(alice_privkey.pub)
txin_p2sh_address = step1_carol_create_p2sh_address(txin_redeemScript)
print("Redeem script:", txin_redeemScript.__repr__())
print(f"Receiving address: P2SH:{txin_p2sh_address}")
print("")
print(f"Send via command:\ncashu send <amount> --secret P2SH:{txin_p2sh_address}")
print("")
txin_signature = step2_carol_sign_tx(txin_redeemScript, alice_privkey).scriptSig
txin_redeemScript_b64 = base64.urlsafe_b64encode(txin_redeemScript).decode()
txin_signature_b64 = base64.urlsafe_b64encode(txin_signature).decode()
print(
f"Receive via command:\ncashu receive <token> --secret P2SH:{txin_p2sh_address} --script {txin_redeemScript_b64} --signature {txin_signature_b64}"
)
@cli.command("burn", help="Burn spent tokens.")
@click.argument("token", required=False, type=str)
@click.option("--all", "-a", default=False, is_flag=True, help="Burn all spent tokens.")

47
cashu/wallet/wallet.py
View File

@@ -3,31 +3,20 @@ import json
import secrets as scrts
import uuid
from typing import List
from loguru import logger
import requests
from loguru import logger
import cashu.core.b_dhke as b_dhke
from cashu.core.base import (
BlindedMessage,
BlindedSignature,
CheckPayload,
MeltPayload,
MintPayloads,
Proof,
SplitPayload,
)
from cashu.core.base import (BlindedMessage, BlindedSignature, CheckPayload,
MeltPayload, MintPayloads, P2SHScript, Proof,
SplitPayload)
from cashu.core.db import Database
from cashu.core.secp import PublicKey
from cashu.core.settings import DEBUG
from cashu.core.split import amount_split
from cashu.wallet.crud import (
get_proofs,
invalidate_proof,
store_proof,
update_proof_reserved,
secret_used,
)
from cashu.wallet.crud import (get_proofs, invalidate_proof, secret_used,
store_proof, update_proof_reserved)
class LedgerAPI:
@@ -132,9 +121,7 @@ class LedgerAPI:
promises = [BlindedSignature.from_dict(p) for p in promises_list]
return self._construct_proofs(promises, secrets, rs)
async def split(
self, proofs, amount, snd_secret: str = None, has_script: bool = False
):
async def split(self, proofs, amount, snd_secret: str = None):
"""Consume proofs and create new promises based on amount split.
If snd_secret is None, random secrets will be generated for the tokens to keep (fst_outputs)
and the promises to send (snd_outputs).
@@ -182,7 +169,7 @@ class LedgerAPI:
else:
raise Exception("Unkown mint error.")
if "error" in promises_dict:
raise Exception("Error: {}".format(promises_dict["error"]))
raise Exception("Mint Error: {}".format(promises_dict["error"]))
promises_fst = [BlindedSignature.from_dict(p) for p in promises_dict["fst"]]
promises_snd = [BlindedSignature.from_dict(p) for p in promises_dict["snd"]]
# Construct proofs from promises (i.e., unblind signatures)
@@ -245,7 +232,11 @@ class Wallet(LedgerAPI):
return proofs
async def redeem(
self, proofs: List[Proof], snd_secret: str = None, snd_script: str = None
self,
proofs: List[Proof],
snd_secret: str = None,
snd_script: str = None,
snd_siganture: str = None,
):
if snd_secret:
logger.debug(f"Redeption secret: {snd_secret}")
@@ -254,13 +245,15 @@ class Wallet(LedgerAPI):
# overload proofs with custom secrets for redemption
for p, s in zip(proofs, snd_secrets):
p.secret = s
if snd_script:
has_script = False
if snd_script and snd_siganture:
has_script = True
logger.debug(f"Unlock script: {snd_script}")
# overload proofs with unlock script
for p in proofs:
p.script = snd_script
p.script = P2SHScript(script=snd_script, signature=snd_siganture)
return await self.split(
proofs, sum(p["amount"] for p in proofs), has_script=snd_script is not None
proofs, sum(p["amount"] for p in proofs), has_script=has_script
)
async def split(
@@ -271,9 +264,7 @@ class Wallet(LedgerAPI):
has_script: bool = False,
):
assert len(proofs) > 0, ValueError("no proofs provided.")
fst_proofs, snd_proofs = await super().split(
proofs, amount, snd_secret, has_script
)
fst_proofs, snd_proofs = await super().split(proofs, amount, snd_secret)
if len(fst_proofs) == 0 and len(snd_proofs) == 0:
raise Exception("received no splits.")
used_secrets = [p["secret"] for p in proofs]

1
tests/test_wallet.py
View File

@@ -1,5 +1,6 @@
import time
from re import S
from cashu.core.helpers import async_unwrap
from cashu.core.migrations import migrate_databases
from cashu.wallet import migrations