basicswap_miserver/basicswap/network.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# Copyright (c) 2020 tecnovert
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.

'''
    Message 2 bytes msg_class, 4 bytes length, [ 2 bytes msg_type, payload ]

    Handshake procedure:
        node0 connecting to node1
        node0 send_handshake
        node1 process_handshake
        node1 send_ping  - With a version field
        node0 recv_ping
            Both nodes are initialised

    XChaCha20_Poly1305 mac is 16bytes
'''

import time
import queue
import random
import select
import socket
import struct
import hashlib
import logging
import secrets
import threading
import traceback

from enum import IntEnum, auto
from collections import OrderedDict
from Crypto.Cipher import ChaCha20_Poly1305  # TODO: Add to libsecp256k1/coincurve fork
from coincurve.keys import PrivateKey, PublicKey
from basicswap.contrib.rfc6979 import (
    rfc6979_hmac_sha256_initialize,
    rfc6979_hmac_sha256_generate)


START_TOKEN = 0xabcd
MSG_START_TOKEN = struct.pack('>H', START_TOKEN)

MSG_MAX_SIZE = 0x200000  # 2MB

MSG_HEADER_LEN = 8


MAX_SEEN_EPHEM_KEYS = 1000
TIMESTAMP_LEEWAY = 8


class NetMessageTypes(IntEnum):
    HANDSHAKE = auto()
    PING = auto()
    PONG = auto()
    DATA = auto()
    ONION_PACKET = auto()

    @classmethod
    def has_value(cls, value):
        return value in cls._value2member_map_


'''
class NetMessage:
    def __init__(self):
        self._msg_class = None  # 2 bytes
        self._len = None # 4 bytes
        self._msg_type = None  # 2 bytes
'''


# Ensure handshake keys are not reused by including the time in the msg, mac and key hash
# Verify timestamp is not too old
# Add keys to db to catch concurrent attempts, records can be cleared periodically, the timestamp should catch older replay attempts
class MsgHandshake:
    __slots__ = ('_timestamp', '_ephem_pk', '_ct', '_mac')

    def __init__(self):
        pass

    def encode_aad(self):  # Additional Authenticated Data
        return struct.pack('>H', NetMessageTypes.HANDSHAKE) + \
            struct.pack('>Q', self._timestamp) + \
            self._ephem_pk

    def encode(self):
        return self.encode_aad() + self._ct + self._mac

    def decode(self, msg_mv):
        o = 2
        self._timestamp = struct.unpack('>Q', msg_mv[o: o + 8])[0]
        o += 8
        self._ephem_pk = bytes(msg_mv[o: o + 33])
        o += 33
        self._ct = bytes(msg_mv[o: -16])
        self._mac = bytes(msg_mv[-16:])


class Peer:
    __slots__ = (
        '_mx', '_pubkey', '_address', '_socket', '_version', '_ready', '_incoming',
        '_connected_at', '_last_received_at', '_bytes_sent', '_bytes_received',
        '_receiving_length', '_receiving_buffer', '_recv_messages', '_misbehaving_score',
        '_ke', '_km', '_dir', '_sent_nonce', '_recv_nonce', '_last_handshake_at',
        '_ping_nonce', '_last_ping_at', '_last_ping_rtt')

    def __init__(self, address, socket, pubkey):
        self._mx = threading.Lock()
        self._pubkey = pubkey
        self._address = address
        self._socket = socket
        self._version = None
        self._ready = False  # True when handshake is complete
        self._incoming = False
        self._connected_at = time.time()
        self._last_received_at = 0
        self._last_handshake_at = 0

        self._bytes_sent = 0
        self._bytes_received = 0

        self._receiving_length = 0
        self._receiving_buffer = None
        self._recv_messages = queue.Queue()  # Built in mutex
        self._misbehaving_score = 0  # TODO: Must be persistent - save to db

        self._ping_nonce = 0
        self._last_ping_at = 0  # ms
        self._last_ping_rtt = 0  # ms

    def close(self):
        self._socket.close()


def listen_thread(cls):
    timeout = 1.0

    max_bytes = 0x10000
    while cls._running:
        # logging.info('[rm] network loop %d', cls._running)
        readable, writable, errored = select.select(cls._read_sockets, cls._write_sockets, cls._error_sockets, timeout)
        cls._mx.acquire()
        try:
            disconnected_peers = []
            for s in readable:
                if s == cls._socket:
                    peer_socket, address = cls._socket.accept()
                    logging.info('Connection from %s', address)
                    new_peer = Peer(address, peer_socket, None)
                    new_peer._incoming = True
                    cls._peers.append(new_peer)
                    cls._error_sockets.append(peer_socket)
                    cls._read_sockets.append(peer_socket)
                else:
                    for peer in cls._peers:
                        if peer._socket == s:
                            try:
                                bytes_recv = s.recv(max_bytes, socket.MSG_DONTWAIT)
                            except socket.error as se:
                                if se.args[0] not in (socket.EWOULDBLOCK, ):
                                    logging.error('Receive error %s', str(se))
                                    disconnected_peers.append(peer)
                                    continue
                            except Exception as e:
                                logging.error('Receive error %s', str(e))
                                disconnected_peers.append(peer)
                                continue

                            if len(bytes_recv) < 1:
                                disconnected_peers.append(peer)
                                continue
                            cls.receive_bytes(peer, bytes_recv)

            for s in errored:
                logging.warning('Socket error')

            for peer in disconnected_peers:
                cls.disconnect(peer)
        finally:
            cls._mx.release()


def msg_thread(cls):
    timeout = 0.1
    while cls._running:
        processed = False

        with cls._mx:
            for peer in cls._peers:
                try:
                    now_us = time.time_ns() // 1000
                    if peer._ready is True:
                        if now_us - peer._last_ping_at >= 5000000:  # 5 seconds  TODO: Make variable
                            cls.send_ping(peer)
                    msg = peer._recv_messages.get(False)
                    cls.process_message(peer, msg)
                    processed = True
                except queue.Empty:
                    pass
                except Exception as e:
                    logging.warning('process message error %s', str(e))
                    if cls._sc.debug:
                        logging.error(traceback.format_exc())

        if processed is False:
            time.sleep(timeout)


class Network:
    __slots__ = (
        '_p2p_host', '_p2p_port', '_network_key', '_network_pubkey',
        '_sc', '_peers', '_max_connections', '_running', '_network_thread', '_msg_thread',
        '_mx', '_socket', '_read_sockets', '_write_sockets', '_error_sockets', '_csprng', '_seen_ephem_keys')

    def __init__(self, p2p_host, p2p_port, network_key, swap_client):
        self._p2p_host = p2p_host
        self._p2p_port = p2p_port
        self._network_key = network_key
        self._network_pubkey = PublicKey.from_secret(network_key).format()
        self._sc = swap_client
        self._peers = []

        self._max_connections = 10
        self._running = False

        self._network_thread = None
        self._msg_thread = None
        self._mx = threading.Lock()
        self._socket = None
        self._read_sockets = []
        self._write_sockets = []
        self._error_sockets = []  # Check for error events
        self._seen_ephem_keys = OrderedDict()

    def startNetwork(self):
        self._mx.acquire()
        try:
            self._csprng = rfc6979_hmac_sha256_initialize(secrets.token_bytes(32))

            self._running = True
            self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self._socket.bind((self._p2p_host, self._p2p_port))
            self._socket.listen(self._max_connections)
            self._read_sockets.append(self._socket)

            self._network_thread = threading.Thread(target=listen_thread, args=(self,))
            self._network_thread.start()

            self._msg_thread = threading.Thread(target=msg_thread, args=(self,))
            self._msg_thread.start()
        finally:
            self._mx.release()

    def stopNetwork(self):
        self._mx.acquire()
        try:
            self._running = False
        finally:
            self._mx.release()

        if self._network_thread:
            self._network_thread.join()
        if self._msg_thread:
            self._msg_thread.join()

        self._mx.acquire()
        try:
            if self._socket:
                self._socket.close()

            for peer in self._peers:
                peer.close()
        finally:
            self._mx.release()

    def add_connection(self, host, port, peer_pubkey):
        self._sc.log.info('Connecting from %s to %s at %s %d', self._network_pubkey.hex(), peer_pubkey.hex(), host, port)
        self._mx.acquire()
        try:
            address = (host, port)
            peer_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            peer_socket.connect(address)
            peer = Peer(address, peer_socket, peer_pubkey)
            self._peers.append(peer)
            self._error_sockets.append(peer_socket)
            self._read_sockets.append(peer_socket)
        finally:
            self._mx.release()

        self.send_handshake(peer)

    def disconnect(self, peer):
        self._sc.log.info('Closing peer socket %s', peer._address)
        self._read_sockets.pop(self._read_sockets.index(peer._socket))
        self._error_sockets.pop(self._error_sockets.index(peer._socket))
        peer.close()
        self._peers.pop(self._peers.index(peer))

    def check_handshake_ephem_key(self, peer, timestamp, ephem_pk, direction=1):
        # assert ._mx.acquire() ?

        used = self._seen_ephem_keys.get(ephem_pk)
        if used:
            raise ValueError('Handshake ephem_pk reused %s peer %s', 'for' if direction == 1 else 'by', used[0])

        self._seen_ephem_keys[ephem_pk] = (peer._address, timestamp)

        while len(self._seen_ephem_keys) > MAX_SEEN_EPHEM_KEYS:
            self._seen_ephem_keys.popitem(last=False)

    def send_handshake(self, peer):
        self._sc.log.debug('send_handshake %s', peer._address)
        peer._mx.acquire()
        try:
            # TODO: Drain peer._recv_messages
            if not peer._recv_messages.empty():
                self._sc.log.warning('send_handshake %s - Receive queue dumped.', peer._address)
                while not peer._recv_messages.empty():
                    peer._recv_messages.get(False)

            msg = MsgHandshake()

            msg._timestamp = int(time.time())
            key_r = rfc6979_hmac_sha256_generate(self._csprng, 32)
            k = PrivateKey(key_r)
            msg._ephem_pk = PublicKey.from_secret(key_r).format()
            self.check_handshake_ephem_key(peer, msg._timestamp, msg._ephem_pk)

            ss = k.ecdh(peer._pubkey)

            hashed = hashlib.sha512(ss + struct.pack('>Q', msg._timestamp)).digest()
            peer._ke = hashed[:32]
            peer._km = hashed[32:]

            nonce = peer._km[24:]

            payload = self._sc._version

            nk = PrivateKey(self._network_key)
            sig = nk.sign_recoverable(peer._km)
            payload += sig

            aad = msg.encode_aad()
            aad += nonce
            cipher = ChaCha20_Poly1305.new(key=peer._ke, nonce=nonce)
            cipher.update(aad)
            msg._ct, msg._mac = cipher.encrypt_and_digest(payload)

            peer._sent_nonce = hashlib.sha256(nonce + msg._mac).digest()
            peer._recv_nonce = hashlib.sha256(peer._km).digest()  # Init nonce

            peer._last_handshake_at = msg._timestamp
            peer._ready = False  # Wait for peer to complete handshake

            self.send_msg(peer, msg)
        finally:
            peer._mx.release()

    def process_handshake(self, peer, msg_mv):
        self._sc.log.debug('process_handshake %s', peer._address)

        # TODO: Drain peer._recv_messages
        if not peer._recv_messages.empty():
            self._sc.log.warning('process_handshake %s - Receive queue dumped.', peer._address)
            while not peer._recv_messages.empty():
                peer._recv_messages.get(False)

        msg = MsgHandshake()
        msg.decode(msg_mv)

        try:
            now = int(time.time())
            if now - peer._last_handshake_at < 30:
                raise ValueError('Too many handshakes from peer %s', peer._address)

            if abs(msg._timestamp - now) > TIMESTAMP_LEEWAY:
                raise ValueError('Bad handshake timestamp from peer %s', peer._address)

            self.check_handshake_ephem_key(peer, msg._timestamp, msg._ephem_pk, direction=2)

            nk = PrivateKey(self._network_key)
            ss = nk.ecdh(msg._ephem_pk)

            hashed = hashlib.sha512(ss + struct.pack('>Q', msg._timestamp)).digest()
            peer._ke = hashed[:32]
            peer._km = hashed[32:]

            nonce = peer._km[24:]

            aad = msg.encode_aad()
            aad += nonce
            cipher = ChaCha20_Poly1305.new(key=peer._ke, nonce=nonce)
            cipher.update(aad)
            plaintext = cipher.decrypt_and_verify(msg._ct, msg._mac)  # Will raise error if mac doesn't match

            peer._version = plaintext[:6]
            sig = plaintext[6:]

            pk_peer = PublicKey.from_signature_and_message(sig, peer._km)
            # TODO: Should pk_peer be linked to public data?

            peer._pubkey = pk_peer.format()
            peer._recv_nonce = hashlib.sha256(nonce + msg._mac).digest()
            peer._sent_nonce = hashlib.sha256(peer._km).digest()  # Init nonce

            peer._last_handshake_at = msg._timestamp
            peer._ready = True
            # Schedule a ping to complete the handshake, TODO: Send here?
            peer._last_ping_at = 0

        except Exception as e:
            # TODO: misbehaving
            self._sc.log.debug('[rm] process_handshake %s', str(e))

    def process_ping(self, peer, msg_mv):
        nonce = peer._recv_nonce[:24]

        cipher = ChaCha20_Poly1305.new(key=peer._ke, nonce=nonce)
        cipher.update(msg_mv[0: 2])
        cipher.update(nonce)

        mac = msg_mv[-16:]
        plaintext = cipher.decrypt_and_verify(msg_mv[2: -16], mac)

        ping_nonce = struct.unpack('>I', plaintext[:4])[0]
        # Version is added to a ping following a handshake message
        if len(plaintext) >= 10:
            peer._ready = True
            version = plaintext[4: 10]
            if peer._version is None:
                peer._version = version
                self._sc.log.debug('Set version from ping %s, %s', peer._pubkey.hex(), peer._version.hex())

        peer._recv_nonce = hashlib.sha256(nonce + mac).digest()

        self.send_pong(peer, ping_nonce)

    def process_pong(self, peer, msg_mv):
        nonce = peer._recv_nonce[:24]

        cipher = ChaCha20_Poly1305.new(key=peer._ke, nonce=nonce)
        cipher.update(msg_mv[0: 2])
        cipher.update(nonce)

        mac = msg_mv[-16:]
        plaintext = cipher.decrypt_and_verify(msg_mv[2: -16], mac)

        pong_nonce = struct.unpack('>I', plaintext[:4])[0]

        if pong_nonce == peer._ping_nonce:
            peer._last_ping_rtt = (time.time_ns() // 1000) - peer._last_ping_at
        else:
            self._sc.log.debug('Pong received out of order %s', peer._address)

        peer._recv_nonce = hashlib.sha256(nonce + mac).digest()

    def send_ping(self, peer):
        ping_nonce = random.getrandbits(32)

        msg_bytes = struct.pack('>H', NetMessageTypes.PING)
        nonce = peer._sent_nonce[:24]

        cipher = ChaCha20_Poly1305.new(key=peer._ke, nonce=nonce)
        cipher.update(msg_bytes)
        cipher.update(nonce)

        payload = struct.pack('>I', ping_nonce)
        if peer._last_ping_at == 0:
            payload += self._sc._version
        ct, mac = cipher.encrypt_and_digest(payload)

        msg_bytes += ct + mac

        peer._sent_nonce = hashlib.sha256(nonce + mac).digest()

        peer._last_ping_at = time.time_ns() // 1000
        peer._ping_nonce = ping_nonce

        self.send_msg(peer, msg_bytes)

    def send_pong(self, peer, ping_nonce):
        msg_bytes = struct.pack('>H', NetMessageTypes.PONG)
        nonce = peer._sent_nonce[:24]

        cipher = ChaCha20_Poly1305.new(key=peer._ke, nonce=nonce)
        cipher.update(msg_bytes)
        cipher.update(nonce)

        payload = struct.pack('>I', ping_nonce)
        ct, mac = cipher.encrypt_and_digest(payload)
        msg_bytes += ct + mac

        peer._sent_nonce = hashlib.sha256(nonce + mac).digest()

        self.send_msg(peer, msg_bytes)

    def send_msg(self, peer, msg):
        msg_encoded = msg if isinstance(msg, bytes) else msg.encode()
        len_encoded = len(msg_encoded)

        msg_packed = bytearray(MSG_START_TOKEN) + struct.pack('>I', len_encoded) + msg_encoded
        peer._socket.sendall(msg_packed)

        peer._bytes_sent += len_encoded

    def process_message(self, peer, msg_bytes):
        logging.info('[rm] process_message %s len %d', peer._address, len(msg_bytes))

        peer._mx.acquire()
        try:
            mv = memoryview(msg_bytes)
            o = 0
            msg_type = struct.unpack('>H', mv[o: o + 2])[0]
            if msg_type == NetMessageTypes.HANDSHAKE:
                self.process_handshake(peer, mv)
            elif msg_type == NetMessageTypes.PING:
                self.process_ping(peer, mv)
            elif msg_type == NetMessageTypes.PONG:
                self.process_pong(peer, mv)
            else:
                self._sc.log.debug('Unknown message type %d', msg_type)
        finally:
            peer._mx.release()

    def receive_bytes(self, peer, bytes_recv):
        # logging.info('[rm] receive_bytes %s %s', peer._address, bytes_recv)

        len_received = len(bytes_recv)
        peer._last_received_at = time.time()
        peer._bytes_received += len_received

        invalid_msg = False
        mv = memoryview(bytes_recv)

        o = 0
        try:
            while o < len_received:
                if peer._receiving_length == 0:
                    if len(bytes_recv) < MSG_HEADER_LEN:
                        raise ValueError('Msg too short')

                    if mv[o: o + 2] != MSG_START_TOKEN:
                        raise ValueError('Invalid start token')
                    o += 2

                    msg_len = struct.unpack('>I', mv[o: o + 4])[0]
                    o += 4
                    if msg_len < 2 or msg_len > MSG_MAX_SIZE:
                        raise ValueError('Invalid data length')

                    # Precheck msg_type
                    msg_type = struct.unpack('>H', mv[o: o + 2])[0]
                    # o += 2  # Don't inc offset, msg includes type
                    if not NetMessageTypes.has_value(msg_type):
                        raise ValueError('Invalid msg type')

                    peer._receiving_length = msg_len
                    len_pkt = (len_received - o)
                    nc = msg_len if len_pkt > msg_len else len_pkt

                    peer._receiving_buffer = mv[o: o + nc]
                    o += nc
                else:
                    len_to_go = peer._receiving_length - len(peer._receiving_buffer)
                    len_pkt = (len_received - o)
                    nc = len_to_go if len_pkt > len_to_go else len_pkt
                    peer._receiving_buffer = mv[o: o + nc]
                    o += nc
                if len(peer._receiving_buffer) == peer._receiving_length:
                    peer._recv_messages.put(peer._receiving_buffer)
                    peer._receiving_length = 0

        except Exception as e:
            if self._sc.debug:
                self._sc.log.error('Invalid message received from %s %s', peer._address, str(e))
            # TODO: misbehaving

    def test_onion(self, path):
        self._sc.log.debug('test_onion packet')

    def get_info(self):
        rv = {}

        peers = []
        with self._mx:
            for peer in self._peers:
                peer_info = {
                    'pubkey': 'Unknown' if not peer._pubkey else peer._pubkey.hex(),
                    'address': '{}:{}'.format(peer._address[0], peer._address[1]),
                    'bytessent': peer._bytes_sent,
                    'bytesrecv': peer._bytes_received,
                    'ready': peer._ready,
                    'incoming': peer._incoming,
                }
                peers.append(peer_info)

        rv['peers'] = peers
        return rv