7c47ae1f7c
The people who actually take USB drives are standing at the cabinet, and most of them have no account here. Check-out and check-in were gated behind a login that was not doing much anyway: both modes already attribute the action to a scanned BADGE rather than to the signed-in user, so the login was gating the screen, not the record. /usb-kiosk is a full-screen touch page modelled on the parts kiosk, off unless a site sets usb_kiosk_enabled - every route 404s until then, rather than 403, so a disabled kiosk does not advertise that there is a door here. It works against the self-hosted tables or an existing cmmc_usb database. THE LIST IS THE TRADE. These drives carry no barcode, so unlike the parts kiosk there is nothing to present and the screen has to offer a list - which exposes more than a scan-first design, where you had to already hold a drive to learn anything. So the list is not anonymous: an SSO comes first and must resolve to a real person before any drive is named, which is the same bar taking a drive already had to clear. It is a POST so the badge stays out of proxy access logs. The main screen shows the caller's own drives, what is free, and a COUNT of everything else; holders of other people's drives appear only behind a second deliberate tap, so nobody reads the whole cabinet off a wall panel in passing. A badge that resolves to no employee is REFUSED, unlike the signed-in route which creates the user row. A checkout nobody can trace to a person defeats the tracking, and at a kiosk there is no operator standing there to vouch for it. Returning somebody else's drive asks first and then records the badge that was SCANNED, so the log says who actually handed it back. That is the honest record and it needs no change to a schema this app does not own. Refusing a mismatch outright is a setting but not the default: a drive left behind by somebody on leave would otherwise be unreturnable at the kiosk, which just moves the problem to a person with a login. The authz sweep caught these writes, twice - once for the two mutations and again for the two list endpoints. Both are exempted with the reasoning written out beside the parts-kiosk entry rather than silenced. The list endpoints are POSTs that change nothing. Verified by driving the real page: SSO on the keypad, tap Take, database moves.
455 lines
18 KiB
Python
455 lines
18 KiB
Python
"""Self-hosted USB check-in/out, backed by the app-owned tables.
|
|
|
|
Used when usb_directory_mode=selfhosted: the USB routes call these functions
|
|
instead of the external cmmc_usb database. Data lives in the app tables
|
|
(usbdevices, usbcheckouts, usbdevicetypes); responses match the same shape the
|
|
external path returns so the frontend is identical in both modes.
|
|
|
|
Field mapping (response contract <- app model):
|
|
device_id <- USBDevice.serialnumber
|
|
device_desc <- USBDevice.label
|
|
status <- derived: retired (not isactive) / checked-out / checked-in
|
|
locker_location <- USBDevice.storagelocation
|
|
current_holder <- USBDevice.currentuserid (SSO)
|
|
checkout_time <- USBDevice.currentcheckoutdate
|
|
History rows are synthesized from usbcheckouts (each row = a check-out event and,
|
|
if returned, a check-in event).
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from shopdb.api import (
|
|
db, success_response, error_response, ErrorCodes,
|
|
get_pagination_params, paginated_response,
|
|
apply_import_timestamps, import_mode_active, parse_import_datetime,
|
|
)
|
|
|
|
from ..models import USBDevice, USBCheckout
|
|
|
|
STATUS_CHECKED_IN = 'checked-in'
|
|
STATUS_CHECKED_OUT = 'checked-out'
|
|
STATUS_RETIRED = 'retired'
|
|
_FILTER = {'available': STATUS_CHECKED_IN, 'checkedout': STATUS_CHECKED_OUT, 'retired': STATUS_RETIRED}
|
|
|
|
|
|
def _resolve_name(sso):
|
|
"""Best-effort name from the self-hosted employee directory."""
|
|
try:
|
|
from plugins.employees.models import DirectoryEmployee
|
|
if sso and str(sso).isdigit():
|
|
emp = db.session.get(DirectoryEmployee, int(sso))
|
|
if emp:
|
|
return f'{emp.firstname} {emp.lastname}'.strip()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def _status(device):
|
|
if not device.isactive:
|
|
return STATUS_RETIRED
|
|
return STATUS_CHECKED_OUT if device.ischeckedout else STATUS_CHECKED_IN
|
|
|
|
|
|
def _device_dict(device):
|
|
return {
|
|
'device_id': device.serialnumber,
|
|
'device_desc': device.label,
|
|
'device_owner': None,
|
|
'owner_name': None,
|
|
'status': _status(device),
|
|
'locker_location': device.storagelocation,
|
|
'current_holder': device.currentuserid,
|
|
'current_holder_name': device.currentusername,
|
|
'checkout_time': device.currentcheckoutdate.isoformat() + 'Z' if device.currentcheckoutdate else None,
|
|
}
|
|
|
|
|
|
def _checkout_events(checkout):
|
|
"""One usbcheckouts row -> its check-out event (+ check-in event if returned)."""
|
|
name = checkout.checkoutname or _resolve_name(checkout.sso)
|
|
events = [{
|
|
'log_id': checkout.checkoutid * 2,
|
|
'badge_number': checkout.sso,
|
|
'badge_name': name,
|
|
'device_id': checkout.device.serialnumber if checkout.device else None,
|
|
'action': 'check-out',
|
|
'timestamp': checkout.checkouttime.isoformat() + 'Z' if checkout.checkouttime else None,
|
|
'scanned_viruses': None,
|
|
'locker_location': None,
|
|
'sanitized': None,
|
|
}]
|
|
if checkout.checkintime:
|
|
events.append({
|
|
'log_id': checkout.checkoutid * 2 + 1,
|
|
'badge_number': checkout.sso,
|
|
'badge_name': name,
|
|
'device_id': checkout.device.serialnumber if checkout.device else None,
|
|
'action': 'check-in',
|
|
'timestamp': checkout.checkintime.isoformat() + 'Z',
|
|
'scanned_viruses': None,
|
|
'locker_location': None,
|
|
'sanitized': checkout.waswiped,
|
|
})
|
|
return events
|
|
|
|
|
|
def _find(device_id):
|
|
return USBDevice.query.filter_by(serialnumber=device_id).first()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def list_devices(request):
|
|
page, per_page = get_pagination_params(request)
|
|
query = USBDevice.query
|
|
status_arg = request.args.get('status', '').strip().lower()
|
|
if status_arg in _FILTER:
|
|
want = _FILTER[status_arg]
|
|
if want == STATUS_RETIRED:
|
|
query = query.filter_by(isactive=False)
|
|
elif want == STATUS_CHECKED_OUT:
|
|
query = query.filter_by(isactive=True, ischeckedout=True)
|
|
else:
|
|
query = query.filter_by(isactive=True, ischeckedout=False)
|
|
search = request.args.get('search', '').strip()
|
|
if search:
|
|
like = f'%{search}%'
|
|
query = query.filter(db.or_(USBDevice.serialnumber.ilike(like),
|
|
USBDevice.label.ilike(like)))
|
|
total = query.count()
|
|
rows = (query.order_by(USBDevice.serialnumber)
|
|
.limit(per_page).offset((page - 1) * per_page).all())
|
|
return paginated_response([_device_dict(d) for d in rows], page, per_page, total)
|
|
|
|
|
|
def get_device(device_id):
|
|
device = _find(device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
|
result = _device_dict(device)
|
|
checkouts = (USBCheckout.query.filter_by(usbdeviceid=device.usbdeviceid)
|
|
.order_by(USBCheckout.checkouttime.desc()).limit(20).all())
|
|
logs = []
|
|
for checkout in checkouts:
|
|
logs.extend(_checkout_events(checkout))
|
|
logs.sort(key=lambda item: item['timestamp'] or '', reverse=True)
|
|
result['checkinoutlog'] = logs[:20]
|
|
return success_response(result)
|
|
|
|
|
|
def create_device(data):
|
|
device_id = (data.get('device_id') or '').strip()
|
|
if not device_id:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'device_id is required')
|
|
if _find(device_id):
|
|
return error_response(ErrorCodes.CONFLICT, 'device_id already exists', http_code=409)
|
|
device = USBDevice(serialnumber=device_id, label=data.get('device_desc'),
|
|
storagelocation=data.get('locker_location'),
|
|
ischeckedout=False, isactive=True)
|
|
db.session.add(device)
|
|
apply_import_timestamps(device, data)
|
|
db.session.commit()
|
|
return success_response(_device_dict(device), message='Device created', http_code=201)
|
|
|
|
|
|
def update_device(device_id, data):
|
|
device = _find(device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
|
if 'device_desc' in data:
|
|
device.label = data['device_desc']
|
|
if 'locker_location' in data:
|
|
device.storagelocation = data['locker_location']
|
|
if 'status' in data:
|
|
if data['status'] == STATUS_RETIRED:
|
|
device.isactive = False
|
|
elif data['status'] in (STATUS_CHECKED_IN, STATUS_CHECKED_OUT):
|
|
device.isactive = True
|
|
device.ischeckedout = data['status'] == STATUS_CHECKED_OUT
|
|
db.session.commit()
|
|
return success_response(_device_dict(device), message='Device updated')
|
|
|
|
|
|
def retire_device(device_id):
|
|
device = _find(device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
|
device.isactive = False
|
|
db.session.commit()
|
|
return success_response(_device_dict(device), message='Device retired')
|
|
|
|
|
|
def checkout_device(device_id, data):
|
|
badge = (data.get('badge') or '').strip()
|
|
if not badge:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'badge is required')
|
|
device = _find(device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
|
if device.ischeckedout:
|
|
return error_response(ErrorCodes.CONFLICT, 'Device is already checked out', http_code=409)
|
|
name = _resolve_name(badge)
|
|
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
# Backdated import: an admin import request may pass the historical
|
|
# checkouttime so migrated usbcheckouts rows keep their real event time.
|
|
eventtime = now
|
|
if import_mode_active():
|
|
override = parse_import_datetime(data.get('checkouttime'))
|
|
if override is not None:
|
|
eventtime = override
|
|
db.session.add(USBCheckout(usbdeviceid=device.usbdeviceid, sso=badge,
|
|
checkoutname=name, checkouttime=eventtime,
|
|
checkoutreason=data.get('reason')))
|
|
device.ischeckedout = True
|
|
device.currentuserid = badge
|
|
device.currentusername = name
|
|
device.currentcheckoutdate = eventtime
|
|
if data.get('locker_location'):
|
|
device.storagelocation = data['locker_location']
|
|
db.session.commit()
|
|
return success_response(_device_dict(device), message='Device checked out', http_code=201)
|
|
|
|
|
|
def checkin_device(device_id, data):
|
|
badge = (data.get('badge') or '').strip()
|
|
if not badge:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'badge is required')
|
|
device = _find(device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
|
if not device.ischeckedout:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'Device is not checked out')
|
|
open_checkout = (USBCheckout.query
|
|
.filter_by(usbdeviceid=device.usbdeviceid, checkintime=None)
|
|
.order_by(USBCheckout.checkouttime.desc()).first())
|
|
if open_checkout:
|
|
checkintime = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
# Backdated import: accept the historical checkintime from an admin
|
|
# import request so returned checkouts keep their real return time.
|
|
if import_mode_active():
|
|
override = parse_import_datetime(data.get('checkintime'))
|
|
if override is not None:
|
|
checkintime = override
|
|
open_checkout.checkintime = checkintime
|
|
open_checkout.waswiped = bool(data.get('sanitized'))
|
|
open_checkout.checkinnotes = data.get('notes')
|
|
device.ischeckedout = False
|
|
device.currentuserid = None
|
|
device.currentusername = None
|
|
device.currentcheckoutdate = None
|
|
if data.get('locker_location'):
|
|
device.storagelocation = data['locker_location']
|
|
db.session.commit()
|
|
return success_response(_device_dict(device), message='Device checked in')
|
|
|
|
|
|
def device_history(device_id, request):
|
|
device = _find(device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND, f'USB device {device_id} not found', http_code=404)
|
|
page, per_page = get_pagination_params(request)
|
|
checkouts = (USBCheckout.query.filter_by(usbdeviceid=device.usbdeviceid)
|
|
.order_by(USBCheckout.checkouttime.desc()).all())
|
|
logs = []
|
|
for checkout in checkouts:
|
|
logs.extend(_checkout_events(checkout))
|
|
logs.sort(key=lambda item: item['timestamp'] or '', reverse=True)
|
|
total = len(logs)
|
|
start = (page - 1) * per_page
|
|
return paginated_response(logs[start:start + per_page], page, per_page, total)
|
|
|
|
|
|
def list_checkouts(request, active_only):
|
|
query = USBCheckout.query
|
|
if active_only:
|
|
query = query.filter_by(checkintime=None)
|
|
badge = request.args.get('badge', '').strip()
|
|
if badge:
|
|
query = query.filter_by(sso=badge)
|
|
rows = query.order_by(USBCheckout.checkouttime.desc()).all()
|
|
data = []
|
|
for checkout in rows:
|
|
event = _checkout_events(checkout)[0]
|
|
event['device_status'] = _status(checkout.device) if checkout.device else None
|
|
data.append(event)
|
|
return success_response(data)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Kiosk (self-hosted mode). Same contract as the external path in routes.py:
|
|
# move a device between checked-out and checked-in, attributed to a badge that
|
|
# resolved to a real person, and nothing else.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def kiosk_device(device_id):
|
|
device = _find(device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'No device matches that code', http_code=404)
|
|
return success_response({
|
|
'device_id': device.serialnumber,
|
|
'device_desc': device.label or device.productname or '',
|
|
'status': _status(device),
|
|
'locker_location': device.storagelocation,
|
|
# Name only. The kiosk needs it for the return-for-someone-else
|
|
# confirm; it does not need their badge.
|
|
'holdername': device.currentusername or '' if device.ischeckedout else '',
|
|
})
|
|
|
|
|
|
def kiosk_session(data):
|
|
"""See routes.kiosk_session for why the list costs a valid badge."""
|
|
badge = (data.get('badge') or '').strip()
|
|
if not badge:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'Scan your badge')
|
|
name = _resolve_name(badge)
|
|
if not name:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'Badge not recognized - see IT', http_code=422)
|
|
|
|
yours, available, others = [], [], 0
|
|
for device in USBDevice.query.filter_by(isactive=True).order_by(
|
|
USBDevice.serialnumber).all():
|
|
entry = {
|
|
'device_id': device.serialnumber,
|
|
'device_desc': device.label or device.productname or '',
|
|
'locker_location': device.storagelocation,
|
|
}
|
|
if not device.ischeckedout:
|
|
available.append(entry)
|
|
elif (device.currentuserid or '').strip() == badge:
|
|
yours.append(entry)
|
|
else:
|
|
others += 1
|
|
|
|
return success_response({'name': name, 'yours': yours,
|
|
'available': available, 'otherscount': others})
|
|
|
|
|
|
def kiosk_others(data):
|
|
badge = (data.get('badge') or '').strip()
|
|
if not badge:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'Scan your badge')
|
|
if not _resolve_name(badge):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'Badge not recognized - see IT', http_code=422)
|
|
|
|
out = []
|
|
for device in USBDevice.query.filter_by(isactive=True, ischeckedout=True
|
|
).order_by(USBDevice.serialnumber).all():
|
|
if (device.currentuserid or '').strip() == badge:
|
|
continue
|
|
out.append({
|
|
'device_id': device.serialnumber,
|
|
'device_desc': device.label or device.productname or '',
|
|
'locker_location': device.storagelocation,
|
|
'holdername': device.currentusername or '',
|
|
})
|
|
return success_response(out)
|
|
|
|
|
|
def kiosk_checkout(data):
|
|
device_id = (data.get('device_id') or '').strip()
|
|
badge = (data.get('badge') or '').strip()
|
|
if not badge:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'Scan your badge')
|
|
|
|
device = _find(device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'No device matches that code', http_code=404)
|
|
if not device.isactive:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'That device is retired - see IT')
|
|
if device.ischeckedout:
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
'That device is already checked out{}'.format(
|
|
' to {}'.format(device.currentusername)
|
|
if device.currentusername else ''),
|
|
http_code=409)
|
|
|
|
# Refused rather than recorded anonymously - see the note on the external
|
|
# path. A checkout nobody can trace to a person defeats the tracking.
|
|
name = _resolve_name(badge)
|
|
if not name:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Badge not recognized - see IT to check this device out',
|
|
http_code=422)
|
|
|
|
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
db.session.add(USBCheckout(usbdeviceid=device.usbdeviceid, sso=badge,
|
|
checkoutname=name, checkouttime=now))
|
|
device.ischeckedout = True
|
|
device.currentuserid = badge
|
|
device.currentusername = name
|
|
device.currentcheckoutdate = now
|
|
db.session.commit()
|
|
return success_response({'device_id': device.serialnumber,
|
|
'holdername': name},
|
|
message='{} checked out to {}'.format(
|
|
device.serialnumber, name))
|
|
|
|
|
|
def kiosk_checkin(data):
|
|
device_id = (data.get('device_id') or '').strip()
|
|
badge = (data.get('badge') or '').strip()
|
|
if not badge:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'Scan your badge')
|
|
forsomeoneelse = bool(data.get('forsomeoneelse'))
|
|
|
|
device = _find(device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'No device matches that code', http_code=404)
|
|
if not device.ischeckedout:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'That device is not checked out')
|
|
|
|
name = _resolve_name(badge)
|
|
if not name:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Badge not recognized - see IT to return this device',
|
|
http_code=422)
|
|
|
|
holderbadge = (device.currentuserid or '').strip()
|
|
if holderbadge and holderbadge != badge:
|
|
if not _kiosk_flag('usb_kiosk_allow_return_for_others', True):
|
|
return error_response(
|
|
ErrorCodes.FORBIDDEN,
|
|
'That device is checked out to somebody else - they need to '
|
|
'return it, or see IT',
|
|
http_code=403)
|
|
if not forsomeoneelse:
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
'That device is checked out to {}. Confirm you are returning '
|
|
'it for them.'.format(device.currentusername or 'somebody else'),
|
|
details={'needsconfirm': True,
|
|
'holdername': device.currentusername or ''},
|
|
http_code=409)
|
|
|
|
open_checkout = (USBCheckout.query
|
|
.filter_by(usbdeviceid=device.usbdeviceid, checkintime=None)
|
|
.order_by(USBCheckout.checkouttime.desc()).first())
|
|
if open_checkout:
|
|
open_checkout.checkintime = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
device.ischeckedout = False
|
|
device.currentuserid = None
|
|
device.currentusername = None
|
|
device.currentcheckoutdate = None
|
|
db.session.commit()
|
|
return success_response({'device_id': device.serialnumber},
|
|
message='{} checked in'.format(device.serialnumber))
|
|
|
|
|
|
def _kiosk_flag(key, default):
|
|
from shopdb.api import Setting
|
|
row = Setting.query.filter_by(key=key).first()
|
|
if row is None or row.value is None or row.value == '':
|
|
return default
|
|
return str(row.value).strip().lower() in ('true', '1', 'yes', 'on')
|