124 lines
2.9 KiB
Python
124 lines
2.9 KiB
Python
# standard imports
|
|
import os
|
|
import json
|
|
import logging
|
|
import tempfile
|
|
import cbor2
|
|
|
|
# external imports
|
|
from hexathon import strip_0x
|
|
|
|
# local imports
|
|
from .base import *
|
|
|
|
logg = logging.getLogger(__name__)
|
|
|
|
|
|
class Proof(Data):
|
|
|
|
def __init__(self, path='.', attachments=None, writer=None):
|
|
super(Proof, self).__init__()
|
|
self.namespace = 'ge'
|
|
self.description = None
|
|
self.path = path
|
|
self.writer = writer
|
|
self.extra_attachments = attachments
|
|
self.attachments = {}
|
|
self.proof_path = os.path.join(self.path, 'proof.json')
|
|
self.temp_proof_path = tempfile.mkstemp()[1]
|
|
|
|
|
|
def load(self):
|
|
super(Proof, self).load()
|
|
|
|
f = open(self.proof_path, 'r')
|
|
o = json.load(f)
|
|
f.close()
|
|
|
|
self.set_version(o['version'])
|
|
self.description = o['description']
|
|
self.namespace = o['namespace']
|
|
self.issuer = o['issuer']
|
|
|
|
self.inited = True
|
|
|
|
|
|
def start(self):
|
|
super(Proof, self).start()
|
|
|
|
proof_template_file_path = os.path.join(data_dir, 'proof_template_v{}.json'.format(self.version()))
|
|
|
|
f = open(proof_template_file_path)
|
|
o = json.load(f)
|
|
f.close()
|
|
|
|
f = open(self.proof_path, 'w')
|
|
json.dump(o, f, sort_keys=True, indent="\t")
|
|
f.close()
|
|
|
|
|
|
def asdict(self):
|
|
return {
|
|
'version': self.version(),
|
|
'namespace': self.namespace,
|
|
'description': self.description,
|
|
'issuer': self.issuer,
|
|
}
|
|
|
|
|
|
def __get_ordered_hashes(self):
|
|
ks = list(self.attachments.keys())
|
|
ks.sort()
|
|
|
|
return ks
|
|
|
|
|
|
def get(self):
|
|
v = self.asdict()
|
|
b = cbor2.dumps(v)
|
|
|
|
f = open(self.temp_proof_path, 'wb')
|
|
f.write(b)
|
|
f.close()
|
|
|
|
hsh = self.hash(b).hex()
|
|
self.attachments[hsh] = self.temp_proof_path
|
|
logg.debug('cbor of {} is {} hashes to {}'.format(v, b.hex(), hsh))
|
|
|
|
if self.extra_attachments != None:
|
|
a = self.extra_attachments.asdict()
|
|
for k in a.keys():
|
|
self.attachments[k] = a[k]
|
|
|
|
hshs = self.__get_ordered_hashes()
|
|
|
|
return (hsh, hshs)
|
|
|
|
|
|
def process(self, token_address=None, writer=None):
|
|
if writer == None:
|
|
writer = self.writer
|
|
|
|
(hsh, hshs) = self.get()
|
|
hshs = list(map(strip_0x, hshs))
|
|
hshs_bin = list(map(bytes.fromhex, hshs))
|
|
hshs_cat = b''.join(hshs_bin)
|
|
|
|
f = open(self.temp_proof_path, 'rb')
|
|
v = f.read()
|
|
f.close()
|
|
writer.write(hsh, v)
|
|
|
|
r = self.hash(hshs_cat)
|
|
r_hex = r.hex()
|
|
|
|
logg.debug('generated proof {} for hashes {}'.format(r_hex, hshs))
|
|
|
|
writer.write(r_hex, hshs_cat)
|
|
|
|
return r_hex
|
|
|
|
|
|
def __str__(self):
|
|
return "description = {}\n".format(self.description)
|