Reverse Engineering Detection of malicious code in GitHub repositories

[sisu]

This Script analyzes the folders by scanning for obfuscation patterns,
Urls, Links, Sockets, WebSockets and other types of remote connections

I hope this helps you avoid those Reverse Shells Containers repositories

import re
import os
import importlib.util
import subprocess
from colorama import Fore, Style,Back, init
from concurrent.futures import ThreadPoolExecutor
import signal
import sys
import time


def manejar_interrupcion(sig, frame):
    """Manejador de señal para Ctrl+C."""
    Detener = input(Fore.GREEN + f"Quiere Detener El Analisis Del Script?" + Style.RESET_ALL)
    if Detener in ["s", "S", "y", "Y"]:
        print (Fore.RED + "Saliendo Del Analisis" + Style.RESET_ALL)
        exit  # Salir del programa
        exit(0)
        return
    if Detener in [ "n", "N" ]:
        print (Fore.GREEN + f'Continuando con el escaneo' + Style.RESET_ALL)
        time.sleep(2)
    else:
        print (Fore.MAGENTA + f"Opcion {Detener} no es valida" + Style.RESET_ALL)
        time.sleep(2)
def Detencion():
    # Asigna el manejador de señal para Ctrl+C
    signal.signal(signal.SIGINT, manejar_interrupcion)

Detencion()

# Inicializar colorama
init(autoreset=True)

# Función para leer el contenido de un archivo y devolver líneas
def read_file_lines(file_path):
    try:
        with open(file_path, 'r', encoding='utf-8') as file:
            return file.readlines()
    except (UnicodeDecodeError, FileNotFoundError) as e:
        print(Fore.RED + f"Error al leer el archivo {file_path}: {e}" + Style.RESET_ALL)
        return None

# Función para buscar patrones en el texto y devolver líneas donde se encuentran
def find_patterns_with_lines(lines, file_path):
    patterns = {
        'ip': r'\b(?:\d{1,3}\.){3}\d{1,3}\b',
        'mac': r'([0-9a-fA-F]{2}[:-]){5}([0-9a-fA-F]{2})',
        'url': r'https?://[^\s]+',
        'socks': r'socks5?://[^\s]+',
        'websocket': r'ws://[^\s]+|wss://[^\s]+',
        'ftp': r'ftp://[^\s]+',
        'smtp': r'smtp://[^\s]+',
        'telnet': r'telnet://[^\s]+',
        'asm': r'\b(?:mov|add|sub|jmp|call|ret|push|pop|xor|shl|shr|inc|dec|cmp|test|lea|nop)\b',
    }
    results = {key: [] for key in patterns.keys()}
  
    for line_number, line in enumerate(lines, start=1):
        for key, pattern in patterns.items():
            if re.search(pattern, line, re.IGNORECASE):
                matches = re.findall(pattern, line, re.IGNORECASE)
                results[key].append((matches, line_number, file_path))

    return results

# Función para analizar imports
def analyze_imports(script_lines):
    imports = re.findall(r'^\s*import\s+(\S+)|^\s*from\s+(\S+)\s+import', ''.join(script_lines), re.MULTILINE)
    return [imp[0] or imp[1] for imp in imports]

# Función para analizar módulos importados
def analyze_imported_modules(imports):
    found_patterns = {}
    for module in imports:
        try:
            spec = importlib.util.find_spec(module)
            if spec is not None and spec.origin is not None:
                module_lines = read_file_lines(spec.origin)
                if module_lines:
                    found_patterns[module] = find_patterns_with_lines(module_lines, spec.origin)
        except Exception as e:
            print(Fore.RED + f"Error al importar el módulo {module}: {e}" + Style.RESET_ALL)
    return found_patterns

# Función para buscar cadenas aleatorias
def find_obfuscated_strings_with_lines(lines, file_path):
    obfuscated_pattern = r'[^a-zA-Z0-9\s]{5,}'
    obfuscated_results = []
  
    for line_number, line in enumerate(lines, start=1):
        if re.search(obfuscated_pattern, line):
            matches = re.findall(obfuscated_pattern, line)
            obfuscated_results.append((matches, line_number, file_path))
  
    return obfuscated_results

# Función para hacer ping a una dirección IP
def ping_host(host):
    try:
        output = subprocess.run(['ping', '-c', '1', host], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        return host, output.returncode == 0
    except Exception as e:
        print(Fore.RED + f"Error al hacer ping a {host}: {e}" + Style.RESET_ALL)
        return host, False

# Función principal para analizar un script
def analyze_script(file_path):
    lines = read_file_lines(file_path)
    if lines is None:
        return

    patterns_found = find_patterns_with_lines(lines, file_path)
    imports = analyze_imports(lines)
    obfuscated_strings = find_obfuscated_strings_with_lines(lines, file_path)
    imported_patterns = analyze_imported_modules(imports)

    # Imprimir resultados ordenados
    print(Fore.YELLOW + f"\nResultados para el archivo: {file_path}" + Style.RESET_ALL)

    print(Fore.BLUE + "Patrones encontrados:" + Style.RESET_ALL)
    for key, values in patterns_found.items():
        if values:
            print(f"  {key.upper()}:")
            for matches, line_number, path in values:
                print(f"    - {matches} (Línea: {line_number}, Archivo: {path})")
        else:
            print(f"  {key.upper()}: Ninguno")
  
    print("Imports encontrados:")
    for imp in imports:
        print(f"  - {imp}")
  
    print("Cadenas ofuscadas encontradas:")
    for matches, line_number, path in obfuscated_strings:
        print(f"  - {matches} (Línea: {line_number}, Archivo: {path})")
  
    print("Patrones encontrados en módulos importados:")
    for module, patterns in imported_patterns.items():
        print(f"  Módulo: {module}")
        for key, values in patterns.items():
            if values:
                print(f"    {key.upper()}:")
                for matches, line_number, path in values:
                    print(f"      - {matches} (Línea: {line_number}, Archivo: {path})")
            else:
                print(f"    {key.upper()}: Ninguno")

    # Hacer ping a las direcciones encontradas
    print(Fore.BLUE + "\nResultados de ping a direcciones externas:" + Style.RESET_ALL)
    all_hosts = []
    for key, values in patterns_found.items():
        if key in ['ip', 'url', 'socks', 'websocket', 'ftp', 'smtp', 'telnet']:
            for matches, line_number, path in values:
                for match in matches:
                    if key == 'ip':
                        host = match
                    else:
                        # Extraer el host de la URL
                        host = re.search(r'//([^/:]+)', match)
                        if host:
                            host = host.group(1)
                        else:
                            continue
                    all_hosts.append((host, line_number, path))

    # Determinar la velocidad de ping según la cantidad de hosts
    num_hosts = len(all_hosts)
    if num_hosts > 10:
        max_workers = 4  # 4 veces más rápido
    elif num_hosts > 5:
        max_workers = 2  # 2 veces más rápido
    else:
        max_workers = 1  # Velocidad normal

    # Hacer ping a los hosts en paralelo
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(ping_host, host): (line_number, path) for host, line_number, path in all_hosts}
        for future in futures:
            host, is_active = future.result()
            line_number, path = futures[future]
            if is_active:
                print(Fore.GREEN + f"Host activo: {host} (Línea: {line_number}, Archivo: {path})" + Style.RESET_ALL)
            else:
                print(Fore.RED + f"Host inactivo: {host} (Línea: {line_number}, Archivo: {path})" + Style.RESET_ALL)

def main():
    # Cambia esta ruta por la del directorio que deseas analizar
    script_path = input('Path/del/directorio: ')
    if os.path.isdir(script_path):
        for root, dirs, files in os.walk(script_path):
            for file in files:
                # Analizar todos los archivos
                analyze_script(os.path.join(root, file))
    else:
        print(Fore.RED + f"La ruta proporcionada no es un directorio válido: {script_path}" + Style.RESET_ALL)

if __name__ == "__main__":
    main()