eth_key_seeker.py

import os
import re
import time
import sqlite3
import hashlib
from datetime import datetime
from github import Github, RateLimitExceededException
from dotenv import load_dotenv
from web3 import Web3
from config import DB_NAME, DATA_DIR

# Load environment variables
load_dotenv()

# Get GitHub token
github_token = os.getenv('GITHUB_TOKEN')
if not github_token:
    raise ValueError("GitHub token not found in .env file")

# Search settings
DELAY_BETWEEN_REQUESTS = 2  # Delay between requests (seconds)
MAX_RETRIES = 3  # Maximum retry attempts

def log_message(message):
    """Display log message with timestamp"""
    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f"[{timestamp}] {message}", flush=True)

def setup_database():
    """Set up database and tables"""
    try:
        # Create data directory
        os.makedirs(DATA_DIR, exist_ok=True)
        log_message(f"Verified data directory: {DATA_DIR}")
        
        conn = sqlite3.connect(DB_NAME)
        cursor = conn.cursor()
        
        # Create search logs table (with file hash)
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS search_logs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            repository TEXT,
            file_path TEXT,
            file_hash TEXT,
            search_date TIMESTAMP,
            status TEXT,
            UNIQUE(repository, file_path, file_hash)
        )
        ''')
        log_message("Created search_logs table")
        
        # Create found keys table
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS found_keys (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            private_key TEXT UNIQUE,
            repository TEXT,
            file_path TEXT,
            found_date TIMESTAMP,
            eth_address TEXT
        )
        ''')
        log_message("Created found_keys table")
        
        conn.commit()
        conn.close()
        log_message("Database setup completed")
    except Exception as e:
        log_message(f"Database setup error: {str(e)}")
        raise

def calculate_file_hash(content):
    """Calculate hash of file content"""
    return hashlib.sha256(content.encode('utf-8')).hexdigest()

def is_file_already_searched(repo_name, file_path, file_hash):
    """Check if file has already been searched"""
    try:
        conn = sqlite3.connect(DB_NAME)
        cursor = conn.cursor()
        cursor.execute(
            'SELECT id FROM search_logs WHERE repository = ? AND file_path = ? AND file_hash = ?',
            (repo_name, file_path, file_hash)
        )
        result = cursor.fetchone() is not None
        conn.close()
        return result
    except Exception as e:
        log_message(f"Search check error: {str(e)}")
        return False

def is_valid_private_key(key):
    """Check if Ethereum private key is valid"""
    try:
        # Remove 0x prefix and convert to hex string
        key = key.replace('0x', '')
        if len(key) != 64:  # Must be 32 bytes (64 characters)
            return False
            
        # Check if valid hex
        int(key, 16)
        
        # Try to generate address using Web3.py
        w3 = Web3()
        acct = w3.eth.account.from_key(key)
        return True
    except Exception:
        return False

def get_eth_address(private_key):
    """Get Ethereum address from private key"""
    try:
        w3 = Web3()
        acct = w3.eth.account.from_key(private_key)
        return acct.address
    except Exception:
        return None

def log_search(repo_name, file_path, file_hash, status):
    """Log search to database"""
    try:
        conn = sqlite3.connect(DB_NAME)
        cursor = conn.cursor()
        cursor.execute(
            'INSERT INTO search_logs (repository, file_path, file_hash, search_date, status) VALUES (?, ?, ?, ?, ?)',
            (repo_name, file_path, file_hash, datetime.now(), status)
        )
        conn.commit()
        log_message(f"Logged search: {repo_name}/{file_path} ({status})")
    except sqlite3.IntegrityError:
        # Skip if already searched
        pass
    except Exception as e:
        log_message(f"Log recording error: {str(e)}")
    finally:
        conn.close()

def save_found_key(private_key, repo_name, file_path):
    """Save discovered key to database"""
    try:
        eth_address = get_eth_address(private_key)
        conn = sqlite3.connect(DB_NAME)
        cursor = conn.cursor()
        cursor.execute(
            'INSERT INTO found_keys (private_key, repository, file_path, found_date, eth_address) VALUES (?, ?, ?, ?, ?)',
            (private_key, repo_name, file_path, datetime.now(), eth_address)
        )
        conn.commit()
        log_message(f"Found new key: {eth_address} in {repo_name}/{file_path}")
    except sqlite3.IntegrityError:
        log_message(f"Found duplicate key: {private_key}")
    except Exception as e:
        log_message(f"Key saving error: {str(e)}")
    finally:
        conn.close()

def search_private_keys(content):
    """Search for private keys in content"""
    # Private key pattern (64 hex characters)
    pattern = r'(?:0x)?[a-fA-F0-9]{64}'
    matches = re.finditer(pattern, content)
    
    valid_keys = []
    for match in matches:
        key = match.group()
        if is_valid_private_key(key):
            valid_keys.append(key)
    
    return valid_keys

def get_search_stats():
    """Get search statistics"""
    try:
        conn = sqlite3.connect(DB_NAME)
        cursor = conn.cursor()
        
        # Get total searched files
        cursor.execute('SELECT COUNT(*) FROM search_logs')
        total_files = cursor.fetchone()[0]
        
        # Get total found keys
        cursor.execute('SELECT COUNT(*) FROM found_keys')
        total_keys = cursor.fetchone()[0]
        
        conn.close()
        return total_files, total_keys
    except Exception as e:
        log_message(f"Stats retrieval error: {str(e)}")
        return 0, 0

def check_rate_limit(g):
    """Check GitHub API rate limit"""
    try:
        rate_limit = g.get_rate_limit()
        remaining = rate_limit.search.remaining
        reset_time = rate_limit.search.reset
        
        if remaining < 5:
            wait_time = (reset_time - datetime.utcnow()).total_seconds()
            if wait_time > 0:
                log_message(f"Approaching API limit. Waiting {wait_time:.0f} seconds...")
                time.sleep(wait_time + 1)
        
        return remaining
    except Exception as e:
        log_message(f"Rate limit check error: {str(e)}")
        return 0

def search_github():
    """Search GitHub code for private keys"""
    try:
        g = Github(github_token)
        
        # Search queries (prioritize hardhat and web3 related repositories)
        queries = [
            'hardhat filename:.env size:>64 sort:updated-desc',  # Prioritize .env files
            'hardhat (extension:js OR extension:ts OR extension:py) size:>64 sort:updated-desc',  # Other files
            'web3 filename:.env size:>64 sort:updated-desc',
            'web3 (extension:js OR extension:ts OR extension:py) size:>64 sort:updated-desc',
            'filename:.env size:>64 sort:updated-desc',
            '(extension:js OR extension:ts OR extension:py) size:>64 sort:updated-desc'
        ]
        
        for query in queries:
            log_message(f"\nSearch query: {query}")
            repositories = g.search_code(query)
            total_count = min(repositories.totalCount, 1000)  # GitHub API limit
            processed_count = 0
            
            for repo_file in repositories:
                processed_count += 1
                
                try:
                    log_message(f"[{processed_count}/{total_count}] Checking: {repo_file.repository.full_name}/{repo_file.path}")
                    
                    # Check rate limit
                    remaining = check_rate_limit(g)
                    log_message(f"Remaining API calls: {remaining}")
                    
                    # Get file content
                    for attempt in range(MAX_RETRIES):
                        try:
                            content = repo_file.decoded_content.decode('utf-8')
                            break
                        except RateLimitExceededException:
                            if attempt < MAX_RETRIES - 1:
                                log_message("Hit API limit. Waiting 60 seconds...")
                                time.sleep(60)
                            else:
                                raise
                        except Exception as e:
                            if attempt < MAX_RETRIES - 1:
                                log_message(f"Error occurred. Retrying... ({attempt + 1}/{MAX_RETRIES})")
                                time.sleep(DELAY_BETWEEN_REQUESTS)
                            else:
                                raise
                    
                    file_hash = calculate_file_hash(content)
                    
                    # Check if already searched
                    if is_file_already_searched(repo_file.repository.full_name, repo_file.path, file_hash):
                        log_message(f"Skip: Already searched - {repo_file.path}")
                        continue
                    
                    # Search for private keys
                    found_keys = search_private_keys(content)
                    
                    if found_keys:
                        log_message(f"Found {len(found_keys)} potential private keys: {repo_file.path}")
                        for key in found_keys:
                            save_found_key(key, repo_file.repository.full_name, repo_file.path)
                        log_search(repo_file.repository.full_name, repo_file.path, file_hash, "found_keys")
                    else:
                        log_search(repo_file.repository.full_name, repo_file.path, file_hash, "no_keys")
                    
                    # Add delay for unsearched files only
                    time.sleep(DELAY_BETWEEN_REQUESTS)
                        
                except Exception as e:
                    log_message(f"Error processing {repo_file.path}: {str(e)}")
                    log_search(repo_file.repository.full_name, repo_file.path, "", f"error: {str(e)}")
                    continue
                    
    except Exception as e:
        log_message(f"GitHub search error: {str(e)}")

def main():
    """Main execution function"""
    try:
        log_message("Setting up database...")
        setup_database()
        
        log_message("\nStarting GitHub search...")
        log_message("Note: 2 second delay for unsearched files only")
        search_github()
        
        # Display statistics
        total_files, total_keys = get_search_stats()
        log_message(f"\nSearch completed!")
        log_message(f"Total files searched: {total_files}")
        log_message(f"Total keys found: {total_keys}")
    except Exception as e:
        log_message(f"Fatal error occurred: {str(e)}")
        raise

if __name__ == "__main__":
    main()