Review safe-polish: docs accuracy, dead imports, no-emoji, geenforce robustness
From the full multi-agent review (0 high, 7 medium, 17 low findings). Applies the mechanical, low-risk items; design/policy findings left for a decision. Docs accuracy: CLAUDE.md contract 0.10.0 -> 0.11.0 and both stale Alembic head citations -> 7d24_customfield_searchable / 31 migrations; Dockerfile bundled- plugin comment fixed (drop nonexistent "equipment", add machines + measuringtools, count eleven). Style/naming (LOCKED rules): remove a CSS-escaped pushpin emoji before location search results (no-emoji policy); rename ManifestEditor shareRoot -> shareroot (variable mirrors the API field verbatim). Dead code: remove confirmed-unused imports across ~20 modules (require_role/ require_permission scaffold residue, stray db/Vendor/Model/current_user/Optional/ error_response); drop unused build_scope import + a stale GEENFORCE_API_KEY docstring clause in geenforce. Migration files left untouched. Correctness: geenforce ingest robustness - record_enforcement_report now 400s on a non-dict counts / non-list results instead of 500; _apply_app_link ignores a non-numeric appid per its docstring instead of 500. Regression tests added. Backend query.get sweep finished: auth.py refresh -> db.session.get (last one). 910 backend tests pass; pyflakes clean; naming green; frontend build green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -54,7 +54,7 @@ def _require_computer_models():
|
||||
return models, None
|
||||
|
||||
|
||||
from shopdb.utils.authz import require_permission, require_role
|
||||
from shopdb.utils.authz import require_permission
|
||||
|
||||
applications_bp = Blueprint('applications', __name__)
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@ from shopdb.utils.responses import (
|
||||
)
|
||||
from shopdb.utils.pagination import get_pagination_params, paginate_query
|
||||
|
||||
from shopdb.utils.authz import require_permission, require_role
|
||||
from shopdb.utils.authz import require_permission
|
||||
from shopdb.utils.import_mode import apply_import_timestamps
|
||||
|
||||
assets_bp = Blueprint('assets', __name__)
|
||||
|
||||
@@ -1,192 +1,192 @@
|
||||
"""Audit log API routes."""
|
||||
|
||||
from flask import Blueprint, request
|
||||
from flask_jwt_extended import jwt_required
|
||||
|
||||
from shopdb.core.models import AuditLog
|
||||
from shopdb.utils.responses import success_response, error_response, ErrorCodes
|
||||
|
||||
auditlogs_bp = Blueprint('auditlogs', __name__)
|
||||
|
||||
|
||||
def _resolve_full_names(ssos):
|
||||
"""Best-effort SSO -> full name map for audit rows. Mode-aware, guarded,
|
||||
degrades to {} so the list still renders if the directory is unreachable."""
|
||||
wanted = {s for s in ssos if s and str(s).isdigit()}
|
||||
if not wanted:
|
||||
return {}
|
||||
names = {}
|
||||
# Self-hosted directory first (no external dependency).
|
||||
try:
|
||||
from plugins.employees.models import DirectoryEmployee
|
||||
rows = DirectoryEmployee.query.filter(
|
||||
DirectoryEmployee.sso.in_(wanted)).all()
|
||||
for emp in rows:
|
||||
full = f'{(emp.firstname or "").strip()} {(emp.lastname or "").strip()}'.strip()
|
||||
if full:
|
||||
names[str(emp.sso)] = full
|
||||
except Exception:
|
||||
pass
|
||||
missing = wanted - set(names)
|
||||
if not missing:
|
||||
return names
|
||||
# External HR directory for anything still unresolved.
|
||||
try:
|
||||
from shopdb.utils.employee_db import employee_connection
|
||||
conn = employee_connection()
|
||||
placeholders = ','.join(['%s'] * len(missing))
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
'SELECT SSO, First_Name, Last_Name FROM employees '
|
||||
f'WHERE SSO IN ({placeholders})', tuple(missing))
|
||||
for emp in cur.fetchall():
|
||||
full = f'{(emp["First_Name"] or "").strip()} {(emp["Last_Name"] or "").strip()}'.strip()
|
||||
if full:
|
||||
names[str(emp['SSO'])] = full
|
||||
conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
return names
|
||||
|
||||
|
||||
@auditlogs_bp.route('', methods=['GET'])
|
||||
@jwt_required()
|
||||
def list_auditlogs():
|
||||
"""
|
||||
List audit logs with filtering and pagination.
|
||||
|
||||
Query params:
|
||||
page: Page number (default 1)
|
||||
perpage: Items per page (default 50, max 200)
|
||||
action: Filter by action (created, updated, deleted)
|
||||
entitytype: Filter by entity type
|
||||
userid: Filter by user ID
|
||||
search: Search in entityname or username
|
||||
from_date: Filter from date (ISO format)
|
||||
to_date: Filter to date (ISO format)
|
||||
"""
|
||||
page = request.args.get('page', 1, type=int)
|
||||
perpage = min(request.args.get('perpage', 50, type=int), 200)
|
||||
|
||||
query = AuditLog.query
|
||||
|
||||
# Filters
|
||||
action = request.args.get('action')
|
||||
if action:
|
||||
query = query.filter(AuditLog.action == action)
|
||||
|
||||
entitytype = request.args.get('entitytype')
|
||||
if entitytype:
|
||||
query = query.filter(AuditLog.entitytype == entitytype)
|
||||
|
||||
userid = request.args.get('userid', type=int)
|
||||
if userid:
|
||||
query = query.filter(AuditLog.userid == userid)
|
||||
|
||||
search = request.args.get('search')
|
||||
if search:
|
||||
search_term = f'%{search}%'
|
||||
query = query.filter(
|
||||
(AuditLog.entityname.ilike(search_term)) |
|
||||
(AuditLog.username.ilike(search_term))
|
||||
)
|
||||
|
||||
from_date = request.args.get('from_date')
|
||||
if from_date:
|
||||
from datetime import datetime
|
||||
try:
|
||||
dt = datetime.fromisoformat(from_date.replace('Z', '+00:00'))
|
||||
query = query.filter(AuditLog.timestamp >= dt)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
to_date = request.args.get('to_date')
|
||||
if to_date:
|
||||
from datetime import datetime
|
||||
try:
|
||||
dt = datetime.fromisoformat(to_date.replace('Z', '+00:00'))
|
||||
query = query.filter(AuditLog.timestamp <= dt)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Order by most recent first
|
||||
query = query.order_by(AuditLog.timestamp.desc())
|
||||
|
||||
# Paginate
|
||||
pagination = query.paginate(page=page, per_page=perpage, error_out=False)
|
||||
|
||||
rows = [log.to_dict() for log in pagination.items]
|
||||
# Attach full names (hover tooltip on the SSO); best-effort.
|
||||
fullnames = _resolve_full_names({r.get('username') for r in rows})
|
||||
for r in rows:
|
||||
r['userfullname'] = fullnames.get(str(r.get('username')))
|
||||
|
||||
return success_response(
|
||||
rows,
|
||||
meta={
|
||||
'page': page,
|
||||
'perpage': perpage,
|
||||
'total': pagination.total,
|
||||
'pages': pagination.pages
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@auditlogs_bp.route('/entity/<entitytype>/<int:entityid>', methods=['GET'])
|
||||
@jwt_required()
|
||||
def get_entity_history(entitytype: str, entityid: int):
|
||||
"""Get audit history for a specific entity."""
|
||||
logs = AuditLog.query.filter_by(
|
||||
entitytype=entitytype,
|
||||
entityid=entityid
|
||||
).order_by(AuditLog.timestamp.desc()).all()
|
||||
|
||||
return success_response([log.to_dict() for log in logs])
|
||||
|
||||
|
||||
@auditlogs_bp.route('/stats', methods=['GET'])
|
||||
@jwt_required()
|
||||
def get_stats():
|
||||
"""Get audit log statistics."""
|
||||
from sqlalchemy import func
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
# Actions by type
|
||||
actions = db_func_count_by(AuditLog.action)
|
||||
|
||||
# Entity types
|
||||
entities = db_func_count_by(AuditLog.entitytype)
|
||||
|
||||
# Recent activity (last 7 days)
|
||||
week_ago = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=7)
|
||||
recent_count = AuditLog.query.filter(AuditLog.timestamp >= week_ago).count()
|
||||
|
||||
# Most active users (last 7 days)
|
||||
from shopdb.extensions import db
|
||||
active_users = db.session.query(
|
||||
AuditLog.username,
|
||||
func.count(AuditLog.auditlogid).label('count')
|
||||
).filter(
|
||||
AuditLog.timestamp >= week_ago,
|
||||
AuditLog.username.isnot(None)
|
||||
).group_by(AuditLog.username).order_by(func.count(AuditLog.auditlogid).desc()).limit(5).all()
|
||||
|
||||
return success_response({
|
||||
'actions': actions,
|
||||
'entities': entities,
|
||||
'recentCount': recent_count,
|
||||
'activeUsers': [{'username': u[0], 'count': u[1]} for u in active_users]
|
||||
})
|
||||
|
||||
|
||||
def db_func_count_by(column):
|
||||
"""Helper to count grouped by a column."""
|
||||
from sqlalchemy import func
|
||||
from shopdb.extensions import db
|
||||
|
||||
results = db.session.query(
|
||||
column,
|
||||
func.count().label('count')
|
||||
).group_by(column).all()
|
||||
|
||||
return {r[0]: r[1] for r in results}
|
||||
"""Audit log API routes."""
|
||||
|
||||
from flask import Blueprint, request
|
||||
from flask_jwt_extended import jwt_required
|
||||
|
||||
from shopdb.core.models import AuditLog
|
||||
from shopdb.utils.responses import success_response
|
||||
|
||||
auditlogs_bp = Blueprint('auditlogs', __name__)
|
||||
|
||||
|
||||
def _resolve_full_names(ssos):
|
||||
"""Best-effort SSO -> full name map for audit rows. Mode-aware, guarded,
|
||||
degrades to {} so the list still renders if the directory is unreachable."""
|
||||
wanted = {s for s in ssos if s and str(s).isdigit()}
|
||||
if not wanted:
|
||||
return {}
|
||||
names = {}
|
||||
# Self-hosted directory first (no external dependency).
|
||||
try:
|
||||
from plugins.employees.models import DirectoryEmployee
|
||||
rows = DirectoryEmployee.query.filter(
|
||||
DirectoryEmployee.sso.in_(wanted)).all()
|
||||
for emp in rows:
|
||||
full = f'{(emp.firstname or "").strip()} {(emp.lastname or "").strip()}'.strip()
|
||||
if full:
|
||||
names[str(emp.sso)] = full
|
||||
except Exception:
|
||||
pass
|
||||
missing = wanted - set(names)
|
||||
if not missing:
|
||||
return names
|
||||
# External HR directory for anything still unresolved.
|
||||
try:
|
||||
from shopdb.utils.employee_db import employee_connection
|
||||
conn = employee_connection()
|
||||
placeholders = ','.join(['%s'] * len(missing))
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
'SELECT SSO, First_Name, Last_Name FROM employees '
|
||||
f'WHERE SSO IN ({placeholders})', tuple(missing))
|
||||
for emp in cur.fetchall():
|
||||
full = f'{(emp["First_Name"] or "").strip()} {(emp["Last_Name"] or "").strip()}'.strip()
|
||||
if full:
|
||||
names[str(emp['SSO'])] = full
|
||||
conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
return names
|
||||
|
||||
|
||||
@auditlogs_bp.route('', methods=['GET'])
|
||||
@jwt_required()
|
||||
def list_auditlogs():
|
||||
"""
|
||||
List audit logs with filtering and pagination.
|
||||
|
||||
Query params:
|
||||
page: Page number (default 1)
|
||||
perpage: Items per page (default 50, max 200)
|
||||
action: Filter by action (created, updated, deleted)
|
||||
entitytype: Filter by entity type
|
||||
userid: Filter by user ID
|
||||
search: Search in entityname or username
|
||||
from_date: Filter from date (ISO format)
|
||||
to_date: Filter to date (ISO format)
|
||||
"""
|
||||
page = request.args.get('page', 1, type=int)
|
||||
perpage = min(request.args.get('perpage', 50, type=int), 200)
|
||||
|
||||
query = AuditLog.query
|
||||
|
||||
# Filters
|
||||
action = request.args.get('action')
|
||||
if action:
|
||||
query = query.filter(AuditLog.action == action)
|
||||
|
||||
entitytype = request.args.get('entitytype')
|
||||
if entitytype:
|
||||
query = query.filter(AuditLog.entitytype == entitytype)
|
||||
|
||||
userid = request.args.get('userid', type=int)
|
||||
if userid:
|
||||
query = query.filter(AuditLog.userid == userid)
|
||||
|
||||
search = request.args.get('search')
|
||||
if search:
|
||||
search_term = f'%{search}%'
|
||||
query = query.filter(
|
||||
(AuditLog.entityname.ilike(search_term)) |
|
||||
(AuditLog.username.ilike(search_term))
|
||||
)
|
||||
|
||||
from_date = request.args.get('from_date')
|
||||
if from_date:
|
||||
from datetime import datetime
|
||||
try:
|
||||
dt = datetime.fromisoformat(from_date.replace('Z', '+00:00'))
|
||||
query = query.filter(AuditLog.timestamp >= dt)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
to_date = request.args.get('to_date')
|
||||
if to_date:
|
||||
from datetime import datetime
|
||||
try:
|
||||
dt = datetime.fromisoformat(to_date.replace('Z', '+00:00'))
|
||||
query = query.filter(AuditLog.timestamp <= dt)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Order by most recent first
|
||||
query = query.order_by(AuditLog.timestamp.desc())
|
||||
|
||||
# Paginate
|
||||
pagination = query.paginate(page=page, per_page=perpage, error_out=False)
|
||||
|
||||
rows = [log.to_dict() for log in pagination.items]
|
||||
# Attach full names (hover tooltip on the SSO); best-effort.
|
||||
fullnames = _resolve_full_names({r.get('username') for r in rows})
|
||||
for r in rows:
|
||||
r['userfullname'] = fullnames.get(str(r.get('username')))
|
||||
|
||||
return success_response(
|
||||
rows,
|
||||
meta={
|
||||
'page': page,
|
||||
'perpage': perpage,
|
||||
'total': pagination.total,
|
||||
'pages': pagination.pages
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@auditlogs_bp.route('/entity/<entitytype>/<int:entityid>', methods=['GET'])
|
||||
@jwt_required()
|
||||
def get_entity_history(entitytype: str, entityid: int):
|
||||
"""Get audit history for a specific entity."""
|
||||
logs = AuditLog.query.filter_by(
|
||||
entitytype=entitytype,
|
||||
entityid=entityid
|
||||
).order_by(AuditLog.timestamp.desc()).all()
|
||||
|
||||
return success_response([log.to_dict() for log in logs])
|
||||
|
||||
|
||||
@auditlogs_bp.route('/stats', methods=['GET'])
|
||||
@jwt_required()
|
||||
def get_stats():
|
||||
"""Get audit log statistics."""
|
||||
from sqlalchemy import func
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
# Actions by type
|
||||
actions = db_func_count_by(AuditLog.action)
|
||||
|
||||
# Entity types
|
||||
entities = db_func_count_by(AuditLog.entitytype)
|
||||
|
||||
# Recent activity (last 7 days)
|
||||
week_ago = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=7)
|
||||
recent_count = AuditLog.query.filter(AuditLog.timestamp >= week_ago).count()
|
||||
|
||||
# Most active users (last 7 days)
|
||||
from shopdb.extensions import db
|
||||
active_users = db.session.query(
|
||||
AuditLog.username,
|
||||
func.count(AuditLog.auditlogid).label('count')
|
||||
).filter(
|
||||
AuditLog.timestamp >= week_ago,
|
||||
AuditLog.username.isnot(None)
|
||||
).group_by(AuditLog.username).order_by(func.count(AuditLog.auditlogid).desc()).limit(5).all()
|
||||
|
||||
return success_response({
|
||||
'actions': actions,
|
||||
'entities': entities,
|
||||
'recentCount': recent_count,
|
||||
'activeUsers': [{'username': u[0], 'count': u[1]} for u in active_users]
|
||||
})
|
||||
|
||||
|
||||
def db_func_count_by(column):
|
||||
"""Helper to count grouped by a column."""
|
||||
from sqlalchemy import func
|
||||
from shopdb.extensions import db
|
||||
|
||||
results = db.session.query(
|
||||
column,
|
||||
func.count().label('count')
|
||||
).group_by(column).all()
|
||||
|
||||
return {r[0]: r[1] for r in results}
|
||||
|
||||
@@ -162,7 +162,7 @@ def login():
|
||||
def refresh():
|
||||
"""Refresh access token using refresh token."""
|
||||
user_id = get_jwt_identity()
|
||||
user = User.query.get(int(user_id))
|
||||
user = db.session.get(User, int(user_id))
|
||||
|
||||
if not user or not user.isactive:
|
||||
return error_response(
|
||||
|
||||
@@ -13,7 +13,7 @@ from shopdb.utils.responses import (
|
||||
)
|
||||
from shopdb.utils.pagination import get_pagination_params, paginate_query
|
||||
|
||||
from shopdb.utils.authz import require_permission, require_role
|
||||
from shopdb.utils.authz import require_role
|
||||
from shopdb.utils.import_mode import apply_import_timestamps
|
||||
|
||||
businessunits_bp = Blueprint('businessunits', __name__)
|
||||
|
||||
@@ -13,7 +13,7 @@ from shopdb.utils.responses import (
|
||||
)
|
||||
from shopdb.utils.pagination import get_pagination_params, paginate_query
|
||||
|
||||
from shopdb.utils.authz import require_permission, require_role
|
||||
from shopdb.utils.authz import require_role
|
||||
from shopdb.utils.import_mode import apply_import_timestamps
|
||||
|
||||
locations_bp = Blueprint('locations', __name__)
|
||||
|
||||
@@ -17,7 +17,7 @@ from shopdb.utils.responses import (
|
||||
)
|
||||
from shopdb.utils.pagination import get_pagination_params, paginate_query
|
||||
|
||||
from shopdb.utils.authz import require_permission, require_role
|
||||
from shopdb.utils.authz import require_role
|
||||
from shopdb.utils.import_mode import apply_import_timestamps
|
||||
|
||||
models_bp = Blueprint('models', __name__)
|
||||
|
||||
@@ -5,7 +5,7 @@ machinetypes; the machines plugin now owns the "machinetypes" name.
|
||||
"""
|
||||
|
||||
from flask import Blueprint, request
|
||||
from flask_jwt_extended import jwt_required, current_user
|
||||
from flask_jwt_extended import jwt_required
|
||||
|
||||
from shopdb.extensions import db
|
||||
from shopdb.core.models import ModelType
|
||||
@@ -17,7 +17,7 @@ from shopdb.utils.responses import (
|
||||
)
|
||||
from shopdb.utils.pagination import get_pagination_params, paginate_query
|
||||
|
||||
from shopdb.utils.authz import require_permission, require_role
|
||||
from shopdb.utils.authz import require_role
|
||||
from shopdb.utils.import_mode import apply_import_timestamps
|
||||
|
||||
modeltypes_bp = Blueprint('modeltypes', __name__)
|
||||
|
||||
@@ -13,7 +13,7 @@ from shopdb.utils.responses import (
|
||||
)
|
||||
from shopdb.utils.pagination import get_pagination_params, paginate_query
|
||||
|
||||
from shopdb.utils.authz import require_permission, require_role
|
||||
from shopdb.utils.authz import require_role
|
||||
from shopdb.utils.import_mode import apply_import_timestamps
|
||||
|
||||
operatingsystems_bp = Blueprint('operatingsystems', __name__)
|
||||
|
||||
@@ -5,7 +5,7 @@ from flask_jwt_extended import jwt_required
|
||||
|
||||
from shopdb.utils.responses import success_response, error_response, ErrorCodes
|
||||
|
||||
from shopdb.utils.authz import require_permission, require_role
|
||||
from shopdb.utils.authz import require_role
|
||||
|
||||
plugins_bp = Blueprint('plugins', __name__)
|
||||
|
||||
|
||||
@@ -13,7 +13,7 @@ from shopdb.utils.responses import (
|
||||
)
|
||||
from shopdb.utils.pagination import get_pagination_params, paginate_query
|
||||
|
||||
from shopdb.utils.authz import require_permission, require_role
|
||||
from shopdb.utils.authz import require_role
|
||||
from shopdb.utils.import_mode import apply_import_timestamps
|
||||
|
||||
vendors_bp = Blueprint('vendors', __name__)
|
||||
|
||||
Reference in New Issue
Block a user