hathorlib blueprint move 2

hathor/crypto/util.py

@@ -12,23 +12,33 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import hashlib
 from typing import Optional
 
-import base58
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.hazmat.primitives.serialization import (
     Encoding,
     KeySerializationEncryption,
     NoEncryption,
     PrivateFormat,
-    PublicFormat,
     load_der_private_key,
 )
 
-from hathor.conf.get_settings import get_global_settings
 from hathor.util import not_none
+from hathorlib.utils.address import (  # noqa: F401
+    decode_address,
+    get_address_b58_from_bytes,
+    get_address_b58_from_public_key,
+    get_address_b58_from_public_key_bytes,
+    get_address_b58_from_public_key_hash,
+    get_address_b58_from_redeem_script_hash,
+    get_address_from_public_key_hash,
+    get_address_from_redeem_script_hash,
+    get_checksum,
+    get_hash160,
+    get_public_key_bytes_compressed,
+    get_public_key_from_bytes_compressed,
+)
 
 _BACKEND = default_backend()
 
@@ -47,26 +57,6 @@ def get_private_key_from_bytes(private_key_bytes: bytes,
     return not_none(load_der_private_key(private_key_bytes, password, _BACKEND))
 
 
-try:
-    hashlib.new('ripemd160', b'')
-except Exception:
-    # XXX: the source says "Test-only pure Python RIPEMD160 implementation", however for our case this is acceptable
-    #      for more details see: https://github.com/bitcoin/bitcoin/pull/23716/files which has a copy of the same code
-    import pycoin.contrib.ripemd160
-
-    def get_hash160(public_key_bytes: bytes) -> bytes:
-        """The input is hashed twice: first with SHA-256 and then with RIPEMD-160"""
-        key_hash = hashlib.sha256(public_key_bytes)
-        return pycoin.contrib.ripemd160.ripemd160(key_hash.digest())
-else:
-    def get_hash160(public_key_bytes: bytes) -> bytes:
-        """The input is hashed twice: first with SHA-256 and then with RIPEMD-160"""
-        key_hash = hashlib.sha256(public_key_bytes)
-        h = hashlib.new('ripemd160')
-        h.update(key_hash.digest())
-        return h.digest()
-
-
 def get_address_from_public_key(public_key):
     """ Get bytes from public key object and call method that expect bytes
 
@@ -93,180 +83,6 @@ def get_address_from_public_key_bytes(public_key_bytes):
     return get_address_from_public_key_hash(public_key_hash)
 
 
-def get_address_b58_from_public_key(public_key: ec.EllipticCurvePublicKey) -> str:
-    """Gets the b58 address from a public key.
-
-    :param: ec.EllipticCurvePublicKey
-    :return: the b58-encoded address
-    :rtype: string
-    """
-    public_key_bytes = get_public_key_bytes_compressed(public_key)
-    return get_address_b58_from_public_key_bytes(public_key_bytes)
-
-
-def get_address_b58_from_public_key_hash(public_key_hash: bytes) -> str:
-    """Gets the b58 address from the hash of a public key.
-
-        :param public_key_hash: hash of public key (sha256 and ripemd160)
-        :param public_key_hash: bytes
-
-        :return: address in base 58
-        :rtype: string
-    """
-    address = get_address_from_public_key_hash(public_key_hash)
-    return base58.b58encode(address).decode('utf-8')
-
-
-def get_address_from_public_key_hash(public_key_hash: bytes, version_byte: Optional[bytes] = None) -> bytes:
-    """Gets the address in bytes from the public key hash
-
-        :param public_key_hash: hash of public key (sha256 and ripemd160)
-        :param public_key_hash: bytes
-
-        :param version_byte: first byte of address to define the version of this address
-        :param version_byte: bytes
-
-        :return: address in bytes
-        :rtype: bytes
-    """
-    settings = get_global_settings()
-    address = b''
-    actual_version_byte: bytes = version_byte if version_byte is not None else settings.P2PKH_VERSION_BYTE
-    # Version byte
-    address += actual_version_byte
-    # Pubkey hash
-    address += public_key_hash
-    checksum = get_checksum(address)
-    address += checksum
-    return address
-
-
-def get_checksum(address_bytes: bytes) -> bytes:
-    """ Calculate double sha256 of address and gets first 4 bytes
-
-        :param address_bytes: address before checksum
-        :param address_bytes: bytes
-
-        :return: checksum of the address
-        :rtype: bytes
-    """
-    return hashlib.sha256(hashlib.sha256(address_bytes).digest()).digest()[:4]
-
-
-def get_address_b58_from_public_key_bytes(public_key_bytes: bytes) -> str:
-    """Gets the b58 address from a public key bytes.
-
-        :param public_key_bytes: public key in bytes
-        :param public_key_bytes: bytes
-
-        :return: address in base 58
-        :rtype: string
-    """
-    public_key_hash = get_hash160(public_key_bytes)
-    return get_address_b58_from_public_key_hash(public_key_hash)
-
-
-def get_address_b58_from_bytes(address):
-    """Gets the b58 address from the address in bytes
-
-        :param address: bytes
-
-        :return: address in base 58
-        :rtype: string
-    """
-    return base58.b58encode(address).decode('utf-8')
-
-
-def get_public_key_bytes_compressed(public_key: ec.EllipticCurvePublicKey) -> bytes:
-    """ Returns the bytes from a cryptography ec.EllipticCurvePublicKey in a compressed format
-
-        :param public_key: Public key object
-        :type public_key: ec.EllipticCurvePublicKey
-
-        :rtype: bytes
-    """
-    return public_key.public_bytes(Encoding.X962, PublicFormat.CompressedPoint)
-
-
-def get_public_key_from_bytes_compressed(public_key_bytes: bytes) -> ec.EllipticCurvePublicKey:
-    """ Returns the cryptography public key from the compressed bytes format
-
-        :param public_key_bytes: Compressed format of public key in bytes
-        :type public_key_bytes: bytes
-
-        :rtype: ec.EllipticCurvePublicKey
-    """
-    return ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256K1(), public_key_bytes)
-
-
-def get_address_b58_from_redeem_script_hash(redeem_script_hash: bytes, version_byte: Optional[bytes] = None) -> str:
-    """Gets the b58 address from the hash of the redeem script in multisig.
-
-        :param redeem_script_hash: hash of the redeem script (sha256 and ripemd160)
-        :param redeem_script_hash: bytes
-
-        :return: address in base 58
-        :rtype: string
-    """
-    settings = get_global_settings()
-    actual_version_byte: bytes = version_byte if version_byte is not None else settings.MULTISIG_VERSION_BYTE
-    address = get_address_from_redeem_script_hash(redeem_script_hash, actual_version_byte)
-    return base58.b58encode(address).decode('utf-8')
-
-
-def get_address_from_redeem_script_hash(redeem_script_hash: bytes, version_byte: Optional[bytes] = None) -> bytes:
-    """Gets the address in bytes from the redeem script hash
-
-        :param redeem_script_hash: hash of redeem script (sha256 and ripemd160)
-        :param redeem_script_hash: bytes
-
-        :param version_byte: first byte of address to define the version of this address
-        :param version_byte: bytes
-
-        :return: address in bytes
-        :rtype: bytes
-    """
-    settings = get_global_settings()
-    actual_version_byte: bytes = version_byte if version_byte is not None else settings.MULTISIG_VERSION_BYTE
-    address = b''
-    # Version byte
-    address += actual_version_byte
-    # redeem script hash
-    address += redeem_script_hash
-    checksum = get_checksum(address)
-    address += checksum
-    return address
-
-
-def decode_address(address58: str) -> bytes:
-    """ Decode address in base58 to bytes
-
-    :param address58: Wallet address in base58
-    :type address58: string
-
-    :raises InvalidAddress: if address58 is not a valid base58 string or
-                            not a valid address or has invalid checksum
-
-    :return: Address in bytes
-    :rtype: bytes
-    """
-    from hathor.wallet.exceptions import InvalidAddress
-    try:
-        decoded_address = base58.b58decode(address58)
-    except ValueError:
-        # Invalid base58 string
-        raise InvalidAddress('Invalid base58 address')
-    # Validate address size [25 bytes]
-    if len(decoded_address) != 25:
-        raise InvalidAddress('Address size must have 25 bytes')
-    # Validate the checksum
-    address_checksum = decoded_address[-4:]
-    valid_checksum = get_checksum(decoded_address[:-4])
-    if address_checksum != valid_checksum:
-        raise InvalidAddress('Invalid checksum of address')
-    return decoded_address
-
-
 def is_pubkey_compressed(pubkey: bytes) -> bool:
     """ Receives a public key bytes and return True if in CompressedPoint format
         This function will not test if this is a valid public key.

hathor/exception.py

@@ -12,10 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
-class HathorError(Exception):
-    """Base class for exceptions in Hathor."""
-    pass
+from hathorlib.exceptions import HathorError  # noqa: F401
 
 
 class BuilderError(Exception):

hathor/nanocontracts/blueprint_syntax_validation.py

@@ -12,88 +12,10 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from __future__ import annotations
-
-import inspect
-from typing import Callable
-
-from hathor.nanocontracts.exception import BlueprintSyntaxError
-
-
-def validate_has_self_arg(fn: Callable, annotation_name: str) -> None:
-    """Validate the `self` arg of a callable."""
-    arg_spec = inspect.getfullargspec(fn)
-    if len(arg_spec.args) == 0:
-        raise BlueprintSyntaxError(f'@{annotation_name} method must have `self` argument: `{fn.__name__}()`')
-
-    if arg_spec.args[0] != 'self':
-        raise BlueprintSyntaxError(
-            f'@{annotation_name} method first argument must be called `self`: `{fn.__name__}()`'
-        )
-
-    if 'self' in arg_spec.annotations.keys():
-        raise BlueprintSyntaxError(f'@{annotation_name} method `self` argument must not be typed: `{fn.__name__}()`')
-
-
-def validate_method_types(fn: Callable) -> None:
-    """Validate the arg and return types of a callable."""
-    special_args = ['self']
-    arg_spec = inspect.getfullargspec(fn)
-
-    if 'return' not in arg_spec.annotations:
-        raise BlueprintSyntaxError(f'missing return type on method `{fn.__name__}`')
-
-    # TODO: This currently fails for types such as unions, probably because this is the wrong
-    #  parsing function to use. Fix this.
-    # from hathor.nanocontracts.fields import get_field_class_for_attr
-    # return_type = arg_spec.annotations['return']
-    # if return_type is not None:
-    #     try:
-    #         get_field_class_for_attr(return_type)
-    #     except UnknownFieldType:
-    #         raise BlueprintSyntaxError(
-    #             f'unsupported return type `{return_type}` on method `{fn.__name__}`'
-    #         )
-
-    for arg_name in arg_spec.args:
-        if arg_name in special_args:
-            continue
-
-        if arg_name not in arg_spec.annotations:
-            raise BlueprintSyntaxError(f'argument `{arg_name}` on method `{fn.__name__}` must be typed')
-
-        # TODO: This currently fails for @view methods with NamedTuple as args for example,
-        #  because API calls use a different parsing function. Fix this.
-        # arg_type = arg_spec.annotations[arg_name]
-        # try:
-        #     get_field_class_for_attr(arg_type)
-        # except UnknownFieldType:
-        #     raise BlueprintSyntaxError(
-        #         f'unsupported type `{arg_type.__name__}` on argument `{arg_name}` of method `{fn.__name__}`'
-        #     )
-
-
-def validate_has_ctx_arg(fn: Callable, annotation_name: str) -> None:
-    """Validate the context arg of a callable."""
-    arg_spec = inspect.getfullargspec(fn)
-
-    if len(arg_spec.args) < 2:
-        raise BlueprintSyntaxError(
-            f'@{annotation_name} method must have `Context` argument: `{fn.__name__}()`'
-        )
-
-    from hathor.nanocontracts import Context
-    second_arg = arg_spec.args[1]
-    if arg_spec.annotations[second_arg] is not Context:
-        raise BlueprintSyntaxError(
-            f'@{annotation_name} method second arg `{second_arg}` argument must be of type `Context`: '
-            f'`{fn.__name__}()`'
-        )
-
-
-def validate_has_not_ctx_arg(fn: Callable, annotation_name: str) -> None:
-    """Validate that a callable doesn't have a `Context` arg."""
-    from hathor.nanocontracts import Context
-    arg_spec = inspect.getfullargspec(fn)
-    if Context in arg_spec.annotations.values():
-        raise BlueprintSyntaxError(f'@{annotation_name} method cannot have arg with type `Context`: `{fn.__name__}()`')
+# Re-export from hathorlib for backward compatibility
+from hathorlib.nanocontracts.blueprint_syntax_validation import (  # noqa: F401
+    validate_has_ctx_arg,
+    validate_has_not_ctx_arg,
+    validate_has_self_arg,
+    validate_method_types,
+)