Shut the parts kiosk behind a device token, and stop the health probe leaking
This commit is contained in:
@@ -1501,6 +1501,17 @@ export const measuringtoolsApi = {
|
||||
}
|
||||
|
||||
// 3D printed parts (printedparts plugin)
|
||||
// Device token for the parts kiosk, provisioned once per screen and kept in
|
||||
// that browser. Absent means the routes answer 404, which is what an
|
||||
// unprovisioned browser should see.
|
||||
export const KIOSK_TOKEN_KEY = 'printedparts-kiosk-token'
|
||||
|
||||
function kioskHeaders() {
|
||||
let token = ''
|
||||
try { token = localStorage.getItem(KIOSK_TOKEN_KEY) || '' } catch { token = '' }
|
||||
return token ? { 'X-Kiosk-Token': token } : {}
|
||||
}
|
||||
|
||||
export const printedpartsApi = {
|
||||
list(params = {}) {
|
||||
return api.get('/printedparts/items', { params })
|
||||
@@ -1541,11 +1552,17 @@ export const printedpartsApi = {
|
||||
adjust(printeditemid, data) {
|
||||
return api.post(`/printedparts/items/${printeditemid}/adjust`, data)
|
||||
},
|
||||
// The kiosk routes are the only unauthenticated write in the product, so they
|
||||
// are gated on a per-DEVICE token rather than on the page. It lives in this
|
||||
// browser's localStorage, put there once by an admin opening the kiosk with
|
||||
// ?token=... - see PartsKiosk.vue. A token shipped inside the page would be
|
||||
// readable by anyone who could load it, which is not a gate.
|
||||
kioskItem(itemcode) {
|
||||
return api.get(`/printedparts/kiosk/item/${encodeURIComponent(itemcode)}`)
|
||||
return api.get(`/printedparts/kiosk/item/${encodeURIComponent(itemcode)}`,
|
||||
{ headers: kioskHeaders() })
|
||||
},
|
||||
kioskTake(data) {
|
||||
return api.post('/printedparts/kiosk/take', data)
|
||||
return api.post('/printedparts/kiosk/take', data, { headers: kioskHeaders() })
|
||||
},
|
||||
listFiles(printeditemid) {
|
||||
return api.get(`/printedparts/items/${printeditemid}/files`)
|
||||
|
||||
@@ -431,11 +431,54 @@ def adjust_item(item_id: int):
|
||||
return success_response(item.to_dict(), message='Stock adjusted')
|
||||
|
||||
|
||||
# --- kiosk: UNauthenticated by decision record --------------------------------
|
||||
# The take endpoint is the product's first open WRITE. The proposal's decision
|
||||
# record sets the bar it must meet: decrement-only, badge-attributed, bounded,
|
||||
# physically rate-limited. It can reduce stock of an active item and nothing
|
||||
# else; identity comes from the badge resolved server-side, never the client.
|
||||
# --- kiosk: off by default, bound to a provisioned device token ---------------
|
||||
# The take endpoint is the product's only open WRITE. The original decision
|
||||
# record justified that with "physically rate-limited: it serves a touch screen
|
||||
# on the shop floor". Nothing in code enforced the physical part, and an HTTP
|
||||
# route is reachable from anywhere the app is, so the justification did not
|
||||
# survive contact with a scanner.
|
||||
#
|
||||
# What was actually reachable with no credential: bare digits in
|
||||
# _kiosk_find_item enumerate the catalogue without knowing a barcode, a badge is
|
||||
# a short numeric SSO, and this route answered 422 for an unknown badge and 200
|
||||
# for a known one - an oracle for finding real ones. The write then decremented
|
||||
# stock and stamped the ledger, which is the audit trail, with whichever SSO the
|
||||
# caller supplied.
|
||||
#
|
||||
# So a badge is ATTRIBUTION, never authorization. Both kiosk routes now require
|
||||
# the kiosk to be switched on for this site AND the device token an administrator
|
||||
# provisioned. The read is gated too, not just the write: leaving it open would
|
||||
# keep the enumeration oracle running against a kiosk that is supposedly off.
|
||||
#
|
||||
# The token is per-DEVICE, not per-page. A kiosk is a browser on a touch screen,
|
||||
# so a token embedded in the served page would be readable by anyone who can
|
||||
# load it. An admin opens the kiosk once with ?token=..., the page keeps it in
|
||||
# that browser's localStorage and sends it as X-Kiosk-Token thereafter.
|
||||
|
||||
import hmac
|
||||
|
||||
# Largest single take the kiosk records. A shop-floor grab is single digits; 25
|
||||
# leaves real use untouched and denies draining a bin in one request.
|
||||
KIOSK_MAX_TAKE_QUANTITY = 25
|
||||
|
||||
|
||||
def _kiosk_open():
|
||||
"""True when this site has the kiosk on AND the caller carries its token."""
|
||||
enabled = Setting.get('printedparts_kiosk_enabled', default=False)
|
||||
if not (enabled is True or str(enabled).strip().lower() in ('true', '1', 'yes', 'on')):
|
||||
return False
|
||||
expected = (Setting.get('printedparts_kiosk_token') or '').strip()
|
||||
presented = (request.headers.get('X-Kiosk-Token') or '').strip()
|
||||
if not expected or not presented:
|
||||
return False
|
||||
# compare_digest, not ==: the token is a secret and == leaks its length and
|
||||
# matching prefix through timing.
|
||||
return hmac.compare_digest(presented, expected)
|
||||
|
||||
|
||||
def _kiosk_shut():
|
||||
"""404, not 403 - a caller without the token does not learn the door exists."""
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'Not found', http_code=404)
|
||||
|
||||
def _kiosk_find_item(itemcode):
|
||||
"""Resolve a scanned or typed code to an active item.
|
||||
@@ -468,7 +511,9 @@ def _kiosk_find_item(itemcode):
|
||||
|
||||
@printedparts_bp.route('/kiosk/item/<itemcode>', methods=['GET'])
|
||||
def kiosk_item(itemcode):
|
||||
"""Item summary for a scanned bin barcode (open read for the kiosk)."""
|
||||
"""Item summary for a scanned bin barcode. Kiosk-token gated."""
|
||||
if not _kiosk_open():
|
||||
return _kiosk_shut()
|
||||
item = _kiosk_find_item(itemcode)
|
||||
if not item:
|
||||
return error_response(ErrorCodes.NOT_FOUND,
|
||||
@@ -478,7 +523,13 @@ def kiosk_item(itemcode):
|
||||
|
||||
@printedparts_bp.route('/kiosk/take', methods=['POST'])
|
||||
def kiosk_take():
|
||||
"""Take parts from a bin. Body: {itemcode, badge, quantity}."""
|
||||
"""Take parts from a bin. Body: {itemcode, badge, quantity}.
|
||||
|
||||
The badge attributes the ledger row. It does not authorize the write - the
|
||||
kiosk flag and the device token do.
|
||||
"""
|
||||
if not _kiosk_open():
|
||||
return _kiosk_shut()
|
||||
data = request.get_json() or {}
|
||||
|
||||
item = _kiosk_find_item(data.get('itemcode'))
|
||||
@@ -487,9 +538,17 @@ def kiosk_take():
|
||||
'No part matches that barcode', http_code=404)
|
||||
|
||||
quantity = data.get('quantity')
|
||||
if not isinstance(quantity, int) or quantity < 1:
|
||||
# isinstance(True, int) is True in Python, so True would have passed as a
|
||||
# quantity of 1. Reject bool explicitly.
|
||||
if (not isinstance(quantity, int) or isinstance(quantity, bool)
|
||||
or quantity < 1):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'Enter how many you are taking')
|
||||
if quantity > KIOSK_MAX_TAKE_QUANTITY:
|
||||
return error_response(
|
||||
ErrorCodes.VALIDATION_ERROR,
|
||||
f'Take at most {KIOSK_MAX_TAKE_QUANTITY} at a time - see the '
|
||||
f'parts team')
|
||||
if quantity > item.quantityonhand:
|
||||
return error_response(
|
||||
ErrorCodes.VALIDATION_ERROR,
|
||||
@@ -498,8 +557,14 @@ def kiosk_take():
|
||||
|
||||
try:
|
||||
sso, name = resolve_badge(data.get('badge'))
|
||||
except BadgeError as badge_error:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, str(badge_error),
|
||||
except BadgeError:
|
||||
# ONE message for every badge failure on the ANONYMOUS route. This is
|
||||
# the only place it matters: distinguishing "malformed" from "not
|
||||
# recognized" let a caller with no account sort guesses into real
|
||||
# employee numbers. The gated restock/adjust handlers keep the specific
|
||||
# message, because there the person is known and the detail is help.
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'Badge not accepted - see the parts team',
|
||||
http_code=422)
|
||||
|
||||
# Print-file revision the scanned part carried: explicit field, else the
|
||||
|
||||
@@ -92,7 +92,7 @@
|
||||
|
||||
<script setup>
|
||||
import { ref, onMounted, onBeforeUnmount, nextTick } from 'vue'
|
||||
import { printedpartsApi, settingsApi } from '@/api'
|
||||
import { printedpartsApi, settingsApi, KIOSK_TOKEN_KEY } from '@/api'
|
||||
import { withBase } from '@/utils/basePath'
|
||||
import TouchKeypad from '@/components/TouchKeypad.vue'
|
||||
|
||||
@@ -116,7 +116,28 @@ const scannedRevision = ref(null)
|
||||
const entryInput = ref(null)
|
||||
let resetTimer = null
|
||||
|
||||
// Claim the device token, once, from the URL an admin used to set this screen
|
||||
// up: /parts-kiosk?token=<value>. It is kept in THIS browser and stripped from
|
||||
// the address bar immediately, so it does not sit in the URL for a shoulder to
|
||||
// read, and does not travel in a bookmark or a screenshot. Every kiosk call
|
||||
// then carries it as X-Kiosk-Token; without it the routes answer 404.
|
||||
function claimTokenFromUrl() {
|
||||
try {
|
||||
const url = new URL(window.location.href)
|
||||
const token = (url.searchParams.get('token') || '').trim()
|
||||
if (!token) return
|
||||
localStorage.setItem(KIOSK_TOKEN_KEY, token)
|
||||
url.searchParams.delete('token')
|
||||
window.history.replaceState({}, '', url.pathname + url.search + url.hash)
|
||||
} catch {
|
||||
// A browser refusing localStorage is a browser this kiosk cannot be
|
||||
// provisioned in. Nothing to recover here; the routes will 404 and the
|
||||
// operator sees the standard error.
|
||||
}
|
||||
}
|
||||
|
||||
onMounted(async () => {
|
||||
claimTokenFromUrl()
|
||||
// A kiosk left running for weeks reads this once at load, which is fine: the
|
||||
// label format is a property of the site's physical tags, not something that
|
||||
// changes during a shift.
|
||||
|
||||
@@ -150,6 +150,24 @@ class PrintedpartsPlugin(BasePlugin):
|
||||
'category': 'printedparts',
|
||||
'description': 'Default low-stock threshold for new items',
|
||||
},
|
||||
{
|
||||
'key': 'printedparts_kiosk_enabled',
|
||||
'value': 'false',
|
||||
'valuetype': 'boolean',
|
||||
'category': 'printedparts',
|
||||
'description': 'Serve the parts kiosk routes. Off by default: '
|
||||
'they are the only unauthenticated write in the '
|
||||
'product, so a site opts in',
|
||||
},
|
||||
{
|
||||
'key': 'printedparts_kiosk_token',
|
||||
'value': '',
|
||||
'valuetype': 'secret',
|
||||
'category': 'printedparts',
|
||||
'description': 'Device token the kiosk sends as X-Kiosk-Token. '
|
||||
'Generate one, then open the kiosk once as '
|
||||
'/parts-kiosk?token=<value> on each screen',
|
||||
},
|
||||
{
|
||||
'key': 'printedparts_unknown_badge',
|
||||
'value': 'deny',
|
||||
|
||||
@@ -317,17 +317,30 @@ def get_widgets():
|
||||
|
||||
@dashboard_bp.route('/health', methods=['GET'])
|
||||
def health_check():
|
||||
"""Health check endpoint (no auth required)."""
|
||||
from shopdb import __version__
|
||||
"""Health check endpoint (no auth required).
|
||||
|
||||
Anonymous callers get a coarse status and nothing else. The probe used to
|
||||
format str(e) straight into the body, and a pymysql OperationalError names
|
||||
the database account, the app host's address, the driver and the failing
|
||||
statement - the internal topology, handed to whoever asks, on a route
|
||||
PUBLIC_ENDPOINTS declares open because "monitoring has no account". The
|
||||
detail is logged server-side instead, where the people who need it are.
|
||||
|
||||
The version goes the same way. It told an anonymous caller exactly which
|
||||
release to look up known issues for, and monitoring does not need it.
|
||||
"""
|
||||
try:
|
||||
db.session.execute(db.text('SELECT 1'))
|
||||
db_status = 'healthy'
|
||||
except Exception as e:
|
||||
db_status = f'unhealthy: {str(e)}'
|
||||
healthy = True
|
||||
except Exception:
|
||||
current_app.logger.exception('Health check database probe failed')
|
||||
healthy = False
|
||||
|
||||
# Status code stays 200 either way. A monitor keyed on the body's status
|
||||
# field is the documented contract here, and turning a degraded database
|
||||
# into a transport-level failure would change what every existing check
|
||||
# sees - a separate decision from not leaking the exception.
|
||||
return success_response({
|
||||
'status': 'ok' if db_status == 'healthy' else 'degraded',
|
||||
'database': db_status,
|
||||
'version': __version__
|
||||
'status': 'ok' if healthy else 'degraded',
|
||||
'database': 'healthy' if healthy else 'unhealthy',
|
||||
})
|
||||
|
||||
@@ -46,10 +46,14 @@ EXEMPT_BLUEPRINTS = {'auth', 'collector', 'setup'}
|
||||
# the same blueprint are JWT+permission gated and ARE swept.
|
||||
EXEMPT_ENDPOINTS = {'knowledgebase.track_click', 'users.update_user',
|
||||
'geenforce.post_report',
|
||||
# Deliberately open kiosk write: decrement-only,
|
||||
# badge-attributed server-side. Decision record in
|
||||
# docs/proposals/printedparts-plugin.md; justification in
|
||||
# the plugin README.
|
||||
# Kiosk write. No JWT - the people at the bin have no
|
||||
# account - but no longer open: it needs the site's kiosk
|
||||
# flag AND the per-device token (X-Kiosk-Token), and 404s
|
||||
# without them. The badge only attributes the ledger row.
|
||||
# The old "physically rate-limited by being a touch screen"
|
||||
# justification did not survive a scanner: bare digits
|
||||
# enumerated the catalogue and the 422/200 split was an
|
||||
# oracle for real employee numbers.
|
||||
'printedparts.kiosk_take',
|
||||
# Deliberately open kiosk writes. NOT the same bar as the
|
||||
# parts kiosk, and the difference is the point: these are
|
||||
|
||||
@@ -60,6 +60,9 @@ PUBLIC_ENDPOINTS = {
|
||||
'slides.feed', 'slides.serve_image',
|
||||
|
||||
# Parts kiosk: one item by its code, and that item's picture.
|
||||
# kiosk_item takes no JWT but is NOT open: same kiosk flag + device token
|
||||
# as the write. Gating only the write would have left the enumeration
|
||||
# oracle running against a kiosk a site had switched off.
|
||||
'printedparts.kiosk_item', 'printedparts.serve_item_image',
|
||||
|
||||
# USB kiosk.
|
||||
|
||||
@@ -121,13 +121,35 @@ def test_anonymous_cannot_mutate(client, item):
|
||||
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; the kiosk stays open."""
|
||||
"""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').status_code == 200
|
||||
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):
|
||||
@@ -139,29 +161,32 @@ def test_member_without_permission_gets_403(client, member_headers, item):
|
||||
headers=member_headers).status_code == 403
|
||||
|
||||
|
||||
def test_kiosk_take_is_open_decrement_only(client, auth_headers, app, item,
|
||||
directory_employee):
|
||||
"""The kiosk endpoint needs no auth but only ever decrements stock."""
|
||||
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)
|
||||
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)
|
||||
'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)
|
||||
'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)
|
||||
'itemcode': itemcode, 'badge': '111111111', 'quantity': 1},
|
||||
headers={**auth_headers, **KIOSK_HEADERS})
|
||||
assert unknown.status_code == 422
|
||||
|
||||
with app.app_context():
|
||||
@@ -197,7 +222,7 @@ def test_lowstock_alert_fires_on_crossing_only(client, auth_headers, app, item,
|
||||
return client.post('/api/printedparts/kiosk/take',
|
||||
json={'itemcode': '3DP-9001',
|
||||
'badge': directory_employee,
|
||||
'quantity': quantity}, headers=auth_headers)
|
||||
'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
|
||||
@@ -245,7 +270,7 @@ def test_alert_recipients_merge_users_and_freetext(client, auth_headers, app,
|
||||
headers=auth_headers)
|
||||
take = client.post('/api/printedparts/kiosk/take',
|
||||
json={'itemcode': '3DP-9001',
|
||||
'badge': directory_employee, 'quantity': 6}, headers=auth_headers)
|
||||
'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']
|
||||
@@ -260,7 +285,7 @@ def test_retire_hides_and_restore_returns(client, auth_headers, item):
|
||||
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 = 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',
|
||||
@@ -270,7 +295,7 @@ def test_retire_hides_and_restore_returns(client, auth_headers, item):
|
||||
|
||||
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).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):
|
||||
@@ -335,7 +360,7 @@ def test_alert_role_members_receive(client, auth_headers, app, item,
|
||||
headers=auth_headers)
|
||||
take = client.post('/api/printedparts/kiosk/take',
|
||||
json={'itemcode': '3DP-9001',
|
||||
'badge': directory_employee, 'quantity': 6}, headers=auth_headers)
|
||||
'badge': directory_employee, 'quantity': 6}, headers={**auth_headers, **KIOSK_HEADERS})
|
||||
assert take.status_code == 200
|
||||
assert captured['to'] == ['crewone@site.test']
|
||||
|
||||
@@ -362,9 +387,9 @@ def test_gagelabtag_assigned_searched_and_kiosk_resolved(client, auth_headers):
|
||||
headers=auth_headers).get_json()['data']
|
||||
assert len(searched) == 1
|
||||
|
||||
by_tag = client.get('/api/printedparts/kiosk/item/WJRP0117', headers=auth_headers)
|
||||
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)
|
||||
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'
|
||||
|
||||
@@ -380,7 +405,7 @@ def test_kiosk_take_records_revision_from_qr(client, app, db, directory_employee
|
||||
iid = row.printeditemid
|
||||
|
||||
take = client.post('/api/printedparts/kiosk/take', json={
|
||||
'itemcode': 'WJRP7000|3', 'badge': directory_employee, 'quantity': 1}, headers=auth_headers)
|
||||
'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():
|
||||
@@ -391,7 +416,7 @@ def test_kiosk_take_records_revision_from_qr(client, app, db, directory_employee
|
||||
|
||||
# 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)
|
||||
'itemcode': 'WJRP7000', 'badge': directory_employee, 'quantity': 1}, headers={**auth_headers, **KIOSK_HEADERS})
|
||||
assert take2.status_code == 200
|
||||
with app.app_context():
|
||||
last = (PrintedItemTransaction.query
|
||||
@@ -418,7 +443,92 @@ def test_lowstock_routes_to_support_team_webhook(client, app, db, directory_empl
|
||||
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)
|
||||
'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
Block a user