Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.filenames: List[str] = [
os.path.join(
"tests",
"plots",
"genesis-plots-"
+ str(k)
+ sha256(int.to_bytes(i, 4, "big")).digest().hex()
+ ".dat",
)
for i in range(num_plots)
]
done_filenames = set()
try:
for pn, filename in enumerate(self.filenames):
if not os.path.exists(filename):
plotter = DiskPlotter()
plotter.create_plot_disk(filename, k, b"genesis", plot_seeds[pn])
done_filenames.add(filename)
except KeyboardInterrupt:
for filename in self.filenames:
if filename not in done_filenames and os.path.exists(filename):
os.remove(filename)
sys.exit(1)
) -> 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,
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)
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."
)
# 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
plot_seed: bytes32 = ProofOfSpace.calculate_plot_seed(
pool_pk, sk.get_public_key()
)
filename: str = f"plot-{i}-{args.size}-{plot_seed}.dat"
full_path: str = os.path.join(plot_root, filename)
if os.path.isfile(full_path):
print(f"Plot {filename} already exists")
else:
# Creates the plot. This will take a long time for larger plots.
plotter: DiskPlotter = DiskPlotter()
plotter.create_plot_disk(full_path, args.size, bytes([]), plot_seed)
# Updates the config if necessary.
if os.path.isfile(plot_config_filename):
plot_config = safe_load(open(plot_config_filename, "r"))
else:
plot_config = {"plots": {}}
plot_config_plots_new = deepcopy(plot_config["plots"])
if filename not in plot_config_plots_new:
plot_config_plots_new[filename] = {
"sk": bytes(sk).hex(),
"pool_pk": bytes(pool_pk).hex(),
}
plot_config["plots"].update(plot_config_plots_new)
# Dumps the new config to disk.
def verify_and_get_quality(self) -> Optional[bytes32]:
v: Verifier = Verifier()
plot_seed: bytes32 = self.get_plot_seed()
quality_str = v.validate_proof(plot_seed, self.size, self.challenge_hash,
bytes(self.proof))
if not quality_str:
return None
return self.quality_str_to_quality(self.challenge_hash, quality_str)
async def request_header_hash(
self, request: farmer_protocol.RequestHeaderHash
) -> OutboundMessageGenerator:
"""
Creates a block body and header, with the proof of space, coinbase, and fee targets provided
by the farmer, and sends the hash of the header data back to the farmer.
"""
plot_seed: bytes32 = request.proof_of_space.get_plot_seed()
# Checks that the proof of space is valid
quality_string: bytes = Verifier().validate_proof(
plot_seed,
request.proof_of_space.size,
request.challenge_hash,
bytes(request.proof_of_space.proof),
)
assert quality_string
async with self.store.lock:
# Retrieves the correct head for the challenge
heads: List[HeaderBlock] = self.blockchain.get_current_tips()
target_head: Optional[HeaderBlock] = None
for head in heads:
assert head.challenge
if head.challenge.get_hash() == request.challenge_hash:
target_head = head
if target_head is None: