lbry-sdk/lbrynet/schema/claim.py

import json
from collections import OrderedDict
from typing import List, Tuple
from decimal import Decimal
from binascii import hexlify, unhexlify

from google.protobuf import json_format  # pylint: disable=no-name-in-module
from google.protobuf.message import DecodeError as DecodeError_pb  # pylint: disable=no-name-in-module,import-error

from torba.client.constants import COIN

from lbrynet.schema.signature import Signature
from lbrynet.schema.constants import CURVE_NAMES, SECP256k1
from lbrynet.schema.error import DecodeError
from lbrynet.schema.types.v2.claim_pb2 import Claim as ClaimMessage, Fee as FeeMessage
from lbrynet.schema.base import b58decode, b58encode
from lbrynet.schema import compat


class ClaimDict(OrderedDict):
    def __init__(self, claim_dict=None, detached_signature: Signature=None):
        if isinstance(claim_dict, legacy_claim_pb2.Claim):
            raise Exception("To initialize %s with a Claim protobuf use %s.load_protobuf" %
                            (self.__class__.__name__, self.__class__.__name__))
        self.detached_signature = detached_signature
        OrderedDict.__init__(self, claim_dict or [])

    @property
    def protobuf_dict(self):
        """Claim dictionary using base64 to represent bytes"""

        return json.loads(json_format.MessageToJson(self.protobuf, True))

    @property
    def protobuf(self):
        """Claim message object"""

        return LegacyClaim.load(self)

    @property
    def serialized(self):
        """Serialized Claim protobuf"""
        if self.detached_signature:
            return self.detached_signature.serialized
        return self.protobuf.SerializeToString()

    @property
    def serialized_no_signature(self):
        """Serialized Claim protobuf without publisherSignature field"""
        claim = self.protobuf
        claim.ClearField("publisherSignature")
        return ClaimDict.load_protobuf(claim).serialized

    @property
    def has_signature(self):
        return self.protobuf.HasField("publisherSignature") or (
                self.detached_signature and self.detached_signature.raw_signature
        )

    @property
    def is_certificate(self):
        claim = self.protobuf
        return CLAIM_TYPE_NAMES[claim.claimType] == "certificate"

    @property
    def is_stream(self):
        claim = self.protobuf
        return CLAIM_TYPE_NAMES[claim.claimType] == "stream"

    @property
    def source_hash(self):
        claim = self.protobuf
        if not CLAIM_TYPE_NAMES[claim.claimType] == "stream":
            return None
        return binascii.hexlify(claim.stream.source.source)

    @property
    def has_fee(self):
        claim = self.protobuf
        if not CLAIM_TYPE_NAMES[claim.claimType] == "stream":
            return None
        if claim.stream.metadata.HasField("fee"):
            return True
        return False

    @property
    def source_fee(self):
        claim = self.protobuf
        if not CLAIM_TYPE_NAMES[claim.claimType] == "stream":
            return None
        if claim.stream.metadata.HasField("fee"):
            return Fee.load_protobuf(claim.stream.metadata.fee)
        return None

    @property
    def certificate_id(self) -> str:
        if self.protobuf.HasField("publisherSignature"):
            return binascii.hexlify(self.protobuf.publisherSignature.certificateId).decode()
        if self.detached_signature and self.detached_signature.certificate_id:
            return binascii.hexlify(self.detached_signature.certificate_id).decode()

    @property
    def signature(self):
        if not self.has_signature:
            return None
        return binascii.hexlify(self.protobuf.publisherSignature.signature)

    @property
    def protobuf_len(self):
        """Length of serialized string"""

        return self.protobuf.ByteSize()

    @property
    def json_len(self):
        """Length of json encoded string"""

        return len(json.dumps(self.claim_dict))

    @property
    def claim_dict(self):
        """Claim dictionary with bytes represented as hex and base58"""

        return dict(encode_fields(self, self.detached_signature))

    @classmethod
    def load_protobuf_dict(cls, protobuf_dict, detached_signature=None):
        """
        Load a ClaimDict from a dictionary with base64 encoded bytes
        (as returned by the protobuf json formatter)
        """

        return cls(decode_b64_fields(protobuf_dict), detached_signature=detached_signature)

    @classmethod
    def load_protobuf(cls, protobuf_claim, detached_signature=None):
        """Load ClaimDict from a protobuf Claim message"""
        return cls.load_protobuf_dict(json.loads(json_format.MessageToJson(protobuf_claim, True)), detached_signature)

    @classmethod
    def load_dict(cls, claim_dict):
        """Load ClaimDict from a dictionary with hex and base58 encoded bytes"""
        try:
            claim_dict, detached_signature = decode_fields(claim_dict)
            return cls.load_protobuf(cls(claim_dict).protobuf, detached_signature)
        except json_format.ParseError as err:
            raise DecodeError(str(err))

    @classmethod
    def deserialize(cls, serialized):
        """Load a ClaimDict from a serialized protobuf string"""
        detached_signature = Signature.flagged_parse(serialized)

        temp_claim = legacy_claim_pb2.Claim()
        try:
            temp_claim.ParseFromString(detached_signature.payload)
        except DecodeError_pb:
            raise DecodeError(DecodeError_pb)
        return cls.load_protobuf(temp_claim, detached_signature=detached_signature)

    @classmethod
    def generate_certificate(cls, private_key, curve=SECP256k1):
        signer = get_signer(curve).load_pem(private_key)
        return cls.load_protobuf(signer.certificate)

    def sign(self, private_key, claim_address, cert_claim_id, curve=SECP256k1, name=None, force_detached=False):
        signer = get_signer(curve).load_pem(private_key)
        signed, signature = signer.sign_stream_claim(self, claim_address, cert_claim_id, name, force_detached)
        return ClaimDict.load_protobuf(signed, signature)

    def validate_signature(self, claim_address, certificate, name=None):
        if isinstance(certificate, ClaimDict):
            certificate = certificate.protobuf
        curve = CURVE_NAMES[certificate.certificate.keyType]
        validator = get_validator(curve).load_from_certificate(certificate, self.certificate_id)
        return validator.validate_claim_signature(self, claim_address, name)

    def validate_private_key(self, private_key, certificate_id):
        certificate = self.protobuf
        if CLAIM_TYPE_NAMES[certificate.claimType] != "certificate":
            return
        curve = CURVE_NAMES[certificate.certificate.keyType]
        validator = get_validator(curve).load_from_certificate(certificate, certificate_id)
        signing_key = validator.signing_key_from_pem(private_key)
        return validator.validate_private_key(signing_key)

    def get_validator(self, certificate_id):
        """
        Get a lbrynet.schema.validator.Validator object for a certificate claim

        :param certificate_id: claim id of this certificate claim
        :return: None or lbrynet.schema.validator.Validator object
        """

        claim = self.protobuf
        if CLAIM_TYPE_NAMES[claim.claimType] != "certificate":
            return
        curve = CURVE_NAMES[claim.certificate.keyType]
        return get_validator(curve).load_from_certificate(claim, certificate_id)


class Claim:

    __slots__ = '_claim', 'signature', 'certificate_id', 'signature_type', 'unsigned_payload'

    def __init__(self, claim_message=None):
        self._claim = claim_message or ClaimMessage()
        self.signature = None
        self.signature_type = 'SECP256k1'
        self.certificate_id = None
        self.unsigned_payload = None

    @property
    def is_undetermined(self):
        return self._claim.WhichOneof('type') is None

    @property
    def is_stream(self):
        return self._claim.WhichOneof('type') == 'stream'

    @property
    def is_channel(self):
        return self._claim.WhichOneof('type') == 'channel'

    @property
    def is_signed(self):
        return self.signature is not None


    @property
    def stream_message(self):
        if self.is_undetermined:
            self._claim.stream.SetInParent()
        if not self.is_stream:
            raise ValueError('Claim is not a stream.')
        return self._claim.stream

    @property
    def stream(self) -> 'Stream':
        return Stream(self)

    @property
    def channel_message(self):
        if self.is_undetermined:
            self._claim.channel.SetInParent()
        if not self.is_channel:
            raise ValueError('Claim is not a channel.')
        return self._claim.channel

    @property
    def channel(self) -> 'Channel':
        return Channel(self)

    def to_bytes(self) -> bytes:
        return self._claim.SerializeToString()

    @classmethod
    def from_bytes(cls, data: bytes) -> 'Claim':
        claim = ClaimMessage()
        if data[0] == 0:
            claim.ParseFromString(data[1:])
            return cls(claim)
        elif data[0] == 1:
            claim.ParseFromString(data[85:])
            return cls(claim).from_message(payload[1:21], payload[21:85])
        elif data[0] == ord('{'):
            return compat.from_old_json_schema(cls(claim), data)
        else:
            return compat.from_types_v1(cls(claim), data)


class Video:

    __slots__ = '_video',

    def __init__(self, video_message):
        self._video = video_message

    @property
    def width(self) -> int:
        return self._video.width

    @width.setter
    def width(self, width: int):
        self._video.width = width

    @property
    def height(self) -> int:
        return self._video.height

    @height.setter
    def height(self, height: int):
        self._video.height = height

    @property
    def dimensions(self) -> Tuple[int, int]:
        return self.width, self.height

    @dimensions.setter
    def dimensions(self, dimensions: Tuple[int, int]):
        self._video.width, self._video.height = dimensions


class File:

    __slots__ = '_file',

    def __init__(self, file_message):
        self._file = file_message

    @property
    def name(self) -> str:
        return self._file.name

    @name.setter
    def name(self, name: str):
        self._file.name = name

    @property
    def size(self) -> int:
        return self._file.size

    @size.setter
    def size(self, size: int):
        self._file.size = size


class Fee:

    __slots__ = '_fee',

    def __init__(self, fee_message):
        self._fee = fee_message

    @property
    def currency(self) -> str:
        return FeeMessage.Currency.Name(self._fee.currency)

    @property
    def address(self) -> str:
        return b58encode(self._fee.address).decode()

    @address.setter
    def address(self, address: str):
        self._fee.address = b58decode(address)

    @property
    def address_bytes(self) -> bytes:
        return self._fee.address

    @address_bytes.setter
    def address_bytes(self, address: bytes):
        self._fee.address = address

    @property
    def amount(self) -> Decimal:
        if self.currency == 'LBC':
            return self.lbc
        if self.currency == 'USD':
            return self.usd

    DEWIES = Decimal(COIN)

    @property
    def lbc(self) -> Decimal:
        if self._fee.currency != FeeMessage.LBC:
            raise ValueError('LBC can only be returned for LBC fees.')
        return Decimal(self._fee.amount / self.DEWIES)

    @lbc.setter
    def lbc(self, amount: Decimal):
        self.dewies = int(amount * self.DEWIES)

    @property
    def dewies(self) -> int:
        if self._fee.currency != FeeMessage.LBC:
            raise ValueError('Dewies can only be returned for LBC fees.')
        return self._fee.amount

    @dewies.setter
    def dewies(self, amount: int):
        self._fee.amount = amount
        self._fee.currency = FeeMessage.LBC

    PENNIES = Decimal(100.0)

    @property
    def usd(self) -> Decimal:
        if self._fee.currency != FeeMessage.USD:
            raise ValueError('USD can only be returned for USD fees.')
        return Decimal(self._fee.amount / self.PENNIES)

    @usd.setter
    def usd(self, amount: Decimal):
        self.pennies = int(amount * self.PENNIES)

    @property
    def pennies(self) -> int:
        if self._fee.currency != FeeMessage.USD:
            raise ValueError('Pennies can only be returned for USD fees.')
        return self._fee.amount

    @pennies.setter
    def pennies(self, amount: int):
        self._fee.amount = amount
        self._fee.currency = FeeMessage.USD


class Channel:

    __slots__ = '_claim', '_channel'

    def __init__(self, claim: Claim = None):
        self._claim = claim or Claim()
        self._channel = self._claim.channel_message

    @property
    def claim(self) -> Claim:
        return self._claim

    @property
    def tags(self) -> List:
        return self._channel.tags

    @property
    def public_key(self) -> str:
        return hexlify(self._channel.public_key).decode()

    @public_key.setter
    def public_key(self, sd_public_key: str):
        self._channel.public_key = unhexlify(sd_public_key.encode())

    @property
    def public_key_bytes(self) -> bytes:
        return self._channel.public_key

    @public_key_bytes.setter
    def public_key_bytes(self, public_key: bytes):
        self._channel.public_key = public_key

    @property
    def language(self) -> str:
        return self._channel.language

    @language.setter
    def language(self, language: str):
        self._channel.language = language

    @property
    def title(self) -> str:
        return self._channel.title

    @title.setter
    def title(self, title: str):
        self._channel.title = title

    @property
    def description(self) -> str:
        return self._channel.description

    @description.setter
    def description(self, description: str):
        self._channel.description = description

    @property
    def contact_email(self) -> str:
        return self._channel.contact_email

    @contact_email.setter
    def contact_email(self, contact_email: str):
        self._channel.contact_email = contact_email

    @property
    def homepage_url(self) -> str:
        return self._channel.homepage_url

    @homepage_url.setter
    def homepage_url(self, homepage_url: str):
        self._channel.homepage_url = homepage_url

    @property
    def thumbnail_url(self) -> str:
        return self._channel.thumbnail_url

    @thumbnail_url.setter
    def thumbnail_url(self, thumbnail_url: str):
        self._channel.thumbnail_url = thumbnail_url

    @property
    def cover_url(self) -> str:
        return self._channel.cover_url

    @cover_url.setter
    def cover_url(self, cover_url: str):
        self._channel.cover_url = cover_url


class Stream:

    __slots__ = '_claim', '_stream'

    def __init__(self, claim: Claim = None):
        self._claim = claim or Claim()
        self._stream = self._claim.stream_message

    @property
    def claim(self) -> Claim:
        return self._claim

    @property
    def video(self) -> Video:
        return Video(self._stream.video)

    @property
    def file(self) -> File:
        return File(self._stream.file)

    @property
    def fee(self) -> Fee:
        return Fee(self._stream.fee)

    @property
    def tags(self) -> List:
        return self._stream.tags

    @property
    def hash(self) -> str:
        return hexlify(self._stream.hash).decode()

    @hash.setter
    def hash(self, sd_hash: str):
        self._stream.hash = unhexlify(sd_hash.encode())

    @property
    def hash_bytes(self) -> bytes:
        return self._stream.hash

    @hash_bytes.setter
    def hash_bytes(self, hash: bytes):
        self._stream.hash = hash

    @property
    def language(self) -> str:
        return self._stream.language

    @language.setter
    def language(self, language: str):
        self._stream.language = language

    @property
    def title(self) -> str:
        return self._stream.title

    @title.setter
    def title(self, title: str):
        self._stream.title = title

    @property
    def author(self) -> str:
        return self._stream.author

    @author.setter
    def author(self, author: str):
        self._stream.author = author

    @property
    def description(self) -> str:
        return self._stream.description

    @description.setter
    def description(self, description: str):
        self._stream.description = description

    @property
    def media_type(self) -> str:
        return self._stream.media_type

    @media_type.setter
    def media_type(self, media_type: str):
        self._stream.media_type = media_type

    @property
    def license(self) -> str:
        return self._stream.license

    @license.setter
    def license(self, license: str):
        self._stream.license = license

    @property
    def license_url(self) -> str:
        return self._stream.license_url

    @license_url.setter
    def license_url(self, license_url: str):
        self._stream.license_url = license_url

    @property
    def thumbnail_url(self) -> str:
        return self._stream.thumbnail_url

    @thumbnail_url.setter
    def thumbnail_url(self, thumbnail_url: str):
        self._stream.thumbnail_url = thumbnail_url

    @property
    def duration(self) -> int:
        return self._stream.duration

    @duration.setter
    def duration(self, duration: int):
        self._stream.duration = duration

    @property
    def release_time(self) -> int:
        return self._stream.release_time

    @release_time.setter
    def release_time(self, release_time: int):
        self._stream.release_time = release_time