from time import time from base64 import b16encode import os.path import libnacl.utils, libnacl.sign from transaction_const import * from io import BytesIO from hashlib import sha3_256 import logging from random import randint logging.basicConfig(level='DEBUG') log = logging.getLogger('tx') ''' [ CA$H ] [signature ] * 16 [load][used] [keyi][txid] [timestamp ] [ uuid ] [ F0X0 ] 21*4 = 84bytes 76 bytes ''' FIELDS = { 'signature': (0, 63), 'enc_key': 64, 'key_id': 65, 'load_amt': (66, 67), 'used_amt': (68, 69), 'tx_id': (70, 71), 'ts': (72, 75), 'uuid': (76, 79) } class Transaction: def __init__(self, tag_id, nfc_bytes=None): # These are needed for proper randomness self.tag_id = tag_id self.data = bytearray(max([(x[1] if isinstance(x, tuple) else x) for x in FIELDS.values()])+1) self.set_field('key_id', KEY_ID) self.set_field('ts', time()) self.set_field('uuid', libnacl.randombytes(4)) if nfc_bytes: self.load(nfc_bytes) def set_field(self, fid, val): if isinstance(FIELDS[fid], int): flen = 1 sbyte = FIELDS[fid] else: flen = 1 + FIELDS[fid][1] - FIELDS[fid][0] sbyte = FIELDS[fid][0] if isinstance(val, float): val = int(val) if isinstance(val, int): val = val.to_bytes(flen) assert len(val) <= flen if flen > 1: log.info(f"Writing {fid} at bytes {sbyte}:{sbyte+flen-1}: 0x{b16encode(val).decode()}") self.data[sbyte:sbyte+flen] = val else: log.info(f"Writing {fid} at byte {sbyte}: 0x{b16encode(val).decode()}") self.data[sbyte] = val[0] def get_field(self, fid): if isinstance(FIELDS[fid], int): flen = 1 sbyte = FIELDS[fid] else: flen = 1 + FIELDS[fid][1] - FIELDS[fid][0] sbyte = FIELDS[fid][0] if flen > 1: val = bytes(self.data[sbyte:sbyte+flen]) else: val = self.data[sbyte] return val def load(self, data): if data.startswith(HEADER): data = data[len(HEADER):] if data.find(FOOTER): data = data[:-data.find(FOOTER)] self.data = data def sign(self): self.set_field('key_id', KEY_ID) signer = libnacl.sign.Signer(KEY_SEED) signature = signer.signature(bytes(self.data[FIELDS['signature'][1]+1:]+self.tag_id)) self.set_field('signature', signature) log.info("Signed message with success!") self.verify() def get_bytes(self): return HEADER + self.data + FOOTER # TODO: implement these def topup(self, amt): self.load_amount += amt self.tx_bytes def spend(self, amt): self.used_amount += amt def verify(self): verifier = libnacl.sign.Verifier(KEYS[self.get_field('key_id')]) verifier.verify(bytes(self.data + self.tag_id)) log.info("Signature has passed!") if __name__ == '__main__': print("Testing transaction.") tx = Transaction(tag_id=bytes(8)) tx.sign() #for fname in FIELDS: # print(' ', fname, tx.get_field(fname), 'len') #res = tx.get_bytes() #tx.verify() #tx = Transaction(tag_id=bytes(8), nfc_bytes=res) #res2 = tx.get_bytes()