Merge remote-tracking branch 'origin/wj' into wilm

This commit is contained in:
benjamin.baker
2026-09-18 17:25:24 -04:00
64 changed files with 35788 additions and 20614 deletions

View File

@@ -1,20 +1,26 @@
"""Every route the app serves is described in docs/api-inventory.json.
The inventory is hand-maintained, and deliberately so: its value is the prose -
what each endpoint is FOR, what it accepts, a worked example - and none of that
can be derived from the code. An audit of all 372 documented operations found
zero phantom routes and zero wrong parameter names, so the maintenance itself is
in good order.
The inventory has two halves. Its STRUCTURE - surface, method, path and the auth
each route actually enforces - is derived from the live url_map and the
decorators on each view by `gen_openapi.py --refresh`. Its PROSE - what an
endpoint is FOR, what it accepts, a worked example - is human, and a refresh
never overwrites words that are already there.
What it cannot do is notice a route somebody added. Twenty-two were serving
traffic undocumented, including an entire plugin's surface and every one of the
dashboard card endpoints - each of them added after its surface was written.
This module is the coverage half of that, and it is the half a generator cannot
do for you: a refresh describes whatever is served at the moment it runs, so
without a gate the file rots the moment the next route lands.
So the prose stays hand-written and the COVERAGE is enforced here. Adding an
endpoint now means describing it, in the same commit, which is when the author
still knows what it is for.
It had. On this branch the inventory was not tracked at all, so the skipif below
fired, this module reported green while checking nothing, and 24 routes went to
production documented nowhere - including the whole endpoint-agent delivery
surface and both kiosks. scripts/check-gates-ran.sh now asserts this module RAN,
in every checkout rather than only in one with wiki prose in it.
So: add a route, run `gen_openapi.py --refresh`, and replace the seeded purpose
with the reason the route exists - which is while the author still knows it.
"""
import importlib.util
import json
import re
from pathlib import Path
@@ -93,3 +99,87 @@ def test_nothing_documented_has_been_removed():
assert not stale, (
'%d documented route(s) are no longer served:\n %s'
% (len(stale), '\n '.join('%s %s' % row for row in stale)))
# ---------------------------------------------------------------------------
# The other half: the auth string, which nothing gated until now
# ---------------------------------------------------------------------------
#
# The two tests above check that a route EXISTS in both directions. Nothing
# checked what the entry SAYS about it, so a wrong or stale auth string passed
# every gate in the repo - and auth is the field a site reads to decide what its
# guest role and its firewall may reach, and the field the published spec turns
# into a machine-readable security requirement.
#
# It had rotted exactly that way: nine operations documented as needing no
# credential while the handler required one, and a kiosk route documented as
# ungated when its gate is a single-use setup code. Both survived a refresh,
# because a refresh only rewrites; nothing compared.
#
# So this re-derives auth from the live app with the generator's own classifier
# and fails on any disagreement with the committed file. The derivation is
# authoritative: hand-editing an auth string in the inventory does not survive
# the next --refresh, so the fix for a wrong one is always to teach the sweep in
# scripts/gen_openapi.py, or to gate the view.
def _generator():
"""scripts/gen_openapi.py, loaded by path.
It is a script, not a package module, so there is nothing to import. Loading
it runs its imports and nothing else - it builds no app until it is asked
to.
"""
path = REPO / 'scripts' / 'gen_openapi.py'
spec = importlib.util.spec_from_file_location('gen_openapi', path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def derived_auth():
"""{(METHOD, path): auth} as the classifier reads the code right now."""
generator = _generator()
app = create_app('testing')
reviewed, blueprints = generator._reviewed_exceptions()
derived = {}
for rule, methods, endpoint, view in generator._served_routes(app):
auth = generator._auth_for(view, endpoint, reviewed, blueprints)
for method in methods:
derived[(method, normalise(rule))] = auth
return derived
def documented_auth():
"""{(METHOD, path): auth} as docs/api-inventory.json has it committed."""
documented = {}
for surface in json.loads(INVENTORY.read_text()):
for entry in surface.get('endpoints', []):
path = normalise(entry.get('path') or '')
for method in re.split(r'[/,|]', entry.get('method') or 'GET'):
method = method.strip().upper()
if method:
documented[(method, path)] = entry.get('auth') or ''
return documented
def test_documented_auth_is_what_the_code_enforces():
"""A stale auth string is a lie about who may call a route.
Only routes present on both sides are compared. A route missing from either
is a coverage fault, and the two tests above already name it; reporting it
twice here would bury this one.
"""
derived = derived_auth()
documented = documented_auth()
disagreements = []
for key in sorted(set(derived) & set(documented)):
if derived[key] != documented[key]:
disagreements.append(
'%s %s\n documented: %s\n enforced: %s'
% (key[0], key[1], documented[key] or '(blank)', derived[key]))
assert not disagreements, (
'%d route(s) document an auth that is not what the code enforces. The '
'inventory is derived, so fix the code or teach the classifier in '
'scripts/gen_openapi.py, then re-run it with --refresh:\n %s'
% (len(disagreements), '\n '.join(disagreements[:15])))

View File

@@ -0,0 +1,143 @@
"""The legacy core collector routes and the shared key.
/pc, /apps, /heartbeat and /bulk name their target PC in the request body, so
the shared collector key is an any-host write key on them and no per-PC
credential narrows it. collector.legacysharedkeyrefused closes them to the
shared key. It defaults OFF so nothing on an un-migrated fleet changes, and the
shared key is logged either way so a site can see what depends on it first.
"""
import pytest
from shopdb.core.api.collector import (
LEGACY_SHARED_KEY_SETTING,
LEGACY_SHARED_KEY_SETTING_DEFAULT,
)
KEY = 'testlegacykey'
LEGACY_POSTS = [
('/api/collector/pc', {'hostname': 'WJLEG01'}),
('/api/collector/apps', {'hostname': 'WJLEG01',
'apps': [{'appname': 'Nothing'}]}),
('/api/collector/heartbeat', {'hostname': 'WJLEG01'}),
('/api/collector/bulk', {'pcs': [{'hostname': 'WJLEG01'}]}),
]
@pytest.fixture
def collector_key(app):
old = app.config.get('COLLECTOR_API_KEY')
app.config['COLLECTOR_API_KEY'] = KEY
yield KEY
app.config['COLLECTOR_API_KEY'] = old
@pytest.fixture
def legacy_closed(db):
"""Turn the closure on for the duration of a test."""
from shopdb.core.models.setting import Setting
Setting.set(LEGACY_SHARED_KEY_SETTING, True, valuetype='boolean',
category='collector')
yield
Setting.query.filter_by(key=LEGACY_SHARED_KEY_SETTING).delete()
db.session.commit()
def _mint_collector_token(client, auth_headers):
resp = client.post('/api/apitokens',
json={'name': 'collector svc',
'scopes': ['collector.ingest']},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
return resp.get_json()['data']['secret']
@pytest.mark.parametrize('path,payload', LEGACY_POSTS)
def test_shared_key_still_works_by_default(client, db, collector_key,
path, payload):
"""No setting row: today's behaviour. The routes answer the shared key.
404/400 is fine - the point is that authorization passed, not that the
hostname exists.
"""
resp = client.post(path, json=payload, headers={'X-API-Key': KEY})
assert resp.status_code != 401, resp.get_json()
@pytest.mark.parametrize('path,payload', LEGACY_POSTS)
def test_shared_key_refused_when_closed(client, db, collector_key,
legacy_closed, path, payload):
"""Setting on: the shared key is refused the way the endpoints plugin
refuses it, before any write happens."""
resp = client.post(path, json=payload, headers={'X-API-Key': KEY})
assert resp.status_code == 401
assert resp.get_json()['data']['error']['code'] == 'SHARED_KEY_REFUSED'
def test_managed_token_unaffected_by_closure(client, db, auth_headers,
legacy_closed):
"""The closure targets the shared key only. A collector-scoped managed
token keeps working on a legacy route."""
secret = _mint_collector_token(client, auth_headers)
resp = client.post('/api/collector/heartbeat',
json={'hostname': 'WJLEG01'},
headers={'X-API-Key': secret})
assert resp.status_code == 200, resp.get_json()
def test_status_stays_open_when_closed(client, db, collector_key,
legacy_closed):
"""/status writes nothing, so the closure does not touch it - it stays a
usable connectivity check for a fleet still on the shared key."""
resp = client.get('/api/collector/status', headers={'X-API-Key': KEY})
assert resp.status_code == 200
def test_generic_collector_unaffected_by_closure(client, db, auth_headers,
collector_key, legacy_closed):
"""The generic per-plugin route has its own shared-key rules (per-PC
credentials). This setting must not reach it."""
from shopdb.core.models import AssetType
db.session.add(AssetType(assettype='computer', pluginname='computers',
tablename='computers', description='PCs'))
db.session.commit()
resp = client.post('/api/collector/computers',
json={'hostname': 'WJLEG02'},
headers={'X-API-Key': KEY})
assert resp.status_code == 200, resp.get_json()
def test_shared_key_use_is_logged(client, db, collector_key, caplog):
"""The warning is what a site reads before flipping the setting, so it has
to be emitted while the routes are still open."""
with caplog.at_level('WARNING'):
client.post('/api/collector/heartbeat', json={'hostname': 'WJLEG01'},
headers={'X-API-Key': KEY})
assert any('Legacy collector route used with the shared key' in r.message
for r in caplog.records)
def test_managed_token_use_is_not_logged(client, db, auth_headers, caplog):
"""A managed token is the supported credential; logging it would bury the
shared-key warnings the site is looking for."""
secret = _mint_collector_token(client, auth_headers)
with caplog.at_level('WARNING'):
client.post('/api/collector/heartbeat', json={'hostname': 'WJLEG01'},
headers={'X-API-Key': secret})
assert not any('used with the shared key' in r.message
for r in caplog.records)
def test_setting_default_is_todays_behaviour():
"""The declared default must leave the routes open: a site mid-migration
has un-enrolled PCs and the shared key is all they have."""
assert LEGACY_SHARED_KEY_SETTING_DEFAULT['key'] == LEGACY_SHARED_KEY_SETTING
assert LEGACY_SHARED_KEY_SETTING_DEFAULT['value'] == 'false'
assert LEGACY_SHARED_KEY_SETTING_DEFAULT['valuetype'] == 'boolean'
description = LEGACY_SHARED_KEY_SETTING_DEFAULT['description']
for route in ('/pc', '/apps', '/heartbeat', '/bulk'):
assert route in description

View File

@@ -86,6 +86,13 @@ PUBLIC_ENDPOINTS = {
# USB kiosk.
'usb.kiosk_device',
# The screen's one-time provisioning hop. It carries no JWT by design: it is
# opened on a wall screen that has never signed in, and it exists precisely
# so the kiosk token stops travelling in a URL. What it accepts is a
# single-use setup code that is redeemed and burned on first use, so it is
# open to anybody the way a password-reset link is open to anybody - the
# secret is the code, not the route.
'usb.kiosk_setup',
# Client scripts that run as SYSTEM before any user session exists.
'printers_asset.printers_for_host',

View File

@@ -0,0 +1,228 @@
"""A switch nobody can flip is not a feature.
Three security switches shipped readable by their enforcement code but with no
declared default and no control, so on an air-gapped box they could only be
changed by editing the database by hand. These tests pin the two halves that
make one real: the default is SEEDED (so the row exists and the settings API
can render it), and a PUT through the public API actually changes behaviour.
collector.legacysharedkeyrefused is the one proven end to end here, because it
is core and its enforcement is a plain route refusal.
Also pinned: the upsert in update_setting must not give an UNDECLARED key
valuetype 'string' when the caller sent JSON true. That was how
plugin.endpoints.bootstrapquerytoken became a string row whose readers, doing
bool(Setting.get(key, True)), then saw the non-empty text 'false' as True.
"""
import pytest
from shopdb.core.api.collector import (
LEGACY_SHARED_KEY_SETTING,
LEGACY_SHARED_KEY_SETTING_DEFAULT,
)
from shopdb.core.api.settings import build_default_settings
from shopdb.core.models import Setting
KEY = 'testlegacykey'
LEGACY_URL = '/api/collector/heartbeat'
LEGACY_BODY = {'hostname': 'WJSW01'}
@pytest.fixture
def collector_key(app):
old = app.config.get('COLLECTOR_API_KEY')
app.config['COLLECTOR_API_KEY'] = KEY
yield KEY
app.config['COLLECTOR_API_KEY'] = old
def _defaults():
return {entry['key']: entry for entry in build_default_settings()}
def test_collector_switch_is_declared_in_the_seed_list():
"""Seeding is what creates the row; without it the toggle has nothing to
read and the UI has no control to render."""
entry = _defaults().get(LEGACY_SHARED_KEY_SETTING)
assert entry is not None, 'collector.legacysharedkeyrefused is not seeded'
assert entry is LEGACY_SHARED_KEY_SETTING_DEFAULT, \
'the seed list must reuse the declaration next to the enforcing code'
# Today's behaviour: the legacy routes stay open to the shared key.
assert entry['value'] == 'false'
assert entry['valuetype'] == 'boolean'
assert entry['category'] == 'collector'
def test_seeding_creates_the_row_as_a_real_boolean(client, db, auth_headers):
resp = client.post('/api/settings/seed', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
row = Setting.query.filter_by(key=LEGACY_SHARED_KEY_SETTING).one()
assert row.valuetype == 'boolean'
assert row.get_typed_value() is False
def test_the_seeded_switch_is_listed_by_the_settings_api(client, db,
auth_headers):
"""The settings hub control reads this list, so the key has to be in it."""
client.post('/api/settings/seed', headers=auth_headers)
resp = client.get('/api/settings', headers=auth_headers)
assert resp.status_code == 200
items = {item['key']: item for item in resp.get_json()['data']}
assert LEGACY_SHARED_KEY_SETTING in items
assert items[LEGACY_SHARED_KEY_SETTING]['value'] is False
assert items[LEGACY_SHARED_KEY_SETTING]['valuetype'] == 'boolean'
def test_put_flips_the_switch_and_the_legacy_route_closes(client, db,
auth_headers,
collector_key):
"""End to end through the public API: default on, PUT, behaviour changes."""
client.post('/api/settings/seed', headers=auth_headers)
# Default: the shared key still works on a legacy route.
assert client.post(LEGACY_URL, json=LEGACY_BODY,
headers={'X-API-Key': KEY}).status_code == 200
flip = client.put('/api/settings/' + LEGACY_SHARED_KEY_SETTING,
json={'value': True}, headers=auth_headers)
assert flip.status_code == 200, flip.get_json()
assert flip.get_json()['data']['value'] is True
refused = client.post(LEGACY_URL, json=LEGACY_BODY,
headers={'X-API-Key': KEY})
assert refused.status_code == 401, refused.get_json()
assert refused.get_json()['data']['error']['code'] == 'SHARED_KEY_REFUSED'
# And back off again, or an operator could not undo a mistake.
client.put('/api/settings/' + LEGACY_SHARED_KEY_SETTING,
json={'value': False}, headers=auth_headers)
assert client.post(LEGACY_URL, json=LEGACY_BODY,
headers={'X-API-Key': KEY}).status_code == 200
def test_an_undeclared_boolean_key_is_not_born_a_string(client, db,
auth_headers):
"""The trap that made bootstrapquerytoken misbehave.
JSON true on a key nobody declares used to create valuetype 'string', and
every boolean reader then got the text 'true'/'false' - both truthy.
"""
key = 'testundeclaredswitch'
resp = client.put('/api/settings/' + key, json={'value': False},
headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
row = Setting.query.filter_by(key=key).one()
assert row.valuetype == 'boolean'
assert row.get_typed_value() is False
assert bool(Setting.get(key, True)) is False
def test_an_undeclared_number_key_is_born_a_number(client, db, auth_headers):
key = 'testundeclaredcount'
client.put('/api/settings/' + key, json={'value': 25},
headers=auth_headers)
row = Setting.query.filter_by(key=key).one()
assert row.valuetype == 'integer'
assert row.get_typed_value() == 25
# And the type is now enforced on the next write, which is the point of
# recording it: a typo cannot silently become 0 for every reader.
bad = client.put('/api/settings/' + key, json={'value': 'twenty-five'},
headers=auth_headers)
assert bad.status_code == 400
def test_an_undeclared_string_that_reads_true_stays_a_string(client, db,
auth_headers):
"""Inference reads the JSON type, never the text. A string setting whose
value happens to be 'true' must not be coerced into a boolean row."""
key = 'testundeclaredlabel'
client.put('/api/settings/' + key, json={'value': 'true'},
headers=auth_headers)
row = Setting.query.filter_by(key=key).one()
assert row.valuetype == 'string'
assert row.get_typed_value() == 'true'
def test_a_declared_key_keeps_its_declared_type(client, db, auth_headers):
"""Declaration wins over inference: the declaration is the contract every
reader was written against."""
client.put('/api/settings/' + LEGACY_SHARED_KEY_SETTING,
json={'value': 'true'}, headers=auth_headers)
row = Setting.query.filter_by(key=LEGACY_SHARED_KEY_SETTING).one()
assert row.valuetype == 'boolean'
assert row.category == 'collector'
assert row.get_typed_value() is True
# --- masking is keyed on the NAME, so it has to know the type ----------------
#
# A settings page cannot render a checkbox it is not allowed to read. The
# secret mask is a name heuristic ('password'/'token'/'secret' anywhere in the
# key), and plugin.endpoints.bootstrapquerytoken is a true/false switch whose
# name contains 'token': the API answered '********' and the control could not
# tell on from off.
def test_a_boolean_named_like_a_secret_is_not_masked(client, db, auth_headers):
key = 'plugin.endpoints.bootstrapquerytoken'
client.put('/api/settings/' + key, json={'value': False},
headers=auth_headers)
resp = client.get('/api/settings/' + key, headers=auth_headers)
assert resp.get_json()['data']['value'] is False
def test_a_string_secret_is_still_masked(client, db, auth_headers):
"""The narrowing must not open the real secrets: they are string rows."""
client.put('/api/settings/zabbix_token', json={'value': 'supersecret'},
headers=auth_headers)
resp = client.get('/api/settings/zabbix_token', headers=auth_headers)
assert resp.get_json()['data']['value'] == '********'
listed = client.get('/api/settings', headers=auth_headers).get_json()['data']
body = client.get('/api/settings', headers=auth_headers).get_data(as_text=True)
assert any(item['key'] == 'zabbix_token' for item in listed)
assert 'supersecret' not in body
def test_the_mask_still_means_unchanged_for_a_string_secret(client, db,
auth_headers):
client.put('/api/settings/zabbix_token', json={'value': 'supersecret'},
headers=auth_headers)
client.put('/api/settings/zabbix_token', json={'value': '********'},
headers=auth_headers)
assert Setting.get('zabbix_token') == 'supersecret'
# --- the control itself ------------------------------------------------------
#
# Seeding alone renders nothing: this UI has no generic key/value settings
# editor, so a core key with no page is still SQL-only on an air-gapped box.
def test_the_security_page_carries_the_collector_toggle():
from pathlib import Path
repo = Path(__file__).resolve().parents[2]
page = (repo / 'frontend' / 'src' / 'views' / 'settings'
/ 'AuthenticationSettings.vue').read_text(encoding='utf-8')
assert LEGACY_SHARED_KEY_SETTING in page, 'no control for the switch'
assert 'toggleLegacySharedKey' in page
# It has to name what stops reporting, or turning it on is a guess.
assert 'Report-AssetToShopDB.ps1' in page
assert 'Report-PrintersToShopDB.ps1' in page

View File

@@ -34,6 +34,14 @@ DOCS = REPO / 'docs'
# published mirror - keying off the directory made this module run there against
# a prose-less tree and fail. A published checkout is one with no authored
# prose, matching the canonical definition in scripts/check-gates-ran.sh.
# KNOWN GAP, left deliberately for a decision rather than closed here: this
# keys on docs/*.md, so prose that lives only in docs/wiki-drafts/ does NOT
# turn the module on, and on gh-wj that is every page there is. Pointed at the
# drafts on 2026-09-18 the scrub fails on 27 lines - site name, internal git
# server, developer home paths - which is a question about what those drafts
# are for (internal wiki, or publication) and not a bug to fix by rewriting
# them. scripts/check-gates-ran.sh matches this definition exactly; change both
# together or the gate runner reports a skip it was told to expect.
pytestmark = pytest.mark.skipif(
not any(DOCS.glob('*.md')),
reason='no docs/ prose in this checkout - it is excluded from publication and lives in the wiki')

View File

@@ -0,0 +1,295 @@
"""Holes in `scripts/check-naming-and-style.sh`, each held open by a test.
The naming gate must read the column names migrations CREATE, not just models.
`scripts/check-naming-and-style.sh` checked `db.Column` attribute names on
models and nothing else, while `migrations/versions/` is filtered out of the
file list every other check in that script uses. A column is born in a
migration, so from the day the gate was written no created column name was ever
looked at: a model attribute reading fine says nothing about what the database
actually got.
Two tests, and the second is the one that matters. A gate that prints nothing
looks identical whether it is clean or broken, and this one was silently
passing everything for its whole life, so the test plants a violation and
requires the gate to fail on it. Without that, a future edit that breaks the
regex would leave every test here green.
"""
import re
import shutil
import subprocess
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parent.parent
GATE = REPO_ROOT / 'scripts' / 'check-naming-and-style.sh'
# Same shapes the gate matches: the name literal of a created column, positional
# or keyword, on the sa.Column( line or on the line after it, plus the target of
# a rename.
COLUMN_LITERAL = re.compile(r"""sa\.Column\(\s*['"]([^'"]+)['"]""", re.S)
COLUMN_KEYWORD = re.compile(
r"""sa\.Column\(\s*name\s*=\s*['"]([^'"]+)['"]""", re.S)
RENAME_LITERAL = re.compile(r"""new_column_name\s*=\s*['"]([^'"]+)['"]""")
# THE RULE, not a proxy for it. The first version of this file asked whether the
# name contained an underscore, which is the same mistake the gate's own regex
# made: 'Monitor_Name' has one but was missed for its leading capital, and a
# CamelCase name with no underscore at all is just as wrong and has nothing to
# find. Lowercase concatenated means lowercase concatenated.
LOWERCASE_CONCATENATED = re.compile(r'[a-z][a-z0-9]*\Z')
PLANTED = '''"""Planted by the test suite. Never applied."""
import sqlalchemy as sa
from alembic import op
revision = 'plantedprobe'
down_revision = None
def upgrade():
op.add_column('widgets', sa.Column('planted_column', sa.Integer(),
nullable=True))
def downgrade():
op.drop_column('widgets', 'planted_column')
'''
# ASSEMBLED FROM FRAGMENTS, like the gate's own patterns and
# tests/test_docs_publishable.py: written literally, a test for the site-literal
# rule is itself a site literal, and both the gate and the publication scrub
# would find it here.
PLANTED_SITE_HOST = 'tsg' 'wp00525' '.geaero' 'space.net'
PLANTED_AGENT = (
'// Planted by the test suite.\n'
'internal static class Probe\n'
'{\n'
' public const string DefaultBaseUrl = "https://'
+ PLANTED_SITE_HOST + '/shopdb";\n'
'}\n'
)
def _migration_files():
found = list((REPO_ROOT / 'migrations' / 'versions').glob('*.py'))
found += list(REPO_ROOT.glob('plugins/*/migrations/versions/*.py'))
return sorted(found)
def test_there_are_migrations_to_check():
"""Guard the guard: a glob that matches nothing passes everything below."""
assert _migration_files(), 'no migrations found - has the layout moved?'
def test_no_migration_creates_a_badly_named_column():
offenders = []
for path in _migration_files():
text = path.read_text(encoding='utf-8')
for pattern in (COLUMN_LITERAL, COLUMN_KEYWORD, RENAME_LITERAL):
for name in pattern.findall(text):
if not LOWERCASE_CONCATENATED.match(name):
offenders.append('{0}: {1}'.format(
path.relative_to(REPO_ROOT), name))
assert not offenders, (
'column names are lowercase concatenated (CONTRIBUTING.md):\n '
+ '\n '.join(offenders))
@pytest.mark.skipif(shutil.which('bash') is None, reason='needs bash')
def test_the_gate_fails_on_a_planted_migration_column(tmp_path):
"""Plant one, and require the gate to say so.
Run against a throwaway tree rather than the repository, so the planted
file cannot be left behind by a failing assert. tmp_path is outside any
checkout, so the script's git file list is unavailable and it takes its
own find fallback - which is the branch a developer without git metadata
gets, and so is worth exercising.
"""
versions = tmp_path / 'migrations' / 'versions'
versions.mkdir(parents=True)
gate = tmp_path / 'scripts' / 'check-naming-and-style.sh'
gate.parent.mkdir(parents=True)
shutil.copy(GATE, gate)
planted = versions / 'zz01_planted.py'
planted.write_text(PLANTED, encoding='utf-8')
dirty = subprocess.run(['bash', str(gate)], cwd=str(tmp_path),
capture_output=True, text=True)
assert dirty.returncode == 1, (
'the gate passed a snake_case migration column:\n' + dirty.stdout)
assert 'planted_column' in dirty.stdout
planted.unlink()
clean = subprocess.run(['bash', str(gate)], cwd=str(tmp_path),
capture_output=True, text=True)
assert clean.returncode == 0, (
'the gate fails with nothing planted:\n' + clean.stdout)
@pytest.mark.skipif(shutil.which('bash') is None, reason='needs bash')
def test_the_gate_fails_on_a_site_literal_in_the_agent(tmp_path):
"""agent/ is in the ADR-015 scan, and has to stay in it.
The C# agent is compiled once and installed on every bay at every site. A
site hostname, base URL or pin baked into a .cs file there ships to the next
site inside a binary, where no reviewer reads it - and agent/ was not one of
the directories the site-literal scan looked at.
"""
gate = tmp_path / 'scripts' / 'check-naming-and-style.sh'
gate.parent.mkdir(parents=True)
shutil.copy(GATE, gate)
agent = tmp_path / 'agent' / 'Probe'
agent.mkdir(parents=True)
planted = agent / 'Probe.cs'
planted.write_text(PLANTED_AGENT, encoding='utf-8')
dirty = subprocess.run(['bash', str(gate)], cwd=str(tmp_path),
capture_output=True, text=True)
assert dirty.returncode == 1, (
'the gate passed a site hostname in the agent:\n' + dirty.stdout)
assert 'Probe.cs' in dirty.stdout
planted.unlink()
clean = subprocess.run(['bash', str(gate)], cwd=str(tmp_path),
capture_output=True, text=True)
assert clean.returncode == 0, (
'the gate fails with nothing planted:\n' + clean.stdout)
# Every shape a migration writes a column name in. The gate's first regex asked
# for ['"][a-z0-9]*_ inside the quotes, so only the first of these was ever
# caught: a leading capital, a keyword argument or a CamelCase name with no
# underscore all walked past it. One case per shape, named so a failure says
# which shape regressed.
PLANTED_COLUMN_SHAPES = {
'positional_snake': (
" op.add_column('widgets', sa.Column('planted_column',\n"
" sa.Integer()))\n", 'planted_column'),
'positional_leading_capital': (
" op.add_column('widgets', sa.Column('Monitor_Name',\n"
" sa.String(50)))\n", 'Monitor_Name'),
'positional_camelcase_no_underscore': (
" op.add_column('widgets', sa.Column('MonitorName', sa.String(50)))\n",
'MonitorName'),
'keyword_name': (
" op.add_column('widgets', sa.Column(name='kw_form',\n"
" type_=sa.String()))\n", 'kw_form'),
'wrapped_after_open_paren': (
" op.add_column('widgets', sa.Column(\n"
" 'wrapped_name', sa.Integer()))\n", 'wrapped_name'),
'alter_column_rename': (
" op.alter_column('widgets', 'oldname',\n"
" new_column_name='New_Name')\n", 'New_Name'),
'create_table_second_column': (
" op.create_table('t2', sa.Column('goodone', sa.Integer()),\n"
" sa.Column('Bad_Two', sa.Integer()))\n", 'Bad_Two'),
'batch_alter_table_add': (
" with op.batch_alter_table('widgets') as batch_op:\n"
" batch_op.add_column(sa.Column('batchCol', sa.Integer()))\n",
'batchCol'),
'batch_alter_table_rename': (
" with op.batch_alter_table('widgets') as batch_op:\n"
" batch_op.alter_column('a', new_column_name='b_c')\n", 'b_c'),
}
PLANTED_CLEAN_BODY = (
" op.add_column('widgets', sa.Column('goodname', sa.Integer()))\n"
" op.add_column('widgets', sa.Column(name='alsogood', type_=sa.String()))\n"
" op.alter_column('widgets', 'oldname', new_column_name='newname')\n"
# drop_column names a column that already exists, so a bad name here is
# deliberately NOT a finding: it was wrong where it was created, and
# reporting it again makes one rename look like two.
" op.drop_column('widgets', 'legacy_old')\n"
)
def _migration_module(body):
return ('"""Planted by the test suite. Never applied."""\n'
'import sqlalchemy as sa\n'
'from alembic import op\n\n'
"revision = 'plantedprobe'\n"
'down_revision = None\n\n\n'
'def upgrade():\n' + body + '\n\n'
'def downgrade():\n'
' pass\n')
def _gate_tree(tmp_path):
gate = tmp_path / 'scripts' / 'check-naming-and-style.sh'
gate.parent.mkdir(parents=True, exist_ok=True)
shutil.copy(GATE, gate)
return gate
@pytest.mark.skipif(shutil.which('bash') is None, reason='needs bash')
@pytest.mark.parametrize('shape', sorted(PLANTED_COLUMN_SHAPES))
def test_the_gate_catches_every_column_shape(tmp_path, shape):
"""Plant one shape at a time and require the gate to name it.
Parametrised rather than one file holding all nine, so a regression says
WHICH shape stopped being seen. A single file would fail on the first one
and hide the rest.
"""
body, expected = PLANTED_COLUMN_SHAPES[shape]
versions = tmp_path / 'migrations' / 'versions'
versions.mkdir(parents=True)
gate = _gate_tree(tmp_path)
planted = versions / 'zz01_planted.py'
planted.write_text(_migration_module(body), encoding='utf-8')
dirty = subprocess.run(['bash', str(gate)], cwd=str(tmp_path),
capture_output=True, text=True)
assert dirty.returncode == 1, (
'the gate passed a {0} column:\n{1}'.format(shape, dirty.stdout))
assert expected in dirty.stdout, (
'the gate failed but never named {0}:\n{1}'.format(
expected, dirty.stdout))
@pytest.mark.skipif(shutil.which('bash') is None, reason='needs bash')
def test_the_gate_passes_well_named_columns(tmp_path):
"""The other half: a gate that fails on everything is no better.
Includes a drop_column with a bad name, which is out of scope on purpose.
"""
versions = tmp_path / 'migrations' / 'versions'
versions.mkdir(parents=True)
gate = _gate_tree(tmp_path)
(versions / 'zz01_clean.py').write_text(
_migration_module(PLANTED_CLEAN_BODY), encoding='utf-8')
clean = subprocess.run(['bash', str(gate)], cwd=str(tmp_path),
capture_output=True, text=True)
assert clean.returncode == 0, (
'the gate failed on lowercase concatenated names:\n' + clean.stdout)
@pytest.mark.skipif(shutil.which('bash') is None, reason='needs bash')
def test_the_site_literal_scan_honours_its_include_list(tmp_path):
"""The ADR-015 scan must read the file types it declares, and no others.
The scan carried --exclude=README.md AHEAD of its --include list, and on GNU
grep a file-level exclude in that position cancels every include on the
command. So the scan was reading EVERY file under agent/ and the other
scanned trees, and --include='*.cs' was decorative - adding agent/ did not
mean the C# agent was covered for the reason the flags claimed, and any
later narrowing of the include list would have been silently inert.
The .cs half is covered by the test above. This is the other half: a file
type the scan does not declare must not fail it, which is only true when the
include list is actually in force. Before the fix this run came back red on
notes.txt.
"""
gate = _gate_tree(tmp_path)
agent = tmp_path / 'agent' / 'Probe'
agent.mkdir(parents=True)
for name in ('notes.txt', 'Probe.csproj', 'README.md'):
(agent / name).write_text(
'shipped to ' + PLANTED_SITE_HOST + '\n', encoding='utf-8')
result = subprocess.run(['bash', str(gate)], cwd=str(tmp_path),
capture_output=True, text=True)
assert result.returncode == 0, (
'the scan read a file type it does not declare:\n' + result.stdout)

View File

@@ -72,15 +72,83 @@ def test_write_verbs_can_carry_a_body(spec):
'from this spec cannot send one:\n ' + '\n '.join(offenders[:15]))
def documented_auth(operation):
"""The auth string the generator wrote into this operation's description.
The spec is the only thing this module reads, and the description is where
the human-readable auth lands ('**Auth:** jwt-optional (...)'). Comparing
the published `security` against it is how a lie gets caught: the two come
from the same inventory entry and must agree.
"""
match = re.search(r'\*\*Auth:\*\*\s*(.+)', operation.get('description') or '')
return match.group(1).strip().lower() if match else ''
def test_optional_auth_is_expressed_as_optional(spec):
"""An endpoint that works logged out must not be published as requiring a
token - that is a lie to every reader, and it is the majority case here."""
optional = [operation for _, _, operation in operations(spec)
if any(item == {} for item in operation.get('security') or [])]
assert len(optional) > 50, (
'The inventory documents over a hundred jwt-optional endpoints; the '
'spec should express them with an empty security requirement '
'alongside the scheme, not as bearer-required.')
"""A route whose auth is optional is published as optional, either way round.
WHY THE COUNT IS GONE. This asserted `len(optional) > 50`, on the premise
that optional auth was the majority case here - it said so in as many
words. It is not any more. The guest role and the permission model moved
the read surface to `permission:*`, which publishes as bearer-required and
is correct, and the optional count fell to 14. The assertion went red
without a single route being wrong.
A count was never the property anyone cares about. It goes red when the
codebase legitimately changes shape, and it stays green while an individual
optional route is published as bearer-required, which is the actual defect.
Lowering the number to 14 would only buy time until the next legitimate
move. So the premise is dropped and the per-route property is checked
instead, whatever the total: an operation whose auth says optional carries
BOTH the empty requirement and a scheme, and an operation that publishes
the empty requirement says optional in its auth.
"""
offenders = []
for path, verb, operation in operations(spec):
auth = documented_auth(operation)
security = operation.get('security') or []
openly = any(item == {} for item in security)
scheme = any(item for item in security)
if 'optional' in auth:
if not openly:
offenders.append(
'%s %s: auth says optional, published as required (%r)'
% (verb.upper(), path, security))
elif not scheme:
offenders.append(
'%s %s: auth says optional, published with no scheme at '
'all, so a caller is never told what the credential is'
% (verb.upper(), path))
elif openly:
offenders.append(
'%s %s: published as optional, but its auth says %r'
% (verb.upper(), path, auth))
assert not offenders, (
'Optional auth means "works without a credential, does more with one", '
'and OpenAPI writes it as the empty requirement ALONGSIDE the '
'scheme:\n ' + '\n '.join(offenders[:15]))
def test_an_operation_needing_no_credential_says_so_in_its_auth(spec):
"""`security: []` is a machine-readable claim that no credential is needed.
Nine operations published it while their handler required one, including
the two that hand a bootstrapping PC its agent package - `security: []` and
a 404 for anyone who does not present the single-use token. The generator
reached that by substring: the auth string started with the word 'none', in
a sentence that went on to name the token. This is the gate on that class
of defect from the spec side, where a reader or a generated client sees it.
"""
offenders = [
'%s %s: %r' % (verb.upper(), path, documented_auth(operation))
for path, verb, operation in operations(spec)
if not operation.get('security')
and 'none' not in documented_auth(operation)
and 'public' not in documented_auth(operation)
]
assert not offenders, (
'These operations publish "no credentials needed" while their auth '
'names a credential:\n ' + '\n '.join(offenders[:15]))
def test_an_endpoint_that_needs_no_token_cannot_fail_with_401(spec):

View File

@@ -15,6 +15,7 @@ These tests pin that contract:
* `flask plugin upgrade-all` runs clean on a fresh DB and is idempotent.
"""
import ast
import re
from pathlib import Path
import pytest
@@ -487,3 +488,634 @@ def test_upgrade_all_seeds_permissions_and_links_them_to_scopes(
(plugin_manager._app, plugin_manager._db, plugin_manager.registry,
plugin_manager.loader, plugin_manager.migration_manager,
plugin_manager._registered_prefixes) = saved
# ---------------------------------------------------------------------------
# ADR-008 lane guard: a plugin migration may only do DDL on its OWN tables.
#
# The endpoints chain broke this and nothing noticed. 0021 (shipped, since
# rewritten to a no-op stub) added five client-certificate columns plus
# idx_assets_cert_fingerprint to the CORE assets table from inside a plugin
# chain, and 0023 drops them again. Both directions are guarded and idempotent
# now, so the breach self-heals on every database - but self-healing is not
# prevention, and the only thing that stopped the next one was that somebody
# happened to read the file. That is what this guard replaces.
#
# The rule, stated as ADR-008 states it: the core Alembic chain owns core
# tables, each plugin chain owns the tables listed for it in
# PLUGIN_TABLE_OWNERS, and no chain reaches across. `assets` is not in
# PLUGIN_TABLE_OWNERS['endpoints'] and cannot be, so the per-plugin MetaData
# filter never saw the columns either: the ownership map was right and the
# migration simply ignored it.
# ---------------------------------------------------------------------------
CORE_MODELS_DIR = Path(__file__).resolve().parent.parent / 'shopdb' / 'core' / 'models'
# Alembic op name -> position of the table name in its positional arguments.
# create_index/drop_index take the INDEX name first and the table second, and
# every constraint op takes the CONSTRAINT name first, which is the whole
# reason this is a map and not "args[0]".
#
# The constraint ops are here because they are DDL on somebody's table just as
# much as add_column is: create_foreign_key / drop_constraint /
# create_unique_constraint / create_check_constraint / create_primary_key can
# all add or drop a constraint on core `assets` from inside a plugin chain.
# Leaving them out let that pass silently.
#
# create_foreign_key's REFERENT table is deliberately not recorded. The DDL
# runs against the source table only; naming another lane's table as a foreign
# key target is a read of its shape, not a write to it, and plugins legitimately
# point at core parents (network 0002 references `models`).
_TABLE_ARG_POSITION = {
'create_table': 0,
'drop_table': 0,
'add_column': 0,
'drop_column': 0,
'alter_column': 0,
'rename_table': 0,
'create_index': 1,
'drop_index': 1,
'create_foreign_key': 1,
'drop_constraint': 1,
'create_unique_constraint': 1,
'create_check_constraint': 1,
'create_primary_key': 1,
}
# Keywords that carry the table when it is not passed positionally. 0023 uses
# `table_name='assets'`, which an args-only scan walks straight past. `source`
# and `source_table` are the two spellings Alembic has used for
# create_foreign_key's own table; `referent`/`referent_table` are absent on
# purpose, for the reason given above.
_TABLE_KEYWORDS = ('table_name', 'source', 'source_table', 'old_table_name')
# Op name -> position of the COLUMN name in its positional arguments, for the
# ops that name exactly one column. The allowlist keys on this as well as the
# table, so an entry excusing a shipped drop_column cannot also excuse a NEW
# drop_column on the same table in the same file.
_COLUMN_ARG_POSITION = {
'add_column': 1,
'drop_column': 1,
'alter_column': 1,
}
# Same, for the batch_alter_table forms, where the table moved onto the context
# manager and every positional argument shifted left by one.
_BATCH_COLUMN_ARG_POSITION = {
'add_column': 0,
'drop_column': 0,
'alter_column': 0,
}
# --- raw SQL -----------------------------------------------------------------
# op.execute() and Connection.exec_driver_sql() hand a string straight to the
# database, so an AST scan that only knows the op.* helpers sees zero ops for
# `op.execute('ALTER TABLE assets DROP COLUMN assetnumber')`. That is the
# cheapest possible way around every other guard in this file, so the statement
# text is parsed too.
#
# Only DDL against a named table counts. UPDATE/DELETE/INSERT/SELECT are data,
# not schema: three plugin revisions already run cross-checked DML
# (backups 0002, backups 0003, computers 0006) and none of them is a lane
# crossing.
#
# Unresolvable text is rendered as a sentinel character rather than dropped, so
# the difference between "interpolates the constraint name" and "interpolates
# the table name" survives into the regex. machines 0002 does the first
# (`f"ALTER TABLE equipment DROP FOREIGN KEY {name}"` - the table is literal
# text, the FK name is not) and stays resolvable; the dodge this guards against
# is the second (`f'ALTER TABLE {t} DROP COLUMN x'`), where the sentinel lands
# inside the captured table name and the op reports UNRESOLVED, which
# test_plugin_migration_ops_all_resolve_to_a_table treats as a failure.
_SQL_HOLE = '\x00'
# Identifier characters plus the sentinel, so an interpolated name is captured
# rather than silently truncated to the literal prefix beside it.
_SQL_IDENT = r'[A-Za-z0-9_$`"\[\]\.' + _SQL_HOLE + r']+'
# (op name, compiled pattern, number of table groups). Ordered only for
# readability; every pattern is tried against every statement.
_SQL_DDL_PATTERNS = (
('execute:alter_table',
re.compile(r'\bALTER\s+TABLE\s+(?:IF\s+EXISTS\s+)?(' + _SQL_IDENT + r')', re.I)),
('execute:drop_table',
re.compile(r'\bDROP\s+TABLE\s+(?:IF\s+EXISTS\s+)?(' + _SQL_IDENT + r')', re.I)),
('execute:create_table',
re.compile(r'\bCREATE\s+(?:TEMPORARY\s+)?TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?('
+ _SQL_IDENT + r')', re.I)),
('execute:rename_table',
re.compile(r'\bRENAME\s+TABLE\s+(' + _SQL_IDENT + r')\s+TO\s+('
+ _SQL_IDENT + r')', re.I)),
('execute:create_index',
re.compile(r'\bCREATE\s+(?:UNIQUE\s+|FULLTEXT\s+|SPATIAL\s+)?INDEX\s+'
+ _SQL_IDENT + r'\s+ON\s+(' + _SQL_IDENT + r')', re.I)),
('execute:drop_index',
re.compile(r'\bDROP\s+INDEX\s+' + _SQL_IDENT + r'\s+ON\s+('
+ _SQL_IDENT + r')', re.I)),
('execute:truncate_table',
re.compile(r'\bTRUNCATE\s+(?:TABLE\s+)?(' + _SQL_IDENT + r')', re.I)),
)
# Statement verbs that are data, not schema. A statement starting with one of
# these is skipped outright, even if its body happens to contain a word like
# "table"; anything starting with something unreadable is recorded UNRESOLVED,
# because `op.execute(sql)` with sql built elsewhere is the same dodge as an
# f-string.
_SQL_DML_VERBS = ('select', 'insert', 'update', 'delete', 'replace', 'with',
'set', 'commit', 'rollback', 'analyze', 'optimize', 'call',
'pragma', 'flush', 'use', 'show', 'start')
# Historical lane crossings that have already shipped to production. Every entry
# is closed - listed so the guard is green today and red for anything new, NOT
# as a precedent to extend. Adding a line here means a plugin chain wrote to
# somebody else's table, which needs an ADR amendment, not an allowlist edit.
#
# The key is (plugin, filename, op, table, column). The column is part of it on
# purpose: keyed on the first four alone, the endpoints 0023 drop_column entry
# would also excuse a brand new `op.drop_column('assets', 'assetnumber')`
# appended to that same shipped file, which is exactly the breach this guard
# exists to stop. `None` means the op names no single column (drop_index) or
# names one the scan cannot resolve.
#
# endpoints 0023 is the CLEANUP of the 0021/0022 client-certificate breach: it
# drops the five cert columns and any index over them from the core assets
# table. It is applied on WJ production, so it cannot be edited - the fix is
# that no 0025 may do the same thing.
#
# 0021 and 0022 themselves are NOT listed, and must not be: they were rewritten
# to stamp-only no-ops, so they present no DDL to scan. Their damage lives on
# only as the 0023 cleanup below.
#
# The drop_index half of that cleanup (0023:71-73) reads alarmingly: it drops
# ANY index on assets whose column set INTERSECTS the cert columns, and never
# recreates one. An index over (assetid, client_cert_fingerprint) is dead once
# the column goes, but an index that merely happened to include a cert column
# alongside columns that still matter would be silently lost. Settled here so
# the next reader does not re-derive it - only these things ever create an index
# on assets:
# * core 68b3947ae14f baseline: idx_asset_active, idx_asset_location,
# idx_asset_status, idx_asset_type_bu, ix_assets_assetnumber,
# ix_assets_serialnumber
# * core 7b02: ix_assets_gaugelabreference, ix_assets_maintenancereference
# * core 7d33: idx_assets_levelid
# * the shipped endpoints 0021: idx_assets_cert_fingerprint over
# (client_cert_fingerprint) alone. The shipped 0022 added two more cert
# columns and no index at all.
# * db.create_all off shopdb/core/models/asset.py. Between 18e5fdc00
# (2026-09-14) and 7480e4fe3 (2026-09-15) that model DID carry the five cert
# columns, with client_cert_fingerprint declared index=True, so create_all
# built ix_assets_client_cert_fingerprint on every database created in that
# window. That is a second cert index, not just the one 0021 named, and the
# inventory above is wrong without it.
# It changes nothing for this loop. ix_assets_client_cert_fingerprint is
# single-column over client_cert_fingerprint, which 0023 drops in the very next
# statement, so the index carries no column that outlives it and there is
# nothing to recreate. The conclusion holds: the only indexes the loop can drop
# are idx_assets_cert_fingerprint and ix_assets_client_cert_fingerprint, both
# dead by construction. No index was lost and no repair migration is owed.
#
# The filter is also unreachable by accident: every name in _CERT_COLUMNS is
# underscore-spelled, which this repo's column convention forbids, so no real
# assets column can collide with it.
ALLOWED_CROSS_LANE_OPS = {
('endpoints', '0023_endpoint_credentials.py', 'drop_index', 'assets', None),
('endpoints', '0023_endpoint_credentials.py', 'drop_column', 'assets',
'client_cert_fingerprint'),
('endpoints', '0023_endpoint_credentials.py', 'drop_column', 'assets',
'cert_issued_at'),
('endpoints', '0023_endpoint_credentials.py', 'drop_column', 'assets',
'cert_expires_at'),
('endpoints', '0023_endpoint_credentials.py', 'drop_column', 'assets',
'cert_revoked_at'),
('endpoints', '0023_endpoint_credentials.py', 'drop_column', 'assets',
'cert_revocation_reason'),
}
def _core_tablenames() -> set:
"""Every table the CORE chain owns, read statically off shopdb/core/models.
Picks up both `__tablename__` on a model and the bare `db.Table(...)`
association tables (userroles, rolepermissions, userpermissions), which have
no model class and would otherwise read as unowned.
"""
names = set()
for source in CORE_MODELS_DIR.glob('*.py'):
tree = ast.parse(source.read_text())
for node in ast.walk(tree):
if isinstance(node, ast.Assign):
targets = [t.id for t in node.targets if isinstance(t, ast.Name)]
if '__tablename__' in targets and isinstance(node.value, ast.Constant):
names.add(node.value.value)
if isinstance(node, ast.Call):
func = node.func
name = func.attr if isinstance(func, ast.Attribute) else getattr(func, 'id', None)
if name == 'Table' and node.args:
first = node.args[0]
if isinstance(first, ast.Constant) and isinstance(first.value, str):
names.add(first.value)
return names
def _table_owner_map() -> dict:
"""table name -> owning lane ('core' or a plugin name)."""
owners = {}
for plugin, tables in PLUGIN_TABLE_OWNERS.items():
for tablename in tables:
owners[tablename] = plugin
for tablename in _core_tablenames():
# A plugin entry wins: pre-cutover the core chain CREATED the plugin
# tables, so a name can look core-shaped and still belong to a plugin.
owners.setdefault(tablename, 'core')
return owners
def _module_constants(tree) -> dict:
"""Module-level `NAME = 'sometable'` bindings.
Half the chains name their tables through constants (CREDENTIALS,
REQUESTS, _APPS), so a scan that only understands string literals sees
almost nothing.
"""
consts = {}
for node in tree.body:
if isinstance(node, ast.Assign) and isinstance(node.value, ast.Constant) \
and isinstance(node.value.value, str):
for target in node.targets:
if isinstance(target, ast.Name):
consts[target.id] = node.value.value
return consts
def _module_sequences(tree) -> dict:
"""Module-level `NAME = ('a', 'b')` bindings of all-string sequences.
Needed for the column half of the allowlist key: endpoints 0023 drops its
five columns with `for column in _CERT_COLUMNS`, and without this the column
resolves to nothing and the allowlist entry degrades back to a blanket
"drop_column on assets from this file is fine".
"""
seqs = {}
for node in tree.body:
if not isinstance(node, ast.Assign):
continue
if not isinstance(node.value, (ast.Tuple, ast.List, ast.Set)):
continue
values = set()
for element in node.value.elts:
if not (isinstance(element, ast.Constant) and isinstance(element.value, str)):
values = set()
break
values.add(element.value)
if not values:
continue
for target in node.targets:
if isinstance(target, ast.Name):
seqs[target.id] = frozenset(values)
return seqs
def _literal(node, consts):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return node.value
if isinstance(node, ast.Name):
return consts.get(node.id)
return None
def _loop_bindings(tree, consts, seqs) -> dict:
"""Map each Call node to the loop variables in scope over a literal sequence.
`for table in (IMPACTS, TRANSITIONS, HEALTHSTATES): op.drop_table(table)`
(endpoints 0017) and the same shape in guides 0001 are real DDL over real
tables. Without this they resolve to nothing, and "resolves to nothing" has
to be a failure or the guard is dodgeable by writing a loop.
A bare module constant as the iterable (`for column in _CERT_COLUMNS`) is
resolved through `seqs` for the same reason.
"""
bindings = {}
for node in ast.walk(tree):
if not isinstance(node, ast.For) or not isinstance(node.target, ast.Name):
continue
values = set()
if isinstance(node.iter, (ast.Tuple, ast.List, ast.Set)):
for element in node.iter.elts:
value = _literal(element, consts)
if value is None:
values = set()
break
values.add(value)
elif isinstance(node.iter, ast.Name):
values = set(seqs.get(node.iter.id, ()))
if not values:
continue
# ast.walk is breadth-first from the top, so an inner loop rebinding the
# same name is visited after the outer one and correctly shadows it.
for sub in ast.walk(node):
if isinstance(sub, ast.Call):
bindings.setdefault(id(sub), {})[node.target.id] = frozenset(values)
return bindings
def _resolve_tables(node, consts, loopvars) -> frozenset:
"""Every table name an argument could be. Empty means unresolvable."""
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return frozenset({node.value})
if isinstance(node, ast.Name):
if node.id in loopvars:
return loopvars[node.id]
if node.id in consts:
return frozenset({consts[node.id]})
return frozenset()
def _resolve_columns(opname, node, args, consts, loopvars, positions) -> frozenset:
"""The column name(s) an op names, or empty when it names none.
Empty is NOT a failure here the way it is for a table: most ops name no
column at all. It only widens the allowlist key to None.
"""
position = positions.get(opname)
if position is None or len(args) <= position:
return frozenset()
argument = args[position]
if opname == 'add_column':
# add_column takes sa.Column('name', ...), not a bare string.
if isinstance(argument, ast.Call) and argument.args:
argument = argument.args[0]
else:
return frozenset()
return _resolve_tables(argument, consts, loopvars)
def _flatten_sql(node, consts) -> str:
"""Statement text with every unresolvable fragment replaced by a sentinel.
Keeping the hole in place (instead of giving up on the whole string) is what
lets `f"ALTER TABLE equipment DROP FOREIGN KEY {name}"` resolve to
`equipment` while `f'ALTER TABLE {t} DROP COLUMN x'` resolves to nothing.
"""
if isinstance(node, ast.Constant):
return node.value if isinstance(node.value, str) else _SQL_HOLE
if isinstance(node, ast.JoinedStr):
return ''.join(_flatten_sql(value, consts) for value in node.values)
if isinstance(node, ast.FormattedValue):
return _SQL_HOLE
if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
return _flatten_sql(node.left, consts) + _flatten_sql(node.right, consts)
if isinstance(node, ast.Name):
return consts.get(node.id, _SQL_HOLE)
if isinstance(node, ast.Call):
# sa.text('...') / text('...') is a wrapper, not a hole: computers 0006
# passes its statement that way.
func = node.func
name = func.attr if isinstance(func, ast.Attribute) else getattr(func, 'id', None)
if name == 'text' and len(node.args) == 1:
return _flatten_sql(node.args[0], consts)
return _SQL_HOLE
def _clean_sql_identifier(raw: str) -> str:
"""`db`.`assets` -> assets. Keeps the sentinel if one landed in the name."""
name = raw.strip().rstrip('(;,')
if '.' in name:
name = name.rsplit('.', 1)[-1]
return name.strip('`"[]')
def _scan_raw_sql(node, consts) -> list:
"""(opname, frozenset-of-tables) for the DDL in one execute-style call.
An empty table set is UNRESOLVED and the caller fails on it. A statement
that is plainly DML returns nothing at all.
"""
statement = _flatten_sql(node, consts)
found = []
# Classify PER STATEMENT, not per call. Reading the first verb and bailing
# let a leading DML verb hide trailing DDL: op.execute("SET @x = 1; ALTER
# TABLE assets DROP COLUMN serialnumber") was skipped whole, because the
# scan saw SET, called it DML and returned nothing.
for part in statement.split(';'):
stripped = part.lstrip().lstrip('(')
verb = stripped.split(None, 1)[0].lower() if stripped.split() else ''
if not verb:
continue
if verb.startswith(_SQL_HOLE):
# The verb itself is interpolated, so nothing about this statement
# can be checked. Unresolved op rather than no op.
found.append(('execute:unknown', frozenset()))
continue
if verb in _SQL_DML_VERBS:
continue
for opname, pattern in _SQL_DDL_PATTERNS:
for match in pattern.finditer(part):
for group in match.groups():
name = _clean_sql_identifier(group)
if not name or _SQL_HOLE in name:
found.append((opname, frozenset()))
else:
found.append((opname, frozenset({name})))
return found
def _scan_plugin_migration_ops() -> list:
"""Every schema op in every plugin chain, as
(plugin, filename, lineno, opname, frozenset-of-tables,
frozenset-of-columns).
An empty table set means the argument could not be resolved statically;
the caller treats that as a failure rather than a pass. An empty column set
only means the op names no resolvable single column.
"""
found = []
for script in sorted(PLUGINS_DIR.glob('*/migrations/versions/*.py')):
plugin = script.parts[-4]
tree = ast.parse(script.read_text())
consts = _module_constants(tree)
seqs = _module_sequences(tree)
loops = _loop_bindings(tree, consts, seqs)
# batch_alter_table('assets') moves the table onto the context manager,
# so batch_op.drop_column('x') names a column where op.drop_column names
# a table. Resolve those against the `with` instead.
batched = {}
for node in ast.walk(tree):
if not isinstance(node, ast.With):
continue
for item in node.items:
call = item.context_expr
if not (isinstance(call, ast.Call) and isinstance(call.func, ast.Attribute)
and call.func.attr == 'batch_alter_table' and call.args):
continue
tables = _resolve_tables(call.args[0], consts, loops.get(id(call), {}))
alias = item.optional_vars.id if isinstance(item.optional_vars, ast.Name) else None
for sub in ast.walk(node):
if isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute) \
and sub.func.attr in _TABLE_ARG_POSITION \
and isinstance(sub.func.value, ast.Name) \
and sub.func.value.id == alias:
batched[id(sub)] = tables
for node in ast.walk(tree):
if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)):
continue
opname = node.func.attr
base = node.func.value
# Raw SQL: op.execute(...), bind.execute(...) and any
# <connection>.exec_driver_sql(...).
#
# Both names are matched on the method name ALONE, without looking at
# what they are called on. Requiring the receiver to be literally `op`
# missed the repo's DOMINANT idiom: 24 call sites across endpoints
# 0006/0007/0010/0019, network 0003 and notifications 0005 all do
# `bind = op.get_bind()` and then `bind.execute(...)`, so a plugin
# could drop a column off the core assets table through a bind and
# the scan recorded nothing at all. Every shipped bind.execute passes
# a literal SELECT or UPDATE, which _SQL_DML_VERBS skips, so widening
# this costs no false positives.
if opname in ('exec_driver_sql', 'execute'):
if not node.args:
continue
for sqlop, tables in _scan_raw_sql(node.args[0], consts):
found.append((plugin, script.name, node.lineno, sqlop,
tables, frozenset()))
continue
if opname not in _TABLE_ARG_POSITION:
continue
loopvars = loops.get(id(node), {})
if id(node) in batched:
columns = _resolve_columns(opname, node, node.args, consts, loopvars,
_BATCH_COLUMN_ARG_POSITION)
found.append((plugin, script.name, node.lineno,
'batch_' + opname, batched[id(node)], columns))
continue
if not (isinstance(base, ast.Name) and base.id == 'op'):
continue
position = _TABLE_ARG_POSITION[opname]
tables = frozenset()
if len(node.args) > position:
tables = _resolve_tables(node.args[position], consts, loopvars)
for keyword in node.keywords:
if keyword.arg in _TABLE_KEYWORDS:
tables = tables | _resolve_tables(keyword.value, consts, loopvars)
columns = _resolve_columns(opname, node, node.args, consts, loopvars,
_COLUMN_ARG_POSITION)
found.append((plugin, script.name, node.lineno, opname, tables, columns))
if opname == 'rename_table' and len(node.args) > 1:
# Both ends of a rename are DDL on a table.
found.append((plugin, script.name, node.lineno, 'rename_table',
_resolve_tables(node.args[1], consts, loopvars),
frozenset()))
return found
def _cross_lane_ops() -> list:
"""Scanned ops whose table belongs to core or to a different plugin.
A table that is in neither the core models nor PLUGIN_TABLE_OWNERS is NOT a
crossing: that is a pre-rename name (machines 0002 renames equipment ->
machines) or a table since dropped, and neither reaches into another lane.
Ownership drift of a table that DOES still exist is already covered by
test_owned_tables_match_declared_models.
"""
owners = _table_owner_map()
crossings = []
for plugin, filename, lineno, opname, tables, columns in _scan_plugin_migration_ops():
for tablename in sorted(tables):
owner = owners.get(tablename)
if owner is None or owner == plugin:
continue
for column in (sorted(columns) or [None]):
crossings.append((plugin, filename, lineno, opname, tablename,
owner, column))
return crossings
def _crossing_keys() -> set:
return {(plugin, filename, opname, tablename, column)
for plugin, filename, _, opname, tablename, _, column in _cross_lane_ops()}
def test_plugin_migration_ops_all_resolve_to_a_table():
"""Every scanned op names a table the scan can actually resolve.
Without this the guard is dodgeable by accident: a table name built at
runtime resolves to nothing, and "nothing" would otherwise read as "touches
no foreign table" and pass. It is also what makes the raw-SQL scan real -
`op.execute(f'ALTER TABLE {t} ...')` lands here rather than going quietly
green.
"""
unresolved = [(plugin, filename, lineno, opname)
for plugin, filename, lineno, opname, tables, _
in _scan_plugin_migration_ops() if not tables]
assert not unresolved, (
'plugin migration ops whose table name could not be resolved '
f'statically: {unresolved}')
def test_scan_sees_the_known_lane_crossing():
"""The scanner is not vacuous: it still finds the 0023 assets cleanup.
If a refactor breaks the AST walk, every other assertion here goes green on
an empty result set. This one goes red.
"""
scanned = _scan_plugin_migration_ops()
assert len(scanned) > 100, f'scan found only {len(scanned)} ops; it is broken'
keys = _crossing_keys()
assert ALLOWED_CROSS_LANE_OPS <= keys, (
'the allowlisted historical crossings are no longer detected: '
f'{ALLOWED_CROSS_LANE_OPS - keys}')
def test_scan_reads_raw_sql_ddl():
"""The raw-SQL half is not vacuous either.
machines 0002 does its whole MySQL rename through exec_driver_sql. If the
statement parser stops working, those ops vanish and nothing else in this
file notices.
"""
raw = [(plugin, filename, opname, tuple(sorted(tables)))
for plugin, filename, _, opname, tables, _
in _scan_plugin_migration_ops() if opname.startswith('execute:')]
assert ('machines', '0002_rename_from_equipment.py', 'execute:rename_table',
('equipmenttypes',)) in raw, raw
assert ('machines', '0002_rename_from_equipment.py', 'execute:alter_table',
('machines',)) in raw, raw
# DML stays out of it, or the three shipped UPDATE statements would read as
# schema changes on tables their own plugin owns.
assert not [entry for entry in raw
if entry[0] == 'backups' or entry[1].startswith('0006_policy')], raw
def test_no_plugin_migration_touches_another_lanes_table():
"""ADR-008: a plugin chain does DDL only on the tables it owns.
Fails on create_table/drop_table/add_column/drop_column/alter_column/
create_index/drop_index/rename_table, the five constraint ops, the
batch_alter_table forms, and DDL inside op.execute/exec_driver_sql, against
a core table or another plugin's table. The allowlist above holds exactly
the entries that already shipped.
"""
offenders = [crossing for crossing in _cross_lane_ops()
if (crossing[0], crossing[1], crossing[3], crossing[4], crossing[6])
not in ALLOWED_CROSS_LANE_OPS]
assert not offenders, '\n'.join(
f'{plugin} chain: {filename}:{lineno} {opname} on {tablename!r}'
+ (f' column {column!r}' if column else '')
+ f', owned by {owner} (ADR-008)'
for plugin, filename, lineno, opname, tablename, owner, column in offenders)
def test_cross_lane_allowlist_has_no_dead_entries():
"""The allowlist does not outlive what it excuses.
An entry left behind after its migration is squashed or rewritten is a
standing licence to write to a core table from that file again.
"""
keys = _crossing_keys()
assert not (ALLOWED_CROSS_LANE_OPS - keys), (
'allowlist entries that match nothing in the tree: '
f'{ALLOWED_CROSS_LANE_OPS - keys}')

View File

@@ -11,6 +11,7 @@ import datetime
import hashlib
import io
import json
import logging
import zipfile
import pytest
@@ -27,17 +28,51 @@ TOKENS_URL = '/api/endpoints/siteagent/bootstrap/tokens'
SCRIPT_URL = '/api/endpoints/siteagent/bootstrap.ps1'
PACKAGE_URL = '/api/endpoints/siteagent/bootstrap/package'
BASE_URL = 'https://shopdb.test/shopdb'
TEMP_SCRIPT = r'C:\Windows\Temp\shopdb-bootstrap.ps1'
TOKEN_HEADER = 'X-Endpoint-Bootstrap-Token'
QUERY_SETTING = 'plugin.endpoints.bootstrapquerytoken'
# Admin-only staging, NOT C:\Windows\Temp: a standard user can pre-plant a
# path there and have it run elevated.
STAGE_DIR = r'C:\Program Files\ShopDB\bootstrap'
STAGED_SCRIPT = STAGE_DIR + r'\shopdb-bootstrap.ps1'
ADMIN_SID = 'S-1-5-32-544'
SYSTEM_SID = 'S-1-5-18'
# The marker the admin "Download .ps1" link carries, so the server can tell an
# admin's browser click apart from a bay still on a pre-change pasted command.
UI_LINK_QUERY = 'via=download'
def _expected_command(secret, base=BASE_URL):
"""Download, run, delete. Never `irm | iex` - Defender kills that shape."""
return ('powershell -ExecutionPolicy Bypass -c "'
"try {{ Invoke-WebRequest -UseBasicParsing "
"'{0}/api/endpoints/siteagent/bootstrap.ps1?t={1}' "
"-OutFile '{2}'; & '{2}' }} finally {{ "
"Remove-Item '{2}' -Force -ErrorAction SilentlyContinue }}\""
).format(base, secret, TEMP_SCRIPT)
"""Stage admin-only, download with the token in a header, check the owner,
run, delete. Never `irm | iex` - Defender kills that shape."""
return (
'powershell -ExecutionPolicy Bypass -c "'
"try {{ New-Item -ItemType Directory -Force -Path '{dir}' "
'| Out-Null; '
"icacls '{dir}' /inheritance:r /grant '*{admin}:(OI)(CI)F' "
"'*{system}:(OI)(CI)F' | Out-Null; "
'if ((Get-Variable LASTEXITCODE -ValueOnly) -ne 0) '
"{{ Write-Host 'Refusing to run: could not restrict "
"{dir} to administrators (icacls failed). Fix the directory "
"or delete it, then run the command again.' -ForegroundColor Red; "
'exit 6 }}; '
"Remove-Item '{path}' -Force -ErrorAction SilentlyContinue; "
"Invoke-WebRequest -UseBasicParsing "
"'{base}/api/endpoints/siteagent/bootstrap.ps1' "
"-Headers @{{'{header}'='{secret}'}} -OutFile '{path}'; "
"if ((Get-Acl '{path}').GetOwner("
'[Security.Principal.SecurityIdentifier]).Value -notin '
"@('{admin}','{system}',"
'[Security.Principal.WindowsIdentity]::GetCurrent().User.Value)) '
"{{ Write-Host 'Refusing to run: the staged bootstrap script is not "
"owned by an administrator.' -ForegroundColor Red; exit 5 }}; "
"& '{path}' }} finally {{ Remove-Item '{path}' -Force "
'-ErrorAction SilentlyContinue }}"'
).format(dir=STAGE_DIR, path=STAGED_SCRIPT, base=base, header=TOKEN_HEADER,
secret=secret, admin=ADMIN_SID, system=SYSTEM_SID)
def _token_header(secret):
return {TOKEN_HEADER: secret}
def _make_cert():
@@ -111,8 +146,8 @@ def test_generate_returns_the_secret_and_the_ready_made_command(
assert len(secret) == len('sdbb_') + 43
assert data['command'] == _expected_command(secret)
assert data['scripturl'] == (
'{0}/api/endpoints/siteagent/bootstrap.ps1?t={1}'.format(BASE_URL,
secret))
'{0}/api/endpoints/siteagent/bootstrap.ps1?t={1}&{2}'.format(
BASE_URL, secret, UI_LINK_QUERY))
token = data['token']
assert token['tokenprefix'] == secret[:12]
assert token['maxuses'] == 3
@@ -136,13 +171,74 @@ def test_the_command_downloads_and_runs_instead_of_piping_into_iex(
assert 'iex' not in command.lower()
assert '$' not in command
assert '\n' not in command
assert data['scripturl'] in command
assert data['secret'] in command
assert "Remove-Item '{0}' -Force".format(TEMP_SCRIPT) in command
assert "Remove-Item '{0}' -Force".format(STAGED_SCRIPT) in command
assert 'finally' in command
# The link the UI offers instead is the route that really serves it.
assert client.get(SCRIPT_URL,
query_string={'t': data['secret']}).status_code == 200
assert client.get(data['scripturl'].split(BASE_URL, 1)[1]
).status_code == 200
def test_the_command_stages_where_only_administrators_can_write(
client, db, auth_headers):
"""C:\\Windows\\Temp lets any standard user create the exact path the
command runs elevated, so the staging directory moved under Program Files,
is created with an admin-only DACL, and the command refuses to run a staged
file owned by anyone but an administrator."""
command = _secret(client, auth_headers)['command']
assert r'C:\Windows\Temp' not in command
assert "-OutFile '{0}'".format(STAGED_SCRIPT) in command
assert "New-Item -ItemType Directory -Force -Path '{0}'".format(
STAGE_DIR) in command
assert "icacls '{0}' /inheritance:r".format(STAGE_DIR) in command
# Well-known SIDs, so a renamed or localized group cannot widen the grant.
assert "/grant '*{0}:(OI)(CI)F' '*{1}:(OI)(CI)F'".format(
ADMIN_SID, SYSTEM_SID) in command
assert "Get-Acl '{0}'".format(STAGED_SCRIPT) in command
assert 'GetOwner([Security.Principal.SecurityIdentifier]).Value' in command
assert "-notin @('{0}','{1}',".format(ADMIN_SID, SYSTEM_SID) in command
assert '[Security.Principal.WindowsIdentity]::GetCurrent().User.Value' \
in command
assert 'not owned by an administrator' in command
assert 'exit 5' in command
# A leftover from an earlier run is cleared before the download, so the
# owner test cannot pass on stale content.
assert command.index("Remove-Item '{0}'".format(STAGED_SCRIPT)) < \
command.index('Invoke-WebRequest')
def test_the_command_sends_the_token_in_a_header_not_the_query_string(
client, db, auth_headers):
"""A query string lands verbatim in the nginx and gunicorn access logs."""
data = _secret(client, auth_headers)
command = data['command']
assert '?t=' not in command
assert "-Headers @{{'{0}'='{1}'}}".format(TOKEN_HEADER,
data['secret']) in command
assert "'{0}/api/endpoints/siteagent/bootstrap.ps1'".format(
BASE_URL) in command
# The browser download link is the one place that cannot send a header,
# and it is marked so its own fetches are not counted as un-migrated bays.
assert data['scripturl'].endswith(
'?t={0}&{1}'.format(data['secret'], UI_LINK_QUERY))
def test_a_failed_acl_hardening_stops_the_command(client, db, auth_headers):
"""icacls is a native command: it raises nothing and $ErrorActionPreference
never applies to it, so an unchecked call leaves a staging directory that
still inherits the parent DACL and the line downloads into it anyway. The
exit code is read sigil-free, because the whole line must hold no `$`."""
command = _secret(client, auth_headers)['command']
assert '(Get-Variable LASTEXITCODE -ValueOnly) -ne 0' in command
assert 'LASTEXITCODE' in command and '$LASTEXITCODE' not in command
assert 'icacls failed' in command
assert 'exit 6' in command
# Checked before anything is fetched into the directory, not after.
assert command.index('LASTEXITCODE') > command.index('icacls')
assert command.index('LASTEXITCODE') < command.index('Invoke-WebRequest')
def test_defaults_are_a_day_and_twenty_five_uses(client, db, auth_headers):
@@ -226,8 +322,8 @@ def test_script_is_served_for_a_good_token(client, db, auth_headers):
assert response.mimetype == 'text/plain'
script = response.get_data(as_text=True)
script.encode('ascii')
assert '{0}/api/endpoints/siteagent/bootstrap/package?t={1}'.format(
BASE_URL, secret) in script
assert '"{0}/api/endpoints/siteagent/bootstrap/package"'.format(
BASE_URL) in script
assert 'Run as administrator' in script
assert 'GE-Enforce' in script
assert 'SHOPDB_BOOTSTRAP_FORCE' in script
@@ -251,6 +347,163 @@ def test_script_bakes_the_staged_binary_hashes(client, db, auth_headers,
assert hashlib.sha256(traypayload).hexdigest() in script
def test_the_generated_script_fetches_the_package_with_the_header(
client, db, auth_headers):
"""Same reason as the one-liner: the package URL must not carry the token,
or every agent install writes it into an access log."""
secret = _secret(client, auth_headers)['secret']
script = client.get(SCRIPT_URL,
headers=_token_header(secret)).get_data(as_text=True)
assert '?t=' not in script
assert '$BootstrapPackageHeaders = @{{ "{0}" = "{1}" }}'.format(
TOKEN_HEADER, secret) in script
assert ('Invoke-WebRequest -Uri $BootstrapPackageUrl '
'-Headers $BootstrapPackageHeaders') in script
def test_both_routes_take_the_token_in_a_header(client, db, auth_headers,
servercert):
secret = _secret(client, auth_headers, maxuses=2)['secret']
script = client.get(SCRIPT_URL, headers=_token_header(secret))
assert script.status_code == 200
assert 'ShopDB endpoint agent bootstrap' in script.get_data(as_text=True)
assert 'install.ps1' in _files(
client.get(PACKAGE_URL, headers=_token_header(secret)))
assert EndpointBootstrapToken.query.one().usecount == 1
@pytest.mark.parametrize('url', [SCRIPT_URL, PACKAGE_URL])
def test_a_bad_header_token_is_not_found(client, db, auth_headers, url):
"""The header is read first, so a good ?t= cannot rescue a bad header - one
request carries one token."""
secret = _secret(client, auth_headers)['secret']
response = client.get(url, query_string={'t': secret},
headers=_token_header('sdbb_' + 'a' * 43))
assert response.status_code == 404
def test_the_query_form_still_works_and_is_audited_as_deprecated(
client, db, auth_headers, caplog):
"""A bay running a command pasted before this change must keep working, so
the query form stays accepted; it is flagged in the audit row and warned
about in the log, with no token value in either."""
secret = _secret(client, auth_headers)['secret']
with caplog.at_level(logging.WARNING):
assert client.get(SCRIPT_URL, query_string={'t': secret},
environ_overrides={'REMOTE_ADDR': '10.0.0.6'}
).status_code == 200
row = AuditLog.query.filter_by(entitytype='EndpointBootstrapToken',
action='requested').one()
assert 'query-string token, deprecated' in row.details
assert secret not in row.details
warnings = [record.getMessage() for record in caplog.records
if record.levelno >= logging.WARNING]
assert any('query string' in message and '10.0.0.6' in message
for message in warnings)
assert all(secret not in message for message in warnings)
def test_the_header_form_is_not_flagged_as_deprecated(client, db, auth_headers,
caplog):
secret = _secret(client, auth_headers)['secret']
with caplog.at_level(logging.WARNING):
client.get(SCRIPT_URL, headers=_token_header(secret))
row = AuditLog.query.filter_by(entitytype='EndpointBootstrapToken',
action='requested').one()
assert 'deprecated' not in row.details
assert not [record for record in caplog.records
if record.levelno >= logging.WARNING]
def test_the_admin_download_link_is_not_counted_as_an_un_migrated_bay(
client, db, auth_headers, caplog):
"""The rollout finds un-migrated bays by the deprecation WARNING, and the
UI Download .ps1 link has to use the query form. Marked, it logs at info
and is audited as the admin path, so an admin testing a token does not
invent a bay that is not there."""
data = _secret(client, auth_headers)
path = data['scripturl'].split(BASE_URL, 1)[1]
with caplog.at_level(logging.INFO):
assert client.get(path, environ_overrides={'REMOTE_ADDR': '10.0.0.7'}
).status_code == 200
row = AuditLog.query.filter_by(entitytype='EndpointBootstrapToken',
action='requested').one()
assert 'admin download link' in row.details
assert 'deprecated' not in row.details
assert data['secret'] not in row.details
assert not [record for record in caplog.records
if record.levelno >= logging.WARNING]
infos = [record.getMessage() for record in caplog.records
if record.levelno == logging.INFO]
assert any('Download .ps1' in message and '10.0.0.7' in message
for message in infos)
assert all(data['secret'] not in message for message in infos)
def test_no_log_or_audit_row_ever_carries_the_token_value(
client, db, auth_headers, caplog, servercert):
secret = _secret(client, auth_headers, maxuses=4)['secret']
with caplog.at_level(logging.DEBUG):
client.get(SCRIPT_URL, query_string={'t': secret})
client.get(SCRIPT_URL, headers=_token_header(secret))
client.get(PACKAGE_URL, query_string={'t': secret})
client.get(PACKAGE_URL, headers=_token_header(secret))
assert all(secret not in record.getMessage()
for record in caplog.records)
rows = AuditLog.query.filter_by(
entitytype='EndpointBootstrapToken').all()
assert rows
for row in rows:
assert secret not in (row.details or '')
# The 12-char display prefix is the identifier the UI already shows.
assert row.entityname == secret[:12]
@pytest.mark.parametrize('url', [SCRIPT_URL, PACKAGE_URL])
def test_the_query_form_can_be_turned_off_once_a_site_has_re_issued(
client, db, auth_headers, url):
"""Off is the hardened end state, not the default: the UI download link
uses the query form too, so a site only turns it off deliberately."""
secret = _secret(client, auth_headers)['secret']
Setting.set(QUERY_SETTING, False, valuetype='boolean')
refused = client.get(url, query_string={'t': secret})
assert refused.status_code == 400
assert TOKEN_HEADER in refused.get_json()['data']['error']['message']
# The admin Download .ps1 link is a query token too: the setting is about
# a live token landing in an access log, and that link puts one there.
assert client.get(url, query_string={'t': secret, 'via': 'download'}
).status_code == 400
# Refused before any lookup, so it confirms nothing about the token.
assert EndpointBootstrapToken.query.one().usecount == 0
assert AuditLog.query.filter_by(
entitytype='EndpointBootstrapToken', action='requested').count() == 0
assert client.get(url, headers=_token_header(secret)).status_code == 200
@pytest.mark.parametrize('url', [SCRIPT_URL, PACKAGE_URL])
def test_no_token_is_still_not_found_when_the_query_form_is_off(
client, db, auth_headers, url):
"""A bare GET presented no token, so it is unusable and answers 404 like
every other unusable request. Calling it a query token answered 400 naming
the header, which handed an anonymous caller the state of the switch."""
_secret(client, auth_headers)
Setting.set(QUERY_SETTING, False, valuetype='boolean')
assert client.get(url).status_code == 404
# An empty ?t= presented nothing either.
assert client.get(url, query_string={'t': ''}).status_code == 404
# And it is the same answer with the query form still on.
Setting.set(QUERY_SETTING, True, valuetype='boolean')
assert client.get(url).status_code == 404
def test_script_fetch_does_not_spend_a_use(client, db, auth_headers):
secret = _secret(client, auth_headers, maxuses=1)['secret']
for _ in range(3):
@@ -421,3 +674,4 @@ def test_the_command_falls_back_to_the_request_when_no_site_url_is_set(
base='https://localhost')
assert data['scripturl'].startswith(
'https://localhost/api/endpoints/siteagent/bootstrap.ps1?t=')
assert data['scripturl'].endswith('&' + UI_LINK_QUERY)

View File

@@ -689,6 +689,7 @@ def test_enrollment_listing_and_settings(client, db, app, keyheaders,
body = response.get_json()
data = body['data']
assert data['settings'] == {'sharedkeyallowed': True, 'autoenroll': True,
'revokeddeniessharedkey': False,
'maxpendingenrollments': 500}
assert data['counts'] == {'enrolled': 1, 'notenrolled': 1,
'pendingcomputers': 1, 'pending': 1}
@@ -711,6 +712,7 @@ def test_enrollment_listing_and_settings(client, db, app, keyheaders,
assert put.status_code == 200
assert put.get_json()['data'] == {'sharedkeyallowed': False,
'autoenroll': True,
'revokeddeniessharedkey': False,
'maxpendingenrollments': 500}
bad = client.put('/api/endpoints/enrollment/settings',
json={'autoenroll': 'yes'}, headers=auth_headers)

View File

@@ -0,0 +1,352 @@
"""What a denial means at the enrollment door, and how long it lasts.
Three denials meet here, and each one used to be undone by the PC on the other
side of it:
reject An admin refuses a request. The agent then sends a new nonce and
the row was re-armed as a fresh cycle - which, with the shared key
and auto-enroll on, auto-issued the credential the admin had just
refused. A rejection binds to the identity (hostname + serial), so
it survives any nonce; only an admin approving it clears it.
cap The pending queue is bounded, and at the cap a PC nobody has seen
before is refused. Abandoned rows come out first, so a queue that
filled with junk drains itself instead of blocking every new PC.
revoke A revoked credential falls back to the shared key, which is how
the un-migrated fleet and the rollback to an older agent keep
reporting. plugin.endpoints.revokeddeniessharedkey (off by
default) turns a revoke into an eviction instead.
"""
from datetime import timedelta
import pytest
from shopdb.core.models import Asset, AssetType, Setting
from plugins.endpoints.api import agentauth
from plugins.endpoints.api.enrollment import (
PENDING_CAP_SETTING, PENDING_STALE_AFTER,
)
from plugins.endpoints.models import EndpointCredential, EndpointEnrollRequest
COLLECTOR_KEY = 'testcollectorkey'
ENROLL_URL = '/api/endpoints/enroll'
ALERTS_URL = '/api/endpoints/alerts'
NONCE = 'n' * 40
OTHER_NONCE = 'm' * 40
@pytest.fixture
def keyheaders(app):
app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY
return {'X-API-Key': COLLECTOR_KEY}
@pytest.fixture(autouse=True)
def nolimiter(app):
"""The per-IP limiter counts in a process-wide cache that outlives a test.
Nothing here is about how FAST a PC asks, and another test in the same
worker may already have spent the budget for this window and address, so the
limiter is off for this module and put back as it was.
"""
key = 'ENDPOINTS_ENROLL_RATELIMIT_ENABLED'
previous = app.config.get(key)
app.config[key] = False
yield
if previous is None:
app.config.pop(key, None)
else:
app.config[key] = previous
def _computer(db, hostname, serialnumber='SER-1'):
from plugins.computers.models import Computer
atype = AssetType.query.filter_by(assettype='computer').first()
if atype is None:
atype = AssetType(assettype='computer')
db.session.add(atype)
db.session.flush()
asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
serialnumber=serialnumber, isactive=True)
db.session.add(asset)
db.session.flush()
db.session.add(Computer(assetid=asset.assetid, hostname=hostname))
db.session.commit()
return asset.assetid
def _body(hostname='BAY-01', serialnumber='SER-1', nonce=NONCE):
return {'hostname': hostname, 'serialnumber': serialnumber,
'enrollnonce': nonce, 'agentversion': '1.20.0'}
def _enroll(client, headers, **kwargs):
return client.post(ENROLL_URL, json=_body(**kwargs), headers=headers)
def _pending_request(client, db, keyheaders, **kwargs):
"""A request waiting for an admin, with auto-enroll back on afterwards.
Auto-enroll off is what puts a clean match in the queue instead of issuing
at once; turning it back on is what made the old code hand the retry a
credential, so every rejection test needs both halves.
"""
_setting(db, 'plugin.endpoints.autoenroll', False)
response = _enroll(client, keyheaders, **kwargs)
assert response.status_code == 202, response.get_data(as_text=True)
_setting(db, 'plugin.endpoints.autoenroll', True)
return response.get_json()['data']['requestid']
def _reject(client, auth_headers, requestid):
return client.post(
f'/api/endpoints/enrollment/requests/{requestid}/reject',
headers=auth_headers)
def _setting(db, key, value, valuetype='boolean'):
Setting.set(key, value, valuetype=valuetype, category='plugin.endpoints')
db.session.commit()
def _code(response):
return response.get_json()['data']['error']['code']
# -- a rejection binds to the identity ---------------------------------------
def test_a_new_nonce_does_not_lift_a_rejection(client, db, app, keyheaders,
auth_headers):
"""The bug: reinstall the agent and the retry was auto-issued the
credential an admin had just refused."""
_computer(db, 'BAY-01')
requestid = _pending_request(client, db, keyheaders)
assert _reject(client, auth_headers, requestid).status_code == 200
retry = _enroll(client, keyheaders, nonce=OTHER_NONCE)
assert retry.status_code == 202, retry.get_data(as_text=True)
assert retry.get_json()['data']['reason'] == 'rejected'
assert retry.get_json()['data']['requestid'] == requestid
assert EndpointCredential.query.count() == 0
row = db.session.get(EndpointEnrollRequest, requestid)
assert row.status == EndpointEnrollRequest.STATUS_REJECTED
assert row.decidedat is not None
# Still one row for this identity: the retry did not queue a second.
assert EndpointEnrollRequest.query.count() == 1
def test_a_rejection_survives_however_often_the_pc_asks(client, db, app,
keyheaders,
auth_headers):
_computer(db, 'BAY-01')
requestid = _pending_request(client, db, keyheaders)
_reject(client, auth_headers, requestid)
for index in range(5):
again = _enroll(client, keyheaders, nonce=('x%d' % index) * 20)
assert again.status_code == 202
assert again.get_json()['data']['reason'] == 'rejected'
assert EndpointCredential.query.count() == 0
assert EndpointEnrollRequest.query.count() == 1
def test_a_rejection_does_not_touch_a_different_pc(client, db, app,
keyheaders, auth_headers):
"""Identity is hostname AND serial, so one PC's rejection is its own."""
_computer(db, 'BAY-01', serialnumber='SER-1')
other = _computer(db, 'BAY-02', serialnumber='SER-2')
requestid = _pending_request(client, db, keyheaders)
_reject(client, auth_headers, requestid)
clean = _enroll(client, keyheaders, hostname='BAY-02',
serialnumber='SER-2')
assert clean.status_code == 201, clean.get_data(as_text=True)
assert EndpointCredential.query.filter_by(assetid=other).one() \
.revokedat is None
def test_an_admin_clears_a_rejection_by_approving_it(client, db, app,
keyheaders, auth_headers):
"""The way back is the admin's, and it works for the reinstalled agent that
now carries a different nonce."""
assetid = _computer(db, 'BAY-01')
requestid = _pending_request(client, db, keyheaders)
_reject(client, auth_headers, requestid)
_enroll(client, keyheaders, nonce=OTHER_NONCE)
approve = client.post(
f'/api/endpoints/enrollment/requests/{requestid}/approve',
json={'assetid': assetid}, headers=auth_headers)
assert approve.status_code == 200, approve.get_data(as_text=True)
issued = _enroll(client, keyheaders, nonce=OTHER_NONCE)
assert issued.status_code == 201, issued.get_data(as_text=True)
assert EndpointCredential.query.filter_by(assetid=assetid).one() \
.enrolledvia == 'approved'
def test_an_approval_out_of_a_rejection_still_binds_to_one_installation(
client, db, app, keyheaders, auth_headers):
"""It binds to the installation the PC was last seen with, so a third
machine claiming the same hostname and serial still cannot collect it."""
assetid = _computer(db, 'BAY-01')
requestid = _pending_request(client, db, keyheaders)
_reject(client, auth_headers, requestid)
_enroll(client, keyheaders, nonce=OTHER_NONCE)
client.post(f'/api/endpoints/enrollment/requests/{requestid}/approve',
json={'assetid': assetid}, headers=auth_headers)
impostor = _enroll(client, keyheaders, nonce='z' * 40)
assert impostor.status_code == 403
assert _code(impostor) == 'ENROLL_NONCE_MISMATCH'
assert EndpointCredential.query.count() == 0
issued = _enroll(client, keyheaders, nonce=OTHER_NONCE)
assert issued.status_code == 201, issued.get_data(as_text=True)
# -- the pending cap ages out abandoned rows ---------------------------------
def _stale(db, hostname, days):
from datetime import datetime, timezone
seen = (datetime.now(timezone.utc).replace(tzinfo=None)
- timedelta(days=days))
row = EndpointEnrollRequest(
hostname=hostname, serialnumber=hostname + '-SER',
enrollnonce=NONCE, status=EndpointEnrollRequest.STATUS_PENDING,
requestedat=seen, lastseenat=seen)
db.session.add(row)
db.session.commit()
return row
def test_the_cap_ages_out_abandoned_rows_before_refusing(client, db, app):
"""A queue full of PCs that stopped asking must not block the one that is."""
_setting(db, PENDING_CAP_SETTING, 2, valuetype='integer')
_stale(db, 'GONE-01', PENDING_STALE_AFTER.days + 1)
_stale(db, 'GONE-02', PENDING_STALE_AFTER.days + 3)
response = client.post(ENROLL_URL, json=_body(hostname='NEW-01',
serialnumber='NEW-SER'))
assert response.status_code == 202, response.get_data(as_text=True)
hostnames = {row.hostname for row in EndpointEnrollRequest.query.all()}
assert hostnames == {'NEW-01'}
def test_the_age_out_spares_rows_still_being_asked_for(client, db, app):
"""Only rows nothing has touched go, and a decided row never goes: the
rejection above has to outlive any queue pressure."""
_setting(db, PENDING_CAP_SETTING, 2, valuetype='integer')
fresh = _stale(db, 'LIVE-01', 0)
stillasking = _stale(db, 'LIVE-02', 0)
decided = _stale(db, 'GONE-01', PENDING_STALE_AFTER.days + 1)
decided.status = EndpointEnrollRequest.STATUS_REJECTED
db.session.commit()
refused = client.post(ENROLL_URL, json=_body(hostname='NEW-01',
serialnumber='NEW-SER'))
assert refused.status_code == 503
assert _code(refused) == 'ENROLL_PENDING_FULL'
for row in (fresh, stillasking, decided):
assert db.session.get(EndpointEnrollRequest, row.requestid) is not None
# -- revocation and the shared key -------------------------------------------
def test_a_revoked_pc_keeps_the_shared_key_by_default(client, db, app,
keyheaders,
auth_headers):
"""Today's behaviour, and the setting default: a revoke is a rotation, not
an eviction, so the PC keeps reporting while it waits to enroll again."""
assetid = _computer(db, 'BAY-01')
assert _enroll(client, keyheaders).status_code == 201
client.post(f'/api/endpoints/enrollment/assets/{assetid}/revoke',
headers=auth_headers)
assert client.get(ALERTS_URL + '?hostname=BAY-01',
headers=keyheaders).status_code == 200
def test_revoked_denies_shared_key_evicts_the_pc(client, db, app, keyheaders,
auth_headers):
assetid = _computer(db, 'BAY-01')
assert _enroll(client, keyheaders).status_code == 201
client.post(f'/api/endpoints/enrollment/assets/{assetid}/revoke',
json={'blockreenroll': True, 'reason': 'stolen'},
headers=auth_headers)
_setting(db, agentauth.REVOKED_DENIES_SHARED_KEY_SETTING, True)
refused = client.get(ALERTS_URL + '?hostname=BAY-01', headers=keyheaders)
assert refused.status_code == 401
assert _code(refused) == 'SHARED_KEY_REFUSED'
assert 'revoked' in refused.get_json()['data']['error']['message']
def test_a_pc_that_never_enrolled_is_never_evicted(client, db, app,
keyheaders, auth_headers):
"""The un-migrated fleet has no credential row at all, so the setting can be
turned on without silencing it."""
_computer(db, 'BAY-01', serialnumber='SER-1')
_computer(db, 'BAY-02', serialnumber='SER-2')
assetid = _computer(db, 'BAY-03', serialnumber='SER-3')
_enroll(client, keyheaders, hostname='BAY-03', serialnumber='SER-3')
client.post(f'/api/endpoints/enrollment/assets/{assetid}/revoke',
headers=auth_headers)
_setting(db, agentauth.REVOKED_DENIES_SHARED_KEY_SETTING, True)
assert client.get(ALERTS_URL + '?hostname=BAY-01',
headers=keyheaders).status_code == 200
assert client.get(ALERTS_URL + '?hostname=BAY-02',
headers=keyheaders).status_code == 200
assert client.get(ALERTS_URL + '?hostname=BAY-03',
headers=keyheaders).status_code == 401
def test_the_collector_contract_follows_the_same_rule(client, db, app,
keyheaders,
auth_headers):
"""The generic collector path asks the plugin, not bind_hostname, so it has
to answer the same way."""
from plugins.endpoints.plugin import EndpointsPlugin
assetid = _computer(db, 'BAY-01')
assert _enroll(client, keyheaders).status_code == 201
client.post(f'/api/endpoints/enrollment/assets/{assetid}/revoke',
headers=auth_headers)
refused = EndpointsPlugin().agent_shared_key_refused
with app.test_request_context():
assert refused('BAY-01') is False
_setting(db, agentauth.REVOKED_DENIES_SHARED_KEY_SETTING, True)
assert refused('BAY-01') is True
assert refused('UNKNOWN') is False
def test_the_eviction_switch_is_editable_and_defaults_to_today(client, db,
auth_headers):
"""A setting nobody can change is a constant with extra steps."""
read = client.get('/api/endpoints/enrollment',
headers=auth_headers).get_json()['data']
assert read['settings']['revokeddeniessharedkey'] is False
put = client.put('/api/endpoints/enrollment/settings',
json={'revokeddeniessharedkey': True},
headers=auth_headers)
assert put.status_code == 200, put.get_data(as_text=True)
assert put.get_json()['data']['revokeddeniessharedkey'] is True
assert Setting.get(agentauth.REVOKED_DENIES_SHARED_KEY_SETTING) is True
bad = client.put('/api/endpoints/enrollment/settings',
json={'revokeddeniessharedkey': 'yes'},
headers=auth_headers)
assert bad.status_code == 400

View File

@@ -0,0 +1,162 @@
"""The two endpoints switches are declared, seeded, and have a control.
Both were readable by their enforcement code before they were declared:
plugin.endpoints.revokeddeniessharedkey agentauth.revoked_denies_shared_key
plugin.endpoints.bootstrapquerytoken bootstrap._query_token_allowed
An undeclared key has no row until somebody writes one, and until it was
declared the settings API upserted it with valuetype 'string' - so
bool(Setting.get(key, True)) read the text 'false' as True and the switch did
nothing. Declaring it fixes both halves: the row is seeded at plugin
install/enable, and a PUT lands as a real boolean.
The UI half is checked by reading the pages. A switch with no control on an
air-gapped box can only be changed with SQL, which is not a setting.
"""
from pathlib import Path
import pytest
from plugins.endpoints.api import agentauth, bootstrap
from plugins.endpoints.plugin import EndpointsPlugin
from shopdb.core.models import Setting
REPO = Path(__file__).resolve().parents[2]
VIEWS = REPO / 'plugins' / 'endpoints' / 'frontend' / 'views'
REVOKED_KEY = agentauth.REVOKED_DENIES_SHARED_KEY_SETTING
QUERY_KEY = bootstrap.QUERY_TOKEN_SETTING
def _declared():
return {entry['key']: entry for entry in
EndpointsPlugin().get_settings_defaults()}
@pytest.mark.parametrize('key,value', [(REVOKED_KEY, 'false'),
(QUERY_KEY, 'true')])
def test_the_switch_is_declared_with_todays_behaviour(key, value):
entry = _declared().get(key)
assert entry is not None, key + ' is read but never declared'
assert entry['valuetype'] == 'boolean'
assert entry['category'] == 'plugin.endpoints'
# The declared default must be what the code does with no row at all, or
# seeding a site silently changes its behaviour.
assert entry['value'] == value
@pytest.mark.parametrize('key', [REVOKED_KEY, QUERY_KEY])
def test_the_description_names_what_turning_it_on_costs(key):
"""An operator on the shop floor decides from this text alone."""
description = _declared()[key]['description']
assert len(description) > 120, description
if key == REVOKED_KEY:
# The revoke + blockreenroll rollback onto an older agent stops working.
assert 'rollback' in description
assert 'enrolls again' in description
else:
assert 're-issued' in description or 're-issue' in description
def test_the_declared_defaults_match_what_the_code_reads_with_no_row(db):
"""No Setting rows at all: the readers fall back to these same values."""
assert Setting.query.filter_by(key=REVOKED_KEY).first() is None
assert Setting.query.filter_by(key=QUERY_KEY).first() is None
assert agentauth.revoked_denies_shared_key() is False
assert bootstrap._query_token_allowed() is True
def test_the_upgrade_pass_seeds_them_on_a_site_that_installed_earlier(app, db):
"""install/enable fire only on a state transition, so a fleet already
running this plugin gets these rows from `flask plugin upgrade-all`, which
calls sync_all_plugin_settings -> _seed_plugin_settings.
This drives _seed_plugin_settings on a manager built for this test rather
than the process-wide plugin_manager singleton, and that is deliberate.
create_app() re-points that singleton, eight test modules build their own
app, and whatever ran last on this xdist worker owns it: the symptoms are
"no such table: settings" (seeding against another app's in-memory SQLite)
and "'NoneType' object has no attribute 'get_all'" (registry left None).
Both land on whichever test follows, so they move with the split - this one
passed locally and failed on CI for exactly that reason.
init_app is not called here either: it re-registers the plugin blueprints,
and twice on one app raises "The name 'backups' is already registered".
Setting _app and _db is all the seeding routine reads.
"""
from shopdb.plugins import PluginManager
from plugins.endpoints.plugin import EndpointsPlugin
assert Setting.query.filter_by(key=QUERY_KEY).first() is None
manager = PluginManager()
manager._app = app
manager._db = db
assert manager._seed_plugin_settings(EndpointsPlugin()) is True
for key, value in ((REVOKED_KEY, False), (QUERY_KEY, True)):
row = Setting.query.filter_by(key=key).one()
assert row.valuetype == 'boolean'
assert row.category == 'plugin.endpoints'
assert row.get_typed_value() is value
def test_a_put_lands_as_a_boolean_and_flips_the_query_token(client, db,
auth_headers):
"""End to end for bootstrapquerytoken through the API the page uses.
The declaration is what makes the upsert boolean; as a string row the read
below would stay True whatever was saved.
"""
saved = client.put('/api/settings/' + QUERY_KEY, json={'value': False},
headers=auth_headers)
assert saved.status_code == 200, saved.get_json()
body = saved.get_json()['data']
assert body['value'] is False
assert body['valuetype'] == 'boolean'
assert body['category'] == 'plugin.endpoints'
assert bootstrap._query_token_allowed() is False
back = client.put('/api/settings/' + QUERY_KEY, json={'value': True},
headers=auth_headers)
assert back.get_json()['data']['value'] is True
assert bootstrap._query_token_allowed() is True
def test_the_settings_api_lists_the_declared_switch(client, db, auth_headers):
"""The agent-key page reads GET /api/settings/<key> to render the box."""
client.put('/api/settings/' + QUERY_KEY, json={'value': True},
headers=auth_headers)
resp = client.get('/api/settings/' + QUERY_KEY, headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert resp.get_json()['data']['value'] is True
def test_the_enrollment_page_carries_the_eviction_checkbox():
page = (VIEWS / 'EnrollmentSettings.vue').read_text(encoding='utf-8')
assert 'toggle-revokeddenies' in page, 'no checkbox for the eviction switch'
assert 'onRevokedDeniesToggle' in page
# It sits with the two switches it belongs beside.
assert 'toggle-sharedkey' in page and 'toggle-autoenroll' in page
# The operator has to be told what it breaks before clicking it.
assert 'rollback' in page
assert 'enrolls again' in page
def test_the_agent_key_page_carries_the_query_token_control():
page = (VIEWS / 'AgentKeySettings.vue').read_text(encoding='utf-8')
assert QUERY_KEY in page, 'the page does not touch the setting'
assert 'toggle-querytoken' in page
assert 'onQueryTokenToggle' in page
# Turning it off breaks commands already handed out, so say so.
assert 're-issue' in page

View File

@@ -289,6 +289,62 @@ def test_leaf_certificate_is_imported_into_root(client, db, auth_headers):
assert json.loads(files['appsettings.json'])['serverpins'] == [leafpin]
def test_installer_restricts_the_registry_key_to_system_and_admins(
client, db, auth_headers, servercert):
"""The shared key stays in HKLM until every consumer is on a per-PC
credential, so what matters meanwhile is who can read it. The key inherits a
DACL that grants BUILTIN\\Users read, which is the exposure the per-PC
migration exists to close - the installer must not leave it that way."""
files = _files(_build(client, auth_headers, collectorkey=SECRET,
includesharedkey='true'))
script = files['install.ps1'].decode('ascii')
assert 'function Protect-HklmShopdbKey' in script
assert 'Protect-HklmShopdbKey -SubKey "SOFTWARE\\GE\\ShopDB"' in script
# SYSTEM and Administrators by SID, so a localised Windows still matches.
assert '"S-1-5-18", "S-1-5-32-544"' in script
# Inheritance broken WITHOUT copying the inherited entries: copying them is
# what silently preserves the Users read.
assert 'SetAccessRuleProtection($true, $false)' in script
assert 'RemoveAccessRule($rule)' in script
def test_the_key_is_written_before_it_is_locked(client, db, auth_headers, servercert):
"""SetAccessControl replaces the DACL rather than merging, so a lock that ran
first would be undone by nothing - but a write that ran after a lock the
installer no longer holds rights for would fail. Order is the guarantee."""
script = _files(_build(client, auth_headers, collectorkey=SECRET,
includesharedkey='true'))['install.ps1'].decode('ascii')
write = script.index('-Name "CollectorKey" -Value $CollectorKey')
lock = script.index('Protect-HklmShopdbKey -SubKey')
assert write < lock
def test_the_registry_lock_failing_does_not_fail_the_install(
client, db, auth_headers, servercert):
"""A bay that cannot take the ACL still needs its agent. It must say plainly
that the key is readable there, rather than dying or staying quiet."""
script = _files(_build(client, auth_headers, collectorkey=SECRET,
includesharedkey='true'))['install.ps1'].decode('ascii')
lock = script.index('Protect-HklmShopdbKey -SubKey')
tail = script[lock:lock + 600]
assert 'catch' in tail
assert 'readable by any logged-on user' in tail
def test_the_keyless_package_still_locks_the_key(client, db, auth_headers, servercert):
"""ServerPins live in the same key and the registry is authoritative for
pinning, so a standard user must not be able to rewrite them either."""
files = _files(_build(client, auth_headers, collectorkey=SECRET,
includesharedkey=False))
script = files['install.ps1'].decode('ascii')
assert b'CollectorKey' not in files['install.ps1']
assert 'Protect-HklmShopdbKey -SubKey "SOFTWARE\\GE\\ShopDB"' in script
def test_install_script_helper_without_pins_or_key():
script = siteagent._install_script('https://shopdb.test', '', 900,
serverpins=[], includesharedkey=False)

View File

@@ -0,0 +1,212 @@
"""Network-position admission for the two unauthenticated kiosks.
The device token covers unattended screens an administrator provisions once. It
does not cover the other caller: somebody opening the kiosk page on their own
PC, who never logs in and whose browser nobody provisioned. Per-browser
provisioning would mean touching every PC on the floor and losing it on the next
reimage, so those callers are admitted by where they are on the network instead.
What these tests hold down:
- the allowlist actually admits a caller with no token (the feature),
- it stays shut when unset, when the caller is outside it, and when the kiosk
itself is off (the flag still wins),
- a malformed entry is skipped rather than widening the match, and
- the two kiosks do not share an allowlist.
"""
import pytest
from shopdb.core.models import Setting
from shopdb.api import ip_in_cidrs
PLANT = '10.134.48.17'
PLANT_CIDR = '10.134.48.0/23'
ELSEWHERE = '10.9.100.5'
# -- the pure matcher ---------------------------------------------------------
# Separated from the request so the fail-closed edges can be stated plainly.
@pytest.mark.parametrize('address, cidrs, expected', [
(PLANT, PLANT_CIDR, True),
(PLANT, ' 10.48.249.0/26 , 10.134.48.0/23 ', True),
(ELSEWHERE, PLANT_CIDR, False),
# Empty or absent setting means the feature is off, not "allow everyone".
(PLANT, '', False),
(PLANT, None, False),
# A malformed entry must be skipped, never treated as a wildcard.
(PLANT, 'not-a-cidr', False),
(PLANT, 'not-a-cidr, 10.134.48.0/23', True),
# An unparseable caller address matches nothing.
('', PLANT_CIDR, False),
('unknown', PLANT_CIDR, False),
])
def test_ip_in_cidrs(address, cidrs, expected):
assert ip_in_cidrs(address, cidrs) is expected
# -- USB kiosk ----------------------------------------------------------------
@pytest.fixture
def usbkiosk(db):
"""Self-hosted USB, kiosk ON, a token set, one employee and one drive."""
from plugins.employees.models import DirectoryEmployee
from plugins.usb.models import USBDevice
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='a-real-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 _allow(db, key, value):
row = Setting.query.filter_by(key=key).first()
if row is None:
db.session.add(Setting(key=key, value=value, valuetype='string',
category='usb'))
else:
row.value = value
db.session.commit()
def _session(client, address):
"""A kiosk call carrying NO token, from `address`."""
return client.post('/api/usb/kiosk/session', json={'badge': '270015376'},
environ_base={'REMOTE_ADDR': address})
def test_usb_kiosk_refuses_tokenless_caller_when_no_allowlist(usbkiosk, client):
"""The state before this feature: no token, no way in."""
assert _session(client, PLANT).status_code == 404
def test_usb_kiosk_admits_tokenless_caller_on_allowed_network(usbkiosk, client):
"""THE FEATURE. Fails without the allowlist branch in _kiosk_open."""
_allow(usbkiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR)
assert _session(client, PLANT).status_code == 200
def test_usb_kiosk_refuses_caller_outside_the_allowlist(usbkiosk, client):
"""The PXE LAN is also 10.x, which is why the range has to be narrow."""
_allow(usbkiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR)
assert _session(client, ELSEWHERE).status_code == 404
def test_usb_kiosk_allowlist_does_not_override_the_off_switch(usbkiosk, client):
"""Turning the kiosk off closes it to everyone, allowlisted or not.
The flag is the site's decision to have this door at all; the allowlist only
decides who walks through an open one.
"""
_allow(usbkiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR)
Setting.query.filter_by(key='usb_kiosk_enabled').first().value = 'false'
usbkiosk.session.commit()
assert _session(client, PLANT).status_code == 404
def test_usb_kiosk_token_still_works_from_any_network(usbkiosk, client):
"""Wall screens keep working, and are not confined to the allowlist."""
response = client.post('/api/usb/kiosk/session',
json={'badge': '270015376'},
headers={'X-Kiosk-Token': 'a-real-token'},
environ_base={'REMOTE_ADDR': ELSEWHERE})
assert response.status_code == 200
def test_usb_allowlist_does_not_open_the_parts_kiosk(usbkiosk, client):
"""Separate screens, separate rooms, separate settings.
The tokens are deliberately separate so provisioning one does not grant the
other. An allowlist shared between them would hand back exactly that.
"""
_allow(usbkiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR)
Setting.query.filter_by(key='usb_kiosk_enabled').first().value = 'true'
usbkiosk.session.add(Setting(key='printedparts_kiosk_enabled', value='true',
valuetype='boolean', category='printedparts'))
usbkiosk.session.commit()
response = client.get('/api/printedparts/kiosk/lookup?code=ANY',
environ_base={'REMOTE_ADDR': PLANT})
assert response.status_code == 404
# -- what the page probes on load ---------------------------------------------
def test_empty_badge_probe_separates_refused_from_admitted(usbkiosk, client):
"""The kiosk page asks the server on load whether it may be used.
It cannot guess: the enabled flag is public and only says the site HAS a
kiosk, and a caller admitted by network carries no token to look for. So the
page sends a badge-less session request and reads the status.
That only works while the gate runs BEFORE badge handling - refused is 404,
admitted is the badge validation error. If someone reorders those, the page
starts telling admitted screens they are not set up, so hold it here.
"""
refused = client.post('/api/usb/kiosk/session', json={'badge': ''},
environ_base={'REMOTE_ADDR': ELSEWHERE})
assert refused.status_code == 404
_allow(usbkiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR)
admitted = client.post('/api/usb/kiosk/session', json={'badge': ''},
environ_base={'REMOTE_ADDR': PLANT})
assert admitted.status_code != 404
assert admitted.status_code < 500
def test_the_probe_records_nothing(usbkiosk, client):
"""A page load must not leave a checkout, a log row, or a resolved badge."""
from plugins.usb.models import USBCheckout
_allow(usbkiosk, 'usb_kiosk_allowed_cidrs', PLANT_CIDR)
client.post('/api/usb/kiosk/session', json={'badge': ''},
environ_base={'REMOTE_ADDR': PLANT})
assert USBCheckout.query.count() == 0
# -- provisioning link --------------------------------------------------------
# The settings API masks any key named '*token*', so the settings page shows
# '********'. An administrator who copies that provisions a screen with the mask
# and gets a silent 404 - the failure that cost real time on 2026-09-16. These
# hold the way out: a named, permission-gated route that hands a working setup
# link to somebody already allowed to set the token.
def test_provisioning_link_does_not_hand_back_the_mask(usbkiosk, client,
auth_headers):
"""It used to answer '/usb-kiosk?token=<value>' and this test pinned that
literal string. The link has since become a single-use setup code, which is
a better answer to the same problem - a query string ends up in history,
referrers and proxy logs - so what is held here is the property rather than
the shape: a usable link, and never the mask."""
response = client.get('/api/usb/kiosk/provisioning-link',
headers=auth_headers)
assert response.status_code == 200
url = response.get_json()['data']['url']
assert '*' not in url
# The token itself must not ride in the URL either.
assert 'a-real-token' not in url
assert client.get(url).status_code == 302
def test_provisioning_link_needs_permission(usbkiosk, client):
"""Anonymous callers cannot read the token out of this route."""
response = client.get('/api/usb/kiosk/provisioning-link')
assert response.status_code in (401, 403)
def test_provisioning_link_says_so_when_no_token_is_set(usbkiosk, client,
auth_headers):
Setting.query.filter_by(key='usb_kiosk_token').first().value = ''
usbkiosk.session.commit()
response = client.get('/api/usb/kiosk/provisioning-link',
headers=auth_headers)
assert response.status_code == 400
assert 'Generate' in response.get_json()['data']['error']['message']

View File

@@ -0,0 +1,77 @@
"""usb and printedparts declare a core_version floor they actually need.
Both reach for the network-position allowlist that arrived on the contract
surface in 0.35.0 (commit "Admit kiosk callers by network position"): ip_in_cidrs
and request_ip_allowlisted landed together in that bump, so either one witnesses
the same floor. Their manifests claimed >=0.1.0 and >=0.16.0, so the loader would
have accepted them on a core whose shopdb.api has no such name - the plugin then
fails at first import of its routes instead of at the version check that exists
to catch exactly this.
The witness is named PER PLUGIN and not shared. usb needs the matcher itself,
because its admission is the allowlist minus the site's proxies and the
addresses no caller can have, which the one-call helper does not express;
printedparts is the plain case and uses ip_in_cidrs. A single shared
name went red the moment usb moved, which said "the floor is wrong" when the
floor was right.
"""
import json
import pathlib
import pytest
from packaging.specifiers import SpecifierSet
from packaging.version import Version
from shopdb import __contract_version__
PLUGINS = pathlib.Path(__file__).resolve().parents[2] / 'plugins'
# The contract version that added request_ip_allowlisted / ip_in_cidrs.
IP_ALLOWLIST_CONTRACT = '0.35.0'
# plugin, the 0.35.0 name it really imports, the floor that name implies
CASES = [
('usb', 'ip_in_cidrs', IP_ALLOWLIST_CONTRACT),
# Was request_ip_allowlisted until 2026-09-18, when the parts kiosk moved to
# client_ip + ip_in_cidrs so it could SUBTRACT proxy and loopback addresses
# before consulting the allowlist. Same contract release, same floor.
('printedparts', 'ip_in_cidrs', IP_ALLOWLIST_CONTRACT),
]
def _manifest(name):
return json.loads((PLUGINS / name / 'manifest.json').read_text())
@pytest.mark.parametrize('name,witness,needed', CASES)
def test_manifest_excludes_a_core_without_the_name(name, witness, needed):
"""A core one minor below the name's arrival must be refused."""
spec = SpecifierSet(_manifest(name)['core_version'])
too_old = Version(needed)
too_old = Version(f'{too_old.major}.{too_old.minor - 1}.0')
assert too_old not in spec
@pytest.mark.parametrize('name,witness,needed', CASES)
def test_manifest_admits_the_running_core(name, witness, needed):
"""And the floor must not have overshot what this core provides."""
spec = SpecifierSet(_manifest(name)['core_version'])
assert Version(__contract_version__) in spec
@pytest.mark.parametrize('name,witness,needed', CASES)
def test_plugin_really_imports_the_name(name, witness, needed):
"""Pins the floor to the reason for it: drop the import and this test says
the floor can come back down."""
sources = (PLUGINS / name).rglob('*.py')
assert any(witness in path.read_text() for path in sources), name
@pytest.mark.parametrize('name,witness,needed', CASES)
def test_the_name_is_on_the_contract_surface(name, witness, needed):
"""Each witness is a contract name, not an internal import a plugin reached
through - which is the only reason the manifest floor can speak for it."""
import shopdb.api
assert witness in shopdb.api.__all__

View File

@@ -0,0 +1,93 @@
"""alertscope is TEXT in the database, a dict in Python.
Migration 0006 ships the column as TEXT because a site still runs MySQL 5.6,
which has no JSON type. The model declared db.JSON, so create-all built a column
the migration path never produces and the two disagreed about the same column.
"""
import json
from sqlalchemy import Text, inspect
from plugins.notifications.models.notification import Notification
def test_column_type_matches_the_migration(app):
"""The model's storage type must be the one migration 0006 creates."""
column = inspect(Notification).columns['alertscope']
assert isinstance(column.type.impl_instance
if hasattr(column.type, 'impl_instance')
else column.type.impl, Text)
def test_ddl_has_no_json_column(app, db):
"""MySQL 5.6 fails on a JSON column with a bare syntax error, so the
emitted DDL must not contain one."""
from sqlalchemy.schema import CreateTable
ddl = str(CreateTable(Notification.__table__).compile(db.engine))
alertscope_line = [line for line in ddl.splitlines()
if 'alertscope' in line][0]
assert 'JSON' not in alertscope_line.upper()
def test_dict_round_trips(app, db):
"""A scope assigned as a dict reads back as a dict."""
row = Notification(notification='fleet alert', alertfleet=True,
alerttimeout=0,
alertscope={'hostname': ['BAY-01', 'BAY-02']})
db.session.add(row)
db.session.commit()
notificationid = row.notificationid
db.session.expire_all()
reread = db.session.get(Notification, notificationid)
assert reread.alertscope == {'hostname': ['BAY-01', 'BAY-02']}
def test_stored_value_is_json_text(app, db):
"""What lands in the column is a JSON string, readable by a site that
queries the table directly."""
row = Notification(notification='fleet alert', alertfleet=True,
alerttimeout=0, alertscope={'computertype': 'CMM PC'})
db.session.add(row)
db.session.commit()
raw = db.session.execute(
db.text('SELECT alertscope FROM notifications WHERE notificationid = :i'),
{'i': row.notificationid}).scalar()
assert isinstance(raw, str)
assert json.loads(raw) == {'computertype': 'CMM PC'}
def test_null_scope_stays_null(app, db):
"""No scope means the whole fleet; it must not become the string 'null'."""
row = Notification(notification='fleet alert', alertfleet=True,
alerttimeout=0, alertscope=None)
db.session.add(row)
db.session.commit()
raw = db.session.execute(
db.text('SELECT alertscope FROM notifications WHERE notificationid = :i'),
{'i': row.notificationid}).scalar()
assert raw is None
assert row.alertscope is None
def test_unparseable_text_is_handed_back_raw(app, db):
"""A hand-edited row must be visible and fixable, not an exception on read.
The endpoints matcher already treats a non-dict scope as matching nothing.
"""
row = Notification(notification='fleet alert', alertfleet=True,
alerttimeout=0)
db.session.add(row)
db.session.commit()
db.session.execute(
db.text('UPDATE notifications SET alertscope = :v '
'WHERE notificationid = :i'),
{'v': 'not json at all', 'i': row.notificationid})
db.session.commit()
db.session.expire_all()
reread = db.session.get(Notification, row.notificationid)
assert reread.alertscope == 'not json at all'

View File

@@ -0,0 +1,92 @@
"""How the parts kiosk decides a caller is admitted by network position.
The USB kiosk closed this on 2026-09-18 and this one did not, so the two read
differently for no reason anybody chose. The hole is the same one: a reverse
proxy normally sits ON the network a site enters in the allowlist, so a request
wearing the proxy's own address matched the range and was admitted while
carrying no evidence at all about who sent it.
Shape of the requests here matches production: the layer in front (IIS with the
X-Forwarded-For rewrite, or nginx plus ProxyFix) has already resolved the caller
into REMOTE_ADDR, and the app never reads the header itself.
"""
import pytest
from shopdb.core.models import Setting
PLANT = '10.134.48.17'
PLANT_CIDR = '10.134.48.0/23'
ELSEWHERE = '10.9.100.5'
# Inside the allowlisted range, which is where a proxy actually lives and the
# reason the un-subtracted check was wrong.
PROXY = '10.134.48.5'
PROXY_CIDR = '10.134.48.5/32'
KIOSK_TOKEN = 'parts-kiosk-admission-token'
def _set(key, value):
Setting.set(key, value, valuetype='string', category='printedparts')
@pytest.fixture
def kiosk(app, db):
_set('printedparts_kiosk_enabled', 'true')
_set('printedparts_kiosk_token', KIOSK_TOKEN)
_set('printedparts_kiosk_allowed_cidrs', '')
_set('printedparts_kiosk_trusted_proxies', '')
return app
def _open_from(app, address):
"""Ask the gate directly: it is the unit under test, not a route's shape.
The request context comes from the REAL app, so Setting.get reaches the same
database the fixture wrote to.
"""
from plugins.printedparts.api.routes import _kiosk_open
with app.test_request_context('/', environ_base={'REMOTE_ADDR': address}):
return _kiosk_open()
def test_a_caller_on_an_allowed_network_is_admitted(kiosk):
_set('printedparts_kiosk_allowed_cidrs', PLANT_CIDR)
assert _open_from(kiosk, PLANT) is True
def test_a_caller_outside_every_allowed_network_is_refused(kiosk):
_set('printedparts_kiosk_allowed_cidrs', PLANT_CIDR)
assert _open_from(kiosk, ELSEWHERE) is False
def test_a_request_wearing_the_proxy_address_fails_closed(kiosk):
"""THE FIX. The proxy forwarded nothing, so its own address arrived - and it
sits inside the allowlist. No evidence about the sender means refuse."""
_set('printedparts_kiosk_allowed_cidrs', PLANT_CIDR)
_set('printedparts_kiosk_trusted_proxies', PROXY_CIDR)
assert _open_from(kiosk, PROXY) is False
# ... and the feature still works for a real caller on the same network, or
# the subtraction has taken the allowlist with it.
assert _open_from(kiosk, PLANT) is True
def test_loopback_is_never_a_kiosk_client(kiosk):
"""Needs no configuration: no site's kiosk is the server itself. This is
also the shape a misconfigured front end produces, where every request
arrives as 127.0.0.1."""
_set('printedparts_kiosk_allowed_cidrs', '127.0.0.0/8, ' + PLANT_CIDR)
assert _open_from(kiosk, '127.0.0.1') is False
def test_the_kiosk_flag_still_governs(kiosk):
"""Network position admits nobody while the kiosk is off."""
_set('printedparts_kiosk_allowed_cidrs', PLANT_CIDR)
_set('printedparts_kiosk_enabled', 'false')
assert _open_from(kiosk, PLANT) is False
def test_an_empty_allowlist_admits_nobody(kiosk):
"""Token-only, which is the shipped default."""
assert _open_from(kiosk, PLANT) is False

View File

@@ -0,0 +1,499 @@
"""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: <an allowlisted address>' 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)

View File

@@ -137,3 +137,66 @@ def test_the_report_is_registered_so_it_appears_in_the_catalogue(client, usbdata
assert report['id'] == 'usb-audit'
# The endpoint is what the reports page fetches; a wrong one renders blank.
assert report['endpoint'] == '/api/usb/reports/audit'
# -- sanitized / virus-scan state in the audit ---------------------------------
# CMMC evidence: the report answered who had which drive and when, but not
# whether it was wiped or scanned on the way back - which is the question the
# record exists to answer. Both stores held the values; neither reported them.
def test_audit_reports_wipe_and_scan_state_on_returns(db, client, auth_headers):
from plugins.usb.models import USBDevice, USBCheckout
from shopdb.core.models import Setting
from datetime import datetime, timedelta
db.session.add(Setting(key='usb_directory_mode', value='selfhosted',
valuetype='string', category='site'))
device = USBDevice(serialnumber='USB-9', label='Grey SanDisk')
db.session.add(device)
db.session.flush()
now = datetime.utcnow()
db.session.add(USBCheckout(
usbdeviceid=device.usbdeviceid, sso='270015376', checkoutname='Dana',
checkouttime=now - timedelta(hours=2), checkintime=now - timedelta(hours=1),
waswiped=True, scannedviruses=False))
db.session.commit()
response = client.get('/api/usb/reports/audit?days=7', headers=auth_headers)
assert response.status_code == 200
data = response.get_json()['data']
assert 'sanitized' in data['columns']
assert 'scanned_viruses' in data['columns']
checkin = [r for r in data['rows'] if r['action'] == 'check-in'][0]
checkout = [r for r in data['rows'] if r['action'] == 'check-out'][0]
assert checkin['sanitized'] == 'yes'
assert checkin['scanned_viruses'] == 'no'
# A take carries neither: both are recorded when the drive comes back.
assert checkout['sanitized'] == ''
assert checkout['scanned_viruses'] == ''
def test_unrecorded_wipe_is_blank_not_no(db, client, auth_headers):
"""A null means nobody wrote it down. Rendering that as 'no' would assert
something no one ever claimed, in the one record kept to be trustworthy."""
from plugins.usb.models import USBDevice, USBCheckout
from shopdb.core.models import Setting
from datetime import datetime, timedelta
db.session.add(Setting(key='usb_directory_mode', value='selfhosted',
valuetype='string', category='site'))
device = USBDevice(serialnumber='USB-10', label='Old stock')
db.session.add(device)
db.session.flush()
now = datetime.utcnow()
db.session.add(USBCheckout(
usbdeviceid=device.usbdeviceid, sso='270015376', checkoutname='Dana',
checkouttime=now - timedelta(hours=3), checkintime=now - timedelta(hours=2),
waswiped=None, scannedviruses=None))
db.session.commit()
response = client.get('/api/usb/reports/audit?days=7', headers=auth_headers)
checkin = [r for r in response.get_json()['data']['rows']
if r['action'] == 'check-in'][0]
assert checkin['sanitized'] == ''
assert checkin['scanned_viruses'] == ''