import struct import hashlib from binascii import hexlify, unhexlify from typing import List, Optional import ecdsa from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.serialization import load_der_public_key from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.asymmetric.utils import Prehashed from cryptography.exceptions import InvalidSignature from torba.client.basetransaction import BaseTransaction, BaseInput, BaseOutput, ReadOnlyList from torba.client.hash import hash160, sha256, Base58 from lbrynet.schema.claim import Claim from lbrynet.schema.url import normalize_name from lbrynet.wallet.account import Account from lbrynet.wallet.script import InputScript, OutputScript class Input(BaseInput): script: InputScript script_class = InputScript class Output(BaseOutput): script: OutputScript script_class = OutputScript __slots__ = 'channel', 'private_key', 'meta' def __init__(self, *args, channel: Optional['Output'] = None, private_key: Optional[str] = None, **kwargs) -> None: super().__init__(*args, **kwargs) self.channel = channel self.private_key = private_key self.meta = {} def update_annotations(self, annotated): super().update_annotations(annotated) self.channel = annotated.channel if annotated else None self.private_key = annotated.private_key if annotated else None def get_fee(self, ledger): name_fee = 0 if self.script.is_claim_name: name_fee = len(self.script.values['claim_name']) * ledger.fee_per_name_char return max(name_fee, super().get_fee(ledger)) @property def is_claim(self) -> bool: return self.script.is_claim_name or self.script.is_update_claim @property def is_support(self) -> bool: return self.script.is_support_claim @property def claim_hash(self) -> bytes: if self.script.is_claim_name: return hash160(self.tx_ref.hash + struct.pack('>I', self.position)) elif self.script.is_update_claim or self.script.is_support_claim: return self.script.values['claim_id'] else: raise ValueError('No claim_id associated.') @property def claim_id(self) -> str: return hexlify(self.claim_hash[::-1]).decode() @property def claim_name(self) -> str: if self.script.is_claim_involved: return self.script.values['claim_name'].decode() raise ValueError('No claim_name associated.') @property def normalized_name(self) -> str: return normalize_name(self.claim_name) @property def claim(self) -> Claim: if self.is_claim: if not isinstance(self.script.values['claim'], Claim): self.script.values['claim'] = Claim.from_bytes(self.script.values['claim']) return self.script.values['claim'] raise ValueError('Only claim name and claim update have the claim payload.') @property def can_decode_claim(self): try: return self.claim except: return False @property def permanent_url(self) -> str: if self.script.is_claim_involved: return f"lbry://{self.claim_name}#{self.claim_id}" raise ValueError('No claim associated.') @property def has_private_key(self): return self.private_key is not None def get_signature_digest(self, ledger): if self.claim.unsigned_payload: pieces = [ Base58.decode(self.get_address(ledger)), self.claim.unsigned_payload, self.claim.signing_channel_hash[::-1] ] else: pieces = [ self.tx_ref.tx.inputs[0].txo_ref.hash, self.claim.signing_channel_hash, self.claim.to_message_bytes() ] return sha256(b''.join(pieces)) def get_encoded_signature(self): signature = hexlify(self.claim.signature) r = int(signature[:int(len(signature)/2)], 16) s = int(signature[int(len(signature)/2):], 16) return ecdsa.util.sigencode_der(r, s, len(signature)*4) @staticmethod def is_signature_valid(encoded_signature, signature_digest, public_key_bytes): try: public_key = load_der_public_key(public_key_bytes, default_backend()) public_key.verify(encoded_signature, signature_digest, ec.ECDSA(Prehashed(hashes.SHA256()))) return True except (ValueError, InvalidSignature): pass return False def is_signed_by(self, channel: 'Output', ledger=None): return self.is_signature_valid( self.get_encoded_signature(), self.get_signature_digest(ledger), channel.claim.channel.public_key_bytes ) def sign(self, channel: 'Output', first_input_id=None): self.channel = channel self.claim.signing_channel_hash = channel.claim_hash digest = sha256(b''.join([ first_input_id or self.tx_ref.tx.inputs[0].txo_ref.hash, self.claim.signing_channel_hash, self.claim.to_message_bytes() ])) self.claim.signature = channel.private_key.sign_digest_deterministic(digest, hashfunc=hashlib.sha256) self.script.generate() def clear_signature(self): self.channel = None self.claim.clear_signature() def generate_channel_private_key(self): self.private_key = ecdsa.SigningKey.generate(curve=ecdsa.SECP256k1, hashfunc=hashlib.sha256) self.claim.channel.public_key_bytes = self.private_key.get_verifying_key().to_der() self.script.generate() return self.private_key def is_channel_private_key(self, private_key): return self.claim.channel.public_key_bytes == private_key.get_verifying_key().to_der() @classmethod def pay_claim_name_pubkey_hash( cls, amount: int, claim_name: str, claim: Claim, pubkey_hash: bytes) -> 'Output': script = cls.script_class.pay_claim_name_pubkey_hash( claim_name.encode(), claim, pubkey_hash) txo = cls(amount, script) return txo @classmethod def pay_update_claim_pubkey_hash( cls, amount: int, claim_name: str, claim_id: str, claim: Claim, pubkey_hash: bytes) -> 'Output': script = cls.script_class.pay_update_claim_pubkey_hash( claim_name.encode(), unhexlify(claim_id)[::-1], claim, pubkey_hash) txo = cls(amount, script) return txo @classmethod def pay_support_pubkey_hash(cls, amount: int, claim_name: str, claim_id: str, pubkey_hash: bytes) -> 'Output': script = cls.script_class.pay_support_pubkey_hash(claim_name.encode(), unhexlify(claim_id)[::-1], pubkey_hash) return cls(amount, script) @classmethod def purchase_claim_pubkey_hash(cls, amount: int, claim_id: str, pubkey_hash: bytes) -> 'Output': script = cls.script_class.purchase_claim_pubkey_hash(unhexlify(claim_id)[::-1], pubkey_hash) return cls(amount, script) class Transaction(BaseTransaction): input_class = Input output_class = Output outputs: ReadOnlyList[Output] inputs: ReadOnlyList[Input] @classmethod def pay(cls, amount: int, address: bytes, funding_accounts: List[Account], change_account: Account): ledger = cls.ensure_all_have_same_ledger(funding_accounts, change_account) output = Output.pay_pubkey_hash(amount, ledger.address_to_hash160(address)) return cls.create([], [output], funding_accounts, change_account) @classmethod def claim_create( cls, name: str, claim: Claim, amount: int, holding_address: str, funding_accounts: List[Account], change_account: Account, signing_channel: Output = None): ledger = cls.ensure_all_have_same_ledger(funding_accounts, change_account) claim_output = Output.pay_claim_name_pubkey_hash( amount, name, claim, ledger.address_to_hash160(holding_address) ) if signing_channel is not None: claim_output.sign(signing_channel, b'placeholder txid:nout') return cls.create([], [claim_output], funding_accounts, change_account, sign=False) @classmethod def claim_update( cls, previous_claim: Output, claim: Claim, amount: int, holding_address: str, funding_accounts: List[Account], change_account: Account, signing_channel: Output = None): ledger = cls.ensure_all_have_same_ledger(funding_accounts, change_account) updated_claim = Output.pay_update_claim_pubkey_hash( amount, previous_claim.claim_name, previous_claim.claim_id, claim, ledger.address_to_hash160(holding_address) ) if signing_channel is not None: updated_claim.sign(signing_channel, b'placeholder txid:nout') else: updated_claim.clear_signature() return cls.create( [Input.spend(previous_claim)], [updated_claim], funding_accounts, change_account, sign=False ) @classmethod def support(cls, claim_name: str, claim_id: str, amount: int, holding_address: str, funding_accounts: List[Account], change_account: Account): ledger = cls.ensure_all_have_same_ledger(funding_accounts, change_account) support_output = Output.pay_support_pubkey_hash( amount, claim_name, claim_id, ledger.address_to_hash160(holding_address) ) return cls.create([], [support_output], funding_accounts, change_account) @classmethod def purchase(cls, claim: Output, amount: int, merchant_address: bytes, funding_accounts: List[Account], change_account: Account): ledger = cls.ensure_all_have_same_ledger(funding_accounts, change_account) claim_output = Output.purchase_claim_pubkey_hash( amount, claim.claim_id, ledger.address_to_hash160(merchant_address) ) return cls.create([], [claim_output], funding_accounts, change_account) @property def my_inputs(self): for txi in self.inputs: if txi.txo_ref.txo is not None and txi.txo_ref.txo.is_my_account: yield txi def _filter_my_outputs(self, f): for txo in self.outputs: if txo.is_my_account and f(txo.script): yield txo def _filter_other_outputs(self, f): for txo in self.outputs: if not txo.is_my_account and f(txo.script): yield txo @property def my_claim_outputs(self): return self._filter_my_outputs(lambda s: s.is_claim_name) @property def my_update_outputs(self): return self._filter_my_outputs(lambda s: s.is_update_claim) @property def my_support_outputs(self): return self._filter_my_outputs(lambda s: s.is_support_claim) @property def other_support_outputs(self): return self._filter_other_outputs(lambda s: s.is_support_claim) @property def my_abandon_outputs(self): for txi in self.inputs: abandon = txi.txo_ref.txo if abandon is not None and abandon.is_my_account and abandon.script.is_claim_involved: is_update = False if abandon.script.is_claim_name or abandon.script.is_update_claim: for update in self.my_update_outputs: if abandon.claim_id == update.claim_id: is_update = True break if not is_update: yield abandon