Skip to content

Commit

Permalink
Change TxIn and TxOut to dataclass
Browse files Browse the repository at this point in the history
  • Loading branch information
@giacomocaironi
giacomocaironi committed Jun 16, 2020
1 parent 34703b5 commit 51ce804
Show file tree
Hide file tree
Showing 4 changed files with 138 additions and 135 deletions.
71 changes: 35 additions & 36 deletions btclib/tests/test_transactions.py
View file
Expand Up @@ -22,18 +22,18 @@ def test_coinbase_1():

block_1_coinbase_output_bytes = b'\x00\xf2\x05*\x01\x00\x00\x00CA\x04\x96\xb58\xe8SQ\x9crj,\x91\xe6\x1e\xc1\x16\x00\xae\x13\x90\x81:b|f\xfb\x8b\xe7\x94{\xe6<R\xdau\x897\x95\x15\xd4\xe0\xa6\x04\xf8\x14\x17\x81\xe6"\x94r\x11f\xbfb\x1es\xa8,\xbf#B\xc8X\xee\xac'

transaction_in = tx_in.deserialize(block_1_coinbase_input_bytes)
transaction_out = tx_out.deserialize(block_1_coinbase_output_bytes)
transaction_in = tx_in.TxIn.deserialize(block_1_coinbase_input_bytes)
transaction_out = tx_out.TxOut.deserialize(block_1_coinbase_output_bytes)
transaction = tx.Tx.deserialize(block_1_coinbase_bytes)

assert transaction.vin[0]["scriptSig"] == transaction_in["scriptSig"]
assert transaction.vout[0]["scriptPubKey"] == transaction_out["scriptPubKey"]
assert transaction.vin[0].scriptSig == transaction_in.scriptSig
assert transaction.vout[0].scriptPubKey == transaction_out.scriptPubKey

assert transaction.serialize() == block_1_coinbase_bytes
assert tx_in.serialize(transaction_in) == block_1_coinbase_input_bytes
assert tx_out.serialize(transaction_out) == block_1_coinbase_output_bytes
assert transaction_in.serialize() == block_1_coinbase_input_bytes
assert transaction_out.serialize() == block_1_coinbase_output_bytes

assert transaction.vin[0]["scriptSig"] == []
assert transaction.vin[0].scriptSig == []

assert (
transaction.txid
Expand All @@ -50,8 +50,8 @@ def test_wiki_transaction():

assert len(transaction.vin) == 1
assert len(transaction.vout) == 2
assert transaction.vout[0]["value"] == 5000000
assert transaction.vout[1]["value"] == 3354000000
assert transaction.vout[0].value == 5000000
assert transaction.vout[1].value == 3354000000

assert transaction.txid == transaction.hash_value

Expand Down Expand Up @@ -101,52 +101,51 @@ def test_double_witness():


def test_invalid_tx_in():
transaction_input: tx_in.TxIn = {
"txid": "00" * 31 + "01",
"vout": 256 ** 4 - 1,
"scriptSig": [],
"scriptSigHex": "",
"sequence": 1,
"txinwitness": [],
}
transaction_input = tx_in.TxIn(
txid="00" * 31 + "01",
vout=256 ** 4 - 1,
scriptSig=[],
scriptSigHex="",
sequence=1,
txinwitness=[],
)

with pytest.raises(ValueError):
tx_in.assert_valid(transaction_input)
transaction_input.assert_valid()


def test_invalid_tx_in2():
transaction_input: tx_in.TxIn = {
"txid": "00" * 32,
"vout": 0,
"scriptSig": [],
"scriptSigHex": "",
"sequence": 1,
"txinwitness": [],
}
transaction_input = tx_in.TxIn(
txid="00" * 32,
vout=0,
scriptSig=[],
scriptSigHex="",
sequence=1,
txinwitness=[],
)

with pytest.raises(ValueError):
tx_in.assert_valid(transaction_input)
transaction_input.assert_valid()


def test_invalid_tx_out():
transaction_output: tx_out.TxOut = {"value": -1, "scriptPubKey": ["OP_RETURN"]}
transaction_output = tx_out.TxOut(value=-1, scriptPubKey=["OP_RETURN"])

with pytest.raises(ValueError):
tx_out.assert_valid(transaction_output)
transaction_output.assert_valid()


def test_invalid_tx_out2():
transaction_output: tx_out.TxOut = {
"value": 2099999997690001,
"scriptPubKey": ["OP_RETURN"],
}
transaction_output = tx_out.TxOut(
value=2099999997690001, scriptPubKey=["OP_RETURN"],
)

with pytest.raises(ValueError):
tx_out.assert_valid(transaction_output)
transaction_output.assert_valid()


def test_invalid_tx_out3():
transaction_output: tx_out.TxOut = {"value": 1, "scriptPubKey": []}
transaction_output = tx_out.TxOut(value=1, scriptPubKey=[])

with pytest.raises(ValueError):
tx_out.assert_valid(transaction_output)
transaction_output.assert_valid()
30 changes: 15 additions & 15 deletions btclib/tx.py
View file
Expand Up @@ -15,12 +15,12 @@
https://bitcoin.stackexchange.com/questions/20721/what-is-the-format-of-the-coinbase-transaction
"""

from typing import List, TypeVar
from typing import List, TypeVar, Type
from dataclasses import dataclass

from . import tx_in, tx_out, varint
from . import varint
from .alias import Octets
from .tx_in import TxIn
from .tx_in import TxIn, witness_serialize, witness_deserialize
from .tx_out import TxOut
from .utils import bytes_from_octets, hash256

Expand All @@ -36,7 +36,7 @@ class Tx:
witness_flag: bool

@classmethod
def deserialize(cls, data: Octets) -> _Tx:
def deserialize(cls: Type[_Tx], data: Octets) -> _Tx:

data = bytes_from_octets(data)

Expand All @@ -52,27 +52,27 @@ def deserialize(cls, data: Octets) -> _Tx:
data = data[len(varint.encode(input_count)) :]
vin: List[TxIn] = []
for _ in range(input_count):
tx_input = tx_in.deserialize(data)
tx_input = TxIn.deserialize(data)
vin.append(tx_input)
data = data[len(tx_in.serialize(tx_input)) :]
data = data[len(tx_input.serialize()) :]

output_count = varint.decode(data)
data = data[len(varint.encode(output_count)) :]
vout: List[TxOut] = []
for _ in range(output_count):
tx_output = tx_out.deserialize(data)
tx_output = TxOut.deserialize(data)
vout.append(tx_output)
data = data[len(tx_out.serialize(tx_output)) :]
data = data[len(tx_output.serialize()) :]

if witness_flag:
for tx_input in vin:
witness = tx_in.witness_deserialize(data)
data = data[len(tx_in.witness_serialize(witness)) :]
tx_input["txinwitness"] = witness
witness = witness_deserialize(data)
data = data[len(witness_serialize(witness)) :]
tx_input.txinwitness = witness

locktime = int.from_bytes(data[:4], "little")

tx: Tx = cls(
tx = cls(
version=version,
locktime=locktime,
vin=vin,
Expand All @@ -90,15 +90,15 @@ def serialize(self, include_witness: bool = True) -> bytes:

out += varint.encode(len(self.vin))
for tx_input in self.vin:
out += tx_in.serialize(tx_input)
out += tx_input.serialize()

out += varint.encode(len(self.vout))
for tx_output in self.vout:
out += tx_out.serialize(tx_output)
out += tx_output.serialize()

if self.witness_flag and include_witness:
for tx_input in self.vin:
out += tx_in.witness_serialize(tx_input["txinwitness"])
out += witness_serialize(tx_input.txinwitness)

out += self.locktime.to_bytes(4, "little")
return out
Expand Down
116 changes: 59 additions & 57 deletions btclib/tx_in.py
View file
Expand Up @@ -8,69 +8,78 @@
# No part of btclib including this file, may be copied, modified, propagated,
# or distributed except according to the terms contained in the LICENSE file.

from typing import List, TypedDict
from typing import List, TypeVar, Type
from dataclasses import dataclass

from . import script, varint
from .alias import Octets, Token
from .utils import bytes_from_octets

_TxIn = TypeVar("_TxIn", bound="TxIn")

class TxIn(TypedDict):

@dataclass
class TxIn:
txid: str
vout: int
scriptSig: List[Token]
scriptSigHex: str
sequence: int
txinwitness: List[str]


def deserialize(data: Octets) -> TxIn:

data = bytes_from_octets(data)

txid = data[:32][::-1].hex()
vout = int.from_bytes(data[32:36], "little")
is_coinbase = False
if txid == "00" * 32 and vout == 256 ** 4 - 1:
is_coinbase = True

script_length = varint.decode(data[36:])
data = data[36 + len(varint.encode(script_length)) :]

scriptSig: List[Token] = []
scriptSigHex = ""
if is_coinbase:
scriptSigHex = data[:script_length].hex()
else:
scriptSig = script.decode(data[:script_length])

sequence = int.from_bytes(data[script_length : script_length + 4], "little")
txinwitness: List[str] = []

tx_in: TxIn = {
"txid": txid,
"vout": vout,
"scriptSig": scriptSig,
"scriptSigHex": scriptSigHex,
"sequence": sequence,
"txinwitness": txinwitness,
}

assert_valid(tx_in)
return tx_in


def serialize(tx_in: TxIn) -> bytes:
out = bytes.fromhex(tx_in["txid"])[::-1]
out += tx_in["vout"].to_bytes(4, "little")
if tx_in["txid"] == "00" * 32 and tx_in["vout"] == 256 ** 4 - 1:
script_bytes = bytes.fromhex(tx_in["scriptSigHex"])
else:
script_bytes = script.encode(tx_in["scriptSig"])
out += varint.encode(len(script_bytes))
out += script_bytes
out += tx_in["sequence"].to_bytes(4, "little")
return out
@classmethod
def deserialize(cls: Type[_TxIn], data: Octets) -> _TxIn:

data = bytes_from_octets(data)

txid = data[:32][::-1].hex()
vout = int.from_bytes(data[32:36], "little")
is_coinbase = False
if txid == "00" * 32 and vout == 256 ** 4 - 1:
is_coinbase = True

script_length = varint.decode(data[36:])
data = data[36 + len(varint.encode(script_length)) :]

scriptSig: List[Token] = []
scriptSigHex = ""
if is_coinbase:
scriptSigHex = data[:script_length].hex()
else:
scriptSig = script.decode(data[:script_length])

sequence = int.from_bytes(data[script_length : script_length + 4], "little")
txinwitness: List[str] = []

tx_in = cls(
txid=txid,
vout=vout,
scriptSig=scriptSig,
scriptSigHex=scriptSigHex,
sequence=sequence,
txinwitness=txinwitness,
)

tx_in.assert_valid()
return tx_in

def serialize(self) -> bytes:
out = bytes.fromhex(self.txid)[::-1]
out += self.vout.to_bytes(4, "little")
if self.txid == "00" * 32 and self.vout == 256 ** 4 - 1:
script_bytes = bytes.fromhex(self.scriptSigHex)
else:
script_bytes = script.encode(self.scriptSig)
out += varint.encode(len(script_bytes))
out += script_bytes
out += self.sequence.to_bytes(4, "little")
return out

def assert_valid(self) -> None:
null_txid = "00" * 32
null_vout = 256 ** 4 - 1
if (self.txid == null_txid) ^ (self.vout == null_vout):
raise ValueError("invalid tx_in")


def witness_deserialize(data: Octets) -> List[str]:
Expand Down Expand Up @@ -102,10 +111,3 @@ def witness_serialize(witness: List[str]) -> bytes:
out += witness_bytes

return out


def assert_valid(tx_in: TxIn) -> None:
null_txid = "00" * 32
null_vout = 256 ** 4 - 1
if (tx_in["txid"] == null_txid) ^ (tx_in["vout"] == null_vout):
raise ValueError("invalid tx_in")

0 comments on commit 51ce804

Please sign in to comment.