diff --git a/app.py b/app.py index 01d418c..ce66588 100644 --- a/app.py +++ b/app.py @@ -49,6 +49,13 @@ def create_app() -> Flask: cfg = get_config() app.config.from_object(cfg) + # Guard against deployment with the insecure default SECRET_KEY + import sys + if not app.debug and app.config.get('SECRET_KEY') == 'change-me-in-production': + print("FATAL: SECRET_KEY is set to the insecure default value. " + "Set SECRET_KEY in your .env file before deploying to production.") + sys.exit(1) + # ------------------------------------------------------------------ # Database initialization # ------------------------------------------------------------------ @@ -100,12 +107,8 @@ def create_app() -> Flask: @app.context_processor def inject_company_name(): - """Make COMPANY_NAME, THEME_NAME, and CURRENT_YEAR available to all templates""" - return { - 'COMPANY_NAME': os.environ.get('COMPANY_NAME', 'QR Code Management System'), - 'THEME_NAME': os.environ.get('THEME_NAME', ''), - 'CURRENT_YEAR': datetime.now().year, - } + """Make COMPANY_NAME available to all templates""" + return {'COMPANY_NAME': os.environ.get('COMPANY_NAME', 'QR Code Management System')} @app.context_processor def inject_logging_status(): @@ -247,10 +250,20 @@ def create_app() -> Flask: @app.errorhandler(500) def internal_error(error): """Handle internal server errors with user-friendly page""" - if app.debug: - return None return render_template('errors/500.html'), 500 + # ------------------------------------------------------------------ + # Startup initialization (runs under gunicorn and flask run alike) + # ------------------------------------------------------------------ + with app.app_context(): + try: + create_tables() + update_existing_qr_codes() + except Exception as e: + from extensions import logger_handler as _startup_lh + _startup_lh.logger.error(f"Startup initialization failed: {e}", exc_info=True) + raise + return app @@ -298,10 +311,8 @@ def update_existing_qr_codes(): """Update existing QR codes with missing URLs or images at startup. Regenerates qr_url slugs without needing a request context. - For qr_code_image, uses QR_BASE_URL (from .env) as the authoritative base - URL so that generated links always match the public-facing domain. - Falls back to FLASK_HOST/FLASK_PORT construction only when QR_BASE_URL is - not configured (development environments without a reverse proxy). + For qr_code_image, constructs the base URL from FLASK_HOST/FLASK_PORT + config so this can run safely outside any HTTP request. """ from extensions import db as _db, logger_handler as lh from utils.helpers import generate_qr_code, get_qr_styling, generate_qr_url @@ -312,19 +323,14 @@ def update_existing_qr_codes(): if not qr_codes: return - # Prefer the explicit QR_BASE_URL env var (required behind a reverse proxy). - # Fall back to FLASK_HOST/PORT for local/dev environments. - qr_base_url = os.environ.get('QR_BASE_URL', '').rstrip('/') - if qr_base_url: - base_url = qr_base_url + '/' - else: - host = os.environ.get('FLASK_HOST', '0.0.0.0') - # 0.0.0.0 is a bind address, not a reachable hostname — default to localhost - if host in ('0.0.0.0', ''): - host = 'localhost' - port = os.environ.get('FLASK_PORT', '5000') - scheme = 'https' if _Cfg.SESSION_COOKIE_SECURE else 'http' - base_url = f"{scheme}://{host}:{port}/" + # Build a base URL that does not require an active request context. + host = os.environ.get('FLASK_HOST', '0.0.0.0') + # 0.0.0.0 is a bind address, not a reachable hostname — default to localhost + if host in ('0.0.0.0', ''): + host = 'localhost' + port = os.environ.get('FLASK_PORT', '5000') + scheme = 'https' if _Cfg.SESSION_COOKIE_SECURE else 'http' + base_url = f"{scheme}://{host}:{port}/" updated_count = 0 for qr_code in qr_codes: @@ -363,9 +369,6 @@ app = create_app() if __name__ == '__main__': with app.app_context(): try: - create_tables() - update_existing_qr_codes() - from extensions import logger_handler logger_handler.logger.info("Initializing performance optimizations") cached_query = initialize_performance_optimizations(app, db, logger_handler) diff --git a/routes/auth.py b/routes/auth.py index be5cbb2..9d1666e 100644 --- a/routes/auth.py +++ b/routes/auth.py @@ -7,6 +7,7 @@ Routes: /, /register, /login, /logout, /profile """ from flask import Blueprint, render_template, request, redirect, flash, session, jsonify, url_for from datetime import datetime +from urllib.parse import urlparse, urljoin import json from extensions import db, logger_handler @@ -105,7 +106,11 @@ def login(): if user and user.check_password(password): # Check if "Remember Me" is checked remember_me = request.form.get('remember_me') == 'on' - + + # Invalidate the pre-login session to prevent session fixation attacks, + # then re-apply the remember_me permanence flag on the fresh session. + session.clear() + # Set session as permanent if "Remember Me" is checked if remember_me: session.permanent = True @@ -143,9 +148,18 @@ def login(): flash(f'Welcome back, {user.full_name}!', 'success') logger_handler.logger.info(f"User {user.username} (ID: {user.id}) logged in successfully") - # Redirect to intended page or dashboard + # Redirect to intended page or dashboard. + # Validate next is a relative path on this host to prevent open-redirect attacks. + def _is_safe_url(target): + ref_url = urlparse(request.host_url) + test_url = urlparse(urljoin(request.host_url, target)) + return (test_url.scheme in ('http', 'https') + and ref_url.netloc == test_url.netloc) + next_page = request.args.get('next') - return redirect(next_page) if next_page else redirect(url_for('attendance.attendance_report')) + if next_page and _is_safe_url(next_page): + return redirect(next_page) + return redirect(url_for('attendance.attendance_report')) else: # Invalid credentials - log failed attempt diff --git a/routes/payroll.py b/routes/payroll.py index e6e4e15..3e4817b 100644 --- a/routes/payroll.py +++ b/routes/payroll.py @@ -117,11 +117,11 @@ def payroll_dashboard(): employee_names = {} if working_hours_data: try: - # Use the same SQL approach as attendance report - JOIN with CAST + # Use parameterized IN clause to avoid SQL injection employee_ids = list(working_hours_data['employees'].keys()) if employee_ids: - # Build a query similar to attendance report - placeholders = ','.join([f"'{emp_id}'" for emp_id in employee_ids]) + placeholders = ','.join([f':emp_{i}' for i in range(len(employee_ids))]) + emp_params = {f'emp_{i}': eid for i, eid in enumerate(employee_ids)} employee_query = db.session.execute(text(f""" SELECT ad.employee_id, @@ -130,7 +130,7 @@ def payroll_dashboard(): LEFT JOIN employee e ON CAST(ad.employee_id AS UNSIGNED) = e.id WHERE ad.employee_id IN ({placeholders}) GROUP BY ad.employee_id, e.firstName, e.lastName - """)) + """), emp_params) for row in employee_query: if row[1]: # Only add if we got a name @@ -240,13 +240,13 @@ def export_payroll_excel(): logger_handler.logger.info(f"Exporting {len(attendance_records)} attendance records to payroll Excel") - # Get employee names using the same method as dashboard + # Get employee names using parameterized IN clause to avoid SQL injection employee_names = {} try: employee_ids = list(set(str(record.employee_id) for record in attendance_records)) if employee_ids: - # Use the same SQL approach as attendance report - JOIN with CAST - placeholders = ','.join([f"'{emp_id}'" for emp_id in employee_ids]) + placeholders = ','.join([f':emp_{i}' for i in range(len(employee_ids))]) + emp_params = {f'emp_{i}': eid for i, eid in enumerate(employee_ids)} employee_query = db.session.execute(text(f""" SELECT ad.employee_id, @@ -255,7 +255,7 @@ def export_payroll_excel(): LEFT JOIN employee e ON CAST(ad.employee_id AS UNSIGNED) = e.id WHERE ad.employee_id IN ({placeholders}) GROUP BY ad.employee_id, e.firstName, e.lastName - """)) + """), emp_params) for row in employee_query: if row[1]: # Only add if we got a name diff --git a/routes/statistics.py b/routes/statistics.py index 7da1252..2b26ee4 100644 --- a/routes/statistics.py +++ b/routes/statistics.py @@ -35,25 +35,34 @@ def qr_statistics(): qr_code_filter = request.args.get('qr_code', '') project_filter = request.args.get('project', '') - # Build date filter - date_filter = "" + # Build parameterized filter conditions (fixes SQL injection) + conditions = [] + params = {} + if date_from: - date_filter += f" AND ad.check_in_date >= '{date_from}'" + conditions.append("ad.check_in_date >= :date_from") + params["date_from"] = date_from if date_to: - date_filter += f" AND ad.check_in_date <= '{date_to}'" - - # QR Code filter - qr_filter = "" + conditions.append("ad.check_in_date <= :date_to") + params["date_to"] = date_to if qr_code_filter: - qr_filter = f" AND ad.qr_code_id = {qr_code_filter}" - - # Project filter - project_filter_clause = "" + try: + params["qr_code_id"] = int(qr_code_filter) + conditions.append("ad.qr_code_id = :qr_code_id") + except (ValueError, TypeError): + logger_handler.logger.warning(f"Invalid qr_code filter value ignored: {qr_code_filter!r}") if project_filter: - project_filter_clause = f" AND qc.project_id = {project_filter}" + try: + params["project_id"] = int(project_filter) + conditions.append("qc.project_id = :project_id") + except (ValueError, TypeError): + logger_handler.logger.warning(f"Invalid project filter value ignored: {project_filter!r}") + + # Compose a reusable AND clause (empty string when no filters applied) + filter_clause = (" AND " + " AND ".join(conditions)) if conditions else "" # 1. General Statistics - general_stats = db.session.execute(text(f""" + general_stats = db.session.execute(text(""" SELECT COUNT(*) as total_scans, COUNT(DISTINCT ad.employee_id) as unique_users, @@ -64,11 +73,11 @@ def qr_statistics(): COUNT(CASE WHEN ad.latitude IS NOT NULL AND ad.longitude IS NOT NULL THEN 1 END) as gps_enabled_scans FROM attendance_data ad LEFT JOIN qr_codes qc ON ad.qr_code_id = qc.id - WHERE 1=1 {date_filter} {qr_filter} {project_filter_clause} - """)).fetchone() + WHERE 1=1 + """ + filter_clause), params).fetchone() # 2. Device Statistics - device_stats = db.session.execute(text(f""" + device_stats = db.session.execute(text(""" SELECT CASE WHEN device_info LIKE '%iPhone%' OR device_info LIKE '%iOS%' THEN 'iOS' @@ -82,13 +91,14 @@ def qr_statistics(): COUNT(DISTINCT employee_id) as unique_users FROM attendance_data ad LEFT JOIN qr_codes qc ON ad.qr_code_id = qc.id - WHERE device_info IS NOT NULL {date_filter} {qr_filter} {project_filter_clause} + WHERE device_info IS NOT NULL + """ + filter_clause + """ GROUP BY device_type ORDER BY scan_count DESC - """)).fetchall() + """), params).fetchall() # 3. Browser Statistics (from User Agent) - browser_stats = db.session.execute(text(f""" + browser_stats = db.session.execute(text(""" SELECT CASE WHEN user_agent LIKE '%Chrome%' AND user_agent NOT LIKE '%Edge%' THEN 'Chrome' @@ -102,13 +112,14 @@ def qr_statistics(): COUNT(DISTINCT employee_id) as unique_users FROM attendance_data ad LEFT JOIN qr_codes qc ON ad.qr_code_id = qc.id - WHERE user_agent IS NOT NULL {date_filter} {qr_filter} {project_filter_clause} + WHERE user_agent IS NOT NULL + """ + filter_clause + """ GROUP BY browser_type ORDER BY scan_count DESC - """)).fetchall() + """), params).fetchall() - # 4. Location Statistics - location_stats = db.session.execute(text(f""" + # 4. Location Statistics + location_stats = db.session.execute(text(""" SELECT qc.name as qr_name, qc.location as qr_location, @@ -120,13 +131,14 @@ def qr_statistics(): MAX(ad.check_in_date) as last_scan FROM attendance_data ad JOIN qr_codes qc ON ad.qr_code_id = qc.id - WHERE 1=1 {date_filter} {qr_filter} {project_filter_clause} + WHERE 1=1 + """ + filter_clause + """ GROUP BY qc.id, qc.name, qc.location, qc.location_event ORDER BY total_scans DESC - """)).fetchall() + """), params).fetchall() # 5. IP Address Analysis (Top 3 Most Active) - ip_stats = db.session.execute(text(f""" + ip_stats = db.session.execute(text(""" SELECT ip_address, COUNT(*) as scan_count, @@ -136,14 +148,15 @@ def qr_statistics(): MAX(check_in_date) as last_scan FROM attendance_data ad LEFT JOIN qr_codes qc ON ad.qr_code_id = qc.id - WHERE ip_address IS NOT NULL {date_filter} {qr_filter} {project_filter_clause} + WHERE ip_address IS NOT NULL + """ + filter_clause + """ GROUP BY ip_address ORDER BY scan_count DESC LIMIT 3 - """)).fetchall() + """), params).fetchall() # 6. Project Statistics (if projects exist) - project_stats = db.session.execute(text(f""" + project_stats = db.session.execute(text(""" SELECT p.id, p.name as project_name, @@ -154,10 +167,11 @@ def qr_statistics(): FROM attendance_data ad JOIN qr_codes qc ON ad.qr_code_id = qc.id LEFT JOIN projects p ON qc.project_id = p.id - WHERE p.id IS NOT NULL {date_filter} {qr_filter} {project_filter_clause} + WHERE p.id IS NOT NULL + """ + filter_clause + """ GROUP BY p.id, p.name ORDER BY total_scans DESC - """)).fetchall() + """), params).fetchall() # Get dropdown options for filters qr_codes_list = db.session.execute(text("""