User Management
@@ -151,14 +204,66 @@
diff --git a/plugins/notifications/models/notification.py b/plugins/notifications/models/notification.py
index d0d30dba..9619aeb2 100644
--- a/plugins/notifications/models/notification.py
+++ b/plugins/notifications/models/notification.py
@@ -1,10 +1,46 @@
"""Notifications plugin models - adapted to existing database schema."""
+import json
+
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
+
+from sqlalchemy import Text
+from sqlalchemy.types import TypeDecorator
+
from shopdb.api import db
+class JSONText(TypeDecorator):
+ """A dict on the Python side, TEXT in the database.
+
+ MySQL 5.6 has no JSON type and a site still runs it, so every JSON-bearing
+ column in this product is TEXT. Declared here rather than imported from the
+ computers plugin: notifications declares no dependency on computers, and a
+ cross-plugin import would be a hidden one.
+ """
+
+ impl = Text
+ cache_ok = True
+
+ def process_bind_param(self, value, dialect):
+ if value is None:
+ return None
+ return json.dumps(value, separators=(',', ':'))
+
+ def process_result_value(self, value, dialect):
+ if value is None or value == '':
+ return None
+ try:
+ return json.loads(value)
+ except (TypeError, ValueError):
+ # Hand-edited or written before this column held JSON. Return the
+ # raw text rather than raising: a scope that cannot be parsed
+ # should be visible and fixable, and the endpoints matcher already
+ # treats a non-dict scope as matching nothing.
+ return value
+
+
_DEFAULT_TZ = 'America/New_York'
# Keys a fleet alert's alertscope may use. Must stay equal to the set the
@@ -155,7 +191,11 @@ class Notification(db.Model):
# Matches migration 0006: NOT NULL with server defaults.
alertfleet = db.Column(db.Boolean, nullable=False, default=False,
server_default=db.false())
- alertscope = db.Column(db.JSON, nullable=True)
+ # JSONText, not db.JSON: migration 0006 ships this column as TEXT because a
+ # site still runs MySQL 5.6, which has no JSON type. db.JSON here meant
+ # create-all built a JSON column the migration path never produces, so the
+ # model and the shipped schema disagreed about the same column.
+ alertscope = db.Column(JSONText, nullable=True)
alerttimeout = db.Column(db.Integer, nullable=False, default=0,
server_default='0')
diff --git a/plugins/printedparts/manifest.json b/plugins/printedparts/manifest.json
index 1343ef2f..0af57b1d 100644
--- a/plugins/printedparts/manifest.json
+++ b/plugins/printedparts/manifest.json
@@ -5,7 +5,7 @@
"display_name": "3D Printed Parts",
"author": "",
"dependencies": ["employees"],
- "core_version": ">=0.16.0,<1.0.0",
+ "core_version": ">=0.35.0,<1.0.0",
"api_prefix": "/api/printedparts",
"default_enabled": false
}
diff --git a/plugins/usb/manifest.json b/plugins/usb/manifest.json
index 5b595c3e..ce0838f2 100644
--- a/plugins/usb/manifest.json
+++ b/plugins/usb/manifest.json
@@ -5,7 +5,7 @@
"description": "USB device checkout management",
"author": "ShopDB Team",
"dependencies": [],
- "core_version": ">=0.1.0,<1.0.0",
+ "core_version": ">=0.35.0,<1.0.0",
"api_prefix": "/api/usb",
"default_enabled": false
}
\ No newline at end of file
diff --git a/shopdb/core/api/collector.py b/shopdb/core/api/collector.py
index e6c5f5f0..e806cdf4 100644
--- a/shopdb/core/api/collector.py
+++ b/shopdb/core/api/collector.py
@@ -27,6 +27,44 @@ COLLECTOR_SCOPE = 'collector.ingest'
# declaring acceptsagentcredential.
AGENT_CREDENTIAL_HEADER = 'X-Endpoint-Credential'
+# How a collector request authorized, recorded on g so a route can tell a
+# managed service token from the shared env key after the check has passed.
+AUTH_MANAGED_TOKEN = 'token'
+AUTH_SHARED_KEY = 'sharedkey'
+
+# The four legacy routes below (/pc, /apps, /heartbeat, /bulk) predate both the
+# generic collector contract and per-PC credentials. They take a hostname from
+# the body and rewrite that PC's logged-in user, boot time, serial number and
+# installed-app inventory, so the shared key is an any-host write key on them.
+# plugin.endpoints.sharedkeyallowed does not reach here - it gates the endpoints
+# collector, not core - so turning it off leaves these open. This setting is the
+# core counterpart. Default false = today's behaviour, unchanged.
+LEGACY_SHARED_KEY_SETTING = 'collector.legacysharedkeyrefused'
+
+# Declared here so the wording stays with the code that enforces it.
+# build_default_settings (shopdb/core/api/settings.py) has to include this dict
+# for the row to be seeded; until it does, the key is created on demand through
+# the settings API and the absent-row default below is what applies.
+LEGACY_SHARED_KEY_SETTING_DEFAULT = {
+ 'key': LEGACY_SHARED_KEY_SETTING,
+ 'value': 'false',
+ 'valuetype': 'boolean',
+ 'category': 'collector',
+ 'description': 'Refuse the shared collector key on the legacy core routes '
+ 'POST /api/collector/pc, /apps, /heartbeat and /bulk. Those '
+ 'routes name their target PC in the body, so any holder of '
+ 'the shared key can rewrite logged-in user, last boot time, '
+ 'serial number and installed apps for ANY hostname. Turning '
+ 'this on closes them to the shared key; a collector-scoped '
+ 'managed token still works. Turn it on only after checking '
+ 'the "Legacy collector route used with the shared key" '
+ 'warnings in the log for a while: whatever still calls them '
+ 'stops working the moment it is on. The two GE-Enforce '
+ 'reporters post to /api/collector/computers and '
+ '/api/collector/printers and are NOT affected. This should '
+ 'follow the move to per-PC credentials, not precede it.',
+}
+
def _computer_models():
"""Lazily import the computers plugin models.
@@ -131,13 +169,17 @@ def _check_collector_auth(expected_key, pluginname=None):
api_key = request.headers.get('X-API-Key')
+ g.collectorauth = None
+
token = _collector_managed_token(api_key)
if token is not None and _token_is_collector(token, pluginname):
touch_apitoken_lastused(token)
+ g.collectorauth = AUTH_MANAGED_TOKEN
return None
if expected_key and api_key == expected_key \
and not _env_fallback_disabled(pluginname):
+ g.collectorauth = AUTH_SHARED_KEY
return None
# Fail-closed only on pure server misconfiguration: no env key AND the
@@ -164,6 +206,54 @@ def require_api_key(f):
return decorated
+def _legacy_shared_key_refused():
+ """True when the site has closed the legacy routes to the shared key.
+
+ Absent setting = false = the routes stay open, which is what every site
+ does today. Read tolerantly because the row can be created by hand through
+ the settings API, where it may land with valuetype string.
+ """
+ from shopdb.core.models.setting import Setting
+
+ value = Setting.get(LEGACY_SHARED_KEY_SETTING, default=False)
+ if isinstance(value, str):
+ return value.strip().lower() in ('true', '1', 'yes')
+ return bool(value)
+
+
+def require_api_key_legacy(f):
+ """require_api_key plus the legacy-route shared-key gate.
+
+ These routes take their target hostname from the body, so the shared key is
+ an any-host write key on them and no per-PC credential narrows it. A site
+ that has finished the move to per-PC credentials can shut it off here
+ without waiting for the routes to be retired.
+
+ Every shared-key use is logged whether the gate is on or off, so a site can
+ read the warnings and learn what still depends on the key BEFORE flipping
+ it - otherwise turning it on is a guess.
+ """
+ from shopdb.utils.clientip import client_ip
+
+ @wraps(f)
+ def decorated(*args, **kwargs):
+ error = _check_collector_auth(current_app.config.get('COLLECTOR_API_KEY'))
+ if error is not None:
+ return error
+ if getattr(g, 'collectorauth', None) == AUTH_SHARED_KEY:
+ if _legacy_shared_key_refused():
+ return error_response(
+ 'SHARED_KEY_REFUSED',
+ 'The shared key is not accepted on the legacy collector '
+ 'routes; use a collector-scoped token',
+ http_code=401)
+ current_app.logger.warning(
+ 'Legacy collector route used with the shared key: %s from %s',
+ request.path, client_ip(request))
+ return f(*args, **kwargs)
+ return decorated
+
+
def _find_pc(hostname):
"""Find a computer by hostname, falling back to its asset number.
@@ -347,7 +437,7 @@ def collector_schemas():
@collector_bp.route('/pc', methods=['POST'])
-@require_api_key
+@require_api_key_legacy
def update_pc_info():
"""Update one PC from the collection script (matched by hostname)."""
data = request.get_json()
@@ -385,7 +475,7 @@ def update_pc_info():
@collector_bp.route('/apps', methods=['POST'])
-@require_api_key
+@require_api_key_legacy
def update_installed_apps():
"""Update installed applications for a PC (matched by hostname)."""
data = request.get_json()
@@ -457,7 +547,7 @@ def update_installed_apps():
@collector_bp.route('/heartbeat', methods=['POST'])
-@require_api_key
+@require_api_key_legacy
def pc_heartbeat():
"""Record PC online status / heartbeat (single or batch)."""
data = request.get_json()
@@ -492,7 +582,7 @@ def pc_heartbeat():
@collector_bp.route('/bulk', methods=['POST'])
-@require_api_key
+@require_api_key_legacy
def bulk_update():
"""Bulk update multiple PCs at once."""
data = request.get_json()
diff --git a/shopdb/core/api/settings.py b/shopdb/core/api/settings.py
index 31b237b4..0582b6b6 100644
--- a/shopdb/core/api/settings.py
+++ b/shopdb/core/api/settings.py
@@ -161,14 +161,50 @@ def _declared_default(key: str) -> dict:
return _plugin_declared_settings().get(key, {})
-def _is_secret(key: str) -> bool:
- return 'password' in key or 'token' in key or 'secret' in key
+def _is_secret(key: str, valuetype=None) -> bool:
+ """True when a key's value must never leave the API in plaintext.
+
+ The test is the key NAME, which is a heuristic, so it is narrowed by the
+ stored type when the caller knows it. A boolean, integer or float setting
+ cannot hold a secret, and masking one breaks the control that renders it:
+ plugin.endpoints.bootstrapquerytoken is a true/false switch whose name
+ contains 'token', and the settings page read back '********' instead of
+ its state, so the checkbox could not show whether it was on.
+
+ valuetype omitted means "judge by name alone", which is the safe direction:
+ an unknown type is treated as a string.
+ """
+ named = 'password' in key or 'token' in key or 'secret' in key
+ if not named:
+ return False
+ return valuetype is None or str(valuetype).lower() == 'string'
+
+
+def _inferred_valuetype(value) -> str:
+ """Valuetype for a key nobody declares, from the JSON type sent.
+
+ Only reached on the upsert path in update_setting. JSON already carries the
+ distinction the row has to keep, so honour it: true/false is a boolean row,
+ a number is integer or float, everything else is a string. Guessing from
+ the TEXT (e.g. the string 'true') is deliberately not done - a genuine
+ string setting whose value happens to read 'true' would then be typed
+ boolean and coerced out from under its reader.
+
+ bool is checked before int because bool subclasses int in Python.
+ """
+ if isinstance(value, bool):
+ return 'boolean'
+ if isinstance(value, int):
+ return 'integer'
+ if isinstance(value, float):
+ return 'float'
+ return 'string'
def _serialize_setting(setting):
"""Serialize a setting, masking secret values so they never leave the API."""
data = setting.to_dict()
- if _is_secret(setting.key):
+ if _is_secret(setting.key, setting.valuetype):
data['value'] = SECRET_MASK if setting.value else ''
return data
@@ -795,7 +831,14 @@ def update_setting(key: str):
setting = Setting(
key=key,
value='',
- valuetype=declared.get('valuetype', 'string'),
+ # An undeclared key used to be born valuetype='string'
+ # unconditionally, and that was a trap: a caller PUTting JSON true
+ # got a string row, get_typed_value then handed every reader the
+ # text 'true', and a reader doing bool(...) or `is True` silently
+ # disagreed with the UI. Infer from the JSON type the client sent
+ # instead, so a boolean key is born boolean.
+ valuetype=declared.get('valuetype') or _inferred_valuetype(
+ data.get('value')),
category=declared.get('category', 'plugin'),
description=declared.get('description'),
)
@@ -808,7 +851,7 @@ def update_setting(key: str):
# A secret submitted as the mask placeholder means "leave unchanged" - the
# client only ever received the mask, so don't overwrite the real secret.
- if _is_secret(key) and value == SECRET_MASK:
+ if _is_secret(key, setting.valuetype) and value == SECRET_MASK:
return success_response(_serialize_setting(setting), message='Setting unchanged')
# The declared valuetype is a promise about what every reader will get, and
@@ -846,7 +889,7 @@ def update_setting(key: str):
setting.value = str(value) if value is not None else None
# Audit log (mask sensitive values)
- is_sensitive = _is_secret(key)
+ is_sensitive = _is_secret(key, setting.valuetype)
AuditLog.log('updated', 'Setting', entityname=key, changes={
'value': {
'old': '***' if is_sensitive else old_value,
@@ -915,10 +958,13 @@ def create_setting():
else:
value_str = str(value) if value is not None else None
+ # Same reasoning as the upsert in update_setting: a caller that omits
+ # valuetype and sends JSON true must not get a string row that every
+ # boolean reader then misreads.
setting = Setting(
key=data['key'],
value=value_str,
- valuetype=data.get('valuetype', 'string'),
+ valuetype=data.get('valuetype') or _inferred_valuetype(value),
category=data.get('category', 'general'),
description=data.get('description')
)
@@ -1268,7 +1314,16 @@ def build_default_settings():
# Collector pc-type -> ComputerType mapping is computers-plugin domain;
# the plugin seeds pctypemap_ settings on install.
+ # Collector authentication. The wording lives with the code that enforces
+ # it (shopdb/core/api/collector.py); seeding it here is what makes the row
+ # exist, and the row is what the settings UI can render a control for.
+ # Imported inside the function because core.api.__init__ imports collector
+ # and settings in the same package - a module-level import would couple
+ # their import order for no gain.
+ from shopdb.core.api.collector import LEGACY_SHARED_KEY_SETTING_DEFAULT
+
defaults = sitedefaults + brandingdefaults + printingdefaults + identifierdefaults + searchdefaults + mapdefaults + [
+ LEGACY_SHARED_KEY_SETTING_DEFAULT,
# ServiceNow ticket links. Each instance points these at its own
# ServiceNow tenant; blank/disabled renders tickets as plain text.
#
diff --git a/tests/test_core/test_collector_legacy_sharedkey.py b/tests/test_core/test_collector_legacy_sharedkey.py
new file mode 100644
index 00000000..abc104b3
--- /dev/null
+++ b/tests/test_core/test_collector_legacy_sharedkey.py
@@ -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
diff --git a/tests/test_core/test_settings_switches_reachable.py b/tests/test_core/test_settings_switches_reachable.py
new file mode 100644
index 00000000..c60ac60c
--- /dev/null
+++ b/tests/test_core/test_settings_switches_reachable.py
@@ -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
diff --git a/tests/test_plugins/test_manifest_core_version_floor.py b/tests/test_plugins/test_manifest_core_version_floor.py
new file mode 100644
index 00000000..5b02632e
--- /dev/null
+++ b/tests/test_plugins/test_manifest_core_version_floor.py
@@ -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__
diff --git a/tests/test_plugins/test_notifications_alertscope_storage.py b/tests/test_plugins/test_notifications_alertscope_storage.py
new file mode 100644
index 00000000..86471a4f
--- /dev/null
+++ b/tests/test_plugins/test_notifications_alertscope_storage.py
@@ -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'