monero.fail/xmrnodes/helpers.py

import sys
import socket
import pickle
from os import path

import geoip2.database
from requests import get as r_get
from urllib.parse import urlparse
from levin.bucket import Bucket
from levin.ctypes import *
from levin.constants import LEVIN_SIGNATURE

from xmrnodes.models import Node
from xmrnodes import config


def make_request(url: str, path="/get_info", data=None):
    headers = {"Origin": "https://monero.fail"}
    if is_onion(url):
        _p = f"socks5h://{config.TOR_HOST}:{config.TOR_PORT}"
        proxies = {"http": _p, "https": _p}
        timeout = 30
    elif is_i2p(url):
        _p = f"http://{config.I2P_HOST}:{config.I2P_PORT}"
        proxies = {"http": _p, "https": _p}
        timeout = 30
    else:
        proxies = None
        timeout = 10
    r = r_get(
        url + path,
        timeout=timeout,
        proxies=proxies,
        json=data,
        headers=headers,
        verify=False,
    )
    r.raise_for_status()
    return r


def determine_crypto(url):
    data = {"method": "get_block_header_by_height", "params": {"height": 0}}
    hashes = {
        "monero": [
            "418015bb9ae982a1975da7d79277c2705727a56894ba0fb246adaabb1f4632e3",  # mainnet
            "48ca7cd3c8de5b6a4d53d2861fbdaedca141553559f9be9520068053cda8430b",  # testnet
            "76ee3cc98646292206cd3e86f74d88b4dcc1d937088645e9b0cbca84b7ce74eb",  # stagenet
        ],
        "wownero": [
            "a3fd635dd5cb55700317783469ba749b5259f0eeac2420ab2c27eb3ff5ffdc5c",  # mainnet
            "d81a24c7aad4628e5c9129f8f2ec85888885b28cf468597a9762c3945e9f29aa",  # testnet
        ],
    }
    try:
        r = make_request(url, "/json_rpc", data)
        assert "result" in r.json()
        hash = r.json()["result"]["block_header"]["hash"]
        crypto = "unknown"
        for c, h in hashes.items():
            if hash in h:
                crypto = c
                break
        return crypto
    except:
        return "unknown"

def is_i2p(url: str):
    _split = url.split(":")
    if len(_split) < 2:
        return False
    if _split[1].endswith(".i2p"):
        return True
    else:
        return False

def is_onion(url: str):
    _split = url.split(":")
    if len(_split) < 2:
        return False
    if _split[1].endswith(".onion"):
        return True
    else:
        return False

def retrieve_peers(host, port):
    try:
        sock = socket.socket()
        sock.settimeout(5)
        sock.connect((host, int(port)))
    except Exception as e:
        return None

    bucket = Bucket.create_handshake_request()

    sock.send(bucket.header())
    sock.send(bucket.payload())

    buckets = []
    peers = []

    while 1:
        buffer = sock.recv(8)
        if not buffer:
            break

        if not buffer.startswith(bytes(LEVIN_SIGNATURE)):
            break

        bucket = Bucket.from_buffer(signature=buffer, sock=sock)
        buckets.append(bucket)

        if bucket.command == 1001:
            _peers = bucket.get_peers() or []

            for peer in _peers:
                try:
                    peers.append("http://%s:%d" % (peer["ip"].ip, peer["port"].value))
                except:
                    pass

            sock.close()
            break

    if peers:
        return peers
    else:
        return None


def get_highest_block(nettype, crypto):
    highest = (
        Node.select()
        .where(Node.validated == True, Node.nettype == nettype, Node.crypto == crypto)
        .order_by(Node.last_height.desc())
        .limit(1)
        .first()
    )
    if highest:
        return highest.last_height
    else:
        return 0


def get_geoip(ip_or_dns):
    host = urlparse(ip_or_dns).netloc.split(':')[0]
    resolved = socket.gethostbyname(host)
    host = host if resolved == host else resolved
    with geoip2.database.Reader("./data/GeoLite2-City.mmdb") as reader:
        return reader.city(host)

def get_whois(ip_or_dns):
    pass

# asn
# asn_cidr
# asn_country_code
# asn_description