Inside Bitcoin

251    use obj.serialize().hex()"""
252    obj.deserialize(BytesIO(bytes.fromhex(hex_string)))
253    return obj
254
255
256def tx_from_hex(hex_string):
257    """Deserialize from hex string to a transaction object"""
258    return from_hex(CTransaction(), hex_string)
259
260
261def malleate_tx_to_invalid_witness(tx):
262    """
263    Create a malleated version of the tx where the witness is replaced with garbage data.
264    Returns a CTransaction object.
265    """
266    tx_bad_wit = tx_from_hex(tx["hex"])
267    tx_bad_wit.wit.vtxinwit = [CTxInWitness()]
268    # Add garbage data to witness 0. We cannot simply strip the witness, as the node would
269    # classify it as a transaction in which the witness was missing rather than wrong.
270    tx_bad_wit.wit.vtxinwit[0].scriptWitness.stack = [b'garbage']
271
272    assert_equal(tx["txid"], tx_bad_wit.txid_hex)
273    assert_not_equal(tx["wtxid"], tx_bad_wit.wtxid_hex)
274
275    return tx_bad_wit
276
277
278# like from_hex, but without the hex part
279def from_binary(cls, stream):
280    """deserialize a binary stream (or bytes object) into an object"""
281    # handle bytes object by turning it into a stream
282    was_bytes = isinstance(stream, bytes)
283    if was_bytes:
284        stream = BytesIO(stream)
285    obj = cls()
286    obj.deserialize(stream)
287    if was_bytes:
288        assert_equal(len(stream.read()), 0)
289    return obj
290
291
292# Objects that map to bitcoind objects, which can be serialized/deserialized
293
294
295class CAddress:
296    __slots__ = ("net", "ip", "nServices", "port", "time")
297
298    # see https://github.com/bitcoin/bips/blob/master/bip-0155.mediawiki
299    NET_IPV4 = 1
300    NET_IPV6 = 2
301    NET_TORV3 = 4
302    NET_I2P = 5
303    NET_CJDNS = 6
304
305    ADDRV2_NET_NAME = {
306        NET_IPV4: "IPv4",
307        NET_IPV6: "IPv6",
308        NET_TORV3: "TorV3",
309        NET_I2P: "I2P",
310        NET_CJDNS: "CJDNS"
311    }
312
313    ADDRV2_ADDRESS_LENGTH = {
314        NET_IPV4: 4,
315        NET_IPV6: 16,
316        NET_TORV3: 32,
317        NET_I2P: 32,
318        NET_CJDNS: 16
319    }
320
321    I2P_PAD = "===="
322
323    def __init__(self):
324        self.time = 0
325        self.nServices = 1
326        self.net = self.NET_IPV4
327        self.ip = "0.0.0.0"
328        self.port = 0
329
330    def __eq__(self, other):
331        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
332
333    def deserialize(self, f, *, with_time=True):
334        """Deserialize from addrv1 format (pre-BIP155)"""
335        if with_time:
336            # VERSION messages serialize CAddress objects without time
337            self.time = int.from_bytes(f.read(4), "little")
338        self.nServices = int.from_bytes(f.read(8), "little")
339        # We only support IPv4 which means skip 12 bytes and read the next 4 as IPv4 address.
340        f.read(12)
341        self.net = self.NET_IPV4
342        self.ip = socket.inet_ntoa(f.read(4))
343        self.port = int.from_bytes(f.read(2), "big")
344
345    def serialize(self, *, with_time=True):
346        """Serialize in addrv1 format (pre-BIP155)"""
347        assert_equal(self.net, self.NET_IPV4)
348        r = b""
349        if with_time:
350            # VERSION messages serialize CAddress objects without time
351            r += self.time.to_bytes(4, "little")
352        r += self.nServices.to_bytes(8, "little")
353        r += b"\x00" * 10 + b"\xff" * 2
354        r += socket.inet_aton(self.ip)
355        r += self.port.to_bytes(2, "big")
356        return r
357
358    def deserialize_v2(self, f):
359        """Deserialize from addrv2 format (BIP155)"""
360        self.time = int.from_bytes(f.read(4), "little")
361
362        self.nServices = deser_compact_size(f)
363
364        self.net = int.from_bytes(f.read(1), "little")
365        assert self.net in self.ADDRV2_NET_NAME
366
367        address_length = deser_compact_size(f)
368        assert_equal(address_length, self.ADDRV2_ADDRESS_LENGTH[self.net])
369
370        addr_bytes = f.read(address_length)
371        if self.net == self.NET_IPV4:
372            self.ip = socket.inet_ntoa(addr_bytes)
373        elif self.net == self.NET_IPV6:
374            self.ip = socket.inet_ntop(socket.AF_INET6, addr_bytes)
375        elif self.net == self.NET_TORV3:
376            prefix = b".onion checksum"
377            version = bytes([3])
378            checksum = sha3(prefix + addr_bytes + version)[:2]
379            self.ip = b32encode(addr_bytes + checksum + version).decode("ascii").lower() + ".onion"
380        elif self.net == self.NET_I2P:
381            self.ip = b32encode(addr_bytes)[0:-len(self.I2P_PAD)].decode("ascii").lower() + ".b32.i2p"
382        elif self.net == self.NET_CJDNS:
383            self.ip = socket.inet_ntop(socket.AF_INET6, addr_bytes)
384        else:
385            raise Exception("Address type not supported")
386
387        self.port = int.from_bytes(f.read(2), "big")
388
389    def serialize_v2(self):
390        """Serialize in addrv2 format (BIP155)"""
391        assert self.net in self.ADDRV2_NET_NAME
392        r = b""
393        r += self.time.to_bytes(4, "little")
394        r += ser_compact_size(self.nServices)
395        r += self.net.to_bytes(1, "little")
396        r += ser_compact_size(self.ADDRV2_ADDRESS_LENGTH[self.net])
397        if self.net == self.NET_IPV4:
398            r += socket.inet_aton(self.ip)
399        elif self.net == self.NET_IPV6:
400            r += socket.inet_pton(socket.AF_INET6, self.ip)
401        elif self.net == self.NET_TORV3:
402            sfx = ".onion"
403            assert self.ip.endswith(sfx)
404            r += b32decode(self.ip[0:-len(sfx)], True)[0:32]
405        elif self.net == self.NET_I2P:
406            sfx = ".b32.i2p"
407            assert self.ip.endswith(sfx)
408            r += b32decode(self.ip[0:-len(sfx)] + self.I2P_PAD, True)
409        elif self.net == self.NET_CJDNS:
410            r += socket.inet_pton(socket.AF_INET6, self.ip)
411        else:
412            raise Exception("Address type not supported")
413        r += self.port.to_bytes(2, "big")
414        return r
415
416    def __repr__(self):
417        return ("CAddress(nServices=%i net=%s addr=%s port=%i)"
418                % (self.nServices, self.ADDRV2_NET_NAME[self.net], self.ip, self.port))
419
420
421class CInv:
422    __slots__ = ("hash", "type")
423
424    typemap = {
425        0: "Error",
426        MSG_TX: "TX",
427        MSG_BLOCK: "Block",
428        MSG_TX | MSG_WITNESS_FLAG: "WitnessTx",
429        MSG_BLOCK | MSG_WITNESS_FLAG: "WitnessBlock",
430        MSG_FILTERED_BLOCK: "filtered Block",
431        MSG_CMPCT_BLOCK: "CompactBlock",
432        MSG_WTX: "WTX",
433    }
434
435    def __init__(self, t=0, h=0):
436        self.type = t
437        self.hash = h
438
439    def deserialize(self, f):
440        self.type = int.from_bytes(f.read(4), "little")
441        self.hash = deser_uint256(f)
442
443    def serialize(self):
444        r = b""
445        r += self.type.to_bytes(4, "little")
446        r += ser_uint256(self.hash)
447        return r
448
449    def __repr__(self):
450        return "CInv(type=%s hash=%064x)" \
451            % (self.typemap[self.type], self.hash)
452
453    def __eq__(self, other):
454        return isinstance(other, CInv) and self.hash == other.hash and self.type == other.type
455
456
457class CBlockLocator:
458    __slots__ = ("nVersion", "vHave")
459
460    def __init__(self):
461        self.vHave = []
462
463    def deserialize(self, f):
464        int.from_bytes(f.read(4), "little", signed=True)  # Ignore version field.
465        self.vHave = deser_uint256_vector(f)
466
467    def serialize(self):
468        r = b""
469        r += (0).to_bytes(4, "little", signed=True)  # Bitcoin Core ignores the version field. Set it to 0.
470        r += ser_uint256_vector(self.vHave)
471        return r
472
473    def __repr__(self):
474        return "CBlockLocator(vHave=%s)" % (repr(self.vHave))
475
476
477class COutPoint:
478    __slots__ = ("hash", "n")
479
480    def __init__(self, hash=0, n=0):
481        self.hash = hash
482        self.n = n
483
484    def deserialize(self, f):
485        self.hash = deser_uint256(f)
486        self.n = int.from_bytes(f.read(4), "little")
487
488    def serialize(self):
489        r = b""
490        r += ser_uint256(self.hash)
491        r += self.n.to_bytes(4, "little")
492        return r
493
494    def __repr__(self):
495        return "COutPoint(hash=%064x n=%i)" % (self.hash, self.n)
496
497
498class CTxIn:
499    __slots__ = ("nSequence", "prevout", "scriptSig")
500