Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@tink_error.use_tink_errors
def public_key_data(self, key_data: tink_pb2.KeyData) -> tink_pb2.KeyData:
return self._cc_key_manager.public_key_data(key_data)
@tink_error.use_tink_errors
def new_decrypting_stream(self, ciphertext_source: BinaryIO,
associated_data: bytes) -> BinaryIO:
stream = decrypting_stream.DecryptingStream(self._streaming_aead,
ciphertext_source,
associated_data)
return typing.cast(BinaryIO, stream)
@tink_error.use_tink_errors
def _get_output_stream_adapter(cc_primitive, aad, destination):
"""Implemented as a separate method to ensure correct error transform."""
return cc_streaming_aead_wrappers.new_cc_encrypting_stream(
cc_primitive, aad, destination)
@tink_error.use_tink_errors
def decrypt(self, plaintext: bytes, associated_data: bytes) -> bytes:
return self._aead.decrypt(plaintext, associated_data)
@tink_error.use_tink_errors
def encrypt(self, plaintext: bytes, associated_data: bytes) -> bytes:
return self._aead.encrypt(plaintext, associated_data)
@tink_error.use_tink_errors
def write(self, b: bytes) -> int:
"""Write the given buffer to the stream.
May use multiple calls to the underlying file object's write() method.
Returns:
The number of bytes written, which will always be the length of b in
bytes.
Raises:
BlockingIOError: if the write could not be fully completed, with
characters_written set to the number of bytes successfully written.
TinkError: if there was a permanent error.
Args:
b: The buffer to write.
@tink_error.use_tink_errors
def close(self) -> None:
"""Flush and close the stream.
This has no effect on a closed stream.
"""
if self.closed:
return
self.flush()
self._output_stream_adapter.close()
self._closed = True
@tink_error.use_tink_errors
def new_key_data(self,
key_template: tink_pb2.KeyTemplate) -> tink_pb2.KeyData:
return self._cc_key_manager.new_key_data(key_template)
@tink_error.use_tink_errors
def new_encrypting_stream(self, ciphertext_destination: BinaryIO,
associated_data: bytes) -> BinaryIO:
stream = encrypting_stream.EncryptingStream(self._streaming_aead,
ciphertext_destination,
associated_data)
return typing.cast(BinaryIO, stream)
@tink_error.use_tink_errors
def primitive(self, key_data: tink_pb2.KeyData) -> P:
return self._primitive_py_wrapper(self._cc_key_manager.primitive(key_data))