da91a29155
The device token assumes a handful of fixed screens an administrator provisions once. That is not how these kiosks are used: most people open the page on their own PC, and there is no login, so per-browser provisioning would mean touching every PC on the floor and losing it again on the next reimage or cleared profile. A screen in that state did not look refused either - the USB kiosk rendered a working page whose every call answered 404. Both kiosks now admit a caller on either ground, the same shape the GE-Enforce client endpoints already use: the per-device token, or a source address inside an allowlist. Wall screens keep the token and are not confined to the allowlist; everyone else is admitted by where they are on the network. Still gated on the site's enable flag, so nothing opens by itself, and both allowlists ship empty - token-only until a site enters its ranges. - shopdb.api gains ip_in_cidrs and request_ip_allowlisted (contract 0.35.0). On the surface because both plugins need the identical gate, and a CIDR parser copied into two of them is where a fail-open creeps in later. They fail closed on an empty setting, an unparseable caller address and a malformed entry, and they read remote_addr rather than raw X-Forwarded-For - so the IIS rewrite rule is what makes them meaningful, which the module says at length. - Separate settings per kiosk, usb_kiosk_allowed_cidrs and printedparts_kiosk_allowed_cidrs. One must not grant the other, for the same reason the tokens are separate. - Both settings pages get a picker listing the network plugin's subnets with their VLANs. It stores resolved CIDR strings, not subnet ids: the setting has to work at a site that does not run that plugin and must not break when an inventory row is deleted. Entries matching no subnet are shown rather than hidden, since a stale one left by an inventory edit looks exactly like a hand-typed range. - USBKiosk says it was not admitted instead of pretending to work, which PartsKiosk already did. - The USB settings page keyed text-vs-boolean off the token alone. Adding a second text setting to the boolean path would have saved it as the word 'true' - the bug that replaced the kiosk token and 404'd every provisioned screen. Both load and save now key off a list, with a spec on it. The trade, stated plainly in the admin UI: anyone on an allowed network can check a drive in or out against a badge number they guess, and the record carries that person's name. Badges are attribution, not authorization. Keeping the ranges narrow is what bounds it; login is the alternative and was ruled out because these users do not sign in.
1260 lines
45 KiB
Python
1260 lines
45 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
|
|
|
|
import hmac
|
|
|
|
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, request_ip_allowlisted
|
|
|
|
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'])
|
|
@require_permission('usb.view')
|
|
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'])
|
|
@require_permission('usb.view')
|
|
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'])
|
|
@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'])
|
|
@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'])
|
|
@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'])
|
|
@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'])
|
|
@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'])
|
|
@require_permission('usb.view')
|
|
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'])
|
|
@require_permission('usb.view')
|
|
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'])
|
|
@require_permission('usb.view')
|
|
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_token_presented():
|
|
"""True when the caller presents the configured device token.
|
|
|
|
The token is per-DEVICE. A kiosk is a browser on a screen by the cabinet, so
|
|
a token served inside the page would be readable by anyone who could load
|
|
it. An admin opens the screen once as /usb-kiosk?token=..., the page keeps
|
|
it in that browser and sends it as X-Kiosk-Token.
|
|
"""
|
|
expected = (Setting.get('usb_kiosk_token') or '').strip()
|
|
presented = (request.headers.get('X-Kiosk-Token') or '').strip()
|
|
if not expected or not presented:
|
|
return False
|
|
# compare_digest, not ==: == leaks the token's length and matching prefix
|
|
# through timing.
|
|
return hmac.compare_digest(presented, expected)
|
|
|
|
|
|
def _kiosk_ip_allowlisted():
|
|
"""True when the caller sits in the configured kiosk network allowlist.
|
|
|
|
Why this exists alongside the token: the token model assumes a handful of
|
|
fixed screens provisioned once. Most people reach this kiosk from their own
|
|
PC instead, and there is no login - so per-browser provisioning would mean
|
|
provisioning every PC on the floor, and losing it on every reimage or
|
|
cleared profile. Network position is what those callers actually have in
|
|
common.
|
|
|
|
Fail-closed behaviour and the IIS rewrite-rule dependency that makes the
|
|
caller address trustworthy are documented on request_ip_allowlisted.
|
|
"""
|
|
return request_ip_allowlisted(
|
|
request, Setting.get('usb_kiosk_allowed_cidrs'))
|
|
|
|
|
|
def _kiosk_open():
|
|
"""True when the kiosk is on for this site AND the caller is admitted.
|
|
|
|
The flag alone was the whole gate, and a flag is a decision, not a
|
|
credential: with the kiosk on, anyone who could reach the app could move a
|
|
device between checked-out and checked-in and have it recorded against a
|
|
badge they guessed. A badge is a short numeric SSO, so it is attribution,
|
|
never authorization - the same reasoning that closed the parts kiosk.
|
|
|
|
Two ways in, either sufficient, mirroring what the GE-Enforce client
|
|
endpoints already do:
|
|
1. the per-device token, for unattended screens by the cabinet, and
|
|
2. a source address inside usb_kiosk_allowed_cidrs, for the people who
|
|
open this page on their own PC and never log in.
|
|
|
|
Both are still gated on usb_kiosk_enabled, so the door stays shut until a
|
|
site opens it deliberately.
|
|
"""
|
|
if not _kiosk_enabled():
|
|
return False
|
|
return _kiosk_token_presented() or _kiosk_ip_allowlisted()
|
|
|
|
|
|
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_open():
|
|
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_open():
|
|
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_open():
|
|
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_open():
|
|
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)
|
|
|
|
# Claim the device FIRST, and let the status transition be the
|
|
# guard: the read-then-write above cannot stop two scans of the
|
|
# same drive that interleave, because both read 'available'. With
|
|
# the status in the WHERE clause the second UPDATE matches no row
|
|
# and there is no log entry to undo.
|
|
cur.execute(
|
|
'UPDATE devices SET status = %s '
|
|
'WHERE device_id = %s AND status <> %s',
|
|
(_STATUS_CHECKED_OUT, device_id, _STATUS_CHECKED_OUT),
|
|
)
|
|
if not cur.rowcount:
|
|
holder = _kiosk_holder_name(cur, device_id)
|
|
conn.rollback()
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
'That device is already checked out{}'.format(
|
|
' to {}'.format(holder) if holder else ''),
|
|
http_code=409)
|
|
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')),
|
|
)
|
|
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_open():
|
|
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'])
|
|
@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']}),
|
|
}
|