niryo_libraries.rosbag.bag._ROSBagAesCbcEncryptor

class documentation

class _ROSBagAesCbcEncryptor(_ROSBagEncryptor):

Constructor: _ROSBagAesCbcEncryptor()

Class for AES-CBC-encrypted bags.

Method	`__init__`	Create AES encryptor.
Method	`add_fields_to_file_header`	Add encryptor information to bag file header. @param header: bag file header @type header: dict
Method	`add_info_rows`	Add rows for rosbag info. @param rows: information on bag encryption @type rows: list of tuples
Method	`decrypt_chunk`	Decrypt chunk. @param encrypted_chunk: chunk to decrypt @type encrypted_chunk: str @return: decrypted chunk @rtype: str @raise ROSBagFormatException: if size of input chunk is not multiple of AES block size...
Method	`encrypt_chunk`	Read chunk from file, encrypt it, and write back to file. @param chunk_size: size of chunk @type chunk_size: int @param chunk_data_pos: position of chunk data portion @type chunk_data_pos: int @param f: file stream @type f: file @return: size of initialization vector and encrypted chunk @rtype: int...
Method	`get_info_str`	Return string for rosbag info. @return: information on bag encryption @rtype: str
Method	`initialize`	Initialize encryptor by composing AES symmetric key. @param bag: bag to be encrypted/decrypted @type bag: Bag @param gpg_key_user: user name of GPG key used for symmetric key encryption @type gpg_key_user: str @raise ROSBagException: if GPG key user has already been set...
Method	`read_encrypted_header`	Read encrypted header from bag file. @param f: file stream @type f: file @param req_op: expected header op code @type req_op: int @return: decrypted header @rtype: dict @raise ROSBagFormatException: if error occurs while decrypting/reading header...
Method	`read_fields_from_file_header`	Read encryptor information from bag file header. @param header: bag file header @type header: dict @raise ROSBagFormatException: if GPG key user is not found in header
Method	`write_encrypted_header`	Write encrypted header to bag file. @param f: file stream @type f: file @param header: unencrypted header @type header: dict @return: encrypted string representing header @rtype: str
Constant	`NAME`	Undocumented
Method	`_build_symmetric_key`	Undocumented
Method	`_decrypt_encrypted_header`	Undocumented
Constant	`_ENCRYPTED_KEY_FIELD_NAME`	Undocumented
Constant	`_GPG_USER_FIELD_NAME`	Undocumented
Instance Variable	`_encrypted_symmetric_key`	Undocumented
Instance Variable	`_gpg_key_user`	Undocumented
Instance Variable	`_gpg_passphrase`	Undocumented
Instance Variable	`_symmetric_key`	Undocumented

Inherited from _ROSBagEncryptor:

Constant _ENCRYPTOR_FIELD_NAME Undocumented

def __init__(self): ¶

overrides niryo_libraries.rosbag.bag._ROSBagEncryptor.__init__

Create AES encryptor.

def add_fields_to_file_header(self, header): ¶

Add encryptor information to bag file header. @param header: bag file header @type header: dict

def add_info_rows(self, rows): ¶

Add rows for rosbag info. @param rows: information on bag encryption @type rows: list of tuples

def decrypt_chunk(self, encrypted_chunk): ¶

Decrypt chunk. @param encrypted_chunk: chunk to decrypt @type encrypted_chunk: str @return: decrypted chunk @rtype: str @raise ROSBagFormatException: if size of input chunk is not multiple of AES block size

def encrypt_chunk(self, chunk_size, chunk_data_pos, f): ¶

Read chunk from file, encrypt it, and write back to file. @param chunk_size: size of chunk @type chunk_size: int @param chunk_data_pos: position of chunk data portion @type chunk_data_pos: int @param f: file stream @type f: file @return: size of initialization vector and encrypted chunk @rtype: int

def get_info_str(self): ¶

Return string for rosbag info. @return: information on bag encryption @rtype: str

def initialize(self, bag, gpg_key_user, passphrase=None): ¶

Initialize encryptor by composing AES symmetric key. @param bag: bag to be encrypted/decrypted @type bag: Bag @param gpg_key_user: user name of GPG key used for symmetric key encryption @type gpg_key_user: str @raise ROSBagException: if GPG key user has already been set

def read_encrypted_header(self, _, f, req_op=None): ¶

Read encrypted header from bag file. @param f: file stream @type f: file @param req_op: expected header op code @type req_op: int @return: decrypted header @rtype: dict @raise ROSBagFormatException: if error occurs while decrypting/reading header

def read_fields_from_file_header(self, header): ¶

Read encryptor information from bag file header. @param header: bag file header @type header: dict @raise ROSBagFormatException: if GPG key user is not found in header

def write_encrypted_header(self, _, f, header): ¶

Write encrypted header to bag file. @param f: file stream @type f: file @param header: unencrypted header @type header: dict @return: encrypted string representing header @rtype: str

NAME: str = ¶

Undocumented

Value

'rosbag/AesCbcEncryptor'

def _build_symmetric_key(self): ¶

Undocumented

def _decrypt_encrypted_header(self, f): ¶

Undocumented

_ENCRYPTED_KEY_FIELD_NAME: str = ¶

Undocumented

Value

'encrypted_key'

_GPG_USER_FIELD_NAME: str = ¶

Undocumented

Value

'gpg_user'

_encrypted_symmetric_key = ¶

Undocumented

_gpg_key_user = ¶

Undocumented

_gpg_passphrase = ¶

Undocumented

_symmetric_key = ¶

Undocumented