ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Import USB records off an external cmmc_usb database, so a site can retire it

Browse Source 
A site running external mode keeps its devices and check-in/out history in a
separate cmmc_usb database. There was no way to bring that in, so the choice
was to keep the second database forever or lose the history. This adds
'flask usb import-external' - dry run by default, --commit to write - which
reads through the same connection settings the routes already use and never
writes to the source.

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

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

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

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

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

Full suite 2131 passed; seven new tests pin the pairing rules.
This commit is contained in:
cproudlock
2026-08-29 13:02:21 -04:00
parent 99c40ae7c3
commit 60c35728e8
7 changed files with 404 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
plugins/usb/api/selfhosted.py
View File

@@ -87,7 +87,10 @@ def _checkout_events(checkout):
'device_id': checkout.device.serialnumber if checkout.device else None,
'action': 'check-in',
'timestamp': checkout.checkintime.isoformat() + 'Z',
'scanned_viruses': None,
# Both read from the checkout row now. scanned_viruses was
# hardcoded None: the shape promised the field and had nowhere to
# read it from, so an imported history would have shown blank.
'scanned_viruses': checkout.scannedviruses,
'locker_location': None,
'sanitized': checkout.waswiped,
})
@@ -233,6 +236,10 @@ def checkin_device(device_id, data):
checkintime = override
open_checkout.checkintime = checkintime
open_checkout.waswiped = bool(data.get('sanitized'))
# None, not False, when the caller said nothing: an unrecorded scan is
# not a failed scan, and an audit reads those very differently.
scanned = data.get('scanned_viruses')
open_checkout.scannedviruses = None if scanned is None else bool(scanned)
open_checkout.checkinnotes = data.get('notes')
device.ischeckedout = False
device.currentuserid = None

235
plugins/usb/cli.py Normal file
View File

@@ -0,0 +1,235 @@
"""CLI: move USB records off an external cmmc_usb database and into ShopDB.
flask usb import-external report what would happen, change nothing
flask usb import-external --commit write it
A site running the external mode keeps its devices and its check-in/out history
in a separate cmmc_usb database. This reads that database through the same
connection settings the routes already use, converts it to the self-hosted
tables, and leaves the source untouched. After a clean run, set
usb_directory_mode to selfhosted and the external database can be retired.
TWO THINGS THIS HAS TO RECONCILE, and they are the reason it is a command
rather than a few INSERTs:
The two schemas disagree about shape. External is an EVENT LOG - one row per
action, check-out or check-in - while self-hosted keeps a PAIRED ROW, one
checkout carrying both its times. Pairing is per device and in time order, and
real logs are not tidy: a check-out with no return is an open checkout, but a
return with no matching check-out, or two check-outs in a row, mean something
was recorded wrong years ago. Those are reported rather than guessed at, and
they never silently vanish.
Identity is the other one. usbdevices.serialnumber is unique and NOT NULL, and
the external devices table has no serial column - only device_id. So device_id
becomes the serial. A blank or duplicated device_id stops the import for that
device instead of inventing a key.
Dry run by default. The point of the first run is the report.
"""
from collections import defaultdict
import click
from flask.cli import AppGroup, with_appcontext
from shopdb.api import db, cmmc_usb_connection
from .models import USBDevice, USBCheckout
usb_cli = AppGroup('usb', help='USB device records.')
# The external status vocabulary, mapped to the self-hosted boolean. Anything
# unrecognised is treated as checked in and reported, rather than assumed.
_CHECKED_OUT = {'checked-out', 'checked out', 'out', 'issued'}
_RETIRED = {'retired', 'disposed', 'destroyed'}
def _external_rows():
"""(devices, log) from the external database, oldest log entry first."""
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
cur.execute('SELECT device_id, device_desc, device_owner, status, '
'locker_location FROM devices')
devices = cur.fetchall()
cur.execute('SELECT log_id, badge_number, device_id, action, timestamp, '
'scanned_viruses, locker_location, sanitized '
'FROM checkinoutlog ORDER BY device_id, timestamp, log_id')
log = cur.fetchall()
finally:
conn.close()
return devices, log
def _is_checkout(action):
return str(action or '').strip().lower().replace('_', '-') in (
'check-out', 'checkout', 'out')
def _is_checkin(action):
return str(action or '').strip().lower().replace('_', '-') in (
'check-in', 'checkin', 'in', 'return')
def _pair(entries, problems, deviceid):
"""Walk one device's log in time order and fold it into checkout pairs."""
pairs, open_row = [], None
for row in entries:
action = row.get('action')
if _is_checkout(action):
if open_row is not None:
# Two check-outs with no return between them. The first is
# unclosed and stays that way; say so rather than inventing a
# return time that nobody recorded.
problems.append(
'device %s: check-out at %s while %s was still out - first '
'left open' % (deviceid, row.get('timestamp'),
open_row.get('badge_number')))
pairs.append((open_row, None))
open_row = row
elif _is_checkin(action):
if open_row is None:
problems.append(
'device %s: check-in at %s with no matching check-out - '
'skipped' % (deviceid, row.get('timestamp')))
continue
pairs.append((open_row, row))
open_row = None
else:
problems.append('device %s: unrecognised action %r at %s - skipped'
% (deviceid, action, row.get('timestamp')))
if open_row is not None:
pairs.append((open_row, None))
return pairs
@usb_cli.command('import-external')
@click.option('--commit', is_flag=True, default=False,
help='Write the records. Without this the command only reports.')
@with_appcontext
def import_external(commit):
"""Copy devices and check-in/out history from the external cmmc_usb DB."""
try:
devices, log = _external_rows()
except Exception as error:
raise click.ClickException(
'Could not read the external database: %s\n'
'Check the cmmc_usb_db_* settings, and that this host can reach it.'
% error)
problems = []
seen_serials = set()
made_devices = 0
made_checkouts = 0
open_checkouts = 0
by_device = defaultdict(list)
for row in log:
by_device[row.get('device_id')].append(row)
for source in devices:
serial = (source.get('device_id') or '').strip()
if not serial:
problems.append('device with no device_id - skipped, it cannot be '
'given a serial number')
continue
if serial in seen_serials:
problems.append('device_id %r appears twice - only the first is '
'imported' % serial)
continue
seen_serials.add(serial)
status = str(source.get('status') or '').strip().lower()
if status and status not in _CHECKED_OUT and status not in _RETIRED \
and status not in ('checked-in', 'checked in', 'in', 'available'):
problems.append('device %s: unrecognised status %r - treated as '
'checked in' % (serial, source.get('status')))
device = USBDevice.query.filter_by(serialnumber=serial).first()
if device is None:
device = USBDevice(serialnumber=serial)
db.session.add(device)
made_devices += 1
device.label = source.get('device_desc') or device.label
device.storagelocation = source.get('locker_location') or device.storagelocation
device.notes = _append_owner(device.notes, source.get('device_owner'))
device.ischeckedout = status in _CHECKED_OUT
db.session.flush()
for out_row, in_row in _pair(by_device.get(source.get('device_id'), []),
problems, serial):
checkout = USBCheckout(
usbdeviceid=device.usbdeviceid,
sso=str(out_row.get('badge_number') or '')[:20],
checkouttime=out_row.get('timestamp'),
checkintime=in_row.get('timestamp') if in_row else None,
)
if in_row:
checkout.waswiped = _tri(in_row.get('sanitized'))
checkout.scannedviruses = _tri(in_row.get('scanned_viruses'))
else:
open_checkouts += 1
db.session.add(checkout)
made_checkouts += 1
# The device's own flag and the history have to agree, or the kiosk
# will offer a drive that the log says somebody still holds.
if device.ischeckedout and not _has_open(device):
problems.append('device %s: marked checked out, but its log has no '
'open check-out' % serial)
click.echo('devices read from external : %d' % len(devices))
click.echo('log rows read : %d' % len(log))
click.echo('devices to create : %d' % made_devices)
click.echo('checkouts to create : %d' % made_checkouts)
click.echo(' of which still open : %d' % open_checkouts)
if problems:
click.echo('')
click.echo('%d row(s) need a human decision:' % len(problems))
for line in problems:
click.echo(' - %s' % line)
if not commit:
db.session.rollback()
click.echo('')
click.echo('DRY RUN - nothing was written. Re-run with --commit when the '
'report above looks right.')
return
db.session.commit()
click.echo('')
click.echo('Written. Next: set usb_directory_mode to selfhosted in Settings, '
'confirm the kiosk and the device list against the external '
'database, and only then retire it.')
def _tri(value):
"""External flags are 0/1/NULL. Keep NULL as None: unrecorded is not false."""
if value is None:
return None
return bool(value)
def _has_open(device):
return USBCheckout.query.filter_by(usbdeviceid=device.usbdeviceid,
checkintime=None).count() > 0
def _append_owner(notes, owner):
"""External device_owner has no column here, so keep it rather than lose it."""
owner = (owner or '').strip()
if not owner:
return notes
line = 'Imported owner: %s' % owner
if notes and line in notes:
return notes
return (notes + '\n' + line) if notes else line
def build_cli():
return [usb_cli]

49
plugins/usb/migrations/versions/0003_checkout_scan_fields.py Normal file
View File

@@ -0,0 +1,49 @@
"""Record whether a returned drive was virus-scanned.
The external cmmc_usb checkinoutlog carries scanned_viruses and sanitized on
every row. The self-hosted usbcheckouts table already has waswiped, which is
the same fact as sanitized under a different name, but nothing corresponds to
scanned_viruses. That gap was harmless while the two backends were
alternatives, and becomes data loss the moment a site imports its history and
cuts over: whether a returned drive was scanned is precisely what a
removable-media audit asks.
Nullable on purpose, and NOT defaulted to false. A checkout recorded before
this column existed is not a checkout where the drive went unscanned; it is one
where nobody wrote it down. False would assert something the data does not say.
Idempotent, so a re-run on an already-migrated database is a no-op.
Revision ID: usb0003scanfields
Revises: usb0002dropmachineid
"""
from alembic import op
import sqlalchemy as sa
revision = 'usb0003scanfields'
down_revision = 'usb0002dropmachineid'
branch_labels = None
depends_on = None
def upgrade():
bind = op.get_bind()
insp = sa.inspect(bind)
if 'usbcheckouts' not in insp.get_table_names():
return
cols = {c['name'] for c in insp.get_columns('usbcheckouts')}
if 'scannedviruses' not in cols:
op.add_column('usbcheckouts',
sa.Column('scannedviruses', sa.Boolean(), nullable=True))
def downgrade():
bind = op.get_bind()
insp = sa.inspect(bind)
if 'usbcheckouts' not in insp.get_table_names():
return
cols = {c['name'] for c in insp.get_columns('usbcheckouts')}
if 'scannedviruses' in cols:
op.drop_column('usbcheckouts', 'scannedviruses')

4
plugins/usb/models/usb_device.py
View File

@@ -133,6 +133,10 @@ class USBCheckout(BaseModel):
checkoutreason = db.Column(db.Text, nullable=True, comment='Reason for checkout')
checkinnotes = db.Column(db.Text, nullable=True)
waswiped = db.Column(db.Boolean, nullable=True, comment='Was device wiped after return')
# Nullable, and never defaulted to False: a checkout with no recorded scan
# is one where nobody wrote it down, not one where the drive went unscanned.
scannedviruses = db.Column(db.Boolean, nullable=True,
comment='Was the device virus-scanned on return')
# Relationships
device = db.relationship('USBDevice', backref=db.backref('checkouts', lazy='dynamic'))

5
plugins/usb/plugin.py
View File

@@ -70,6 +70,11 @@ class USBPlugin(BasePlugin):
'docs': 'plugins/usb/README.md',
}
def get_cli_commands(self) -> List:
"""CLI: import records off an external cmmc_usb database."""
from .cli import build_cli
return build_cli()
def get_reports(self) -> List[Dict]:
"""Removable-media audit.

6
tests/test_plugin_migrations.py
View File

@@ -70,8 +70,10 @@ EXPECTED_HEAD_REVISION['printers'] = 'printers0005observedqueues'
EXPECTED_HEAD_REVISION['machines'] = 'machines0002rename'
# employees adds the photofilename column on top of its cutover anchor.
EXPECTED_HEAD_REVISION['employees'] = 'employees0002photo'
# usb drops the dead usbcheckouts.machineid column on top of its anchor.
EXPECTED_HEAD_REVISION['usb'] = 'usb0002dropmachineid'
# usb drops the dead usbcheckouts.machineid column on top of its anchor, then
# adds scannedviruses so an imported external history keeps the scan result -
# the field a removable-media audit actually asks about.
EXPECTED_HEAD_REVISION['usb'] = 'usb0003scanfields'
# network links a device to a catalog model, which is where its photo comes
# from - machines, PCs and printers already had that link.
EXPECTED_HEAD_REVISION['network'] = 'network0003prefix'

99
tests/test_plugins/test_usb_import_pairing.py Normal file
View File

@@ -0,0 +1,99 @@
"""Folding an external check-in/out LOG into self-hosted checkout PAIRS.
The external cmmc_usb database records one row per action; the self-hosted
usbcheckouts table records one row per checkout, carrying both times. The
import has to convert between them, and the interesting cases are all the ways
a real log is untidy after years of use: a drive still out, a return nobody
has a matching issue for, two issues in a row.
The rule these tests pin down is that NOTHING is invented and NOTHING vanishes
silently. An unclosed checkout stays unclosed; anything that cannot be paired
is reported. A quieter import that guessed the missing halves would produce a
tidier database and a false audit history, which for removable-media records is
the worst of the available outcomes.
"""
from plugins.usb.cli import _pair, _tri
def _row(action, when, badge='B1'):
return {'action': action, 'timestamp': when, 'badge_number': badge}
def test_a_clean_out_and_in_becomes_one_closed_checkout():
problems = []
pairs = _pair([_row('check-out', 1), _row('check-in', 2)], problems, 'DEV1')
assert len(pairs) == 1
out, back = pairs[0]
assert out['timestamp'] == 1
assert back['timestamp'] == 2
assert problems == []
def test_a_drive_still_out_stays_open_and_is_not_reported_as_a_problem():
# This is the normal state of any drive currently issued to somebody, not
# a data fault, so it must not appear in the problem list.
problems = []
pairs = _pair([_row('check-out', 1)], problems, 'DEV1')
assert len(pairs) == 1
out, back = pairs[0]
assert back is None
assert problems == []
def test_a_return_with_no_matching_issue_is_skipped_and_reported():
problems = []
pairs = _pair([_row('check-in', 5)], problems, 'DEV1')
assert pairs == []
assert len(problems) == 1
assert 'no matching check-out' in problems[0]
assert 'DEV1' in problems[0]
def test_two_issues_in_a_row_leave_the_first_open_rather_than_inventing_a_return():
problems = []
pairs = _pair([_row('check-out', 1, 'ALICE'),
_row('check-out', 2, 'BOB'),
_row('check-in', 3, 'BOB')], problems, 'DEV1')
assert len(pairs) == 2
first, second = pairs
assert first[0]['badge_number'] == 'ALICE'
assert first[1] is None, 'the first checkout must not be given a return time'
assert second[0]['badge_number'] == 'BOB'
assert second[1]['timestamp'] == 3
assert len(problems) == 1
assert 'still out' in problems[0]
def test_an_unrecognised_action_is_skipped_and_reported():
problems = []
pairs = _pair([_row('destroyed', 1)], problems, 'DEV1')
assert pairs == []
assert len(problems) == 1
assert 'unrecognised action' in problems[0]
def test_action_spelling_variants_are_accepted():
# Years of a hand-maintained log produce more than one spelling.
for out_word, in_word in (('check-out', 'check-in'),
('checkout', 'checkin'),
('out', 'in'),
('CHECK_OUT', 'RETURN')):
problems = []
pairs = _pair([_row(out_word, 1), _row(in_word, 2)], problems, 'DEV1')
assert len(pairs) == 1, out_word
assert pairs[0][1] is not None, out_word
assert problems == [], out_word
def test_unrecorded_flags_stay_unknown_rather_than_becoming_false():
# NULL in the source means nobody wrote it down. False would assert that
# the drive was checked and failed, which is a different claim entirely.
assert _tri(None) is None
assert _tri(0) is False
assert _tri(1) is True