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.
172 lines
5.7 KiB
Python
172 lines
5.7 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)
|
|
|
|
# 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
|