mysql_exploiter.py

#!/usr/bin/env python3
"""
MySQL Security Assessor
Author: RFS - Security Researcher

Professional MySQL security assessment tool for authorized testing.
Focuses on vulnerability detection and security analysis without exploitation.
"""

import mysql.connector
import logging
import json
from datetime import datetime
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
import hashlib
import socket

@dataclass
class AssessmentConfig:
    """Configuration for MySQL security assessment"""
    host: str
    port: int = 3306
    username: str = ""
    password: str = ""
    database: Optional[str] = None
    timeout: int = 30
    ssl_disabled: bool = False

class MySQLSecurityAssessor:
    """Professional MySQL security assessment tool"""
    
    def __init__(self, config: AssessmentConfig, verbose: bool = True):
        self.config = config
        self.verbose = verbose
        self.connection = None
        self.vulnerabilities = []
        self.security_findings = []
        self.assessment_report = {}
        
        # Setup logging
        logging.basicConfig(
            level=logging.INFO if verbose else logging.WARNING,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
        
    def log_info(self, message: str):
        """Log information message"""
        if self.verbose:
            self.logger.info(f"[ASSESS] {message}")
            
    def log_warning(self, message: str):
        """Log warning message"""
        self.logger.warning(f"[WARNING] {message}")
        
    def log_finding(self, severity: str, finding: str, description: str):
        """Log security finding"""
        self.security_findings.append({
            'severity': severity,
            'finding': finding,
            'description': description,
            'timestamp': datetime.now().isoformat()
        })
        self.logger.warning(f"[{severity}] {finding}: {description}")
        
    def connect(self) -> bool:
        """Establish connection to MySQL server"""
        try:
            self.connection = mysql.connector.connect(
                host=self.config.host,
                port=self.config.port,
                user=self.config.username,
                password=self.config.password,
                database=self.config.database,
                connection_timeout=self.config.timeout,
                autocommit=True
            )
            self.log_info(f"Successfully connected to {self.config.host}:{self.config.port}")
            return True
            
        except mysql.connector.Error as e:
            self.log_warning(f"Connection failed: {e}")
            return False
            
    def disconnect(self):
        """Close MySQL connection"""
        if self.connection and self.connection.is_connected():
            self.connection.close()
            self.log_info("Database connection closed")
            
    def assess_server_version(self) -> Dict[str, Any]:
        """Assess MySQL server version for known vulnerabilities"""
        try:
            cursor = self.connection.cursor()
            cursor.execute("SELECT VERSION()")
            version = cursor.fetchone()[0]
            cursor.close()
            
            self.log_info(f"MySQL Version: {version}")
            
            # Check for known vulnerable versions
            version_info = {
                'version': version,
                'vulnerabilities': []
            }
            
            # Example version checks (add more as needed)
            if '5.7' in version and version < '5.7.44':
                self.log_finding(
                    'HIGH',
                    'Outdated MySQL Version',
                    f'MySQL {version} may contain known security vulnerabilities'
                )
                version_info['vulnerabilities'].append('CVE-2023-XXXX')
                
            return version_info
            
        except Exception as e:
            self.log_warning(f"Version assessment failed: {e}")
            return {}
            
    def assess_user_accounts(self) -> List[Dict[str, Any]]:
        """Assess user accounts for security issues"""
        try:
            cursor = self.connection.cursor()
            
            # Get user information
            cursor.execute("""
                SELECT User, Host, authentication_string, 
                       plugin, password_expired, account_locked
                FROM mysql.user
            """)
            
            users = []
            for row in cursor.fetchall():
                user_info = {
                    'username': row[0],
                    'host': row[1],
                    'has_password': bool(row[2]),
                    'plugin': row[3],
                    'password_expired': row[4],
                    'account_locked': row[5]
                }
                users.append(user_info)
                
                # Security checks
                if not user_info['has_password'] and user_info['username'] != '':
                    self.log_finding(
                        'HIGH',
                        'Empty Password',
                        f"User '{user_info['username']}' has empty password"
                    )
                    
                if user_info['host'] == '%':
                    self.log_finding(
                        'MEDIUM',
                        'Wildcard Host Access',
                        f"User '{user_info['username']}' can connect from any host"
                    )
                    
            cursor.close()
            return users
            
        except Exception as e:
            self.log_warning(f"User account assessment failed: {e}")
            return []
            
    def assess_privileges(self) -> Dict[str, Any]:
        """Assess current user privileges"""
        try:
            cursor = self.connection.cursor()
            
            # Check current user privileges
            cursor.execute("SHOW GRANTS FOR CURRENT_USER()")
            grants = [row[0] for row in cursor.fetchall()]
            
            privilege_info = {
                'current_user': self.config.username,
                'grants': grants,
                'dangerous_privileges': []
            }
            
            # Check for dangerous privileges
            dangerous_privs = ['FILE', 'SUPER', 'PROCESS', 'SHUTDOWN', 'CREATE USER']
            
            for grant in grants:
                for priv in dangerous_privs:
                    if priv in grant.upper():
                        privilege_info['dangerous_privileges'].append(priv)
                        self.log_finding(
                            'HIGH',
                            'Excessive Privileges',
                            f"User has {priv} privilege: {grant}"
                        )
                        
            cursor.close()
            return privilege_info
            
        except Exception as e:
            self.log_warning(f"Privilege assessment failed: {e}")
            return {}
            
    def assess_configuration(self) -> Dict[str, Any]:
        """Assess MySQL configuration for security issues"""
        try:
            cursor = self.connection.cursor()
            
            # Security-relevant configuration variables
            security_vars = [
                'log_bin', 'general_log', 'slow_query_log',
                'local_infile', 'secure_file_priv', 'sql_mode',
                'validate_password%'
            ]
            
            config_info = {}
            
            for var in security_vars:
                cursor.execute(f"SHOW VARIABLES LIKE '{var}'")
                results = cursor.fetchall()
                
                for variable, value in results:
                    config_info[variable] = value
                    
                    # Security checks
                    if variable == 'local_infile' and value == 'ON':
                        self.log_finding(
                            'MEDIUM',
                            'Local Infile Enabled',
                            'local_infile allows reading local files from client'
                        )
                        
                    if variable == 'secure_file_priv' and value == '':
                        self.log_finding(
                            'HIGH',
                            'Unrestricted File Operations',
                            'secure_file_priv is empty, allowing file operations anywhere'
                        )
                        
            cursor.close()
            return config_info
            
        except Exception as e:
            self.log_warning(f"Configuration assessment failed: {e}")
            return {}
            
    def assess_databases(self) -> List[Dict[str, Any]]:
        """Assess available databases"""
        try:
            cursor = self.connection.cursor()
            cursor.execute("SHOW DATABASES")
            
            databases = []
            for (database,) in cursor.fetchall():
                db_info = {
                    'name': database,
                    'accessible': True
                }
                
                # Check for default/test databases
                if database in ['test', 'information_schema', 'performance_schema']:
                    if database == 'test':
                        self.log_finding(
                            'LOW',
                            'Test Database Present',
                            'Default test database should be removed in production'
                        )
                        
                databases.append(db_info)
                
            cursor.close()
            return databases
            
        except Exception as e:
            self.log_warning(f"Database assessment failed: {e}")
            return []
            
    def assess_ssl_configuration(self) -> Dict[str, Any]:
        """Assess SSL/TLS configuration"""
        try:
            cursor = self.connection.cursor()
            
            # Check SSL status
            cursor.execute("SHOW STATUS LIKE 'Ssl%'")
            ssl_status = {row[0]: row[1] for row in cursor.fetchall()}
            
            cursor.execute("SHOW VARIABLES LIKE 'have_ssl'")
            have_ssl = cursor.fetchone()
            
            ssl_info = {
                'ssl_available': have_ssl[1] if have_ssl else 'NO',
                'ssl_status': ssl_status
            }
            
            # Security assessment
            if ssl_info['ssl_available'] != 'YES':
                self.log_finding(
                    'HIGH',
                    'SSL Not Available',
                    'MySQL server does not support SSL/TLS encryption'
                )
            elif not ssl_status.get('Ssl_cipher'):
                self.log_finding(
                    'MEDIUM',
                    'SSL Not Used',
                    'Current connection is not using SSL encryption'
                )
                
            cursor.close()
            return ssl_info
            
        except Exception as e:
            self.log_warning(f"SSL assessment failed: {e}")
            return {}
            
    def check_information_disclosure(self) -> Dict[str, Any]:
        """Check for information disclosure vulnerabilities"""
        try:
            cursor = self.connection.cursor()
            
            disclosure_info = {
                'exposed_information': []
            }
            
            # Check if we can access sensitive system information
            sensitive_queries = [
                ("User list", "SELECT User, Host FROM mysql.user LIMIT 5"),
                ("Process list", "SHOW PROCESSLIST"),
                ("Variables", "SHOW VARIABLES LIKE 'version%'")
            ]
            
            for check_name, query in sensitive_queries:
                try:
                    cursor.execute(query)
                    results = cursor.fetchall()
                    
                    if results:
                        disclosure_info['exposed_information'].append({
                            'type': check_name,
                            'accessible': True,
                            'sample_data': str(results[0]) if results else None
                        })
                        
                        if check_name == "User list":
                            self.log_finding(
                                'MEDIUM',
                                'User Information Disclosure',
                                'Current user can access mysql.user table'
                            )
                            
                except mysql.connector.Error:
                    # Access denied is actually good for security
                    disclosure_info['exposed_information'].append({
                        'type': check_name,
                        'accessible': False
                    })
                    
            cursor.close()
            return disclosure_info
            
        except Exception as e:
            self.log_warning(f"Information disclosure check failed: {e}")
            return {}
            
    def run_full_assessment(self) -> Dict[str, Any]:
        """Run comprehensive security assessment"""
        self.log_info("Starting comprehensive MySQL security assessment")
        
        if not self.connect():
            return {'error': 'Failed to connect to MySQL server'}
            
        try:
            # Run all assessment modules
            self.assessment_report = {
                'target': {
                    'host': self.config.host,
                    'port': self.config.port,
                    'username': self.config.username
                },
                'timestamp': datetime.now().isoformat(),
                'version_info': self.assess_server_version(),
                'user_accounts': self.assess_user_accounts(),
                'privileges': self.assess_privileges(),
                'configuration': self.assess_configuration(),
                'databases': self.assess_databases(),
                'ssl_configuration': self.assess_ssl_configuration(),
                'information_disclosure': self.check_information_disclosure(),
                'security_findings': self.security_findings,
                'summary': self._generate_summary()
            }
            
            self.log_info("Security assessment completed")
            return self.assessment_report
            
        finally:
            self.disconnect()
            
    def _generate_summary(self) -> Dict[str, Any]:
        """Generate assessment summary"""
        severity_counts = {'HIGH': 0, 'MEDIUM': 0, 'LOW': 0}
        
        for finding in self.security_findings:
            severity = finding['severity']
            if severity in severity_counts:
                severity_counts[severity] += 1
                
        risk_score = (severity_counts['HIGH'] * 3 + 
                     severity_counts['MEDIUM'] * 2 + 
                     severity_counts['LOW'] * 1)
                     
        risk_level = 'LOW'
        if risk_score >= 10:
            risk_level = 'CRITICAL'
        elif risk_score >= 7:
            risk_level = 'HIGH'
        elif risk_score >= 4:
            risk_level = 'MEDIUM'
            
        return {
            'total_findings': len(self.security_findings),
            'severity_breakdown': severity_counts,
            'risk_score': risk_score,
            'risk_level': risk_level,
            'recommendations': self._get_recommendations()
        }
        
    def _get_recommendations(self) -> List[str]:
        """Generate security recommendations"""
        recommendations = []
        
        for finding in self.security_findings:
            if 'Empty Password' in finding['finding']:
                recommendations.append("Implement strong password policies for all user accounts")
            elif 'SSL' in finding['finding']:
                recommendations.append("Enable and enforce SSL/TLS encryption for all connections")
            elif 'Excessive Privileges' in finding['finding']:
                recommendations.append("Implement principle of least privilege for user accounts")
            elif 'local_infile' in finding['description']:
                recommendations.append("Disable local_infile to prevent local file access")
                
        # Remove duplicates while preserving order
        return list(dict.fromkeys(recommendations))
        
    def save_report(self, filename: str = None) -> str:
        """Save assessment report to file"""
        if not filename:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"mysql_security_assessment_{self.config.host}_{timestamp}.json"
            
        with open(filename, 'w') as f:
            json.dump(self.assessment_report, f, indent=2)
            
        self.log_info(f"Assessment report saved to {filename}")
        return filename


def main():
    """Example usage of MySQL Security Assessor"""
    import argparse
    
    parser = argparse.ArgumentParser(description="MySQL Security Assessment Tool")
    parser.add_argument('-H', '--host', required=True, help='MySQL host')
    parser.add_argument('-P', '--port', type=int, default=3306, help='MySQL port')
    parser.add_argument('-u', '--username', required=True, help='MySQL username')
    parser.add_argument('-p', '--password', required=True, help='MySQL password')
    parser.add_argument('-d', '--database', help='MySQL database')
    parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output')
    parser.add_argument('-o', '--output', help='Output file for report')
    
    args = parser.parse_args()
    
    # Create assessment configuration
    config = AssessmentConfig(
        host=args.host,
        port=args.port,
        username=args.username,
        password=args.password,
        database=args.database
    )
    
    # Run assessment
    assessor = MySQLSecurityAssessor(config, verbose=args.verbose)
    report = assessor.run_full_assessment()
    
    # Save report
    if args.output:
        assessor.save_report(args.output)
    else:
        assessor.save_report()
    
    # Print summary
    if 'summary' in report:
        summary = report['summary']
        print(f"\n=== ASSESSMENT SUMMARY ===")
        print(f"Risk Level: {summary['risk_level']}")
        print(f"Total Findings: {summary['total_findings']}")
        print(f"High: {summary['severity_breakdown']['HIGH']}, "
              f"Medium: {summary['severity_breakdown']['MEDIUM']}, "
              f"Low: {summary['severity_breakdown']['LOW']}")


if __name__ == "__main__":
    main()