ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Promote the 2026-08-30 review security fixes and supplies ordering to main

Browse Source
This commit is contained in:
Cameron Proudlock
2026-08-31 09:24:13 -04:00
parent 6680c3c5de
commit 0e3d5a6226
30 changed files with 991 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
tests/test_core/test_auth_ratelimit.py
View File

@@ -32,11 +32,18 @@ def ratelimit_on(app):
def _bad_login(client, ip):
"""Attempt a login for a non-existent user from a given source IP."""
"""Attempt a login for a non-existent user from a given source IP.
The caller is set through REMOTE_ADDR, the peer that actually connected,
because that is what the limiter keys on. It used to be sent as
X-Forwarded-For, which meant these tests changed identity using the very
header a real attacker would rotate to escape the limit - so the suite
could not have told the difference.
"""
return client.post(
'/api/auth/login',
json={'username': 'ghost', 'password': 'wrong'},
headers={'X-Forwarded-For': ip},
environ_base={'REMOTE_ADDR': ip},
)

98
tests/test_core/test_clientip.py
View File

@@ -1,4 +1,12 @@
"""strip_port / client_ip normalization (ARR forwards clientip:port)."""
"""client_ip: which address the app treats as the caller, and port normalization.
The answer is remote_addr. X-Forwarded-For is written by the caller, and the
login rate limit, the account lockout and the audit log all key on this value,
so honoring the header let a caller rotate it for an unbounded budget and an
untraceable trail. Where a proxy really is in front, TRUSTED_PROXY_COUNT turns
on ProxyFix and the header arrives already resolved into remote_addr - ports
included, because ARR forwards clientip:port.
"""
from shopdb.utils.clientip import strip_port, client_ip
@@ -34,12 +42,94 @@ class _FakeRequest:
self.remote_addr = remote_addr
def test_client_ip_prefers_forwarded_first_hop_stripped():
request_obj = _FakeRequest(headers={'X-Forwarded-For': '10.0.0.5:51000, 10.0.0.1'},
remote_addr='127.0.0.1')
def test_client_ip_ignores_the_forwarded_header():
"""The caller sends X-Forwarded-For, so it cannot decide who the caller is.
This test used to assert the opposite. The login rate limit, the account
lockout and the audit log all key on client_ip, so honoring the header let
a caller rotate it for an unbounded budget and an untraceable trail. Where
a proxy really is in front, TRUSTED_PROXY_COUNT turns on ProxyFix and the
header reaches us already resolved into remote_addr.
"""
request_obj = _FakeRequest(
headers={'X-Forwarded-For': '10.0.0.5:51000, 10.0.0.1'},
remote_addr='127.0.0.1')
assert client_ip(request_obj) == '127.0.0.1'
def test_client_ip_strips_a_port_proxyfix_passed_through():
"""ARR writes clientip:port into the header and ProxyFix copies it verbatim
into remote_addr, so the port still has to come off."""
request_obj = _FakeRequest(headers={}, remote_addr='10.0.0.5:51000')
assert client_ip(request_obj) == '10.0.0.5'
def test_client_ip_falls_back_to_remote_addr_stripped():
request_obj = _FakeRequest(headers={}, remote_addr='192.168.1.9:60000')
assert client_ip(request_obj) == '192.168.1.9'
# ---------------------------------------------------------------------------
# Through a real request, and through the login limiter that keys on this.
# ---------------------------------------------------------------------------
def test_the_forwarded_header_does_not_decide_the_client_ip(app):
from shopdb.utils.clientip import client_ip
with app.test_request_context(
'/', headers={'X-Forwarded-For': '10.0.0.99'},
environ_base={'REMOTE_ADDR': '203.0.113.7'}):
from flask import request
assert client_ip(request) == '203.0.113.7'
def test_the_peer_address_decides_the_client_ip(app):
from shopdb.utils.clientip import client_ip
with app.test_request_context(
'/', environ_base={'REMOTE_ADDR': '203.0.113.8'}):
from flask import request
assert client_ip(request) == '203.0.113.8'
def test_a_forwarded_port_is_still_stripped(app):
"""IIS ARR writes "clientip:port" and ProxyFix passes it through verbatim,
so the port has to come off remote_addr or every request looks like a new
client."""
from shopdb.utils.clientip import client_ip
with app.test_request_context(
'/', environ_base={'REMOTE_ADDR': '203.0.113.9:54321'}):
from flask import request
assert client_ip(request) == '203.0.113.9'
def test_rotating_the_forwarded_header_does_not_reset_the_login_limiter(app,
client,
db):
"""The budget follows the connection, not a header the caller writes.
Sprayed across many usernames on purpose: the per-account lockout bounds
guessing at ONE account, so the IP limiter is the only thing standing
between a caller and unlimited attempts spread across the directory. That
is what rotating the header used to defeat.
"""
caller = {'REMOTE_ADDR': '203.0.113.50'}
previous = dict(app.config)
app.config['AUTH_RATELIMIT_ENABLED'] = True
app.config['AUTH_RATELIMIT_MAX'] = 3
try:
seen = set()
for hop in range(10):
response = client.post(
'/api/auth/login',
headers={'X-Forwarded-For': '198.51.100.{}'.format(hop)},
environ_base=caller,
json={'username': 'nobody{}'.format(hop), 'password': 'wrong'})
seen.add(response.status_code)
if response.status_code == 429:
break
finally:
app.config['AUTH_RATELIMIT_ENABLED'] = previous.get(
'AUTH_RATELIMIT_ENABLED', False)
app.config['AUTH_RATELIMIT_MAX'] = previous.get('AUTH_RATELIMIT_MAX', 30)
assert 429 in seen, 'rotating X-Forwarded-For bought an unbounded budget'

61
tests/test_core/test_deactivated_user.py Normal file
View File

@@ -0,0 +1,61 @@
"""Deactivating an account ends its sessions.
There is no token denylist in this product, so the JWT user loader is the only
place a deactivated account can be stopped. It did a bare lookup by id, which
meant deactivating somebody left their access token working until it expired and
their refresh token minting new ones for thirty days. /auth/refresh and the API
token path already checked isactive; the loader did not.
"""
from werkzeug.security import generate_password_hash
def _login(client, username='deactivateme', password='testpass'):
response = client.post('/api/auth/login',
json={'username': username, 'password': password})
assert response.status_code == 200, response.get_json()
return response.get_json()['data']
def _make_user(db, username='deactivateme'):
from shopdb.core.models import User
user = User(
username=username,
email=f'{username}@test.local',
passwordhash=generate_password_hash('testpass'),
)
db.session.add(user)
db.session.commit()
return user
def test_a_live_token_stops_working_once_the_account_is_deactivated(client, db):
user = _make_user(db)
tokens = _login(client)
headers = {'Authorization': 'Bearer {}'.format(tokens['access_token'])}
assert client.get('/api/auth/me', headers=headers).status_code == 200
user.isactive = False
db.session.commit()
assert client.get('/api/auth/me', headers=headers).status_code == 401
def test_a_deactivated_account_cannot_refresh(client, db):
user = _make_user(db, 'refreshme')
tokens = _login(client, 'refreshme')
headers = {'Authorization': 'Bearer {}'.format(tokens['refresh_token'])}
user.isactive = False
db.session.commit()
assert client.post('/api/auth/refresh', headers=headers).status_code == 401
def test_an_active_account_is_unaffected(client, db):
"""The loader must not reject everybody; isactive defaults true."""
_make_user(db, 'stillhere')
tokens = _login(client, 'stillhere')
headers = {'Authorization': 'Bearer {}'.format(tokens['access_token'])}
assert client.get('/api/auth/me', headers=headers).status_code == 200

58
tests/test_core/test_guest_role.py
View File

@@ -98,12 +98,39 @@ def test_a_site_with_no_guest_role_behaves_as_it_always_did(client, db):
# The guardrail
# ---------------------------------------------------------------------------
def test_a_view_permission_is_grantable(app):
def test_a_declared_permission_is_grantable(app):
from shopdb.core.models import guest_grantable_permissions
with app.app_context():
grantable = guest_grantable_permissions()
assert 'assets.view' in grantable
assert 'reports.view' in grantable
assert 'applications.view' in grantable
def test_a_view_permission_is_not_grantable_merely_for_being_a_view(app):
"""The suffix stopped deciding. These are all '.view' and none is declared.
settings.view gates every setting row; audit.view the before/after of every
change; users.view nothing at all. Each was anonymous-grantable purely
because of how it was spelled.
assets.view is deliberately NOT in this list. It was, while global search
answered twelve domains behind it; now that search gates each domain on its
own permission the grant reaches assets alone, and the floor needs it to
look an asset number up without signing in.
"""
from shopdb.core.models import guest_grantable_permissions
with app.app_context():
grantable = guest_grantable_permissions()
for name in ('settings.view', 'audit.view', 'users.view'):
assert name not in grantable, name
def test_the_bytes_behind_applications_are_not_grantable(app):
"""applications.view is grantable only because the bytes moved off it."""
from shopdb.core.models import guest_grantable_permissions
with app.app_context():
grantable = guest_grantable_permissions()
assert 'applications.download' not in grantable
def test_a_write_permission_is_not_grantable(app):
@@ -127,13 +154,29 @@ def test_the_api_refuses_to_grant_a_guest_a_write(client, db, auth_headers,
assert guestrole.permissions == []
def test_the_api_allows_granting_a_guest_a_view(client, db, auth_headers,
guestrole):
def test_the_api_allows_granting_a_guest_a_declared_read(client, db,
auth_headers,
guestrole):
response = client.put('/api/users/roles/{}'.format(guestrole.roleid),
headers=auth_headers,
json={'permissions': ['reports.view']})
assert response.status_code == 200
assert [p.name for p in guestrole.permissions] == ['reports.view']
def test_the_api_refuses_an_undeclared_read(client, db, auth_headers,
guestrole):
"""This test used to grant guests audit.view and pass, which is the bug.
audit.view ends in '.view' and gates usernames, addresses and the
before/after of every change in the database.
"""
response = client.put('/api/users/roles/{}'.format(guestrole.roleid),
headers=auth_headers,
json={'permissions': ['audit.view']})
assert response.status_code == 200
assert [p.name for p in guestrole.permissions] == ['audit.view']
assert response.status_code == 400
assert 'audit.view' in response.get_json()['data']['error']['message']
assert guestrole.permissions == []
def test_the_guest_role_cannot_be_deleted(client, auth_headers, guestrole):
@@ -155,8 +198,9 @@ def test_the_role_list_says_what_a_guest_may_hold(client, auth_headers,
rows = client.get('/api/users/roles', headers=auth_headers).get_json()['data']
guest = next(r for r in rows if r['rolename'] == 'guest')
assert guest['isguest'] is True
assert 'assets.view' in guest['grantable']
assert 'reports.view' in guest['grantable']
assert 'assets.delete' not in guest['grantable']
assert 'audit.view' not in guest['grantable']
other = next((r for r in rows if r['rolename'] != 'guest'), None)
if other is not None:

94
tests/test_core/test_search_domain_permissions.py Normal file
View File

@@ -0,0 +1,94 @@
"""Global search answers each domain only to a caller who holds ITS permission.
The route is gated on assets.view, and that used to be the entire check. One
permission then queried twelve domains, so anyone who could look up an asset
number also read out of the staff directory, the USB checkout history, the
notification log and the network inventory.
That is why assets.view could not be handed to guests, which cost the floor its
search. Gating each domain separately is what makes the grant safe again: it
reaches assets and nothing else.
"""
from unittest.mock import patch
import pytest
from shopdb.core.models import Permission, Role
def _grant_guest(db, *names):
guest = Role.guest()
guest.permissions = [p for p in Permission.query.filter(
Permission.name.in_(names)).all()]
db.session.commit()
@pytest.fixture
def guestrole(db):
"""The seeded guest role, holding nothing."""
Permission.seed()
Role.seedguest()
db.session.commit()
return Role.guest()
def _search(client, term='a'):
response = client.get('/api/search?q={}'.format(term))
assert response.status_code == 200, response.get_json()
return response.get_json()['data']
FAKE_EMPLOYEE = [{
'First_Name': 'Ada', 'Last_Name': 'Smith', 'SSO': 508123,
'Team': 'Machining', 'Role': 'Operator',
}]
def _with_employee_directory():
"""The HR directory is an external database and is unreachable in tests, so
the domain silently returned nothing and a leak test over it proved nothing.
Patched here so there is something real to leak."""
return patch('shopdb.core.api.search._employee_rows',
return_value=FAKE_EMPLOYEE)
def test_assets_view_alone_does_not_reach_the_staff_directory(client, db,
guestrole):
"""The finding this exists for: ?q=smith returned employee records to a
signed-out visitor holding only assets.view."""
_grant_guest(db, 'assets.view')
with _with_employee_directory():
data = _search(client, 'smith')
assert 'employee' not in {row['type'] for row in data['results']}
def test_employees_view_does_reach_it(client, db, guestrole):
"""The other direction, so the gate is not passing by refusing everything."""
_grant_guest(db, 'assets.view', 'employees.view')
with _with_employee_directory():
data = _search(client, 'smith')
employees = [row for row in data['results'] if row['type'] == 'employee']
assert len(employees) == 1
assert employees[0]['title'] == 'Ada Smith'
def test_search_still_answers_a_guest_with_assets_view(client, db, guestrole):
"""The regression that prompted this: guests lost search entirely.
assets.view left the grantable set because of the leak above, which took
the floor's asset lookup with it.
"""
_grant_guest(db, 'assets.view')
response = client.get('/api/search?q=press')
assert response.status_code == 200
def test_a_guest_with_nothing_granted_is_refused(client, db, guestrole):
_grant_guest(db)
assert client.get('/api/search?q=press').status_code in (401, 403)
def test_assets_view_is_grantable_again(app):
from shopdb.core.models import guest_grantable_permissions
with app.app_context():
assert 'assets.view' in guest_grantable_permissions()