ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
347402896ccf7735e0c1752fbe41df37f6cafb60
shopdb-flask/plugins/usb/api/selfhosted.py
cproudlock 60c35728e8 Import USB records off an external cmmc_usb database, so a site can retire it 
A site running external mode keeps its devices and check-in/out history in a
separate cmmc_usb database. There was no way to bring that in, so the choice
was to keep the second database forever or lose the history. This adds
'flask usb import-external' - dry run by default, --commit to write - which
reads through the same connection settings the routes already use and never
writes to the source.

Two things it has to reconcile, and they are why it is a command rather than a
few INSERTs.

The schemas disagree about SHAPE. External is an event log, one row per action;
self-hosted is a paired row carrying both times. Pairing walks each device in
time order, and real logs after years of use are untidy. The rule is that
nothing is invented and nothing vanishes: a drive still out becomes an open
checkout and is NOT flagged (that is just a drive somebody has), a return with
no matching issue is skipped and named, and two issues in a row leave the first
open rather than inventing a return time nobody recorded.

Identity is the other. usbdevices.serialnumber is unique and NOT NULL and the
external devices table has no serial, so device_id becomes it. Blank or
duplicated ids stop that device rather than inventing a key. external
device_owner has no column here, so it is appended to the device notes rather
than dropped.

Also adds usbcheckouts.scannedviruses, the one field with nowhere to land.
waswiped already means what external calls sanitized - selfhosted.py was
mapping it - but scanned_viruses was HARDCODED None in the log output: the API
promised the field and had nothing behind it, so an imported history would have
rendered blank. It now reads and writes properly.

Nullable, never defaulted to false. An unrecorded scan is not a failed scan,
and for removable-media records those are very different claims.

Full suite 2131 passed; seven new tests pin the pairing rules.
2026-08-29 13:02:21 -04:00

506 lines
21 KiB
Python
Raw Blame History

"""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
Reference in New Issue View Git Blame Copy Permalink