Class SSHBruteforce

actions/ssh_connector.py:27–80

Class to handle the SSH brute force process.

b_parent = None

class SSHBruteforce:
    """
    Class to handle the SSH brute force process.
    """
    def __init__(self, shared_data):
        self.shared_data = shared_data
        self.ssh_connector = SSHConnector(shared_data)
        logger.info("SSHConnector initialized.")

    def bruteforce_ssh(self, ip, port):
        """
        Run the SSH brute force attack on the given IP and port.
        """
        logger.info(f"Running bruteforce_ssh on {ip}:{port}...")
        return self.ssh_connector.run_bruteforce(ip, port)
    
    def execute(self, ip, port, row, status_key):
        """
        Execute the brute force attack and update status.
        Optimization: Skip bruteforce if valid credentials already exist for this host.
        """
        logger.info(f"Executing SSHBruteforce on {ip}:{port}...")
        
        # Check if we already have valid credentials for this host
        existing_creds = CredentialChecker.check_existing_credentials(
            self.shared_data.sshfile, ip
        )
        if existing_creds:
            logger.info(f"SSH credentials already exist for {ip} - verifying instead of bruteforcing...")
            # Verify credentials still work
            if self._verify_credentials(ip, existing_creds):
                logger.success(f"Existing SSH credentials verified for {ip}: {len(existing_creds)} account(s)")
                return 'success'
            else:
                logger.warning(f"Existing credentials for {ip} no longer valid, will re-bruteforce")
        
        self.shared_data.ragnarorch_status = "SSHBruteforce"
        success, results = self.bruteforce_ssh(ip, port)
        if success and results:
            for mac_address, ip_addr, hostname, user, password, used_port in results:
                logger.success(
                    f"SSH credentials confirmed | MAC: {mac_address} | IP: {ip_addr} | Host: {hostname} | User: {user} | Password: {password} | Port: {used_port}"
                )
        else:
            logger.info(f"SSHBruteforce completed for {ip}:{port} with no valid credentials discovered")
        return 'success' if success else 'failed'
    
    def _verify_credentials(self, ip, credentials):
        """Verify that existing credentials still work (quick check)."""
        for user, password in credentials:
            if self.ssh_connector.ssh_connect(ip, user, password):
                logger.debug(f"Verified credential for {ip}: {user}")
                return True
        return False

class SSHConnector:
    """