440 lines
18 KiB
Python
440 lines
18 KiB
Python
"""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()
|