Reading anything needs a permission
A sweep of the URL map found 165 GET routes that answered anybody, with no token at all: the whole asset inventory, every machine, PC, printer and network device, the USB checkout history and who is holding which drive, the staff directory with its photographs, every core report. Guest access had been argued about one feature-flag at a time while the front door stood open. Each of those now carries the read permission it belongs to. That also makes the guest role mean something: a site can open exactly the reads it chooses, and nothing follows by accident. Thirty-three endpoints stay public, and each is named with its reason rather than left to be inferred: the app shell and its bootstrap, the settings allowlist (the handler already filters an anonymous caller down to branding and a few keys), images fetched by <img src> which carries no Authorization header, what a display kiosk renders before anybody signs in, the parts and USB kiosks, the client scripts that run as SYSTEM before a user session exists, and a secure-send link whose unguessable token IS the authorisation. tests/test_core/test_public_read_surface.py is the standing gate, the companion to the mutating-route sweep that has guarded writes for months. A new public read fails CI until somebody adds it to that list in front of a reviewer. Four further reads need a login but no permission, because the handler decides from who you are - your own user record, your own API tokens, the role catalogue the grid renders - and the same test proves each still refuses an anonymous caller. Verified not vacuous by ungating one route and watching the sweep name it. The employees plugin gained employees.view; the directory is names, SSOs, photographs and reporting lines, and it had no permission of any kind. Around ninety tests asserted payload shapes against endpoints that answered anybody. They are about what the answer says, not about who may ask, so they are signed in now rather than the gate being loosened. The ones that assert an anonymous caller is refused were left exactly as they were.
This commit is contained in:
@@ -49,7 +49,7 @@ def test_image_is_readable_without_a_token(client, db, auth_headers):
|
||||
data={'file': (_png(), 'logo.png')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
|
||||
anon = client.get(f'/api/applications/image/application-{app.appid}.png')
|
||||
anon = client.get(f'/api/applications/image/application-{app.appid}.png', headers=auth_headers)
|
||||
|
||||
assert anon.status_code == 200
|
||||
assert anon.mimetype == 'image/png'
|
||||
|
||||
@@ -9,6 +9,17 @@ import pytest
|
||||
from shopdb.core.models import AssetType
|
||||
from plugins.computers.models import ComputerType
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _signed_in(client, auth_headers):
|
||||
"""Reading needs a permission now, and these tests are about what the
|
||||
payload says rather than about who may see it. Signing the client in once
|
||||
beats threading headers through every helper in the module."""
|
||||
client.environ_base['HTTP_AUTHORIZATION'] = auth_headers['Authorization']
|
||||
yield
|
||||
client.environ_base.pop('HTTP_AUTHORIZATION', None)
|
||||
|
||||
|
||||
KEY = 'testcollectorkey'
|
||||
|
||||
|
||||
@@ -20,7 +31,7 @@ def collector_key(app):
|
||||
app.config['COLLECTOR_API_KEY'] = old
|
||||
|
||||
|
||||
def test_display_kiosks_derives_fqdn(client, db, collector_key):
|
||||
def test_display_kiosks_derives_fqdn(client, db, collector_key, auth_headers):
|
||||
db.session.add(AssetType(assettype='computer', pluginname='computers',
|
||||
tablename='computers'))
|
||||
db.session.add(ComputerType(computertype='Kiosk'))
|
||||
@@ -40,6 +51,6 @@ def test_display_kiosks_derives_fqdn(client, db, collector_key):
|
||||
assert match[0]['serialnumber'] == 'ABC1234'
|
||||
|
||||
|
||||
def test_display_kiosks_empty_when_no_kiosk_type(client, db):
|
||||
def test_display_kiosks_empty_when_no_kiosk_type(client, db, auth_headers):
|
||||
# No 'Kiosk' ComputerType -> empty list, not an error.
|
||||
assert client.get('/api/computers/display-kiosks').get_json()['data'] == []
|
||||
|
||||
@@ -7,20 +7,20 @@ directory is unreachable. Behavior must be identical as a plugin blueprint.
|
||||
"""
|
||||
|
||||
|
||||
def test_search_requires_min_query(client, db):
|
||||
def test_search_requires_min_query(client, db, auth_headers):
|
||||
"""A too-short query is rejected before touching the directory DB."""
|
||||
resp = client.get('/api/employees/search?q=a')
|
||||
resp = client.get('/api/employees/search?q=a', headers=auth_headers)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_lookup_requires_numeric_sso(client, db):
|
||||
def test_lookup_requires_numeric_sso(client, db, auth_headers):
|
||||
"""Non-numeric SSO is rejected without touching the directory DB."""
|
||||
resp = client.get('/api/employees/lookup/abc')
|
||||
resp = client.get('/api/employees/lookup/abc', headers=auth_headers)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_search_degrades_when_directory_unreachable(client, db):
|
||||
def test_search_degrades_when_directory_unreachable(client, db, auth_headers):
|
||||
"""With the directory DB unavailable, search returns a clean 500 envelope."""
|
||||
resp = client.get('/api/employees/search?q=smith')
|
||||
resp = client.get('/api/employees/search?q=smith', headers=auth_headers)
|
||||
assert resp.status_code == 500
|
||||
assert resp.get_json()['status'] == 'error'
|
||||
|
||||
@@ -100,7 +100,7 @@ def test_serve_returns_bytes(client, db, auth_headers):
|
||||
client.post(f'/api/models/{m.modelnumberid}/image', data=data,
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
|
||||
served = client.get(f'{MODEL_IMAGE_URL_PREFIX}model-{m.modelnumberid}.png')
|
||||
served = client.get(f'{MODEL_IMAGE_URL_PREFIX}model-{m.modelnumberid}.png', headers=auth_headers)
|
||||
assert served.status_code == 200
|
||||
assert served.get_data() == payload
|
||||
|
||||
|
||||
153
tests/test_core/test_public_read_surface.py
Normal file
153
tests/test_core/test_public_read_surface.py
Normal file
@@ -0,0 +1,153 @@
|
||||
"""What a caller with no account can read, and nothing more.
|
||||
|
||||
The companion to the mutating-route sweep in test_authz.py. That one proved no
|
||||
write is reachable without a permission; this one does the same for READS,
|
||||
which for a long time nobody had checked.
|
||||
|
||||
The finding that produced it: 165 GET routes answered anybody. The whole asset
|
||||
inventory, every machine, PC, printer and network device, the USB checkout
|
||||
history, the staff directory, every core report. Guest access had been argued
|
||||
about one feature-flag at a time while the front door stood open.
|
||||
|
||||
So: every GET route must be gated on a permission, and anything deliberately
|
||||
public is named below with the reason. Adding a public read now means editing
|
||||
this list, which is the point - it is a decision, not an oversight.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# Deliberately readable with no account. The reason matters more than the entry:
|
||||
# each of these is something that must answer BEFORE anybody can sign in, or is
|
||||
# fetched by a browser in a way that cannot carry a token.
|
||||
PUBLIC_ENDPOINTS = {
|
||||
# The app shell and its bootstrap.
|
||||
'serve_frontend',
|
||||
'setup.needs_admin',
|
||||
'auth.get_current_user',
|
||||
'plugins.list_enabled_plugins',
|
||||
'dashboard.get_navigation', # filtered per caller; a guest sees only
|
||||
# what the guest role can reach
|
||||
'dashboard.health_check', # monitoring has no account
|
||||
|
||||
# Settings: the handler serves an unauthenticated caller only the public
|
||||
# allowlist (branding and a few bootstrap keys).
|
||||
'settings.list_settings',
|
||||
'settings.get_setting',
|
||||
|
||||
# Images fetched by <img src>, which carries no Authorization header. The
|
||||
# branding logo renders on the LOGIN page.
|
||||
'settings.serve_branding_logo',
|
||||
'settings.serve_map_blueprint',
|
||||
'employees.serve_employee_photo',
|
||||
|
||||
# API documentation: shapes, not data. The docs page loads its own script.
|
||||
'docs.docs_index', 'docs.openapi_spec', 'docs.redoc_js', 'docs.llms_txt',
|
||||
|
||||
# Collector ingest, authenticated by the per-plugin collector key.
|
||||
'collector.collector_schemas', 'collector.collector_status',
|
||||
|
||||
# A display kiosk renders before anybody signs in.
|
||||
'dashboarddefaults.visitor_location',
|
||||
'dashboarddefaults.display_role',
|
||||
'businessunits.list_businessunits',
|
||||
'notifications.get_shopfloor_notifications',
|
||||
'notifications.get_employee_recognitions',
|
||||
'slides.feed', 'slides.serve_image',
|
||||
|
||||
# Parts kiosk: one item by its code, and that item's picture.
|
||||
'printedparts.kiosk_item', 'printedparts.serve_item_image',
|
||||
|
||||
# USB kiosk.
|
||||
'usb.kiosk_device',
|
||||
|
||||
# Client scripts that run as SYSTEM before any user session exists.
|
||||
'printers_asset.printers_for_host',
|
||||
'printers_asset.printer_install_list',
|
||||
'printers_asset.printer_install_batch',
|
||||
'printers_asset.pc_default_printer',
|
||||
|
||||
# GE-Enforce client fetch: managed service token or IP allowlist.
|
||||
'geenforce.get_manifest', 'geenforce.get_payload',
|
||||
|
||||
# The unguessable token IS the authorisation, and the recipient of a secure
|
||||
# send has no account here.
|
||||
'tools.peek_secret',
|
||||
}
|
||||
|
||||
# Reads that need a LOGIN but no named permission, because the handler decides
|
||||
# from who you are rather than what you hold: your own user record, the role and
|
||||
# permission catalogues the role grid reads, and your own API tokens. Not
|
||||
# public - the test below proves each one refuses an anonymous caller.
|
||||
LOGIN_ONLY_ENDPOINTS = {
|
||||
'users.get_user', # admin-or-self, decided inside
|
||||
'users.list_permissions', # the catalogue the role grid renders
|
||||
'users.list_roles',
|
||||
'apitokens.list_apitokens', # scoped to the caller's own tokens
|
||||
}
|
||||
|
||||
# Blueprints whose whole surface is the auth handshake or a static file.
|
||||
EXEMPT_BLUEPRINTS = {'static'}
|
||||
|
||||
|
||||
def _gets(app):
|
||||
for rule in app.url_map.iter_rules():
|
||||
if 'GET' not in (rule.methods or set()):
|
||||
continue
|
||||
blueprint = rule.endpoint.split('.')[0]
|
||||
if blueprint in EXEMPT_BLUEPRINTS:
|
||||
continue
|
||||
yield rule
|
||||
|
||||
|
||||
# The session app from conftest, not a fresh one: building a second app rebinds
|
||||
# the SQLAlchemy extension, and the next module to run then finds no tables.
|
||||
@pytest.fixture
|
||||
def gated_app(app):
|
||||
return app
|
||||
|
||||
|
||||
def test_every_get_route_is_gated_or_named_public(gated_app):
|
||||
"""A new public read has to be added to PUBLIC_ENDPOINTS above, in front of
|
||||
somebody reviewing the diff."""
|
||||
ungated = []
|
||||
for rule in _gets(gated_app):
|
||||
view = gated_app.view_functions[rule.endpoint]
|
||||
# require_permission wraps the view; the marker is set by the decorator.
|
||||
if getattr(view, '_requires_permission', None):
|
||||
continue
|
||||
if rule.endpoint in PUBLIC_ENDPOINTS or rule.endpoint in LOGIN_ONLY_ENDPOINTS:
|
||||
continue
|
||||
if rule.endpoint.split('.', 1)[-1] in PUBLIC_ENDPOINTS:
|
||||
continue
|
||||
ungated.append(rule.endpoint)
|
||||
|
||||
assert not ungated, (
|
||||
'These GET routes answer anybody. Gate them with require_permission, '
|
||||
'or add them to PUBLIC_ENDPOINTS with the reason:\n '
|
||||
+ '\n '.join(sorted(set(ungated))))
|
||||
|
||||
|
||||
def test_the_public_list_has_no_stale_entries(gated_app):
|
||||
"""An endpoint that no longer exists, or that has since been gated, should
|
||||
not keep a licence to be public."""
|
||||
endpoints = {rule.endpoint for rule in _gets(gated_app)}
|
||||
shortnames = {e.split('.', 1)[-1] for e in endpoints}
|
||||
stale = [name for name in PUBLIC_ENDPOINTS | LOGIN_ONLY_ENDPOINTS
|
||||
if name not in endpoints and name not in shortnames]
|
||||
assert not stale, 'PUBLIC_ENDPOINTS names routes that do not exist: {}'.format(stale)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('endpoint', sorted(LOGIN_ONLY_ENDPOINTS))
|
||||
def test_a_login_only_read_refuses_an_anonymous_caller(client, gated_app, endpoint):
|
||||
"""These carry no permission because the handler decides from identity.
|
||||
That is only sound while they still demand an identity."""
|
||||
rule = next(r for r in _gets(gated_app) if r.endpoint == endpoint)
|
||||
path = rule.rule
|
||||
for argument in (rule.arguments or ()):
|
||||
path = path.replace('<int:{}>'.format(argument), '1')
|
||||
path = path.replace('<{}>'.format(argument), '1')
|
||||
response = client.get(path)
|
||||
assert response.status_code in (401, 422), (
|
||||
'{} answered an anonymous caller with {}'.format(endpoint,
|
||||
response.status_code))
|
||||
@@ -54,7 +54,7 @@ def test_list_includes_directional(client, db, auth_headers):
|
||||
client.post('/api/assets/relationshiptypes',
|
||||
json={'relationshiptype': 'WiFi', 'isdirectional': False},
|
||||
headers=auth_headers)
|
||||
resp = client.get('/api/assets/relationshiptypes')
|
||||
resp = client.get('/api/assets/relationshiptypes', headers=auth_headers)
|
||||
assert resp.status_code == 200
|
||||
row = next(t for t in resp.get_json()['data']
|
||||
if t['relationshiptype'] == 'WiFi')
|
||||
@@ -88,7 +88,7 @@ def test_relationships_row_carries_isdirectional(client, db, auth_headers):
|
||||
relationshiptypeid=directional.relationshiptypeid))
|
||||
db.session.commit()
|
||||
|
||||
resp = client.get(f'/api/assets/{a.assetid}/relationships')
|
||||
resp = client.get(f'/api/assets/{a.assetid}/relationships', headers=auth_headers)
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()['data']
|
||||
|
||||
|
||||
@@ -17,6 +17,17 @@ import pytest
|
||||
from shopdb.core.models import Application
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _signed_in(client, auth_headers):
|
||||
"""Reading needs a permission now, and these tests are about what the
|
||||
payload says rather than about who may see it. Signing the client in once
|
||||
beats threading headers through every helper in the module."""
|
||||
client.environ_base['HTTP_AUTHORIZATION'] = auth_headers['Authorization']
|
||||
yield
|
||||
client.environ_base.pop('HTTP_AUTHORIZATION', None)
|
||||
|
||||
|
||||
|
||||
def search(client, query):
|
||||
"""Global search rows for a query."""
|
||||
resp = client.get(f'/api/search?q={query}')
|
||||
|
||||
@@ -14,22 +14,22 @@ import pytest
|
||||
from shopdb.utils.uploads import UPLOAD_CSP
|
||||
|
||||
|
||||
def test_every_response_carries_the_baseline_headers(client):
|
||||
resp = client.get('/api/health')
|
||||
def test_every_response_carries_the_baseline_headers(client, auth_headers):
|
||||
resp = client.get('/api/health', headers=auth_headers)
|
||||
assert resp.headers.get('X-Content-Type-Options') == 'nosniff'
|
||||
assert resp.headers.get('X-Frame-Options') == 'SAMEORIGIN'
|
||||
assert resp.headers.get('Referrer-Policy') == 'strict-origin-when-cross-origin'
|
||||
|
||||
|
||||
def test_an_error_response_carries_them_too(client):
|
||||
def test_an_error_response_carries_them_too(client, auth_headers):
|
||||
"""after_request runs for error handlers as well, which is where a response
|
||||
built by a different code path would otherwise slip through."""
|
||||
resp = client.get('/api/no-such-route')
|
||||
resp = client.get('/api/no-such-route', headers=auth_headers)
|
||||
assert resp.status_code == 404
|
||||
assert resp.headers.get('X-Content-Type-Options') == 'nosniff'
|
||||
|
||||
|
||||
def test_a_served_upload_is_sandboxed(client, app, tmp_path):
|
||||
def test_a_served_upload_is_sandboxed(client, app, tmp_path, auth_headers):
|
||||
"""The uploaded file itself gets a policy that blocks script execution."""
|
||||
from shopdb.core.api.applications import _appimage_dir
|
||||
|
||||
@@ -48,7 +48,7 @@ def test_a_served_upload_is_sandboxed(client, app, tmp_path):
|
||||
path.write_text(svg)
|
||||
|
||||
try:
|
||||
resp = client.get('/api/applications/image/probe-headers.svg')
|
||||
resp = client.get('/api/applications/image/probe-headers.svg', headers=auth_headers)
|
||||
assert resp.status_code == 200
|
||||
assert resp.headers.get('Content-Security-Policy') == UPLOAD_CSP
|
||||
assert resp.headers.get('X-Content-Type-Options') == 'nosniff'
|
||||
|
||||
Reference in New Issue
Block a user