shopdb-flask/tests/test_plugins/test_usb_kiosk.py

"""The USB kiosk: an UNAUTHENTICATED write, so the guards are the feature.

The kiosk exists so somebody at the drive cabinet - with no ShopDB login - can
take a drive and put it back, with the action attributed to their badge instead
of to a signed-in operator. Everything here is about the bar that opens the door
only as far as it has to: it can move a device between checked-out and
checked-in and nothing else, it refuses a badge it cannot put a name to, and it
does not exist at all until a site turns it on.

Self-hosted mode. The external cmmc_usb path is the same contract against raw
SQL and needs that database to exercise.
"""

import pytest

from shopdb.core.models import Setting


KIOSK_TOKEN = 'usb-kiosk-device-token-for-tests'
# What a provisioned screen sends. Turning the kiosk on is a decision; this is
# the credential.
KIOSK_HEADERS = {'X-Kiosk-Token': KIOSK_TOKEN}


@pytest.fixture
def kiosk(db):
    """Self-hosted USB, kiosk ON, one employee and one available drive."""
    from plugins.employees.models import DirectoryEmployee
    from plugins.usb.models import USBDevice

    db.session.add(Setting(key='usb_directory_mode', value='selfhosted',
                           valuetype='string', category='site'))
    db.session.add(Setting(key='usb_kiosk_enabled', value='true',
                           valuetype='boolean', category='usb'))
    db.session.add(Setting(key='usb_kiosk_token', value=KIOSK_TOKEN,
                           valuetype='string', category='usb'))
    db.session.add(DirectoryEmployee(sso=270015376, firstname='Dana',
                                     lastname='Reyes'))
    db.session.add(DirectoryEmployee(sso=270099999, firstname='Sam',
                                     lastname='Okafor'))
    db.session.add(USBDevice(serialnumber='USB-1', label='Blue Kingston',
                             ischeckedout=False))
    db.session.commit()
    return db


def _setting(db, key, value):
    row = Setting.query.filter_by(key=key).first()
    if row is None:
        db.session.add(Setting(key=key, value=value, valuetype='boolean',
                               category='usb'))
    else:
        row.value = value
    db.session.commit()


def _out(client, badge='270015376', **extra):
    return client.post('/api/usb/kiosk/checkout',
                       json=dict({'device_id': 'USB-1', 'badge': badge}, **extra), headers=KIOSK_HEADERS)


def _in(client, badge='270015376', **extra):
    return client.post('/api/usb/kiosk/checkin',
                       json=dict({'device_id': 'USB-1', 'badge': badge}, **extra), headers=KIOSK_HEADERS)


# --- the door itself --------------------------------------------------------

def test_the_kiosk_does_not_exist_until_a_site_turns_it_on(client, kiosk):
    """Default off, and 404 rather than 403.

    An unauthenticated write is a decision about the door. A site that has not
    made that decision should not have these routes answering at all, and a
    disabled kiosk should not advertise that there is a door here.
    """
    _setting(kiosk, 'usb_kiosk_enabled', 'false')

    assert client.get('/api/usb/kiosk/device/USB-1', headers=KIOSK_HEADERS).status_code == 404
    assert _out(client).status_code == 404
    assert _in(client).status_code == 404


def test_it_ships_off(app):
    """The declared default is what a fresh site gets."""
    from plugins.usb.plugin import USBPlugin
    defaults = {s['key']: s for s in USBPlugin().get_settings_defaults()}
    assert defaults['usb_kiosk_enabled']['value'] == 'false'


def test_no_login_is_required(client, kiosk):
    """The whole point: no Authorization header anywhere in this file."""
    response = client.get('/api/usb/kiosk/device/USB-1', headers=KIOSK_HEADERS)
    assert response.status_code == 200
    assert response.get_json()['data']['device_id'] == 'USB-1'


def test_the_kiosk_cannot_create_edit_or_delete(client, kiosk):
    """It moves a device between two states. It is not an admin surface."""
    assert client.post('/api/usb/kiosk/device/USB-1', headers=KIOSK_HEADERS).status_code in (404, 405)
    assert client.delete('/api/usb/kiosk/device/USB-1').status_code in (404, 405)
    # And the authenticated write routes are still gated.
    assert client.post('/api/usb/USB-1/checkout',
                       json={'badge': '270015376'}).status_code == 401


# --- badges -----------------------------------------------------------------

def test_a_badge_it_cannot_name_is_refused(client, kiosk):
    """A checkout nobody can trace to a person defeats the tracking.

    The signed-in route creates the user row for an unknown badge; the kiosk
    must not, because there is no operator standing there to vouch for it.
    """
    response = _out(client, badge='999999999')
    assert response.status_code == 422
    assert 'not recognized' in response.get_json()['data']['error']['message']

    from plugins.usb.models import USBDevice
    assert USBDevice.query.filter_by(serialnumber='USB-1').first().ischeckedout is False


def test_a_missing_badge_is_refused(client, kiosk):
    assert client.post('/api/usb/kiosk/checkout',
                       json={'device_id': 'USB-1'}, headers=KIOSK_HEADERS).status_code == 400


# --- taking one -------------------------------------------------------------

def test_taking_a_drive_records_who_took_it(client, kiosk):
    from plugins.usb.models import USBDevice

    response = _out(client)
    assert response.status_code == 200, response.get_json()

    device = USBDevice.query.filter_by(serialnumber='USB-1').first()
    assert device.ischeckedout is True
    assert device.currentuserid == '270015376'
    assert device.currentusername == 'Dana Reyes'


def test_a_drive_already_out_cannot_be_taken_again(client, kiosk):
    _out(client)
    response = _out(client, badge='270099999')
    assert response.status_code == 409
    assert 'Dana Reyes' in response.get_json()['data']['error']['message']


def test_an_unknown_code_is_a_404(client, kiosk):
    response = client.post('/api/usb/kiosk/checkout',
                           json={'device_id': 'NOPE', 'badge': '270015376'}, headers=KIOSK_HEADERS)
    assert response.status_code == 404


# --- putting one back -------------------------------------------------------

def test_the_holder_returns_their_own_drive(client, kiosk):
    from plugins.usb.models import USBDevice
    _out(client)

    assert _in(client).status_code == 200
    device = USBDevice.query.filter_by(serialnumber='USB-1').first()
    assert device.ischeckedout is False
    assert device.currentuserid is None


def test_returning_somebody_elses_drive_asks_first(client, kiosk):
    """Not a silent success: the record is about to say they handed it back."""
    from plugins.usb.models import USBDevice
    _out(client)

    response = _in(client, badge='270099999')
    assert response.status_code == 409
    details = response.get_json()['data']['error']['details']
    assert details['needsconfirm'] is True
    assert details['holdername'] == 'Dana Reyes'
    assert USBDevice.query.filter_by(serialnumber='USB-1').first().ischeckedout is True


def test_confirming_returns_it_for_them(client, kiosk):
    """A drive left on a bench by somebody on leave still gets put back.

    Refusing outright just moves the problem to a person with a login.
    """
    from plugins.usb.models import USBDevice
    _out(client)

    response = _in(client, badge='270099999', forsomeoneelse=True)
    assert response.status_code == 200, response.get_json()
    assert USBDevice.query.filter_by(serialnumber='USB-1').first().ischeckedout is False


def test_a_site_can_refuse_returns_for_others_outright(client, kiosk):
    _setting(kiosk, 'usb_kiosk_allow_return_for_others', 'false')
    _out(client)

    response = _in(client, badge='270099999', forsomeoneelse=True)
    assert response.status_code == 403


def test_a_drive_that_is_not_out_cannot_be_returned(client, kiosk):
    assert _in(client).status_code == 400


# --- what the screen is allowed to know -------------------------------------

def test_the_lookup_names_the_holder_but_not_their_badge(client, kiosk):
    """The confirm step has to show whose it is; it does not need their SSO."""
    _out(client)
    data = client.get('/api/usb/kiosk/device/USB-1', headers=KIOSK_HEADERS).get_json()['data']

    assert data['holdername'] == 'Dana Reyes'
    assert '270015376' not in str(data)


# --- the picker -------------------------------------------------------------
# The drives carry no barcode, so the kiosk cannot ask somebody to present one -
# it has to offer a LIST. That is a real widening of what a logged-out screen
# exposes, so the list is not anonymous: it costs a badge that resolves to a
# real person, the same bar taking a drive already had to clear.

def _session(client, badge='270015376'):
    return client.post('/api/usb/kiosk/session', json={'badge': badge}, headers=KIOSK_HEADERS)


def test_the_device_list_costs_a_valid_badge(client, kiosk):
    assert client.post('/api/usb/kiosk/session', json={}, headers=KIOSK_HEADERS).status_code == 400
    assert _session(client, badge='999999999').status_code == 422


def test_the_list_is_not_reachable_when_the_kiosk_is_off(client, kiosk):
    _setting(kiosk, 'usb_kiosk_enabled', 'false')
    assert _session(client).status_code == 404
    assert client.post('/api/usb/kiosk/others',
                       json={'badge': '270015376'}, headers=KIOSK_HEADERS).status_code == 404


def test_the_session_separates_yours_from_available(client, kiosk):
    from plugins.usb.models import USBDevice
    kiosk.session.add(USBDevice(serialnumber='USB-2', label='Second',
                                ischeckedout=False))
    kiosk.session.commit()
    _out(client)   # 270015376 takes USB-1

    data = _session(client).get_json()['data']
    assert data['name'] == 'Dana Reyes'
    assert [d['device_id'] for d in data['yours']] == ['USB-1']
    assert [d['device_id'] for d in data['available']] == ['USB-2']
    assert data['otherscount'] == 0


def test_somebody_elses_drive_is_a_count_not_a_name(client, kiosk):
    """A passer-by should not read the cabinet's holders off a wall screen."""
    _out(client)   # Dana takes USB-1

    data = _session(client, badge='270099999').get_json()['data']
    assert data['yours'] == []
    assert data['otherscount'] == 1
    assert 'Dana' not in str(data)


def test_names_appear_only_on_the_deliberate_second_step(client, kiosk):
    _out(client)

    rows = client.post('/api/usb/kiosk/others',
                       json={'badge': '270099999'}, headers=KIOSK_HEADERS).get_json()['data']
    assert [r['device_id'] for r in rows] == ['USB-1']
    assert rows[0]['holdername'] == 'Dana Reyes'


def test_the_others_list_is_gone_when_returns_for_others_are_refused(client, kiosk):
    """No point showing a list of drives the kiosk will not let anyone return."""
    _setting(kiosk, 'usb_kiosk_allow_return_for_others', 'false')
    _out(client)

    assert client.post('/api/usb/kiosk/others',
                       json={'badge': '270099999'}, headers=KIOSK_HEADERS).status_code == 404


def test_a_retired_drive_is_not_offered(client, kiosk):
    from plugins.usb.models import USBDevice
    device = USBDevice.query.filter_by(serialnumber='USB-1').first()
    device.isactive = False
    kiosk.session.commit()

    data = _session(client).get_json()['data']
    assert data['available'] == []


# --- the flag is a decision, the token is the credential ---------------------
#
# usb_kiosk_enabled alone used to be the whole gate. With the kiosk on, anyone
# who could reach the app could move a drive between checked-out and checked-in
# and have it recorded against a badge they guessed.

def test_kiosk_is_shut_without_the_device_token(client, kiosk):
    assert client.get('/api/usb/kiosk/device/USB-1').status_code == 404
    assert client.post('/api/usb/kiosk/checkout',
                       json={'device_id': 'USB-1',
                             'badge': '270015376'}).status_code == 404


def test_kiosk_is_shut_with_the_wrong_device_token(client, kiosk):
    wrong = {'X-Kiosk-Token': KIOSK_TOKEN + 'x'}
    assert client.get('/api/usb/kiosk/device/USB-1',
                      headers=wrong).status_code == 404
    assert client.post('/api/usb/kiosk/checkout', headers=wrong,
                       json={'device_id': 'USB-1',
                             'badge': '270015376'}).status_code == 404


def test_the_token_alone_is_not_enough_when_the_kiosk_is_off(client, kiosk):
    """Both halves are required: a provisioned screen at a site that has since
    switched the kiosk off gets nothing."""
    _setting(kiosk, 'usb_kiosk_enabled', 'false')
    assert client.get('/api/usb/kiosk/device/USB-1',
                      headers=KIOSK_HEADERS).status_code == 404


# --- the race, and who returned it ------------------------------------------
#
# The kiosk decided "is this drive available?" in Python and then wrote. Two
# scans of the same drive that interleaved both read "available", so both
# inserted a custody row: one drive recorded out to two people, and the
# documented 409 never fired. The fix puts ischeckedout in the WHERE clause, so
# the database settles it at write time.

def test_a_checkout_that_passed_its_availability_check_is_still_refused(
        client, kiosk, monkeypatch):
    """Simulate the race window: the request holds a device object that says
    available while the row has already been claimed by somebody else."""
    from plugins.usb.api import selfhosted
    from plugins.usb.models import USBDevice

    first = _out(client, badge='270015376')
    assert first.status_code == 200

    # What the losing request sees: its own read, taken before the winner
    # committed, still says the drive is on the shelf.
    stale = USBDevice.query.filter_by(serialnumber='USB-1').first()
    monkeypatch.setattr(
        selfhosted, '_find',
        lambda device_id: _stale_copy(stale))

    second = _out(client, badge='270099999')
    assert second.status_code == 409, second.get_json()

    open_rows = _open_checkouts(USBDevice)
    assert len(open_rows) == 1, 'one drive, one open custody row'
    assert open_rows[0].sso == '270015376'


def _stale_copy(device):
    """A detached stand-in that still believes the drive is available."""
    class Stale:
        usbdeviceid = device.usbdeviceid
        serialnumber = device.serialnumber
        isactive = True
        ischeckedout = False
        currentuserid = None
        currentusername = None
    return Stale()


def _open_checkouts(_):
    from plugins.usb.models import USBCheckout
    return USBCheckout.query.filter_by(checkintime=None).all()


def test_a_return_records_who_handed_it_back_not_only_the_holder(client, kiosk):
    """Returning for a colleague is a supported flow, so the check-in has to
    name the person who actually walked it back."""
    from plugins.usb.models import USBCheckout

    assert _out(client, badge='270015376').status_code == 200
    returned = _in(client, badge='270099999', forsomeoneelse=True)
    assert returned.status_code == 200, returned.get_json()

    row = USBCheckout.query.order_by(USBCheckout.checkoutid.desc()).first()
    assert row.sso == '270015376', 'the holder still owns the checkout'
    assert row.checkinsso == '270099999', 'the returner is recorded'
    assert row.checkinname and 'Sam' in row.checkinname


def test_a_check_in_closes_every_open_row_for_that_device(client, kiosk, db):
    """Defence in depth for history written before the guard existed: a device
    can only be out once, so a second open row is corruption, and closing only
    the newest left the older ones open forever."""
    from plugins.usb.models import USBCheckout, USBDevice

    assert _out(client, badge='270015376').status_code == 200
    device = USBDevice.query.filter_by(serialnumber='USB-1').first()
    db.session.add(USBCheckout(usbdeviceid=device.usbdeviceid, sso='270099999',
                               checkoutname='Sam Okafor'))
    db.session.commit()
    assert len(_open_checkouts(USBDevice)) == 2

    assert _in(client, badge='270015376').status_code == 200
    assert _open_checkouts(USBDevice) == []


# --- one badge vocabulary for the whole site --------------------------------
#
# The parts kiosk understood both shapes a scanner emits; this one took the
# string raw and compared it to an SSO. So an operator whose keyboard-wedge
# scanner emits the PayNo wrapper could take a 3D-printed part and could not
# check out a drive, with the same badge, at two cabinets a few metres apart.
#
# The same change closed a subtler hole: str.isdigit() is true for Arabic-Indic
# digits, so a badge written in them resolved through int() to a REAL employee
# and was then stored as an SSO string that joins to nothing.

def test_the_payno_wrapper_works_at_this_kiosk_too(client, kiosk):
    from plugins.usb.models import USBCheckout

    response = _out(client, badge='0270015376BZ')
    assert response.status_code == 200, response.get_json()

    row = USBCheckout.query.order_by(USBCheckout.checkoutid.desc()).first()
    assert row.sso == '270015376', \
        'the custody row records the SSO, not the wrapper it arrived in'
    assert 'Dana' in (row.checkoutname or '')


def test_a_badge_in_non_ascii_digits_is_refused(client, kiosk):
    """It looks like a number to str.isdigit() and parses through int(), so it
    resolved to a real employee under an SSO nothing else matches."""
    # Written as escapes so this file stays ASCII, which the style gate
    # enforces: these are Arabic-Indic digits, U+0669 and friends.
    arabic_indic = '\u0669\u0667\u0660\u0660\u0660\u0667\u0661'
    response = _out(client, badge=arabic_indic)
    assert response.status_code == 422, response.get_json()


def test_the_same_person_scanning_either_shape_is_one_holder(client, kiosk):
    """Wrapper out, bare number back: the check-in has to recognise it as the
    same person, or the drive can never be returned by its holder."""
    assert _out(client, badge='0270015376BZ').status_code == 200
    returned = _in(client, badge='270015376')
    assert returned.status_code == 200, returned.get_json()