coldcard/firmware
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

vdisk support

Browse Source
This commit is contained in:
Peter D. Gray 2021-11-15 10:22:54 -05:00
parent ad032a26da
commit a2928000ef
No known key found for this signature in database
GPG Key ID: F0E6CC6AFC16CF7B
5 changed files with 85 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
shared/auth.py
View File

@ -263,7 +263,7 @@ def sign_txt_file(filename):
# copy message into memory
with CardSlot() as card:
with open(filename, 'rt') as fd:
with card.open(filename, 'rt') as fd:
text = fd.readline().strip()
subpath = fd.readline().strip()
@ -312,7 +312,7 @@ def sign_txt_file(filename):
for path in [orig_path, None]:
try:
with CardSlot() as card:
with CardSlot(readonly=True) as card:
out_full, out_fn = card.pick_filename(target_fname, path)
out_path = path
if out_full: break
@ -327,7 +327,7 @@ def sign_txt_file(filename):
# attempt write-out
try:
with CardSlot() as card:
with open(out_full, 'wt') as fd:
with card.open(out_full, 'wt') as fd:
# save in full RFC style
fd.write(RFC_SIGNATURE_TEMPLATE.format(addr=address, msg=text,
blockchain='BITCOIN', sig=sig))
@ -695,7 +695,7 @@ class ApproveTransaction(UserAuthorizedAction):
left = self.psbt.num_outputs - len(largest) - num_change
if left > 0:
msg.write('.. plus %d more smaller output(s), not shown here, which total: ' % left)
msg.write('.. plus %d smaller output(s), not shown here, which total: ' % left)
# calculate left over value
mtot = self.psbt.total_value_out - sum(v for v,t in largest)
@ -749,7 +749,7 @@ async def sign_psbt_file(filename, force_vdisk=False):
# - can't work in-place on the card because we want to support writing out to different card
# - accepts hex or base64 encoding, but binary prefered
with CardSlot(force_vdisk, readonly=True) as card:
with open(filename, 'rb') as fd:
with card.open(filename, 'rb') as fd:
dis.fullscreen('Reading...')
# see how long it is
@ -833,7 +833,7 @@ async def sign_psbt_file(filename, force_vdisk=False):
# don't write signed PSBT if we'd just delete it anyway
out_fn = None
else:
with output_encoder(open(out_full, 'wb')) as fd:
with output_encoder(card.open(out_full, 'wb')) as fd:
# save as updated PSBT
psbt.serialize(fd)
@ -843,7 +843,7 @@ async def sign_psbt_file(filename, force_vdisk=False):
base+'-final.txn' if not del_after else 'tmp.txn', out_path)
if out2_full:
with HexWriter(open(out2_full, 'w+t')) as fd:
with HexWriter(card.open(out2_full, 'w+t')) as fd:
# save transaction, in hex
txid = psbt.finalize(fd)

17
shared/files.py
View File

@ -7,6 +7,7 @@ from uerrno import ENOENT
async def needs_microsd():
# Standard msg shown if no SD card detected when we need one.
from ux import ux_show_story
return await ux_show_story("Please insert a MicroSD card before attempting this operation.")
def _is_ejected():
@ -210,6 +211,7 @@ class CardSlot:
self.mountpt = None
self.force_vdisk = force_vdisk
self.readonly = readonly
self.wrote_files = set()
def __enter__(self):
# Mk4: maybe use our virtual disk in preference to SD Card
@ -243,10 +245,19 @@ class CardSlot:
if self.mountpt == '/sd':
self._recover()
else:
glob.VD.unmount()
glob.VD.unmount(self.wrote_files)
self.mountpt = None
return False
def open(self, fname, mode='r', **kw):
# open a file for read/write
# - track new files for virtdisk case
if 'w' in mode:
assert not self.readonly
self.wrote_files.add(fname)
return open(fname, mode, **kw)
def _recover(self):
# done using the microSD -- unpower it
@ -297,7 +308,7 @@ class CardSlot:
# - no UI here please
import ure
assert self.mountpt # used out of context mgr
assert self.mountpt # else: we got used out of context mgr
# put it back where we found it
path = path or (self.mountpt + '/')
@ -345,6 +356,8 @@ class CardSlot:
# NOTE: we know the FAT filesystem code is simple, see
# ../external/micropython/extmod/vfs_fat.[ch]
self.wrote_files.discard(full_path)
path, basename = full_path.rsplit('/', 1)
try:

66
shared/nvstore.py
View File

@ -21,7 +21,7 @@
import os, sys, ujson, ustruct, ckcc, gc, ngu, aes256ctr
from uio import BytesIO
from uhashlib import sha256
from random import shuffle
from random import shuffle, randbelow
from utils import call_later_ms
from version import mk_num
from glob import PSRAM
@ -66,12 +66,14 @@ from glob import PSRAM
if mk_num <= 3:
# where in SPI Flash we work (last 128k)
SLOTS = range((1024-128)*1024, 1024*1024, 4096)
NUM_SLOTS = 32
from sffile import SFFile
from sflash import SF
else:
# work in LFS2 filesystem instead, but same terminology
SLOTS = range(0, 64)
NUM_SLOTS = 64
MK4_WORKDIR = '/flash/settings/'
@ -287,6 +289,11 @@ class SettingsObject:
fd.write(aes(chk.digest()))
def _used_slots(self):
# mk4: faster list of slots in use; doesn't open them
files = os.listdir(MK4_WORKDIR)
return [int(fn[0:-4], 16) for fn in files if fn.endswith('.aes')]
def _nonempty_slots(self, dis=None):
# generate slots that are non-empty
taste = bytearray(4)
@ -307,12 +314,10 @@ class SettingsObject:
yield pos, taste
else:
# use directory listing
files = os.listdir(MK4_WORKDIR)
self.num_empty = len(SLOTS) - len(files)
files = self._used_slots()
self.num_empty = NUM_SLOTS - len(files)
for i, fn in enumerate(files):
if not fn.endswith('.aes'): continue
pos = int(fn[0:-4], 16)
for i, pos in enumerate(files):
if dis:
dis.progress_bar_show(i / len(files))
@ -331,11 +336,13 @@ class SettingsObject:
self.my_pos = None
self.is_dirty = 0
self.capacity = 0
nonempty = set()
for pos, taste in self._nonempty_slots(dis):
# check if first 2 bytes makes sense for JSON
aes = self.get_aes(pos)
chk = aes.copy().cipher(b'{"')
nonempty.add(pos)
if chk != taste[0:2]:
# doesn't look like JSON meant for me
@ -377,16 +384,15 @@ class SettingsObject:
# nothing found, use defaults
self.current = self.default_values()
# pick a random home
blks = list(SLOTS)
shuffle(blks)
self.my_pos = blks.pop()
# pick a (new) random home
self.my_pos = self.find_spot(-1)
#print("NV: empty")
if self.num_empty == len(SLOTS):
if self.num_empty == NUM_SLOTS:
# Whole thing is blank. Bad for plausible deniability. Write 3 slots
# with white noise. They will be wasted space until it fills up.
for pos in blks[0:3]:
for _ in range(4):
pos = self.find_spot(-1)
self._deny_slot(pos)
def get(self, kn, default=None):
@ -442,23 +448,33 @@ class SettingsObject:
# - check randomly and pick first blank one (wear leveling, deniability)
# - we will write and then erase old slot
# - if "full", blow away a random one
options = [s for s in SLOTS if s != not_here]
shuffle(options)
if mk_num <= 3:
options = [s for s in SLOTS if s != not_here]
shuffle(options)
buf = bytearray(4)
for pos in options:
if self._slot_is_blank(pos, buf):
# found a blank area
return pos
buf = bytearray(4)
for pos in options:
if self._slot_is_blank(pos, buf):
# found a blank area
return pos
# No where to write! (probably a bug because we have lots of slots)
# ... so pick a random slot and kill what it had
#print("ERROR: nvram full?")
# No-where to write! (probably a bug because we have lots of slots)
# ... so pick a random slot and kill what it had
victim = options[0]
else:
# on mk4, use the filesystem to see what's already taken
avail = set(SLOTS) - set(self._used_slots())
avail.discard(not_here)
victem = options[0]
self._wipe_slot(victem)
if avail:
return avail.pop()
return victem
victim = randbelow(NUM_SLOTS)
#print("ERROR: nvram full")
self._wipe_slot(victim)
return victim
def save(self):
# render as JSON, encrypt and write it.

5
shared/usb.py
View File

@ -14,7 +14,6 @@ import uselect as select
from utils import problem_file_line, call_later_ms
from version import has_fatram, is_devmode, has_psram
from exceptions import FramingError, CCBusyError, HSMDenied
from glob import settings
# Unofficial, unpermissioned... numbers
COINKITE_VID = 0xd13e
@ -513,6 +512,7 @@ class USBHandler:
# bip39 passphrase provided, maybe use it if authorized
assert self.encrypted_req, 'must encrypt'
from auth import start_bip39_passphrase
from glob import settings
assert settings.get('words', True), 'no seed'
assert len(args) < 400, 'too long'
@ -629,6 +629,7 @@ class USBHandler:
self.encrypt = ctr.cipher
self.decrypt = ctr.copy().cipher
from glob import settings
xfp = settings.get('xfp', 0)
xpub = settings.get('xpub', '')
@ -797,7 +798,7 @@ class USBHandler:
def handle_bag_number(self, bag_num):
import version, callgate
from glob import dis
from glob import dis, settings
from pincodes import pa
if version.is_factory_mode and bag_num:

37
shared/vdisk.py
View File

@ -6,14 +6,10 @@
import os, sys, pyb, ckcc, version, glob, uasyncio, utime
from sigheader import FW_MIN_LENGTH
from public_constants import MAX_UPLOAD_LEN
from glob import settings
from usb import enable_usb, disable_usb
from uasyncio import sleep_ms
# block device implemented on half the PSRAM
VBLKDEV = ckcc.PSRAM()
MAX_PSRAM_FILE = const(2<<20) # 2 megs
MAX_PSRAM_FILE = const(2<<20) # 2 megs
MIN_QUIET_TIME = 250 # (ms) delay after host writes disk, before we look at it.
def _host_done_cb(_psram):
@ -22,35 +18,41 @@ def _host_done_cb(_psram):
if glob.VD:
glob.VD.host_done_handler()
# singleton: block device implemented on half of the PSRAM
VBLKDEV = ckcc.PSRAM()
class VirtDisk:
def __init__(self):
# Feature is enabled, altho USB might be off.
print("vdisk: init")
glob.VD = self
self.ignore = set()
self.contents = self.sample()
assert ckcc.PSRAM
VBLKDEV.callback(_host_done_cb)
VBLKDEV.set_inserted(True)
print("vdisk: started")
def shutdown(self):
# we've been disabled, stop
print("vdisk: shutdown")
VBLKDEV.set_inserted(False)
VBLKDEV.callback(None)
glob.VD = None
def unmount(self):
def unmount(self, written_files):
# just unmount; ignore errors
try:
os.umount('/vdisk')
enable_usb()
except:
pass
# ignore the files we write ourselves
for fn in written_files:
if fn.startswith('/vdisk/'):
self.ignore.add(fn[7:])
# allow host to change again
enable_usb()
if glob.VD:
VBLKDEV.set_inserted(True)
@ -75,9 +77,10 @@ class VirtDisk:
return '/vdisk'
except OSError as exc:
# corrupt or unformated?
# XXX incomlpete error handling here; needs work
# XXX incomplete error handling here; needs work
VBLKDEV.set_inserted(True)
sys.print_exception(exc)
return None
def sample(self):
@ -87,7 +90,8 @@ class VirtDisk:
try:
os.mount(VBLKDEV, '/vdisk', readonly=True)
return list(sorted((fn, sz) for (fn,ty,_,sz) in os.ilistdir('/vdisk') if ty == 0x8000))
return list(sorted(('/vdisk/'+fn, sz) for (fn,ty,_,sz) in os.ilistdir('/vdisk')
if ty == 0x8000))
except BaseException as exc:
sys.print_exception(exc)
@ -101,7 +105,7 @@ class VirtDisk:
# I could not resist doing this in C... since we already have the
# data in memory, why mess around with file concepts?
actual = VBLKDEV.copy_file(0, filename)
actual = VBLKDEV.copy_file(0, filename.split('/')[-1])
assert actual == sz
@ -109,9 +113,8 @@ class VirtDisk:
def new_psbt(self, filename, sz):
# New incoming PSBT has been detected, start to sign it.
print("new PSBT: " + filename)
from auth import sign_psbt_file
uasyncio.create_task(sign_psbt_file('/vdisk/'+filename, force_vdisk=True))
uasyncio.create_task(sign_psbt_file(filename, force_vdisk=True))
def new_firmware(self, filename, sz):
# potential new firmware file detected
@ -120,7 +123,7 @@ class VirtDisk:
uasyncio.create_task(psram_upgrade(filename, sz))
def host_done_handler(self):
print('host-wrote')
from glob import settings
if settings.get('vdsk', 0) != 2:
# auto mode not enabled, so ignore changes
@ -130,7 +133,6 @@ class VirtDisk:
if now == self.contents:
# no-op change, common, ignore
# - timestamp changes, hidden files, MacOS BS, etc.
print('no file change')
return
# clear ignored items once they are deleted
@ -141,7 +143,6 @@ class VirtDisk:
# Look for files we want to taste; assume they have
# been fully written-out because we are called after a
# fairly long timeout
print('New files? %r' % now)
for fn, sz in now:
if fn in self.ignore: