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

Security: mask settings secrets, remove hardcoded employee-DB creds

Browse Source 
- GET /settings now masks password/token values (were returned in plaintext
  to anonymous callers); sending the mask back on update is a no-op so the
  real secret is never clobbered.
- Move the employee-directory DB credentials out of source into env-backed
  config (shopdb.utils.employee_db); employees + notification recognition use
  the shared helper. Employee lookups stop leaking exception strings.
- Fix low-supplies report using loc.location instead of loc.locationname.

Employee/notification read endpoints stay unauthenticated by design (public
shopfloor kiosk displays consume them).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-06-26 08:42:16 -04:00
parent e631564377
commit b516b9b771
6 changed files with 203 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
plugins/notifications/api/routes.py
View File

@@ -12,6 +12,7 @@ from shopdb.utils.responses import (
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
from shopdb.utils.employee_db import employee_connection
from ..models import Notification, NotificationType
@@ -522,11 +523,7 @@ def get_shopfloor_notifications():
# Try to get picture from wjf_employees
if n.employeesso and n.employeesso.isdigit():
try:
import pymysql
conn = pymysql.connect(
host='localhost', user='root', password='rootpassword',
database='wjf_employees', cursorclass=pymysql.cursors.DictCursor
)
conn = employee_connection()
with conn.cursor() as cur:
cur.execute('SELECT Picture FROM employees WHERE SSO = %s', (int(n.employeesso),))
emp = cur.fetchone()
@@ -555,11 +552,7 @@ def get_shopfloor_notifications():
picture = None
if sso.isdigit():
try:
import pymysql
conn = pymysql.connect(
host='localhost', user='root', password='rootpassword',
database='wjf_employees', cursorclass=pymysql.cursors.DictCursor
)
conn = employee_connection()
with conn.cursor() as cur:
cur.execute('SELECT Picture FROM employees WHERE SSO = %s', (int(sso),))
emp = cur.fetchone()
@@ -591,11 +584,7 @@ def get_shopfloor_notifications():
picture = None
if sso.isdigit():
try:
import pymysql
conn = pymysql.connect(
host='localhost', user='root', password='rootpassword',
database='wjf_employees', cursorclass=pymysql.cursors.DictCursor
)
conn = employee_connection()
with conn.cursor() as cur:
cur.execute('SELECT Picture FROM employees WHERE SSO = %s', (int(sso),))
emp = cur.fetchone()

2
plugins/printers/api/asset_routes.py
View File

@@ -565,7 +565,7 @@ def _get_low_supplies_data():
from shopdb.core.models import Location
loc = Location.query.get(asset.locationid)
if loc:
location_name = loc.location
location_name = loc.locationname
results.append({
'printerid': printer.printerid,

7
shopdb/config.py
View File

@@ -57,6 +57,13 @@ class Config:
ZABBIX_URL = os.environ.get('ZABBIX_URL', '')
ZABBIX_TOKEN = os.environ.get('ZABBIX_TOKEN', '')
# Read-only HR/employee directory database (separate from the app DB).
# Credentials come from the environment; never hardcode them in source.
EMPLOYEE_DB_HOST = os.environ.get('EMPLOYEE_DB_HOST', 'localhost')
EMPLOYEE_DB_USER = os.environ.get('EMPLOYEE_DB_USER', 'root')
EMPLOYEE_DB_PASSWORD = os.environ.get('EMPLOYEE_DB_PASSWORD', 'rootpassword')
EMPLOYEE_DB_NAME = os.environ.get('EMPLOYEE_DB_NAME', 'wjf_employees')
CACHE_TYPE = 'SimpleCache'
CACHE_DEFAULT_TIMEOUT = 600

305
shopdb/core/api/employees.py
View File

@@ -1,161 +1,144 @@
"""Employee lookup API endpoints."""
from flask import Blueprint, request
from shopdb.utils.responses import success_response, error_response, ErrorCodes
employees_bp = Blueprint('employees', __name__)
@employees_bp.route('/search', methods=['GET'])
def search_employees():
"""
Search employees by name.
Query parameters:
- q: Search query (searches first and last name)
- limit: Max results (default 10)
"""
query = request.args.get('q', '').strip()
limit = min(int(request.args.get('limit', 10)), 50)
if len(query) < 2:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'Search query must be at least 2 characters'
)
try:
import pymysql
conn = pymysql.connect(
host='localhost',
user='root',
password='rootpassword',
database='wjf_employees',
cursorclass=pymysql.cursors.DictCursor
)
with conn.cursor() as cur:
# Search by first name, last name, or SSO
cur.execute('''
SELECT SSO, First_Name, Last_Name, Team, Role, Picture
FROM employees
WHERE First_Name LIKE %s
OR Last_Name LIKE %s
OR CAST(SSO AS CHAR) LIKE %s
ORDER BY Last_Name, First_Name
LIMIT %s
''', (f'%{query}%', f'%{query}%', f'%{query}%', limit))
employees = cur.fetchall()
conn.close()
return success_response(employees)
except Exception as e:
return error_response(
ErrorCodes.DATABASE_ERROR,
f'Employee lookup failed: {str(e)}',
http_code=500
)
@employees_bp.route('/lookup/<sso>', methods=['GET'])
def lookup_employee(sso):
"""Look up a single employee by SSO."""
if not sso.isdigit():
return error_response(
ErrorCodes.VALIDATION_ERROR,
'SSO must be numeric'
)
try:
import pymysql
conn = pymysql.connect(
host='localhost',
user='root',
password='rootpassword',
database='wjf_employees',
cursorclass=pymysql.cursors.DictCursor
)
with conn.cursor() as cur:
cur.execute(
'SELECT SSO, First_Name, Last_Name, Team, Role, Picture FROM employees WHERE SSO = %s',
(int(sso),)
)
employee = cur.fetchone()
conn.close()
if not employee:
return error_response(
ErrorCodes.NOT_FOUND,
f'Employee with SSO {sso} not found',
http_code=404
)
return success_response(employee)
except Exception as e:
return error_response(
ErrorCodes.DATABASE_ERROR,
f'Employee lookup failed: {str(e)}',
http_code=500
)
@employees_bp.route('/lookup', methods=['GET'])
def lookup_employees():
"""
Look up multiple employees by SSO list.
Query parameters:
- sso: Comma-separated list of SSOs
"""
sso_list = request.args.get('sso', '')
ssos = [s.strip() for s in sso_list.split(',') if s.strip().isdigit()]
if not ssos:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'At least one valid SSO is required'
)
try:
import pymysql
conn = pymysql.connect(
host='localhost',
user='root',
password='rootpassword',
database='wjf_employees',
cursorclass=pymysql.cursors.DictCursor
)
with conn.cursor() as cur:
placeholders = ','.join(['%s'] * len(ssos))
cur.execute(
f'SELECT SSO, First_Name, Last_Name, Team, Role, Picture FROM employees WHERE SSO IN ({placeholders})',
[int(s) for s in ssos]
)
employees = cur.fetchall()
conn.close()
# Build name string
names = ', '.join(
f"{e['First_Name'].strip()} {e['Last_Name'].strip()}"
for e in employees
)
return success_response({
'employees': employees,
'names': names
})
except Exception as e:
return error_response(
ErrorCodes.DATABASE_ERROR,
f'Employee lookup failed: {str(e)}',
http_code=500
)
"""Employee lookup API endpoints.
These read from the separate employee directory DB (see shopdb.utils.employee_db).
They are intentionally reachable by the unauthenticated shopfloor kiosk displays
(recognition wall), so they are not JWT-gated; keep them read-only and never
return more than the directory fields below.
"""
import logging
from flask import Blueprint, request
from shopdb.utils.responses import success_response, error_response, ErrorCodes
from shopdb.utils.employee_db import employee_connection
logger = logging.getLogger(__name__)
employees_bp = Blueprint('employees', __name__)
# Columns safe to expose to the directory/recognition UI
_FIELDS = 'SSO, First_Name, Last_Name, Team, Role, Picture'
@employees_bp.route('/search', methods=['GET'])
def search_employees():
"""
Search employees by name.
Query parameters:
- q: Search query (searches first and last name)
- limit: Max results (default 10)
"""
query = request.args.get('q', '').strip()
limit = min(int(request.args.get('limit', 10)), 50)
if len(query) < 2:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'Search query must be at least 2 characters'
)
try:
conn = employee_connection()
with conn.cursor() as cur:
cur.execute(f'''
SELECT {_FIELDS}
FROM employees
WHERE First_Name LIKE %s
OR Last_Name LIKE %s
OR CAST(SSO AS CHAR) LIKE %s
ORDER BY Last_Name, First_Name
LIMIT %s
''', (f'%{query}%', f'%{query}%', f'%{query}%', limit))
employees = cur.fetchall()
conn.close()
return success_response(employees)
except Exception:
logger.exception('Employee search failed')
return error_response(
ErrorCodes.DATABASE_ERROR,
'Employee lookup failed',
http_code=500
)
@employees_bp.route('/lookup/<sso>', methods=['GET'])
def lookup_employee(sso):
"""Look up a single employee by SSO."""
if not sso.isdigit():
return error_response(
ErrorCodes.VALIDATION_ERROR,
'SSO must be numeric'
)
try:
conn = employee_connection()
with conn.cursor() as cur:
cur.execute(
f'SELECT {_FIELDS} FROM employees WHERE SSO = %s',
(int(sso),)
)
employee = cur.fetchone()
conn.close()
if not employee:
return error_response(
ErrorCodes.NOT_FOUND,
f'Employee with SSO {sso} not found',
http_code=404
)
return success_response(employee)
except Exception:
logger.exception('Employee lookup failed for SSO %s', sso)
return error_response(
ErrorCodes.DATABASE_ERROR,
'Employee lookup failed',
http_code=500
)
@employees_bp.route('/lookup', methods=['GET'])
def lookup_employees():
"""
Look up multiple employees by SSO list.
Query parameters:
- sso: Comma-separated list of SSOs
"""
sso_list = request.args.get('sso', '')
ssos = [s.strip() for s in sso_list.split(',') if s.strip().isdigit()]
if not ssos:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'At least one valid SSO is required'
)
try:
conn = employee_connection()
with conn.cursor() as cur:
placeholders = ','.join(['%s'] * len(ssos))
cur.execute(
f'SELECT {_FIELDS} FROM employees WHERE SSO IN ({placeholders})',
[int(s) for s in ssos]
)
employees = cur.fetchall()
conn.close()
names = ', '.join(
f"{e['First_Name'].strip()} {e['Last_Name'].strip()}"
for e in employees
)
return success_response({
'employees': employees,
'names': names
})
except Exception:
logger.exception('Employee multi-lookup failed')
return error_response(
ErrorCodes.DATABASE_ERROR,
'Employee lookup failed',
http_code=500
)

32
shopdb/core/api/settings.py
View File

@@ -13,6 +13,22 @@ settings_bp = Blueprint('settings', __name__)
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 = '********'
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."""
@@ -42,7 +58,7 @@ def list_settings():
query = query.filter_by(category=category)
settings = query.order_by(Setting.category, Setting.key).all()
return success_response([s.to_dict() for s in settings])
return success_response([_serialize_setting(s) for s in settings])
@settings_bp.route('/<key>', methods=['GET'])
@@ -54,7 +70,7 @@ def get_setting(key: str):
if not setting:
return error_response(ErrorCodes.NOT_FOUND, f'Setting {key} not found', http_code=404)
return success_response(setting.to_dict())
return success_response(_serialize_setting(setting))
@settings_bp.route('/<key>', methods=['PUT'])
@@ -74,15 +90,21 @@ def update_setting(key: str):
# Track old value for audit
old_value = setting.value
# Convert value to string for storage
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 = 'password' in key or 'token' in key or 'secret' in key
is_sensitive = _is_secret(key)
AuditLog.log('updated', 'Setting', entityname=key, changes={
'value': {
'old': '***' if is_sensitive else old_value,
@@ -93,7 +115,7 @@ def update_setting(key: str):
db.session.commit()
invalidate_settings_cache()
return success_response(setting.to_dict(), message='Setting updated')
return success_response(_serialize_setting(setting), message='Setting updated')
@settings_bp.route('', methods=['POST'])

20
shopdb/utils/employee_db.py Normal file
View File

@@ -0,0 +1,20 @@
"""Connection helper for the read-only employee directory database.
Credentials are pulled from app config (env-backed, see Config.EMPLOYEE_DB_*),
never hardcoded. Used by the employee lookup API and the notification
recognition feature.
"""
import pymysql
from flask import current_app
def employee_connection():
"""Open a pymysql connection to the employee directory DB."""
return pymysql.connect(
host=current_app.config['EMPLOYEE_DB_HOST'],
user=current_app.config['EMPLOYEE_DB_USER'],
password=current_app.config['EMPLOYEE_DB_PASSWORD'],
database=current_app.config['EMPLOYEE_DB_NAME'],
cursorclass=pymysql.cursors.DictCursor,
)