Bitmessage/PyBitmessage-2024-12-10
Archived
2
0
Fork 0
Code Issues 168 Pull Requests 53 Actions Packages Projects Releases 5 Wiki Activity

formatted, refactored & cleaned code

Browse Source
This commit is contained in:
cis-kuldeep 2021-07-28 18:32:27 +05:30
parent 4a5645d9aa
commit 078c183ae9
No known key found for this signature in database
GPG Key ID: 67B47D8A06FA45E4
1 changed files with 85 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
src/protocol.py
View File

@ -9,6 +9,7 @@ import hashlib
import random import random
import socket import socket
import sys import sys
import six
import time import time
from binascii import hexlify from binascii import hexlify
from struct import Struct, pack, unpack from struct import Struct, pack, unpack
@ -36,8 +37,6 @@ except ImportError:
from .helper_sql import sqlExecute from .helper_sql import sqlExecute
from .version import softwareVersion from .version import softwareVersion
PY3 = sys.version_info[0] >= 3
# Service flags # Service flags
#: This is a normal network node #: This is a normal network node
NODE_NETWORK = 1 NODE_NETWORK = 1
@ -107,26 +106,20 @@ def isBitSetWithinBitfield(fourByteString, n):
# ip addresses # ip addresses
if PY3: def encodeHost(host):
def encodeHost(host):
"""Encode a given host to be used in low-level socket operations""" """Encode a given host to be used in low-level socket operations"""
if host.find('.onion') > -1: if six.PY2:
return b'\xfd\x87\xd8\x7e\xeb\x43' + base64.b32decode( ONION_PREFIX = '\xfd\x87\xd8\x7e\xeb\x43'
host.split(".")[0], True) IP_PREFIX = '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF'
elif host.find(':') == -1: else:
return b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF' + \ ONION_PREFIX = b'\xfd\x87\xd8\x7e\xeb\x43'
socket.inet_aton(host) IP_PREFIX = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF'
return socket.inet_pton(socket.AF_INET6, host)
else:
def encodeHost(host):
"""Encode a given host to be used in low-level socket operations"""
if host.find('.onion') > -1: if host.find('.onion') > -1:
return '\xfd\x87\xd8\x7e\xeb\x43' + base64.b32decode( return ONION_PREFIX + base64.b32decode(
host.split(".")[0], True) host.split(".")[0], True)
elif host.find(':') == -1: elif host.find(':') == -1:
return '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF' + \ return IP_PREFIX + socket.inet_aton(host)
socket.inet_aton(host)
return socket.inet_pton(socket.AF_INET6, host) return socket.inet_pton(socket.AF_INET6, host)
@ -167,42 +160,22 @@ def network_group(host):
return network_type return network_type
if PY3: def checkIPAddress(host, private=False):
def checkIPAddress(host, private=False):
""" """
Returns hostStandardFormat if it is a valid IP address, Returns hostStandardFormat if it is a valid IP address,
otherwise returns False otherwise returns False
""" """
if host[0:12] == b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF': if six.PY3:
hostStandardFormat = socket.inet_ntop(socket.AF_INET, host[12:]) IP_PREFIX1 = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF'
return checkIPv4Address(host[12:], hostStandardFormat, private) IP_PREFIX2 = b'\xfd\x87\xd8\x7e\xeb\x43'
elif host[0:6] == b'\xfd\x87\xd8\x7e\xeb\x43':
# Onion, based on BMD/bitcoind
hostStandardFormat = base64.b32encode(host[6:]).lower() + ".onion"
if private:
return False
return hostStandardFormat
else: else:
try: IP_PREFIX1 = '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF'
hostStandardFormat = socket.inet_ntop(socket.AF_INET6, host) IP_PREFIX2 = '\xfd\x87\xd8\x7e\xeb\x43'
except ValueError:
return False
if hostStandardFormat == "":
# This can happen on Windows systems which are
# not 64-bit compatible so let us drop the IPv6 address.
return False
return checkIPv6Address(host, hostStandardFormat, private)
else: if host[0:12] == IP_PREFIX1:
def checkIPAddress(host, private=False):
"""
Returns hostStandardFormat if it is a valid IP address,
otherwise returns False
"""
if host[0:12] == '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF':
hostStandardFormat = socket.inet_ntop(socket.AF_INET, host[12:]) hostStandardFormat = socket.inet_ntop(socket.AF_INET, host[12:])
return checkIPv4Address(host[12:], hostStandardFormat, private) return checkIPv4Address(host[12:], hostStandardFormat, private)
elif host[0:6] == '\xfd\x87\xd8\x7e\xeb\x43': elif host[0:6] == IP_PREFIX2:
# Onion, based on BMD/bitcoind # Onion, based on BMD/bitcoind
hostStandardFormat = base64.b32encode(host[6:]).lower() + ".onion" hostStandardFormat = base64.b32encode(host[6:]).lower() + ".onion"
if private: if private:
@ -220,58 +193,41 @@ else:
return checkIPv6Address(host, hostStandardFormat, private) return checkIPv6Address(host, hostStandardFormat, private)
if PY3: def checkIPv4Address(host, hostStandardFormat, private=False):
def checkIPv4Address(host, hostStandardFormat, private=False):
""" """
Returns hostStandardFormat if it is an IPv4 address, Returns hostStandardFormat if it is an IPv4 address,
otherwise returns False otherwise returns False
""" """
if host[0] == 127: # 127/8 if six.PY2:
if not private: IP_OCT1 = '\x7F' # 127/8
logger.debug( IP_OCT2 = '\x0A' # 10/8
'Ignoring IP address in loopback range: %s', IP_OCT3 = '\xC0\xA8' # 192.168/16
hostStandardFormat) IP_OCT4 = '\xAC\x10' # 172.16
return hostStandardFormat if private else False IP_OCT5 = '\xAC\x20' # 12
if host[0] == 10: # 10/8 else:
if not private: IP_OCT1 = 127 # 127/8
logger.debug( IP_OCT2 = 10 # 10/8
'Ignoring IP address in private range: %s', hostStandardFormat) IP_OCT3 = b'\xC0\xA8' # 192.168/16
return hostStandardFormat if private else False IP_OCT4 = b'\xAC\x10' # 172.16
if host[0:2] == b'\xC0\xA8': # 192.168/16 IP_OCT5 = b'\xAC\x20' # 12
if not private:
logger.debug(
'Ignoring IP address in private range: %s', hostStandardFormat)
return hostStandardFormat if private else False
if host[0:2] >= b'\xAC\x10' and host[0:2] < '\xAC\x20': # 172.16/12
if not private:
logger.debug(
'Ignoring IP address in private range: %s', hostStandardFormat)
return hostStandardFormat if private else False
return False if private else hostStandardFormat
else: if host[0] == IP_OCT1:
def checkIPv4Address(host, hostStandardFormat, private=False):
"""
Returns hostStandardFormat if it is an IPv4 address,
otherwise returns False
"""
if host[0] == '\x7F': # 127/8
if not private: if not private:
logger.debug( logger.debug(
'Ignoring IP address in loopback range: %s', 'Ignoring IP address in loopback range: %s',
hostStandardFormat) hostStandardFormat)
return hostStandardFormat if private else False return hostStandardFormat if private else False
if host[0] == '\x0A': # 10/8 if host[0] == IP_OCT2:
if not private: if not private:
logger.debug( logger.debug(
'Ignoring IP address in private range: %s', hostStandardFormat) 'Ignoring IP address in private range: %s', hostStandardFormat)
return hostStandardFormat if private else False return hostStandardFormat if private else False
if host[0:2] == '\xC0\xA8': # 192.168/16 if host[0:2] == IP_OCT3:
if not private: if not private:
logger.debug( logger.debug(
'Ignoring IP address in private range: %s', hostStandardFormat) 'Ignoring IP address in private range: %s', hostStandardFormat)
return hostStandardFormat if private else False return hostStandardFormat if private else False
if host[0:2] >= '\xAC\x10' and host[0:2] < '\xAC\x20': # 172.16/12 if host[0:2] >= IP_OCT4 and host[0:2] < IP_OCT5:
if not private: if not private:
logger.debug( logger.debug(
'Ignoring IP address in private range: %s', hostStandardFormat) 'Ignoring IP address in private range: %s', hostStandardFormat)