We're working to fix this issue. Please try again later.
← Back to Home ''', 500 @self.app.errorhandler(404) def handle_not_found(error): """Log 404 errors for security monitoring""" self.log_security_event( event_type="page_not_found", description=f"404 error: {request.path}", severity="LOW" ) return '''The page you're looking for doesn't exist.
← Back to Home ''', 404 def _get_request_context(self): """Get current request context information""" if not request: return {} return { 'ip_address': request.environ.get('HTTP_X_FORWARDED_FOR', request.remote_addr), 'user_agent': request.headers.get('User-Agent', ''), 'request_path': request.path, 'request_method': request.method, 'session_id': session.get('_id', 'anonymous'), 'user_id': session.get('user_id'), 'username': session.get('username') } def _log_to_database(self, event_type, event_category, description, event_data=None, severity='INFO'): """Log critical events to database table""" try: context = self._get_request_context() event_id = str(uuid.uuid4()) insert_sql = """ INSERT INTO log_events ( event_id, event_type, event_category, user_id, username, event_description, event_data, ip_address, user_agent, request_path, session_id, severity_level ) VALUES ( :event_id, :event_type, :event_category, :user_id, :username, :description, :event_data, :ip_address, :user_agent, :request_path, :session_id, :severity ) """ self.db.session.execute(text(insert_sql), { 'event_id': event_id, 'event_type': event_type, 'event_category': event_category, 'user_id': context.get('user_id'), 'username': context.get('username'), 'description': description, 'event_data': json.dumps(event_data) if event_data else None, 'ip_address': context.get('ip_address'), 'user_agent': context.get('user_agent'), 'request_path': context.get('request_path'), 'session_id': context.get('session_id'), 'severity': severity }) self.db.session.commit() except Exception as e: # Don't let logging errors break the application print(f"Database logging error: {e}") try: self.db.session.rollback() except: pass # USER LOGIN/LOGOUT LOGGING METHODS def log_user_login(self, user_id, username, success=True, failure_reason=None): """Log user login attempts with detailed session information""" context = self._get_request_context() event_data = { 'user_id': user_id, 'username': username, 'success': success, 'login_timestamp': datetime.now().isoformat(), 'session_info': { 'session_id': context.get('session_id'), 'ip_address': context.get('ip_address'), 'user_agent': context.get('user_agent') } } if failure_reason: event_data['failure_reason'] = failure_reason if success: message = f"User login successful: {username} (ID: {user_id})" self.logger.info(json.dumps({ 'event': 'user_login_success', 'data': event_data })) # Log to database for security monitoring self._log_to_database( event_type='user_login_success', event_category='authentication', description=message, event_data=event_data, severity='INFO' ) else: message = f"User login failed: {username} - {failure_reason}" self.security_logger.warning(json.dumps({ 'event': 'user_login_failure', 'data': event_data })) # Log failed logins to database for security analysis self._log_to_database( event_type='user_login_failure', event_category='security', description=message, event_data=event_data, severity='WARNING' ) def log_user_logout(self, user_id, username, session_duration=None): """Log user logout events with session duration""" context = self._get_request_context() event_data = { 'user_id': user_id, 'username': username, 'logout_timestamp': datetime.now().isoformat(), 'session_duration_minutes': session_duration, 'session_info': { 'session_id': context.get('session_id'), 'ip_address': context.get('ip_address') } } message = f"User logout: {username} (ID: {user_id})" if session_duration: message += f" - Session duration: {session_duration} minutes" self.logger.info(json.dumps({ 'event': 'user_logout', 'data': event_data })) # Log to database self._log_to_database( event_type='user_logout', event_category='authentication', description=message, event_data=event_data ) # QR CODE LOGGING METHODS def log_qr_code_created(self, qr_code_id, qr_code_name, created_by_user_id, qr_data): """Log QR code creation events""" event_data = { 'qr_code_id': qr_code_id, 'qr_code_name': qr_code_name, 'created_by_user_id': created_by_user_id, 'created_timestamp': datetime.now().isoformat(), 'qr_code_details': { 'location': qr_data.get('location'), 'location_address': qr_data.get('location_address'), 'location_event': qr_data.get('location_event'), 'has_coordinates': qr_data.get('has_coordinates', False) } } if qr_data.get('has_coordinates'): event_data['qr_code_details']['coordinates'] = { 'latitude': qr_data.get('latitude'), 'longitude': qr_data.get('longitude'), 'accuracy': qr_data.get('coordinate_accuracy') } message = f"QR code created: {qr_code_name} (ID: {qr_code_id}) by user {created_by_user_id}" self.logger.info(json.dumps({ 'event': 'qr_code_created', 'data': event_data })) # Log to database self._log_to_database( event_type='qr_code_created', event_category='qr_management', description=message, event_data=event_data ) def log_qr_code_updated(self, qr_code_id, qr_code_name, updated_by_user_id, changes): """Log QR code modification events""" event_data = { 'qr_code_id': qr_code_id, 'qr_code_name': qr_code_name, 'updated_by_user_id': updated_by_user_id, 'updated_timestamp': datetime.now().isoformat(), 'changes': changes } message = f"QR code updated: {qr_code_name} (ID: {qr_code_id}) by user {updated_by_user_id}" self.logger.info(json.dumps({ 'event': 'qr_code_updated', 'data': event_data })) # Log to database self._log_to_database( event_type='qr_code_updated', event_category='qr_management', description=message, event_data=event_data ) def log_qr_code_deleted(self, qr_code_id, qr_code_name, deleted_by_user_id): """Log QR code deletion events""" event_data = { 'qr_code_id': qr_code_id, 'qr_code_name': qr_code_name, 'deleted_by_user_id': deleted_by_user_id, 'deleted_timestamp': datetime.now().isoformat() } message = f"QR code deleted: {qr_code_name} (ID: {qr_code_id}) by user {deleted_by_user_id}" self.logger.warning(json.dumps({ 'event': 'qr_code_deleted', 'data': event_data })) # Log to database with higher severity self._log_to_database( event_type='qr_code_deleted', event_category='qr_management', description=message, event_data=event_data, severity='WARNING' ) def log_qr_code_accessed(self, qr_code_id, qr_code_name, access_method='scan'): """Log QR code access/scan events""" context = self._get_request_context() event_data = { 'qr_code_id': qr_code_id, 'qr_code_name': qr_code_name, 'access_method': access_method, 'access_timestamp': datetime.now().isoformat(), 'access_info': { 'ip_address': context.get('ip_address'), 'user_agent': context.get('user_agent') } } message = f"QR code accessed: {qr_code_name} (ID: {qr_code_id}) via {access_method}" self.logger.info(json.dumps({ 'event': 'qr_code_accessed', 'data': event_data })) # DATABASE ERROR LOGGING METHODS def log_database_error(self, operation, error, query=None, parameters=None): """Log database operation errors""" event_data = { 'operation': operation, 'error_type': type(error).__name__, 'error_message': str(error), 'error_timestamp': datetime.now().isoformat(), 'database_info': { 'query': query[:500] if query else None, # Truncate long queries 'parameters': str(parameters)[:200] if parameters else None } } if isinstance(error, SQLAlchemyError): event_data['sqlalchemy_error'] = True if hasattr(error, 'orig'): event_data['original_error'] = str(error.orig) message = f"Database error in {operation}: {type(error).__name__} - {str(error)}" self.logger.error(json.dumps({ 'event': 'database_error', 'data': event_data })) # Log to database if possible (try/catch to avoid recursive errors) try: self._log_to_database( event_type='database_error', event_category='database', description=message, event_data=event_data, severity='ERROR' ) except: # If database logging fails, just continue pass def log_database_connection_error(self, error): """Log database connection failures""" event_data = { 'error_type': 'database_connection_failure', 'error_message': str(error), 'error_timestamp': datetime.now().isoformat() } message = f"Database connection error: {str(error)}" self.logger.critical(json.dumps({ 'event': 'database_connection_error', 'data': event_data })) # FLASK ERROR LOGGING METHODS def log_flask_error(self, error_type, error_message, stack_trace=None, request_data=None): """Log Flask application errors""" context = self._get_request_context() event_data = { 'error_type': error_type, 'error_message': error_message, 'error_timestamp': datetime.now().isoformat(), 'request_context': context, 'stack_trace': stack_trace[:2000] if stack_trace else None # Truncate long traces } if request_data: event_data['request_data'] = request_data message = f"Flask error: {error_type} - {error_message}" self.logger.error(json.dumps({ 'event': 'flask_error', 'data': event_data })) # Log to database self._log_to_database( event_type='flask_error', event_category='application', description=message, event_data=event_data, severity='ERROR' ) # SECURITY EVENT LOGGING METHODS def log_security_event(self, event_type, description, severity='MEDIUM', additional_data=None): """Log security-related events""" context = self._get_request_context() event_data = { 'security_event_type': event_type, 'severity': severity, 'event_timestamp': datetime.now().isoformat(), 'request_context': context } if additional_data: event_data['additional_data'] = additional_data message = f"Security event: {event_type} - {description}" self.security_logger.warning(json.dumps({ 'event': 'security_event', 'data': event_data })) # Log to database with high priority self._log_to_database( event_type='security_event', event_category='security', description=message, event_data=event_data, severity='WARNING' ) # UTILITY METHODS def get_log_statistics(self, days=7): """Get logging statistics for the specified number of days""" try: cutoff_date = datetime.now() - timedelta(days=days) # Get total events total_sql = """ SELECT COUNT(*) as total_events FROM log_events WHERE created_timestamp >= :cutoff_date """ total_result = self.db.session.execute(text(total_sql), { 'cutoff_date': cutoff_date }).fetchone() # Get events by category category_sql = """ SELECT event_category, COUNT(*) as event_count FROM log_events WHERE created_timestamp >= :cutoff_date GROUP BY event_category """ category_result = self.db.session.execute(text(category_sql), { 'cutoff_date': cutoff_date }).fetchall() # Build simple statistics dictionary (not nested) stats = { 'total_events': total_result.total_events if total_result else 0, 'security_events': 0, 'database_errors': 0, 'user_activities': 0, 'system_events': 0 } # Process category results for row in category_result: if row.event_category == 'security': stats['security_events'] = row.event_count elif row.event_category == 'database': stats['database_errors'] = row.event_count elif row.event_category == 'user_activity': stats['user_activities'] = row.event_count elif row.event_category == 'system': stats['system_events'] = row.event_count return stats except Exception as e: self.log_database_error('get_log_statistics', e) print(f"Error in get_log_statistics: {e}") # Return default stats structure return { 'total_events': 0, 'security_events': 0, 'database_errors': 0, 'user_activities': 0, 'system_events': 0 } def cleanup_old_logs(self, days_to_keep=90): """Clean up old log entries from database""" try: cutoff_date = datetime.now() - timedelta(days=days_to_keep) # First, count how many records will be deleted count_sql = """ SELECT COUNT(*) as count_to_delete FROM log_events WHERE created_timestamp < :cutoff_date AND severity_level NOT IN ('ERROR', 'CRITICAL', 'HIGH') """ count_result = self.db.session.execute(text(count_sql), { 'cutoff_date': cutoff_date }).fetchone() count_to_delete = count_result.count_to_delete if count_result else 0 if count_to_delete == 0: return 0 # Perform the cleanup - exclude critical logs cleanup_sql = """ DELETE FROM log_events WHERE created_timestamp < :cutoff_date AND severity_level NOT IN ('ERROR', 'CRITICAL', 'HIGH') """ result = self.db.session.execute(text(cleanup_sql), { 'cutoff_date': cutoff_date }) deleted_count = result.rowcount self.db.session.commit() # Log the cleanup operation self.logger.info(f"Log cleanup completed: {deleted_count} entries removed (keeping entries older than {days_to_keep} days)") # Also log to security log for audit self.log_security_event( event_type="log_cleanup", description=f"Admin cleaned up {deleted_count} old log entries (keeping last {days_to_keep} days)", severity="MEDIUM", additional_data={ 'days_to_keep': days_to_keep, 'deleted_count': deleted_count, 'cutoff_date': cutoff_date.isoformat() } ) return deleted_count except Exception as e: self.db.session.rollback() self.log_database_error('cleanup_old_logs', e) print(f"Error in cleanup_old_logs: {e}") return 0 def get_recent_logs(self, days=7, limit=100, category_filter=None, severity_filter=None, search_term=None): """Enhanced method to get recent logs with filtering options""" try: cutoff_date = datetime.now() - timedelta(days=days) # Build the base query base_sql = """ SELECT event_id, event_type, event_category, event_description, severity_level, created_timestamp, username, ip_address, user_id FROM log_events WHERE created_timestamp >= :cutoff_date """ # Add filters params = {'cutoff_date': cutoff_date} if category_filter: base_sql += " AND event_category = :category_filter" params['category_filter'] = category_filter if severity_filter: base_sql += " AND severity_level = :severity_filter" params['severity_filter'] = severity_filter if search_term: base_sql += " AND (event_description LIKE :search_term OR event_type LIKE :search_term OR username LIKE :search_term)" params['search_term'] = f"%{search_term}%" # Add ordering and limit base_sql += " ORDER BY created_timestamp DESC LIMIT :limit" params['limit'] = limit result = self.db.session.execute(text(base_sql), params).fetchall() logs = [] for row in result: logs.append({ 'event_id': row.event_id, 'event_type': row.event_type, 'event_category': row.event_category, 'description': row.event_description, 'severity': row.severity_level, 'timestamp': row.created_timestamp.isoformat(), 'username': row.username or 'System', 'ip_address': row.ip_address or '-', 'user_id': row.user_id }) return logs except Exception as e: self.log_database_error('get_recent_logs', e) print(f"Error in get_recent_logs: {e}") return [] # DECORATOR FUNCTIONS FOR AUTOMATIC LOGGING def log_user_activity(activity_type): """Decorator to automatically log user activities""" def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): try: result = f(*args, **kwargs) # Log successful activity if hasattr(g, 'app') and hasattr(g.app, 'logger_handler'): logger = g.app.logger_handler logger.logger.info(json.dumps({ 'event': f'user_activity_{activity_type}', 'data': { 'user_id': session.get('user_id'), 'username': session.get('username'), 'activity': activity_type, 'timestamp': datetime.now().isoformat() } })) return result except Exception as e: # Log error if hasattr(g, 'app') and hasattr(g.app, 'logger_handler'): logger = g.app.logger_handler logger.log_flask_error( error_type=f"activity_error_{activity_type}", error_message=str(e), stack_trace=traceback.format_exc() ) raise return decorated_function return decorator def log_database_operations(operation_name): """Decorator to automatically log database operations""" def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): try: result = f(*args, **kwargs) return result except Exception as e: # Log database error if hasattr(g, 'app') and hasattr(g.app, 'logger_handler'): logger = g.app.logger_handler logger.log_database_error( operation=operation_name, error=e ) raise return decorated_function return decorator def verify_log_table_exists(self): """Verify that the log_events table exists and has the correct structure""" try: # Check if table exists check_table_sql = """ SELECT COUNT(*) as table_exists FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = 'log_events' """ result = self.db.session.execute(text(check_table_sql)).fetchone() if result.table_exists == 0: print("⚠️ log_events table does not exist. Creating it now...") self._create_log_table() return True # Check if table has records count_sql = "SELECT COUNT(*) as record_count FROM log_events" count_result = self.db.session.execute(text(count_sql)).fetchone() print(f"✅ log_events table exists with {count_result.record_count} records") return True except Exception as e: print(f"❌ Error verifying log table: {e}") return False # INITIALIZATION FUNCTION def init_logging(app, db): """Initialize the logging system with the Flask app""" logger_handler = AppLogger(app, db) return logger_handler