自研AES-128(ECB模式)加密结果与标准库不符求排查
AES-128 ECB模式加密结果与标准库不一致问题排查
我正在实现AES-128的ECB模式复刻版本(仅适配128位密钥,忽略192/256位),当前代码能生成加密输出,但结果和Python的Crypto.Cipher库输出不一致。已确认密钥调度模块正常,问题定位在encrypt方法中,以下是完整代码:
AES类代码
class AES: def __init__(self, key: str) -> None: assert len(key) in [16, 24, 32], "Invalid key length. AES allows 16, 24, or 32 key lengths." self.key = key self.rounds = {4: 10, 6: 12, 8: 14}[len(key) // 4] self.expanded_key = self.key_schedule(key) print(f"Original Key: {self.key} | len={len(self.key)}") print(f"Expanded Key: {self.expanded_key}") @record_time def key_schedule(self, key: str) -> list: """Expand the key to be used in encryption.""" expanded_key = to_matrix(list(key.encode("utf-8"))) # 4x4 matrix for index in range(4, 4 * (self.rounds + 1)): word_block = expanded_key[index - 1] # 1 byte if index % 4 == 0: word_block = self.sub_bytes(self.rot_bytes(word_block)) # Rot & Sub word_block[0] ^= RCON[index // 4] # RCON elif self.rounds > 6 and self.rounds % 4 == 4: # 256 word_block = self.sub_bytes(word_block) word_block = [i ^ j for i, j in zip(word_block, expanded_key[index - 4])] # XOR expanded_key.append(word_block) return expanded_key # 44 word_blocks => 11 keys @record_time def encrypt(self, text: str) -> str: """Encrypt the given text using the key.""" plaintext = pkcs7_padding(text.encode("utf-8")) # encodes and adds padding word_block_matrix = to_matrix(plaintext) self.add_round(word_block_matrix, self.expanded_key[:4]) for index in range(1, self.rounds): word_block_matrix = [self.sub_bytes(wb) for wb in word_block_matrix] self.shift_rows(word_block_matrix) self.mix_columns(word_block_matrix) self.add_round(word_block_matrix, self.expanded_key[4 * index: 4 * (index + 1)]) # Last round (without mix_columns) word_block_matrix = [self.sub_bytes(wb) for wb in word_block_matrix] self.shift_rows(word_block_matrix) self.add_round(word_block_matrix, self.expanded_key[40:]) # encrypted_bytes = bytes(encrypted_blocks) # to bytes list encrypted_base64 = base64.b64encode(to_bytes(word_block_matrix)) # to base64 encrypted_text = encrypted_base64.decode('utf-8') return encrypted_text @record_time def decrypt(self, text: str) -> str: word_block_matrix = to_matrix(text.encode('utf-8')) # encodes to 4x4 matrix # print(word_block_matrix) # self.add_round(word_block_matrix, self.expanded_key[-4:]) def rot_bytes(self, word_block: list) -> list: """Rotate the bytes in the block to the left by one position.""" return word_block[1:] + word_block[:1] def sub_bytes(self, word_block: list, inverse=False) -> list: """Apply sub bytes transformation to the block.""" if inverse: return [SBOX_INV[wb] for wb in word_block] return [SBOX[wb] for wb in word_block] def shift_rows(self, word_block: list, inverse=False) -> None: """Shift the rows in the block according to AES specifications.""" if inverse: word_block[1] = [word_block[1][3], word_block[1][0], word_block[1][1], word_block[1][2]] # shifts -1 word_block[2] = [word_block[2][2], word_block[2][3], word_block[2][0], word_block[2][1]] # shifts -2 word_block[3] = [word_block[3][1], word_block[3][2], word_block[3][3], word_block[3][0]] # shifts -3 else: word_block[1] = [word_block[1][1], word_block[1][2], word_block[1][3], word_block[1][0]] # shifts 1 word_block[2] = [word_block[2][2], word_block[2][3], word_block[2][0], word_block[2][1]] # shifts 2 word_block[3] = [word_block[3][3], word_block[3][0], word_block[3][1], word_block[3][2]] # shifts 3 def mix_columns(self, word_block_matrix: list, inverse=False) -> None: """Apply mix columns transformation to the block.""" for i in range(4): wb0 = word_block_matrix[i][0] wb1 = word_block_matrix[i][1] wb2 = word_block_matrix[i][2] wb3 = word_block_matrix[i][3] if inverse: # inverse galois variable b= differs than a non inverse galois word_block_matrix[i][0] = galois(wb0, 0xE) ^ galois(wb1, 0xB) ^ galois(wb2, 0xD) ^ galois(wb3, 0x9) word_block_matrix[i][1] = galois(wb1, 0xE) ^ galois(wb2, 0xB) ^ galois(wb3, 0xD) ^ galois(wb0, 0x9) word_block_matrix[i][2] = galois(wb2, 0xE) ^ galois(wb3, 0xB) ^ galois(wb0, 0xD) ^ galois(wb1, 0x9) word_block_matrix[i][3] = galois(wb3, 0xE) ^ galois(wb0, 0xB) ^ galois(wb1, 0xD) ^ galois(wb2, 0x9) else: word_block_matrix[i][0] = galois(wb0, 2) ^ galois(wb1, 3) ^ wb2 ^ wb3 word_block_matrix[i][1] = galois(wb1, 2) ^ galois(wb2, 3) ^ wb3 ^ wb0 word_block_matrix[i][2] = galois(wb2, 2) ^ galois(wb3, 3) ^ wb0 ^ wb1 word_block_matrix[i][3] = galois(wb3, 2) ^ galois(wb0, 3) ^ wb1 ^ wb2 def add_round(self, word_block_matrix: list, key: list) -> None: """Apply key schedule transformation to the block using the key.""" for i in range(4): for j in range(4): word_block_matrix[i][j] ^= key[i][j]
工具函数代码
## UTILS ## def to_matrix(block: list) -> list: """Convert a 16-byte block to a 4x4 matrix.""" return [list(block[index:index+4]) for index in range(0, len(block), 4)] def to_bytes(block: list) -> bytes: """Convert a 4x4 matrix to a 16-byte block.""" return bytes(sum(block, [])) def pkcs7_padding(data: bytes) -> bytes: """Apply PKCS#7 padding to the data to make it a multiple of 16.""" padding_length = 16 - len(data) % 16 padding = bytes([padding_length] * padding_length) return data + padding def pkcs7_padding_undo(): pass def galois(a, b): p = 0 for _ in range(8): if b & 1: p ^= a carry = a & 0x80 a <<= 1 if carry: a ^= 0x1B # xtime(0x80) in the Galois field (0x1B) b >>= 1 return p % 256 def record_time(func): def wrapper(*args, **kwargs): start_time = time.time() result = func(*args, **kwargs) end_time = time.time() execution_time = end_time - start_time print(f"{func.__name__}:\t{execution_time:.6f} seconds") return result return wrapper
核心问题排查与修正
状态矩阵存储方式错误(行/列颠倒)
AES的状态矩阵采用列优先存储,即输入的16字节按列填充到4x4矩阵中,但当前to_matrix函数是按行分组存储,导致后续所有变换逻辑完全错位。修正后的
to_matrix函数:def to_matrix(block: list) -> list: """Convert a 16-byte block to a 4x4 column-major matrix (AES standard).""" return [ [block[0], block[4], block[8], block[12]], [block[1], block[5], block[9], block[13]], [block[2], block[6], block[10], block[14]], [block[3], block[7], block[11], block[15]] ]对应的
to_bytes函数也需要调整为按列读取:def to_bytes(block: list) -> bytes: """Convert a 4x4 column-major matrix to a 16-byte block.""" return bytes([ block[0][0], block[1][0], block[2][0], block[3][0], block[0][1], block[1][1], block[2][1], block[3][1], block[0][2], block[1][2], block[2][2], block[3][2], block[0][3], block[1][3], block[2][3], block[3][3] ])ShiftRows实现逻辑错误
原代码中ShiftRows的行移位操作是基于行优先矩阵的,修正为列优先矩阵后,ShiftRows需要对每一行进行循环左移:def shift_rows(self, word_block: list, inverse=False) -> None: """Shift the rows in the block according to AES specifications (column-major matrix).""" if inverse: # 逆操作:循环右移对应位数 word_block[1] = [word_block[1][3]] + word_block[1][:3] # 右移1位 word_block[2] = word_block[2][2:] + word_block[2][:2] # 右移2位 word_block[3] = word_block[3][1:] + word_block[3][:1] # 右移3位 else: # 正操作:循环左移对应位数 word_block[1] = word_block[1][1:] + [word_block[1][0]] # 左移1位 word_block[2] = word_block[2][2:] + word_block[2][:2] # 左移2位 word_block[3] = word_block[3][3:] + word_block[3][:3] # 左移3位MixColumns实现逻辑错误
AES的MixColumns是对矩阵的列进行变换,原代码错误地对行进行处理,修正后:def mix_columns(self, word_block_matrix: list, inverse=False) -> None: """Apply mix columns transformation to the block (column-major matrix).""" for col in range(4): # 提取当前列的四个字节 c0 = word_block_matrix[0][col] c1 = word_block_matrix[1][col] c2 = word_block_matrix[2][col] c3 = word_block_matrix[3][col] if inverse: # 逆MixColumns变换 word_block_matrix[0][col] = galois(c0, 0xE) ^ galois(c1, 0xB) ^ galois(c2, 0xD) ^ galois(c3, 0x9) word_block_matrix[1][col] = galois(c0, 0x9) ^ galois(c1, 0xE) ^ galois(c2, 0xB) ^ galois(c3, 0xD) word_block_matrix[2][col] = galois(c0, 0xD) ^ galois(c1, 0x9) ^ galois(c2, 0xE) ^ galois(c3, 0xB) word_block_matrix[3][col] = galois(c0, 0xB) ^ galois(c1, 0xD) ^ galois(c2, 0x9) ^ galois(c3, 0xE) else: # 正MixColumns变换 word_block_matrix[0][col] = galois(c0, 2) ^ galois(c1, 3) ^ c2 ^ c3 word_block_matrix[1][col] = c0 ^ galois(c1, 2) ^ galois(c2, 3) ^ c3 word_block_matrix[2][col] = c0 ^ c1 ^ galois(c2, 2) ^ galois(c3, 3) word_block_matrix[3][col] = galois(c0, 3) ^ c1 ^ c2 ^ galois(c3, 2)SubBytes调用方式问题
原代码中word_block_matrix = [self.sub_bytes(wb) for wb in word_block_matrix]是对每一行进行SubBytes,修正为列优先矩阵后,保持逐行处理即可(SubBytes是逐字节操作,行或列遍历不影响结果):# 替换原SubBytes调用代码 for i in range(4): word_block_matrix[i] = self.sub_bytes(word_block_matrix[i])
内容的提问来源于stack exchange,提问作者user21729171
相关产品推荐
相关产品推荐

