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 harvester_handshake(
self, harvester_handshake: harvester_protocol.HarvesterHandshake
):
"""
Handshake between the harvester and farmer. The harvester receives the pool public keys,
which must be put into the plots, before the plotting process begins. We cannot
use any plots which don't have one of the pool keys.
"""
for partial_filename, plot_config in self.plot_config["plots"].items():
if "plot_root" in self.config:
filename = os.path.join(self.config["plot_root"], partial_filename)
else:
filename = os.path.join(ROOT_DIR, "plots", partial_filename)
pool_pubkey = PublicKey.from_bytes(bytes.fromhex(plot_config["pool_pk"]))
# Only use plots that correct pools associated with them
if pool_pubkey in harvester_handshake.pool_pubkeys:
if os.path.isfile(filename):
self.provers[partial_filename] = DiskProver(filename)
else:
log.warn(f"Plot at {filename} does not exist.")
else:
log.warning(
f"Plot {filename} has a pool key that is not in the farmer's pool_pk list."
)
def _pubkey_from_bytes(pubkey: BLSPubkey) -> PublicKey:
try:
return PublicKey.from_bytes(pubkey)
except (RuntimeError, ValueError) as error:
raise ValidationError(f"Bad public key: {pubkey}, {error}")
"-p", "--pool_pub_key", help="Hex public key of pool", type=str, default=""
)
# We need the keys file, to access pool keys (if the exist), and the sk_seed.
args = parser.parse_args()
if not os.path.isfile(key_config_filename):
raise RuntimeError("Keys not generated. Run ./scripts/regenerate_keys.py.")
# The seed is what will be used to generate a private key for each plot
key_config = safe_load(open(key_config_filename, "r"))
sk_seed: bytes = bytes.fromhex(key_config["sk_seed"])
pool_pk: PublicKey
if len(args.pool_pub_key) > 0:
# Use the provided pool public key, useful for using an external pool
pool_pk = PublicKey.from_bytes(bytes.fromhex(args.pool_pub_key))
else:
# Use the pool public key from the config, useful for solo farming
pool_sk = PrivateKey.from_bytes(bytes.fromhex(key_config["pool_sks"][0]))
pool_pk = pool_sk.get_public_key()
print(
f"Creating {args.num_plots} plots of size {args.size}, sk_seed {sk_seed.hex()} ppk {pool_pk}"
)
for i in range(args.num_plots):
# Generate a sk based on the seed, plot size (k), and index
sk: PrivateKey = PrivateKey.from_seed(
sk_seed + args.size.to_bytes(1, "big") + i.to_bytes(4, "big")
)
# The plot seed is based on the pool and plot pks
response: Optional[harvester_protocol.RespondProofOfSpace] = None
try:
# Using the quality find the right plot and index from our solutions
challenge_hash, filename, index = self.challenge_hashes[request.quality]
except KeyError:
log.warning(f"Quality {request.quality} not found")
return
if index is not None:
proof_xs: bytes
try:
proof_xs = self.provers[filename].get_full_proof(challenge_hash, index)
except RuntimeError:
self.provers[filename] = DiskProver(filename)
proof_xs = self.provers[filename].get_full_proof(challenge_hash, index)
pool_pubkey = PublicKey.from_bytes(
bytes.fromhex(self.plot_config["plots"][filename]["pool_pk"])
)
plot_pubkey = PrivateKey.from_bytes(
bytes.fromhex(self.plot_config["plots"][filename]["sk"])
).get_public_key()
proof_of_space: ProofOfSpace = ProofOfSpace(
challenge_hash,
pool_pubkey,
plot_pubkey,
uint8(self.provers[filename].get_size()),
[uint8(b) for b in proof_xs],
)
response = harvester_protocol.RespondProofOfSpace(
request.quality, proof_of_space
)