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.
140 lines
5.5 KiB
Python
140 lines
5.5 KiB
Python
"""USB check-in / check-out audit report.
|
|
|
|
CMMC removable-media tracking is an audit obligation, and the question an audit
|
|
asks is "who had this drive, when" - which a total cannot answer. So this
|
|
reports EVENTS over a window, with the window named rather than expressed as a
|
|
day count nobody should have to work out.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from shopdb.core.models import Setting
|
|
|
|
|
|
def _now():
|
|
return datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
|
|
@pytest.fixture
|
|
def usbdata(db):
|
|
"""Self-hosted USB with a loan returned today and one from long ago."""
|
|
from plugins.usb.models import USBDevice, USBCheckout
|
|
|
|
db.session.add(Setting(key='usb_directory_mode', value='selfhosted',
|
|
valuetype='string', category='site'))
|
|
recent = USBDevice(serialnumber='USB-1', label='Blue Kingston')
|
|
old = USBDevice(serialnumber='USB-2', label='Red SanDisk')
|
|
db.session.add_all([recent, old])
|
|
db.session.flush()
|
|
|
|
# Taken and returned today: two events.
|
|
db.session.add(USBCheckout(usbdeviceid=recent.usbdeviceid, sso='270015376',
|
|
checkoutname='Dana Reyes',
|
|
checkouttime=_now() - timedelta(hours=3),
|
|
checkintime=_now() - timedelta(hours=1)))
|
|
# Taken 200 days ago and never returned: one event, outside most windows.
|
|
db.session.add(USBCheckout(usbdeviceid=old.usbdeviceid, sso='270099999',
|
|
checkoutname='Sam Okafor',
|
|
checkouttime=_now() - timedelta(days=200)))
|
|
db.session.commit()
|
|
return db
|
|
|
|
|
|
def _report(client, headers, **params):
|
|
query = '&'.join(f'{k}={v}' for k, v in params.items())
|
|
response = client.get(f'/api/usb/reports/audit?{query}', headers=headers)
|
|
assert response.status_code == 200, response.get_json()
|
|
return response.get_json()['data']
|
|
|
|
|
|
def test_a_loan_produces_both_a_take_and_a_return(client, usbdata, auth_headers):
|
|
"""One stored row holds both ends of a loan.
|
|
|
|
Reporting it once would make a returned drive look identical to one still
|
|
out, which is the opposite of what the log is for.
|
|
"""
|
|
data = _report(client, auth_headers, period='month')
|
|
actions = sorted(row['action'] for row in data['rows'])
|
|
|
|
assert actions == ['check-in', 'check-out']
|
|
assert data['summary']['checkouts'] == 1
|
|
assert data['summary']['checkins'] == 1
|
|
|
|
|
|
def test_the_window_actually_bounds_the_rows(client, usbdata, auth_headers):
|
|
month = _report(client, auth_headers, period='month')
|
|
year = _report(client, auth_headers, period='year')
|
|
|
|
assert {row['device_id'] for row in month['rows']} == {'USB-1'}
|
|
assert {row['device_id'] for row in year['rows']} == {'USB-1', 'USB-2'}
|
|
|
|
|
|
def test_all_time_keeps_everything(client, usbdata, auth_headers):
|
|
data = _report(client, auth_headers, period='all')
|
|
assert data['days'] == 0
|
|
assert data['summary']['events'] == 3
|
|
|
|
|
|
def test_today_is_a_window_too(client, usbdata, auth_headers):
|
|
"""A shift handover asks what happened today, not this month."""
|
|
data = _report(client, auth_headers, period='day')
|
|
assert data['days'] == 1
|
|
assert data['summary']['events'] == 2
|
|
|
|
|
|
@pytest.mark.parametrize('period,days', [('day', 1), ('month', 30),
|
|
('sixmonths', 182), ('year', 365),
|
|
('all', 0)])
|
|
def test_every_named_window_maps_to_a_day_count(client, usbdata, auth_headers,
|
|
period, days):
|
|
assert _report(client, auth_headers, period=period)['days'] == days
|
|
|
|
|
|
def test_an_unknown_period_falls_back_rather_than_failing(client, usbdata,
|
|
auth_headers):
|
|
"""A bad query string should not 500 a report somebody is mid-audit on."""
|
|
assert _report(client, auth_headers, period='fortnight')['days'] == 30
|
|
|
|
|
|
def test_rows_are_newest_first(client, usbdata, auth_headers):
|
|
rows = _report(client, auth_headers, period='all')['rows']
|
|
stamps = [row['timestamp'] for row in rows]
|
|
assert stamps == sorted(stamps, reverse=True)
|
|
|
|
|
|
def test_it_says_who_and_which_drive(client, usbdata, auth_headers):
|
|
rows = _report(client, auth_headers, period='month')['rows']
|
|
row = rows[0]
|
|
assert row['name'] == 'Dana Reyes'
|
|
assert row['badge'] == '270015376'
|
|
assert row['device_id'] == 'USB-1'
|
|
assert row['device_desc'] == 'Blue Kingston'
|
|
|
|
|
|
def test_it_exports_as_csv(client, usbdata, auth_headers):
|
|
response = client.get('/api/usb/reports/audit?period=all&format=csv',
|
|
headers=auth_headers)
|
|
assert response.status_code == 200
|
|
assert response.mimetype == 'text/csv'
|
|
body = response.get_data(as_text=True)
|
|
assert body.splitlines()[0].startswith('timestamp,action,device_id')
|
|
assert 'Dana Reyes' in body
|
|
|
|
|
|
def test_reading_the_audit_needs_permission(client, usbdata, member_headers):
|
|
"""A role-less member cannot read who had which drive."""
|
|
response = client.get('/api/usb/reports/audit?period=all',
|
|
headers=member_headers)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_the_report_is_registered_so_it_appears_in_the_catalogue(client, usbdata,
|
|
auth_headers):
|
|
from plugins.usb.plugin import USBPlugin
|
|
report = USBPlugin().get_reports()[0]
|
|
assert report['id'] == 'usb-audit'
|
|
# The endpoint is what the reports page fetches; a wrong one renders blank.
|
|
assert report['endpoint'] == '/api/usb/reports/audit'
|