StandalonePasswordChange/python-flask/smeserver-password-app/smeserver_utils.py

#!/usr/bin/env python3
"""
Corrected SME Server Utilities Module - Python 3.6.8 Compatible

This module provides utilities for interfacing with SME Server's
configuration database using the correct passwordstrength structure:
passwordstrength=configuration
    Admin=strong
    Ibays=strong  
    Users=strong

Compatible with Python 3.6.8
"""

import subprocess
import os
import logging
import re

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SMEServerError(Exception):
    """Custom exception for SME Server related errors"""
    pass

class SMEConfigDB:
    """Interface for SME Server configuration database"""
    
    def __init__(self):
        self.db_command = '/sbin/e-smith/db'
        
    def _run_db_command(self, args):
        """Run a database command and return success status and output"""
        try:
            cmd = [self.db_command] + args
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=30
            )
            return result.returncode == 0, result.stdout.strip()
        except subprocess.TimeoutExpired:
            logger.error("Database command timed out: {}".format(args))
            return False, "Command timed out"
        except FileNotFoundError:
            logger.error("Database command not found: {}".format(self.db_command))
            return False, "Database command not available"
        except Exception as e:
            logger.error("Error running database command: {}".format(e))
            return False, str(e)
    
    def get_account_info(self, username):
        """Get account information from the accounts database"""
        success, output = self._run_db_command(['accounts', 'show', username])
        
        if not success:
            return None
            
        # Parse the output into a dictionary
        info = {'username': username}
        for line in output.split('\n'):
            if '=' in line:
                key, value = line.split('=', 1)
                info[key] = value
        
        return info
    
    def account_exists(self, username):
        """Check if an account exists in the accounts database"""
        success, _ = self._run_db_command(['accounts', 'show', username])
        return success
    
    def get_account_type(self, username):
        """Get the type of an account (user, admin, etc.)"""
        info = self.get_account_info(username)
        return info.get('type') if info else None
    
    def is_user_account(self, username):
        """Check if the account is a user account (not system account)"""
        account_type = self.get_account_type(username)
        return account_type == 'user'
    
    def is_admin_account(self, username):
        """Check if the account is an admin account"""
        account_type = self.get_account_type(username)
        return account_type == 'admin'
    
    def get_password_strength_setting(self, account_type='Users'):
        """Get the password strength setting for specific account type
        
        Args:
            account_type: 'Users', 'Admin', or 'Ibays'
        """
        success, output = self._run_db_command(['configuration', 'getprop', 'passwordstrength', account_type])
        
        if success and output:
            strength = output.strip().lower()
            if strength in ['none', 'normal', 'strong']:
                return strength
        
        # Default to 'normal' if not set or invalid
        return 'normal'
    
    def set_password_strength_setting(self, strength, account_type='Users'):
        """Set the password strength setting for specific account type
        
        Args:
            strength: 'none', 'normal', or 'strong'
            account_type: 'Users', 'Admin', or 'Ibays'
        """
        if strength.lower() not in ['none', 'normal', 'strong']:
            return False, "Invalid strength level. Must be 'none', 'normal', or 'strong'"
        
        success, output = self._run_db_command(['configuration', 'setprop', 'passwordstrength', account_type, strength.lower()])
        return success, output
    
    def get_all_password_strength_settings(self):
        """Get all password strength settings"""
        success, output = self._run_db_command(['configuration', 'show', 'passwordstrength'])
        
        settings = {
            'Users': 'normal',
            'Admin': 'normal', 
            'Ibays': 'normal'
        }
        
        if success and output:
            for line in output.split('\n'):
                if '=' in line:
                    key, value = line.split('=', 1)
                    key = key.strip()
                    value = value.strip().lower()
                    if key in ['Users', 'Admin', 'Ibays'] and value in ['none', 'normal', 'strong']:
                        settings[key] = value
        
        return settings

class BasicPasswordValidator:
    """Basic password validation for fallback when external library not available"""
    
    def __init__(self):
        # Common weak passwords for basic validation
        self.common_passwords = {
            'password', 'password123', '123456', '123456789', 'qwerty', 'abc123',
            'password1', 'admin', 'administrator', 'root', 'user', 'guest',
            'welcome', 'login', 'pass', 'secret', 'default', 'changeme',
            'letmein', 'monkey', 'dragon', 'master', 'shadow', 'superman'
        }
    
    def validate_password_strength(self, password, strength_level='normal'):
        """Basic password validation"""
        errors = []
        
        if strength_level == 'none':
            if len(password) < 1:
                errors.append("Password cannot be empty")
            return errors
        
        elif strength_level == 'normal':
            errors.extend(self._validate_normal_strength(password))
            
        elif strength_level == 'strong':
            errors.extend(self._validate_normal_strength(password))
            errors.extend(self._validate_strong_basic(password))
        
        return errors
    
    def _validate_normal_strength(self, password):
        """Validate normal strength requirements"""
        errors = []
        
        if len(password) < 12:
            errors.append("Password must be at least 12 characters long")
        
        if len(password) > 127:
            errors.append("Password must be no more than 127 characters long")
        
        has_upper = any(c.isupper() for c in password)
        has_lower = any(c.islower() for c in password)
        has_numeric = any(c.isdigit() for c in password)
        has_non_alpha = any(not c.isalnum() for c in password)
        
        missing_types = []
        if not has_upper:
            missing_types.append("uppercase letter")
        if not has_lower:
            missing_types.append("lowercase letter")
        if not has_numeric:
            missing_types.append("number")
        if not has_non_alpha:
            missing_types.append("special character")
        
        if missing_types:
            errors.append("Password must contain at least one: {}".format(", ".join(missing_types)))
        
        return errors
    
    def _validate_strong_basic(self, password):
        """Basic strong validation"""
        errors = []
        
        if password.lower() in self.common_passwords:
            errors.append("Password is too common and easily guessable")
        
        # Basic keyboard pattern check
        keyboard_patterns = ['qwerty', 'asdfgh', 'zxcvbn', '123456', '654321']
        password_lower = password.lower()
        for pattern in keyboard_patterns:
            if pattern in password_lower:
                errors.append("Password contains keyboard patterns that are easily guessable")
                break
        
        return errors

class SMEPasswordManager:
    """Handle password operations for SME Server with corrected DB structure"""
    
    def __init__(self):
        self.config_db = SMEConfigDB()
        self.basic_validator = BasicPasswordValidator()
        
        # Try to import external password validation library
        self.external_validator = None
        try:
            import zxcvbn
            self.external_validator = zxcvbn
            logger.info("Using zxcvbn library for password validation")
        except ImportError:
            logger.info("zxcvbn library not available, using basic validation")
    
    def validate_username(self, username):
        """Validate username format and existence"""
        if not username:
            return False, "Username cannot be empty"
        
        # Check username format (alphanumeric, underscore, hyphen)
        if not re.match(r'^[a-zA-Z0-9_-]+$', username):
            return False, "Username contains invalid characters"
        
        # Check if account exists
        if not self.config_db.account_exists(username):
            return False, "User account does not exist"
        
        # Check if it's a user account (not system account)
        if not self.config_db.is_user_account(username):
            return False, "Account is not a user account"
        
        return True, "Username is valid"
    
    def validate_password_strength(self, password, username=None):
        """Validate password using external library or fallback to basic validation"""
        # Get appropriate password strength setting
        strength_level = self.config_db.get_password_strength_setting('Users')
        
        if self.external_validator:
            return self._validate_with_zxcvbn(password, strength_level, username)
        else:
            return self.basic_validator.validate_password_strength(password, strength_level)
    
    def _validate_with_zxcvbn(self, password, strength_level, username=None):
        """Validate password using zxcvbn library"""
        errors = []
        
        if strength_level == 'none':
            if len(password) < 1:
                errors.append("Password cannot be empty")
            return errors
        
        # Basic length and complexity checks first
        if len(password) < 12:
            errors.append("Password must be at least 12 characters long")
        
        if len(password) > 127:
            errors.append("Password must be no more than 127 characters long")
        
        # Character type requirements for normal and strong
        if strength_level in ['normal', 'strong']:
            has_upper = any(c.isupper() for c in password)
            has_lower = any(c.islower() for c in password)
            has_numeric = any(c.isdigit() for c in password)
            has_non_alpha = any(not c.isalnum() for c in password)
            
            missing_types = []
            if not has_upper:
                missing_types.append("uppercase letter")
            if not has_lower:
                missing_types.append("lowercase letter")
            if not has_numeric:
                missing_types.append("number")
            if not has_non_alpha:
                missing_types.append("special character")
            
            if missing_types:
                errors.append("Password must contain at least one: {}".format(", ".join(missing_types)))
        
        # Use zxcvbn for advanced validation on strong passwords
        if strength_level == 'strong' and not errors:
            user_inputs = [username] if username else []
            result = self.external_validator.zxcvbn(password, user_inputs)
            
            # zxcvbn scores: 0-4 (0 = very weak, 4 = very strong)
            if result["score"] < 3:  # Require score of 3 or higher for strong
                feedback = result.get('feedback', {})
                warning = feedback.get('warning', '')
                suggestions = feedback.get('suggestions', [])
                
                if warning:
                    errors.append("Password weakness: {}".format(warning))
                
                for suggestion in suggestions:
                    errors.append("Suggestion: {}".format(suggestion))
                
                if not warning and not suggestions:
                    errors.append("Password is not strong enough (zxcvbn score: {}/4)".format(result['score']))
        
        return errors
    
    def get_password_strength_info(self, account_type='Users'):
        """Get current password strength setting and requirements"""
        strength_level = self.config_db.get_password_strength_setting(account_type)
        
        descriptions = {
            'none': 'No specific password requirements',
            'normal': 'Minimum 12 characters with uppercase, lowercase, number, and special character',
            'strong': 'Normal requirements plus advanced security validation using zxcvbn library' if self.external_validator else 'Normal requirements plus basic pattern protection'
        }
        
        return {
            'level': strength_level,
            'description': descriptions.get(strength_level, 'Unknown strength level'),
            'account_type': account_type,
            'using_zxcvbn': self.external_validator is not None
        }
    
    def verify_current_password(self, username, password):
        """Verify the current password for a user using system authentication"""
        try:
            # Use the 'su' command to verify password
            process = subprocess.Popen(
                ['su', username, '-c', 'true'],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )
            
            stdout, stderr = process.communicate(input=password + '\n', timeout=10)
            return process.returncode == 0
            
        except subprocess.TimeoutExpired:
            logger.error("Password verification timed out")
            return False
        except Exception as e:
            logger.error("Error verifying password: {}".format(e))
            return False
    
    def change_password(self, username, new_password):
        """Change user password and signal the update event"""
        try:
            # First, change the password using passwd command
            process = subprocess.Popen(
                ['passwd', username],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )
            
            # Send the new password twice (passwd asks for confirmation)
            input_data = "{}\n{}\n".format(new_password, new_password)
            stdout, stderr = process.communicate(input=input_data, timeout=30)
            
            if process.returncode != 0:
                logger.error("passwd command failed: {}".format(stderr))
                return False, "Failed to change password: {}".format(stderr)
            
            # Signal the password update event
            signal_result = subprocess.run(
                ['/sbin/e-smith/signal-event', 'password-update', username],
                capture_output=True,
                text=True,
                timeout=60
            )
            
            if signal_result.returncode != 0:
                logger.error("signal-event failed: {}".format(signal_result.stderr))
                return False, "Password changed but failed to update system: {}".format(signal_result.stderr)
            
            logger.info("Password successfully changed for user: {}".format(username))
            return True, "Password changed successfully"
            
        except subprocess.TimeoutExpired:
            logger.error("Password change operation timed out")
            return False, "Password change operation timed out"
        except FileNotFoundError as e:
            logger.error("Required command not found: {}".format(e))
            return False, "System command not available"
        except Exception as e:
            logger.error("Unexpected error changing password: {}".format(e))
            return False, "Unexpected error: {}".format(str(e))

class SMESystemInfo:
    """Get SME Server system information"""
    
    @staticmethod
    def get_version():
        """Get SME Server version"""
        try:
            version_files = [
                '/etc/e-smith-release',
                '/etc/sme-release',
                '/etc/redhat-release'
            ]
            
            for version_file in version_files:
                if os.path.exists(version_file):
                    with open(version_file, 'r') as f:
                        return f.read().strip()
            
            return "SME Server (version unknown)"
            
        except Exception:
            return "SME Server (version unknown)"
    
    @staticmethod
    def get_copyright_info():
        """Get copyright information"""
        return {
            'mitel': 'Copyright 1999-2006 Mitel Corporation',
            'rights': 'All rights reserved.',
            'koozali': 'Copyright (C) 2013 - 2021 Koozali Foundation Inc.'
        }