diff options
Diffstat (limited to 'python/collaboration/examples')
| -rw-r--r-- | python/collaboration/examples/download_everything.py | 66 | ||||
| -rw-r--r-- | python/collaboration/examples/multitool.py | 297 | ||||
| -rw-r--r-- | python/collaboration/examples/sync_test.py | 84 | ||||
| -rw-r--r-- | python/collaboration/examples/upload_everything.py | 127 |
4 files changed, 574 insertions, 0 deletions
diff --git a/python/collaboration/examples/download_everything.py b/python/collaboration/examples/download_everything.py new file mode 100644 index 00000000..a3f908aa --- /dev/null +++ b/python/collaboration/examples/download_everything.py @@ -0,0 +1,66 @@ +# coding=utf-8 +# Copyright (c) 2015-2024 Vector 35 Inc +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN 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. + +from pathlib import Path +import sys + +import binaryninja +import binaryninja.enterprise as enterprise +import binaryninja.collaboration as collaboration + + +def main(): + with enterprise.LicenseCheckout(): + # Connect to remote from Enterprise + remote = collaboration.enterprise_remote() + if not remote: + return + if not remote.is_connected: + # Will pull default credentials from Enterprise + remote.connect() + + # Pull every file from every project + for project in remote.projects: + for file in project.files: + bndb_path = file.default_bndb_path + print(f"{project.name}/{file.name} BNDB at {bndb_path}") + + try: + metadata: binaryninja.FileMetadata = file.download_to_bndb(bndb_path) + + for v in metadata.existing_views: + if v == 'Raw': + continue + bv = metadata.get_view_of_type(v) + # Show the entry point to demonstrate we have pulled the analyzed file + print(f"{project.name}/{file.name} {v} Entrypoint @ 0x{bv.entry_point:08x}") + except InterruptedError as e: + # In case of ^C + raise e + + except Exception as e: + # In case download or open fails + print(f"{project.name}/{file.name} Load failed: {e}", file=sys.stderr) + + +if __name__ == '__main__': + main() + diff --git a/python/collaboration/examples/multitool.py b/python/collaboration/examples/multitool.py new file mode 100644 index 00000000..e2ea3360 --- /dev/null +++ b/python/collaboration/examples/multitool.py @@ -0,0 +1,297 @@ +# coding=utf-8 +# Copyright (c) 2015-2024 Vector 35 Inc +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN 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. + +import argparse +import os +import sys + +import binaryninja.enterprise as enterprise +import binaryninja.collaboration as collaboration +from binaryninja.collaboration import Remote, RemoteProject, RemoteFile + + +def print_remote_info(remote: Remote, extended=False): + try: + if not remote.is_connected: + remote.connect() + except: + pass + if extended: + print(f"Name: {remote.name}\n" + f"Address: {remote.address}" + ) + if remote.is_connected: + print(f"Id: {remote.unique_id}\n" + f"Is Enterprise: {remote.is_enterprise}\n" + f"Connected username: {remote.username}\n" + f"Is administrator: {remote.is_admin}\n" + f"Server version: {remote.server_version}\n" + f"Server build id: {remote.server_build_id}\n" + f"Project count: {len(remote.projects)}") + if remote.is_admin: + print(f"Group count: {len(remote.groups)}\n" + f"User count: {len(remote.users)}") + else: + if remote.is_connected: + print(f"{remote.name} @ {remote.address} (id: {remote.unique_id})") + else: + print(f"{remote.name} @ {remote.address} (id: ???)") + +def print_project_info(project: RemoteProject, extended=False): + if extended: + print(f"Name: {project.name}\n" + f"Description: {project.description}\n" + f"Id: {project.id}\n" + f"Url: {project.url}\n" + f"Created date: {project.created}\n" + f"Last modified date: {project.last_modified}\n" + f"Default path on disk: {project.default_path}\n" + f"Is admin: {project.is_admin}\n" + f"File count: {len(project.files)}") + else: + print(f"{project.remote.name}/{project.name} @ {project.url} (id: {project.id})") + +def print_file_info(file: RemoteFile, extended=False): + if extended: + print(f"Name: {file.name}\n" + f"Description: {file.description}\n" + f"Id: {file.id}\n" + f"Url: {file.url}\n" + f"Created date: {file.created}\n" + f"Last modified date: {file.last_modified}\n" + f"Last snapshot date: {file.last_snapshot}\n" + f"Contents hash: {file.hash}\n" + f"Contents size: {file.size}\n" + f"Default path on disk: {file.default_bndb_path}\n" + f"Snapshot count: {len(file.snapshots)}") + else: + print(f"{file.remote.name}/{file.project.name}/{file.name} @ {file.url} (id: {file.id})") + + +def main(): + parser = argparse.ArgumentParser(prog=sys.argv[0]) + subparsers = parser.add_subparsers(dest='command', help="Commands", required=True) + + remote_parser = subparsers.add_parser("remote", help="Remote-specific commands") + remote_subparsers = remote_parser.add_subparsers(dest='remote_command', help="Remote-specific commands", required=True) + + remote_add_parser = remote_subparsers.add_parser("add", help="Add new Remote") + remote_add_parser.add_argument("name", type=str) + remote_add_parser.add_argument("address", type=str) + + remote_remove_parser = remote_subparsers.add_parser("remove", help="Remove existing Remote") + remote_remove_parser.add_argument("name", type=str) + + remote_list_parser = remote_subparsers.add_parser("list", help="List existing Remotes") + + remote_info_parser = remote_subparsers.add_parser("info", help="Lookup remote information") + remote_info_parser.add_argument("name", type=str) + + project_parser = subparsers.add_parser("project", help="Project-specific commands") + project_subparsers = project_parser.add_subparsers(dest='project_command', help="Project-specific commands", required=True) + + project_create_parser = project_subparsers.add_parser("create", help="Create new Project") + project_create_parser.add_argument("remote", type=str) + project_create_parser.add_argument("name", type=str) + project_create_parser.add_argument("description", type=str) + + project_delete_parser = project_subparsers.add_parser("delete", help="Delete existing Project") + project_delete_parser.add_argument("remote", type=str) + project_delete_parser.add_argument("name", type=str) + + project_list_parser = project_subparsers.add_parser("list", help="List Projects in Remote") + project_list_parser.add_argument("remote", type=str) + + project_info_parser = project_subparsers.add_parser("info", help="Lookup Project information") + project_info_parser.add_argument("remote", type=str) + project_info_parser.add_argument("name", type=str) + + file_parser = subparsers.add_parser("file", help="File-specific commands") + file_subparsers = file_parser.add_subparsers(dest='file_command', help="File-specific commands", required=True) + + file_upload_parser = file_subparsers.add_parser("upload", help="Upload new File") + file_upload_parser.add_argument("remote", type=str) + file_upload_parser.add_argument("project", type=str) + file_upload_parser.add_argument("filepath", type=str) + file_upload_parser.add_argument("description", type=str) + + file_download_parser = file_subparsers.add_parser("download", help="Download existing File") + file_download_parser.add_argument("remote", type=str) + file_download_parser.add_argument("project", type=str) + file_download_parser.add_argument("name", type=str) + + file_delete_parser = file_subparsers.add_parser("delete", help="Delete existing File") + file_delete_parser.add_argument("remote", type=str) + file_delete_parser.add_argument("project", type=str) + file_delete_parser.add_argument("name", type=str) + + file_list_parser = file_subparsers.add_parser("list", help="List Files in Project") + file_list_parser.add_argument("remote", type=str) + file_list_parser.add_argument("project", type=str) + + file_info_parser = file_subparsers.add_parser("info", help="Lookup File information") + file_info_parser.add_argument("remote", type=str) + file_info_parser.add_argument("project", type=str) + file_info_parser.add_argument("name", type=str) + + args = parser.parse_args(sys.argv[1:]) + if args.command == "remote": + if args.remote_command == "add": + remote = Remote(args.name, args.address) + collaboration.add_known_remote(remote) + collaboration.save_remotes() + print_remote_info(remote) + elif args.remote_command == "remove": + remote = collaboration.get_remote_by_name(args.name) + if remote is None: + print(f"Unknown Remote: {args.name}") + sys.exit(1) + collaboration.remove_known_remote(remote) + collaboration.save_remotes() + print_remote_info(remote) + elif args.remote_command == "list": + for remote in collaboration.known_remotes(): + print_remote_info(remote) + elif args.remote_command == "info": + remote = collaboration.get_remote_by_name(args.name) + if remote is None: + print(f"Unknown Remote: {args.name}") + sys.exit(1) + print_remote_info(remote, extended=True) + elif args.command == "project": + # Connect to remote + remote = collaboration.get_remote_by_name(args.remote) + if remote is None: + print(f"Unknown Remote: {args.remote}") + sys.exit(1) + + try: + remote.connect() + except RuntimeError as e: + print(f"Could not connect to Remote {args.remote}: {e}") + sys.exit(1) + + if args.project_command == "create": + project = remote.create_project(args.name, args.description) + print_project_info(project) + elif args.project_command == "delete": + project = remote.get_project_by_name(args.name) + if project is None: + print(f"Unknown Project: {args.name}") + sys.exit(1) + remote.delete_project(project) + print_project_info(project) + elif args.project_command == "list": + for project in remote.projects: + print_project_info(project) + elif args.project_command == "info": + project = remote.get_project_by_name(args.name) + if project is None: + print(f"Unknown Project: {args.name}") + sys.exit(1) + print_project_info(project, extended=True) + elif args.command == "file": + # Connect to remote + remote = collaboration.get_remote_by_name(args.remote) + if remote is None: + print(f"Unknown Remote: {args.remote}") + sys.exit(1) + + try: + remote.connect() + except RuntimeError as e: + print(f"Could not connect to Remote {args.remote}: {e}") + sys.exit(1) + + project = remote.get_project_by_name(args.project) + if project is None: + print(f"Unknown Project: {args.project}") + sys.exit(1) + + if args.file_command == "upload": + with TqdmProgress(desc="", leave=False) as t: + file = project.upload_new_file(args.filepath, progress=lambda cur, max: t.progress(cur, max)) + file.description = args.description + print_file_info(file) + elif args.file_command == "download": + file = project.get_file_by_name(args.name) + if file is None: + print(f"Unknown File: {args.name}") + sys.exit(1) + with TqdmProgress(desc="", leave=False) as t: + file.download(lambda cur, max: t.progress(cur, max)) + print_file_info(file) + elif args.file_command == "delete": + file = project.get_file_by_name(args.name) + if file is None: + print(f"Unknown File: {args.name}") + sys.exit(1) + print_file_info(file) + project.delete_file(file) + elif args.file_command == "list": + for file in project.files: + print_file_info(file) + elif args.file_command == "info": + file = project.get_file_by_name(args.name) + if file is None: + print(f"Unknown File: {args.name}") + sys.exit(1) + print_file_info(file, extended=True) + else: + pass + + +try: + from tqdm import tqdm + + class TqdmProgress(tqdm): + def __init__(self, *args, **kwargs): + kwargs['total'] = 100 + super(TqdmProgress, self).__init__(*args, **kwargs) + + def progress(self, cur, max): + new_n = cur / max * self.total + self.update(new_n - self.n) + return True + +except ImportError: + # Stub class + class TqdmProgress: + def __init__(self, *args, **kwargs): + pass + + def __enter__(self): + return TqdmProgress() + + def __exit__(self, exc_type, exc_val, exc_tb): + pass + + def progress(self, cur, max): + pass + + +if __name__ == '__main__': + os.environ["BN_DISABLE_USER_PLUGINS"] = "True" + + # Short checkout and don't release, in case you want to run this many times in a row + with enterprise.LicenseCheckout(duration=120, release=False): + main() + diff --git a/python/collaboration/examples/sync_test.py b/python/collaboration/examples/sync_test.py new file mode 100644 index 00000000..6b81d03c --- /dev/null +++ b/python/collaboration/examples/sync_test.py @@ -0,0 +1,84 @@ +# coding=utf-8 +# Copyright (c) 2015-2024 Vector 35 Inc +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN 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. + +import shutil +from pathlib import Path + +import binaryninja +import binaryninja.enterprise as enterprise +import binaryninja.collaboration as collaboration + + +def main(): + with enterprise.LicenseCheckout(): + # Connect to remote from Enterprise + remote = collaboration.enterprise_remote() + if not remote: + return + if not remote.is_connected: + # Will pull default credentials from Enterprise + remote.connect() + + # Create test project + project = remote.create_project("Test Project", "Test project for test purposes") + file = None + project_dir = None + try: + print(f'Created project {project.name}') + file = project.upload_new_file('/bin/ls') + print(f'Created file {project.name}/{file.name}') + + project_dir = Path(file.default_bndb_path).parent + + print(f'Snapshots: {[snapshot.id for snapshot in file.snapshots]}') + bv = binaryninja.load(file.default_bndb_path) + view_type = bv.view_type + assert collaboration.RemoteFile.get_for_bv(bv) == file + + print(f'Setting entry function at 0x{bv.entry_function.start:08x} name to \'entry_function\'') + bv.entry_function.name = 'entry_function' + bv.file.save_auto_snapshot() + + file.sync(bv, lambda conflicts: False) + print(f'Snapshots: {[snapshot.id for snapshot in file.snapshots]}') + + # Try deleting the bndb, redownload and see if the function name is preserved + bv.file.close() + Path(file.default_bndb_path).unlink() + + print(f'Redownloading {project.name}/{file.name}...') + metadata = file.download_to_bndb() + bv = metadata.get_view_of_type(view_type) + print(f'Entry function name: {bv.entry_function.name}') + assert bv.entry_function.name == 'entry_function' + + finally: + # Clean up + if file is not None: + project.delete_file(file) + remote.delete_project(project) + if project_dir is not None: + shutil.rmtree(project_dir) + + +if __name__ == '__main__': + main() + diff --git a/python/collaboration/examples/upload_everything.py b/python/collaboration/examples/upload_everything.py new file mode 100644 index 00000000..b8bef950 --- /dev/null +++ b/python/collaboration/examples/upload_everything.py @@ -0,0 +1,127 @@ +# coding=utf-8 +# Copyright (c) 2015-2024 Vector 35 Inc +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to +# deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +# sell copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN 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. +import hashlib +import os +import traceback +import sys +from pathlib import Path +from typing import Optional + +from tqdm import tqdm + +import binaryninja +import binaryninja.enterprise as enterprise +import binaryninja.collaboration as collaboration + + +def main(): + with enterprise.LicenseCheckout(): + if len(sys.argv) < 2: + print(f"Usage: {sys.argv[0]} <remote name> <project name>") + print("") + print("Here is a list of remotes available to you:") + for remote in sorted(collaboration.known_remotes(), key=lambda remote: remote.name): + print(f"{remote.name}") + sys.exit(1) + + # Connect to remote as specified + remote = collaboration.get_remote_by_name(sys.argv[1]) + if not remote: + return + if not remote.is_connected: + # Will pull default credentials from either Enterprise or Keychain + remote.connect() + + if len(sys.argv) < 3: + print(f"Usage: {sys.argv[0]} <remote name> <project name>") + print("") + print("Here is a list of projects available to you:") + for project in sorted(remote.projects, key=lambda project: project.name): + print(f"{project.name}") + sys.exit(1) + + # Create test project + project = remote.get_project_by_name(sys.argv[2]) + if project is None: + print(f"Creating new project '{sys.argv[2]}'") + project = remote.create_project(sys.argv[2], "") + + known_hashes = set() + project.pull_folders() + for file in project.files: + known_hashes.add(file.hash.lower()) + + # Find all the files in the current directory and upload them + for file in tqdm(list(sorted(Path(os.curdir).rglob("*"))), desc="Files"): + if not file.is_file(): + continue + if file.name == ".DS_Store": + continue + try: + # Check sha256 + with open(file, 'rb') as f: + hash = hashlib.sha256(f.read()).hexdigest() + if hash.lower() in known_hashes: + continue + + # Create parents + base = Path(os.curdir) + parents = [] + parent = file.parent + while parent != base and parent is not None: + parents.insert(0, parent.name) + parent = parent.parent + + def get_folder_by_name(name: str, parent: Optional[collaboration.RemoteFolder]) -> Optional[collaboration.RemoteFolder]: + for folder in project.folders: + if folder.name == name and folder.parent == parent: + return folder + return None + + folder = None + for p in parents: + next_folder = get_folder_by_name(p, folder) + if next_folder is None: + next_folder = project.create_folder(p, "", folder) + folder = next_folder + + tqdm.write(f"Uploading {file}...") + with TqdmProgress(desc="", leave=False) as t: + with binaryninja.load(file, update_analysis=False, progress_func=lambda cur, max: t.progress(cur, max)) as bv: + project.upload_new_file(bv.file, folder, progress=lambda cur, max: t.progress(cur, max)) + except: + tqdm.write(traceback.format_exc()) + + +class TqdmProgress(tqdm): + def __init__(self, *args, **kwargs): + kwargs['total'] = 100 + super(TqdmProgress, self).__init__(*args, **kwargs) + + def progress(self, cur, max): + new_n = cur / max * self.total + self.update(new_n - self.n) + return True + + +if __name__ == '__main__': + main() + |