Sign Up
Log In
Log In
or
Sign Up
Places
All Projects
Status Monitor
Collapse sidebar
openSUSE:Leap:15.5:Update
python-pycryptodomex
CVE-2023-52323-const_time-decoding.patch
Overview
Repositories
Revisions
Requests
Users
Attributes
Meta
File CVE-2023-52323-const_time-decoding.patch of Package python-pycryptodomex
From 0deea1bfe1489e8c80d2053bbb06a1aa0b181ebd Mon Sep 17 00:00:00 2001 From: Helder Eijs <helderijs@gmail.com> Date: Wed, 27 Dec 2023 09:39:22 +0100 Subject: [PATCH] Use constant-time (faster) padding decoding also for OAEP --- lib/Crypto/Cipher/PKCS1_OAEP.py | 34 +- lib/Crypto/Cipher/PKCS1_v1_5.py | 28 +- lib/Crypto/Cipher/_pkcs1_oaep_decode.py | 41 +++ lib/Crypto/SelfTest/Cipher/test_pkcs1_15.py | 6 lib/Crypto/Util/py3compat.py | 10 setup.py | 4 src/pkcs1_decode.c | 354 ++++++++++++++++++++++++++++ src/test/Makefile | 5 8 files changed, 449 insertions(+), 33 deletions(-) create mode 100644 lib/Crypto/Cipher/_pkcs1_oaep_decode.py --- a/lib/Crypto/Cipher/PKCS1_OAEP.py +++ b/lib/Crypto/Cipher/PKCS1_OAEP.py @@ -23,11 +23,13 @@ from Crypto.Signature.pss import MGF1 import Crypto.Hash.SHA1 -from Crypto.Util.py3compat import bord, _copy_bytes +from Crypto.Util.py3compat import _copy_bytes import Crypto.Util.number -from Crypto.Util.number import ceil_div, bytes_to_long, long_to_bytes -from Crypto.Util.strxor import strxor +from Crypto.Util.number import ceil_div, bytes_to_long, long_to_bytes +from Crypto.Util.strxor import strxor from Crypto import Random +from ._pkcs1_oaep_decode import oaep_decode + class PKCS1OAEP_Cipher: """Cipher object for PKCS#1 v1.5 OAEP. @@ -68,7 +70,7 @@ class PKCS1OAEP_Cipher: if mgfunc: self._mgf = mgfunc else: - self._mgf = lambda x,y: MGF1(x,y,self._hashObj) + self._mgf = lambda x, y: MGF1(x, y, self._hashObj) self._label = _copy_bytes(None, None, label) self._randfunc = randfunc @@ -105,7 +107,7 @@ class PKCS1OAEP_Cipher: # See 7.1.1 in RFC3447 modBits = Crypto.Util.number.size(self._key.n) - k = ceil_div(modBits, 8) # Convert from bits to bytes + k = ceil_div(modBits, 8) # Convert from bits to bytes hLen = self._hashObj.digest_size mLen = len(message) @@ -159,11 +161,11 @@ class PKCS1OAEP_Cipher: # See 7.1.2 in RFC3447 modBits = Crypto.Util.number.size(self._key.n) - k = ceil_div(modBits,8) # Convert from bits to bytes + k = ceil_div(modBits, 8) # Convert from bits to bytes hLen = self._hashObj.digest_size # Step 1b and 1c - if len(ciphertext) != k or k<hLen+2: + if len(ciphertext) != k or k < hLen+2: raise ValueError("Ciphertext with incorrect length.") # Step 2a (O2SIP) ct_int = bytes_to_long(ciphertext) @@ -171,8 +173,6 @@ class PKCS1OAEP_Cipher: em = self._key._decrypt(ct_int) # Step 3a lHash = self._hashObj.new(self._label).digest() - # Step 3b - y = em[0] # y must be 0, but we MUST NOT check it here in order not to # allow attacks like Manger's (http://dl.acm.org/citation.cfm?id=704143) maskedSeed = em[1:hLen+1] @@ -185,19 +185,12 @@ class PKCS1OAEP_Cipher: dbMask = self._mgf(seed, k-hLen-1) # Step 3f db = strxor(maskedDB, dbMask) - # Step 3g - one_pos = db[hLen:].find(b'\x01') - lHash1 = db[:hLen] - invalid = bord(y) | int(one_pos < 0) - hash_compare = strxor(lHash1, lHash) - for x in hash_compare: - invalid |= bord(x) - for x in db[hLen:one_pos]: - invalid |= bord(x) - if invalid != 0: + # Step 3b + 3g + res = oaep_decode(em, lHash, db) + if res <= 0: raise ValueError("Incorrect decryption.") # Step 4 - return db[hLen + one_pos + 1:] + return db[res:] def new(key, hashAlgo=None, mgfunc=None, label=b'', randfunc=None): """Return a cipher object :class:`PKCS1OAEP_Cipher` that can be used to perform PKCS#1 OAEP encryption or decryption. @@ -234,4 +227,3 @@ def new(key, hashAlgo=None, mgfunc=None, if randfunc is None: randfunc = Random.get_random_bytes return PKCS1OAEP_Cipher(key, hashAlgo, mgfunc, label, randfunc) - --- a/lib/Crypto/Cipher/PKCS1_v1_5.py +++ b/lib/Crypto/Cipher/PKCS1_v1_5.py @@ -18,15 +18,18 @@ # ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. -# =================================================================== +# ================================================================== __all__ = [ 'new', 'PKCS115_Cipher' ] from Crypto.Util.number import ceil_div, bytes_to_long, long_to_bytes -from Crypto.Util.py3compat import bord, _copy_bytes +from Crypto.Util.py3compat import bord, is_bytes, _copy_bytes import Crypto.Util.number from Crypto import Random +from ._pkcs1_oaep_decode import pkcs1_decode + + class PKCS115_Cipher: """This cipher can perform PKCS#1 v1.5 RSA encryption or decryption. Do not instantiate directly. Use :func:`Crypto.Cipher.PKCS1_v1_5.new` instead.""" @@ -89,7 +92,6 @@ class PKCS115_Cipher: continue ps.append(new_byte) ps = b"".join(ps) - assert(len(ps) == k - mLen - 3) # Step 2b em = b'\x00\x02' + ps + b'\x00' + _copy_bytes(None, None, message) # Step 3a (OS2IP) @@ -100,7 +102,7 @@ class PKCS115_Cipher: c = long_to_bytes(m_int, k) return c - def decrypt(self, ciphertext, sentinel): + def decrypt(self, ciphertext, sentinel, expected_pt_len=0): r"""Decrypt a PKCS#1 v1.5 ciphertext. This function is named ``RSAES-PKCS1-V1_5-DECRYPT``, and is specified in @@ -167,13 +169,17 @@ class PKCS115_Cipher: ct_int = bytes_to_long(ciphertext) # Step 2b (RSADP) and Step 2c (I2OSP) em = self._key._decrypt(ct_int) - # Step 3 - sep = em.find(b'\x00', 2) - if not em.startswith(b'\x00\x02') or sep < 10: - return sentinel - # Step 4 - return em[sep + 1:] - + # Step 3 (not constant time when the sentinel is not a byte string) + output = bytes(bytearray(k)) + if not is_bytes(sentinel) or len(sentinel) > k: + size = pkcs1_decode(em, b'', expected_pt_len, output) + if size < 0: + return sentinel + else: + return output[size:] + # Step 3 (somewhat constant time) + size = pkcs1_decode(em, sentinel, expected_pt_len, output) + return output[size:] def new(key, randfunc=None): """Create a cipher for performing PKCS#1 v1.5 encryption or decryption. --- /dev/null +++ b/lib/Crypto/Cipher/_pkcs1_oaep_decode.py @@ -0,0 +1,41 @@ +from Crypto.Util._raw_api import (load_pycryptodome_raw_lib, c_size_t, + c_uint8_ptr) + + +_raw_pkcs1_decode = load_pycryptodome_raw_lib("Crypto.Cipher._pkcs1_decode", + """ + int pkcs1_decode(const uint8_t *em, size_t len_em, + const uint8_t *sentinel, size_t len_sentinel, + size_t expected_pt_len, + uint8_t *output); + + int oaep_decode(const uint8_t *em, + size_t em_len, + const uint8_t *lHash, + size_t hLen, + const uint8_t *db, + size_t db_len); + """) + + +def pkcs1_decode(em, sentinel, expected_pt_len, output): + if len(em) != len(output): + raise ValueError("Incorrect output length") + + ret = _raw_pkcs1_decode.pkcs1_decode(c_uint8_ptr(em), + c_size_t(len(em)), + c_uint8_ptr(sentinel), + c_size_t(len(sentinel)), + c_size_t(expected_pt_len), + c_uint8_ptr(output)) + return ret + + +def oaep_decode(em, lHash, db): + ret = _raw_pkcs1_decode.oaep_decode(c_uint8_ptr(em), + c_size_t(len(em)), + c_uint8_ptr(lHash), + c_size_t(len(lHash)), + c_uint8_ptr(db), + c_size_t(len(db))) + return ret --- a/lib/Crypto/SelfTest/Cipher/test_pkcs1_15.py +++ b/lib/Crypto/SelfTest/Cipher/test_pkcs1_15.py @@ -23,6 +23,7 @@ from __future__ import print_function import json +import sys import unittest from binascii import unhexlify @@ -151,7 +152,10 @@ HKukWBcq9f/UOmS0oEhai/6g+Uf7VHJdWaeO5Lzu pt_int = bytes_to_long(pt) ct_int = self.key1024._encrypt(pt_int) ct = long_to_bytes(ct_int, 128) - self.assertEqual("---", cipher.decrypt(ct, "---")) + if sys.version_info[0] == 2: + self.assertEqual("---", cipher.decrypt(ct, "---")) + else: + self.assertEqual(b"", cipher.decrypt(ct, "---")) def testEncryptVerify1(self): # Encrypt/Verify messages of length [0..RSAlen-11] --- a/lib/Crypto/Util/py3compat.py +++ b/lib/Crypto/Util/py3compat.py @@ -104,6 +104,11 @@ if sys.version_info[0] == 2: def is_string(x): return isinstance(x, basestring) + def is_bytes(x): + return isinstance(x, str) or \ + isinstance(x, bytearray) or \ + isinstance(x, memoryview) + ABC = abc.ABCMeta('ABC', (object,), {'__slots__': ()}) else: @@ -146,6 +151,11 @@ else: def is_string(x): return isinstance(x, str) + def is_bytes(x): + return isinstance(x, bytes) or \ + isinstance(x, bytearray) or \ + isinstance(x, memoryview) + from abc import ABC --- a/setup.py +++ b/setup.py @@ -381,6 +381,10 @@ ext_modules = [ include_dirs=['src/'], sources=['src/cpuid.c']), + Extension("Crypto.Cipher._pkcs1_decode", + include_dirs=['src/'], + sources=['src/pkcs1_decode.c']), + # Chaining modes Extension("Crypto.Cipher._raw_ecb", include_dirs=['src/'], --- /dev/null +++ b/src/pkcs1_decode.c @@ -0,0 +1,354 @@ +/* =================================================================== + * + * Copyright (c) 2021, Helder Eijs <helderijs@gmail.com> + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions + * are met: + * + * 1. Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in + * the documentation and/or other materials provided with the + * distribution. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS + * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE + * COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, + * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, + * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; + * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER + * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT + * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN + * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + * =================================================================== + */ + +#include "common.h" + +FAKE_INIT(pkcs1_decode) + +#define SIZE_T_MAX ((size_t)-1) +#define SIZE_T_LEN (sizeof(size_t)) + +STATIC uint8_t rol8(uint8_t x) +{ + return (uint8_t)((x << 1) | (x >> 7)); +} + +/* + * Return 0 if x is 0, otherwise SIZE_T_MAX (all bits set to 1) + */ +STATIC size_t propagate_ones(uint8_t x) +{ + unsigned i; + uint8_t result8; + size_t result; + + result8 = x; + for (i=0; i<8; i++) { + x = rol8(x); + result8 |= x; + } + result = 0; + for (i=0; i<sizeof(result); i++) { + result |= ((size_t)result8) << (i*8); + } + + return result; +} + +/* + * Set all bits to 1 in the flag if term1 == term2 + * or leave the flag untouched otherwise. + */ +STATIC void set_if_match(uint8_t *flag, size_t term1, size_t term2) +{ + unsigned i; + uint8_t x; + + x = 0; + for (i=0; i<sizeof(size_t); i++) { + x |= (uint8_t)((term1 ^ term2) >> (i*8)); + } + *flag |= (uint8_t)~propagate_ones(x); +} + +/* + * Set all bits to 1 in the flag if term1 != term2 + * or leave the flag untouched otherwise. + */ +STATIC void set_if_no_match(uint8_t *flag, size_t term1, size_t term2) +{ + size_t i; + uint8_t x; + + x = 0; + for (i=0; i<sizeof(size_t); i++) { + x |= (uint8_t)((term1 ^ term2) >> (i*8)); + } + *flag |= (uint8_t)propagate_ones(x); +} + +/* + * Copy in1[] into out[] if choice is 0, otherwise copy in2[] + */ +STATIC void safe_select(const uint8_t *in1, const uint8_t *in2, uint8_t *out, uint8_t choice, size_t len) +{ + size_t i; + uint8_t mask1, mask2; + + mask1 = (uint8_t)propagate_ones(choice); + mask2 = (uint8_t)~mask1; + for (i=0; i<len; i++) { + out[i] = (in1[i] & mask2) | (in2[i] & mask1); + /* yes, these rol8s are redundant, but we try to avoid compiler optimizations */ + mask1 = rol8(mask1); + mask2 = rol8(mask2); + } +} + +/* + * Return in1 if choice is 0, in2 otherwise. + */ +STATIC size_t safe_select_idx(size_t in1, size_t in2, uint8_t choice) +{ + size_t mask; + + mask = propagate_ones(choice); + return (in1 & ~mask) | (in2 & mask); +} + +/* + * Return 0 if all these conditions hold: + * - in1[] is equal to in2[] where eq_mask[] is 0xFF, + * - in1[] is NOT equal to in2[] where neq_mask[] is 0xFF. + * Return non-zero otherwise. + */ +STATIC uint8_t safe_cmp_masks(const uint8_t *in1, const uint8_t *in2, + const uint8_t *eq_mask, const uint8_t *neq_mask, + size_t len) +{ + size_t i; + uint8_t c, result; + + result = 0; + for (i=0; i<len; i++) { + c = (uint8_t)propagate_ones(*in1++ ^ *in2++); + result |= (uint8_t)(c & *eq_mask++); /* Set all bits to 1 if *in1 and *in2 differed + and eq_mask was 0xff */ + result |= (uint8_t)(~c & *neq_mask++); /* Set all bits to 1 if *in1 and *in2 matched + and neq_mask was 0xff */ + } + + return result; +} + +/* + * Return the index of the first byte with value c, + * the length of in1[] when c is not present, + * or SIZE_T_MAX in case of problems. + */ +STATIC size_t safe_search(const uint8_t *in1, uint8_t c, size_t len) +{ + size_t result, mask1, mask2, i; + uint8_t *in1_c; + + if (NULL == in1 || 0 == len) { + return SIZE_T_MAX; + } + + /* + * Create a second byte string and put c at the end, + * so that at least we will find it there. + */ + in1_c = (uint8_t*) malloc(len + 1); + if (NULL == in1_c) { + return SIZE_T_MAX; + } + memcpy(in1_c, in1, len); + in1_c[len] = c; + + result = 0; + mask2 = 0; + for (i=0; i<(len+1); i++) { + mask1 = ~mask2 & ~propagate_ones(in1_c[i] ^ c); /* Set mask1 to 0xff if there is a match + and it is the first one. */ + result |= i & mask1; + mask2 |= mask1; + } + + free(in1_c); + return result; +} + +#define PKCS1_PREFIX_LEN 10 + +/* + * Decode and verify the PKCS#1 padding, then put either the plaintext + * or the sentinel value into the output buffer, all in constant time. + * + * The output is a buffer of equal length as the encoded message (em). + * + * The sentinel is put into the buffer when decryption fails. + * + * The caller may already know the expected length of the plaintext M, + * which they should put into expected_pt_len. + * Otherwise, expected_pt_len must be set to 0. + * + * Either the plaintext or the sentinel will be put into the buffer + * with padding on the left. + * + * The function returns the number of bytes to ignore at the beginning + * of the output buffer, or -1 in case of problems. + */ +EXPORT_SYM int pkcs1_decode(const uint8_t *em, size_t len_em_output, + const uint8_t *sentinel, size_t len_sentinel, + size_t expected_pt_len, + uint8_t *output) +{ + size_t pos; + uint8_t match, selector; + uint8_t *padded_sentinel; + int result; + + result = -1; + + if (NULL == em || NULL == output || NULL == sentinel) { + return -1; + } + if (len_em_output < (PKCS1_PREFIX_LEN + 2)) { + return -1; + } + if (len_sentinel > len_em_output) { + return -1; + } + if (expected_pt_len > 0 && expected_pt_len > (len_em_output - PKCS1_PREFIX_LEN - 1)) { + return -1; + } + + /** Pad sentinel (on the left) so that it matches em in length **/ + padded_sentinel = (uint8_t*) calloc(1, len_em_output); + if (NULL == padded_sentinel) { + return -1; + } + memcpy(padded_sentinel + (len_em_output - len_sentinel), sentinel, len_sentinel); + + /** The first 10 bytes must follow the pattern **/ + match = safe_cmp_masks(em, + (const uint8_t*)"\x00\x02" "\x00\x00\x00\x00\x00\x00\x00\x00", + (const uint8_t*)"\xFF\xFF" "\x00\x00\x00\x00\x00\x00\x00\x00", + (const uint8_t*)"\x00\x00" "\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF", + 10); + + /* + * pos is the index of the first 0 byte. It is followed by the plaintext M. + * It can be (len_em_output-1) when the 0 is at the end (empty M). + * It can be len_em_output when the 0 is not present. + * It can SIZE_T_MAX in case of other errors. + */ + pos = safe_search(em + 10, 0, len_em_output - 10) + 10; + if (pos == SIZE_T_MAX) { + result = -1; + goto end; + } + + /* + * selector is 0 if: + * - there is a match for the first 10 bytes AND + * - a 0 byte is found in the remainder of em, AND + * - the length of the plaintext matches the expectation (if there is one) + */ + selector = match; + set_if_match(&selector, pos, len_em_output); + if (expected_pt_len > 0) { + size_t pt_len; + + pt_len = len_em_output - pos - 1; + set_if_no_match(&selector, pt_len, expected_pt_len); + } + + /** Select the correct data to output **/ + safe_select(em, padded_sentinel, output, selector, len_em_output); + + /** Select the number of bytes that the caller will skip in output **/ + result = (int)safe_select_idx(pos + 1, len_em_output - len_sentinel, selector); + +end: + free(padded_sentinel); + return result; +} + +/* + * Decode and verify the OAEP padding in constant time. + * + * The function returns the number of bytes to ignore at the beginning + * of db (the rest is the plaintext), or -1 in case of problems. + */ + +EXPORT_SYM int oaep_decode(const uint8_t *em, + size_t em_len, + const uint8_t *lHash, + size_t hLen, + const uint8_t *db, + size_t db_len) /* em_len - 1 - hLen */ +{ + int result; + size_t one_pos, search_len, i; + uint8_t wrong_padding; + uint8_t *eq_mask = NULL; + uint8_t *neq_mask = NULL; + uint8_t *target_db = NULL; + + if (NULL == em || NULL == lHash || NULL == db) { + return -1; + } + + if (em_len < 2*hLen+2 || db_len != em_len-1-hLen) { + return -1; + } + + /* Allocate */ + eq_mask = (uint8_t*) calloc(1, db_len); + neq_mask = (uint8_t*) calloc(1, db_len); + target_db = (uint8_t*) calloc(1, db_len); + if (NULL == eq_mask || NULL == neq_mask || NULL == target_db) { + result = -1; + goto cleanup; + } + + /* Step 3g */ + search_len = db_len - hLen; + + one_pos = safe_search(db + hLen, 0x01, search_len); + if (SIZE_T_MAX == one_pos) { + result = -1; + goto cleanup; + } + + memset(eq_mask, 0xAA, db_len); + memcpy(target_db, lHash, hLen); + memset(eq_mask, 0xFF, hLen); + + for (i=0; i<search_len; i++) { + eq_mask[hLen + i] = propagate_ones(i < one_pos); + } + + wrong_padding = em[0]; + wrong_padding |= safe_cmp_masks(db, target_db, eq_mask, neq_mask, db_len); + set_if_match(&wrong_padding, one_pos, search_len); + + result = wrong_padding ? -1 : (int)(hLen + 1 + one_pos); + +cleanup: + free(eq_mask); + free(neq_mask); + free(target_db); + + return result; +} --- a/src/test/Makefile +++ b/src/test/Makefile @@ -96,6 +96,11 @@ build/clmul.o: ../ghash_clmul.c build/test_clmul: test_clmul.c ../common.h build/clmul.o $(CC) $(CFLAGS) -mssse3 -mpclmul $(CPPFLAGS) -o $@ $(filter %.c %.o, $^) +# pkcs1 + +build/pkcs1_decode.o: ../pkcs1_decode.c + $(CC) $(CFLAGS) $(CPPFLAGS) -o $@ $^ -c + # Poly1305 build/poly1305.o: ../poly1305.c
Locations
Projects
Search
Status Monitor
Help
OpenBuildService.org
Documentation
API Documentation
Code of Conduct
Contact
Support
@OBShq
Terms
openSUSE Build Service is sponsored by
The Open Build Service is an
openSUSE project
.
Sign Up
Log In
Places
Places
All Projects
Status Monitor
