f59aca5a1b
CMMC removable-media tracking is an audit obligation, and answering "who had this drive, and when" meant paging through the check-in/out list by hand. GET /api/usb/reports/audit returns the events - a take and a return are separate rows, because a count cannot answer the question an audit actually asks - with summary figures riding along for the people who only want the shape of it, and a CSV for the people who want it in a spreadsheet. The window is named rather than a day count: today, last month, six months, a year, all time. Nobody should have to work out that a year is 365. Both directory modes answer in one voice. The external path reads the cmmc_usb checkinoutlog; self-hosted expands one usbcheckouts row into its two ends, so a returned drive is distinguishable from one still out, and the window filter runs in SQL against either end of the loan - a drive taken before the window and returned inside it is a check-in the window has to show. Rows come back from the mode-specific half; the CSV, the summary and the response shape are built once in the route. Ordering is total in both modes. Timestamp alone ties for two events in the same second, and the same query then returns them in a different order each run, which reads as the log being unstable.
1193 lines
42 KiB
Python
1193 lines
42 KiB
Python
"""USB plugin API endpoints.
|
|
|
|
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)
|
|
status is one of 'checked-in' | 'checked-out' | 'retired'
|
|
users(badge_number PK, first_name, last_name)
|
|
checkinoutlog(log_id PK auto, badge_number, device_id, action, timestamp,
|
|
scanned_viruses, locker_location, sanitized)
|
|
action is one of 'check-in' | 'check-out'; timestamp is a
|
|
'YYYY-MM-DD HH:MM:SS' string; scanned_viruses/sanitized are '0'|'1'|null.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
|
|
from flask import Blueprint, request, Response
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
from shopdb.api import (
|
|
cmmc_usb_connection,
|
|
employee_connection,
|
|
success_response,
|
|
error_response,
|
|
paginated_response,
|
|
ErrorCodes,
|
|
get_pagination_params,
|
|
require_permission,
|
|
)
|
|
from shopdb.api 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'
|
|
_STATUS_RETIRED = 'retired'
|
|
|
|
# map the ?status= query values to the stored devices.status strings
|
|
_STATUS_FILTER = {
|
|
'available': _STATUS_CHECKED_IN,
|
|
'checkedout': _STATUS_CHECKED_OUT,
|
|
'retired': _STATUS_RETIRED,
|
|
}
|
|
|
|
# columns pulled for a devices row
|
|
_DEVICE_COLS = 'device_id, device_desc, device_owner, status, locker_location'
|
|
|
|
# columns pulled for a checkinoutlog row
|
|
_LOG_COLS = (
|
|
'log_id, badge_number, device_id, action, timestamp, '
|
|
'scanned_viruses, locker_location, sanitized'
|
|
)
|
|
|
|
# a badge like 0123456BZ carries a PayNo (the digits) wrapped in 0...BZ
|
|
_PAYNO_BADGE = re.compile(r'^0(\d+)BZ$', re.IGNORECASE)
|
|
|
|
|
|
# =============================================================================
|
|
# Module helpers
|
|
# =============================================================================
|
|
|
|
def _now():
|
|
"""Current local time as a 'YYYY-MM-DD HH:MM:SS' string for the log."""
|
|
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
|
|
def _flag(value):
|
|
"""Coerce a truthy/falsy input to a '1'/'0' string, or None when unset."""
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bool):
|
|
return '1' if value else '0'
|
|
text = str(value).strip().lower()
|
|
if text in ('1', 'true', 'yes', 'y'):
|
|
return '1'
|
|
if text in ('0', 'false', 'no', 'n', ''):
|
|
return '0'
|
|
return None
|
|
|
|
|
|
def _lookup_employee_name(badge):
|
|
"""Look up "First Last" in the HR directory for a badge. Empty on any miss.
|
|
|
|
A numeric badge is an SSO. A 0<digits>BZ badge carries a PayNo (the middle
|
|
digits). Any other shape has no HR mapping so we return "".
|
|
"""
|
|
badge = (badge or '').strip()
|
|
if not badge:
|
|
return ''
|
|
|
|
try:
|
|
conn = employee_connection()
|
|
except Exception:
|
|
logger.exception('HR directory connection failed for badge %s', badge)
|
|
return ''
|
|
|
|
try:
|
|
with conn.cursor() as ecur:
|
|
if badge.isdigit():
|
|
ecur.execute(
|
|
'SELECT First_Name, Last_Name FROM employees WHERE SSO = %s',
|
|
(badge,),
|
|
)
|
|
else:
|
|
match = _PAYNO_BADGE.match(badge)
|
|
if not match:
|
|
return ''
|
|
ecur.execute(
|
|
'SELECT First_Name, Last_Name FROM employees WHERE PayNo = %s',
|
|
(match.group(1),),
|
|
)
|
|
emp = ecur.fetchone()
|
|
except Exception:
|
|
logger.exception('HR directory lookup failed for badge %s', badge)
|
|
return ''
|
|
finally:
|
|
conn.close()
|
|
|
|
if not emp:
|
|
return ''
|
|
first = (emp.get('First_Name') or '').strip()
|
|
last = (emp.get('Last_Name') or '').strip()
|
|
return (first + ' ' + last).strip()
|
|
|
|
|
|
def _resolve_badge_name(cur, badge):
|
|
"""Resolve a badge to "First Last". Try cmmc_usb users, then HR directory.
|
|
|
|
cur is an open cmmc_usb cursor. Returns "" when the badge cannot be
|
|
resolved anywhere.
|
|
"""
|
|
badge = (badge or '').strip() if badge is not None else ''
|
|
if not badge:
|
|
return ''
|
|
|
|
cur.execute(
|
|
'SELECT first_name, last_name FROM users WHERE badge_number = %s',
|
|
(badge,),
|
|
)
|
|
row = cur.fetchone()
|
|
if row:
|
|
first = (row.get('first_name') or '').strip()
|
|
last = (row.get('last_name') or '').strip()
|
|
name = (first + ' ' + last).strip()
|
|
if name:
|
|
return name
|
|
|
|
return _lookup_employee_name(badge)
|
|
|
|
|
|
def _ensure_user(cur, badge, name):
|
|
"""Insert a badge into cmmc_usb users if absent. Splits name to first/last."""
|
|
badge = (badge or '').strip() if badge is not None else ''
|
|
if not badge:
|
|
return
|
|
|
|
cur.execute('SELECT badge_number FROM users WHERE badge_number = %s', (badge,))
|
|
if cur.fetchone():
|
|
return
|
|
|
|
first, last = '', ''
|
|
if name:
|
|
parts = name.split()
|
|
if parts:
|
|
first = parts[0]
|
|
last = ' '.join(parts[1:])
|
|
|
|
cur.execute(
|
|
'INSERT INTO users (badge_number, first_name, last_name) VALUES (%s, %s, %s)',
|
|
(badge, first, last),
|
|
)
|
|
|
|
|
|
def _fetch_device(cur, device_id):
|
|
"""Return the devices row for device_id, or None."""
|
|
cur.execute(
|
|
'SELECT ' + _DEVICE_COLS + ' FROM devices WHERE device_id = %s',
|
|
(device_id,),
|
|
)
|
|
return cur.fetchone()
|
|
|
|
|
|
def _device_to_dict(cur, row):
|
|
"""Build the device response dict, resolving owner + current holder names."""
|
|
device = {
|
|
'device_id': row.get('device_id'),
|
|
'device_desc': row.get('device_desc'),
|
|
'device_owner': row.get('device_owner'),
|
|
'owner_name': _resolve_badge_name(cur, row.get('device_owner')),
|
|
'status': row.get('status'),
|
|
'locker_location': row.get('locker_location'),
|
|
'current_holder': None,
|
|
'current_holder_name': None,
|
|
'checkout_time': None,
|
|
}
|
|
|
|
if row.get('status') == _STATUS_CHECKED_OUT:
|
|
cur.execute(
|
|
"SELECT badge_number, timestamp FROM checkinoutlog "
|
|
"WHERE device_id = %s AND action = 'check-out' "
|
|
"ORDER BY timestamp DESC, log_id DESC LIMIT 1",
|
|
(row.get('device_id'),),
|
|
)
|
|
log = cur.fetchone()
|
|
if log:
|
|
device['current_holder'] = log.get('badge_number')
|
|
device['current_holder_name'] = _resolve_badge_name(
|
|
cur, log.get('badge_number')
|
|
)
|
|
device['checkout_time'] = log.get('timestamp')
|
|
|
|
return device
|
|
|
|
|
|
def _log_to_dict(cur, row):
|
|
"""Build a checkinoutlog response dict with the resolved badge name."""
|
|
result = {
|
|
'log_id': row.get('log_id'),
|
|
'badge_number': row.get('badge_number'),
|
|
'badge_name': _resolve_badge_name(cur, row.get('badge_number')),
|
|
'device_id': row.get('device_id'),
|
|
'action': row.get('action'),
|
|
'timestamp': row.get('timestamp'),
|
|
'scanned_viruses': row.get('scanned_viruses'),
|
|
'locker_location': row.get('locker_location'),
|
|
'sanitized': row.get('sanitized'),
|
|
}
|
|
# /checkouts joins in the current device status; pass it through when present
|
|
if 'device_status' in row:
|
|
result['device_status'] = row.get('device_status')
|
|
return result
|
|
|
|
|
|
# =============================================================================
|
|
# USB Devices
|
|
# =============================================================================
|
|
|
|
@usb_bp.route('', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_usb_devices():
|
|
"""List USB devices with checkout status.
|
|
|
|
Query parameters:
|
|
- page, per_page: pagination
|
|
- 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 = []
|
|
params = []
|
|
|
|
status_arg = request.args.get('status', '').strip().lower()
|
|
if status_arg in _STATUS_FILTER:
|
|
where.append('status = %s')
|
|
params.append(_STATUS_FILTER[status_arg])
|
|
|
|
search = request.args.get('search', '').strip()
|
|
if search:
|
|
where.append('(device_id LIKE %s OR device_desc LIKE %s)')
|
|
like = '%' + search + '%'
|
|
params.extend([like, like])
|
|
|
|
where_sql = (' WHERE ' + ' AND '.join(where)) if where else ''
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT COUNT(*) AS total FROM devices' + where_sql, params)
|
|
total = cur.fetchone()['total']
|
|
|
|
cur.execute(
|
|
'SELECT ' + _DEVICE_COLS + ' FROM devices' + where_sql
|
|
+ ' ORDER BY device_id LIMIT %s OFFSET %s',
|
|
params + [per_page, (page - 1) * per_page],
|
|
)
|
|
rows = cur.fetchall()
|
|
data = [_device_to_dict(cur, row) for row in rows]
|
|
finally:
|
|
conn.close()
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@usb_bp.route('/<device_id>', methods=['GET'])
|
|
@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:
|
|
row = _fetch_device(cur, device_id)
|
|
if not row:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'USB device {device_id} not found',
|
|
http_code=404,
|
|
)
|
|
|
|
result = _device_to_dict(cur, row)
|
|
|
|
cur.execute(
|
|
'SELECT ' + _LOG_COLS + ' FROM checkinoutlog WHERE device_id = %s '
|
|
'ORDER BY timestamp DESC, log_id DESC LIMIT 20',
|
|
(device_id,),
|
|
)
|
|
logs = cur.fetchall()
|
|
result['checkinoutlog'] = [_log_to_dict(cur, log) for log in logs]
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@usb_bp.route('', methods=['POST'])
|
|
@jwt_required()
|
|
@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()
|
|
if not device_id:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'device_id is required')
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'SELECT device_id FROM devices WHERE device_id = %s', (device_id,)
|
|
)
|
|
if cur.fetchone():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT, 'device_id already exists', http_code=409
|
|
)
|
|
|
|
cur.execute(
|
|
'INSERT INTO devices '
|
|
'(device_id, device_desc, device_owner, status, locker_location) '
|
|
'VALUES (%s, %s, %s, %s, %s)',
|
|
(
|
|
device_id,
|
|
data.get('device_desc'),
|
|
data.get('device_owner'),
|
|
_STATUS_CHECKED_IN,
|
|
data.get('locker_location'),
|
|
),
|
|
)
|
|
conn.commit()
|
|
|
|
with conn.cursor() as cur:
|
|
row = _fetch_device(cur, device_id)
|
|
result = _device_to_dict(cur, row)
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response(result, message='Device created', http_code=201)
|
|
|
|
|
|
@usb_bp.route('/<device_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@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()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
if not _fetch_device(cur, device_id):
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'USB device {device_id} not found',
|
|
http_code=404,
|
|
)
|
|
|
|
fields = []
|
|
params = []
|
|
for col in ('device_desc', 'device_owner', 'locker_location', 'status'):
|
|
if col in data:
|
|
fields.append(col + ' = %s')
|
|
params.append(data[col])
|
|
|
|
if fields:
|
|
params.append(device_id)
|
|
cur.execute(
|
|
'UPDATE devices SET ' + ', '.join(fields) + ' WHERE device_id = %s',
|
|
params,
|
|
)
|
|
conn.commit()
|
|
|
|
with conn.cursor() as cur:
|
|
row = _fetch_device(cur, device_id)
|
|
result = _device_to_dict(cur, row)
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response(result, message='Device updated')
|
|
|
|
|
|
@usb_bp.route('/<device_id>/retire', methods=['POST'])
|
|
@jwt_required()
|
|
@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:
|
|
if not _fetch_device(cur, device_id):
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'USB device {device_id} not found',
|
|
http_code=404,
|
|
)
|
|
cur.execute(
|
|
'UPDATE devices SET status = %s WHERE device_id = %s',
|
|
(_STATUS_RETIRED, device_id),
|
|
)
|
|
conn.commit()
|
|
|
|
with conn.cursor() as cur:
|
|
row = _fetch_device(cur, device_id)
|
|
result = _device_to_dict(cur, row)
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response(result, message='Device retired')
|
|
|
|
|
|
# =============================================================================
|
|
# Check-in / check-out operations
|
|
# =============================================================================
|
|
|
|
@usb_bp.route('/<device_id>/checkout', methods=['POST'])
|
|
@jwt_required()
|
|
@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()
|
|
if not badge:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'badge is required')
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
device = _fetch_device(cur, device_id)
|
|
if not device:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'USB device {device_id} not found',
|
|
http_code=404,
|
|
)
|
|
if device.get('status') == _STATUS_CHECKED_OUT:
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
'Device is already checked out',
|
|
http_code=409,
|
|
)
|
|
|
|
name = _resolve_badge_name(cur, badge)
|
|
_ensure_user(cur, badge, name)
|
|
|
|
locker = data.get('locker_location')
|
|
log_locker = locker if locker is not None else device.get('locker_location')
|
|
|
|
cur.execute(
|
|
'INSERT INTO checkinoutlog '
|
|
'(badge_number, device_id, action, timestamp, locker_location) '
|
|
"VALUES (%s, %s, 'check-out', %s, %s)",
|
|
(badge, device_id, _now(), log_locker),
|
|
)
|
|
|
|
if locker is not None:
|
|
cur.execute(
|
|
'UPDATE devices SET status = %s, locker_location = %s '
|
|
'WHERE device_id = %s',
|
|
(_STATUS_CHECKED_OUT, locker, device_id),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
'UPDATE devices SET status = %s WHERE device_id = %s',
|
|
(_STATUS_CHECKED_OUT, device_id),
|
|
)
|
|
conn.commit()
|
|
|
|
with conn.cursor() as cur:
|
|
row = _fetch_device(cur, device_id)
|
|
result = _device_to_dict(cur, row)
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response(result, message='Device checked out', http_code=201)
|
|
|
|
|
|
@usb_bp.route('/<device_id>/checkin', methods=['POST'])
|
|
@jwt_required()
|
|
@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()
|
|
if not badge:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'badge is required')
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
device = _fetch_device(cur, device_id)
|
|
if not device:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'USB device {device_id} not found',
|
|
http_code=404,
|
|
)
|
|
if device.get('status') != _STATUS_CHECKED_OUT:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Device is not currently checked out',
|
|
http_code=400,
|
|
)
|
|
|
|
name = _resolve_badge_name(cur, badge)
|
|
_ensure_user(cur, badge, name)
|
|
|
|
locker = data.get('locker_location')
|
|
log_locker = locker if locker is not None else device.get('locker_location')
|
|
sanitized = _flag(data.get('sanitized'))
|
|
scanned_viruses = _flag(data.get('scanned_viruses'))
|
|
|
|
cur.execute(
|
|
'INSERT INTO checkinoutlog '
|
|
'(badge_number, device_id, action, timestamp, scanned_viruses, '
|
|
'locker_location, sanitized) '
|
|
"VALUES (%s, %s, 'check-in', %s, %s, %s, %s)",
|
|
(badge, device_id, _now(), scanned_viruses, log_locker, sanitized),
|
|
)
|
|
|
|
if locker is not None:
|
|
cur.execute(
|
|
'UPDATE devices SET status = %s, locker_location = %s '
|
|
'WHERE device_id = %s',
|
|
(_STATUS_CHECKED_IN, locker, device_id),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
'UPDATE devices SET status = %s WHERE device_id = %s',
|
|
(_STATUS_CHECKED_IN, device_id),
|
|
)
|
|
conn.commit()
|
|
|
|
with conn.cursor() as cur:
|
|
row = _fetch_device(cur, device_id)
|
|
result = _device_to_dict(cur, row)
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response(result, message='Device checked in')
|
|
|
|
|
|
# =============================================================================
|
|
# Check-in/out log queries
|
|
# =============================================================================
|
|
|
|
@usb_bp.route('/<device_id>/history', methods=['GET'])
|
|
@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()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'SELECT COUNT(*) AS total FROM checkinoutlog WHERE device_id = %s',
|
|
(device_id,),
|
|
)
|
|
total = cur.fetchone()['total']
|
|
|
|
cur.execute(
|
|
'SELECT ' + _LOG_COLS + ' FROM checkinoutlog WHERE device_id = %s '
|
|
'ORDER BY timestamp DESC, log_id DESC LIMIT %s OFFSET %s',
|
|
(device_id, per_page, (page - 1) * per_page),
|
|
)
|
|
rows = cur.fetchall()
|
|
data = [_log_to_dict(cur, row) for row in rows]
|
|
finally:
|
|
conn.close()
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@usb_bp.route('/checkouts', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_all_checkouts():
|
|
"""List check-out log rows.
|
|
|
|
Query parameters:
|
|
- 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'"]
|
|
params = []
|
|
|
|
if request.args.get('active', '').lower() == 'true':
|
|
where.append("d.status = 'checked-out'")
|
|
|
|
badge = request.args.get('badge', '').strip()
|
|
if badge:
|
|
where.append('l.badge_number = %s')
|
|
params.append(badge)
|
|
|
|
where_sql = ' WHERE ' + ' AND '.join(where)
|
|
base = (
|
|
'FROM checkinoutlog l JOIN devices d ON l.device_id = d.device_id'
|
|
+ where_sql
|
|
)
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT COUNT(*) AS total ' + base, params)
|
|
total = cur.fetchone()['total']
|
|
|
|
cur.execute(
|
|
'SELECT l.log_id, l.badge_number, l.device_id, l.action, '
|
|
'l.timestamp, l.scanned_viruses, l.locker_location, l.sanitized, '
|
|
'd.status AS device_status ' + base
|
|
+ ' ORDER BY l.timestamp DESC, l.log_id DESC LIMIT %s OFFSET %s',
|
|
params + [per_page, (page - 1) * per_page],
|
|
)
|
|
rows = cur.fetchall()
|
|
data = [_log_to_dict(cur, row) for row in rows]
|
|
finally:
|
|
conn.close()
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@usb_bp.route('/checkouts/active', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_active_checkouts():
|
|
"""List the latest check-out log row for every currently checked-out device.
|
|
|
|
Query parameters:
|
|
- badge: filter by badge_number
|
|
"""
|
|
if _usb_selfhosted():
|
|
return selfhosted.list_checkouts(request, active_only=True)
|
|
where_badge = ''
|
|
params = []
|
|
badge = request.args.get('badge', '').strip()
|
|
if badge:
|
|
where_badge = 'AND l.badge_number = %s '
|
|
params.append(badge)
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'SELECT l.log_id, l.badge_number, l.device_id, l.action, '
|
|
'l.timestamp, l.scanned_viruses, l.locker_location, l.sanitized, '
|
|
'd.status AS device_status '
|
|
'FROM devices d '
|
|
'JOIN checkinoutlog l ON l.device_id = d.device_id '
|
|
"AND l.action = 'check-out' "
|
|
"WHERE d.status = 'checked-out' "
|
|
+ where_badge +
|
|
'AND l.log_id = ('
|
|
' SELECT MAX(l2.log_id) FROM checkinoutlog l2 '
|
|
" WHERE l2.device_id = d.device_id AND l2.action = 'check-out'"
|
|
') '
|
|
'ORDER BY l.timestamp DESC, l.log_id DESC',
|
|
params,
|
|
)
|
|
rows = cur.fetchall()
|
|
data = [_log_to_dict(cur, row) for row in rows]
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response(data)
|
|
|
|
|
|
# =============================================================================
|
|
# Kiosk: UNAUTHENTICATED check-out / check-in
|
|
# =============================================================================
|
|
# The bar these have to meet, same as the parts kiosk: they can move a device
|
|
# between checked-out and checked-in and NOTHING else. No create, no edit, no
|
|
# retire, no browsing the fleet, no reading who holds what without presenting
|
|
# the device. Identity is the scanned badge, resolved server-side; the client
|
|
# never asserts who it is.
|
|
#
|
|
# Off unless a site turns it on. An unauthenticated write is a decision about
|
|
# the door, not about the person walking through it.
|
|
|
|
def _kiosk_setting(key, default=False):
|
|
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 _kiosk_enabled():
|
|
return _kiosk_setting('usb_kiosk_enabled', False)
|
|
|
|
|
|
def _kiosk_disabled_response():
|
|
# 404, not 403: when the kiosk is off these routes should not read as
|
|
# "there is a door here and you may not use it".
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Not found', http_code=404)
|
|
|
|
|
|
def _kiosk_badge(data):
|
|
"""The scanned badge, or an error response. Never trusts a client name."""
|
|
badge = (data.get('badge') or '').strip()
|
|
if not badge:
|
|
return None, error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'Scan your badge')
|
|
return badge, None
|
|
|
|
|
|
@usb_bp.route('/kiosk/device/<device_id>', methods=['GET'])
|
|
def kiosk_device(device_id):
|
|
"""What the kiosk shows after a device is scanned.
|
|
|
|
Deliberately narrow: the description, whether it is out, and the NAME of
|
|
whoever holds it - which the return-for-someone-else confirm has to show, or
|
|
the person standing there cannot tell whether they are doing the right
|
|
thing. No badge number, no history, no contact details.
|
|
"""
|
|
if not _kiosk_enabled():
|
|
return _kiosk_disabled_response()
|
|
if _usb_selfhosted():
|
|
return selfhosted.kiosk_device(device_id)
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
device = _fetch_device(cur, device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'No device matches that code',
|
|
http_code=404)
|
|
holder = ''
|
|
if device.get('status') == _STATUS_CHECKED_OUT:
|
|
holder = _kiosk_holder_name(cur, device_id)
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response({
|
|
'device_id': device.get('device_id'),
|
|
'device_desc': device.get('device_desc'),
|
|
'status': device.get('status'),
|
|
'locker_location': device.get('locker_location'),
|
|
'holdername': holder,
|
|
})
|
|
|
|
|
|
def _kiosk_holder_badge(cur, device_id):
|
|
"""Badge on the most recent check-out for this device, or ''."""
|
|
cur.execute(
|
|
'SELECT badge_number FROM checkinoutlog '
|
|
"WHERE device_id = %s AND action = 'check-out' "
|
|
'ORDER BY timestamp DESC, log_id DESC LIMIT 1',
|
|
(device_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return (row.get('badge_number') or '').strip() if row else ''
|
|
|
|
|
|
def _kiosk_holder_name(cur, device_id):
|
|
badge = _kiosk_holder_badge(cur, device_id)
|
|
return _resolve_badge_name(cur, badge) if badge else ''
|
|
|
|
|
|
@usb_bp.route('/kiosk/session', methods=['POST'])
|
|
def kiosk_session():
|
|
"""What one badge may do right now. Body: {badge}.
|
|
|
|
The drives here carry no barcode, so the kiosk cannot ask somebody to
|
|
present one - it has to offer a LIST. That is a real widening: before, you
|
|
had to already hold a drive to learn anything. So the list is not anonymous.
|
|
It costs a badge that resolves to a real person, which is the same bar
|
|
check-out itself has to clear, and it is a POST so the badge does not end up
|
|
in a proxy access log the way a query string would.
|
|
|
|
Returns the caller's own drives (to return), what is free (to take), and a
|
|
COUNT of everything else - names for those are a separate, deliberate step,
|
|
so a passer-by does not read the whole cabinet's holders off a wall screen.
|
|
"""
|
|
if not _kiosk_enabled():
|
|
return _kiosk_disabled_response()
|
|
data = request.get_json() or {}
|
|
if _usb_selfhosted():
|
|
return selfhosted.kiosk_session(data)
|
|
|
|
badge, failure = _kiosk_badge(data)
|
|
if failure:
|
|
return failure
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
name = _resolve_badge_name(cur, badge)
|
|
if not name:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Badge not recognized - see IT',
|
|
http_code=422)
|
|
|
|
cur.execute(
|
|
'SELECT ' + _DEVICE_COLS + ' FROM devices '
|
|
'WHERE status IN (%s, %s) ORDER BY device_id',
|
|
(_STATUS_CHECKED_IN, _STATUS_CHECKED_OUT),
|
|
)
|
|
rows = cur.fetchall() or []
|
|
|
|
yours, available, others = [], [], 0
|
|
for row in rows:
|
|
entry = {
|
|
'device_id': row.get('device_id'),
|
|
'device_desc': row.get('device_desc'),
|
|
'locker_location': row.get('locker_location'),
|
|
}
|
|
if row.get('status') == _STATUS_CHECKED_IN:
|
|
available.append(entry)
|
|
elif _kiosk_holder_badge(cur, row.get('device_id')) == badge:
|
|
yours.append(entry)
|
|
else:
|
|
others += 1
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response({'name': name, 'yours': yours,
|
|
'available': available, 'otherscount': others})
|
|
|
|
|
|
@usb_bp.route('/kiosk/others', methods=['POST'])
|
|
def kiosk_others():
|
|
"""Drives checked out to OTHER people, with holder names. Body: {badge}.
|
|
|
|
Behind its own tap, and only when returning for somebody else is allowed at
|
|
all. Somebody has to be able to put back a drive left on a bench by a person
|
|
who is on leave - but the names that makes visible should not be the first
|
|
thing on a shopfloor screen.
|
|
"""
|
|
if not _kiosk_enabled():
|
|
return _kiosk_disabled_response()
|
|
if not _kiosk_setting('usb_kiosk_allow_return_for_others', True):
|
|
return _kiosk_disabled_response()
|
|
data = request.get_json() or {}
|
|
if _usb_selfhosted():
|
|
return selfhosted.kiosk_others(data)
|
|
|
|
badge, failure = _kiosk_badge(data)
|
|
if failure:
|
|
return failure
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
if not _resolve_badge_name(cur, badge):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'Badge not recognized - see IT',
|
|
http_code=422)
|
|
cur.execute(
|
|
'SELECT ' + _DEVICE_COLS + ' FROM devices WHERE status = %s '
|
|
'ORDER BY device_id',
|
|
(_STATUS_CHECKED_OUT,),
|
|
)
|
|
out = []
|
|
for row in cur.fetchall() or []:
|
|
holderbadge = _kiosk_holder_badge(cur, row.get('device_id'))
|
|
if holderbadge == badge:
|
|
continue
|
|
out.append({
|
|
'device_id': row.get('device_id'),
|
|
'device_desc': row.get('device_desc'),
|
|
'locker_location': row.get('locker_location'),
|
|
'holdername': _resolve_badge_name(cur, holderbadge),
|
|
})
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response(out)
|
|
|
|
|
|
@usb_bp.route('/kiosk/checkout', methods=['POST'])
|
|
def kiosk_checkout():
|
|
"""Check a device out at the kiosk. Body: {device_id, badge}."""
|
|
if not _kiosk_enabled():
|
|
return _kiosk_disabled_response()
|
|
data = request.get_json() or {}
|
|
if _usb_selfhosted():
|
|
return selfhosted.kiosk_checkout(data)
|
|
|
|
device_id = (data.get('device_id') or '').strip()
|
|
badge, failure = _kiosk_badge(data)
|
|
if failure:
|
|
return failure
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
device = _fetch_device(cur, device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'No device matches that code',
|
|
http_code=404)
|
|
if device.get('status') == _STATUS_RETIRED:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'That device is retired - see IT')
|
|
if device.get('status') == _STATUS_CHECKED_OUT:
|
|
holder = _kiosk_holder_name(cur, device_id)
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
'That device is already checked out{}'.format(
|
|
' to {}'.format(holder) if holder else ''),
|
|
http_code=409)
|
|
|
|
# An unresolvable badge is REFUSED here, unlike the signed-in route
|
|
# which creates the user row. A checkout nobody can trace to a
|
|
# person is the one thing this tracking exists to prevent, and at a
|
|
# kiosk there is no signed-in operator to vouch for it.
|
|
name = _resolve_badge_name(cur, badge)
|
|
if not name:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Badge not recognized - see IT to check this device out',
|
|
http_code=422)
|
|
|
|
cur.execute(
|
|
'INSERT INTO checkinoutlog '
|
|
'(badge_number, device_id, action, timestamp, locker_location) '
|
|
"VALUES (%s, %s, 'check-out', %s, %s)",
|
|
(badge, device_id, _now(), device.get('locker_location')),
|
|
)
|
|
cur.execute(
|
|
'UPDATE devices SET status = %s WHERE device_id = %s',
|
|
(_STATUS_CHECKED_OUT, device_id),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response({'device_id': device_id, 'holdername': name},
|
|
message='{} checked out to {}'.format(
|
|
device_id, name))
|
|
|
|
|
|
@usb_bp.route('/kiosk/checkin', methods=['POST'])
|
|
def kiosk_checkin():
|
|
"""Check a device in at the kiosk. Body: {device_id, badge, forsomeoneelse}.
|
|
|
|
Holder by default. A mismatch is refused with the holder's name and a flag
|
|
the screen uses to offer "returning it for them", because the alternative -
|
|
a drive whose owner is on leave that literally cannot be returned at the
|
|
kiosk - just moves the problem to somebody with a login.
|
|
|
|
The badge RECORDED is the one that was scanned: the log says who actually
|
|
handed the device back, which is the honest record and needs no change to a
|
|
schema this app does not own.
|
|
"""
|
|
if not _kiosk_enabled():
|
|
return _kiosk_disabled_response()
|
|
data = request.get_json() or {}
|
|
if _usb_selfhosted():
|
|
return selfhosted.kiosk_checkin(data)
|
|
|
|
device_id = (data.get('device_id') or '').strip()
|
|
badge, failure = _kiosk_badge(data)
|
|
if failure:
|
|
return failure
|
|
forsomeoneelse = bool(data.get('forsomeoneelse'))
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
device = _fetch_device(cur, device_id)
|
|
if not device:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'No device matches that code',
|
|
http_code=404)
|
|
if device.get('status') != _STATUS_CHECKED_OUT:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'That device is not checked out')
|
|
|
|
name = _resolve_badge_name(cur, badge)
|
|
if not name:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Badge not recognized - see IT to return this device',
|
|
http_code=422)
|
|
|
|
holderbadge = _kiosk_holder_badge(cur, device_id)
|
|
if holderbadge and holderbadge != badge:
|
|
if not _kiosk_setting('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:
|
|
holdername = _resolve_badge_name(cur, holderbadge)
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
'That device is checked out to {}. Confirm you are '
|
|
'returning it for them.'.format(holdername or 'somebody else'),
|
|
details={'needsconfirm': True,
|
|
'holdername': holdername},
|
|
http_code=409)
|
|
|
|
cur.execute(
|
|
'INSERT INTO checkinoutlog '
|
|
'(badge_number, device_id, action, timestamp, locker_location) '
|
|
"VALUES (%s, %s, 'check-in', %s, %s)",
|
|
(badge, device_id, _now(), device.get('locker_location')),
|
|
)
|
|
cur.execute(
|
|
'UPDATE devices SET status = %s WHERE device_id = %s',
|
|
(_STATUS_CHECKED_IN, device_id),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
return success_response({'device_id': device_id},
|
|
message='{} checked in'.format(device_id))
|
|
|
|
|
|
# =============================================================================
|
|
# Reports
|
|
# =============================================================================
|
|
# CMMC removable-media tracking is an audit obligation, and until now the only
|
|
# way to answer "who had which drive, when" was to read the check-in/out log
|
|
# page by page. This is the same question over a window: a day for "what
|
|
# happened this shift", a month or six for a periodic review, a year or all time
|
|
# for an audit.
|
|
|
|
# The windows the UI offers, in days. 0 means everything ever recorded.
|
|
AUDITPERIODS = {'day': 1, 'month': 30, 'sixmonths': 182, 'year': 365, 'all': 0}
|
|
|
|
|
|
def _auditdays():
|
|
"""Days of history to report. Accepts ?period= or a raw ?days=."""
|
|
period = (request.args.get('period') or '').strip().lower()
|
|
if period in AUDITPERIODS:
|
|
return AUDITPERIODS[period]
|
|
days = request.args.get('days', type=int)
|
|
return max(0, days) if days is not None else 30
|
|
|
|
|
|
def _auditcsv(rows, columns, filename):
|
|
"""CSV download; generate_csv is not on the contract surface."""
|
|
import csv
|
|
import io
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
writer.writerow(columns)
|
|
for row in rows:
|
|
writer.writerow([row.get(column, '') for column in columns])
|
|
return Response(output.getvalue(), mimetype='text/csv',
|
|
headers={'Content-Disposition':
|
|
'attachment; filename={}'.format(filename)})
|
|
|
|
|
|
AUDITCOLUMNS = ['timestamp', 'action', 'device_id', 'device_desc', 'badge',
|
|
'name', 'locker_location']
|
|
|
|
|
|
@usb_bp.route('/reports/audit', methods=['GET'])
|
|
@jwt_required()
|
|
@require_permission('usb.view')
|
|
def report_audit():
|
|
"""Every check-out and check-in in the window, newest first.
|
|
|
|
Rows, not just totals: an audit asks who had a specific drive on a specific
|
|
day, and a count cannot answer that. Summary figures ride along for the
|
|
people who only want the shape of it.
|
|
"""
|
|
days = _auditdays()
|
|
if _usb_selfhosted():
|
|
rows = selfhosted.audit_rows(days)
|
|
return _auditresponse(rows, days)
|
|
|
|
cutoff = None
|
|
if days > 0:
|
|
cutoff = (datetime.utcnow() - timedelta(days=days)).strftime(
|
|
'%Y-%m-%d %H:%M:%S')
|
|
|
|
conn = cmmc_usb_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
sql = ('SELECT l.timestamp, l.action, l.device_id, l.badge_number, '
|
|
'l.locker_location, d.device_desc, '
|
|
'u.first_name, u.last_name '
|
|
'FROM checkinoutlog l '
|
|
'LEFT JOIN devices d ON d.device_id = l.device_id '
|
|
'LEFT JOIN users u ON u.badge_number = l.badge_number ')
|
|
params = []
|
|
if cutoff:
|
|
sql += 'WHERE l.timestamp >= %s '
|
|
params.append(cutoff)
|
|
# A total order: timestamp alone ties for two events in the same
|
|
# second and the same query then returns them in a different order
|
|
# each run, which reads as the log being unstable.
|
|
sql += 'ORDER BY l.timestamp DESC, l.log_id DESC'
|
|
cur.execute(sql, tuple(params))
|
|
rows = []
|
|
for row in cur.fetchall() or []:
|
|
name = '{} {}'.format(row.get('first_name') or '',
|
|
row.get('last_name') or '').strip()
|
|
stamp = row.get('timestamp')
|
|
rows.append({
|
|
'timestamp': stamp.isoformat() if hasattr(stamp, 'isoformat')
|
|
else (stamp or ''),
|
|
'action': row.get('action') or '',
|
|
'device_id': row.get('device_id') or '',
|
|
'device_desc': row.get('device_desc') or '',
|
|
'badge': row.get('badge_number') or '',
|
|
'name': name,
|
|
'locker_location': row.get('locker_location') or '',
|
|
})
|
|
finally:
|
|
conn.close()
|
|
|
|
return _auditresponse(rows, days)
|
|
|
|
|
|
def _auditresponse(rows, days):
|
|
"""One voice for both directory modes: same CSV, same JSON, same summary."""
|
|
if request.args.get('format') == 'csv':
|
|
return _auditcsv(rows, AUDITCOLUMNS, 'usb-audit.csv')
|
|
return success_response({
|
|
'columns': AUDITCOLUMNS,
|
|
'rows': rows,
|
|
'days': days,
|
|
'summary': _auditsummary(rows),
|
|
})
|
|
|
|
|
|
def _auditsummary(rows):
|
|
checkouts = sum(1 for row in rows if row['action'] == 'check-out')
|
|
return {
|
|
'events': len(rows),
|
|
'checkouts': checkouts,
|
|
'checkins': sum(1 for row in rows if row['action'] == 'check-in'),
|
|
'devices': len({row['device_id'] for row in rows if row['device_id']}),
|
|
'people': len({row['badge'] for row in rows if row['badge']}),
|
|
}
|