crypto.py

"""
Adopted from:

https://github.com/raiden-network/microraiden/blob/436562a60521c0b321edec822980807c3ac6bca4/microraiden/utils/crypto.py

MIT License

Copyright (c) 2017

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""

from typing import List, Tuple, Any

import rlp
from coincurve import PrivateKey, PublicKey
from eth_utils import (
    encode_hex,
    decode_hex,
    remove_0x_prefix,
    keccak,
    is_0x_prefixed,
    to_checksum_address,
)
from ethereum.transactions import Transaction

Type = str
Name = str
TypedData = Tuple[Type, Name, Any]


def generate_privkey() -> bytes:
    return encode_hex(PrivateKey().secret)


def pubkey_to_addr(pubkey) -> str:
    if isinstance(pubkey, PublicKey):
        pubkey = pubkey.format(compressed=False)
    assert isinstance(pubkey, bytes)
    return encode_hex(keccak256(pubkey[1:])[-20:])


def privkey_to_addr(privkey: str) -> str:
    return to_checksum_address(
        pubkey_to_addr(PrivateKey.from_hex(remove_0x_prefix(privkey)).public_key)
    )


def addr_from_sig(sig: bytes, msg: bytes):_
    assert len(sig) == 65
    # Support Ethereum's EC v value of 27 and EIP 155 values of > 35.
    if sig[-1] >= 35:
        network_id = (sig[-1] - 35) // 2
        sig = sig[:-1] + bytes([sig[-1] - 35 - 2 * network_id])
    elif sig[-1] >= 27:
        sig = sig[:-1] + bytes([sig[-1] - 27])

    receiver_pubkey = PublicKey.from_signature_and_message(sig, msg, hasher=None)
    return pubkey_to_addr(receiver_pubkey)


def pack(*args) -> bytes:
    """
    Simulates Solidity's keccak256 packing. Integers can be passed as tuples where the second tuple
    element specifies the variable's size in bits, e.g.:
    keccak256((5, 32))
    would be equivalent to Solidity's
    keccak256(uint32(5))
    Default size is 256.
    """

    def format_int(value, size):
        assert isinstance(value, int)
        assert isinstance(size, int)
        if value >= 0:
            return decode_hex('{:x}'.format(value).zfill(size // 4))
        else:
            return decode_hex('{:x}'.format((1 << size) + value))

    msg = b''
    for arg in args:
        assert arg is not None
        if isinstance(arg, bytes):
            msg += arg
        elif isinstance(arg, str):
            if is_0x_prefixed(arg):
                msg += decode_hex(arg)
            else:
                msg += arg.encode()
        elif isinstance(arg, bool):
            msg += format_int(int(arg), 8)
        elif isinstance(arg, int):
            msg += format_int(arg, 256)  # note: its a trap - use tuple instead
        elif isinstance(arg, tuple):
            msg += format_int(arg[0], arg[1])
        else:
            raise ValueError('Unsupported type: {}.'.format(type(arg)))

    return msg


def keccak256(*args) -> bytes:
    return keccak(pack(*args))


def sign(privkey: str, msg: bytes, v=0) -> bytes:
    assert isinstance(msg, bytes)
    assert isinstance(privkey, str)

    pk = PrivateKey.from_hex(remove_0x_prefix(privkey))
    assert len(msg) == 32

    sig = pk.sign_recoverable(msg, hasher=None)
    assert len(sig) == 65

    sig = sig[:-1] + bytes([sig[-1] + v])

    return sig


def sign_transaction(tx: Transaction, privkey: str, network_id: int):
    # Implementing EIP 155.
    tx.v = network_id
    sig = sign(privkey, keccak256(rlp.encode(tx)), v=35 + 2 * network_id)
    v, r, s = sig[-1], sig[0:32], sig[32:-1]
    tx.v = v
    tx.r = int.from_bytes(r, byteorder='big')
    tx.s = int.from_bytes(s, byteorder='big')


def eth_message_hash(msg: str) -> bytes:
    msg = '\x19Ethereum Signed Message:\n' + str(len(msg)) + msg
    return keccak256(msg)


def eth_sign(privkey: str, msg: str) -> bytes:
    assert isinstance(msg, str)
    sig = sign(privkey, eth_message_hash(msg), v=27)
    return sig


def eth_verify(sig: bytes, msg: str) -> str:
    return addr_from_sig(sig, eth_message_hash(msg))


def eth_typed_data_message(message) -> bytes:
    types = ['%s %s' % (x['type'], x['name']) for x in message]
    values = [x['value'] for x in message]

    return keccak256(keccak256(*types), keccak256(*values))


def eth_sign_typed_data(privkey: str, typed_data: List[TypedData]) -> bytes:
    msg = eth_typed_data_message(typed_data)
    return sign(privkey, msg, v=27)


def eth_typed_data_message_eip(message) -> bytes:
    types = ['%s %s' % (x['type'], x['name']) for x in message]
    values = [x['value'] for x in message]

    return keccak256(keccak256(*types), *values)


def eth_sign_typed_data_eip(privkey: str, typed_data: List[TypedData]) -> bytes:
    msg = eth_typed_data_message_eip(typed_data)
    return sign(privkey, msg, v=27)