Merge pull request #10 from CoreSecFrame/feature/file-manager

Refactor and improve code quality

This commit includes several code improvements and refactorings:

Refactored app/core/models.py for log counting:

Introduced a private helper method _count_logs_since in the SystemLog model to consolidate logic for counting logs based on time, level, and security status.
Updated get_error_count_last_24h and get_security_events_count_last_24h to use this new helper method, reducing code duplication.
Made security log retention configurable:

Added a SECURITY_LOG_RETENTION_DAYS configuration variable in app/config.py (defaulting to 90 days, settable via environment variable).
Modified the cleanup_old_logs method in SystemLog to use this configuration value instead of a hardcoded one.
Improved app/core/models.py documentation and readability:

Added a docstring to LogSearchQuery.to_dict().
Ensured dictionary definitions and SQLAlchemy queries in SystemLog methods are formatted for better readability.
Removed some redundant local imports of datetime and timedelta.
Refactored exception handling in app/auth/routes.py:

Introduced a private helper function _handle_auth_exception to standardize logging and flash messages for exceptions occurring during user login and registration.
Updated the login and register routes to use this helper, reducing boilerplate.
Also default terminal path changed to /home/$USER

Author: sPROFFEs

app/auth/routes.py

@@ -9,6 +9,12 @@
 
 auth_bp = Blueprint('auth', __name__, url_prefix='/auth')
 
+def _handle_auth_exception(e, operation_name):
+    """Helper function to log and flash authentication errors."""
+    current_app.logger.error(f"{operation_name} error: {str(e)}")
+    current_app.logger.error(traceback.format_exc())
+    flash(f'An error occurred during {operation_name.lower()}. Please try again.', 'danger')
+
 @auth_bp.route('/login', methods=['GET', 'POST'])
 def login():
     if current_user.is_authenticated:
@@ -29,9 +35,7 @@ def login():
                 next_page = url_for('core.index')
             return redirect(next_page)
         except Exception as e:
-            current_app.logger.error(f"Login error: {str(e)}")
-            current_app.logger.error(traceback.format_exc())
-            flash('An error occurred during login. Please try again.', 'danger')
+            _handle_auth_exception(e, "Login")
     return render_template('auth/login.html', title='Sign In', form=form)
 
 @auth_bp.route('/logout')
@@ -59,10 +63,8 @@ def register():
             flash('Congratulations, you are now a registered user!', 'success')
             return redirect(url_for('auth.login'))
         except Exception as e:
-            db.session.rollback()
-            current_app.logger.error(f"Registration error: {str(e)}")
-            current_app.logger.error(traceback.format_exc())
-            flash('An error occurred during registration. Please try again.', 'danger')
+            db.session.rollback() # Keep rollback here as it's specific to DB operations
+            _handle_auth_exception(e, "Registration")
     return render_template('auth/register.html', title='Register', form=form)
 
 @auth_bp.route('/profile')

app/config.py

@@ -21,6 +21,9 @@ class Config:
     # Modules repository URL
     MODULES_REPOSITORY_URL = os.environ.get('MODULES_REPOSITORY_URL') or \
         'https://github.com/CoreSecFrame/CoreSecFrame-Modules'
+
+    # Log retention settings
+    SECURITY_LOG_RETENTION_DAYS = int(os.environ.get('SECURITY_LOG_RETENTION_DAYS', 90))
 
     # Security settings
     SESSION_COOKIE_SECURE = False  # Set to True in production

app/core/models.py

@@ -1,6 +1,7 @@
 # app/core/models.py
+from flask import current_app
 from app import db
-from datetime import datetime
+from datetime import datetime, timedelta
 from sqlalchemy import Index
 
 class SystemLog(db.Model):
@@ -102,6 +103,20 @@ def is_error(self):
     def is_warning(self):
         """Check if this is a warning log"""
         return self.level == 'WARNING'
+
+    @classmethod
+    def _count_logs_since(cls, since_hours, levels=None, is_security_event=None):
+        """Helper method to count logs within a time window with optional filters."""
+        since = datetime.utcnow() - timedelta(hours=since_hours)
+        query = cls.query.filter(cls.timestamp >= since)
+
+        if levels:
+            query = query.filter(cls.level.in_(levels))
+        
+        if is_security_event is not None:
+            query = query.filter(cls.is_security_event == is_security_event)
+            
+        return query.count()
 
     @classmethod
     def get_recent_logs(cls, limit=50, level=None, is_security=None, user_id=None):
@@ -122,31 +137,16 @@ def get_recent_logs(cls, limit=50, level=None, is_security=None, user_id=None):
     @classmethod
     def get_error_count_last_24h(cls):
         """Get count of errors in the last 24 hours"""
-        from datetime import datetime, timedelta
-        
-        since = datetime.utcnow() - timedelta(hours=24)
-        
-        return cls.query.filter(
-            cls.timestamp >= since,
-            cls.level.in_(['ERROR', 'CRITICAL'])
-        ).count()
+        return cls._count_logs_since(since_hours=24, levels=['ERROR', 'CRITICAL'])
 
     @classmethod
     def get_security_events_count_last_24h(cls):
         """Get count of security events in the last 24 hours"""
-        from datetime import datetime, timedelta
-        
-        since = datetime.utcnow() - timedelta(hours=24)
-        
-        return cls.query.filter(
-            cls.timestamp >= since,
-            cls.is_security_event == True
-        ).count()
+        return cls._count_logs_since(since_hours=24, is_security_event=True)
 
     @classmethod
     def get_log_level_stats(cls, hours=24):
         """Get statistics by log level for the specified time period"""
-        from datetime import datetime, timedelta
         from sqlalchemy import func
 
         since = datetime.utcnow() - timedelta(hours=hours)
@@ -163,7 +163,6 @@ def get_log_level_stats(cls, hours=24):
     @classmethod
     def get_module_stats(cls, hours=24, limit=10):
         """Get top modules by log count"""
-        from datetime import datetime, timedelta
         from sqlalchemy import func
 
         since = datetime.utcnow() - timedelta(hours=hours)
@@ -182,12 +181,11 @@ def get_module_stats(cls, hours=24, limit=10):
     @classmethod
     def cleanup_old_logs(cls, days_to_keep=30):
         """Clean up old log entries"""
-        from datetime import datetime, timedelta
-        
         cutoff_date = datetime.utcnow() - timedelta(days=days_to_keep)
 
-        # Keep security events longer (90 days)
-        security_cutoff = datetime.utcnow() - timedelta(days=90)
+        # Keep security events longer, based on config
+        security_retention_days = current_app.config['SECURITY_LOG_RETENTION_DAYS']
+        security_cutoff = datetime.utcnow() - timedelta(days=security_retention_days)
 
         # Delete old non-security logs
         deleted_count = cls.query.filter(
@@ -227,7 +225,10 @@ class LogSearchQuery(db.Model):
     use_count = db.Column(db.Integer, default=0)
 
     def to_dict(self):
-        """Convert to dictionary for API"""
+        """
+        Returns a dictionary representation of the log search query, 
+        suitable for API responses.
+        """
         return {
             'id': self.id,
             'name': self.name,

app/terminal/manager.py

@@ -55,7 +55,10 @@ def create_session(app, session_id, socketio, allow_create=True):
             env = os.environ.copy()
             env['TERM'] = 'xterm-256color'
             env['BASH_COMPLETION_ENABLED'] = '1'
-            
+
+            # Get user's home directory
+            user_home = os.path.expanduser('~')
+
             # Start bash with completion enabled
             process = subprocess.Popen(
                 ['/bin/bash', '--login'],
@@ -64,6 +67,7 @@ def create_session(app, session_id, socketio, allow_create=True):
                 stderr=slave,
                 start_new_session=True,
                 env=env,
+                cwd=user_home,  # Agregar esta línea
                 close_fds=True
             )
 

app/terminal/utils.py

@@ -25,6 +25,9 @@ def create_terminal_process(session):
         if check_session.returncode == 0:
             # Session already exists
             return True
+        
+        # Get user's home directory
+        user_home = os.path.expanduser('~')
 
         # Create a new session
         if session.session_type == 'terminal':
@@ -34,6 +37,7 @@ def create_terminal_process(session):
                 '-d',  # Start detached
                 '-s', session.session_id,  # Session name
                 '-n', 'main',  # Window name
+                '-c', user_home,  # Set working directory to user's home
                 'bash'  # Command to run
             ], check=True)
         elif session.session_type in ['guided', 'direct']:
@@ -52,6 +56,7 @@ def create_terminal_process(session):
                 '-d',  # Start detached
                 '-s', session.session_id,  # Session name
                 '-n', session.module_name,  # Window name
+                '-c', user_home,  # Set working directory to user's home
                 module_cmd  # Command to run
             ], check=True)