This is a follow-up to this question after following some of the advice kindly given by other users.
The full repo is here, but I'll post the key excerpts below.
background
I'm trying to get a grip on blockchain and public key cryptography, and thought it would be fun to create a government regulation drafting and procurement program – the government in question is one I just invented.
The system should work as follows:
- A government official who has the required passwords creates a decree, stamps it with a digital stamp, and then adds it to the registry.
-
A private person then meets decrees at work:
- If he comes across a decree in isolation, he can judge its authenticity by means of its stamp and the public key.
- If he has access to the register, he can check that the government has not tried to hide an earlier decree by checking the chain of hashes.
The code
This is the code in which I make and review mine Stamps, the certificates that authenticate decrees:
"""
This code defines two classes: one of which produces a digital stamp for
documents issued by the Chancellor of Cyprus, and the other of which verfies
the same.
"""
# Standard imports.
import getpass
import os
from pathlib import Path
# Non-standard imports.
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa
# Local constants.
PATH_TO_DIGISTAMP = str(Path.home())+"/chancery-b/digistamp/"
PATH_TO_PRIVATE_KEY = PATH_TO_DIGISTAMP+"stamp_private_key.pem"
PATH_TO_PUBLIC_KEY = PATH_TO_DIGISTAMP+"stamp_public_key.pem"
PUBLIC_EXPONENT = 65537
KEY_SIZE = 2048
ENCODING = "utf-8"
################
# MAIN CLASSES #
################
class StampMachine:
""" A class which produces a string which testifies as to the
authenticity of a given document. """
def __init__(self, data):
if not os.path.exists(PATH_TO_PRIVATE_KEY):
raise Exception("No private key on disk.")
self.private_key = load_private_key()
self.data = data
def make_stamp(self):
""" Ronseal. """
data_bytes = bytes(self.data, ENCODING)
result_bytes = self.private_key.sign(
data_bytes,
padding.PSS(mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH),
hashes.SHA256())
result = result_bytes.hex()
return result
class Verifier:
""" A class which allows the user to verify a stamp produced as
above. """
def __init__(self, stamp, data):
self.stamp_bytes = bytes.fromhex(stamp)
self.public_key = load_public_key()
self.data = data
def verify(self):
""" Decide whether the stamp in question is authentic or not. """
data_bytes = bytes(self.data, ENCODING)
try:
self.public_key.verify(
self.stamp_bytes,
data_bytes,
padding.PSS(mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH),
hashes.SHA256())
except InvalidSignature:
return False
else:
return True
####################
# HELPER FUNCTIONS #
####################
def get_bytes_password():
""" Get a password from the user, and convert it into bytes. """
password = getpass.getpass(prompt="Digistamp password: ")
result = bytes(password, ENCODING)
return result
def get_bytes_password_new():
""" Get a NEW password from the user, and convert it into bytes. """
password = getpass.getpass(prompt="Digistamp password: ")
password_ = getpass.getpass(prompt="Confirm password: ")
if password != password_:
print("Passwords do not match.")
return get_bytes_password_new()
result = bytes(password, ENCODING)
return result
def generate_public_key(private_key):
""" Generate a public key from a private key object. """
public_key = private_key.public_key()
pem = public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo)
with open(PATH_TO_PUBLIC_KEY, "wb") as public_key_file:
public_key_file.write(pem)
def generate_keys():
""" Generate a new private and public key. """
if os.path.exists(PATH_TO_PRIVATE_KEY):
raise Exception("Private key file already exists.")
bpw = get_bytes_password_new()
private_key = rsa.generate_private_key(public_exponent=PUBLIC_EXPONENT,
key_size=KEY_SIZE,
backend=default_backend())
pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.BestAvailableEncryption(bpw))
with open(PATH_TO_PRIVATE_KEY, "wb") as private_key_file:
private_key_file.write(pem)
generate_public_key(private_key)
def load_private_key():
""" Load the private key from its file. """
bpw = get_bytes_password()
with open(PATH_TO_PRIVATE_KEY, "rb") as key_file:
result = serialization.load_pem_private_key(
key_file.read(),
password=bpw,
backend=default_backend())
return result
def load_public_key():
""" Load the public key from its file. """
with open(PATH_TO_PUBLIC_KEY, "rb") as key_file:
result = serialization.load_pem_public_key(
key_file.read(),
backend=default_backend())
return result
###########
# TESTING #
###########
def test():
""" Run the unit tests. """
stamp = StampMachine("123").make_stamp()
assert Verifier(stamp, "123").verify()
assert not Verifier(stamp, "abc").verify()
print("Tests passed!")
###################
# RUN AND WRAP UP #
###################
def run():
test()
if __name__ == "__main__":
run()
This is the code that contains a decree uploaded to register:
"""
This code defines a class, which uploads a record to the ledger.
"""
# Standard imports.
import datetime
import hashlib
import os
import sqlite3
# Local imports.
import ordinance_inputs
from digistamp.digistamp import StampMachine
# Local constants.
ENCODING = "utf-8"
##############
# MAIN CLASS #
##############
class Uploader:
""" The class question. """
def __init__(self):
self.connection = None
self.cursor = None
self.block = BlockOfLedger()
def make_connection(self):
""" Ronseal. """
self.connection = sqlite3.connect("ledger.db")
self.connection.row_factory = dict_factory
self.cursor = self.connection.cursor()
def close_connection(self):
""" Ronseal. """
self.connection.close()
def add_ordinal_and_prev(self):
""" Add the ordinal and the previous block's hash to the block. """
self.make_connection()
query = "SELECT * FROM Block ORDER BY ordinal DESC;"
self.cursor.execute(query)
result = self.cursor.fetchone()
self.close_connection()
if result is None:
self.block.set_ordinal(1)
self.block.set_prev("genesis")
else:
self.block.set_ordinal(result("ordinal")+1)
self.block.set_prev(result("hash"))
def add_hash(self):
""" Add the hash to the present block. """
hash_maker = hashlib.sha256()
hash_maker.update(bytes(self.block.ordinal))
hash_maker.update(bytes(self.block.ordinance_type, ENCODING))
hash_maker.update(bytes(self.block.latex, ENCODING))
hash_maker.update(bytes(self.block.year))
hash_maker.update(bytes(self.block.month))
hash_maker.update(bytes(self.block.day))
if self.block.annexe:
hash_maker.update(self.block.annexe)
hash_maker.update(bytes(self.block.prev, ENCODING))
self.block.set_the_hash(hash_maker.hexdigest())
def add_new_block(self):
""" Add a new block to the legder. """
new_block_tuple = (self.block.ordinal, self.block.ordinance_type,
self.block.latex, self.block.year,
self.block.month, self.block.day,
self.block.stamp, self.block.annexe,
self.block.prev, self.block.the_hash)
query = ("INSERT INTO Block (ordinal, ordinanceType, latex, year, "+
" month, day, stamp, annexe, prev, "+
" hash) "+
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);")
self.make_connection()
self.cursor.execute(query, new_block_tuple)
self.connection.commit()
self.close_connection()
def upload(self):
""" Construct a new block and add it to the chain. """
self.add_ordinal_and_prev()
self.add_hash()
self.add_new_block()
################################
# HELPER CLASSES AND FUNCTIONS #
################################
class BlockOfLedger:
""" A class to hold the properties of a block of the ledger. """
def __init__(self):
date_and_time = datetime.datetime.now()
self.ordinal = None
self.ordinance_type = ordinance_inputs.ordinance_type
self.latex = ordinance_inputs.latex
self.year = date_and_time.year
self.month = date_and_time.month
self.day = date_and_time.day
self.annexe = annexe_to_bytes()
self.prev = None
self.the_hash = None
self.stamp = None
def set_ordinal(self, ordinal):
""" Assign a value to the "ordinal" field of this object. """
self.ordinal = ordinal
def set_prev(self, prev):
""" Assign a value to the "prev" field of this object. """
self.prev = prev
def set_the_hash(self, the_hash):
""" Assign a value to the "the_hash" field of this object. """
self.the_hash = the_hash
self.stamp = StampMachine(self.the_hash).make_stamp()
def dict_factory(cursor, row):
""" A function which allows queries to return dictionaries, rather than
default tuples. """
result = {}
for idx, col in enumerate(cursor.description):
result(col(0)) = row(idx)
return result
def annexe_to_bytes():
""" Convert the annexe folder to a zip, load the bytes thereof into
memory, and then delete the zip. """
if len(os.listdir("annexe/")) == 0:
return None
os.system("zip -r annexe.zip annexe/")
with open("annexe.zip", "rb") as annexe_zip:
result = annexe_zip.read()
os.system("rm annexe.zip")
return result
###################
# RUN AND WRAP UP #
###################
def run():
Uploader().upload()
if __name__ == "__main__":
run()
And that's the code in which decrees are extracted off the register and into something more legible, d. H. a PDF:
"""
This code defines a class which takes a given record in the ledger and
converts it into a directory.
"""
# Standard imports.
import os
import sqlite3
import sys
# Non-standard imports.
from pdfrw import PdfReader, PdfWriter
# Local imports.
from digistamp.digistamp import Verifier
from uploader import dict_factory
# Local constants.
MONTH_NAMES = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug",
"Sep", "Oct", "Nov", "Dec")
VERIFICATION_INSTRUCTIONS = ("To verify this Ordinance: (1) Verify that "+
"the hash matches the data. It should be the "+
"hex digest of the SHA256 hash of the data "+
"points. (2) Verify that the stamp matches "+
"the hash, using the public key, standard "+
"padding and, again, SHA256. (To make your "+
"life easier, you could just use the "+
"verification software provided by this "+
"office.)")
##############
# MAIN CLASS #
##############
class Extractor:
""" The class in question. """
def __init__(self, ordinal):
self.ordinal = ordinal
self.block = self.fetch_block()
self.main_tex = self.make_main_tex()
def fetch_block(self):
""" Fetch the block matching this object's ordinal from the
ledger. """
connection = sqlite3.connect("ledger.db")
connection.row_factory = dict_factory
cursor = connection.cursor()
query = "SELECT * FROM Block WHERE ordinal = ?;"
cursor.execute(query, (self.ordinal,))
result = cursor.fetchone()
connection.close()
if result is None:
raise Exception("No block with ordinal "+str(self.ordinal)+
" in the ledger.")
return result
def get_base(self):
""" Get the base for main.tex, given the type of the Ordinance. """
if self.block("ordinanceType") == "Declaration":
path_to_base = "latexery/base_declaration.tex"
elif self.block("ordinanceType") == "Order":
path_to_base = "latexery/base_order.tex"
else:
raise Exception("Invalid ordinanceType: "+
self.block("ordinanceType"))
with open(path_to_base, "r") as base_file:
result = base_file.read()
return result
def make_main_tex(self):
""" Make the code for main.tex, which will then be used build our
PDF. """
day_str = str(self.block("day"))
if len(day_str) == 1:
day_str = "0"+day_str
packed_ordinal = str(self.ordinal)
while len(packed_ordinal) < 3:
packed_ordinal = "0"+packed_ordinal
month_str = MONTH_NAMES(self.block("month")-1)
result = self.get_base()
result = result.replace("#BODY", self.block("latex"))
result = result.replace("#DAY_STR", day_str)
result = result.replace("#MONTH_STR", month_str)
result = result.replace("#YEAR", str(self.block("year")))
result = result.replace("#PACKED_ORDINAL", packed_ordinal)
return result
def authenticate(self):
""" Check that the block isn't a forgery. """
self.compare_hashes()
self.verify_stamp()
def compare_hashes(self):
""" Compare the "prev" field of this block with the hash of the
previous. """
if self.ordinal == 1:
if self.block("prev") != "genesis":
raise Exception("Block with ordinal=1 should be the "+
"genesis block.")
return
prev_ordinal = self.ordinal-1
connection = sqlite3.connect("ledger.db")
cursor = connection.cursor()
query = "SELECT hash FROM Block WHERE ordinal = ?;"
cursor.execute(query, (prev_ordinal,))
extract = cursor.fetchone()
connection.close()
prev_hash = extract("0")
if prev_hash != self.block("prev"):
raise Exception("Block with ordinal="+str(self.ordinal)+" is "+
"not authentic: "prev" does not match "+
"previous "hash".")
def verify_stamp(self):
""" Check that this block's stamp is in order. """
verifier = Verifier(self.block("stamp"), self.block("hash"))
if not verifier.verify():
raise Exception("Block with ordinal="+str(self.ordinal)+" is "+
"not authentic: "prev" does not match "+
"previous "hash".")
def write_main_tex(self):
""" Ronseal. """
with open("latexery/main.tex", "w") as main_tex:
main_tex.write(self.main_tex)
def compile_main_tex(self):
""" Compile the PDF. """
script = ("cd latexery/n"+
"pdflatex main.tex")
os.system(script)
def add_metadata(self):
""" Add the verification metadata to the PDF. """
os.system("mv latexery/main.pdf latexery/main_old.pdf")
trailer = PdfReader("latexery/main_old.pdf")
trailer.Info.instructions = VERIFICATION_INSTRUCTIONS
trailer.Info.data_ordinal = self.block("ordinal")
trailer.Info.data_ordinanceType = self.block("ordinanceType")
trailer.Info.data_latex = self.block("latex")
trailer.Info.data_year = self.block("year")
trailer.Info.data_month = self.block("month")
trailer.Info.data_day = self.block("day")
decoded_annexe = self.block("annexe").hex()
trailer.Info.data_annexe = decoded_annexe
trailer.Info.data_prev = self.block("prev")
trailer.Info.hash = self.block("hash")
trailer.Info.stamp = self.block("stamp")
PdfWriter("latexery/main.pdf", trailer=trailer).write()
def create_and_copy(self):
""" Create the directory, and copy the PDF into it. """
if os.path.isdir("extracts/"+str(self.ordinal)+"/"):
os.system("rm -r extracts/"+str(self.ordinal)+"/")
os.system("mkdir extracts/"+str(self.ordinal)+"/")
os.system("cp latexery/main.pdf extracts/"+str(self.ordinal)+"/")
def write_annexe_zip(self):
""" Write annexe to a file in the directory. """
if self.block("annexe") is None:
return
with open("extracts/"+str(self.ordinal)+"/annexe.zip", "wb")
as annexe_zip:
annexe_zip.write(self.block("annexe"))
def extract(self):
""" Do the thing. """
self.authenticate()
self.write_main_tex()
self.compile_main_tex()
self.add_metadata()
self.create_and_copy()
self.write_annexe_zip()
def zip_and_delete(self):
""" Ronseal. """
script = ("cd extracts/n"+
"zip -r ordinance_"+str(self.ordinal)+".zip "+
str(self.ordinal)+"/n"+
"rm -r "+str(self.ordinal)+"/")
if os.path.exists("extracts/ordinance_"+str(self.ordinal)+".zip"):
os.system("rm extracts/ordinance_"+str(self.ordinal)+".zip")
os.system(script)
###########
# TESTING #
###########
def demo():
""" Run a demonstration. """
extractor = Extractor(1)
extractor.extract()
###################
# RUN AND WRAP UP #
###################
def run():
if len(sys.argv) == 2:
extractor = Extractor(int(sys.argv(1)))
extractor.extract()
else:
print("Please run me with exactly one argument, the number of the "+
"Ordinance you wish to extract.")
if __name__ == "__main__":
run()
What I would like to know
- Do I use blockchain and public key cryptography wisely? Or have I misunderstood some of the basic concepts?
- Are there any obvious security issues? Is there an obvious exploit where someone could start making convincing fakes?
I feel like I've already raised the more style-related issue in my previous question, but if you still see blatant issues in this regard, you don't have to keep it to yourself.
