ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
405a2ef919c04f13bb1de7d765e5fd896d0044d1
shopdb-flask/tests/test_plugins/test_printedparts_ledger.py
cproudlock 744bc4341a Shut the parts kiosk behind a device token, and stop the health probe leaking 
Both from the second Darkwing scan. Every claim in it was verified against the
code before acting.

The kiosk was the product's only unauthenticated write, justified by a decision
record saying it was "physically rate-limited: it serves a touch screen on the
shop floor". Nothing in code enforced the physical half, and an HTTP route is
reachable from wherever the app is. What that left open:

  - _kiosk_find_item falls back to matching bare digits against the numeric
    tail of every active item, so 1, 2, 3 walk the catalogue without knowing a
    barcode;
  - a badge is a short numeric SSO, and the route answered 422 for an unknown
    one and 200 for a real one - an oracle for sorting guesses into employees;
  - the write then decremented stock and stamped the ledger, which IS the audit
    trail, with whatever SSO the caller supplied;
  - quantity was bounded only by what was on hand, so one request emptied a bin.

A badge is attribution, never authorization. Both kiosk routes now need the
site's kiosk flag AND a per-device token, and 404 rather than 403 without them
so a caller does not learn the door is there. The READ is gated too: gating only
the write would have left the enumeration oracle running against a kiosk a site
had switched off.

The token is per-DEVICE, not per-page, because a kiosk is a browser on a touch
screen and a token served inside the page is readable by anyone who can load it.
An admin opens the screen once as /parts-kiosk?token=..., the page moves it into
that browser's localStorage and strips it from the address bar.

Also: quantity is capped per take, JSON true no longer passes as 1
(isinstance(True, int) is True), and every badge failure on the anonymous route
returns one message. The gated restock and adjust handlers keep the specific
message - there the person is known and the detail is help, not an oracle. I
changed the wrong one of those three handlers first; caught by the test.

/api/dashboard/health formatted str(e) straight into the body of a route
PUBLIC_ENDPOINTS declares open because "monitoring has no account". A pymysql
OperationalError names the database account, the app host's address, the driver
and the failing statement. It is logged server-side now and the body carries a
coarse status. The version goes with it - it told an anonymous caller which
release to look up. Status stays 200 either way: a monitor keyed on the body's
status field is the contract here, and turning a degraded database into a
transport failure is a separate decision from not leaking the exception.

Nine tests. The three gate tests were confirmed to fail with the gate stubbed
out, rather than assumed to cover it. Suite 2325, frontend 316, build clean.

Scope: product
2026-09-02 17:42:54 -04:00

535 lines
25 KiB
Python
Raw Blame History

"""Printedparts ledger + badge tests.
The invariants that make the plugin trustworthy: itemcode minting, the
single-commit cache==ledger rule, the below-zero guard, quantity edits
forced through the ledger, and the badge contract (SSO digits, PayNo
wrap, unknown-badge policy).
"""
import pytest
from shopdb.api import db
from shopdb.core.models import Setting
from plugins.printedparts.models import PrintedItem, PrintedItemTransaction
@pytest.fixture
def item(app, db):
with app.app_context():
row = PrintedItem(itemcode='3DP-9001', itemname='Test clip',
quantityonhand=0, lowstockthreshold=5)
db.session.add(row)
db.session.commit()
yield row.printeditemid
@pytest.fixture
def directory_employee(app):
with app.app_context():
from plugins.employees.models import DirectoryEmployee
if not db.session.get(DirectoryEmployee, 502000001):
db.session.add(DirectoryEmployee(
sso=502000001, firstname='Pat', lastname='Printer'))
db.session.commit()
return '502000001'
def test_create_mints_itemcode(client, auth_headers):
response = client.post('/api/printedparts/items', json={'itemname': 'Bracket'},
headers=auth_headers)
assert response.status_code == 201
data = response.get_json()['data']
assert data['itemcode'] == f"3DP{data['printeditemid']:04d}"
assert data['quantityonhand'] == 0
def test_update_refuses_quantity(client, auth_headers, item):
response = client.put(f'/api/printedparts/items/{item}',
json={'quantityonhand': 50}, headers=auth_headers)
assert response.status_code == 400
def test_restock_writes_ledger_and_cache(client, auth_headers, app, item,
directory_employee):
response = client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': 10, 'badge': directory_employee},
headers=auth_headers)
assert response.status_code == 200
assert response.get_json()['data']['quantityonhand'] == 10
with app.app_context():
rows = PrintedItemTransaction.query.filter_by(printeditemid=item).all()
assert len(rows) == 1
assert rows[0].transactiontype == 'restock'
assert rows[0].quantitychange == 10
assert rows[0].employeename == 'Pat Printer'
cached = db.session.get(PrintedItem, item).quantityonhand
assert cached == sum(r.quantitychange for r in rows)
def test_payno_badge_shape_resolves(client, auth_headers, item,
directory_employee):
response = client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': 1,
'badge': f'0{directory_employee}BZ'},
headers=auth_headers)
assert response.status_code == 200
def test_adjust_requires_reason_and_floors_at_zero(client, auth_headers, item,
directory_employee):
no_reason = client.post(f'/api/printedparts/items/{item}/adjust',
json={'quantitychange': -1,
'badge': directory_employee},
headers=auth_headers)
assert no_reason.status_code == 400
below_zero = client.post(f'/api/printedparts/items/{item}/adjust',
json={'quantitychange': -1, 'reason': 'test',
'badge': directory_employee},
headers=auth_headers)
assert below_zero.status_code == 400
def test_unknown_badge_denied_then_allowed_by_policy(client, auth_headers, app,
item):
denied = client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': 1, 'badge': '999999999'},
headers=auth_headers)
assert denied.status_code == 422
with app.app_context():
Setting.set('printedparts_unknown_badge', 'allow',
valuetype='string', category='printedparts')
db.session.commit()
try:
allowed = client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': 1, 'badge': '999999999'},
headers=auth_headers)
assert allowed.status_code == 200
assert allowed.get_json()['data']['quantityonhand'] == 1
finally:
with app.app_context():
Setting.set('printedparts_unknown_badge', 'deny',
valuetype='string', category='printedparts')
db.session.commit()
def test_anonymous_cannot_mutate(client, item):
assert client.post('/api/printedparts/items',
json={'itemname': 'X'}).status_code == 401
assert client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': 1, 'badge': '1'}).status_code == 401
KIOSK_TOKEN = 'kiosk-device-token-for-tests'
# The header a provisioned screen sends. A module constant rather than the
# fixture's return value, so a test does not have to name the fixture just to
# reach the header.
KIOSK_HEADERS = {'X-Kiosk-Token': KIOSK_TOKEN}
@pytest.fixture(autouse=True)
def kiosk(app, db):
"""Provision the kiosk the way an administrator would, and hand back the
header a provisioned screen sends. Without this every kiosk route 404s,
which is the point: the routes are the only unauthenticated write in the
product, so they are off until a site opts in."""
with app.app_context():
Setting.set('printedparts_kiosk_enabled', 'true')
Setting.set('printedparts_kiosk_token', KIOSK_TOKEN)
db.session.commit()
return KIOSK_HEADERS
def test_catalog_reads_require_view_permission(client, member_headers, item):
"""Browsing the catalog is printedparts.view-gated; a PROVISIONED kiosk
reads without an account."""
assert client.get('/api/printedparts/items').status_code == 401
assert client.get(f'/api/printedparts/items/{item}').status_code == 401
assert client.get('/api/printedparts/items',
headers=member_headers).status_code == 403
assert client.get('/api/printedparts/kiosk/item/3DP-9001',
headers=KIOSK_HEADERS).status_code == 200
def test_member_without_permission_gets_403(client, member_headers, item):
"""Authentication alone is not authorization: a role-less user is denied."""
assert client.post('/api/printedparts/items', json={'itemname': 'X'},
headers=member_headers).status_code == 403
assert client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': 1, 'badge': '1'},
headers=member_headers).status_code == 403
def test_kiosk_take_is_decrement_only(client, auth_headers, app, item,
directory_employee, kiosk):
"""A provisioned kiosk needs no account but only ever decrements stock."""
itemcode = '3DP-9001'
stocked = client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': 5, 'badge': directory_employee},
headers=auth_headers)
assert stocked.status_code == 200
lookup = client.get(f'/api/printedparts/kiosk/item/{itemcode}', headers={**auth_headers, **KIOSK_HEADERS})
assert lookup.status_code == 200
take = client.post('/api/printedparts/kiosk/take', json={
'itemcode': itemcode, 'badge': directory_employee, 'quantity': 2},
headers={**auth_headers, **KIOSK_HEADERS})
assert take.status_code == 200, take.get_json()
assert take.get_json()['data']['quantityonhand'] == 3
too_many = client.post('/api/printedparts/kiosk/take', json={
'itemcode': itemcode, 'badge': directory_employee, 'quantity': 99},
headers={**auth_headers, **KIOSK_HEADERS})
assert too_many.status_code == 400
unknown = client.post('/api/printedparts/kiosk/take', json={
'itemcode': itemcode, 'badge': '111111111', 'quantity': 1},
headers={**auth_headers, **KIOSK_HEADERS})
assert unknown.status_code == 422
with app.app_context():
rows = PrintedItemTransaction.query.filter_by(
printeditemid=item, transactiontype='take').all()
assert len(rows) == 1
assert rows[0].quantitychange == -2
assert rows[0].employeename == 'Pat Printer'
cached = db.session.get(PrintedItem, item).quantityonhand
ledgersum = sum(r.quantitychange for r in
PrintedItemTransaction.query.filter_by(
printeditemid=item).all())
assert cached == ledgersum
def test_lowstock_alert_fires_on_crossing_only(client, auth_headers, app, item,
directory_employee, monkeypatch):
"""One alert when stock CROSSES the threshold downward; restocking above
rearms it; staying below does not re-fire."""
sent = []
import plugins.printedparts.api.routes as printedparts_routes
monkeypatch.setattr(
printedparts_routes, '_send_lowstock_alert',
lambda alerted_item: sent.append(alerted_item.itemcode))
def restock(quantity):
return client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': quantity,
'badge': directory_employee},
headers=auth_headers)
def take(quantity):
return client.post('/api/printedparts/kiosk/take',
json={'itemcode': '3DP-9001',
'badge': directory_employee,
'quantity': quantity}, headers={**auth_headers, **KIOSK_HEADERS})
restock(10) # 10 on hand, threshold 5
assert take(3).status_code == 200 # 7: above threshold, no alert
assert sent == []
assert take(3).status_code == 200 # 4: CROSSES 5 -> one alert
assert sent == ['3DP-9001']
assert take(2).status_code == 200 # 2: still below, no re-fire
assert sent == ['3DP-9001']
restock(20) # 22: rearmed
assert take(18).status_code == 200 # 4: crosses again -> second alert
assert sent == ['3DP-9001', '3DP-9001']
def test_alert_recipients_merge_users_and_freetext(client, auth_headers, app,
item, directory_employee,
monkeypatch):
"""Selected shopdb users' account emails merge with the free-text list,
deduped; inactive users are skipped."""
import shopdb.api as contract_surface
captured = {}
monkeypatch.setattr(contract_surface, 'send_email',
lambda to, subject, html, text=None:
captured.setdefault('to', to) or True)
with app.app_context():
from shopdb.api import User
from werkzeug.security import generate_password_hash
active = User(username='partslead', email='lead@site.test',
passwordhash=generate_password_hash('x'), isactive=True)
inactive = User(username='oldtimer', email='gone@site.test',
passwordhash=generate_password_hash('x'),
isactive=False)
db.session.add_all([active, inactive])
db.session.commit()
Setting.set('printedparts_alert_userids',
f'{active.userid},{inactive.userid}',
valuetype='string', category='printedparts')
Setting.set('printedparts_alert_email',
'extra@site.test, lead@site.test',
valuetype='string', category='printedparts')
db.session.commit()
client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': 10, 'badge': directory_employee},
headers=auth_headers)
take = client.post('/api/printedparts/kiosk/take',
json={'itemcode': '3DP-9001',
'badge': directory_employee, 'quantity': 6}, headers={**auth_headers, **KIOSK_HEADERS})
assert take.status_code == 200 # 4 on hand: crossed threshold 5
assert captured['to'] == ['lead@site.test', 'extra@site.test']
def test_retire_hides_and_restore_returns(client, auth_headers, item):
"""Retire drops the item from the default list and the kiosk; restore
brings it back with history intact."""
assert client.delete(f'/api/printedparts/items/{item}',
headers=auth_headers).status_code == 200
listed = client.get('/api/printedparts/items',
headers=auth_headers).get_json()['data']
assert all(row['printeditemid'] != item for row in listed)
kiosk = client.get('/api/printedparts/kiosk/item/3DP-9001', headers={**auth_headers, **KIOSK_HEADERS})
assert kiosk.status_code == 404
including = client.get('/api/printedparts/items?active=false',
headers=auth_headers)
assert any(row['printeditemid'] == item
for row in including.get_json()['data'])
assert client.post(f'/api/printedparts/items/{item}/restore',
headers=auth_headers).status_code == 200
assert client.get('/api/printedparts/kiosk/item/3DP-9001', headers={**auth_headers, **KIOSK_HEADERS}).status_code == 200
def test_file_revisions_append_and_download(client, auth_headers, item, tmp_path):
"""Uploads mint sequential revisions; download returns the original name."""
import io
first = client.post(f'/api/printedparts/items/{item}/files',
data={'file': (io.BytesIO(b'solid part'), 'clip_v1.stl'),
'note': 'initial'},
headers=auth_headers,
content_type='multipart/form-data')
assert first.status_code == 201, first.get_json()
assert first.get_json()['data']['revision'] == 1
second = client.post(f'/api/printedparts/items/{item}/files',
data={'file': (io.BytesIO(b'G1 X0 Y0'), 'clip_v2.gcode')},
headers=auth_headers,
content_type='multipart/form-data')
assert second.get_json()['data']['revision'] == 2
bad = client.post(f'/api/printedparts/items/{item}/files',
data={'file': (io.BytesIO(b'x'), 'malware.exe')},
headers=auth_headers,
content_type='multipart/form-data')
assert bad.status_code == 400
listing = client.get(f'/api/printedparts/items/{item}/files',
headers=auth_headers).get_json()['data']
assert [f['revision'] for f in listing] == [2, 1]
fileid = listing[1]['fileid']
download = client.get(f'/api/printedparts/files/{fileid}/download', headers=auth_headers)
assert download.status_code == 200
assert download.data == b'solid part'
assert 'clip_v1.stl' in download.headers['Content-Disposition']
def test_alert_role_members_receive(client, auth_headers, app, item,
directory_employee, monkeypatch):
"""Every active member of a selected role gets the alert."""
import shopdb.api as contract_surface
captured = {}
monkeypatch.setattr(contract_surface, 'send_email',
lambda to, subject, html, text=None:
captured.setdefault('to', to) or True)
with app.app_context():
from shopdb.api import User, Role
from werkzeug.security import generate_password_hash
role = Role(rolename='partscrew', description='3D parts crew')
member = User(username='crewone', email='crewone@site.test',
passwordhash=generate_password_hash('x'), isactive=True)
member.roles.append(role)
db.session.add_all([role, member])
db.session.commit()
Setting.set('printedparts_alert_roleids', str(role.roleid),
valuetype='string', category='printedparts')
db.session.commit()
client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': 10, 'badge': directory_employee},
headers=auth_headers)
take = client.post('/api/printedparts/kiosk/take',
json={'itemcode': '3DP-9001',
'badge': directory_employee, 'quantity': 6}, headers={**auth_headers, **KIOSK_HEADERS})
assert take.status_code == 200
assert captured['to'] == ['crewone@site.test']
def test_gagelabtag_assigned_searched_and_kiosk_resolved(client, auth_headers):
"""The internal code stays auto-minted; the gage-lab tag is optional,
unique, searchable, and the kiosk resolves it (exact and bare digits)."""
created = client.post('/api/printedparts/items',
json={'itemname': 'Gage block holder',
'gagelabtag': 'wjrp0117'},
headers=auth_headers)
assert created.status_code == 201
data = created.get_json()['data']
assert data['gagelabtag'] == 'WJRP0117'
assert data['itemcode'].startswith('3DP') # internal code untouched
duplicate = client.post('/api/printedparts/items',
json={'itemname': 'Other',
'gagelabtag': 'WJRP0117'},
headers=auth_headers)
assert duplicate.status_code == 409
searched = client.get('/api/printedparts/items?search=WJRP0117',
headers=auth_headers).get_json()['data']
assert len(searched) == 1
by_tag = client.get('/api/printedparts/kiosk/item/WJRP0117', headers={**auth_headers, **KIOSK_HEADERS})
assert by_tag.status_code == 200
by_digits = client.get('/api/printedparts/kiosk/item/117', headers={**auth_headers, **KIOSK_HEADERS})
assert by_digits.status_code == 200
assert by_digits.get_json()['data']['gagelabtag'] == 'WJRP0117'
def test_kiosk_take_records_revision_from_qr(client, app, db, directory_employee, auth_headers):
"""A label QR payload 'TAG|rev' resolves the item by TAG and records the
revision on the take (traceability of which print-file rev was consumed)."""
with app.app_context():
row = PrintedItem(itemcode='3DP-7000', gagelabtag='WJRP7000',
itemname='Rev part', quantityonhand=10, lowstockthreshold=2)
db.session.add(row)
db.session.commit()
iid = row.printeditemid
take = client.post('/api/printedparts/kiosk/take', json={
'itemcode': 'WJRP7000|3', 'badge': directory_employee, 'quantity': 1}, headers={**auth_headers, **KIOSK_HEADERS})
assert take.status_code == 200, take.get_json()
with app.app_context():
txn = PrintedItemTransaction.query.filter_by(
printeditemid=iid, transactiontype='take').first()
assert txn is not None
assert txn.revision == 3
# Manual entry (no |rev) records a null revision, still resolves.
take2 = client.post('/api/printedparts/kiosk/take', json={
'itemcode': 'WJRP7000', 'badge': directory_employee, 'quantity': 1}, headers={**auth_headers, **KIOSK_HEADERS})
assert take2.status_code == 200
with app.app_context():
last = (PrintedItemTransaction.query
.filter_by(printeditemid=iid, transactiontype='take')
.order_by(PrintedItemTransaction.transactionid.desc()).first())
assert last.revision is None
def test_lowstock_routes_to_support_team_webhook(client, app, db, directory_employee, auth_headers):
"""A low-stock crossing posts to the selected support team's webhook URL."""
from unittest.mock import patch
from shopdb.core.models import SupportTeam, Setting
with app.app_context():
team = SupportTeam(teamname='3D Print Team',
webhookurl='https://teams.example/3dprint')
db.session.add(team)
db.session.commit()
Setting.set('printedparts_alert_supportteamid', str(team.supportteamid))
row = PrintedItem(itemcode='3DP-7100', gagelabtag='WJRP7100',
itemname='Widget', quantityonhand=6, lowstockthreshold=5)
db.session.add(row)
db.session.commit()
with patch('shopdb.api.send_webhook') as mock_send:
# take 2 -> 6 crosses to 4 (<= threshold 5) -> alert fires
resp = client.post('/api/printedparts/kiosk/take', json={
'itemcode': 'WJRP7100', 'badge': directory_employee, 'quantity': 2}, headers={**auth_headers, **KIOSK_HEADERS})
assert resp.status_code == 200, resp.get_json()
assert mock_send.called
assert mock_send.call_args.kwargs['url'] == 'https://teams.example/3dprint'
# --- the kiosk is off until a site opts in -----------------------------------
#
# These are the cases the Darkwing scan turned into an exploit path. Every one
# of them returned 200 and wrote to the ledger before the gate existed.
def _clear_kiosk(app, db):
with app.app_context():
Setting.set('printedparts_kiosk_enabled', 'false')
Setting.set('printedparts_kiosk_token', '')
db.session.commit()
def test_kiosk_is_shut_when_the_site_has_not_enabled_it(client, app, db, item):
_clear_kiosk(app, db)
assert client.get('/api/printedparts/kiosk/item/3DP-9001').status_code == 404
assert client.post('/api/printedparts/kiosk/take',
json={'itemcode': '3DP-9001', 'badge': '1',
'quantity': 1}).status_code == 404
def test_kiosk_is_shut_without_the_device_token(client, item):
"""Enabled is not enough: an anonymous caller has no token."""
assert client.get('/api/printedparts/kiosk/item/3DP-9001').status_code == 404
assert client.post('/api/printedparts/kiosk/take',
json={'itemcode': '3DP-9001', 'badge': '1',
'quantity': 1}).status_code == 404
def test_kiosk_is_shut_with_the_wrong_device_token(client, item):
wrong = {'X-Kiosk-Token': KIOSK_TOKEN + 'x'}
assert client.get('/api/printedparts/kiosk/item/3DP-9001',
headers=wrong).status_code == 404
assert client.post('/api/printedparts/kiosk/take', headers=wrong,
json={'itemcode': '3DP-9001', 'badge': '1',
'quantity': 1}).status_code == 404
def test_a_single_take_cannot_drain_a_bin(client, app, item, auth_headers,
directory_employee):
"""quantity was bounded only by what was on hand, so one request emptied
the bin. It is capped per transaction now."""
from plugins.printedparts.api.routes import KIOSK_MAX_TAKE_QUANTITY
client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': KIOSK_MAX_TAKE_QUANTITY + 10,
'badge': directory_employee}, headers=auth_headers)
over = client.post('/api/printedparts/kiosk/take', headers=KIOSK_HEADERS,
json={'itemcode': '3DP-9001',
'badge': directory_employee,
'quantity': KIOSK_MAX_TAKE_QUANTITY + 1})
assert over.status_code == 400
assert str(KIOSK_MAX_TAKE_QUANTITY) in over.get_json()['data']['error']['message']
def test_true_is_not_a_quantity(client, item, directory_employee):
"""isinstance(True, int) is True in Python, so a JSON true used to sail
through the type check and take one."""
resp = client.post('/api/printedparts/kiosk/take', headers=KIOSK_HEADERS,
json={'itemcode': '3DP-9001',
'badge': directory_employee, 'quantity': True})
assert resp.status_code == 400
def test_the_badge_route_is_not_an_enumeration_oracle(client, item, auth_headers,
directory_employee):
"""An unknown badge and a malformed one must be indistinguishable, or the
route tells an anonymous caller which numbers are real employees.
Stock first: the on-hand check runs BEFORE the badge is resolved, so an
empty bin would answer 400 for both and the test would pass without ever
reaching the code it is about."""
client.post(f'/api/printedparts/items/{item}/restock',
json={'quantity': 5, 'badge': directory_employee},
headers=auth_headers)
unknown = client.post('/api/printedparts/kiosk/take', headers=KIOSK_HEADERS,
json={'itemcode': '3DP-9001', 'badge': '999999999',
'quantity': 1})
malformed = client.post('/api/printedparts/kiosk/take', headers=KIOSK_HEADERS,
json={'itemcode': '3DP-9001', 'badge': 'not-a-badge',
'quantity': 1})
assert unknown.status_code == malformed.status_code == 422
assert (unknown.get_json()['data']['error']['message']
== malformed.get_json()['data']['error']['message'])
Reference in New Issue View Git Blame Copy Permalink