Move collaboration to core

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()
+