Files
shopdb-flask/plugins/usb/models/usb_device.py
cproudlock aa8028f96e Let the row refuse the write both kiosks were deciding in Python
Wave 1 of the remediation plan for the 2026-09-03 feature test run. Two kiosks,
one shape of defect, one critical consequence.

Both kiosks read the state, decided in Python, and then wrote. Nothing stops two
requests interleaving between the read and the write, so both see the same
pre-state and both proceed. Reproduced on MySQL 5.6 over HTTP, five concurrent
takes of 5 from a bin holding 5: all five were accepted and quantityonhand
landed on -20, because each request wrote its own arithmetic. The ledger had
recorded every one of them, so the cached quantity and the ledger disagreed
permanently, which is the number the shop floor actually reads. The USB kiosk
had the same shape: concurrent scans of one drive all succeeded and inserted
several open custody rows, so one drive was recorded out to two people and the
documented 409 never fired.

The comparison now lives in the WHERE clause, where the database settles it at
write time. printedparts._move_quantity issues
UPDATE ... WHERE quantityonhand >= :n and treats rowcount 0 as the refusal,
raising InsufficientStock, which take and adjust turn back into the message they
already gave - carrying the quantity the database holds now, not the stale one
the request read. The low-stock alert derives its before-value from what the
write produced for the same reason. Both USB paths claim the device with
ischeckedout (self-hosted) or status (external cmmc_usb) in the WHERE clause;
the external one claims BEFORE writing the log line, so a refused claim leaves
nothing to undo. Check-in is guarded the same way.

After the fix, the same five concurrent takes: one accepted, four refused,
quantityonhand 0 and the ledger sum 0.

Check-in also closes EVERY open row for the device rather than the newest.
One device can only be out once, so a second open row is corruption, and
closing only the newest left the older ones open forever - the device then read
as returned while its custody history said otherwise.

Separately, usbcheckouts now records who handed a drive back
(checkinsso/checkinname, migration usb0004checkinidentity). Returning for a
colleague is a supported flow that the kiosk asks to confirm, but the row it
wrote attributed the check-in to the HOLDER, so the custody history and the
audit report both named somebody who was not there. Nullable and deliberately
not backfilled: copying the holder's badge into old rows would manufacture
evidence for the exact claim this stops making.

Four tests. Both guards verified by planting: strip the condition out of either
UPDATE and its test fails.

Scope: product
2026-09-03 13:58:26 -04:00

179 lines
6.1 KiB
Python

"""USB device plugin models."""
from datetime import datetime, timezone
from shopdb.api import db, BaseModel, AuditMixin
def _utcnow():
# naive UTC for DB columns (stored without tzinfo)
return datetime.now(timezone.utc).replace(tzinfo=None)
class USBDeviceType(BaseModel):
"""
USB device type classification.
Examples: Flash Drive, External HDD, External SSD, Card Reader
"""
__tablename__ = 'usbdevicetypes'
usbdevicetypeid = db.Column(db.Integer, primary_key=True)
typename = db.Column(db.String(50), unique=True, nullable=False)
description = db.Column(db.Text)
icon = db.Column(db.String(50), default='usb', comment='Icon name for UI')
def __repr__(self):
return f"<USBDeviceType {self.typename}>"
class USBDevice(BaseModel, AuditMixin):
"""
USB device model.
Tracks USB storage devices that can be checked out by users.
"""
__tablename__ = 'usbdevices'
usbdeviceid = db.Column(db.Integer, primary_key=True)
# Identification
serialnumber = db.Column(db.String(100), unique=True, nullable=False)
label = db.Column(db.String(100), nullable=True, comment='Human-readable label')
assetnumber = db.Column(db.String(50), nullable=True, comment='Optional asset tag')
# Classification
usbdevicetypeid = db.Column(
db.Integer,
db.ForeignKey('usbdevicetypes.usbdevicetypeid'),
nullable=True
)
# Specifications
capacitygb = db.Column(db.Integer, nullable=True, comment='Capacity in GB')
vendorid = db.Column(db.String(10), nullable=True, comment='USB Vendor ID (hex)')
productid = db.Column(db.String(10), nullable=True, comment='USB Product ID (hex)')
manufacturer = db.Column(db.String(100), nullable=True)
productname = db.Column(db.String(100), nullable=True)
# Current status
ischeckedout = db.Column(db.Boolean, default=False)
currentuserid = db.Column(db.String(50), nullable=True, comment='SSO of current user')
currentusername = db.Column(db.String(100), nullable=True, comment='Name of current user')
currentcheckoutdate = db.Column(db.DateTime, nullable=True)
# Location
storagelocation = db.Column(db.String(200), nullable=True, comment='Where device is stored when not checked out')
# Security
pin = db.Column(db.String(50), nullable=True, comment='PIN for encrypted devices')
# Notes
notes = db.Column(db.Text, nullable=True)
# Relationships
devicetype = db.relationship('USBDeviceType', backref='devices')
# Indexes
__table_args__ = (
db.Index('idx_usb_checkedout', 'ischeckedout'),
db.Index('idx_usb_type', 'usbdevicetypeid'),
db.Index('idx_usb_currentuser', 'currentuserid'),
)
def __repr__(self):
return f"<USBDevice {self.label or self.serialnumber}>"
@property
def display_name(self):
"""Get display name (label if set, otherwise serial number)."""
return self.label or self.serialnumber
def to_dict(self):
"""Convert to dictionary with related data."""
result = super().to_dict()
# Add type info
if self.devicetype:
result['typename'] = self.devicetype.typename
result['typeicon'] = self.devicetype.icon
# Add computed property
result['displayname'] = self.display_name
return result
class USBCheckout(BaseModel):
"""
USB device checkout history.
Tracks when devices are checked out and returned.
Maps to existing usbcheckouts table from classic ShopDB.
"""
__tablename__ = 'usbcheckouts'
checkoutid = db.Column(db.Integer, primary_key=True)
# Device reference (new column linking to usbdevices table)
usbdeviceid = db.Column(
db.Integer,
db.ForeignKey('usbdevices.usbdeviceid', ondelete='CASCADE'),
nullable=True
)
# User info
sso = db.Column(db.String(20), nullable=False, comment='SSO of user')
checkoutname = db.Column(db.String(100), nullable=True, comment='Name of user')
# Checkout details
checkouttime = db.Column(db.DateTime, nullable=False, default=_utcnow)
checkintime = db.Column(db.DateTime, nullable=True)
# WHO handed it back. Not always the holder: returning for a colleague is a
# supported flow, and recording the check-in against the holder named
# somebody who was not there. See migration usb0004checkinidentity.
checkinsso = db.Column(db.String(20), nullable=True,
comment='SSO of whoever returned it')
checkinname = db.Column(db.String(100), nullable=True,
comment='Name of whoever returned it')
# Metadata
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'))
def __repr__(self):
return f"<USBCheckout device={self.usbdeviceid} user={self.sso}>"
@property
def is_active(self):
"""Check if this checkout is currently active (not returned)."""
return self.checkintime is None
@property
def duration_days(self):
"""Get duration of checkout in days."""
end = self.checkintime or datetime.now(timezone.utc).replace(tzinfo=None)
delta = end - self.checkouttime
return delta.days
def to_dict(self):
"""Convert to dictionary with computed fields."""
result = super().to_dict()
result['isactivecheckout'] = self.is_active
result['durationdays'] = self.duration_days
# Add device info if loaded
if self.device:
result['devicelabel'] = self.device.label
result['deviceserialnumber'] = self.device.serialnumber
return result