59 lines
1.3 KiB
Python
59 lines
1.3 KiB
Python
from kasten import Kasten
|
|
|
|
from time import time
|
|
from math import floor
|
|
from typing import Iterable
|
|
|
|
import binascii
|
|
|
|
MAX_FUTURE_SKEW_SECONDS = 60 * 2
|
|
TOTAL_MAX_SIZE = 6 * (10**6)
|
|
|
|
|
|
class BlockRulesException(Exception):
|
|
pass
|
|
|
|
|
|
class BlockExpired(BlockRulesException):
|
|
pass
|
|
|
|
class BlockFutureSkewedBeyondMax(BlockRulesException):
|
|
pass
|
|
|
|
class BlockExistsError(BlockRulesException):
|
|
pass
|
|
|
|
class BlockTooLarge(BlockRulesException):
|
|
pass
|
|
|
|
|
|
def checksum_exists_in_list(checksum: bytes, arr: 'Iterable'):
|
|
if checksum in arr:
|
|
raise BlockExistsError(
|
|
binascii.hexlify(
|
|
checksum).decode("utf-8") + " Already exists in given set")
|
|
|
|
|
|
def check_block_sanity(raw_bytes):
|
|
kasten: Kasten = Kasten(None, raw_bytes, None, auto_check_generator=False)
|
|
|
|
# Ensure a block is not too large
|
|
if len(kasten.get_packed()) > TOTAL_MAX_SIZE:
|
|
raise BlockTooLarge()
|
|
|
|
# Ensure block is not future-skewed or expired
|
|
timestamp: int = floor(time())
|
|
block_timestamp = kasten.get_timestamp()
|
|
difference = timestamp - block_timestamp
|
|
|
|
if difference < -MAX_FUTURE_SKEW_SECONDS:
|
|
raise BlockFutureSkewedBeyondMax()
|
|
|
|
try:
|
|
if block_timestamp + kasten.get_metadata()['expire'] >= timestamp:
|
|
raise BlockExpired()
|
|
except (IndexError, TypeError, KeyError):
|
|
pass
|
|
|
|
|