7ac1ee3077
Both kiosks admit a check-in or check-out with no login when the caller sits in a configured network. The subtraction that makes that safe was missing. A reverse proxy normally sits ON the plant network a site enters in the allowlist. When the proxy forwards nothing - misconfigured, or something bypassed it - the request arrives wearing the proxy's own address, which matches the allowlist, while carrying no evidence whatever about who sent it. That is the one case that has to refuse, and it was being admitted. The previous fail-closed test only looked closed because the proxy in it happened to sit outside the range. Addresses that belong to the path are now subtracted before the allowlist is consulted: loopback always, since no site's kiosk is the server itself, plus whatever the site declares as its proxies. USB and printedparts get the same treatment - they had diverged for no reason anybody chose. Three more, same surface: A range that admits everybody can no longer be saved by accident. 0.0.0.0/0 and ::/0 are refused outright, anything wider than a site plausibly needs is refused unless an explicit override is set, and the refusal says why. The kiosk token no longer travels in a URL. The provisioning link carries a single-use setup code redeemed on first open, so the token stays out of browser history, referrers and proxy logs. The route is declared public deliberately: the secret is the code, the way a password-reset link is the secret. The badge-to-name lookup is bounded. It answers without a login and is now reachable from a whole network rather than one screen, so a source that keeps missing gets refused rather than being allowed to walk the directory. On by default, budget and kill switch are settings. All four new settings carry controls on their settings pages. A setting that can stop the kiosk, or widen it, must not be reachable only by hand-editing rows. The installer's client-IP advice is corrected: it described the opposite of what the shipped waitress flags do, and named only GE-Enforce as the thing that breaks when the rule is absent, when the two kiosk allowlists depend on it too.
623 lines
26 KiB
Python
623 lines
26 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.
|
|
|
|
The badge goes through the shared shape rule first. This compared the raw
|
|
string, so the PayNo wrapper some scanners emit (0<digits>BZ) resolved to
|
|
nobody and the kiosk answered "Badge not recognized - see IT" - for a badge
|
|
the parts kiosk a few metres away accepted. It also used str.isdigit(),
|
|
true for Arabic-Indic digits, so a badge in those characters reached int()
|
|
and matched a real employee.
|
|
"""
|
|
from shopdb.api import BadgeShapeError, normalise_badge
|
|
|
|
try:
|
|
digits = normalise_badge(sso)
|
|
except BadgeShapeError:
|
|
return None
|
|
try:
|
|
from plugins.employees.models import DirectoryEmployee
|
|
emp = db.session.get(DirectoryEmployee, int(digits))
|
|
if emp:
|
|
return f'{emp.firstname} {emp.lastname}'.strip()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def _kiosk_badge_digits(raw):
|
|
"""Return (digits, errorresponse) for a scanned badge.
|
|
|
|
Every kiosk write stores the DIGITS rather than the string that arrived:
|
|
the PayNo wrapper (0<digits>BZ) and the bare number are the same person,
|
|
and recording one row under each spelling makes a custody history that
|
|
cannot be joined to itself.
|
|
|
|
An empty scan and an unreadable one get different answers, because they
|
|
are different problems for the person standing at the cabinet: one has not
|
|
scanned yet, the other has scanned something this site does not know.
|
|
"""
|
|
from shopdb.api import BadgeShapeError, normalise_badge
|
|
|
|
if not (raw or '').strip():
|
|
return None, error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'Scan your badge')
|
|
try:
|
|
return normalise_badge(raw), None
|
|
except BadgeShapeError:
|
|
return None, error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Badge not recognized - see IT', http_code=422)
|
|
|
|
|
|
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 _badge_not_recognised(suffix=''):
|
|
"""One unrecognised scan, counted against the caller, then the refusal.
|
|
|
|
WHY COUNTED. The answer to a recognised badge is an employee's NAME, and a
|
|
badge is a short number, so this is a lookup somebody admitted to the kiosk
|
|
can walk. Network-position admission can now mean a whole VLAN rather than
|
|
one screen by the cabinet, so the number of badges a caller may guess wrong
|
|
has to be bounded. The budget lives in routes.py alongside the admission
|
|
checks; the import is deferred because routes imports this module.
|
|
|
|
The wording stays per-action ("to check this device out"), because the
|
|
person at the screen is being told what did not happen.
|
|
"""
|
|
from .routes import record_badge_miss
|
|
|
|
record_badge_miss()
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'Badge not recognized - see IT' + suffix,
|
|
http_code=422)
|
|
|
|
|
|
def kiosk_session(data):
|
|
"""See routes.kiosk_session for why the list costs a valid badge."""
|
|
badge, badgeerror = _kiosk_badge_digits(data.get('badge'))
|
|
if badgeerror:
|
|
return badgeerror
|
|
name = _resolve_name(badge)
|
|
if not name:
|
|
return _badge_not_recognised()
|
|
|
|
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, badgeerror = _kiosk_badge_digits(data.get('badge'))
|
|
if badgeerror:
|
|
return badgeerror
|
|
if not _resolve_name(badge):
|
|
return _badge_not_recognised()
|
|
|
|
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, badgeerror = _kiosk_badge_digits(data.get('badge'))
|
|
if badgeerror:
|
|
return badgeerror
|
|
|
|
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 _badge_not_recognised(' to check this device out')
|
|
|
|
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
# The transition IS the guard. The ischeckedout test above reads, decides
|
|
# in Python and then writes, so two kiosk scans of the same drive that
|
|
# interleave both read "available" and both insert a custody row - one
|
|
# drive recorded out to two people, and the documented 409 never fires.
|
|
# Making the flag part of the WHERE clause moves the decision into the
|
|
# write, where the database settles it: the second UPDATE matches no row.
|
|
claimed = USBDevice.query.filter(
|
|
USBDevice.usbdeviceid == device.usbdeviceid,
|
|
USBDevice.ischeckedout == False).update(
|
|
{USBDevice.ischeckedout: True,
|
|
USBDevice.currentuserid: badge,
|
|
USBDevice.currentusername: name,
|
|
USBDevice.currentcheckoutdate: now},
|
|
synchronize_session=False)
|
|
if not claimed:
|
|
db.session.rollback()
|
|
# Re-read the row rather than refreshing the object this request is
|
|
# holding: the whole point of arriving here is that the object is
|
|
# stale, and the caller deserves the name of whoever actually has it.
|
|
holder = db.session.query(USBDevice.currentusername).filter(
|
|
USBDevice.usbdeviceid == device.usbdeviceid).scalar()
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
'That device is already checked out{}'.format(
|
|
' to {}'.format(holder) if holder else ''),
|
|
http_code=409)
|
|
db.session.add(USBCheckout(usbdeviceid=device.usbdeviceid, sso=badge,
|
|
checkoutname=name, checkouttime=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, badgeerror = _kiosk_badge_digits(data.get('badge'))
|
|
if badgeerror:
|
|
return badgeerror
|
|
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 _badge_not_recognised(' to return this device')
|
|
|
|
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)
|
|
|
|
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
# Guarded the same way as the checkout: only a device that is still out can
|
|
# be brought back, and the database decides that, not a Python read from
|
|
# earlier in this request.
|
|
released = USBDevice.query.filter(
|
|
USBDevice.usbdeviceid == device.usbdeviceid,
|
|
USBDevice.ischeckedout == True).update(
|
|
{USBDevice.ischeckedout: False,
|
|
USBDevice.currentuserid: None,
|
|
USBDevice.currentusername: None,
|
|
USBDevice.currentcheckoutdate: None},
|
|
synchronize_session=False)
|
|
if not released:
|
|
db.session.rollback()
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'That device is not checked out')
|
|
|
|
# EVERY open row, not just the newest. One device can only be out once, so
|
|
# a second open row is corruption - closing only the newest one left the
|
|
# older ones open forever, and the device then reads as returned while its
|
|
# custody history says otherwise.
|
|
#
|
|
# checkinsso/checkinname record WHO HANDED IT BACK, which is not always the
|
|
# holder: returning for a colleague is a supported flow, and attributing
|
|
# the check-in to the holder made the audit trail name somebody who was not
|
|
# there.
|
|
(USBCheckout.query
|
|
.filter_by(usbdeviceid=device.usbdeviceid, checkintime=None)
|
|
.update({USBCheckout.checkintime: now,
|
|
USBCheckout.checkinsso: badge,
|
|
USBCheckout.checkinname: name},
|
|
synchronize_session=False))
|
|
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 auditflag(value):
|
|
"""Render a stored tri-state as 'yes', 'no', or '' for the report.
|
|
|
|
THREE states, not two. Both stores keep these nullable on purpose: a null
|
|
means nobody recorded it, which is not the same claim as 'no'. Rendering an
|
|
unrecorded return as 'No' would put a assertion in an audit record that
|
|
nobody ever made - the opposite of what the record is for. So unknown stays
|
|
blank and reads as the gap it is.
|
|
"""
|
|
if value is None or value == '':
|
|
return ''
|
|
if isinstance(value, str):
|
|
return 'yes' if value.strip() in ('1', 'true', 'True') else 'no'
|
|
return 'yes' if value else 'no'
|
|
|
|
|
|
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):
|
|
# No wipe or scan state on a take: both are recorded on the return,
|
|
# so putting them here would invent a fact.
|
|
rows.append(dict(base, timestamp=checkout.checkouttime.isoformat(),
|
|
action='check-out',
|
|
sanitized='', scanned_viruses=''))
|
|
if checkout.checkintime and (cutoff is None or checkout.checkintime >= cutoff):
|
|
rows.append(dict(base, timestamp=checkout.checkintime.isoformat(),
|
|
action='check-in',
|
|
sanitized=auditflag(checkout.waswiped),
|
|
scanned_viruses=auditflag(checkout.scannedviruses)))
|
|
|
|
# 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
|