# (c) Copyright 2018 by Coinkite Inc. This file is covered by license found in COPYING-CC.
#
# usb.py - USB related things
#
import ckcc, pyb, callgate, sys, ux, ngu, stash, aes256ctr
from uasyncio import sleep_ms, core
from uhashlib import sha256
from public_constants import MAX_MSG_LEN, MAX_TXN_LEN, MAX_BLK_LEN, MAX_UPLOAD_LEN, AFC_SCRIPT
from public_constants import STXN_FLAGS_MASK
from ustruct import pack, unpack_from
from ubinascii import hexlify as b2a_hex
from ckcc import watchpoint, is_simulator
import uselect as select
from utils import problem_file_line, call_later_ms
from version import has_fatram, is_devmode
from exceptions import FramingError, CCBusyError, HSMDenied
from nvstore import settings

# Unofficial, unpermissioned... numbers
COINKITE_VID = 0xd13e
CKCC_PID     = 0xcc10

# Based on the U2F descriptor:
# see <https://fidoalliance.org/specs/fido-u2f-v1.0-ps-20141009/fido-u2f-hid-protocol-ps-20141009.html>
# however, we don't want to be detected as a U2F device, because we 
# don't support that protocol, and we don't want web browsers and such
# trying to speak U2F/HID/USB protocol at us. This descriptor is just to
# keep the HID class drivers happy, we will detect based on VID/PID.
#
hid_descp = bytes([
            0x06, 0xcc, 0x10,  # USAGE_PAGE (CC10 = Coldcard v1.0)
            0x09, 0x01,        # USAGE (0x01)
            0xa1, 0x01,        # COLLECTION (Application)
            0x09, 0x20,        # USAGE (Input Report Data)
            0x15, 0x00,        # LOGICAL_MINIMUM (0)
            0x26, 0xff, 0x00,  # LOGICAL_MAXIMUM (255)
            0x75, 0x08,        # REPORT_SIZE (8)
            0x95, 0x40,        # REPORT_COUNT (64)
            0x81, 0x02,        # INPUT (Data,Var,Abs)
            0x09, 0x21,        # USAGE (Output Report Data)
            0x15, 0x00,        # LOGICAL_MINIMUM (0)
            0x26, 0xff, 0x00,  # LOGICAL_MAXIMUM (255)
            0x75, 0x08,        # REPORT_SIZE (8)
            0x95, 0x40,        # REPORT_COUNT (64)
            0x91, 0x02,        # OUTPUT (Data,Var,Abs)
            0xc0,              # END_COLLECTION
        ])

# Only these whitelisted USB commands are allowed once we enter HSM mode.
# NOTE: 'robo' here would allow firmware changes during HSM mode!
HSM_WHITELIST = frozenset({
    'logo', 'ping', 'vers',     # harmless/boring
    'upld', 'sha2', 'dwld', 'stxn',     # up/download/sign PSBT needed
    'mitm','ncry',              # maybe limited by policy tho
    'smsg',                     # limited by policy
    'blkc', 'hsts',             # report status values
    'stok', 'smok',             # completion check: sign txn or msg
    'xpub', 'msck',             # quick status checks
    'p2sh', 'show',             # limited by HSM policy
    'user',                     # auth HSM user, other user cmds not allowed
    'gslr',                     # read storage locker; hsm mode only, limited usage
})



# singleton instance of USBHandler()
handler = None

def enable_usb():
    # We can't change it on the fly; must be disabled before here
    cur = pyb.usb_mode()
    if cur:
        print("USB already enabled: %s" % cur)
    else:
        # subclass, protocol, max packet length, polling interval, report descriptor
        hid_info = (0x0, 0x0, 64, 5, hid_descp )
        pyb.usb_mode('VCP+HID', vid=COINKITE_VID, pid=CKCC_PID, hid=hid_info)

    global handler
    if not handler:
        handler = USBHandler()
        from imptask import IMPT
        IMPT.start_task('USB', handler.usb_hid_recv())

def disable_usb():
    # pull the plug
    pyb.usb_mode(None)

def is_vcp_active():
    # VCP = Virtual Comm Port
    en = ckcc.vcp_enabled(None)
    cur = pyb.usb_mode()

    return cur and ('VCP' in cur) and en

class USBHandler:
    def __init__(self):
        self.dev = pyb.USB_HID()

        # We keep a running hash over whatever has been uploaded
        # - reset at offset zero, can be read back anytime
        self.file_checksum = sha256()
        self.is_fw_upgrade = False

        # handle simulator
        self.blockable = getattr(self.dev, 'pipe', self.dev)

        #self.msg = bytearray(MAX_MSG_LEN)
        from sram2 import usb_buf
        self.msg = usb_buf
        assert len(self.msg) == MAX_MSG_LEN

        self.encrypted_req = False

        # these will be objects later
        self.encrypt = None
        self.decrypt = None

    def get_packet(self):
        # read next packet (64 bytes) waiting on the wire. Unframe it and return
        # active part of packet, flags associated.
        buf = self.dev.recv(64, timeout=5000)

        if not buf:
            raise FramingError('timeout')
        elif len(buf) < 64:
            raise FramingError('short')
        elif len(buf) > 64:
            raise FramingError('long')

        # first byte gives us the actual size, status
        # all illegal combos here may become special messages someday
        flag = buf[0]
        is_last  = bool(flag & 0x80)
        len_here = int(flag & 0x3f)
        is_encrypted = bool(flag & 0x40)

        return buf[1:1+len_here], is_last, is_encrypted

    async def usb_hid_recv(self):
        # blocks and builds up a full-length command packet in memory
        # - calls self.handle() once complete msg on hand
        msg_len = 0

        while 1:
            yield core._io_queue.queue_read(self.blockable)

            try:
                here, is_last, is_encrypted = self.get_packet()

                #print('Rx[%d]' % len(here))
                if here:
                    lh = len(here)
                    if msg_len+lh > MAX_MSG_LEN:
                        raise FramingError('xlong')

                    self.msg[msg_len:msg_len + lh] = here
                    msg_len += lh
                else:
                    # treat zero-length packets as a reset request
                    # do not echo anything back on link.. used to resync connection
                    msg_len = 0
                    continue

                if not is_last:
                    # need more content
                    continue

                if not(4 <= msg_len <= MAX_MSG_LEN):
                    raise FramingError('badsz')

                if is_encrypted:
                    if self.decrypt is None:
                        raise FramingError('no key')

                    self.encrypted_req = True
                    self.decrypt_inplace(msg_len)
                else:
                    self.encrypted_req = False

                # process request
                try:
                    # this saves memory over a simple slice (confirmed)
                    args = memoryview(self.msg)[4:msg_len]
                    resp = await self.handle(self.msg[0:4], args)
                    msg_len = 0
                except CCBusyError:
                    # auth UX is doing something else
                    resp = b'busy'
                    msg_len = 0
                except HSMDenied:
                    resp = b'err_Not allowed in HSM mode'
                    msg_len = 0
                except (ValueError, AssertionError) as exc:
                    # some limited invalid args feedback
                    #print("USB request caused assert: ", end='')
                    #sys.print_exception(exc)
                    msg = str(exc)
                    if not msg:
                        msg = 'Assertion ' + problem_file_line(exc)
                    resp = b'err_' + msg.encode()[0:80]
                    msg_len = 0
                except MemoryError:
                    # prefer to catch at higher layers, but sometimes can't
                    resp = b'err_Out of RAM'
                    msg_len = 0
                except Exception as exc:
                    # catch bugs and fuzzing too
                    if is_simulator() or is_devmode:
                        print("USB request caused this: ", end='')
                        sys.print_exception(exc)
                    resp = b'err_Confused ' + problem_file_line(exc)
                    msg_len = 0

                # aways send a reply if they get this far
                await self.send_response(resp)

            except FramingError as exc:
                reason = exc.args[0]
                #print("Framing: %s" % reason)
                self.framing_error(reason)
                msg_len = 0

            except BaseException as exc:
                # recover from general issues/keep going
                #print("USB!")
                #sys.print_exception(exc)
                msg_len = 0

    def decrypt_inplace(self, msg_len):
        # self.msg is encrypted. decode it in place
        # - seems dangerous to use memview here, but works
        # - some memory alloc still happens here tho
        self.msg[0:msg_len] = self.decrypt(memoryview(self.msg)[0:msg_len])

    def encrypt_response(self, msg):
        # encrypt what we'll send to desktop

        return self.encrypt(msg)

    async def send_response(self, resp):
        # send a python object as the response
        # - we know how to encode a few things, or send binary
        # - sadly we cannot stream here because we cannot subclass streams
        # - cannot reuse rx buffer either!

        # handle simple types here

        if isinstance(resp, (bytes, bytearray)):
            # preformated
            assert len(resp) >= 4
        elif resp is None:
            resp = b'okay'
        elif isinstance(resp, int):
            resp = pack('<4sI', 'int1', resp)
        else:
            #print("Unknown resp: " + repr(resp))
            raise NotImplementedError()

        assert len(resp) >= 4

        msg = bytearray(64)

        if self.encrypt and self.encrypted_req:
            resp = self.encrypt_response(resp)
            final_flag = 0x80 | 0x40
        else:
            final_flag = 0x80

        pos = 0
        left = len(resp)
        while left:
            # sent up to 63 bytes per packet
            here = min(left, 63)
            msg[0] = here
            msg[1:1+here] = resp[pos:pos+here]
            if here == left:
                # no more to come
                assert 0 <= here < 64
                msg[0] |= final_flag

            left -= here
            pos += here

            for retries in range(100):
                chk = self.dev.send(msg)
                if chk == 64: break

                # Host may not have read previous value yet, so might need
                # to wait for it. Data loss possible here, but also the
                # host may stop reading the EP forever, so not our fault.
                # Let other stuff run during this delay.
                await sleep_ms(10)

    def framing_error(self, why):
        # send error about framing, and recover
        self.dev.send(b'%cfram%-59s' % (4+len(why), why))


    async def handle(self, cmd, args):
        # Dispatch incoming message, and provide reply.
        from glob import hsm_active

        try:
            cmd = bytes(cmd).decode()
        except:
            raise FramingError('decode')

        if cmd[0].isupper() and (is_simulator() or is_devmode):
            # special hacky commands to support testing w/ the simulator
            try:
                from usb_test_commands import do_usb_command
                return do_usb_command(cmd, args)
            except: 
                pass

        if hsm_active:
            # only a few commands are allowed during HSM mode
            if cmd not in HSM_WHITELIST:
                raise HSMDenied

        if cmd == 'dfu_':
            # only useful in factory, undocumented.
            return self.call_after(callgate.enter_dfu)

        if cmd == 'rebo':
            from auth import UserAuthorizedAction, FirmwareUpgradeRequest
            import machine

            req = UserAuthorizedAction.active_request
            if req and isinstance(req, FirmwareUpgradeRequest):
                # We're waiting on firmware upgrade approval, so don't reboot
                # (which would not apply the upgrade anyway) and also don't
                # give an error, because ckcc-protocol and other clients
                # send the reboot as part of the old upgrade process.
                return
            return self.call_after(machine.reset)

        if cmd == 'logo':
            from utils import clean_shutdown
            return self.call_after(clean_shutdown)

        if cmd == 'ping':
            return b'biny' + args

        if cmd == 'upld':
            offset, total_size = unpack_from('<II', args)
            data = memoryview(args)[4+4:]

            return await self.handle_upload(offset, total_size, data)

        if cmd == 'dwld':
            offset, length, fileno = unpack_from('<III', args)
            return await self.handle_download(offset, length, fileno)

        if cmd == 'ncry':
            version, his_pubkey = unpack_from('<I64s', args)

            return self.handle_crypto_setup(version, his_pubkey)

        if cmd == 'vers':
            from version import get_mpy_version, hw_label
            from callgate import get_bl_version

            # Returning: date, version(human), bootloader version, full date version
            # BUT: be ready for additions!
            rv = list(get_mpy_version())
            rv.insert(2, get_bl_version()[0])
            rv.append(hw_label)

            return b'asci' + ('\n'.join(rv)).encode()

        if cmd == 'sha2':
            return b'biny' + self.file_checksum.digest()

        if cmd == 'xpub':
            assert self.encrypted_req, 'must encrypt'
            return self.handle_xpub(args)

        if cmd == 'mitm':
            assert self.encrypted_req, 'must encrypt'
            return await self.handle_mitm_check()

        if cmd == 'smsg':
            # sign message
            addr_fmt, len_subpath, len_msg = unpack_from('<III', args)
            subpath = args[12:12+len_subpath]
            msg = args[12+len_subpath:]
            assert len(msg) == len_msg, "badlen"

            from auth import sign_msg
            sign_msg(msg, subpath, addr_fmt)
            return None

        if cmd == 'p2sh':
            # show P2SH (probably multisig) address on screen (also provides it back)
            # - must provide redeem script, and list of [xfp+path]
            from auth import start_show_p2sh_address

            if hsm_active and not hsm_active.approve_address_share(is_p2sh=True):
                raise HSMDenied

            # new multsig goodness, needs mapping from xfp->path and M values
            addr_fmt, M, N, script_len = unpack_from('<IBBH', args)

            assert addr_fmt & AFC_SCRIPT
            assert 1 <= M <= N <= 20
            assert 30 <= script_len <= 520

            offset = 8
            witdeem_script = args[offset:offset+script_len]
            offset += script_len

            assert len(witdeem_script) == script_len

            xfp_paths = []
            for i in range(N):
                ln = args[offset]
                assert 1 <= ln <= 16, 'badlen'
                xfp_paths.append(unpack_from('<%dI' % ln, args, offset+1))
                offset += (ln*4) + 1

            assert offset == len(args)

            return b'asci' + start_show_p2sh_address(M, N, addr_fmt, xfp_paths,
                                                        witdeem_script)

        if cmd == 'show':
            # simple cases, older code: text subpath
            from auth import start_show_address

            addr_fmt, = unpack_from('<I', args)
            assert not (addr_fmt & AFC_SCRIPT)

            return b'asci' + start_show_address(addr_fmt, subpath=args[4:])

        if cmd == 'enrl':
            # Enroll new xpubkey to be involved in multisigs.
            # - text config file must already be uploaded

            file_len, file_sha = unpack_from('<I32s', args)
            if file_sha != self.file_checksum.digest():
                return b'err_Checksum'
            assert 100 < file_len <= (20*200), "badlen"

            # Start an UX interaction, return immediately here
            from auth import maybe_enroll_xpub
            maybe_enroll_xpub(sf_len=file_len, ux_reset=True)

            return None

        if cmd == 'msck':
            # Quick check to test if we have a wallet already installed.
            from multisig import MultisigWallet
            M, N, xfp_xor = unpack_from('<3I', args)

            return int(MultisigWallet.quick_check(M, N, xfp_xor))

        if cmd == 'stxn':
            # sign transaction
            txn_len, flags, txn_sha = unpack_from('<II32s', args)
            if txn_sha != self.file_checksum.digest():
                return b'err_Checksum'

            assert 50 < txn_len <= MAX_TXN_LEN, "badlen"

            from auth import sign_transaction
            sign_transaction(txn_len, (flags & STXN_FLAGS_MASK), txn_sha)
            return None

        if cmd == 'stok' or cmd == 'bkok' or cmd == 'smok' or cmd == 'pwok':
            # Have we finished (whatever) the transaction,
            # which needed user approval? If so, provide result.
            from auth import UserAuthorizedAction

            req = UserAuthorizedAction.active_request
            if not req:
                return b'err_No active request'

            if req.refused:
                UserAuthorizedAction.cleanup()
                return b'refu'

            if req.failed:
                rv = b'err_' + req.failed.encode()
                UserAuthorizedAction.cleanup()
                return rv

            if not req.result:
                # STILL waiting on user
                return None


            if cmd == 'pwok':
                # return new root xpub
                xpub = req.result
                UserAuthorizedAction.cleanup()
                return b'asci' + bytes(xpub, 'ascii')
            elif cmd == 'smok':
                # signed message done: just give them the signature
                addr, sig = req.address, req.result
                UserAuthorizedAction.cleanup()
                return pack('<4sI', 'smrx', len(addr)) + addr.encode() + sig
            else:
                # generic file response
                resp_len, sha = req.result
                UserAuthorizedAction.cleanup()
                return pack('<4sI32s', 'strx', resp_len, sha)

        if cmd == 'pass':
            # bip39 passphrase provided, maybe use it if authorized
            assert self.encrypted_req, 'must encrypt'
            from auth import start_bip39_passphrase

            assert settings.get('words', True), 'no seed'
            assert len(args) < 400, 'too long'
            pw = str(args, 'utf8')
            assert len(pw) < 100, 'too long'

            return start_bip39_passphrase(pw)

        if cmd == 'back':
            # start backup: asks user, takes long time.
            from auth import start_remote_backup
            return start_remote_backup()

        if cmd == 'blkc':
            # report which blockchain we are configured for
            from chains import current_chain
            chain = current_chain()
            return b'asci' + chain.ctype

        if cmd == 'bagi':
            return self.handle_bag_number(args)

        if has_fatram:
            # HSM and user-related features only supported on larger-memory Mk3

            if cmd == 'hsms':
                # HSM mode "start" -- requires user approval
                if args:
                    file_len, file_sha = unpack_from('<I32s', args)
                    if file_sha != self.file_checksum.digest():
                        return b'err_Checksum'
                    assert 2 <= file_len <= (200*1000), "badlen"
                else:
                    file_len = 0

                # Start an UX interaction but return (mostly) immediately here
                from hsm_ux import start_hsm_approval
                await start_hsm_approval(sf_len=file_len, usb_mode=True)

                return None

            if cmd == 'hsts':
                # can always query HSM mode
                from hsm import hsm_status_report
                import ujson
                return b'asci' + ujson.dumps(hsm_status_report())

            if cmd == 'gslr':
                # get the value held in the Storage Locker
                assert hsm_active, 'need hsm'
                return b'biny' + hsm_active.fetch_storage_locker()


            # User Mgmt
            if cmd == 'nwur':     # new user
                from users import Users
                auth_mode, ul, sl = unpack_from('<BBB', args)
                username = bytes(args[3:3+ul]).decode('ascii')
                secret = bytes(args[3+ul:3+ul+sl])

                return b'asci' + Users.create(username, auth_mode, secret).encode('ascii')

            if cmd == 'rmur':     # delete user
                from users import Users
                ul, = unpack_from('<B', args)
                username = bytes(args[1:1+ul]).decode('ascii')

                return Users.delete(username)

            if cmd == 'user':       # auth user (HSM mode)
                from users import Users
                totp_time, ul, tl = unpack_from('<IBB', args)
                username = bytes(args[6:6+ul]).decode('ascii')
                token = bytes(args[6+ul:6+ul+tl])

                if hsm_active:
                    # just queues these details, can't be checked until PSBT on-hand
                    hsm_active.usb_auth_user(username, token, totp_time)
                    return None
                else:
                    # dryrun/testing purposes: validate only, doesn't unlock nothing
                    return b'asci' + Users.auth_okay(username, token, totp_time).encode('ascii')

        #print("USB garbage: %s +[%d]" % (cmd, len(args)))

        return b'err_Unknown cmd'


    def call_after(self, func, *args):
        # we want to provide nice response before dying
        call_later_ms(500, func, *args)

    def handle_crypto_setup(self, version, his_pubkey):
        # pick a one-time key pair for myself, and return the pubkey for that
        # determine what the session key will be for this connection
        assert version == 0x1
        assert len(his_pubkey) == 64

        # pick a random key pair, just for this session
        pair = ngu.secp256k1.keypair()
        my_pubkey = pair.pubkey().to_bytes(True)        # un-compressed

        #print('my pubkey = ' + str(b2a_hex(my_pubkey)))
        #print('his pubkey = ' + str(b2a_hex(his_pubkey)))

        self.session_key = pair.ecdh_multiply(b'\x04' + his_pubkey)
        del pair

        #print("session = " + str(b2a_hex(self.session_key)))

        # Would be nice to have nonce in addition to the counter, but
        # harder on the desktop side.
        ctr = aes256ctr.new(self.session_key)
        self.encrypt = ctr.cipher
        self.decrypt = ctr.copy().cipher

        xfp = settings.get('xfp', 0)
        xpub = settings.get('xpub', '')

        #assert my_pubkey[0] == 0x04
        return b'mypb' + my_pubkey[1:] + pack('<II', xfp, len(xpub)) +  xpub

    async def handle_mitm_check(self):
        # Sign the current session key using our master (bitcoin) key.
        # - proves our identity and that no-one is between 

        # Rate limit and fuzz timing in case we have timing sensitivity
        await sleep_ms(100 + ngu.random.uniform(100))

        with stash.SensitiveValues() as sv:
            pk = sv.node.privkey()
            sv.register(pk)

            signature = ngu.secp256k1.sign(pk, self.session_key, 0).to_bytes()

            assert len(signature) == 65

        return b'biny' + signature


    async def handle_download(self, offset, length, file_number):
        # let them read from where we store the signed txn
        # - filenumber can be 0 or 1: uploaded txn, or result
        from sflash import SF

        # limiting memory use here, should be MAX_BLK_LEN really
        length = min(length, MAX_BLK_LEN)

        assert 0 <= file_number < 2, 'bad fnum'
        assert 0 <= offset <= MAX_TXN_LEN, "bad offset"
        assert 1 <= length, 'len'

        # maintain a running SHA256 over what's sent
        if offset == 0:
            self.file_checksum = sha256()

        pos = (MAX_TXN_LEN * file_number) + offset

        resp = bytearray(4 + length)
        resp[0:4] = b'biny'
        SF.read(pos, memoryview(resp)[4:])

        self.file_checksum.update(memoryview(resp)[4:])

        return resp

    async def handle_upload(self, offset, total_size, data):
        from sflash import SF
        from glob import dis, hsm_active
        from utils import check_firmware_hdr
        from sigheader import FW_HEADER_OFFSET, FW_HEADER_SIZE, FW_HEADER_MAGIC

        # maintain a running SHA256 over what's received
        if offset == 0:
            self.file_checksum = sha256()
            self.is_fw_upgrade = False

        assert offset % 256 == 0, 'alignment'
        assert offset+len(data) <= total_size <= MAX_UPLOAD_LEN, 'long'

        if hsm_active:
            # additional restrictions in HSM mode
            assert offset+len(data) <= total_size <= MAX_TXN_LEN, 'psbt'
            if offset == 0:
                assert data[0:5] == b'psbt\xff', 'psbt'

        for pos in range(offset, offset+len(data), 256):
            if pos % 4096 == 0:
                # erase here
                dis.fullscreen("Receiving...", offset/total_size)

                SF.sector_erase(pos)

                # expect 10-22 ms delay here
                await sleep_ms(12)
                while SF.is_busy():
                    await sleep_ms(2)

            # write up to 256 bytes
            here = data[pos-offset:pos-offset+256]
            self.file_checksum.update(here)

            # Very special case for firmware upgrades: intercept and modify
            # header contents on the fly, and also fail faster if wouldn't work
            # on this specific hardware.
            # - workaround: ckcc-protocol upgrade process understates the file
            #   length and appends hdr, but that's kinda a bug, so support both
            is_trailer = (pos == (total_size - FW_HEADER_SIZE) or pos == total_size)

            if pos == (FW_HEADER_OFFSET & ~255):
                hdr = memoryview(here)[-128:]
                magic, = unpack_from('<I', hdr[0:4])
                if magic == FW_HEADER_MAGIC:
                    self.is_fw_upgrade = bytes(hdr)

                    prob = check_firmware_hdr(hdr, total_size)
                    if prob:
                        raise ValueError(prob)

            if is_trailer and self.is_fw_upgrade:
                # expect the trailer to exactly match the original one
                assert len(here) == 128      # == FW_HEADER_SIZE
                hdr = memoryview(here)[-128:]
                assert hdr == self.is_fw_upgrade        # indicates hacking

                # but don't write it, instead offer user a chance to abort
                from auth import authorize_upgrade
                authorize_upgrade(self.is_fw_upgrade, pos)

                # pretend we wrote it, so ckcc-protocol or whatever gives normal feedback
                return offset

            SF.write(pos, here)

            # full page write: 0.6 to 3ms
            while SF.is_busy():
                await sleep_ms(1)


        if offset+len(data) >= total_size and not hsm_active:
            # probably done
            dis.progress_bar_show(1.0)
            ux.restore_menu()

        return offset

    def handle_xpub(self, subpath):
        # Share the xpub for the indicated subpath. Expects
        # a text string which is the path derivation.

        # TODO: might not have a privkey yet

        from chains import current_chain
        from utils import cleanup_deriv_path

        subpath = cleanup_deriv_path(subpath)

        from glob import hsm_active
        if hsm_active and not hsm_active.approve_xpub_share(subpath):
            raise HSMDenied

        chain = current_chain()

        with stash.SensitiveValues() as sv:
            node = sv.derive_path(subpath)

            xpub = chain.serialize_public(node)

            return b'asci' + xpub.encode()

    def handle_bag_number(self, bag_num):
        import version, callgate
        from glob import dis
        from pincodes import pa

        if version.is_factory_mode and bag_num:
            # check state first
            assert settings.get('tested', False)
            assert pa.is_blank()
            assert 8 <= len(bag_num) < 32

            # do the change
            failed = callgate.set_bag_number(bag_num)
            assert not failed

            callgate.set_rdp_level(2 if not is_devmode else 0)
            pa.greenlight_firmware()
            dis.fullscreen(bytes(bag_num).decode())

            self.call_after(callgate.show_logout, 1)

        # always report the existing/new value
        val = callgate.get_bag_number() or b''

        return b'asci' + val

# EOF