🔒 Secure Coding

# ✅ SECURE - Environment variables
import os
import boto3
from cryptography.fernet import Fernet
import logging

class SecureConnector:
    def __init__(self):
        # Load from environment variables
        self.db_host = os.getenv('DB_HOST')
        self.username = os.getenv('DB_USERNAME')
        self.password = self._get_secret('DB_PASSWORD')
        self.api_key = self._get_secret('API_KEY')
        
        if not all([self.db_host, self.username, 
                   self.password, self.api_key]):
            raise ValueError("Missing required env vars")
    
    def _get_secret(self, secret_name):
        """Retrieve secret from AWS Secrets Manager"""
        try:
            client = boto3.client('secretsmanager')
            response = client.get_secret_value(
                SecretId=secret_name)
            return response['SecretString']
        except Exception as e:
            logging.error(f"Failed to retrieve secret: "
                         f"{secret_name}")
            raise
    
    def connect(self):
        # Use connection pooling and SSL
        return {
            'host': self.db_host,
            'username': self.username,
            'password': self.password,
            'sslmode': 'require'
        }

# Secure API calling
def call_external_api():
    api_key = os.getenv('EXTERNAL_API_KEY')
    if not api_key:
        raise ValueError("API key not configured")
    
    headers = {
        'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }
    
    try:
        response = requests.get(
            'https://api.example.com/data', 
            headers=headers, timeout=30)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        logging.error("API call failed", exc_info=True)
        raise

# Dynamic key generation
def generate_encryption_key():
    return Fernet.generate_key()

# ✅ SECURE - Proper error handling and logging
import sqlite3
import logging
import hashlib
import secrets
from typing import Optional, Dict, Any

# Configure secure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - '
           '%(message)s',
    handlers=[
        logging.FileHandler('/var/log/app/security.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

class SecurityError(Exception):
    """Custom exception for security-related errors"""
    pass

class DatabaseManager:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._setup_database()
    
    def _setup_database(self):
        """Initialize database with proper security"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                conn.execute("PRAGMA foreign_keys = ON")
                conn.execute("PRAGMA secure_delete = ON")
        except sqlite3.Error as e:
            logger.error("Database initialization failed", 
                        exc_info=True)
            raise SecurityError("Database config error")

    def get_user_data(self, user_id: int) -> Optional[Dict]:
        """Retrieve user data with secure error handling"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                conn.row_factory = sqlite3.Row
                cursor = conn.cursor()
                
                # Parameterized query prevents SQL injection
                cursor.execute("SELECT * FROM users WHERE id = ?",
                             (user_id,))
                result = cursor.fetchone()
                
                if result:
                    return dict(result)
                else:
                    # Generic error message
                    logger.warning(f"User lookup failed for "
                                 f"ID: {user_id}")
                    return None
                    
        except sqlite3.Error as e:
            # Log technical details internally
            logger.error(f"Database query failed for "
                        f"user_id {user_id}", exc_info=True)
            # Return generic error to client
            raise SecurityError("Data retrieval failed")
        except Exception as e:
            logger.error("Unexpected error in get_user_data", 
                        exc_info=True)
            raise SecurityError("Internal server error")

class AuthenticationManager:
    def __init__(self, db_manager: DatabaseManager):
        self.db_manager = db_manager
        self.max_attempts = 5
        self.lockout_duration = 300  # 5 minutes

    def authenticate_user(self, username: str, 
                         password: str) -> bool:
        """Secure user authentication"""
        # Input validation
        if not username or not password:
            logger.warning("Authentication attempt with "
                          "empty credentials")
            raise SecurityError("Invalid credentials")
        
        if len(username) > 255 or len(password) > 255:
            logger.warning(f"Authentication attempt with "
                          f"oversized input from "
                          f"{username[:50]}...")
            raise SecurityError("Invalid credentials")
        
        try:
            # Check for account lockout
            if self._is_account_locked(username):
                logger.warning(f"Authentication attempt on "
                              f"locked account: {username}")
                raise SecurityError("Account temporarily locked")
            
            # Retrieve user data
            user_data = self._get_user_by_username(username)
            if not user_data:
                # Same error for non-existent users
                self._log_failed_attempt(username)
                raise SecurityError("Invalid credentials")
            
            # Verify password
            if self._verify_password(password, 
                                   user_data['password_hash']):
                logger.info(f"Successful authentication for "
                           f"user: {username}")
                self._clear_failed_attempts(username)
                return True
            else:
                self._log_failed_attempt(username)
                raise SecurityError("Invalid credentials")
                
        except SecurityError:
            # Re-raise security errors as-is
            raise
        except Exception as e:
            # Log unexpected errors
            logger.error("Unexpected authentication error", 
                        exc_info=True)
            raise SecurityError("Authentication service unavailable")
    
    def _get_user_by_username(self, username: str) -> Optional[Dict]:
        """Retrieve user by username using parameterized query"""
        try:
            with sqlite3.connect(self.db_manager.db_path) as conn:
                conn.row_factory = sqlite3.Row
                cursor = conn.cursor()
                cursor.execute(
                    "SELECT id, username, password_hash FROM users WHERE username = ?", 
                    (username,)
                )
                result = cursor.fetchone()
                return dict(result) if result else None
        except sqlite3.Error as e:
            logger.error("Database query failed", exc_info=True)
            return None
    
    def _verify_password(self, password: str, stored_hash: str) -> bool:
        """Secure password verification"""
        try:
            # Use secure comparison to prevent timing attacks
            return secrets.compare_digest(
                hashlib.pbkdf2_hmac('sha256', password.encode(), b'salt', 100000),
                bytes.fromhex(stored_hash)
            )
        except Exception as e:
            logger.error("Password verification error", exc_info=True)
            return False
    
    def _log_failed_attempt(self, username: str):
        """Log failed authentication attempt"""
        # Log without exposing password
        logger.warning(f"Failed authentication attempt for username: {username}")
    
    def _is_account_locked(self, username: str) -> bool:
        """Check if account is locked due to failed attempts"""
        # Implementation would check failed attempt count and timing
        return False
    
    def _clear_failed_attempts(self, username: str):
        """Clear failed attempt counter after successful login"""
        pass

# Usage example
def main():
    try:
        db_manager = DatabaseManager("/secure/path/app.db")
        auth_manager = AuthenticationManager(db_manager)
        
        result = auth_manager.authenticate_user("john_doe", "secure_password")
        if result:
            print("Authentication successful")
        
    except SecurityError as e:
        # Return generic error to client
        print(f"Error: {str(e)}")
    except Exception as e:
        # Log unexpected errors, return generic message
        logger.error("Unexpected application error", exc_info=True)
        print("Service temporarily unavailable")