diff --git a/plugins/notifications/api/routes.py b/plugins/notifications/api/routes.py index 486b6a2..fac86e4 100644 --- a/plugins/notifications/api/routes.py +++ b/plugins/notifications/api/routes.py @@ -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() diff --git a/plugins/printers/api/asset_routes.py b/plugins/printers/api/asset_routes.py index dd7237f..8ff7463 100644 --- a/plugins/printers/api/asset_routes.py +++ b/plugins/printers/api/asset_routes.py @@ -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, diff --git a/shopdb/config.py b/shopdb/config.py index eb910d5..655b7ca 100644 --- a/shopdb/config.py +++ b/shopdb/config.py @@ -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 diff --git a/shopdb/core/api/employees.py b/shopdb/core/api/employees.py index 42791e8..5624655 100644 --- a/shopdb/core/api/employees.py +++ b/shopdb/core/api/employees.py @@ -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/', 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/', 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 + ) diff --git a/shopdb/core/api/settings.py b/shopdb/core/api/settings.py index 023246f..9f058d0 100644 --- a/shopdb/core/api/settings.py +++ b/shopdb/core/api/settings.py @@ -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('/', 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('/', 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']) diff --git a/shopdb/utils/employee_db.py b/shopdb/utils/employee_db.py new file mode 100644 index 0000000..7e6b7e9 --- /dev/null +++ b/shopdb/utils/employee_db.py @@ -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, + )