"""How the USB kiosk decides a caller is admitted, and what that costs. The kiosk writes without a login, so everything here is about the two things standing in for one: where the caller is on the network, and a device token. Three properties this file holds down, each of which was missing: - a caller cannot choose its own address. X-Forwarded-For is a request header; unless the site has said where its proxies are, it is ignored. - a range that admits everybody cannot be saved by accident. - the device token does not travel in a URL, and the badge-to-name answer cannot be walked by whoever is admitted. """ import pytest from shopdb.api import cache from shopdb.core.models import Setting PLANT = '10.134.48.17' PLANT_CIDR = '10.134.48.0/23' ELSEWHERE = '10.9.100.5' # The reverse proxy sits ON the plant network, which is the normal place for it # and the reason the old fail-closed claim was false: its own address matched # the allowlist. A proxy parked outside the range, as this file used to have, # proves nothing. PROXY = '10.134.48.5' PROXY_CIDR = '10.134.48.5/32' KIOSK_TOKEN = 'usb-kiosk-admission-token' @pytest.fixture def kiosk(db): """Self-hosted USB, kiosk ON, a token set, one employee, one drive.""" from plugins.employees.models import DirectoryEmployee from plugins.usb.models import USBDevice # Session-scoped app means one SimpleCache for the whole run, and the badge # limiter counts in it. Start every test with an empty budget. cache.clear() db.session.add(Setting(key='usb_directory_mode', value='selfhosted', valuetype='string', category='site')) db.session.add(Setting(key='usb_kiosk_enabled', value='true', valuetype='boolean', category='usb')) db.session.add(Setting(key='usb_kiosk_token', value=KIOSK_TOKEN, valuetype='string', category='usb')) db.session.add(DirectoryEmployee(sso=270015376, firstname='Dana', lastname='Reyes')) db.session.add(USBDevice(serialnumber='USB-1', label='Blue Kingston', ischeckedout=False)) db.session.commit() return db def _set(db, key, value, valuetype='string'): row = Setting.query.filter_by(key=key).first() if row is None: db.session.add(Setting(key=key, value=value, valuetype=valuetype, category='usb')) else: row.value = value db.session.commit() def _session(client, address, badge='270015376', **headers): return client.post('/api/usb/kiosk/session', json={'badge': badge}, environ_base={'REMOTE_ADDR': address}, headers=headers) # -- where the caller is ------------------------------------------------------ # The stack that ships resolves this BEFORE Flask: IIS -> HttpPlatformHandler -> # waitress on loopback with --trusted-proxy=127.0.0.1 --trusted-proxy-count=1, # which takes the last X-Forwarded-For hop into REMOTE_ADDR. So REMOTE_ADDR in # these tests is what that layer decided, and the header beside it is what the # app is required to ignore. def test_the_app_admits_on_the_address_the_wsgi_layer_resolved(kiosk, client): """Production shape: waitress already promoted the caller into REMOTE_ADDR.""" _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) assert _session(client, PLANT).status_code == 200 def test_a_caller_cannot_forward_itself_into_the_allowlist(kiosk, client): """THE FIX. The forwarding header is not evidence of anything here. Without it, anyone who could reach the app sent 'X-Forwarded-For: ' and was admitted to an unauthenticated write from anywhere on the network. """ _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) assert _session(client, ELSEWHERE, **{'X-Forwarded-For': PLANT}).status_code == 404 def test_the_header_cannot_take_an_admitted_caller_away_either(kiosk, client): """Read in either direction, the header is simply not consulted. Stated as its own case because a half-measure that only ignored the header when it would have ADMITTED would still leave the app parsing it, and the next person would wire the other branch back up. """ _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) assert _session(client, PLANT, **{'X-Forwarded-For': ELSEWHERE}).status_code == 200 def test_declaring_a_proxy_does_not_turn_header_reading_back_on(kiosk, client): """usb_kiosk_trusted_proxies subtracts addresses; it never adds a source. The version this replaced read X-Forwarded-For once this setting named the peer. On the Windows stack the peer is never the proxy - waitress replaced it with what the proxy forwarded - so that branch could not fire in production, and anywhere it could it handed the header back its authority. """ _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) _set(kiosk, 'usb_kiosk_trusted_proxies', '10.20.30.40/32') assert _session(client, '10.20.30.40', **{'X-Forwarded-For': PLANT}).status_code == 404 def test_a_request_wearing_the_proxy_address_fails_closed(kiosk, client): """THE FIX, and the case the old "fails closed" test did not cover. A proxy on the plant network forwards nothing - it is misconfigured, or something bypassed it - so the request arrives as the proxy's own address, which is inside usb_kiosk_allowed_cidrs. It carries no evidence about who sent it and must be refused, not admitted. """ _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) _set(kiosk, 'usb_kiosk_trusted_proxies', PROXY_CIDR) assert _session(client, PROXY).status_code == 404 # And a caller on the same allowlisted network is still admitted, or the # subtraction has taken the feature with it. assert _session(client, PLANT).status_code == 200 def test_the_proxy_is_refused_even_carrying_a_forwarding_header(kiosk, client): """Its own address is what arrived, whatever it says about somebody else.""" _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) _set(kiosk, 'usb_kiosk_trusted_proxies', PROXY_CIDR) assert _session(client, PROXY, **{'X-Forwarded-For': PLANT}).status_code == 404 def test_loopback_is_never_a_kiosk_client(kiosk, client): """The shipped Windows default with no X-Forwarded-For at all. HttpPlatformHandler connects from 127.0.0.1, waitress has nothing to promote, and every caller then wears the app server's own address. A site whose allowlist happens to cover loopback must not turn that into admission for everyone who can reach the server. """ _set(kiosk, 'usb_kiosk_allowed_cidrs', '127.0.0.0/8, ' + PLANT_CIDR) assert _session(client, '127.0.0.1').status_code == 404 assert _session(client, PLANT).status_code == 200 def test_a_proxy_address_is_refused_even_when_nobody_declared_it(kiosk, client): """Honest limit, written down. Loopback is refused with no setting at all; any OTHER proxy address is indistinguishable from a PC on that network, so the site has to name it. This test is what says so out loud.""" _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) assert _session(client, PROXY).status_code == 200 _set(kiosk, 'usb_kiosk_trusted_proxies', PROXY_CIDR) assert _session(client, PROXY).status_code == 404 def test_the_token_still_works_from_a_proxy_address(kiosk, client): """Subtracting proxies is about admission BY NETWORK. A screen presenting its device token has said who it is and is not affected.""" _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) _set(kiosk, 'usb_kiosk_trusted_proxies', PROXY_CIDR) assert _session(client, PROXY, **{'X-Kiosk-Token': KIOSK_TOKEN}).status_code == 200 def test_todays_deployment_is_unchanged(kiosk, client): """No trusted proxy set - the default - still admits by connecting peer.""" _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) assert _session(client, PLANT).status_code == 200 # -- what may be saved in an allowlist --------------------------------------- @pytest.mark.parametrize('value, refused', [ ('', False), ('192.0.2.0/24', False), (' 10.134.48.0/23 , 10.48.249.0/26 ', False), ('0.0.0.0/0', True), ('::/0', True), ('10.0.0.0/8', True), ('192.0.2.0/24, 0.0.0.0/0', True), ('not-a-cidr', True), ]) def test_kiosk_cidr_problem(value, refused): from plugins.usb.api.routes import kiosk_cidr_problem assert (kiosk_cidr_problem(value) is not None) is refused def test_a_wide_range_is_allowed_when_a_site_says_so_deliberately(): from plugins.usb.api.routes import kiosk_cidr_problem assert kiosk_cidr_problem('10.0.0.0/8', allowwide=True) is None # Never, though: the override is about a big site, not about no bound. assert kiosk_cidr_problem('0.0.0.0/0', allowwide=True) is not None def test_saving_everything_is_refused_with_a_reason(kiosk, client, auth_headers): """THE FIX, at the place the administrator is standing.""" response = client.put('/api/settings/usb_kiosk_allowed_cidrs', json={'value': '0.0.0.0/0'}, headers=auth_headers) assert response.status_code == 400 message = response.get_json()['data']['error']['message'] assert 'without a login' in message assert Setting.query.filter_by(key='usb_kiosk_allowed_cidrs').first() is None def test_a_sane_range_still_saves(kiosk, client, auth_headers): response = client.put('/api/settings/usb_kiosk_allowed_cidrs', json={'value': PLANT_CIDR}, headers=auth_headers) assert response.status_code == 200 assert Setting.get('usb_kiosk_allowed_cidrs') == PLANT_CIDR def test_the_override_lets_a_flat_site_save_its_range(kiosk, client, auth_headers): assert client.put('/api/settings/usb_kiosk_allowed_cidrs', json={'value': '10.0.0.0/8'}, headers=auth_headers).status_code == 400 _set(kiosk, 'usb_kiosk_allow_wide_cidrs', 'true', valuetype='boolean') assert client.put('/api/settings/usb_kiosk_allowed_cidrs', json={'value': '10.0.0.0/8'}, headers=auth_headers).status_code == 200 def test_an_override_turned_off_stays_off(kiosk, client, auth_headers): """Stored as a string, 'false' is a truthy Python value. It must not be one.""" _set(kiosk, 'usb_kiosk_allow_wide_cidrs', 'false') assert client.put('/api/settings/usb_kiosk_allowed_cidrs', json={'value': '10.0.0.0/8'}, headers=auth_headers).status_code == 400 def test_the_trusted_proxy_list_is_validated_too(kiosk, client, auth_headers): """A proxy list that matches everything would trust every caller's header.""" assert client.put('/api/settings/usb_kiosk_trusted_proxies', json={'value': '0.0.0.0/0'}, headers=auth_headers).status_code == 400 def test_other_settings_are_not_touched_by_the_guard(kiosk, client, auth_headers): """The before_request hook runs on every save; it may only judge its own.""" response = client.put('/api/settings/site_name', json={'value': '0.0.0.0/0'}, headers=auth_headers) assert response.status_code == 200 # -- who may find out what the rules are -------------------------------------- # The validator runs in a before_request, which is ahead of the view and so # ahead of @require_permission. That ordering answered an anonymous caller 400 # with the reason a range was refused, while an anonymous caller who happened to # send a sane range fell through to 401 - the difference being a free oracle on # the allowlist rules. def test_an_anonymous_probe_is_refused_before_the_value_is_judged(kiosk, client): """THE FIX. Authentication decides first.""" refused = client.put('/api/settings/usb_kiosk_allowed_cidrs', json={'value': '0.0.0.0/0'}) assert refused.status_code in (401, 403) assert 'without a login' not in refused.get_data(as_text=True) def test_the_probe_cannot_tell_a_bad_range_from_a_good_one(kiosk, client): """Both answers must be the same one, or the oracle is still there.""" bad = client.put('/api/settings/usb_kiosk_allowed_cidrs', json={'value': '0.0.0.0/0'}) good = client.put('/api/settings/usb_kiosk_allowed_cidrs', json={'value': PLANT_CIDR}) assert bad.status_code == good.status_code assert (bad.get_json()['data']['error']['code'] == good.get_json()['data']['error']['code']) assert Setting.query.filter_by(key='usb_kiosk_allowed_cidrs').first() is None # -- the badge budget is a setting, not a config file ------------------------- # It was app-config only, which on an air-gapped server means editing # instance/config.py and restarting IIS. In practice that is no off switch at # all, and the only lever anybody had was turning the whole kiosk off. def test_the_badge_limiter_settings_are_declared(kiosk): """Declared where the app seeds defaults, or nothing creates the rows.""" from plugins.usb.plugin import USBPlugin declared = {entry['key']: entry for entry in USBPlugin().get_settings_defaults()} assert declared['usb_kiosk_badge_limit_enabled']['value'] == 'true' assert declared['usb_kiosk_badge_limit_enabled']['valuetype'] == 'boolean' assert declared['usb_kiosk_badge_miss_max']['value'] == '15' assert declared['usb_kiosk_badge_miss_max']['valuetype'] == 'integer' def test_the_budget_setting_moves_the_bound(kiosk, client, app): """A site dials the number down and the kiosk stops there.""" from plugins.usb.api.routes import KIOSK_BADGE_MISS_MAX _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) _set(kiosk, KIOSK_BADGE_MISS_MAX, '2', valuetype='integer') # The app-config override the suite normally uses has to be out of the way, # or it would answer instead of the setting. previous = app.config.pop('USB_KIOSK_BADGE_MISS_MAX', None) app.config['USB_KIOSK_BADGE_LIMIT_ENABLED'] = True try: for _ in range(2): assert _session(client, PLANT, badge='111111111').status_code == 422 assert _session(client, PLANT, badge='111111111').status_code == 429 finally: app.config.pop('USB_KIOSK_BADGE_LIMIT_ENABLED', None) if previous is not None: app.config['USB_KIOSK_BADGE_MISS_MAX'] = previous def test_a_nonsense_budget_falls_back_to_the_shipped_one(kiosk, app): """Zero is not "no limit" here - it would refuse the first scan of every window - and a settings page can store a blank or a typo just as easily.""" from plugins.usb.api.routes import (BADGE_MISS_MAX, KIOSK_BADGE_MISS_MAX, _badge_miss_budget) previous = app.config.pop('USB_KIOSK_BADGE_MISS_MAX', None) try: for stored in ('', '0', '-4', 'lots'): _set(kiosk, KIOSK_BADGE_MISS_MAX, stored, valuetype='integer') with app.test_request_context('/'): assert _badge_miss_budget()[1] == BADGE_MISS_MAX finally: if previous is not None: app.config['USB_KIOSK_BADGE_MISS_MAX'] = previous def test_the_kill_switch_is_reachable_without_editing_a_config_file(kiosk, app): """And it defaults to today's behaviour: on. Off under TESTING is an arrangement for this suite, not the default, so the first assertion here is also what stops "the limiter is off" quietly becoming true everywhere. """ from plugins.usb.api.routes import (KIOSK_BADGE_LIMIT_ENABLED, _badge_limit_enabled) previous = app.config.pop('USB_KIOSK_BADGE_LIMIT_ENABLED', None) try: with app.test_request_context('/'): app.config['TESTING'] = False try: assert _badge_limit_enabled() is True _set(kiosk, KIOSK_BADGE_LIMIT_ENABLED, 'false', valuetype='boolean') assert _badge_limit_enabled() is False finally: app.config['TESTING'] = True finally: if previous is not None: app.config['USB_KIOSK_BADGE_LIMIT_ENABLED'] = previous # -- provisioning a screen ---------------------------------------------------- def test_the_setup_link_does_not_carry_the_token(kiosk, client, auth_headers): """THE FIX. A query string ends up in history, referrers and proxy logs.""" response = client.get('/api/usb/kiosk/provisioning-link', headers=auth_headers) assert response.status_code == 200 url = response.get_json()['data']['url'] assert KIOSK_TOKEN not in url assert url.startswith('/api/usb/kiosk/setup?code=') def test_opening_the_setup_link_provisions_that_browser(kiosk, client, auth_headers): """One open on the screen, and its later calls are admitted with no token.""" url = client.get('/api/usb/kiosk/provisioning-link', headers=auth_headers).get_json()['data']['url'] redirected = client.get(url) assert redirected.status_code == 302 assert redirected.headers['Location'].endswith('/usb-kiosk') cookie = redirected.headers['Set-Cookie'] assert 'usbkiosktoken=' + KIOSK_TOKEN in cookie assert 'HttpOnly' in cookie # No X-Kiosk-Token header, no allowlist: the cookie is the whole credential. assert _session(client, ELSEWHERE).status_code == 200 def test_a_setup_code_is_good_once(kiosk, client, auth_headers): url = client.get('/api/usb/kiosk/provisioning-link', headers=auth_headers).get_json()['data']['url'] assert client.get(url).status_code == 302 second = client.get(url) assert second.status_code == 410 assert 'Generate' in second.get_json()['data']['error']['message'] def test_a_made_up_setup_code_provisions_nothing(kiosk, client): response = client.get('/api/usb/kiosk/setup?code=not-a-real-code') assert response.status_code == 410 assert 'Set-Cookie' not in response.headers def test_an_expired_setup_code_is_dead(kiosk, client, auth_headers): from plugins.usb.api.routes import KIOSK_SETUP_CODE url = client.get('/api/usb/kiosk/provisioning-link', headers=auth_headers).get_json()['data']['url'] pending = Setting.get(KIOSK_SETUP_CODE) _set(kiosk, KIOSK_SETUP_CODE, pending.split('|')[0] + '|1') assert client.get(url).status_code == 410 def test_the_old_header_still_admits_screens_already_on_the_floor(kiosk, client): """Screens provisioned before this change must not need a visit.""" assert _session(client, ELSEWHERE, **{'X-Kiosk-Token': KIOSK_TOKEN}).status_code == 200 # -- the badge-to-name answer ------------------------------------------------- @pytest.fixture def smallbudget(app): """Three misses, so the test states the rule rather than counting to 15. Also turns the limiter on: it defaults off under TESTING, because the counter lives in a cache that outlives a test and is shared across whatever else runs on the same worker. Same arrangement as the login limiter. """ previous = {key: app.config.get(key) for key in ('USB_KIOSK_BADGE_MISS_MAX', 'USB_KIOSK_BADGE_LIMIT_ENABLED')} app.config['USB_KIOSK_BADGE_MISS_MAX'] = 3 app.config['USB_KIOSK_BADGE_LIMIT_ENABLED'] = True yield for key, value in previous.items(): if value is None: app.config.pop(key, None) else: app.config[key] = value def test_badge_guessing_is_bounded_per_source(kiosk, client, smallbudget): """THE FIX. The kiosk answers with a NAME, so it is a lookup to be walked. Admission by network position can mean a whole VLAN, not one screen by the cabinet, which is what makes an unbounded lookup worth somebody's loop. """ _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) for _ in range(3): assert _session(client, PLANT, badge='111111111').status_code == 422 refused = _session(client, PLANT, badge='111111111') assert refused.status_code == 429 # And a badge that WOULD resolve gets nothing either, or the bound is a # speed bump with a hole in it. assert _session(client, PLANT).status_code == 429 def test_the_bound_is_per_caller_not_per_kiosk(kiosk, client, smallbudget): """One PC guessing must not shut the cabinet screen down.""" _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) for _ in range(4): _session(client, PLANT, badge='111111111') assert _session(client, PLANT).status_code == 429 assert _session(client, '10.134.48.99').status_code == 200 def test_ordinary_use_never_spends_the_budget(kiosk, client, smallbudget): """Only a badge this site has never heard of counts.""" _set(kiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR) for _ in range(6): assert _session(client, PLANT).status_code == 200 def test_enabling_the_plugin_on_a_running_server_still_works(): """init_app also runs when an operator switches the plugin on live. Flask refuses a setup method once the app has served a request, and the loader turns any exception out of init_app into a plugin that failed to load - so an unguarded before_request would have made "enable USB" fail on every running server. """ from flask import Flask from plugins.usb.plugin import USBPlugin running = Flask(__name__) running.test_client().get('/') USBPlugin().init_app(running, None)