506 lines
21 KiB
Python
506 lines
21 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, timedelta, 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',
|
|
# Both read from the checkout row now. scanned_viruses was
|
|
# hardcoded None: the shape promised the field and had nowhere to
|
|
# read it from, so an imported history would have shown blank.
|
|
'scanned_viruses': checkout.scannedviruses,
|
|
'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'))
|
|
# None, not False, when the caller said nothing: an unrecorded scan is
|
|
# not a failed scan, and an audit reads those very differently.
|
|
scanned = data.get('scanned_viruses')
|
|
open_checkout.scannedviruses = None if scanned is None else bool(scanned)
|
|
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')
|
|
|
|
|
|
def audit_rows(days):
|
|
"""Check-out / check-in events over a window (self-hosted mode).
|
|
|
|
One usbcheckouts row holds BOTH ends of a loan, so it expands into two
|
|
events - the take and, when it has come back, the return. Reporting the row
|
|
once would make a returned drive indistinguishable from one still out.
|
|
|
|
Rows only: the caller owns the CSV, the summary and the response shape, so
|
|
both directory modes answer in exactly one voice.
|
|
"""
|
|
cutoff = None
|
|
if days > 0:
|
|
cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=days)
|
|
|
|
query = (db.session.query(USBCheckout, USBDevice)
|
|
.join(USBDevice, USBDevice.usbdeviceid == USBCheckout.usbdeviceid))
|
|
if cutoff is not None:
|
|
# Either end of the loan can fall in the window: a drive taken before it
|
|
# and returned inside it is a check-in the window has to show.
|
|
query = query.filter(db.or_(USBCheckout.checkouttime >= cutoff,
|
|
USBCheckout.checkintime >= cutoff))
|
|
|
|
rows = []
|
|
for checkout, device in query.all():
|
|
base = {
|
|
'device_id': device.serialnumber,
|
|
'device_desc': device.label or device.productname or '',
|
|
'badge': checkout.sso or '',
|
|
'name': checkout.checkoutname or '',
|
|
'locker_location': device.storagelocation or '',
|
|
}
|
|
if checkout.checkouttime and (cutoff is None or checkout.checkouttime >= cutoff):
|
|
rows.append(dict(base, timestamp=checkout.checkouttime.isoformat(),
|
|
action='check-out'))
|
|
if checkout.checkintime and (cutoff is None or checkout.checkintime >= cutoff):
|
|
rows.append(dict(base, timestamp=checkout.checkintime.isoformat(),
|
|
action='check-in'))
|
|
|
|
# Newest first, and a total order so two events in the same second do not
|
|
# swap places between identical runs.
|
|
rows.sort(key=lambda row: (row['timestamp'], row['device_id']), reverse=True)
|
|
return rows
|