119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
#
|
|
# Store and watch all **status** values in system.
|
|
#
|
|
import sys, logging, asyncio
|
|
from pprint import pprint, pformat
|
|
from decimal import Decimal
|
|
from chrono import NOW
|
|
from objstruct import ObjectStruct
|
|
from hashlib import sha256
|
|
from copy import deepcopy
|
|
from utils import WatchableMixin
|
|
|
|
logging.getLogger(__name__).addHandler(logging.NullHandler())
|
|
|
|
class SystemStatus(WatchableMixin, ObjectStruct):
|
|
|
|
def __init__(self):
|
|
# define all values here. keep simple, small!
|
|
# - values must be JSON-able.
|
|
super(SystemStatus, self).__init__()
|
|
|
|
self.connected = False
|
|
self.serial_number = None
|
|
|
|
#self.xfp = None
|
|
|
|
self.hsm = dict(users=[], wallets=[]) # short for "hsm_status"
|
|
self.is_testnet = False
|
|
|
|
# storage locker has been read ok.
|
|
self.sl_loaded = False
|
|
|
|
# user doesn't want Tor regardless of other settings (also disables login process)
|
|
self.force_local_mode = False
|
|
|
|
# we are in setup mode
|
|
self.setup_mode = False
|
|
|
|
# PSBT related
|
|
self._pending_psbt = None # raw binary
|
|
self.psbt_hash = None # hex digits (sha256)
|
|
self.psbt_size = None # size of binary
|
|
self.local_code = None # string of 6 digits
|
|
self.psbt_preview = None # text
|
|
self.busy_signing = False
|
|
|
|
# tor related
|
|
self.tord_good = False # local tord control connection good
|
|
self.onion_addr = None # our present onion addr, if any
|
|
self.tor_enabled = False # config calls for tor (ie. BP['tor_enabled'])
|
|
|
|
# list of structs about creditials given by remote users
|
|
self.pending_auth = []
|
|
|
|
def reset_pending_auth(self):
|
|
# clear and setup pending auth list
|
|
from persist import BP
|
|
|
|
# make a list of users that might need to auth
|
|
ul = self.hsm.get('users')
|
|
if not ul:
|
|
if BP.get('policy'):
|
|
ul = set()
|
|
try:
|
|
for r in BP['policy']['rules']:
|
|
ul.union(r.users)
|
|
except KeyError: pass
|
|
ul = list(sorted(ul))
|
|
|
|
# they might have picked privacy over UX, so provide some "slots"
|
|
# regardless of above.
|
|
if not ul:
|
|
ul = ['' for i in range(5)]
|
|
|
|
# construct an obj for UX purposes, but keep the actual secrets separate
|
|
self.pending_auth = [ObjectStruct(name=n, has_name=bool(n),
|
|
has_guess='', totp=0) for n in ul]
|
|
self._auth_guess = [None]*len(ul)
|
|
|
|
|
|
def clear_psbt(self):
|
|
# wipe knowledge of PSBT
|
|
self._pending_psbt = None
|
|
self.psbt_hash = None
|
|
self.psbt_size = None
|
|
self.local_code = None
|
|
self.psbt_preview = None
|
|
|
|
def import_psbt(self, psbt):
|
|
from ckcc.utils import calc_local_pincode
|
|
from utils import cleanup_psbt
|
|
from binascii import b2a_hex
|
|
|
|
self.clear_psbt()
|
|
|
|
self._pending_psbt = cleanup_psbt(psbt)
|
|
|
|
self.psbt_size = len(self._pending_psbt)
|
|
|
|
hh = sha256(self._pending_psbt).digest()
|
|
self.psbt_hash = b2a_hex(hh).decode('ascii')
|
|
|
|
# local PIN code will be wrong/stale now.
|
|
if self.hsm and self.hsm.get('next_local_code'):
|
|
self.local_code = calc_local_pincode(hh, self.hsm.next_local_code)
|
|
|
|
logging.info("Imported PSBT with hash: " + self.psbt_hash)
|
|
|
|
def as_dict(self):
|
|
# we stream changes to web clients, so provide JSON
|
|
return dict((k, deepcopy(self[k]))
|
|
for k in self.keys() if k[0] != '_' and not callable(self[k]))
|
|
|
|
|
|
# singleton
|
|
STATUS = SystemStatus()
|
|
|
|
# EOF
|