Upload 4 files

codebase/auth.py

@@ -0,0 +1,88 @@
+import re
+import hashlib
+import time
+from datetime import datetime, timedelta
+
+class AuthenticationManager:
+    def __init__(self):
+        self.failed_attempts = {}
+        self.session_timeout = 7200  # 2 hours in seconds
+        
+    def validate_password(self, password):
+        """Validate password meets security requirements"""
+        if len(password) < 8:
+            return False, "Password must be at least 8 characters"
+        
+        # Fixed regex to properly handle special characters
+        if not re.match(r'^[a-zA-Z0-9!@#$%^&*()_+\-=\[\]{};:,.<>?]+$', password):
+            return False, "Password contains invalid characters"
+            
+        return True, "Password valid"
+    
+    def authenticate_user(self, username, password):
+        """Authenticate user with username and password"""
+        try:
+            # Check for account lockout
+            if self.is_account_locked(username):
+                return False, "Account temporarily locked due to too many failed attempts"
+            
+            # Validate credentials
+            if self.verify_credentials(username, password):
+                self.reset_failed_attempts(username)
+                return True, "Authentication successful"
+            else:
+                self.record_failed_attempt(username)
+                return False, "Invalid username or password"
+                
+        except Exception as e:
+            return False, f"Authentication error: {str(e)}"
+    
+    def verify_credentials(self, username, password):
+        """Verify user credentials against database"""
+        # Simplified credential check
+        valid_users = {
+            "admin": "admin123",
+            "user1": "password123",
+            "[email protected]": "test123!"
+        }
+        return valid_users.get(username) == password
+    
+    def is_account_locked(self, username):
+        """Check if account is temporarily locked"""
+        if username not in self.failed_attempts:
+            return False
+        
+        attempts, last_attempt = self.failed_attempts[username]
+        if attempts >= 5 and time.time() - last_attempt < 900:  # 15 minutes
+            return True
+        return False
+    
+    def record_failed_attempt(self, username):
+        """Record a failed login attempt"""
+        current_time = time.time()
+        if username in self.failed_attempts:
+            attempts, _ = self.failed_attempts[username]
+            self.failed_attempts[username] = (attempts + 1, current_time)
+        else:
+            self.failed_attempts[username] = (1, current_time)
+    
+    def reset_failed_attempts(self, username):
+        """Reset failed attempts for user"""
+        if username in self.failed_attempts:
+            del self.failed_attempts[username]
+    
+    def create_session(self, username):
+        """Create user session with timeout"""
+        session_data = {
+            'username': username,
+            'created_at': datetime.now(),
+            'expires_at': datetime.now() + timedelta(seconds=self.session_timeout),
+            'session_id': hashlib.sha256(f"{username}{time.time()}".encode()).hexdigest()
+        }
+        return session_data
+    
+    def validate_session(self, session_id):
+        """Validate if session is still active"""
+        # This would typically check against a session store
+        # For demo purposes, assuming session validation logic
+        return True  # Simplified 

codebase/database.py

@@ -0,0 +1,3 @@
+def connect_to_db():
+    print("Connecting to database...")
+    return True

codebase/email.py

@@ -0,0 +1,133 @@
+import smtplib
+import ssl
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+from email.template import Template
+import logging
+
+class EmailService:
+    def __init__(self, smtp_server="smtp.gmail.com", smtp_port=587):
+        self.smtp_server = smtp_server
+        self.smtp_port = smtp_port
+        self.username = None
+        self.password = None
+        self.logger = logging.getLogger(__name__)
+        
+    def configure_smtp(self, username, password):
+        """Configure SMTP credentials"""
+        self.username = username
+        self.password = password
+        
+    def send_password_reset_email(self, recipient_email, reset_token):
+        """Send password reset email to user"""
+        try:
+            # Create message
+            message = MIMEMultipart("alternative")
+            message["Subject"] = "Password Reset Request"
+            message["From"] = self.username
+            message["To"] = recipient_email
+            
+            # Create HTML content
+            html_content = self._generate_reset_email_html(reset_token)
+            html_part = MIMEText(html_content, "html")
+            message.attach(html_part)
+            
+            # Send email
+            return self._send_email(message)
+            
+        except Exception as e:
+            self.logger.error(f"Error sending password reset email: {str(e)}")
+            return False, f"Failed to send email: {str(e)}"
+    
+    def send_notification_email(self, recipient_email, subject, content):
+        """Send general notification email"""
+        try:
+            message = MIMEMultipart()
+            message["Subject"] = subject
+            message["From"] = self.username
+            message["To"] = recipient_email
+            
+            text_part = MIMEText(content, "plain")
+            message.attach(text_part)
+            
+            return self._send_email(message)
+            
+        except Exception as e:
+            self.logger.error(f"Error sending notification email: {str(e)}")
+            return False, f"Failed to send email: {str(e)}"
+    
+    def _send_email(self, message):
+        """Internal method to send email via SMTP"""
+        try:
+            # Create secure connection
+            context = ssl.create_default_context()
+            
+            with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
+                server.starttls(context=context)
+                
+                # Check if credentials are configured
+                if not self.username or not self.password:
+                    raise Exception("SMTP credentials not configured")
+                    
+                server.login(self.username, self.password)
+                
+                # Send email
+                text = message.as_string()
+                server.sendmail(message["From"], message["To"], text)
+                
+            self.logger.info(f"Email sent successfully to {message['To']}")
+            return True, "Email sent successfully"
+            
+        except smtplib.SMTPAuthenticationError:
+            error_msg = "SMTP authentication failed - check credentials"
+            self.logger.error(error_msg)
+            return False, error_msg
+            
+        except smtplib.SMTPConnectError:
+            error_msg = "Failed to connect to SMTP server"
+            self.logger.error(error_msg)
+            return False, error_msg
+            
+        except Exception as e:
+            error_msg = f"Unexpected error sending email: {str(e)}"
+            self.logger.error(error_msg)
+            return False, error_msg
+    
+    def _generate_reset_email_html(self, reset_token):
+        """Generate HTML content for password reset email"""
+        html_template = """
+        <!DOCTYPE html>
+        <html>
+        <head>
+            <meta charset="utf-8">
+            <title>Password Reset</title>
+        </head>
+        <body>
+            <div style="max-width: 600px; margin: 0 auto; padding: 20px;">
+                <h2>Password Reset Request</h2>
+                <p>You have requested a password reset for your account.</p>
+                <p>Click the link below to reset your password:</p>
+                <a href="https://yourapp.com/reset-password?token={token}" 
+                   style="background-color: #4CAF50; color: white; padding: 10px 20px; 
+                          text-decoration: none; border-radius: 4px;">
+                    Reset Password
+                </a>
+                <p>If you did not request this reset, please ignore this email.</p>
+                <p>This link will expire in 24 hours.</p>
+            </div>
+        </body>
+        </html>
+        """
+        return html_template.format(token=reset_token)
+    
+    def test_connection(self):
+        """Test SMTP connection and configuration"""
+        try:
+            context = ssl.create_default_context()
+            with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
+                server.starttls(context=context)
+                if self.username and self.password:
+                    server.login(self.username, self.password)
+                return True, "SMTP connection successful"
+        except Exception as e:
+            return False, f"SMTP connection failed: {str(e)}" 

codebase/profile.py

@@ -0,0 +1,228 @@
+import json
+import sqlite3
+from datetime import datetime
+from typing import Dict, Optional, List
+
+class UserProfileManager:
+    """Manages user profile data and operations"""
+    
+    def __init__(self, db_path="users.db"):
+        self.db_path = db_path
+        self.init_database()
+    
+    def init_database(self):
+        """Initialize the user profile database"""
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                cursor.execute("""
+                    CREATE TABLE IF NOT EXISTS user_profiles (
+                        id INTEGER PRIMARY KEY AUTOINCREMENT,
+                        username VARCHAR(50) UNIQUE NOT NULL,
+                        email VARCHAR(100) UNIQUE NOT NULL,
+                        first_name VARCHAR(50),
+                        last_name VARCHAR(50),
+                        profile_data TEXT,
+                        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    )
+                """)
+                conn.commit()
+        except Exception as e:
+            print(f"Database initialization error: {e}")
+    
+    def create_profile(self, username: str, email: str, first_name: str = "", 
+                      last_name: str = "", additional_data: Dict = None) -> bool:
+        """Create a new user profile"""
+        try:
+            profile_data = json.dumps(additional_data or {})
+            
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                cursor.execute("""
+                    INSERT INTO user_profiles 
+                    (username, email, first_name, last_name, profile_data)
+                    VALUES (?, ?, ?, ?, ?)
+                """, (username, email, first_name, last_name, profile_data))
+                conn.commit()
+                return True
+                
+        except sqlite3.IntegrityError:
+            return False  # Username or email already exists
+        except Exception as e:
+            print(f"Profile creation error: {e}")
+            return False
+    
+    def get_profile(self, username: str) -> Optional[Dict]:
+        """Retrieve user profile by username"""
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                cursor.execute("""
+                    SELECT id, username, email, first_name, last_name, 
+                           profile_data, created_at, updated_at
+                    FROM user_profiles 
+                    WHERE username = ?
+                """, (username,))
+                
+                row = cursor.fetchone()
+                if row:
+                    return {
+                        'id': row[0],
+                        'username': row[1],
+                        'email': row[2],
+                        'first_name': row[3],
+                        'last_name': row[4],
+                        'profile_data': json.loads(row[5] or '{}'),
+                        'created_at': row[6],
+                        'updated_at': row[7]
+                    }
+                return None
+                
+        except Exception as e:
+            print(f"Profile retrieval error: {e}")
+            return None
+    
+    def update_profile(self, username: str, updates: Dict) -> bool:
+        """Update user profile information"""
+        try:
+            # Build dynamic update query
+            update_fields = []
+            update_values = []
+            
+            allowed_fields = ['email', 'first_name', 'last_name', 'profile_data']
+            
+            for field, value in updates.items():
+                if field in allowed_fields:
+                    if field == 'profile_data':
+                        value = json.dumps(value)
+                    update_fields.append(f"{field} = ?")
+                    update_values.append(value)
+            
+            if not update_fields:
+                return False  # No valid fields to update
+            
+            # Add updated_at timestamp
+            update_fields.append("updated_at = ?")
+            update_values.append(datetime.now().isoformat())
+            
+            # Add username for WHERE clause
+            update_values.append(username)
+            
+            query = f"""
+                UPDATE user_profiles 
+                SET {', '.join(update_fields)}
+                WHERE username = ?
+            """
+            
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                cursor.execute(query, update_values)
+                conn.commit()
+                
+                return cursor.rowcount > 0  # True if row was updated
+                
+        except sqlite3.IntegrityError:
+            return False  # Constraint violation (e.g., duplicate email)
+        except Exception as e:
+            print(f"Profile update error: {e}")
+            return False
+    
+    def delete_profile(self, username: str) -> bool:
+        """Delete user profile"""
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                cursor.execute("DELETE FROM user_profiles WHERE username = ?", (username,))
+                conn.commit()
+                return cursor.rowcount > 0
+                
+        except Exception as e:
+            print(f"Profile deletion error: {e}")
+            return False
+    
+    def list_profiles(self, limit: int = 50, offset: int = 0) -> List[Dict]:
+        """List user profiles with pagination"""
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                cursor.execute("""
+                    SELECT id, username, email, first_name, last_name, created_at
+                    FROM user_profiles 
+                    ORDER BY created_at DESC
+                    LIMIT ? OFFSET ?
+                """, (limit, offset))
+                
+                profiles = []
+                for row in cursor.fetchall():
+                    profiles.append({
+                        'id': row[0],
+                        'username': row[1],
+                        'email': row[2],
+                        'first_name': row[3],
+                        'last_name': row[4],
+                        'created_at': row[5]
+                    })
+                
+                return profiles
+                
+        except Exception as e:
+            print(f"Profile listing error: {e}")
+            return []
+    
+    def search_profiles(self, search_term: str) -> List[Dict]:
+        """Search profiles by username, email, or name"""
+        try:
+            search_pattern = f"%{search_term}%"
+            
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                cursor.execute("""
+                    SELECT id, username, email, first_name, last_name, created_at
+                    FROM user_profiles 
+                    WHERE username LIKE ? OR email LIKE ? 
+                       OR first_name LIKE ? OR last_name LIKE ?
+                    ORDER BY username
+                """, (search_pattern, search_pattern, search_pattern, search_pattern))
+                
+                profiles = []
+                for row in cursor.fetchall():
+                    profiles.append({
+                        'id': row[0],
+                        'username': row[1],
+                        'email': row[2],
+                        'first_name': row[3],
+                        'last_name': row[4],
+                        'created_at': row[5]
+                    })
+                
+                return profiles
+                
+        except Exception as e:
+            print(f"Profile search error: {e}")
+            return []
+
+# API endpoints for profile management
+def handle_profile_update_api(request_data: Dict) -> Dict:
+    """Handle API request for profile updates"""
+    try:
+        username = request_data.get('username')
+        updates = request_data.get('updates', {})
+        
+        if not username:
+            return {'error': 'Username is required', 'status': 400}
+        
+        profile_manager = UserProfileManager()
+        
+        # Check if profile exists
+        if not profile_manager.get_profile(username):
+            return {'error': 'Profile not found', 'status': 404}
+        
+        # Attempt to update profile
+        if profile_manager.update_profile(username, updates):
+            return {'message': 'Profile updated successfully', 'status': 200}
+        else:
+            return {'error': 'Failed to update profile', 'status': 500}
+            
+    except Exception as e:
+        return {'error': f'Internal server error: {str(e)}', 'status': 500}