Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
) -> FullBlock:
"""
Creates a block with the specified details. Uses the stored plots to create a proof of space,
and also evaluates the VDF for the proof of time.
"""
prover = None
plot_pk = None
plot_sk = None
qualities: List[bytes] = []
for pn in range(num_plots):
# Allow passing in seed, to create reorgs and different chains
seeded_pn = (pn + 17 * int.from_bytes(seed, "big")) % num_plots
filename = self.filenames[seeded_pn]
plot_pk = plot_pks[seeded_pn]
plot_sk = plot_sks[seeded_pn]
prover = DiskProver(filename)
qualities = prover.get_qualities_for_challenge(challenge_hash)
if len(qualities) > 0:
break
assert prover
assert plot_pk
assert plot_sk
if len(qualities) == 0:
raise NoProofsOfSpaceFound("No proofs for this challenge")
proof_xs: bytes = prover.get_full_proof(challenge_hash, 0)
proof_of_space: ProofOfSpace = ProofOfSpace(
challenge_hash, pool_pk, plot_pk, k, [uint8(b) for b in proof_xs]
)
number_iters: uint64 = pot_iterations.calculate_iterations(
proof_of_space, difficulty, ips, test_constants["MIN_BLOCK_TIME"]
def create_genesis_block(challenge_hash=bytes([0]*32)) -> FullBlock:
plot_seed: bytes32 = ProofOfSpace.calculate_plot_seed(pool_pk, plot_pk)
filename: str = "genesis-plot-" + token_hex(10)
plotter = DiskPlotter()
try:
plotter.create_plot_disk(filename, k, b"genesis", plot_seed)
prover = DiskProver(filename)
qualities = prover.get_qualities_for_challenge(challenge_hash)
if len(qualities) == 0:
os.remove(filename)
raise RuntimeError("No proofs for this challenge")
proof_xs: bytes = prover.get_full_proof(challenge_hash, 0)
proof_of_space: ProofOfSpace = ProofOfSpace(pool_pk, plot_pk, k, list(proof_xs))
except KeyboardInterrupt:
os.remove(filename)
sys.exit(1)
os.remove(filename)
number_iters: uint64 = pot_iterations.calculate_iterations(proof_of_space, challenge_hash,
if not os.path.isfile(key_config_filename):
raise RuntimeError(
"Keys not generated. Run python3.7 ./scripts/regenerate_keys.py."
)
if not os.path.isfile(plot_config_filename):
raise RuntimeError(
"Plots not generated. Run python3.7 ./scripts/create_plots.py."
)
self.config = safe_load(open(config_filename, "r"))["harvester"]
self.key_config = safe_load(open(key_config_filename, "r"))
self.plot_config = safe_load(open(plot_config_filename, "r"))
# From filename to prover
self.provers: Dict[str, DiskProver] = {}
# From quality to (challenge_hash, filename, index)
self.challenge_hashes: Dict[bytes32, Tuple[bytes32, str, uint8]] = {}
The farmer requests a signature on the header hash, for one of the proofs that we found.
We look up the correct plot based on the quality, lookup the proof, and return it.
"""
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],
)
"""
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."
)