Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def respond_header_signature(
self, response: harvester_protocol.RespondHeaderSignature
):
"""
Receives a signature on a block header hash, which is required for submitting
a block to the blockchain.
"""
header_hash: bytes32 = self.harvester_responses_header_hash[response.quality]
proof_of_space: bytes32 = self.harvester_responses_proofs[response.quality]
plot_pubkey = self.harvester_responses_proofs[response.quality].plot_pubkey
assert response.header_hash_signature.verify(
[Util.hash256(header_hash)], [plot_pubkey]
)
pos_hash: bytes32 = proof_of_space.get_hash()
request = farmer_protocol.HeaderSignature(
pos_hash, header_hash, response.header_hash_signature
)
yield OutboundMessage(
NodeType.FULL_NODE, Message("header_signature", request), Delivery.BROADCAST
)
):
"""
The farmer requests a signature on the header hash, for one of the proofs that we found.
A signature is created on the header hash using the plot private key.
"""
_, filename, _ = self.challenge_hashes[request.quality]
plot_sk = PrivateKey.from_bytes(
bytes.fromhex(self.plot_config["plots"][filename]["sk"])
)
header_hash_signature: PrependSignature = plot_sk.sign_prepend(
request.header_hash
)
assert header_hash_signature.verify(
[Util.hash256(request.header_hash)], [plot_sk.get_public_key()]
)
response: harvester_protocol.RespondHeaderSignature = harvester_protocol.RespondHeaderSignature(
request.quality, header_hash_signature,
)
yield OutboundMessage(
NodeType.FARMER,
Message("respond_header_signature", response),
Delivery.RESPOND,
)
if block.body.coinbase.height != prev_block.body.coinbase.height + 1:
return False
else:
if block.body.coinbase.height != 0:
return False
# 12. Check coinbase amount
if (
calculate_block_reward(block.body.coinbase.height)
!= block.body.coinbase.amount
):
return False
# 13. Check coinbase signature with pool pk
if not block.body.coinbase_signature.verify(
[blspy.Util.hash256(bytes(block.body.coinbase))],
[block.header_block.proof_of_space.pool_pubkey],
):
return False
# TODO: 14a. check transactions
# TODO: 14b. Aggregate transaction results into signature
if block.body.aggregated_signature:
# TODO: 15. check that aggregate signature is valid, based on pubkeys, and messages
pass
# TODO: 16. check fees
# TODO: 17. check cost
return True
async def respond_partial_proof(
self, response: harvester_protocol.RespondPartialProof
):
"""
Receives a signature on the hash of the farmer payment target, which is used in a pool
share, to tell the pool where to pay the farmer.
"""
farmer_target_hash = sha256(
bytes.fromhex(self.key_config["farmer_target"])
).digest()
plot_pubkey = self.harvester_responses_proofs[response.quality].plot_pubkey
assert response.farmer_target_signature.verify(
[Util.hash256(farmer_target_hash)], [plot_pubkey]
)
assert prev_block.header_block.challenge
challenge_hash = prev_block.header_block.challenge.get_hash()
# 8. Check challenge hash of prev is the same as in pos
if challenge_hash != block.header_block.proof_of_space.challenge_hash:
return False
else:
assert block.header_block.proof_of_time
challenge_hash = block.header_block.proof_of_time.challenge_hash
if challenge_hash != block.header_block.proof_of_space.challenge_hash:
return False
# 9. Check harvester signature of header data is valid based on harvester key
if not block.header_block.header.harvester_signature.verify(
[blspy.Util.hash256(block.header_block.header.data.get_hash())],
[block.header_block.proof_of_space.plot_pubkey],
):
return False
# 10. Check proof of space based on challenge
pos_quality = block.header_block.proof_of_space.verify_and_get_quality()
if not pos_quality:
return False
# 11. Check coinbase height = parent coinbase height + 1
if not genesis:
assert prev_block
if block.body.coinbase.height != prev_block.body.coinbase.height + 1:
return False
else:
if block.body.coinbase.height != 0: