""" A small demo script that shows how to export the datastore (unrestricted API key) """ import requests import json import nacl.encoding import nacl.signing import nacl.secret import nacl.utils from nacl.public import PrivateKey, PublicKey, Box import binascii import socket api_key_id = '1794337d-de80-4aa0-8509-7070448221e6' api_key_private_key = '5c315e95703afd125d59bd26e5d7013683707a7671957b997b6cf11bb5670999' api_key_secret_key = '06f97520b4462565713851435c297ae8f70ee87fa6a8a4072bad87ccdb6d8d89' server_url = 'https://psonoclient.chickahoona.com/server' server_public_key = '02da2ad857321d701d754a7e60d0a147cdbc400ff4465e1f57bc2d9fbfeddf0b' server_signature = '4ce9e761e1d458fe18af577c50eb8249a0de535c9bd6b7a97885c331b46dcbd1' SSL_VERIFY = False def get_device_description(): """ This info is later shown in the "Open sessions" overview in the client. Should be something so the user knows where this session is coming from. :return: :rtype: """ return 'Console Client ' + socket.gethostname() def generate_client_login_info(): """ Generates and signs the login info Returns a tuple of the session private key and the login info :return: :rtype: """ box = PrivateKey.generate() session_private_key = box.encode(encoder=nacl.encoding.HexEncoder).decode() session_public_key = box.public_key.encode(encoder=nacl.encoding.HexEncoder).decode() info = { 'api_key_id': api_key_id, 'session_public_key': session_public_key, 'device_description': get_device_description(), } info = json.dumps(info) signing_box = nacl.signing.SigningKey(api_key_private_key, encoder=nacl.encoding.HexEncoder) # The first 128 chars (512 bits or 64 bytes) are the actual signature, the rest the binary encoded info signed = signing_box.sign(info.encode()) signature = binascii.hexlify(signed.signature) return session_private_key, { 'info': info, 'signature': signature.decode(), } def decrypt_server_login_info(login_info_hex, login_info_nonce_hex, session_public_key, session_private_key): """ Takes the login info and nonce together with the session public and private key. Will decrypt the login info and interpret it as json and return the json parsed object. :param login_info: :type login_info: :param login_info_nonce: :type login_info_nonce: :param session_public_key: :type session_public_key: :param session_private_key: :type session_private_key: :return: :rtype: """ crypto_box = Box(PrivateKey(session_private_key, encoder=nacl.encoding.HexEncoder), PublicKey(session_public_key, encoder=nacl.encoding.HexEncoder)) login_info = nacl.encoding.HexEncoder.decode(login_info_hex) login_info_nonce = nacl.encoding.HexEncoder.decode(login_info_nonce_hex) login_info = json.loads(crypto_box.decrypt(login_info, login_info_nonce).decode()) return login_info def verify_signature(login_info, login_info_signature): """ Takes the login info and the provided signature and will validate it with the help of server_signature. Will raise an exception if it does not match. :param login_info: :type login_info: :param login_info_signature: :type login_info_signature: :return: :rtype: """ verify_key = nacl.signing.VerifyKey(server_signature, encoder=nacl.encoding.HexEncoder) verify_key.verify(login_info.encode(), binascii.unhexlify(login_info_signature)) def decrypt_symmetric(text_hex, nonce_hex, secret): """ Decryts an encrypted text with nonce with the given secret :param text_hex: :type text_hex: :param nonce_hex: :type nonce_hex: :param secret: :type secret: :return: :rtype: """ text = nacl.encoding.HexEncoder.decode(text_hex) nonce = nacl.encoding.HexEncoder.decode(nonce_hex) secret_box = nacl.secret.SecretBox(secret, encoder=nacl.encoding.HexEncoder) return secret_box.decrypt(text, nonce) def encrypt_symmetric(msg, secret): """ Encrypts a message with a random nonce and a given secret :param msg: The message as str :type msg: str :param secret: The secret as hex encoded str :type secret: str :return: A dict, containing the encrypted message with the text and nonce being returned separately :rtype: dict """ # generate random nonce nonce = nacl.utils.random(nacl.secret.SecretBox.NONCE_SIZE) # open crypto box with session secret secret_box = nacl.secret.SecretBox(secret, encoder=nacl.encoding.HexEncoder) # encrypt msg with crypto box and nonce encrypted = secret_box.encrypt(msg.encode(), nonce) # cut away the nonce text = encrypted[len(nonce):] # convert nonce and encrypted msg to hex nonce_hex = nacl.encoding.HexEncoder.encode(nonce).decode() text_hex = nacl.encoding.HexEncoder.encode(text).decode() return {'text': text_hex, 'nonce': nonce_hex} def decrypt_with_api_secret_key(secret_hex, secret_nonce_hex): """ take anything that is encrypted with the api keys secret and decrypts it. e.g. the users secret and private key :param secret_hex: :type secret_hex: :param secret_nonce_hex: :type secret_nonce_hex: :return: :rtype: """ return decrypt_symmetric(secret_hex, secret_nonce_hex, api_key_secret_key) def api_request(method, endpoint, data = None, token = None, session_secret_key = None): if token: headers = {'content-type': 'application/json', 'authorization': 'Token ' + token} else: headers = {'content-type': 'application/json'} r = requests.request(method, server_url + endpoint, data=data, headers=headers, verify=SSL_VERIFY) if not session_secret_key: return r.json() else: encrypted_content = r.json() decrypted_content = decrypt_symmetric(encrypted_content['text'], encrypted_content['nonce'], session_secret_key) return json.loads(decrypted_content) def api_login(client_login_info): """ API Request: Sends the actual login :param client_login_info: :type client_login_info: :return: :rtype: """ method = 'POST' endpoint = '/api-key/login/' data = json.dumps(client_login_info) return api_request(method, endpoint, data) def api_read_datastores(token, session_secret_key): """ Reads all datastores :param token: :type token: :param session_secret_key: :type session_secret_key: :return: :rtype: """ method = 'GET' endpoint = '/datastore/' return api_request(method, endpoint, token=token, session_secret_key=session_secret_key) def api_logout(token, session_secret_key): """ Destroys the session again. :param token: :type token: :param session_secret_key: :type session_secret_key: :return: :rtype: """ method = 'POST' endpoint = '/authentication/logout/' return api_request(method, endpoint, token=token, session_secret_key=session_secret_key) def api_read_datastore(token, session_secret_key, datastore_id): """ Reads the content of a specific datastore :param token: :type token: :param session_secret_key: :type session_secret_key: :param datastore_id: :type datastore_id: :return: :rtype: """ method = 'GET' endpoint = '/datastore/' + datastore_id + '/' return api_request(method, endpoint, token=token, session_secret_key=session_secret_key) def api_read_secret(token, session_secret_key, secret_id): """ Reads the content of a specific secret :param token: :type token: :param session_secret_key: :type session_secret_key: :param secret_id: :type secret_id: :return: :rtype: """ method = 'GET' endpoint = '/secret/' + secret_id + '/' return api_request(method, endpoint, token=token, session_secret_key=session_secret_key) def api_read_share(token, session_secret_key, share_id): """ Reads the content of a specific share :param token: :type token: :param session_secret_key: :type session_secret_key: :param secret_id: :type secret_id: :return: :rtype: """ method = 'GET' endpoint = '/share/' + share_id + '/' return api_request(method, endpoint, token=token, session_secret_key=session_secret_key) def get_all_shares(folder, token, session_secret_key): def handle_share(share): share_read_result = api_read_share(token, session_secret_key, share['share_id']) share_content = None try: share_content = json.loads(decrypt_symmetric(share_read_result['data'], share_read_result['data_nonce'], share['share_secret_key'])) except: pass return share_content if "folders" in folder: new_folders = [] for f in folder["folders"]: if 'share_id' in f and 'share_secret_key' in f: f = handle_share(f) if f is None: continue new_folders.append(f) get_all_shares(f, token, session_secret_key) folder["folders"] = new_folders if "items" in folder: new_items = [] for i in folder["items"]: if 'share_id' in i and 'share_secret_key' in i: i = handle_share(i) if i is None: continue new_items.append(i) folder["items"] = new_items def get_all_secrets(datastore, token, session_secret_key, include_trash_bin_items=False): def handle_items(items): for item in items: if not "secret_id" in item: continue if not "secret_key" in item: continue if not include_trash_bin_items and 'deleted' in item and item['deleted']: continue secret_read_result = api_read_secret(token, session_secret_key, item['secret_id']) if 'non_field_errors' in secret_read_result: # we skip over entries that don't exist or that we have lost permission for continue secret_content = json.loads(decrypt_symmetric(secret_read_result['data'], secret_read_result['data_nonce'], item['secret_key'])) for key in secret_content.keys(): item[key] = secret_content[key] item["create_date"] = secret_read_result["create_date"] item["write_date"] = secret_read_result["write_date"] item["callback_url"] = secret_read_result["callback_url"] item["callback_user"] = secret_read_result["callback_user"] item["callback_pass"] = secret_read_result["callback_pass"] def handle_folders(folders): for folder in folders: if "folders" in folder: handle_folders(folder["folders"]) if "items" in folder: handle_items(folder["items"]) if "folders" in datastore: handle_folders(datastore["folders"]) if "items" in datastore: handle_items(datastore["items"]) def filter_datastore_export(folder, include_trash_bin_items=False): unwanted_folder_properties = [ "id", "datastore_id", "is_folder", "parent_datastore_id", "share_index", "parent_share_id", "share_id", "path", "share_rights", "share_secret_key", ] unwanted_item_properties = [ "id", "datastore_id", "is_folder", "parent_datastore_id", "parent_share_id", "secret_id", "secret_key", "share_id", "path", "share_rights", "share_secret_key", ] # filter out unwanted folder properties for p in unwanted_folder_properties: if p in folder: del folder[p] # Delete items that have been marked as deleted if includeTrashBinItems is not set if "items" in folder and not include_trash_bin_items: folder['items'] = [i for i in folder['items'] if 'deleted' not in i or not i['deleted']] # Delete items attribute if its empty if "items" in folder and len(folder['items']) == 0: del folder['items'] # filter out unwanted item properties if "items" in folder: for p in unwanted_item_properties: for item in folder["items"]: if p in item: del item[p] # Delete folders that have been marked as deleted if includeTrashBinItems is not set if "folders" in folder and not include_trash_bin_items: folder['folders'] = [f for f in folder['folders'] if 'deleted' not in f or not f['deleted']] # Delete folders attribute if its empty if "folders" in folder and len(folder['folders']) == 0: del folder['folders'] # filter folders recursive if "folders" in folder: folder["folders"] = [filter_datastore_export(f) for f in folder["folders"]] return folder def main(): # 1. Generate the login info including the private key for PFS session_private_key, client_login_info = generate_client_login_info() # 2. Send the login request and handle eventual exceptions, problems and so on ... json_response = api_login(client_login_info) # 3. Verify the signature in order to proof that we are really communicating with the server # (or someone who is in the posession of the servers private key :D) verify_signature(json_response['login_info'], json_response['login_info_signature']) # 4. Decrypt the actual login info with the token and session_secret_key for the transport encryption decrypted_sever_login_info = decrypt_server_login_info(json_response['login_info'], json_response['login_info_nonce'], json_response['server_session_public_key'], session_private_key) token = decrypted_sever_login_info['token'] # That is the token that we have to send always as header session_secret_key = decrypted_sever_login_info['session_secret_key'] # that is the symmetric secret for the transport encryption user_username = decrypted_sever_login_info['user']['username'] # The username user_public_key = decrypted_sever_login_info['user']['public_key'] # The user's public key if decrypted_sever_login_info['api_key_restrict_to_secrets']: print("api key is restricted. it should only be used to read specific secrets") return # if the api key is unrestricted then the request will also return the encrypted secret and private key of the user, symmetric encrypted with the api secret key user_private_key = decrypt_with_api_secret_key(decrypted_sever_login_info['user']['private_key'], decrypted_sever_login_info['user']['private_key_nonce']) # The user's private key user_secret_key = decrypt_with_api_secret_key(decrypted_sever_login_info['user']['secret_key'], decrypted_sever_login_info['user']['secret_key_nonce']) # The user's secret key # 5. Now we can start actual reading the datastore and secrets e.g. to read the datastore: content = api_read_datastores(token, session_secret_key) # 6. Read content of the first password datastore including all its shares, all secrets and filter the secrets datastore_content = None for datastore in content['datastores']: if datastore['type'] != 'password': continue datastore_read_result = api_read_datastore(token, session_secret_key, datastore['id']) datastore_secret = decrypt_symmetric(datastore_read_result['secret_key'], datastore_read_result['secret_key_nonce'], user_secret_key) datastore_content = json.loads(decrypt_symmetric(datastore_read_result['data'], datastore_read_result['data_nonce'], datastore_secret)) break # 7. Reads all shares recursively get_all_shares(datastore_content, token, session_secret_key) # 8. Reads all the secrets get_all_secrets(datastore_content, token, session_secret_key) # 9. Filter the datastore content to remove unnecessary data datastore_content = filter_datastore_export(datastore_content) # 10. Logout api_logout(token, session_secret_key) # 11. Write export.json with open('export.json', 'w') as outfile: json.dump(datastore_content, outfile) if __name__ == '__main__': main()