Files
shopdb-flask/shopdb/core/api/settings.py
cproudlock 5fa5160420 Apply skill-driven review fixes: security, hook isolation, tests, docs
Addresses findings from a 6-lens review against the project skills
(defining-asset-contract, enforcing-plugin-contract, hardening-flask-config,
integrating-plugin-hooks, pinning-flask-behavior, simplifying-python).

Security (hardening-flask-config):
- Load per-plugin COLLECTOR_API_KEY_<PLUGIN> from env in create_app. from_object
  only copies class attributes, so per-plugin keys (ADR-006) were dead in real
  deploys and silently fell back to the shared key.
- EMPLOYEE_DB_USER/PASSWORD no longer default to root/rootpassword (no safe
  default for a secret; unset fails loud). Documented in .env.example + DEPLOY.md.
- COLLECTOR_API_KEY + per-plugin + EMPLOYEE_DB_* added to .env.example/DEPLOY.md.

Hook isolation (integrating-plugin-hooks):
- collector _collector_plugins and dashboard get_navigation now re-raise in
  dev/test and log+isolate in prod, instead of silently swallowing a broken
  plugin hook.

Plugin loader (enforcing-plugin-contract):
- enable_plugin/install_plugin read dependencies+version from the manifest
  instead of instantiating the plugin class.
- _register_plugin_components rejects a second plugin claiming an already-used
  api_prefix (reset per app in init_app).

Tests (pinning-flask-behavior):
- test_identifiers.py: gauge/maintenance round-trip on computer/printer/network
  create+update; per-type seed yields the 12 identifier keys.
- contract tests for apply_collector_payload presence + schema-declarers-implement.
- security tests for per-plugin key env loading + no employee-db password default.

Docs/contract sync (defining-asset-contract):
- PLUGIN-HOOKS.md documents apply_collector_payload; stale 0.2.0 -> 0.3.0.
- ADR-006 documents apply_collector_payload + single-dispatch rationale.
- ADR-001 enumerates the expanded shopdb.api import surface.

Simplify (simplifying-python):
- De-duplicate the 21-entry settings defaults: shared build_default_settings()
  used by both the /settings/seed route and the CLI (were drifting copies).
- Remove dead AssetStatus import + redundant AssetType local import in computers
  plugin; comment the statusid=1 collector default.

153 tests pass (was 145), naming/style green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-26 19:25:52 -04:00

350 lines
11 KiB
Python

"""Settings API routes."""
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.extensions import db, cache
from shopdb.core.models import Setting, AuditLog
from shopdb.utils.responses import success_response, error_response, ErrorCodes
settings_bp = Blueprint('settings', __name__)
# Cache key for settings
SETTINGS_CACHE_KEY = 'system_settings'
SETTINGS_CACHE_TTL = 300 # 5 minutes
# Placeholder returned in API responses for secret values so they are never
# exposed in plaintext. Sending it back on update is treated as "unchanged".
SECRET_MASK = '********'
# Optional asset identifiers and the asset types they can be toggled on.
# Drives per-type seed keys and the Settings matrix UI. The asset type names
# match the AssetType.assettype values seeded by each plugin.
IDENTIFIER_LABELS = {
'gaugelabreference': 'Gauge Lab Reference',
'maintenancereference': 'Maintenance Reference',
'fqdn': 'FQDN / hostname',
}
IDENTIFIER_ASSETTYPES = ['equipment', 'computer', 'printer', 'network_device']
def _is_secret(key: str) -> bool:
return 'password' in key or 'token' in key or 'secret' in key
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):
data['value'] = SECRET_MASK if setting.value else ''
return data
def get_cached_settings():
"""Get all settings from cache or database."""
cached = cache.get(SETTINGS_CACHE_KEY)
if cached is not None:
return cached
settings = Setting.query.all()
result = {s.key: s.get_typed_value() for s in settings}
cache.set(SETTINGS_CACHE_KEY, result, timeout=SETTINGS_CACHE_TTL)
return result
def invalidate_settings_cache():
"""Clear the settings cache."""
cache.delete(SETTINGS_CACHE_KEY)
@settings_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_settings():
"""List all settings, optionally filtered by category."""
category = request.args.get('category')
query = Setting.query
if category:
query = query.filter_by(category=category)
settings = query.order_by(Setting.category, Setting.key).all()
return success_response([_serialize_setting(s) for s in settings])
@settings_bp.route('/<key>', methods=['GET'])
@jwt_required(optional=True)
def get_setting(key: str):
"""Get a single setting by key."""
setting = Setting.query.filter_by(key=key).first()
if not setting:
return error_response(ErrorCodes.NOT_FOUND, f'Setting {key} not found', http_code=404)
return success_response(_serialize_setting(setting))
@settings_bp.route('/<key>', methods=['PUT'])
@jwt_required()
def update_setting(key: str):
"""Update a setting value."""
data = request.get_json()
if data is None or 'value' not in data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'value is required')
setting = Setting.query.filter_by(key=key).first()
if not setting:
return error_response(ErrorCodes.NOT_FOUND, f'Setting {key} not found', http_code=404)
# Track old value for audit
old_value = setting.value
value = data['value']
# 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:
return success_response(_serialize_setting(setting), message='Setting unchanged')
# Convert value to string for storage
if isinstance(value, bool):
setting.value = 'true' if value else 'false'
else:
setting.value = str(value) if value is not None else None
# Audit log (mask sensitive values)
is_sensitive = _is_secret(key)
AuditLog.log('updated', 'Setting', entityname=key, changes={
'value': {
'old': '***' if is_sensitive else old_value,
'new': '***' if is_sensitive else setting.value
}
})
db.session.commit()
invalidate_settings_cache()
return success_response(_serialize_setting(setting), message='Setting updated')
@settings_bp.route('', methods=['POST'])
@jwt_required()
def create_setting():
"""Create a new setting (admin only)."""
data = request.get_json()
if not data or not data.get('key'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'key is required')
if Setting.query.filter_by(key=data['key']).first():
return error_response(ErrorCodes.CONFLICT, f"Setting '{data['key']}' already exists", http_code=409)
value = data.get('value')
if isinstance(value, bool):
value_str = 'true' if value else 'false'
else:
value_str = str(value) if value is not None else None
setting = Setting(
key=data['key'],
value=value_str,
valuetype=data.get('valuetype', 'string'),
category=data.get('category', 'general'),
description=data.get('description')
)
db.session.add(setting)
db.session.commit()
invalidate_settings_cache()
return success_response(setting.to_dict(), message='Setting created', http_code=201)
def build_default_settings():
"""Return the full default-settings list (identifier toggles + static).
Shared by the /settings/seed route and the `flask seed settings` CLI so
the two definitions never drift.
"""
# Asset identifier feature toggles, per identifier AND per asset type.
# Key format: identifier_<name>_<assettype>_enabled (boolean). Admins pick
# which optional identifiers show on which asset types. See ADR-001.
identifierdefaults = [
{
'key': f'identifier_{name}_{assettype}_enabled',
'value': 'true',
'valuetype': 'boolean',
'category': 'identifiers',
'description': f'Show the {label} identifier on {assettype} assets',
}
for name, label in IDENTIFIER_LABELS.items()
for assettype in IDENTIFIER_ASSETTYPES
]
defaults = identifierdefaults + [
# Zabbix integration
{
'key': 'zabbix_enabled',
'value': 'false',
'valuetype': 'boolean',
'category': 'integrations',
'description': 'Enable Zabbix integration for printer supply monitoring'
},
{
'key': 'zabbix_url',
'value': '',
'valuetype': 'string',
'category': 'integrations',
'description': 'Zabbix API URL (e.g., http://zabbix.example.com:8080)'
},
{
'key': 'zabbix_token',
'value': '',
'valuetype': 'string',
'category': 'integrations',
'description': 'Zabbix API authentication token'
},
# Email/SMTP settings
{
'key': 'smtp_enabled',
'value': 'false',
'valuetype': 'boolean',
'category': 'email',
'description': 'Enable email notifications and alerts'
},
{
'key': 'smtp_host',
'value': '',
'valuetype': 'string',
'category': 'email',
'description': 'SMTP server hostname'
},
{
'key': 'smtp_port',
'value': '587',
'valuetype': 'integer',
'category': 'email',
'description': 'SMTP server port (usually 587 for TLS, 465 for SSL, 25 for unencrypted)'
},
{
'key': 'smtp_username',
'value': '',
'valuetype': 'string',
'category': 'email',
'description': 'SMTP authentication username'
},
{
'key': 'smtp_password',
'value': '',
'valuetype': 'string',
'category': 'email',
'description': 'SMTP authentication password'
},
{
'key': 'smtp_use_tls',
'value': 'true',
'valuetype': 'boolean',
'category': 'email',
'description': 'Use TLS encryption for SMTP connection'
},
{
'key': 'smtp_from_address',
'value': '',
'valuetype': 'string',
'category': 'email',
'description': 'From address for outgoing emails'
},
{
'key': 'smtp_from_name',
'value': 'ShopDB',
'valuetype': 'string',
'category': 'email',
'description': 'From name for outgoing emails'
},
{
'key': 'alert_recipients',
'value': '',
'valuetype': 'string',
'category': 'email',
'description': 'Default email recipients for alerts (comma-separated)'
},
# Audit log settings
{
'key': 'audit_retention_days',
'value': '90',
'valuetype': 'integer',
'category': 'audit',
'description': 'Number of days to retain audit logs (0 = keep forever)'
},
# Authentication settings
{
'key': 'saml_enabled',
'value': 'false',
'valuetype': 'boolean',
'category': 'auth',
'description': 'Enable SAML SSO authentication'
},
{
'key': 'saml_idp_metadata_url',
'value': '',
'valuetype': 'string',
'category': 'auth',
'description': 'SAML Identity Provider metadata URL'
},
{
'key': 'saml_entity_id',
'value': '',
'valuetype': 'string',
'category': 'auth',
'description': 'SAML Service Provider entity ID (e.g., https://shopdb.example.com)'
},
{
'key': 'saml_acs_url',
'value': '',
'valuetype': 'string',
'category': 'auth',
'description': 'SAML Assertion Consumer Service URL'
},
{
'key': 'saml_allow_local_login',
'value': 'true',
'valuetype': 'boolean',
'category': 'auth',
'description': 'Allow local username/password login when SAML is enabled'
},
{
'key': 'saml_auto_create_users',
'value': 'true',
'valuetype': 'boolean',
'category': 'auth',
'description': 'Automatically create users on first SAML login'
},
{
'key': 'saml_admin_group',
'value': '',
'valuetype': 'string',
'category': 'auth',
'description': 'SAML group name that grants admin role'
},
]
return defaults
@settings_bp.route('/seed', methods=['POST'])
@jwt_required()
def seed_default_settings():
"""Seed default settings if they don't exist."""
created = 0
for d in build_default_settings():
if not Setting.query.filter_by(key=d['key']).first():
setting = Setting(**d)
db.session.add(setting)
created += 1
db.session.commit()
invalidate_settings_cache()
return success_response({'created': created}, message=f'{created} default settings created')