Second-pass review fixes: kill last hardcoded creds, wire get_services, dedup
Verification audit (re-run of the 6 skill lenses) confirmed the prior fixes hold and surfaced a few misses: Security (HIGH): - search.py _check_smart_redirect still opened a raw pymysql connection with root/rootpassword (reachable on any 9-digit SSO query). Now uses the shared env-backed employee_connection helper. - Deleted dead shopdb/core/services/employee_service.py (zero importers; carried another root/rootpassword literal). No hardcoded credentials remain in app logic; config.py dev defaults stay gated by ProductionConfig.validate. Dead hook: - get_services was implemented by the printers plugin but had no consumer (docs claimed otherwise). Added PluginManager.get_service(name) that resolves a service from enabled plugins; updated PLUGIN-HOOKS.md. Tests: - search disabled-plugin exclusion (the high-value gap): enabled plugin's hostname appears, disabled plugin's hostname drops out (searched by a hostname distinct from assetnumber so only the gated domain can match). - get_service consumer test (unknown name -> None). Simplify: - Extract the triplicated GE_LOGO_SVG + loadLogo + drawLogoOverlay into shared frontend/src/views/print/qrLogo.js (renderQrDataUrl); both QR views use it. - applications.py: lift the misplaced pagination import to the top; drop unused Computer unpacking in the 3 endpoints that only touch ComputerInstalledApp. 154 tests pass, naming/style green, app boots, QR render verified. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -522,14 +522,9 @@ def _check_smart_redirect(query, classification):
|
||||
# Exact SSO match
|
||||
if classification['is_sso']:
|
||||
try:
|
||||
import pymysql
|
||||
emp_conn = pymysql.connect(
|
||||
host='localhost',
|
||||
user='root',
|
||||
password='rootpassword',
|
||||
database='wjf_employees',
|
||||
cursorclass=pymysql.cursors.DictCursor
|
||||
)
|
||||
# Shared env-backed connection helper; never hardcode creds.
|
||||
from shopdb.utils.employee_db import employee_connection
|
||||
emp_conn = employee_connection()
|
||||
with emp_conn.cursor() as cur:
|
||||
cur.execute(
|
||||
'SELECT SSO, First_Name, Last_Name FROM employees WHERE SSO = %s LIMIT 1',
|
||||
|
||||
@@ -1,104 +0,0 @@
|
||||
"""Employee lookup service - queries wjf_employees database."""
|
||||
|
||||
from typing import Optional, Dict, List
|
||||
import pymysql
|
||||
from flask import current_app
|
||||
|
||||
|
||||
def get_employee_connection():
|
||||
"""Get connection to wjf_employees database."""
|
||||
return pymysql.connect(
|
||||
host='localhost',
|
||||
user='root',
|
||||
password='rootpassword',
|
||||
database='wjf_employees',
|
||||
cursorclass=pymysql.cursors.DictCursor
|
||||
)
|
||||
|
||||
|
||||
def lookup_employee(sso: str) -> Optional[Dict]:
|
||||
"""
|
||||
Look up employee by SSO.
|
||||
|
||||
Returns dict with: SSO, First_Name, Last_Name, full_name, Picture, etc.
|
||||
"""
|
||||
if not sso or not sso.strip().isdigit():
|
||||
return None
|
||||
|
||||
try:
|
||||
conn = get_employee_connection()
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
'SELECT * FROM employees WHERE SSO = %s',
|
||||
(int(sso.strip()),)
|
||||
)
|
||||
row = cur.fetchone()
|
||||
if row:
|
||||
# Add computed full_name
|
||||
first = (row.get('First_Name') or '').strip()
|
||||
last = (row.get('Last_Name') or '').strip()
|
||||
row['full_name'] = f"{first} {last}".strip()
|
||||
return row
|
||||
conn.close()
|
||||
except Exception as e:
|
||||
current_app.logger.error(f"Employee lookup error: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def lookup_employees(sso_list: str) -> List[Dict]:
|
||||
"""
|
||||
Look up multiple employees by comma-separated SSO list.
|
||||
|
||||
Returns list of employee dicts.
|
||||
"""
|
||||
if not sso_list:
|
||||
return []
|
||||
|
||||
ssos = [s.strip() for s in sso_list.split(',') if s.strip().isdigit()]
|
||||
if not ssos:
|
||||
return []
|
||||
|
||||
try:
|
||||
conn = get_employee_connection()
|
||||
with conn.cursor() as cur:
|
||||
placeholders = ','.join(['%s'] * len(ssos))
|
||||
cur.execute(
|
||||
f'SELECT * FROM employees WHERE SSO IN ({placeholders})',
|
||||
[int(s) for s in ssos]
|
||||
)
|
||||
rows = cur.fetchall()
|
||||
|
||||
# Add computed full_name to each
|
||||
for row in rows:
|
||||
first = (row.get('First_Name') or '').strip()
|
||||
last = (row.get('Last_Name') or '').strip()
|
||||
row['full_name'] = f"{first} {last}".strip()
|
||||
|
||||
return rows
|
||||
conn.close()
|
||||
except Exception as e:
|
||||
current_app.logger.error(f"Employee lookup error: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def get_employee_names(sso_list: str) -> str:
|
||||
"""
|
||||
Get comma-separated list of employee names from SSO list.
|
||||
|
||||
Input: "212574611,212637451"
|
||||
Output: "Brandon Saltz, Jon Kolkmann"
|
||||
"""
|
||||
employees = lookup_employees(sso_list)
|
||||
if not employees:
|
||||
return sso_list # Return SSOs as fallback
|
||||
|
||||
return ', '.join(emp['full_name'] for emp in employees if emp.get('full_name'))
|
||||
|
||||
|
||||
def get_employee_picture_url(sso: str) -> Optional[str]:
|
||||
"""Get URL to employee picture if available."""
|
||||
emp = lookup_employee(sso)
|
||||
if emp and emp.get('Picture'):
|
||||
# Pictures are stored relative paths like "Support/212574611.png"
|
||||
return f"/static/employees/{emp['Picture']}"
|
||||
return None
|
||||
@@ -317,6 +317,25 @@ class PluginManager:
|
||||
"""Get all loaded plugins."""
|
||||
return self.loader.get_all_loaded()
|
||||
|
||||
def get_service(self, name: str):
|
||||
"""Resolve a service exposed by an enabled plugin via get_services().
|
||||
|
||||
Consumer for the BasePlugin.get_services hook: searches enabled plugins
|
||||
for one that registers `name` and returns the registered value (a service
|
||||
class or factory). Returns None if no enabled plugin provides it. This is
|
||||
how one plugin obtains another's service (e.g. the Zabbix service).
|
||||
"""
|
||||
for plugin_name, plugin in self.get_all_plugins().items():
|
||||
if not self.registry.is_enabled(plugin_name):
|
||||
continue
|
||||
try:
|
||||
services = plugin.get_services() or {}
|
||||
except Exception:
|
||||
continue
|
||||
if name in services:
|
||||
return services[name]
|
||||
return None
|
||||
|
||||
|
||||
# Global plugin manager instance
|
||||
plugin_manager = PluginManager()
|
||||
|
||||
Reference in New Issue
Block a user