Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/Crypto/Util/_raw_api.py: 48%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# ===================================================================
2#
3# Copyright (c) 2014, Legrandin <helderijs@gmail.com>
4# All rights reserved.
5#
6# Redistribution and use in source and binary forms, with or without
7# modification, are permitted provided that the following conditions
8# are met:
9#
10# 1. Redistributions of source code must retain the above copyright
11# notice, this list of conditions and the following disclaimer.
12# 2. Redistributions in binary form must reproduce the above copyright
13# notice, this list of conditions and the following disclaimer in
14# the documentation and/or other materials provided with the
15# distribution.
16#
17# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
18# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
19# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
20# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
21# COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
22# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
23# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
24# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
25# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
26# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
27# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
28# POSSIBILITY OF SUCH DAMAGE.
29# ===================================================================
31from __future__ import annotations
33import _thread
34import abc
35import os
36import sys
37import threading
38import weakref
39from importlib import machinery
40from typing import Any, Optional, Union
42from Crypto.Util._file_system import pycryptodome_filename
44# List of file suffixes for Python extensions
45extension_suffixes = machinery.EXTENSION_SUFFIXES
47# Which types with buffer interface we support (apart from byte strings)
48_buffer_type = (bytearray, memoryview)
50# While a C function runs, the GIL is released, so another thread could
51# resize a bytearray, or release a memoryview, whose memory the C function
52# is using. CPython prevents that for as long as the object returned by
53# c_uint8_ptr() is alive, but PyPy may not: PyPy 7.3 cannot lock any buffer,
54# PyPy 8.0 can lock a bytearray but not a memoryview.
55# So, on PyPy, if another thread exists, C functions get a private copy of
56# the buffers that PyPy cannot lock (see c_uint8_ptr and c_uint8_ptr_out).
57_is_pypy = "__pypy__" in sys.builtin_module_names
59# The buffer types that this PyPy keeps locked (found when cffi is loaded)
60_pypy_locked_types: tuple = ()
63def _must_copy(data: Any) -> bool:
64 """Whether C code must get a private copy of the bytearray or
65 memoryview data, instead of its memory (see above).
67 With one thread only, nothing can change data while a C function runs:
68 C code never starts Python threads, and signal handlers only run
69 when it returns.
70 """
71 return _is_pypy and not isinstance(data, _pypy_locked_types) and _thread._count() > 0
74class _BytesOutput:
75 """A new bytes object that C code writes into (see create_output_buffer)"""
77 __slots__ = ("obj", "ptr")
79 def __init__(self, obj: bytes, ptr: Any) -> None:
80 self.obj = obj
81 self.ptr = ptr
83 def __len__(self) -> int:
84 return len(self.obj)
87class _VoidPointer:
88 @abc.abstractmethod
89 def get(self) -> Any:
90 """Return the memory location we point to"""
91 return
93 @abc.abstractmethod
94 def address_of(self) -> Any:
95 """Return a raw pointer to this pointer"""
96 return
99try:
100 # Starting from v2.18, pycparser (used by cffi for in-line ABI mode)
101 # stops working correctly when PYOPTIMIZE==2 or the parameter -OO is
102 # passed. In that case, we fall back to ctypes.
103 # Note that PyPy ships with an old version of pycparser so we can keep
104 # using cffi there.
105 # See https://github.com/Legrandin/pycryptodome/issues/228
106 if "__pypy__" not in sys.builtin_module_names and sys.flags.optimize == 2:
107 raise ImportError("CFFI with optimize=2 fails due to pycparser bug.")
109 # cffi < 1.16.0 crashes on CPython 3.12+ for Windows due to a removed C API.
110 # See https://groups.google.com/g/python-cffi/c/oZkOIZ_zi5k
111 import cffi
113 if (
114 sys.version_info >= (3, 12)
115 and os.name == "nt"
116 and sys.implementation.name == "cpython"
117 and cffi.__version_info__ < (1, 16)
118 ):
119 raise ImportError("CFFI 1.16.0+ is required with CPython 3.12+ on Windows")
121 # ffi.from_buffer() with a type (used by c_uint8_ptr) requires cffi 1.12+
122 if cffi.__version_info__ < (1, 12):
123 raise ImportError("CFFI 1.12.0+ is required")
125 ffi: Any = cffi.FFI()
126 null_pointer: Any = ffi.NULL
128 _Array = ffi.new("uint8_t[1]").__class__.__bases__
130 def _find_pypy_locked_types() -> tuple:
131 """Return the buffer types that cannot be resized or released
132 while ffi.from_buffer() uses them"""
134 locked: list[type] = []
135 data = bytearray(1)
136 ptr = ffi.from_buffer("uint8_t[]", data)
137 try:
138 data.append(0)
139 except BufferError:
140 locked.append(bytearray)
141 data = bytearray(1)
142 view = memoryview(data)
143 ptr = ffi.from_buffer("uint8_t[]", view)
144 try:
145 view.release()
146 except BufferError:
147 try:
148 data.append(0)
149 except BufferError:
150 locked.append(memoryview)
151 del ptr
152 return tuple(locked)
154 if _is_pypy:
155 _pypy_locked_types = _find_pypy_locked_types()
157 # Declarations already passed to ffi.cdef(), and the lock that
158 # protects it when modules are imported from several threads
159 _declared: set = set()
160 _declared_lock = threading.Lock()
162 def load_lib(name: str, cdecl: str) -> Any:
163 """Load a shared library and return a handle to it.
165 @name, either an absolute path or the name of a library
166 in the system search path.
168 @cdecl, the C function declarations.
169 """
171 if hasattr(ffi, "RTLD_DEEPBIND") and not os.getenv("PYCRYPTODOME_DISABLE_DEEPBIND"):
172 lib = ffi.dlopen(name, ffi.RTLD_DEEPBIND)
173 else:
174 lib = ffi.dlopen(name)
175 # Several libraries can export the same functions (for instance,
176 # a portable and an optimized build of the same code), but
177 # cffi rejects a second declaration of a function.
178 with _declared_lock:
179 if cdecl not in _declared:
180 ffi.cdef(cdecl)
181 _declared.add(cdecl)
182 return lib
184 def c_ulong(x: int) -> Any:
185 """Convert a Python integer to unsigned long"""
186 return x
188 c_ulonglong = c_ulong
189 c_uint = c_ulong
190 c_ubyte = c_ulong
192 def c_size_t(x: int) -> Any:
193 """Convert a Python integer to size_t"""
194 return x
196 def create_string_buffer(init_or_size: Union[bytes, int], size: Optional[int] = None) -> Any:
197 """Allocate the given amount of bytes (initially set to 0)"""
199 if isinstance(init_or_size, bytes):
200 size = max(len(init_or_size) + 1, size or 0)
201 result = ffi.new("uint8_t[]", size)
202 result[:] = init_or_size
203 else:
204 if size:
205 raise ValueError("Size must be specified once only")
206 result = ffi.new("uint8_t[]", init_or_size)
207 return result
209 def get_c_string(c_string: Any) -> bytes:
210 """Convert a C string into a Python byte sequence"""
211 return ffi.string(c_string)
213 def get_raw_buffer(buf: Any) -> bytes:
214 """Convert a C buffer into a Python byte sequence"""
215 if isinstance(buf, _BytesOutput):
216 return buf.obj
217 return ffi.buffer(buf)[:]
219 def c_uint8_ptr(data: Union[bytes, memoryview, bytearray]) -> Any:
220 """Pass data to C code, which only reads it"""
221 if isinstance(data, _buffer_type):
222 if _is_pypy and _must_copy(data):
223 return bytes(data)
224 # The returned object holds the buffer of data, which cannot
225 # be resized or freed (e.g. by another thread, while the GIL
226 # is released during a C call) for as long as it is alive.
227 return ffi.from_buffer("uint8_t[]", data)
228 elif isinstance(data, (bytes, _Array)):
229 return data
230 else:
231 raise TypeError("Object type %s cannot be passed to C code" % type(data))
233 def _c_uint8_ptr_into_bytes(obj: bytes) -> Any:
234 # On CPython, it points to the content of obj (no copy)
235 return ffi.from_buffer("uint8_t[]", obj)
237 class VoidPointer_cffi(_VoidPointer):
238 """Model a newly allocated pointer to void"""
240 def __init__(self):
241 self._pp = ffi.new("void *[1]")
243 def get(self):
244 return self._pp[0]
246 def address_of(self):
247 return self._pp
249 def VoidPointer() -> _VoidPointer:
250 return VoidPointer_cffi()
252 backend: str = "cffi"
254except ImportError:
255 import ctypes
256 from ctypes import ( # type: ignore[assignment]
257 CDLL,
258 byref,
259 c_size_t,
260 c_uint,
261 c_ulong,
262 c_ulonglong,
263 c_void_p,
264 create_string_buffer,
265 )
266 from ctypes import Array as _Array # type: ignore[assignment]
267 from ctypes.util import find_library
269 null_pointer = None
270 cached_architecture: list[str] = []
272 def c_ubyte(c: int) -> Any: # type: ignore[misc]
273 if not (0 <= c < 256):
274 raise OverflowError()
275 return ctypes.c_ubyte(c)
277 def load_lib(name: str, cdecl: str) -> Any:
278 if not cached_architecture:
279 # platform.architecture() creates a subprocess, so caching the
280 # result makes successive imports faster.
281 import platform
283 cached_architecture[:] = platform.architecture()
284 _bits, linkage = cached_architecture
285 if "." not in name and not linkage.startswith("Win"):
286 full_name = find_library(name)
287 if full_name is None:
288 raise OSError("Cannot load library '%s'" % name)
289 name = full_name
290 return CDLL(name)
292 def get_c_string(c_string: Any) -> bytes:
293 return c_string.value
295 def get_raw_buffer(buf: Any) -> bytes:
296 if isinstance(buf, _BytesOutput):
297 return buf.obj
298 return buf.raw
300 # ---- Get raw pointer ---
302 _c_ssize_t = ctypes.c_ssize_t
304 _PyBUF_SIMPLE = 0
305 _PyObject_GetBuffer = ctypes.pythonapi.PyObject_GetBuffer
306 _PyBuffer_Release = ctypes.pythonapi.PyBuffer_Release
307 _py_object = ctypes.py_object
308 _c_ssize_p = ctypes.POINTER(_c_ssize_t)
310 # See Include/object.h for CPython
311 # and https://github.com/pallets/click/blob/master/src/click/_winconsole.py
312 class _Py_buffer(ctypes.Structure):
313 _fields_ = [
314 ("buf", c_void_p),
315 ("obj", ctypes.py_object),
316 ("len", _c_ssize_t),
317 ("itemsize", _c_ssize_t),
318 ("readonly", ctypes.c_int),
319 ("ndim", ctypes.c_int),
320 ("format", ctypes.c_char_p),
321 ("shape", _c_ssize_p),
322 ("strides", _c_ssize_p),
323 ("suboffsets", _c_ssize_p),
324 ("internal", c_void_p),
325 ]
327 def c_uint8_ptr(data: Union[bytes, memoryview, bytearray]) -> Any:
328 if isinstance(data, (bytes, _Array)):
329 return data
330 elif isinstance(data, bytearray):
331 # Fast path: the array holds the buffer of data until it is
332 # garbage collected, so data cannot be resized or freed
333 # (e.g. by another thread, while the GIL is released during
334 # a C call).
335 # The memoryview locks the buffer and reads its length at once:
336 # with len(data) first, another thread could shrink data
337 # before from_buffer() locks it.
338 view = memoryview(data)
339 return (ctypes.c_ubyte * view.nbytes).from_buffer(view)
340 elif isinstance(data, _buffer_type):
341 # memoryview objects can be read-only, which from_buffer()
342 # does not accept
343 obj = _py_object(data)
344 buf = _Py_buffer()
345 _PyObject_GetBuffer(obj, byref(buf), _PyBUF_SIMPLE)
346 try:
347 buffer_type = ctypes.c_ubyte * buf.len
348 result = buffer_type.from_address(buf.buf)
349 # Hold the buffer of data until the returned array is
350 # garbage collected: until then, data cannot be resized
351 # or freed (e.g. by another thread, while the GIL is
352 # released during a C call).
353 weakref.finalize(result, _PyBuffer_Release, byref(buf))
354 except BaseException:
355 _PyBuffer_Release(byref(buf))
356 raise
357 return result
358 else:
359 raise TypeError("Object type %s cannot be passed to C code" % type(data))
361 def _c_uint8_ptr_into_bytes(obj: bytes) -> Any:
362 return (ctypes.c_ubyte * len(obj)).from_address(_PyBytes_AsString(obj))
364 # ---
366 class VoidPointer_ctypes(_VoidPointer):
367 """Model a newly allocated pointer to void"""
369 def __init__(self):
370 self._p = c_void_p()
372 def get(self):
373 return self._p
375 def address_of(self):
376 return byref(self._p)
378 def VoidPointer() -> _VoidPointer:
379 return VoidPointer_ctypes()
381 backend = "ctypes"
384# Only on CPython, C code can write the output directly into a new bytes
385# object, which is then returned without copying it. This is what C extensions
386# do with PyBytes_FromStringAndSize(NULL, size): the content is undefined,
387# and it must be filled in before anything else gets the object.
388# Private prototypes, so that the ones in ctypes.pythonapi are not changed.
389_PyBytes_FromStringAndSize: Any = None
390_PyBytes_AsString: Any = None
391if sys.implementation.name == "cpython":
392 try:
393 import ctypes as _ctypes
395 _PyBytes_FromStringAndSize = _ctypes.PYFUNCTYPE(
396 _ctypes.py_object, _ctypes.c_void_p, _ctypes.c_ssize_t
397 )(("PyBytes_FromStringAndSize", _ctypes.pythonapi))
398 _PyBytes_AsString = _ctypes.PYFUNCTYPE(_ctypes.c_void_p, _ctypes.py_object)(
399 ("PyBytes_AsString", _ctypes.pythonapi)
400 )
401 except (ImportError, AttributeError):
402 _PyBytes_FromStringAndSize = None
404# For shorter outputs, the usual copy is cheap
405_MIN_BYTES_OUTPUT = 64 * 1024
408def create_output_buffer(size: int) -> Any:
409 """Allocate the memory where C code writes the output of a function
410 into, of the given size::
412 out = create_output_buffer(size)
413 with c_uint8_ptr_out(out) as ptr:
414 lib.function(..., ptr, ...)
415 return get_raw_buffer(out)
417 Usually, it is create_string_buffer(size), which get_raw_buffer() copies.
418 On CPython, for long outputs, it is a new bytes object instead, which C code
419 writes directly into, and which get_raw_buffer() returns as it is.
420 Nothing else gets that bytes object until the C code has written it.
421 """
423 # The empty bytes object is a shared singleton
424 if _PyBytes_FromStringAndSize is None or size < max(_MIN_BYTES_OUTPUT, 1):
425 return create_string_buffer(size)
426 obj = _PyBytes_FromStringAndSize(None, size)
427 return _BytesOutput(obj, _c_uint8_ptr_into_bytes(obj))
430def c_uint8_ptr_len(data: Union[bytes, memoryview, bytearray]) -> tuple[Any, int]:
431 """Like c_uint8_ptr(), but also return the length of the memory that
432 C code gets. Pass C code that length, never len(data): another thread
433 could resize data in the meantime, but not the memory that C code gets,
434 for as long as the returned pointer is in use."""
436 ptr = c_uint8_ptr(data)
437 return ptr, len(ptr)
440class c_uint8_ptr_out:
441 """Pass a buffer to C code that writes into it::
443 with c_uint8_ptr_out(buf) as ptr:
444 lib.function(..., ptr, ...)
446 Usually, ptr is just c_uint8_ptr(buf). If the C code must not get the
447 memory of buf (see _must_copy), it gets a private copy instead, which is
448 copied back into buf when the C function returns (without an exception).
449 """
451 def __init__(self, data: Any) -> None:
452 self._data = data
453 self._copy: Any = None
455 def __enter__(self) -> Any:
456 if isinstance(self._data, _BytesOutput):
457 return self._data.ptr
458 if _is_pypy and isinstance(self._data, _buffer_type) and _must_copy(self._data):
459 self._copy = create_string_buffer(len(self._data))
460 self._copy[0 : len(self._data)] = bytes(self._data)
461 return self._copy
462 return c_uint8_ptr(self._data)
464 def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
465 if self._copy is not None and exc_type is None:
466 self._data[:] = get_raw_buffer(self._copy)
469class SmartPointer:
470 """Class to hold a non-managed piece of memory"""
472 def __init__(self, raw_pointer: Any, destructor: Any) -> None:
473 self._raw_pointer = raw_pointer
474 self._destructor = destructor
476 def get(self) -> Any:
477 return self._raw_pointer
479 def release(self) -> Any:
480 rp, self._raw_pointer = self._raw_pointer, None
481 return rp
483 def __del__(self):
484 try:
485 if self._raw_pointer is not None:
486 self._destructor(self._raw_pointer)
487 self._raw_pointer = None
488 except AttributeError:
489 pass
492def load_pycryptodome_raw_lib(name: str, cdecl: str) -> Any:
493 """Load a shared library and return a handle to it.
495 @name, the name of the library expressed as a PyCryptodome module,
496 for instance Crypto.Cipher._raw_cbc.
498 @cdecl, the C function declarations.
499 """
501 split = name.split(".")
502 dir_comps, basename = split[:-1], split[-1]
503 attempts = []
504 for ext in extension_suffixes:
505 try:
506 filename = basename + ext
507 full_name = pycryptodome_filename(dir_comps, filename)
508 if not os.path.isfile(full_name):
509 attempts.append("Not found '%s'" % filename)
510 continue
511 return load_lib(full_name, cdecl)
512 except OSError as exp:
513 attempts.append("Cannot load '%s': %s" % (filename, str(exp)))
514 raise OSError("Cannot load native module '%s': %s" % (name, ", ".join(attempts)))
517def is_buffer(x: Any) -> bool:
518 """Return True if object x supports the buffer interface"""
519 return isinstance(x, (bytes, bytearray, memoryview))
522def is_writeable_buffer(x: Any) -> bool:
523 return isinstance(x, bytearray) or (isinstance(x, memoryview) and not x.readonly)