diff --git a/plugins/usb/api/routes.py b/plugins/usb/api/routes.py index a5b1f89..2863d1d 100644 --- a/plugins/usb/api/routes.py +++ b/plugins/usb/api/routes.py @@ -1,10 +1,14 @@ """USB plugin API endpoints. -Check-in/out state lives in a separate MySQL database (cmmc_usb), reached with -raw parameterized pymysql via cmmc_usb_connection(), mirroring the read-only -employee-directory helper pattern. The SQLAlchemy USBDevice/USBCheckout/ -USBDeviceType models in ../models are NOT used by these routes anymore; they -are left in place for any legacy importers. +Two modes, chosen by the usb_directory_mode setting: + - 'selfhosted' (default): the app-owned tables (usbdevices/usbcheckouts/ + usbdevicetypes) via SQLAlchemy - see selfhosted.py. Each route below + delegates when _usb_selfhosted() is true. + - 'external': a separate MySQL database (cmmc_usb) reached with raw + parameterized pymysql via cmmc_usb_connection(), for a site that already + runs the cmmc_usb solution. That is the raw-SQL code in this file. + +Both modes return the same response shape so the frontend is identical. cmmc_usb schema used here: devices(device_id PK, device_desc, device_owner, status, locker_location) @@ -33,11 +37,20 @@ from shopdb.api import ( get_pagination_params, require_permission, ) +from shopdb.core.models import Setting + +from . import selfhosted logger = logging.getLogger(__name__) usb_bp = Blueprint('usb', __name__) + +def _usb_selfhosted(): + """True when USB uses the app-owned tables, not the external cmmc_usb DB.""" + row = Setting.query.filter_by(key='usb_directory_mode').first() + return (row.value if row and row.value else 'external').lower() == 'selfhosted' + # public status strings on the devices row _STATUS_CHECKED_IN = 'checked-in' _STATUS_CHECKED_OUT = 'checked-out' @@ -253,6 +266,8 @@ def list_usb_devices(): - status: available | checkedout | retired - search: match on device_id or device_desc """ + if _usb_selfhosted(): + return selfhosted.list_devices(request) page, per_page = get_pagination_params(request) where = [] @@ -294,6 +309,8 @@ def list_usb_devices(): @jwt_required(optional=True) def get_usb_device(device_id): """Get a single device plus its last 20 check-in/out log rows.""" + if _usb_selfhosted(): + return selfhosted.get_device(device_id) conn = cmmc_usb_connection() try: with conn.cursor() as cur: @@ -325,6 +342,8 @@ def get_usb_device(device_id): @require_permission('usb.create') def create_usb_device(): """Add a device (starts checked-in).""" + if _usb_selfhosted(): + return selfhosted.create_device(request.get_json() or {}) data = request.get_json() or {} device_id = (data.get('device_id') or '').strip() @@ -370,6 +389,8 @@ def create_usb_device(): @require_permission('usb.edit') def update_usb_device(device_id): """Edit device_desc / device_owner / locker_location / status.""" + if _usb_selfhosted(): + return selfhosted.update_device(device_id, request.get_json() or {}) data = request.get_json() or {} conn = cmmc_usb_connection() @@ -411,6 +432,8 @@ def update_usb_device(device_id): @require_permission('usb.edit') def retire_usb_device(device_id): """Retire a device (status -> retired).""" + if _usb_selfhosted(): + return selfhosted.retire_device(device_id) conn = cmmc_usb_connection() try: with conn.cursor() as cur: @@ -444,6 +467,8 @@ def retire_usb_device(device_id): @require_permission('usb.create') def checkout_device(device_id): """Check a device out to a badge.""" + if _usb_selfhosted(): + return selfhosted.checkout_device(device_id, request.get_json() or {}) data = request.get_json() or {} badge = (data.get('badge') or '').strip() @@ -507,6 +532,8 @@ def checkout_device(device_id): @require_permission('usb.create') def checkin_device(device_id): """Check a device back in from a badge.""" + if _usb_selfhosted(): + return selfhosted.checkin_device(device_id, request.get_json() or {}) data = request.get_json() or {} badge = (data.get('badge') or '').strip() @@ -576,6 +603,8 @@ def checkin_device(device_id): @jwt_required(optional=True) def get_device_history(device_id): """Paginated check-in/out log for one device.""" + if _usb_selfhosted(): + return selfhosted.device_history(device_id, request) page, per_page = get_pagination_params(request) conn = cmmc_usb_connection() @@ -609,6 +638,8 @@ def list_all_checkouts(): - active: true -> only rows whose device is currently checked out - badge: filter by badge_number """ + if _usb_selfhosted(): + return selfhosted.list_checkouts(request, active_only=False) page, per_page = get_pagination_params(request) where = ["l.action = 'check-out'"] @@ -653,6 +684,8 @@ def list_all_checkouts(): @jwt_required(optional=True) def list_active_checkouts(): """List the latest check-out log row for every currently checked-out device.""" + if _usb_selfhosted(): + return selfhosted.list_checkouts(request, active_only=True) conn = cmmc_usb_connection() try: with conn.cursor() as cur: diff --git a/plugins/usb/api/selfhosted.py b/plugins/usb/api/selfhosted.py new file mode 100644 index 0000000..1c51065 --- /dev/null +++ b/plugins/usb/api/selfhosted.py @@ -0,0 +1,260 @@ +"""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 + +from shopdb.api import ( + db, success_response, error_response, ErrorCodes, + get_pagination_params, paginated_response, +) + +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 = DirectoryEmployee.query.get(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) + 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.utcnow() + db.session.add(USBCheckout(usbdeviceid=device.usbdeviceid, machineid=0, sso=badge, + checkoutname=name, checkouttime=now, + checkoutreason=data.get('reason'))) + device.ischeckedout = True + device.currentuserid = badge + device.currentusername = name + device.currentcheckoutdate = now + 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: + open_checkout.checkintime = datetime.utcnow() + 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)