read permissions on every GET
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