세그윗

#트랜잭션 클래스를 수정한다. class Tx: command = b'tx' #세그윗 필드를 추가한다. def __init__(self, version, tx_ins, tx_outs, locktime, testnet=False, segwit=False): self.version = version self.tx_ins = tx_ins self.tx_outs = tx_outs self.locktime = locktime self.testnet = testnet self.segwit = segwit self._hash_prevouts = None self._hash_sequence = None self._hash_outputs = None #...생략 @classmethod #세그윗 파싱을 위한 함수 #기존의 parse는 parse_legacy로 변경하였다. def parse(cls, s, testnet=False): #세그윗 마커 확인 s.read(4) #세그윗 트랜잭션 확인. 다섯번째 바이트가 0이면 세그윗이다. if s.read(1) == b'\x00': #세그윗 파싱을 실행한다. parse_method = cls.parse_segwit else: parse_method = cls.parse_legacy #스트림 포인터를 원위치한다. s.seek(-5, 1) return parse_method(s, testnet=testnet) @classmethod def parse_segwit(cls, s, testnet=False): version = little_endian_to_int(s.read(4)) marker = s.read(2) #세그윗 마커 if marker != b'\x00\x01': raise RuntimeError('Not a segwit transaction {}'.format(marker)) num_inputs = read_varint(s) inputs = [] for _ in range(num_inputs): inputs.append(TxIn.parse(s)) num_outputs = read_varint(s) outputs = [] for _ in range(num_outputs): outputs.append(TxOut.parse(s)) #증인(witness)필드 파싱 for tx_in in inputs: num_items = read_varint(s) items = [] for _ in range(num_items): item_len = read_varint(s) if item_len == 0: items.append(0) else: items.append(s.read(item_len)) tx_in.witness = items locktime = little_endian_to_int(s.read(4)) return cls(version, inputs, outputs, locktime, testnet=testnet, segwit=True) #세그윗 직렬화. 기존의 직렬화는 serialize_legacy로 변경되었다. def serialize_segwit(self): result = int_to_little_endian(self.version, 4) #마커 추가 result += b'\x00\x01' result += encode_varint(len(self.tx_ins)) for tx_in in self.tx_ins: result += tx_in.serialize() result += encode_varint(len(self.tx_outs)) for tx_out in self.tx_outs: result += tx_out.serialize() #증인 필드 직렬화 for tx_in in self.tx_ins: result += int_to_little_endian(len(tx_in.witness), 1) for item in tx_in.witness: if type(item) == int: result += int_to_little_endian(item, 1) else: result += encode_varint(len(item)) + item result += int_to_little_endian(self.locktime, 4) return result def verify_input(self, input_index): tx_in = self.tx_ins[input_index] script_pubkey = tx_in.script_pubkey(testnet=self.testnet) #P2SH 이면 if script_pubkey.is_p2sh_script_pubkey(): cmd = tx_in.script_sig.cmds[-1] raw_redeem = int_to_little_endian(len(cmd), 1) + cmd redeem_script = Script.parse(BytesIO(raw_redeem)) #P2SH-P2WPKH인 경우 if redeem_script.is_p2wpkh_script_pubkey(): z = self.sig_hash_bip143(input_index, redeem_script) witness = tx_in.witness #P2SH-P2WSH인 경우 elif redeem_script.is_p2wsh_script_pubkey(): cmd = tx_in.witness[-1] raw_witness = encode_varint(len(cmd)) + cmd witness_script = Script.parse(BytesIO(raw_witness)) z = self.sig_hash_bip143(input_index, witness_script=witness_script) witness = tx_in.witness else: z = self.sig_hash(input_index, redeem_script) witness = None else: #P2WPKH if script_pubkey.is_p2wpkh_script_pubkey(): z = self.sig_hash_bip143(input_index) witness = tx_in.witness #P2WSH elif script_pubkey.is_p2wsh_script_pubkey(): cmd = tx_in.witness[-1] raw_witness = encode_varint(len(cmd)) + cmd witness_script = Script.parse(BytesIO(raw_witness)) z = self.sig_hash_bip143(input_index, witness_script=witness_script) witness = tx_in.witness else: z = self.sig_hash(input_index) witness = None combined = tx_in.script_sig + script_pubkey return combined.evaluate(z, witness)

class Script: #...생략 #P2WPKH def p2wpkh_script(h160): #증인 필드 발동 조건 return Script([0x00, h160]) #P2WSH def p2wsh_script(h256): #증인 필드 발동 조건 return Script([0x00, h256]) #...생략 #P2PKH 확인 def is_p2pkh_script_pubkey(self): # 5개의 명령어가 있어야 한다. # OP_DUP (0x76), OP_HASH160 (0xa9), 20-byte hash, OP_EQUALVERIFY (0x88), # OP_CHECKSIG (0xac) return len(self.cmds) == 5 and self.cmds[0] == 0x76 \ and self.cmds[1] == 0xa9 \ and type(self.cmds[2]) == bytes and len(self.cmds[2]) == 20 \ and self.cmds[3] == 0x88 and self.cmds[4] == 0xac #P2SH 확인 def is_p2sh_script_pubkey(self): # 3개의 명령어가 있어야 한다. # OP_HASH160 (0xa9), 20-byte hash, OP_EQUAL (0x87) return len(self.cmds) == 3 and self.cmds[0] == 0xa9 \ and type(self.cmds[1]) == bytes and len(self.cmds[1]) == 20 \ and self.cmds[2] == 0x87 #P2WPKH 확인 def is_p2wpkh_script_pubkey(self): return len(self.cmds) == 2 and self.cmds[0] == 0x00 \ and type(self.cmds[1]) == bytes and len(self.cmds[1]) == 20 #P2WSH 확인 def is_p2wsh_script_pubkey(self): return len(self.cmds) == 2 and self.cmds[0] == 0x00 \ and type(self.cmds[1]) == bytes and len(self.cmds[1]) == 32 #...생략 # 스크립트 검증 def evaluate(self, z, witness): cmds = self.cmds[:] stack = [] altstack = [] while len(cmds) > 0: cmd = cmds.pop(0) if type(cmd) == int: operation = OP_CODE_FUNCTIONS[cmd] if cmd in (99, 100): if not operation(stack, cmds): LOGGER.info('bad op: {}'.format(OP_CODE_NAMES[cmd])) return False elif cmd in (107, 108): if not operation(stack, altstack): LOGGER.info('bad op: {}'.format(OP_CODE_NAMES[cmd])) return False elif cmd in (172, 173, 174, 175): if not operation(stack, z): LOGGER.info('bad op: {}'.format(OP_CODE_NAMES[cmd])) return False else: if not operation(stack): LOGGER.info('bad op: {}'.format(OP_CODE_NAMES[cmd])) return False else: stack.append(cmd) if len(cmds) == 3 and cmds[0] == 0xa9 \ and type(cmds[1]) == bytes and len(cmds[1]) == 20 \ and cmds[2] == 0x87: redeem_script = encode_varint(len(cmd)) + cmd cmds.pop() h160 = cmds.pop() cmds.pop() if not op_hash160(stack): return False stack.append(h160) if not op_equal(stack): return False if not op_verify(stack): LOGGER.info('bad p2sh h160') return False redeem_script = encode_varint(len(cmd)) + cmd stream = BytesIO(redeem_script) cmds.extend(Script.parse(stream).cmds) # P2WPKH # 0 <20 byte hash> P2WPKH의 발동 조건 if len(stack) == 2 and stack[0] == b'' and len(stack[1]) == 20: h160 = stack.pop() stack.pop() cmds.extend(witness) cmds.extend(p2pkh_script(h160).cmds) # P2WSH # 0 <32 byte hash> P2WSH의 발동 조건 if len(stack) == 2 and stack[0] == b'' and len(stack[1]) == 32: #해시값 s256 = stack.pop() #세그윗 버전 stack.pop() #증인 스크립트를 제외하고 명령어 집합 확장 cmds.extend(witness[:-1]) #증인 스크립트 witness_script = witness[-1] #증인 스크립트 해시 검사 if s256 != sha256(witness_script): print('bad sha256 {} vs {}'.format (s256.hex(), sha256(witness_script).hex())) return False stream = BytesIO(encode_varint(len(witness_script)) + witness_script) #증인 스크립트 파싱 witness_script_cmds = Script.parse(stream).cmds cmds.extend(witness_script_cmds) if len(stack) == 0: return False if stack.pop() == b'': return False return True