diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..399d0c5 --- /dev/null +++ b/.env.example @@ -0,0 +1,12 @@ +# Database Configuration +DB_HOST=localhost +DB_USER=your_username +DB_PASSWORD=your_secure_password +DB_NAME=DATAVERSE + +# Security Configuration +SECRET_KEY=your_secret_key_here_change_this_in_production +SESSION_TIMEOUT=3600 + +# Application Configuration +APP_DEBUG=False \ No newline at end of file diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..81bc06a --- /dev/null +++ b/.gitignore @@ -0,0 +1,102 @@ +# Environment variables +.env + +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +.hypothesis/ +.pytest_cache/ + +# Jupyter Notebook +.ipynb_checkpoints + +# pyenv +.python-version + +# celery beat schedule file +celerybeat-schedule + +# SageMath parsed files +*.sage.py + +# Environments +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS +.DS_Store +.DS_Store? +._* +.Spotlight-V100 +.Trashes +ehthumbs.db +Thumbs.db + +# Application specific +appconfig +*.log +plot.png +*.pdf \ No newline at end of file diff --git a/SECURITY.md b/SECURITY.md index 32eebb2..c62ae65 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -1,39 +1,199 @@ -# Dataverse's Security Policies and Procedures +# Security Implementation Documentation -This document outlines security procedures and general policies for the -Dataverse project. +## Overview -- [Reporting a Vulnerability](#reporting-a-vulnerability) -- [Disclosure Policy](#disclosure-policy) +This document outlines the security enhancements implemented in Dataverse to protect user data and prevent common security vulnerabilities. -## Reporting a Vulnerability +## Security Features Implemented -The Dataverse team and community takes all security vulnerabilities -seriously. Thank you for improving the security of our open source -software. We appreciate your efforts and responsible disclosure and will -make every effort to acknowledge your contributions. +### šŸ” Password Security -Report security vulnerabilities by emailing the Dataverse team at: +- **Secure Hashing**: Replaced custom encryption with bcrypt hashing +- **Salt Generation**: Each password gets a unique salt +- **Password Strength**: Enforced strong password requirements +- **Migration Support**: Existing passwords are migrated securely +### šŸŒ Environment Configuration + +- **Environment Variables**: Database credentials moved to `.env` file +- **Configuration Management**: Centralized configuration with defaults +- **Secret Management**: Secure handling of sensitive configuration + +### šŸ›”ļø SQL Injection Prevention + +- **Parameterized Queries**: All database queries use parameter binding +- **Input Validation**: Comprehensive validation for all user inputs +- **Data Sanitization**: Automatic sanitization of user data + +### šŸŽ« Session Management + +- **Token-Based Sessions**: Secure session tokens using cryptographic random generation +- **Session Timeout**: Automatic session expiration +- **Session Validation**: Proper session validation and cleanup + +### šŸš« Account Protection + +- **Failed Attempt Tracking**: Monitor and limit failed login attempts +- **Account Lockout**: Temporary account lockout after multiple failures +- **Brute Force Protection**: Protection against brute force attacks + +## Security Configuration + +### Environment Variables + +Create a `.env` file with the following variables: + +```env +# Database Configuration +DB_HOST=localhost +DB_USER=your_username +DB_PASSWORD=your_secure_password +DB_NAME=DATAVERSE + +# Security Configuration +SECRET_KEY=your_secret_key_here +SESSION_TIMEOUT=3600 + +# Application Configuration +APP_DEBUG=False ``` -multiverse.letter@gmail.com + +### Password Requirements + +- Minimum 8 characters +- At least one uppercase letter +- At least one lowercase letter +- At least one digit +- At least one special character +- Protection against common weak passwords + +### Session Configuration + +- Default timeout: 1 hour (3600 seconds) +- Automatic cleanup of expired sessions +- Secure token generation using `secrets` module + +## Migration Process + +### Before Migration + +1. **Backup Database**: The migration script automatically creates a backup +2. **Install Dependencies**: Ensure `bcrypt` and `python-dotenv` are installed +3. **Configure Environment**: Set up your `.env` file + +### Running Migration + +```bash +python migration_script.py ``` -The lead maintainer will acknowledge your email within 24 hours and will -send a more detailed response within 48 hours indicating the next steps in -handling your report. After the initial reply to your report, the security -team will endeavor to keep you informed of the progress towards a fix and -full announcement, and may ask for additional information or guidance. +### After Migration + +1. **Test Login**: Verify existing users can still log in +2. **Test Registration**: Create new users to test the system +3. **Remove Backup**: Once satisfied, remove the backup table + +## Security Best Practices + +### For Developers + +1. **Never Hardcode Credentials**: Always use environment variables +2. **Use Parameterized Queries**: Never use string formatting for SQL +3. **Validate All Inputs**: Validate and sanitize all user inputs +4. **Handle Errors Securely**: Don't expose sensitive information in errors + +### For Deployment + +1. **Secure Environment File**: Protect the `.env` file with proper permissions +2. **Use Strong Secrets**: Generate strong, unique secret keys +3. **Regular Updates**: Keep dependencies updated for security patches +4. **Monitor Logs**: Monitor for suspicious activities + +## Security Testing + +### Manual Testing + +1. **SQL Injection**: Test with malicious SQL in input fields +2. **Password Strength**: Test weak password rejection +3. **Session Management**: Test session timeout and validation +4. **Account Lockout**: Test failed login attempt limits + +### Automated Testing + +Consider implementing: +- Unit tests for validation functions +- Integration tests for authentication +- Security scanning tools +- Dependency vulnerability scanning + +## Incident Response + +### If Security Issue Detected + +1. **Immediate Action**: Change all secrets and passwords +2. **Assess Impact**: Determine what data may have been compromised +3. **Update System**: Apply security patches immediately +4. **Monitor**: Increase monitoring for suspicious activities + +### Reporting Security Issues + +If you discover a security vulnerability: +1. **Do Not** create a public issue +2. Contact the maintainers privately +3. Provide detailed information about the vulnerability +4. Allow time for the issue to be fixed before disclosure + +## Compliance Notes + +This implementation follows security best practices including: +- OWASP Top 10 protection guidelines +- Password hashing standards (bcrypt) +- Session management best practices +- Input validation and sanitization standards + +## Dependencies + +### Security-Related Dependencies + +- `bcrypt>=4.0.1`: Secure password hashing +- `python-dotenv>=1.0.0`: Environment variable management + +### Security Considerations + +- Keep dependencies updated +- Monitor for security advisories +- Use dependency scanning tools + +## Changelog + +### Version 1.0 (Security Enhancement) + +- āœ… Implemented bcrypt password hashing +- āœ… Added environment variable configuration +- āœ… Implemented parameterized queries +- āœ… Added comprehensive input validation +- āœ… Implemented secure session management +- āœ… Added account lockout protection +- āœ… Created migration script for existing data + +## Future Enhancements + +### Planned Security Improvements + +- [ ] Two-factor authentication (2FA) +- [ ] Password reset functionality +- [ ] Audit logging +- [ ] Rate limiting for API endpoints +- [ ] CSRF protection for web interface +- [ ] Content Security Policy (CSP) headers -Report security vulnerabilities in third-party modules to the person or -team maintaining the module. +### Monitoring and Alerting -## Disclosure Policy +- [ ] Failed login attempt monitoring +- [ ] Unusual activity detection +- [ ] Security event logging +- [ ] Automated security scanning -When the security team receives a security bug report, they will assign it -to a primary handler. This person will coordinate the fix and release -process, involving the following steps: +--- - * Confirm the problem. - * Audit code to find any potential similar problems. - * Prepare fixes and release them as fast as possible. \ No newline at end of file +For questions about security implementation, please refer to the implementation guide or contact the development team. \ No newline at end of file diff --git a/installation/requirements.txt b/installation/requirements.txt index ee35841..ee5cd65 100644 --- a/installation/requirements.txt +++ b/installation/requirements.txt @@ -1,16 +1,8 @@ -PIL -time +Pillow numpy -ctypes -tkinter -datetime tabulate -functools -webbrowser matplotlib customtkinter -mysql.connector -matplotlib.pyplot -matplotlib.gridspec -matplotlib.widgets -mpl_toolkits.mplot3d \ No newline at end of file +mysql-connector-python +bcrypt>=4.0.1 +python-dotenv>=1.0.0 \ No newline at end of file diff --git a/migration_script.py b/migration_script.py new file mode 100644 index 0000000..f892900 --- /dev/null +++ b/migration_script.py @@ -0,0 +1,257 @@ +#!/usr/bin/env python3 +""" +Migration script to update existing passwords to secure hashes +Run this ONCE after implementing the new security system + +Usage: python migration_script.py +""" + +import sys +import os + +# Add software directory to path +sys.path.append('software') + +import mysql.connector as my +from mysql.connector import Error +import db_config +from security import security_manager + +def decrypt_old_password(encrypted_password): + """ + Decrypt password using old method (for migration only) + This function replicates the old decryption logic + + Args: + encrypted_password (str): Old encrypted password + + Returns: + str: Decrypted password + """ + try: + password_length = len(encrypted_password) + decrypted_password = "" + + if password_length % 2 == 0: + transformed_pwd = encrypted_password[int(password_length / 2):] + transformed_pwd += encrypted_password[:int(password_length / 2)] + else: + transformed_pwd = encrypted_password[int(password_length / 2) + 1:] + transformed_pwd += encrypted_password[:int(password_length / 2) + 1] + + for character in transformed_pwd: + decrypted_password += chr(ord(character) // 2) + + return decrypted_password + except Exception as e: + print(f"Error decrypting old password: {e}") + return None + +def backup_database(): + """ + Create a backup of the user table before migration + + Returns: + bool: True if backup successful + """ + try: + connection = my.connect( + host=db_config.DB_HOST, + user=db_config.DB_USER, + password=db_config.DB_PASSWORD, + database=db_config.DB_NAME + ) + cursor = connection.cursor() + + # Create backup table + backup_query = """ + CREATE TABLE IF NOT EXISTS user_backup_pre_security AS + SELECT * FROM user + """ + cursor.execute(backup_query) + connection.commit() + + print("āœ… Database backup created successfully (user_backup_pre_security)") + return True + + except Error as e: + print(f"āŒ Failed to create backup: {e}") + return False + finally: + if connection: + connection.close() + +def migrate_passwords(): + """ + Migrate existing passwords to secure hashes + + Returns: + bool: True if migration successful + """ + try: + connection = my.connect( + host=db_config.DB_HOST, + user=db_config.DB_USER, + password=db_config.DB_PASSWORD, + database=db_config.DB_NAME + ) + cursor = connection.cursor() + + # Get all users with old password format + cursor.execute("SELECT u_id, u_name, pwd FROM user") + users = cursor.fetchall() + + if not users: + print("ā„¹ļø No users found to migrate") + return True + + migrated_count = 0 + failed_count = 0 + + for u_id, username, old_pwd in users: + try: + # Check if password is already hashed (bcrypt hashes start with $2b$) + if old_pwd.startswith('$2b$'): + print(f"ā­ļø User {username} already has secure password, skipping") + continue + + # Decrypt old password + decrypted = decrypt_old_password(old_pwd) + if not decrypted: + print(f"āŒ Failed to decrypt password for user: {username}") + failed_count += 1 + continue + + # Hash with new secure method + new_hash = security_manager.hash_password(decrypted) + + # Update database with parameterized query + update_query = "UPDATE user SET pwd = %s WHERE u_id = %s" + cursor.execute(update_query, (new_hash, u_id)) + + migrated_count += 1 + print(f"āœ… Migrated password for user: {username}") + + except Exception as e: + print(f"āŒ Failed to migrate password for user {username}: {e}") + failed_count += 1 + continue + + connection.commit() + + print(f"\nšŸ“Š Migration Summary:") + print(f" āœ… Successfully migrated: {migrated_count} users") + print(f" āŒ Failed migrations: {failed_count} users") + print(f" šŸ“ Total users processed: {len(users)} users") + + if failed_count == 0: + print("\nšŸŽ‰ Password migration completed successfully!") + else: + print(f"\nāš ļø Migration completed with {failed_count} failures") + + return failed_count == 0 + + except Error as e: + print(f"āŒ Migration failed with database error: {e}") + if connection: + connection.rollback() + return False + except Exception as e: + print(f"āŒ Migration failed with unexpected error: {e}") + if connection: + connection.rollback() + return False + finally: + if connection: + connection.close() + +def verify_migration(): + """ + Verify that migration was successful + + Returns: + bool: True if verification successful + """ + try: + connection = my.connect( + host=db_config.DB_HOST, + user=db_config.DB_USER, + password=db_config.DB_PASSWORD, + database=db_config.DB_NAME + ) + cursor = connection.cursor() + + # Check if all passwords are now bcrypt hashes + cursor.execute("SELECT u_name, pwd FROM user") + users = cursor.fetchall() + + non_bcrypt_count = 0 + for username, pwd in users: + if not pwd.startswith('$2b$'): + print(f"āš ļø User {username} still has non-bcrypt password") + non_bcrypt_count += 1 + + if non_bcrypt_count == 0: + print("āœ… Verification successful: All passwords are now securely hashed") + return True + else: + print(f"āŒ Verification failed: {non_bcrypt_count} users still have insecure passwords") + return False + + except Error as e: + print(f"āŒ Verification failed: {e}") + return False + finally: + if connection: + connection.close() + +def main(): + """Main migration function""" + print("šŸ”’ Dataverse Security Migration Script") + print("=" * 50) + + # Check if .env file exists + if not os.path.exists('.env'): + print("āŒ .env file not found!") + print("Please create a .env file with your database configuration.") + print("You can copy .env.example and modify it with your settings.") + return False + + print("šŸ“‹ Starting password migration process...") + + # Step 1: Create backup + print("\n1ļøāƒ£ Creating database backup...") + if not backup_database(): + print("āŒ Backup failed. Migration aborted for safety.") + return False + + # Step 2: Migrate passwords + print("\n2ļøāƒ£ Migrating passwords to secure hashes...") + if not migrate_passwords(): + print("āŒ Migration failed. Please check the errors above.") + return False + + # Step 3: Verify migration + print("\n3ļøāƒ£ Verifying migration...") + if not verify_migration(): + print("āŒ Verification failed. Please check the migration manually.") + return False + + print("\nšŸŽ‰ Migration completed successfully!") + print("\nšŸ“ Next steps:") + print(" 1. Test login with existing users") + print(" 2. Create new users to test registration") + print(" 3. Remove backup table when satisfied: DROP TABLE user_backup_pre_security;") + + return True + +if __name__ == "__main__": + try: + success = main() + sys.exit(0 if success else 1) + except KeyboardInterrupt: + print("\n\nāš ļø Migration interrupted by user") + sys.exit(1) + except Exception as e: + print(f"\nāŒ Unexpected error: {e}") + sys.exit(1) \ No newline at end of file diff --git a/security_implementation_guide.md b/security_implementation_guide.md new file mode 100644 index 0000000..7f88581 --- /dev/null +++ b/security_implementation_guide.md @@ -0,0 +1,386 @@ +# Security Implementation Guide for Dataverse + +## Step 1: Update Requirements + +First, add the new security dependencies to `installation/requirements.txt`: + +```txt +bcrypt>=4.0.1 +python-dotenv>=1.0.0 +``` + +## Step 2: Create Environment Configuration + +### Create `.env.example`: +```env +# Database Configuration +DB_HOST=localhost +DB_USER=your_username +DB_PASSWORD=your_secure_password +DB_NAME=DATAVERSE + +# Security Configuration +SECRET_KEY=your_secret_key_here +SESSION_TIMEOUT=3600 +``` + +### Update `software/db_config.py`: +```python +import os +from dotenv import load_dotenv + +# Load environment variables +load_dotenv() + +DB_HOST = os.getenv('DB_HOST', 'localhost') +DB_USER = os.getenv('DB_USER', 'root') +DB_PASSWORD = os.getenv('DB_PASSWORD', '') +DB_NAME = os.getenv('DB_NAME', 'DATAVERSE') +SECRET_KEY = os.getenv('SECRET_KEY', 'default-secret-key') +SESSION_TIMEOUT = int(os.getenv('SESSION_TIMEOUT', 3600)) +``` + +## Step 3: Implement Secure Password Hashing + +### Create `software/security.py`: +```python +import bcrypt +import secrets +import time +from typing import Optional, Dict + +class SecurityManager: + def __init__(self): + self.active_sessions: Dict[str, Dict] = {} + + def hash_password(self, password: str) -> str: + """Hash password using bcrypt with salt""" + salt = bcrypt.gensalt() + hashed = bcrypt.hashpw(password.encode('utf-8'), salt) + return hashed.decode('utf-8') + + def verify_password(self, password: str, hashed: str) -> bool: + """Verify password against hash""" + try: + return bcrypt.checkpw(password.encode('utf-8'), hashed.encode('utf-8')) + except Exception: + return False + + def create_session(self, user_id: str) -> str: + """Create secure session token""" + token = secrets.token_urlsafe(32) + self.active_sessions[token] = { + 'user_id': user_id, + 'created_at': time.time(), + 'last_activity': time.time() + } + return token + + def validate_session(self, token: str) -> Optional[str]: + """Validate session token and return user_id if valid""" + if token not in self.active_sessions: + return None + + session = self.active_sessions[token] + current_time = time.time() + + # Check if session expired + if current_time - session['last_activity'] > SESSION_TIMEOUT: + del self.active_sessions[token] + return None + + # Update last activity + session['last_activity'] = current_time + return session['user_id'] + + def destroy_session(self, token: str) -> bool: + """Destroy session token""" + if token in self.active_sessions: + del self.active_sessions[token] + return True + return False + +# Global security manager instance +security_manager = SecurityManager() +``` + +## Step 4: Input Validation + +### Create `software/validators.py`: +```python +import re +from typing import Optional + +class InputValidator: + @staticmethod + def validate_username(username: str) -> bool: + """Validate username format""" + if not username or len(username) < 3 or len(username) > 50: + return False + # Allow alphanumeric and underscore only + return re.match(r'^[a-zA-Z0-9_]+$', username) is not None + + @staticmethod + def validate_password(password: str) -> bool: + """Validate password strength""" + if not password or len(password) < 8: + return False + # At least one uppercase, lowercase, digit + has_upper = re.search(r'[A-Z]', password) + has_lower = re.search(r'[a-z]', password) + has_digit = re.search(r'\d', password) + return all([has_upper, has_lower, has_digit]) + + @staticmethod + def validate_country(country: str) -> bool: + """Validate country name""" + if not country or len(country) > 100: + return False + # Allow letters, spaces, hyphens only + return re.match(r'^[a-zA-Z\s\-]+$', country) is not None + + @staticmethod + def sanitize_input(input_str: str) -> str: + """Basic input sanitization""" + if not input_str: + return "" + # Remove potential SQL injection characters + dangerous_chars = ["'", '"', ';', '--', '/*', '*/', 'xp_', 'sp_'] + sanitized = input_str + for char in dangerous_chars: + sanitized = sanitized.replace(char, '') + return sanitized.strip() +``` + +## Step 5: Update Database Operations + +### Update `software/manage_data.py`: +```python +import mysql.connector as my +from mysql.connector import Error +import db_config +from security import security_manager +from validators import InputValidator + +# Secure database connection +def get_db_connection(): + try: + connection = my.connect( + host=db_config.DB_HOST, + user=db_config.DB_USER, + password=db_config.DB_PASSWORD, + database=db_config.DB_NAME, + autocommit=False # Enable transactions + ) + return connection + except Error as e: + print(f"Database connection error: {e}") + return None + +def create_user_secure(username: str, password: str, country: str) -> tuple: + """Securely create new user with proper validation""" + validator = InputValidator() + + # Validate inputs + if not validator.validate_username(username): + return False, "Invalid username format" + + if not validator.validate_password(password): + return False, "Password must be at least 8 characters with uppercase, lowercase, and digit" + + if not validator.validate_country(country): + return False, "Invalid country format" + + connection = get_db_connection() + if not connection: + return False, "Database connection failed" + + try: + cursor = connection.cursor() + + # Check if username already exists (parameterized query) + check_query = "SELECT u_name FROM user WHERE u_name = %s" + cursor.execute(check_query, (username,)) + + if cursor.fetchone(): + return False, "Username already exists" + + # Hash password securely + hashed_password = security_manager.hash_password(password) + + # Generate user ID + import datetime + u_id = datetime.datetime.now().strftime("%y%m%d%H%M%S") + + # Insert user with parameterized query + insert_query = """ + INSERT INTO user (u_id, u_name, pwd, country) + VALUES (%s, %s, %s, %s) + """ + cursor.execute(insert_query, (u_id, username, hashed_password, country)) + + connection.commit() + return True, f"Account created successfully! User ID: {u_id}" + + except Error as e: + connection.rollback() + return False, f"Database error: {e}" + finally: + if connection: + connection.close() + +def authenticate_user_secure(username: str, password: str) -> tuple: + """Securely authenticate user""" + validator = InputValidator() + + if not validator.validate_username(username): + return False, "Invalid username format", None + + connection = get_db_connection() + if not connection: + return False, "Database connection failed", None + + try: + cursor = connection.cursor() + + # Get user data with parameterized query + query = "SELECT u_id, pwd FROM user WHERE u_name = %s" + cursor.execute(query, (username,)) + result = cursor.fetchone() + + if not result: + return False, "User not found", None + + u_id, stored_hash = result + + # Verify password + if security_manager.verify_password(password, stored_hash): + # Create session token + session_token = security_manager.create_session(str(u_id)) + return True, "Login successful", session_token + else: + return False, "Invalid password", None + + except Error as e: + return False, f"Database error: {e}", None + finally: + if connection: + connection.close() +``` + +## Step 6: Migration Script + +### Create `migration_script.py`: +```python +""" +Migration script to update existing passwords to secure hashes +Run this ONCE after implementing the new security system +""" + +import mysql.connector as my +from security import security_manager +import db_config + +def migrate_passwords(): + """Migrate existing passwords to secure hashes""" + connection = my.connect( + host=db_config.DB_HOST, + user=db_config.DB_USER, + password=db_config.DB_PASSWORD, + database=db_config.DB_NAME + ) + + cursor = connection.cursor() + + try: + # Get all users with old password format + cursor.execute("SELECT u_id, u_name, pwd FROM user") + users = cursor.fetchall() + + for u_id, username, old_pwd in users: + # Decrypt old password (using the old decrypt function) + decrypted = decrypt_old_password(old_pwd) + + # Hash with new secure method + new_hash = security_manager.hash_password(decrypted) + + # Update database + update_query = "UPDATE user SET pwd = %s WHERE u_id = %s" + cursor.execute(update_query, (new_hash, u_id)) + + print(f"Migrated password for user: {username}") + + connection.commit() + print("Password migration completed successfully!") + + except Exception as e: + connection.rollback() + print(f"Migration failed: {e}") + finally: + connection.close() + +def decrypt_old_password(encrypted_password): + """Decrypt password using old method (for migration only)""" + password_length = len(encrypted_password) + decrypted_password = "" + if password_length % 2 == 0: + transformed_pwd = encrypted_password[int(password_length / 2):] + transformed_pwd += encrypted_password[:int(password_length / 2)] + else: + transformed_pwd = encrypted_password[int(password_length / 2) + 1:] + transformed_pwd += encrypted_password[:int(password_length / 2) + 1] + for character in transformed_pwd: + decrypted_password += chr(ord(character) // 2) + return decrypted_password + +if __name__ == "__main__": + migrate_passwords() +``` + +## Step 7: Update Main Application + +### Update relevant functions in `software/main.py`: +```python +from security import security_manager +from validators import InputValidator +import manage_data + +# Replace the old encrypt/decrypt functions with secure versions +def create_account_secure(username, password, country): + """Secure account creation""" + success, message = manage_data.create_user_secure(username, password, country) + if success: + messagebox.showinfo("Success", message) + else: + messagebox.showerror("Error", message) + +def login_secure(username, password): + """Secure login process""" + success, message, session_token = manage_data.authenticate_user_secure(username, password) + if success: + # Store session token for the user session + global current_session_token + current_session_token = session_token + messagebox.showinfo("Success", message) + # Proceed to user menu + else: + messagebox.showerror("Error", message) +``` + +## Testing the Implementation + +1. **Create `.env` file** with your database credentials +2. **Install new dependencies**: `pip install bcrypt python-dotenv` +3. **Run migration script** to update existing passwords +4. **Test user registration** with new validation +5. **Test login** with secure authentication +6. **Verify session management** works correctly + +## Security Benefits + +āœ… **Strong Password Hashing**: bcrypt with salt prevents rainbow table attacks +āœ… **Environment Variables**: No more hardcoded credentials +āœ… **SQL Injection Prevention**: Parameterized queries eliminate injection risks +āœ… **Input Validation**: Prevents malicious data entry +āœ… **Session Management**: Secure token-based authentication +āœ… **Backward Compatibility**: Migration script preserves existing users \ No newline at end of file diff --git a/software/db_config.py b/software/db_config.py index 09f3c2e..70c5f3e 100644 --- a/software/db_config.py +++ b/software/db_config.py @@ -1,3 +1,13 @@ -DB_HOST = "localhost" -DB_USER = "root" -DB_PASSWORD = "tejas123" \ No newline at end of file +import os +from dotenv import load_dotenv + +# Load environment variables from .env file +load_dotenv() + +DB_HOST = os.getenv('DB_HOST', 'localhost') +DB_USER = os.getenv('DB_USER', 'root') +DB_PASSWORD = os.getenv('DB_PASSWORD', '') +DB_NAME = os.getenv('DB_NAME', 'DATAVERSE') +SECRET_KEY = os.getenv('SECRET_KEY', 'default-secret-key-change-in-production') +SESSION_TIMEOUT = int(os.getenv('SESSION_TIMEOUT', 3600)) +APP_DEBUG = os.getenv('APP_DEBUG', 'False').lower() == 'true' \ No newline at end of file diff --git a/software/main.py b/software/main.py index 13519ea..a901f82 100644 --- a/software/main.py +++ b/software/main.py @@ -21,6 +21,9 @@ APP_THEME = "appconfig" +# Global session management +current_session_token = None + #===================================================================================================================plot colors colors=["#440154", "#3b528b","#21918c", "#5ec962", "#fde725","#f89540", "#e16462","#b12a90", "#6a00a8", "#0d0887", "#3474eb", "#5ec962", "yellow", "#f89540", "tomato","tan"] maxLimit = 16 # maximum limit for colors @@ -125,8 +128,12 @@ def switch_theme(): cursor.execute("CREATE TABLE IF NOT EXISTS finance (u_id BIGINT, salary FLOAT DEFAULT 0, gold FLOAT DEFAULT 0, stocks FLOAT DEFAULT 0, commodity FLOAT DEFAULT 0, sales FLOAT DEFAULT 0, expenditure FLOAT DEFAULT 0, total FLOAT AS (salary + gold + stocks + commodity + sales - expenditure), entryDate date);") #=================================================================================================================================== #============================================================================================Dataverse Operations -#======================================================================password encryption +#======================================================================password encryption (DEPRECATED - kept for migration only) def encrypt(pwd): + """ + DEPRECATED: Old encryption method - kept only for migration purposes + New code should use security_manager.hash_password() instead + """ n=len(pwd) e="" t=pwd[int(n/2):] @@ -139,20 +146,29 @@ def login(b1,b2,b3,b4,preview_image): switch(b1,b2,b3,b4) preview_image.place_forget() def show_message(): - message=manage_data.check_credentials(f"{user.get()}",f"{pwd.get()}") - word=message.split(' ')[0] - if word=='Login': - messagebox.showinfo(title="", message=message,icon="info") - global menu + username = f"{user.get()}" + password = f"{pwd.get()}" + + # Use secure authentication + success, message, session_token = manage_data.authenticate_user_secure(username, password) + + if success: + messagebox.showinfo(title="Success", message="Login successful!",icon="info") + global menu, current_session_token menu.pack_forget() form.pack_forget() - user_menu(message.split(' ')[-1],f"{user.get()}") - elif word=='No': - messagebox.showinfo(title="", message=message,icon="info") - elif word=='Incorrect': - messagebox.showinfo(title="", message=message,icon="error") + + # Store session token globally for the session + current_session_token = session_token + + # Get user info from session + session_info = manage_data.security_manager.validate_session(session_token) + if session_info: + user_menu(session_info['user_id'], username) + else: + messagebox.showerror(title="Error", message="Session creation failed",icon="error") else: - messagebox.showinfo(title="", message="Some unknown error occured.",icon="warning") + messagebox.showerror(title="Login Failed", message=message,icon="error") global relation relation.pack_forget() global text @@ -181,20 +197,19 @@ def create(b2,b1,b3,b4,preview_image): def show_message(u_name,pwd,country,names): u_name=f"{u_name.get()}" pwd=f"{pwd.get()}" - pwd=encrypt(pwd) country=f"{country.get()}" - if u_name in names: - messagebox.showinfo(title="Username Not Available", message="That username is already taken. Try another one.",icon="info") - else: - u_id = datetime.datetime.now().strftime("%y%m%d%H%M%S") - q="insert into user values({},'{}','{}','{}')".format(u_id,u_name,pwd,country) - cursor.execute(q) - mycon.commit() - msg="Account Created Successfully! āœ“\nYour User ID is: {}\n".format(u_id) + + # Use secure user creation + success, message, user_id = manage_data.create_user_secure(u_name, pwd, country) + + if success: global text - text=Label(font="poppins 10 bold",fg=theme["White"],bg=theme["Black"],text=msg) + text=Label(font="poppins 10 bold",fg=theme["White"],bg=theme["Black"],text=message) text.pack(fill=X,pady=35,padx=(0,200)) - messagebox.showinfo(title="Important", message="This User ID is required while deleting the account.",icon="warning") + messagebox.showinfo(title="Success", message="Account created successfully!",icon="info") + messagebox.showinfo(title="Important", message=f"Your User ID is: {user_id}\nThis User ID is required while deleting the account.",icon="warning") + else: + messagebox.showerror(title="Error", message=message,icon="error") switch(b2,b1,b3,b4) preview_image.place_forget() diff --git a/software/manage_data.py b/software/manage_data.py index d0ff845..68c45a5 100644 --- a/software/manage_data.py +++ b/software/manage_data.py @@ -11,38 +11,271 @@ from matplotlib.widgets import Cursor as lines import mysql.connector as my from mysql.connector import Error +from security import security_manager +from validators import InputValidator #===============================================================================================================plot colors colors=["#440154", "#3b528b","#21918c", "#5ec962", "#fde725","#f89540", "#e16462","#b12a90", "#6a00a8", "#0d0887", "#3474eb", "#5ec962", "yellow", "#f89540", "tomato","tan"] #==================================================================================================connecting MySQL +def get_db_connection(): + """ + Create a secure database connection with proper error handling + + Returns: + mysql.connector.connection: Database connection or None if failed + """ + try: + connection = my.connect( + host=db_config.DB_HOST, + user=db_config.DB_USER, + password=db_config.DB_PASSWORD, + database=db_config.DB_NAME, + autocommit=False, # Enable transactions + charset='utf8mb4', + collation='utf8mb4_unicode_ci', + sql_mode='STRICT_TRANS_TABLES,NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO' + ) + return connection + except Error as e: + print(f"Database connection error: {e}") + if db_config.APP_DEBUG: + messagebox.showerror("Database Error", f"Failed to connect to the database: {e}\n\nPlease check your .env configuration.") + return None + +# Initialize database connection for backward compatibility try: - mycon = my.connect( - host=db_config.DB_HOST, - user=db_config.DB_USER, - passwd=db_config.DB_PASSWORD -) - cursor = mycon.cursor() + mycon = get_db_connection() + if mycon: + cursor = mycon.cursor() + # Ensure database and tables exist + cursor.execute(f"CREATE DATABASE IF NOT EXISTS {db_config.DB_NAME}") + cursor.execute(f"USE {db_config.DB_NAME}") + mycon.commit() + else: + raise Error("Failed to establish database connection") except Error as e: - print(f"Error connecting to MySQL: {e}") - messagebox.showerror("Database Error", f"Failed to connect to the database: {e}\n\nExecute the following query in your MYSQL Workbench or MySQL Shell and then try again:\n\nCREATE DATABASE DATAVERSE;") - exit() # Exit the program if the database connection fails + print(f"Error initializing database: {e}") + if not db_config.APP_DEBUG: + messagebox.showerror("Database Error", "Failed to connect to the database. Please check your configuration.") + exit() z = 0 # Global variable to track failed login attempts +#==================================================================================================secure user operations +def create_user_secure(username: str, password: str, country: str) -> tuple: + """ + Securely create new user with proper validation and hashing + + Args: + username (str): Username for new account + password (str): Plain text password + country (str): User's country + + Returns: + tuple: (success: bool, message: str, user_id: str or None) + """ + validator = InputValidator() + + # Validate inputs + is_valid, error_msg = validator.validate_username(username) + if not is_valid: + return False, f"Invalid username: {error_msg}", None + + is_valid, error_msg = validator.validate_password(password) + if not is_valid: + return False, f"Invalid password: {error_msg}", None + + is_valid, error_msg = validator.validate_country(country) + if not is_valid: + return False, f"Invalid country: {error_msg}", None + + connection = get_db_connection() + if not connection: + return False, "Database connection failed", None + + try: + cursor = connection.cursor() + + # Check if username already exists (parameterized query) + check_query = "SELECT u_name FROM user WHERE u_name = %s" + cursor.execute(check_query, (username,)) + + if cursor.fetchone(): + return False, "Username already exists", None + + # Hash password securely + try: + hashed_password = security_manager.hash_password(password) + except Exception as e: + return False, f"Password hashing failed: {e}", None + + # Generate user ID + u_id = datetime.datetime.now().strftime("%y%m%d%H%M%S") + + # Insert user with parameterized query + insert_query = """ + INSERT INTO user (u_id, u_name, pwd, country) + VALUES (%s, %s, %s, %s) + """ + cursor.execute(insert_query, (u_id, username, hashed_password, country)) + + connection.commit() + return True, f"Account created successfully! User ID: {u_id}", u_id + + except Error as e: + connection.rollback() + print(f"Database error during user creation: {e}") + return False, "Failed to create account due to database error", None + except Exception as e: + connection.rollback() + print(f"Unexpected error during user creation: {e}") + return False, "Failed to create account due to unexpected error", None + finally: + if connection: + connection.close() + +def authenticate_user_secure(username: str, password: str) -> tuple: + """ + Securely authenticate user with proper validation and session management + + Args: + username (str): Username to authenticate + password (str): Plain text password + + Returns: + tuple: (success: bool, message: str, session_token: str or None) + """ + validator = InputValidator() + + # Basic input validation + is_valid, error_msg = validator.validate_username(username) + if not is_valid: + return False, f"Invalid username format: {error_msg}", None + + if not password: + return False, "Password cannot be empty", None + + # Check if account is locked due to failed attempts + if security_manager.is_account_locked(username): + return False, "Account temporarily locked due to multiple failed login attempts. Please try again later.", None + + connection = get_db_connection() + if not connection: + return False, "Database connection failed", None + + try: + cursor = connection.cursor() + + # Get user data with parameterized query + query = "SELECT u_id, pwd FROM user WHERE u_name = %s" + cursor.execute(query, (username,)) + result = cursor.fetchone() + + if not result: + # Record failed attempt even for non-existent users to prevent enumeration + security_manager.record_failed_attempt(username) + return False, "Invalid username or password", None + + u_id, stored_hash = result + + # Verify password + if security_manager.verify_password(password, stored_hash): + # Clear any failed attempts on successful login + security_manager.clear_failed_attempts(username) + + # Create session token + session_token = security_manager.create_session(str(u_id), username) + return True, "Login successful", session_token + else: + # Record failed attempt + is_locked = security_manager.record_failed_attempt(username) + if is_locked: + return False, "Too many failed attempts. Account temporarily locked.", None + else: + return False, "Invalid username or password", None + + except Error as e: + print(f"Database error during authentication: {e}") + return False, "Authentication failed due to database error", None + except Exception as e: + print(f"Unexpected error during authentication: {e}") + return False, "Authentication failed due to unexpected error", None + finally: + if connection: + connection.close() + +def validate_user_session(session_token: str) -> tuple: + """ + Validate user session and return user information + + Args: + session_token (str): Session token to validate + + Returns: + tuple: (is_valid: bool, user_info: dict or None) + """ + validator = InputValidator() + + if not validator.validate_session_token(session_token): + return False, None + + session_info = security_manager.validate_session(session_token) + if session_info: + return True, { + 'user_id': session_info['user_id'], + 'username': session_info['username'], + 'last_activity': session_info['last_activity'] + } + + return False, None + +def logout_user_secure(session_token: str) -> bool: + """ + Securely logout user by destroying session + + Args: + session_token (str): Session token to destroy + + Returns: + bool: True if logout successful + """ + return security_manager.destroy_session(session_token) + #=========================================================================================view data def view_data(user_id): + """ + Securely view user's financial data using parameterized queries + + Args: + user_id: User ID to fetch data for + + Returns: + str: Formatted data table or error message + """ + connection = get_db_connection() + if not connection: + return "Database connection failed" + try: - query = "select * from finance where u_id={}".format(user_id) - cursor.execute(query) + cursor = connection.cursor() + # Use parameterized query to prevent SQL injection + query = "SELECT * FROM finance WHERE u_id = %s ORDER BY entryDate DESC" + cursor.execute(query, (user_id,)) result_set = cursor.fetchall() + if len(result_set) == 0: result_message = "Your dataSet is empty." else: columns = [col[0] for col in cursor.description] result_table = tabulate.tabulate(result_set, headers=columns, tablefmt="pretty") result_message = result_table + except Error as e: result_message = f"Error fetching data: {e}" print(f"Database error: {e}") + finally: + if connection: + connection.close() + return result_message #==============================================================================================password decryption @@ -59,42 +292,54 @@ def decrypt(encrypted_password): decrypted_password += chr(ord(character) // 2) return decrypted_password -#==================================================================================================add data +#==================================================================================================add data (legacy function - kept for backward compatibility) def check_credentials(username, password): - try: - query = "select u_name from user" - cursor.execute(query) - user_list = cursor.fetchall() - usernames = [user[0] for user in user_list] - if str(username) not in usernames: - message = "No account exists with that username." + """ + Legacy credential checking function - redirects to secure authentication + + Args: + username: Username to check + password: Password to verify + + Returns: + str: Authentication result message + """ + success, message, session_token = authenticate_user_secure(username, password) + + if success: + # Extract user ID from session for backward compatibility + session_info = security_manager.validate_session(session_token) + if session_info: + return f"Login Successful. āœ“\nUser ID: {session_info['user_id']}" else: - query = "select pwd from user where u_name='{}'".format(username) - cursor.execute(query) - fetched_password = cursor.fetchall() - decrypted_password = decrypt(fetched_password[0][0]) - if decrypted_password == password: - query = "select u_id from user where u_name='{}'".format(username) - cursor.execute(query) - user_id = cursor.fetchall()[0][0] - message = f"Login Successful. āœ“\nUser ID: {user_id}" - else: - message = "Incorrect password! āœ–" - global z - z += 1 - if z >= 2: - print("There have been more than 1 failed login attempts. Closing the system.") - except Error as e: - message = f"Error during login: {e}" - print(f"Database error: {e}") - return message + return "Login successful but session creation failed" + else: + return message #============================================================================================================fetch user's finance data def fetch_data(user_id): + """ + Securely fetch user's financial data using parameterized queries + + Args: + user_id: User ID to fetch data for + + Returns: + list or None: [column_names, data_pool] or None if no data + """ + connection = get_db_connection() + if not connection: + print("Database connection failed") + return None + try: - query = "select * from finance where u_id={}".format(user_id) - cursor.execute(query) + cursor = connection.cursor() + + # Use parameterized query + query = "SELECT * FROM finance WHERE u_id = %s ORDER BY entryDate" + cursor.execute(query, (user_id,)) result_set = cursor.fetchall() + if len(result_set) == 0: return None else: @@ -102,18 +347,24 @@ def fetch_data(user_id): schema = cursor.fetchall() column_names = [column[0] for column in schema] data_pool = {} + for column_name in column_names: column_data = [] - query = "select {} from finance where u_id={}".format(column_name, user_id) - cursor.execute(query) + # Use parameterized query for each column + query = f"SELECT {column_name} FROM finance WHERE u_id = %s ORDER BY entryDate" + cursor.execute(query, (user_id,)) values = cursor.fetchall() for value in values: column_data.append(value[0]) data_pool[column_name] = column_data return [column_names, data_pool] + except Error as e: print(f"Error fetching user data: {e}") return None + finally: + if connection: + connection.close() #=======================================================================================================Predictive Analytics (Linear Regression) ''' def predict_future_expenditure(pool): diff --git a/software/security.py b/software/security.py new file mode 100644 index 0000000..c2f61f0 --- /dev/null +++ b/software/security.py @@ -0,0 +1,210 @@ +import bcrypt +import secrets +import time +from typing import Optional, Dict +import db_config + +class SecurityManager: + """ + Comprehensive security manager for password hashing and session management + """ + + def __init__(self): + self.active_sessions: Dict[str, Dict] = {} + self.failed_attempts: Dict[str, Dict] = {} + self.max_failed_attempts = 5 + self.lockout_duration = 900 # 15 minutes + + def hash_password(self, password: str) -> str: + """ + Hash password using bcrypt with salt + + Args: + password (str): Plain text password + + Returns: + str: Hashed password + """ + if not password: + raise ValueError("Password cannot be empty") + + salt = bcrypt.gensalt(rounds=12) # Higher rounds for better security + hashed = bcrypt.hashpw(password.encode('utf-8'), salt) + return hashed.decode('utf-8') + + def verify_password(self, password: str, hashed: str) -> bool: + """ + Verify password against hash + + Args: + password (str): Plain text password + hashed (str): Stored hash + + Returns: + bool: True if password matches + """ + try: + if not password or not hashed: + return False + return bcrypt.checkpw(password.encode('utf-8'), hashed.encode('utf-8')) + except Exception as e: + print(f"Password verification error: {e}") + return False + + def create_session(self, user_id: str, username: str) -> str: + """ + Create secure session token + + Args: + user_id (str): User ID + username (str): Username + + Returns: + str: Session token + """ + token = secrets.token_urlsafe(32) + current_time = time.time() + + self.active_sessions[token] = { + 'user_id': user_id, + 'username': username, + 'created_at': current_time, + 'last_activity': current_time, + 'ip_address': 'localhost' # Can be extended for web version + } + + # Clean up old sessions + self._cleanup_expired_sessions() + + return token + + def validate_session(self, token: str) -> Optional[Dict]: + """ + Validate session token and return session info if valid + + Args: + token (str): Session token + + Returns: + Optional[Dict]: Session info if valid, None otherwise + """ + if not token or token not in self.active_sessions: + return None + + session = self.active_sessions[token] + current_time = time.time() + + # Check if session expired + if current_time - session['last_activity'] > db_config.SESSION_TIMEOUT: + del self.active_sessions[token] + return None + + # Update last activity + session['last_activity'] = current_time + return session + + def destroy_session(self, token: str) -> bool: + """ + Destroy session token + + Args: + token (str): Session token + + Returns: + bool: True if session was destroyed + """ + if token in self.active_sessions: + del self.active_sessions[token] + return True + return False + + def record_failed_attempt(self, username: str) -> bool: + """ + Record failed login attempt and check if account should be locked + + Args: + username (str): Username that failed login + + Returns: + bool: True if account is now locked + """ + current_time = time.time() + + if username not in self.failed_attempts: + self.failed_attempts[username] = { + 'count': 0, + 'first_attempt': current_time, + 'locked_until': 0 + } + + attempt_info = self.failed_attempts[username] + + # Reset counter if it's been more than lockout duration since first attempt + if current_time - attempt_info['first_attempt'] > self.lockout_duration: + attempt_info['count'] = 0 + attempt_info['first_attempt'] = current_time + + attempt_info['count'] += 1 + + # Lock account if too many failed attempts + if attempt_info['count'] >= self.max_failed_attempts: + attempt_info['locked_until'] = current_time + self.lockout_duration + return True + + return False + + def is_account_locked(self, username: str) -> bool: + """ + Check if account is currently locked + + Args: + username (str): Username to check + + Returns: + bool: True if account is locked + """ + if username not in self.failed_attempts: + return False + + attempt_info = self.failed_attempts[username] + current_time = time.time() + + if attempt_info['locked_until'] > current_time: + return True + + # Unlock account if lockout period has passed + if attempt_info['locked_until'] > 0: + attempt_info['count'] = 0 + attempt_info['locked_until'] = 0 + + return False + + def clear_failed_attempts(self, username: str): + """ + Clear failed attempts for successful login + + Args: + username (str): Username to clear attempts for + """ + if username in self.failed_attempts: + del self.failed_attempts[username] + + def _cleanup_expired_sessions(self): + """Clean up expired sessions""" + current_time = time.time() + expired_tokens = [] + + for token, session in self.active_sessions.items(): + if current_time - session['last_activity'] > db_config.SESSION_TIMEOUT: + expired_tokens.append(token) + + for token in expired_tokens: + del self.active_sessions[token] + + def get_active_sessions_count(self) -> int: + """Get count of active sessions""" + self._cleanup_expired_sessions() + return len(self.active_sessions) + +# Global security manager instance +security_manager = SecurityManager() \ No newline at end of file diff --git a/software/validators.py b/software/validators.py new file mode 100644 index 0000000..5ca92d0 --- /dev/null +++ b/software/validators.py @@ -0,0 +1,256 @@ +import re +from typing import Optional, List + +class InputValidator: + """ + Comprehensive input validation and sanitization utilities + """ + + # SQL injection patterns to detect + SQL_INJECTION_PATTERNS = [ + r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|CREATE|ALTER|EXEC|UNION|SCRIPT)\b)", + r"(--|#|/\*|\*/)", + r"(\b(OR|AND)\s+\d+\s*=\s*\d+)", + r"(\bOR\s+\w+\s*=\s*\w+)", + r"(\';|\"\;)", + r"(\bxp_|\bsp_)" + ] + + @staticmethod + def validate_username(username: str) -> tuple[bool, str]: + """ + Validate username format and security + + Args: + username (str): Username to validate + + Returns: + tuple[bool, str]: (is_valid, error_message) + """ + if not username: + return False, "Username cannot be empty" + + if len(username) < 3: + return False, "Username must be at least 3 characters long" + + if len(username) > 50: + return False, "Username cannot exceed 50 characters" + + # Allow alphanumeric, underscore, and hyphen only + if not re.match(r'^[a-zA-Z0-9_-]+$', username): + return False, "Username can only contain letters, numbers, underscore, and hyphen" + + # Check for SQL injection patterns + if InputValidator._contains_sql_injection(username): + return False, "Username contains invalid characters" + + # Username cannot start with number + if username[0].isdigit(): + return False, "Username cannot start with a number" + + return True, "" + + @staticmethod + def validate_password(password: str) -> tuple[bool, str]: + """ + Validate password strength and security + + Args: + password (str): Password to validate + + Returns: + tuple[bool, str]: (is_valid, error_message) + """ + if not password: + return False, "Password cannot be empty" + + if len(password) < 8: + return False, "Password must be at least 8 characters long" + + if len(password) > 128: + return False, "Password cannot exceed 128 characters" + + # Check for required character types + has_upper = bool(re.search(r'[A-Z]', password)) + has_lower = bool(re.search(r'[a-z]', password)) + has_digit = bool(re.search(r'\d', password)) + has_special = bool(re.search(r'[!@#$%^&*(),.?":{}|<>]', password)) + + missing_requirements = [] + if not has_upper: + missing_requirements.append("uppercase letter") + if not has_lower: + missing_requirements.append("lowercase letter") + if not has_digit: + missing_requirements.append("digit") + if not has_special: + missing_requirements.append("special character") + + if missing_requirements: + return False, f"Password must contain at least one: {', '.join(missing_requirements)}" + + # Check for common weak patterns + if InputValidator._is_weak_password(password): + return False, "Password is too common or weak" + + return True, "" + + @staticmethod + def validate_country(country: str) -> tuple[bool, str]: + """ + Validate country name format + + Args: + country (str): Country name to validate + + Returns: + tuple[bool, str]: (is_valid, error_message) + """ + if not country: + return False, "Country cannot be empty" + + if len(country) > 100: + return False, "Country name cannot exceed 100 characters" + + # Allow letters, spaces, hyphens, and apostrophes only + if not re.match(r"^[a-zA-Z\s\-']+$", country): + return False, "Country name contains invalid characters" + + # Check for SQL injection + if InputValidator._contains_sql_injection(country): + return False, "Country name contains invalid characters" + + return True, "" + + @staticmethod + def validate_numeric_input(value: str, field_name: str, min_val: float = None, max_val: float = None) -> tuple[bool, str]: + """ + Validate numeric input for financial data + + Args: + value (str): Value to validate + field_name (str): Name of the field for error messages + min_val (float, optional): Minimum allowed value + max_val (float, optional): Maximum allowed value + + Returns: + tuple[bool, str]: (is_valid, error_message) + """ + if not value: + return False, f"{field_name} cannot be empty" + + try: + num_value = float(value) + except ValueError: + return False, f"{field_name} must be a valid number" + + if min_val is not None and num_value < min_val: + return False, f"{field_name} cannot be less than {min_val}" + + if max_val is not None and num_value > max_val: + return False, f"{field_name} cannot exceed {max_val}" + + # Check for reasonable financial values (not more than 1 trillion) + if abs(num_value) > 1_000_000_000_000: + return False, f"{field_name} value is unreasonably large" + + return True, "" + + @staticmethod + def sanitize_input(input_str: str) -> str: + """ + Basic input sanitization + + Args: + input_str (str): Input string to sanitize + + Returns: + str: Sanitized string + """ + if not input_str: + return "" + + # Remove potential dangerous characters + sanitized = input_str.strip() + + # Remove null bytes + sanitized = sanitized.replace('\x00', '') + + # Remove control characters except newline and tab + sanitized = ''.join(char for char in sanitized if ord(char) >= 32 or char in '\n\t') + + return sanitized + + @staticmethod + def _contains_sql_injection(input_str: str) -> bool: + """ + Check if input contains potential SQL injection patterns + + Args: + input_str (str): String to check + + Returns: + bool: True if potential SQL injection detected + """ + input_upper = input_str.upper() + + for pattern in InputValidator.SQL_INJECTION_PATTERNS: + if re.search(pattern, input_upper, re.IGNORECASE): + return True + + return False + + @staticmethod + def _is_weak_password(password: str) -> bool: + """ + Check if password is commonly used or weak + + Args: + password (str): Password to check + + Returns: + bool: True if password is weak + """ + # Common weak passwords + weak_passwords = { + 'password', '12345678', 'qwerty123', 'abc123456', + 'password123', '123456789', 'welcome123', 'admin123', + 'letmein123', 'monkey123', 'dragon123', 'master123' + } + + if password.lower() in weak_passwords: + return True + + # Check for keyboard patterns + keyboard_patterns = ['qwerty', 'asdfgh', 'zxcvbn', '123456', '654321'] + password_lower = password.lower() + + for pattern in keyboard_patterns: + if pattern in password_lower: + return True + + # Check for repeated characters (more than 3 in a row) + if re.search(r'(.)\1{3,}', password): + return True + + return False + + @staticmethod + def validate_session_token(token: str) -> bool: + """ + Validate session token format + + Args: + token (str): Session token to validate + + Returns: + bool: True if token format is valid + """ + if not token: + return False + + # Session tokens should be URL-safe base64 strings of specific length + if not re.match(r'^[A-Za-z0-9_-]{43}$', token): + return False + + return True \ No newline at end of file