diff options
Diffstat (limited to 'test/functional/test_framework/messages.py')
| -rwxr-xr-x | test/functional/test_framework/messages.py | 30 |
1 files changed, 24 insertions, 6 deletions
diff --git a/test/functional/test_framework/messages.py b/test/functional/test_framework/messages.py index 504c8c70d4..065e8961ae 100755 --- a/test/functional/test_framework/messages.py +++ b/test/functional/test_framework/messages.py @@ -18,6 +18,7 @@ ser_*, deser_*: functions that handle serialization/deserialization. Classes use __slots__ to ensure extraneous attributes aren't accidentally added by tests, compromising their intended effect. """ +from base64 import b32decode, b32encode from codecs import encode import copy import hashlib @@ -213,15 +214,20 @@ class CAddress: # see https://github.com/bitcoin/bips/blob/master/bip-0155.mediawiki NET_IPV4 = 1 + NET_I2P = 5 ADDRV2_NET_NAME = { - NET_IPV4: "IPv4" + NET_IPV4: "IPv4", + NET_I2P: "I2P" } ADDRV2_ADDRESS_LENGTH = { - NET_IPV4: 4 + NET_IPV4: 4, + NET_I2P: 32 } + I2P_PAD = "====" + def __init__(self): self.time = 0 self.nServices = 1 @@ -229,6 +235,9 @@ class CAddress: self.ip = "0.0.0.0" self.port = 0 + def __eq__(self, other): + return self.net == other.net and self.ip == other.ip and self.nServices == other.nServices and self.port == other.port and self.time == other.time + def deserialize(self, f, *, with_time=True): """Deserialize from addrv1 format (pre-BIP155)""" if with_time: @@ -261,24 +270,33 @@ class CAddress: self.nServices = deser_compact_size(f) self.net = struct.unpack("B", f.read(1))[0] - assert self.net == self.NET_IPV4 + assert self.net in (self.NET_IPV4, self.NET_I2P) address_length = deser_compact_size(f) assert address_length == self.ADDRV2_ADDRESS_LENGTH[self.net] - self.ip = socket.inet_ntoa(f.read(4)) + addr_bytes = f.read(address_length) + if self.net == self.NET_IPV4: + self.ip = socket.inet_ntoa(addr_bytes) + else: + self.ip = b32encode(addr_bytes)[0:-len(self.I2P_PAD)].decode("ascii").lower() + ".b32.i2p" self.port = struct.unpack(">H", f.read(2))[0] def serialize_v2(self): """Serialize in addrv2 format (BIP155)""" - assert self.net == self.NET_IPV4 + assert self.net in (self.NET_IPV4, self.NET_I2P) r = b"" r += struct.pack("<I", self.time) r += ser_compact_size(self.nServices) r += struct.pack("B", self.net) r += ser_compact_size(self.ADDRV2_ADDRESS_LENGTH[self.net]) - r += socket.inet_aton(self.ip) + if self.net == self.NET_IPV4: + r += socket.inet_aton(self.ip) + else: + sfx = ".b32.i2p" + assert self.ip.endswith(sfx) + r += b32decode(self.ip[0:-len(sfx)] + self.I2P_PAD, True) r += struct.pack(">H", self.port) return r |