From 3c28d7e6eaa84041ec0fe5352a618759c13e6cb3 Mon Sep 17 00:00:00 2001 From: Josh Ferrell Date: Mon, 8 Apr 2024 14:13:16 -0400 Subject: Move collaboration to core --- python/CMakeLists.txt | 23 +- python/collaboration/__init__.py | 163 ++++ python/collaboration/changeset.py | 94 +++ python/collaboration/compatibility.py | 21 + python/collaboration/databasesync.py | 310 ++++++++ .../collaboration/examples/download_everything.py | 66 ++ python/collaboration/examples/multitool.py | 297 ++++++++ python/collaboration/examples/sync_test.py | 84 +++ python/collaboration/examples/upload_everything.py | 127 ++++ python/collaboration/file.py | 460 ++++++++++++ python/collaboration/folder.py | 164 ++++ python/collaboration/group.py | 112 +++ python/collaboration/merge.py | 240 ++++++ python/collaboration/permission.py | 159 ++++ python/collaboration/project.py | 826 +++++++++++++++++++++ python/collaboration/remote.py | 722 ++++++++++++++++++ python/collaboration/snapshot.py | 514 +++++++++++++ python/collaboration/user.py | 119 +++ python/collaboration/util.py | 157 ++++ python/generator.cpp | 3 + 20 files changed, 4658 insertions(+), 3 deletions(-) create mode 100644 python/collaboration/__init__.py create mode 100644 python/collaboration/changeset.py create mode 100644 python/collaboration/compatibility.py create mode 100644 python/collaboration/databasesync.py create mode 100644 python/collaboration/examples/download_everything.py create mode 100644 python/collaboration/examples/multitool.py create mode 100644 python/collaboration/examples/sync_test.py create mode 100644 python/collaboration/examples/upload_everything.py create mode 100644 python/collaboration/file.py create mode 100644 python/collaboration/folder.py create mode 100644 python/collaboration/group.py create mode 100644 python/collaboration/merge.py create mode 100644 python/collaboration/permission.py create mode 100644 python/collaboration/project.py create mode 100644 python/collaboration/remote.py create mode 100644 python/collaboration/snapshot.py create mode 100644 python/collaboration/user.py create mode 100644 python/collaboration/util.py (limited to 'python') diff --git a/python/CMakeLists.txt b/python/CMakeLists.txt index 5cbe6b23..96870fe2 100644 --- a/python/CMakeLists.txt +++ b/python/CMakeLists.txt @@ -2,12 +2,17 @@ cmake_minimum_required(VERSION 3.9 FATAL_ERROR) project(python-api) -file(GLOB PYTHON_SOURCES CONFIGURE_DEPENDS ${PROJECT_SOURCE_DIR}/*.py ${PROJECT_SOURCE_DIR}/*.typed) +file(GLOB PYTHON_SOURCES CONFIGURE_DEPENDS + ${PROJECT_SOURCE_DIR}/*.py ${PROJECT_SOURCE_DIR}/*.typed + ${PROJECT_SOURCE_DIR}/collaboration/*.py ${PROJECT_SOURCE_DIR}/collaboration/*.typed +) + list(REMOVE_ITEM PYTHON_SOURCES ${PROJECT_SOURCE_DIR}/_binaryninjacore.py) list(REMOVE_ITEM PYTHON_SOURCES ${PROJECT_SOURCE_DIR}/enums.py) if(NOT ENTERPRISE) list(REMOVE_ITEM PYTHON_SOURCES ${PROJECT_SOURCE_DIR}/enterprise.py) + list(FILTER PYTHON_SOURCES EXCLUDE REGEX ".*/collaboration/.*$") endif() add_executable(generator @@ -26,13 +31,25 @@ if(BN_INTERNAL_BUILD) COMMAND ${CMAKE_COMMAND} -E copy ${BN_CORE_OUTPUT_DIR}/binaryninjacore.dll ${PROJECT_BINARY_DIR}/) endif() + # Generate script to copy python sources with the correct paths + file(GENERATE OUTPUT ${PROJECT_BINARY_DIR}/copy_python_sources.cmake + CONTENT " + foreach(PYTHON_SOURCE ${PYTHON_SOURCES}) + cmake_path(RELATIVE_PATH PYTHON_SOURCE BASE_DIRECTORY ${PROJECT_SOURCE_DIR} OUTPUT_VARIABLE OUTPUT_SUBPATH) + cmake_path(REMOVE_FILENAME OUTPUT_SUBPATH) + file(COPY $\{PYTHON_SOURCE\} DESTINATION ${BN_RESOURCE_DIR}/python/binaryninja/$\{OUTPUT_SUBPATH\}) + endforeach() + " + ) + add_custom_target(generator_copy ALL BYPRODUCTS ${PROJECT_SOURCE_DIR}/_binaryninjacore.py ${PROJECT_SOURCE_DIR}/enums.py DEPENDS ${PYTHON_SOURCES} ${PROJECT_SOURCE_DIR}/../binaryninjacore.h $ COMMENT "Copying API Python sources" COMMAND ${CMAKE_COMMAND} -E make_directory ${BN_RESOURCE_DIR}/python/binaryninja/ COMMAND ${CMAKE_COMMAND} -E env ASAN_OPTIONS=detect_leaks=0 $ ${PROJECT_SOURCE_DIR}/../binaryninjacore.h ${PROJECT_SOURCE_DIR}/_binaryninjacore.py ${PROJECT_SOURCE_DIR}/enums.py - COMMAND ${CMAKE_COMMAND} -E copy ${PYTHON_SOURCES} ${BN_RESOURCE_DIR}/python/binaryninja/ + COMMAND ${CMAKE_COMMAND} -P ${PROJECT_BINARY_DIR}/copy_python_sources.cmake COMMAND ${CMAKE_COMMAND} -E copy ${PROJECT_SOURCE_DIR}/_binaryninjacore.py ${BN_RESOURCE_DIR}/python/binaryninja/ - COMMAND ${CMAKE_COMMAND} -E copy ${PROJECT_SOURCE_DIR}/enums.py ${BN_RESOURCE_DIR}/python/binaryninja/) + COMMAND ${CMAKE_COMMAND} -E copy ${PROJECT_SOURCE_DIR}/enums.py ${BN_RESOURCE_DIR}/python/binaryninja/ + ) endif() diff --git a/python/collaboration/__init__.py b/python/collaboration/__init__.py new file mode 100644 index 00000000..63331d83 --- /dev/null +++ b/python/collaboration/__init__.py @@ -0,0 +1,163 @@ +import binaryninja + +from .changeset import * +from .databasesync import * +from .file import * +from .folder import * +from .group import * +from .project import * +from .remote import * +from .snapshot import * +from .user import * +from .util import _last_error + +""" +Collaboration Remote API +Python wrappers around C++ wrappers around the various REST apis exposed by the collaboration server. +Many methods throw RuntimeError, as documented. + +None of these classes are thread-safe. Python's GIL probably helps with this but if you are +doing heavily multi-threaded work with these you may want the C++ api. +""" + + +def active_remote() -> Optional['Remote']: + """ + Get the single actively connected Remote (for ux simplification) + + :return: Active Remote, if one is set. None, otherwise. + """ + binaryninja._init_plugins() + value = core.BNCollaborationGetActiveRemote() + if value is None: + return None + result = Remote(handle=value) + return result + + +def set_active_remote(remote: Optional['Remote']): + """ + Set the single actively connected Remote + + :param remote: New active Remote, or None to clear it. + """ + binaryninja._init_plugins() + if remote is not None: + core.BNCollaborationSetActiveRemote(remote._handle) + else: + core.BNCollaborationSetActiveRemote(None) + + +def known_remotes() -> List['Remote']: + """ + List of known/connected Remotes + + :return: All known remotes + """ + binaryninja._init_plugins() + count = ctypes.c_size_t() + value = core.BNCollaborationGetRemotes(count) + result = [] + for i in range(count.value): + result.append(Remote(handle=value[i])) + return result + + +def enterprise_remote() -> Optional['Remote']: + """ + Get whichever known Remote has the same address as the Enterprise license server + + :return: Relevant known Remote, or None if one is not found + """ + for remote in known_remotes(): + if remote.is_enterprise: + return remote + return None + + +def add_known_remote(remote: 'Remote') -> None: + """ + Add a Remote to the list of known remotes (saved to Settings) + + :param remote: New Remote to add + """ + binaryninja._init_plugins() + core.BNCollaborationAddRemote(remote._handle) + + +def remove_known_remote(remote: 'Remote') -> None: + """ + Remove a Remote from the list of known remotes (saved to Settings) + + :param remote: Remote to remove + """ + binaryninja._init_plugins() + core.BNCollaborationRemoveRemote(remote._handle) + + +def get_remote_by_id(id: str) -> Optional['Remote']: + """ + Get Remote by unique id + + :param id: Unique id of the Remote + :return: Remote, if known, else None + """ + binaryninja._init_plugins() + value = core.BNCollaborationGetRemoteById(id) + if value is None: + return None + result = Remote(handle=value) + return result + + +def get_remote_by_address(address: str) -> Optional['Remote']: + """ + Get Remote by address + + :param address: Base address of remote api + :return: Remote, if found, else None + """ + binaryninja._init_plugins() + value = core.BNCollaborationGetRemoteByAddress(address) + if value is None: + return None + result = Remote(handle=value) + return result + + +def get_remote_by_name(name: str) -> Optional['Remote']: + """ + Get Remote by name + + :param name: Name of Remote + :return: Remote, if found, else None + """ + binaryninja._init_plugins() + value = core.BNCollaborationGetRemoteByName(name) + if value is None: + return None + result = Remote(handle=value) + return result + + +def load_remotes(): + """ + Load the list of known Remotes from local Settings + + :raises RuntimeError: If there was an error + """ + binaryninja._init_plugins() + if not core.BNCollaborationLoadRemotes(): + raise RuntimeError(_last_error()) + + +def save_remotes(): + """ + Save the list of known Remotes to local Settings + + :raises RuntimeError: If there was an error + """ + binaryninja._init_plugins() + if not core.BNCollaborationSaveRemotes(): + raise RuntimeError(_last_error()) + diff --git a/python/collaboration/changeset.py b/python/collaboration/changeset.py new file mode 100644 index 00000000..931dcbe2 --- /dev/null +++ b/python/collaboration/changeset.py @@ -0,0 +1,94 @@ + +import ctypes +from typing import List + +from .. import _binaryninjacore as core +from ..database import Database +from . import file, user, util + + +class Changeset: + """ + Class representing a collection of snapshots in a local database + """ + def __init__(self, handle: core.BNCollaborationChangesetHandle): + """ + :param handle: FFI handle for internal use + """ + self._handle = ctypes.cast(handle, core.BNCollaborationChangesetHandle) + + def __del__(self): + if core is not None: + core.BNFreeCollaborationChangeset(self._handle) + + @property + def database(self) -> 'Database': + """ + Owning database for snapshots + + :return: Database object + """ + value = core.BNCollaborationChangesetGetDatabase(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + result = Database(handle=value) + return result + + @property + def file(self) -> 'file.RemoteFile': + """ + Relevant remote File object + + :return: File object + """ + value = core.BNCollaborationChangesetGetFile(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return file.RemoteFile(handle=value) + + @property + def snapshot_ids(self) -> List[int]: + """ + List of snapshot ids in the database + + :return: Snapshot id list + """ + count = ctypes.c_size_t() + snapshot_ids = core.BNCollaborationChangesetGetSnapshotIds(self._handle, count) + if snapshot_ids is None: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(snapshot_ids[i]) + core.BNCollaborationFreeSnapshotIdList(snapshot_ids, count.value) + return result + + @property + def author(self) -> 'user.User': + """ + Relevant remote author User + + :return: Author User + """ + value = core.BNCollaborationChangesetGetAuthor(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return user.User(handle=value) + + @property + def name(self) -> str: + """ + Changeset name + + :return: Name string + """ + return core.BNCollaborationChangesetGetName(self._handle) + + @name.setter + def name(self, name: str): + """ + Set the name of the changeset, e.g. in a name changeset function. + + :param name: New changeset name + """ + core.BNCollaborationChangesetSetName(self._handle, name) diff --git a/python/collaboration/compatibility.py b/python/collaboration/compatibility.py new file mode 100644 index 00000000..e723cd69 --- /dev/null +++ b/python/collaboration/compatibility.py @@ -0,0 +1,21 @@ +import warnings + +from .file import RemoteFile +from .folder import RemoteFolder +from .project import RemoteProject + + +class File(RemoteFile): + def __init__(self, handle): + warnings.warn('Legacy class "File" will be removed in a future version. Please migrate to "binaryninja.collaboration.file.RemoteFile".') + super().__init__(handle) + +class Folder(RemoteFolder): + def __init__(self, handle): + warnings.warn('Legacy class "Folder" will be removed in a future version. Please migrate to "binaryninja.collaboration.folder.RemoteFolder".') + super().__init__(handle) + +class Project(RemoteProject): + def __init__(self, handle): + warnings.warn('Legacy class "Project" will be removed in a future version. Please migrate to "binaryninja.collaboration.project.RemoteProject".') + super().__init__(handle) diff --git a/python/collaboration/databasesync.py b/python/collaboration/databasesync.py new file mode 100644 index 00000000..62805e87 --- /dev/null +++ b/python/collaboration/databasesync.py @@ -0,0 +1,310 @@ +import ctypes +from typing import Optional + +from .. import _binaryninjacore as core +from . import file, folder, merge, project, remote, snapshot, util + +from ..database import Database, Snapshot +from ..filemetadata import FileMetadata + +""" +Database syncing and choreography between BN api and remote +""" + + +def default_project_path(project_: 'project.RemoteProject') -> str: + """ + Get the default directory path for a remote Project. This is based off the Setting for + collaboration.directory, the project's id, and the project's remote's id. + + :param project_: Remote Project + :return: Default project path + :raises RuntimeError: If there was an error + """ + value = core.BNCollaborationDefaultProjectPath(project_._handle) + if value is None: + raise RuntimeError(util._last_error()) + return value + + +def default_file_path(file_: 'file.RemoteFile') -> str: + """ + Get the default filepath for a remote File. This is based off the Setting for + collaboration.directory, the file's id, the file's project's id, and the file's + remote's id. + + :param file_: Remote File + :return: Default file path + :raises RuntimeError: If there was an error + """ + value = core.BNCollaborationDefaultFilePath(file_._handle) + if value is None: + raise RuntimeError(util._last_error()) + return value + + +def download_file(file_: 'file.RemoteFile', db_path: str, progress: 'util.ProgressFuncType' = util.nop) -> 'FileMetadata': + """ + Download a file from its remote, saving all snapshots to a database in the + specified location. Returns a FileContext for opening the file later. + + :param file_: Remote File to download and open + :param db_path: File path for saved database + :param progress: Function to call for progress updates + :return: FileContext for opening + :raises RuntimeError: If there was an error + """ + value = core.BNCollaborationDownloadFile(file_._handle, db_path, util.wrap_progress(progress), None) + if value is None: + raise RuntimeError(util._last_error()) + return FileMetadata(handle=ctypes.cast(value, core.BNFileMetadataHandle)) + + +def upload_database(metadata: 'FileMetadata', project: 'project.RemoteProject', parent_folder: Optional['folder.RemoteFolder'] = None, progress: 'util.ProgressFuncType' = util.nop, name_changeset: 'util.NameChangesetFuncType' = util.nop) -> 'file.RemoteFile': + """ + Upload a file, with database, to the remote under the given project + + :param metadata: Local file with database + :param project: Remote project under which to place the new file + :param progress: Function to call for progress updates + :param name_changeset: Function to call for naming a pushed changeset, if necessary + :param parent_folder: Optional parent folder in which to place this file + :return: Remote File created + :raises RuntimeError: If there was an error + """ + folder_handle = parent_folder._handle if parent_folder is not None else None + value = core.BNCollaborationUploadDatabase(ctypes.cast(metadata.handle, core.BNFileMetadataHandle), project._handle, folder_handle, util.wrap_progress(progress), None, util.wrap_name_changeset(name_changeset), None) + if value is None: + raise RuntimeError(util._last_error()) + return file.RemoteFile(handle=value) + + +def is_collaboration_database(database: Database) -> bool: + """ + Test if a database is valid for use in collaboration + + :param database: Database to test + :return: True if valid + """ + return core.BNCollaborationIsCollaborationDatabase(ctypes.cast(database.handle, core.BNDatabaseHandle)) + + +def get_remote_for_local_database(database: Database) -> Optional['remote.Remote']: + """ + Get the Remote for a Database + + :param database: BN database, potentially with collaboration metadata + :return: Remote from one of the connected remotes, or None if not found + :raises RuntimeError: If there was an error + """ + value = core.BNRemoteHandle() + if not core.BNCollaborationGetRemoteForLocalDatabase(ctypes.cast(database.handle, core.BNDatabaseHandle), value): + raise RuntimeError(util._last_error()) + if not value: + return None + return remote.Remote(handle=value) + + +def get_remote_project_for_local_database(database: Database) -> Optional['project.RemoteProject']: + """ + Get the Remote Project for a Database + + :param database: BN database, potentially with collaboration metadata + :return: Remote project from one of the connected remotes, or None if not found + or if projects are not pulled + :raises RuntimeError: If there was an error + """ + value = core.BNRemoteProjectHandle() + if not core.BNCollaborationGetRemoteProjectForLocalDatabase(ctypes.cast(database.handle, core.BNDatabaseHandle), value): + raise RuntimeError(util._last_error()) + if not value: + return None + return project.RemoteProject(handle=value) + + +def get_remote_file_for_local_database(database: Database) -> Optional['file.RemoteFile']: + """ + Get the Remote File for a Database + + :param database: BN database, potentially with collaboration metadata + :return: Remote file from one of the connected remotes, or None if not found + or if files are not pulled + :raises RuntimeError: If there was an error + """ + value = core.BNRemoteFileHandle() + if not core.BNCollaborationGetRemoteFileForLocalDatabase(ctypes.cast(database.handle, core.BNDatabaseHandle), value): + raise RuntimeError(util._last_error()) + if not value: + return None + return file.RemoteFile(handle=value) + + +def assign_snapshot_map(local_snapshot: Snapshot, remote_snapshot: snapshot.CollabSnapshot): + """ + Add a snapshot to the id map in a database + + :param local_snapshot: Local snapshot, will use this snapshot's database + :param remote_snapshot: Remote snapshot + :raises RuntimeError: If there was an error + """ + if not core.BNCollaborationAssignSnapshotMap(ctypes.cast(local_snapshot.handle, core.BNSnapshotHandle), remote_snapshot._handle): + raise RuntimeError(util._last_error()) + + +def get_remote_snapshot_for_local(snap: Snapshot) -> Optional['snapshot.CollabSnapshot']: + """ + Get the remote snapshot associated with a local snapshot (if it exists) + + :param snap: Local snapshot + :return: Remote snapshot if it exists, or None if not + :raises RuntimeError: If there was an error + """ + value = core.BNCollaborationSnapshotHandle() + if not core.BNCollaborationGetRemoteSnapshotFromLocal(ctypes.cast(snap.handle, core.BNSnapshotHandle), value): + raise RuntimeError(util._last_error()) + if not value: + return None + return snapshot.CollabSnapshot(handle=value) + + +def get_local_snapshot_for_remote(snapshot: snapshot.CollabSnapshot, database: Database) -> Optional['Snapshot']: + """ + Get the local snapshot associated with a remote snapshot (if it exists) + + :param snapshot: Remote snapshot + :param database: Local database to search + :return: Snapshot reference if it exists, or None reference if not + :raises RuntimeError: If there was an error + """ + value = core.BNSnapshotHandle() + if not core.BNCollaborationGetLocalSnapshotFromRemote(snapshot._handle, ctypes.cast(database.handle, core.BNSnapshotHandle), value): + raise RuntimeError(util._last_error()) + if not value: + return None + return Snapshot(handle=ctypes.cast(value, ctypes.POINTER(core.BNSnapshot))) + + +def sync_database(database: Database, file_: 'file.RemoteFile', conflict_handler: 'util.ConflictHandlerType', progress: 'util.ProgressFuncType' = util.nop, name_changeset: 'util.NameChangesetFuncType' = util.nop): + """ + Completely sync a database, pushing/pulling/merging/applying changes + + :param database: Database to sync + :param file_: File to sync with + :param conflict_handler: Function to call to resolve snapshot conflicts + :param progress: Function to call for progress updates + :param name_changeset: Function to call for naming a pushed changeset, if necessary + :raises RuntimeError: If there was an error (or the operation was cancelled) + """ + + if not core.BNCollaborationSyncDatabase(ctypes.cast(database.handle, core.BNDatabaseHandle), file_._handle, util.wrap_conflict_handler(conflict_handler), None, util.wrap_progress(progress), None, util.wrap_name_changeset(name_changeset), None): + raise RuntimeError(util._last_error()) + + +def pull_database(database: Database, file_: 'file.RemoteFile', conflict_handler: 'util.ConflictHandlerType', progress: 'util.ProgressFuncType' = util.nop, name_changeset: 'util.NameChangesetFuncType' = util.nop): + """ + Pull updated snapshots from the remote. Merge local changes with remote changes and + potentially create a new snapshot for unsaved changes, named via name_changeset. + + :param database: Database to pull + :param file_: Remote File to pull to + :param conflict_handler: Function to call to resolve snapshot conflicts + :param progress: Function to call for progress updates + :param name_changeset: Function to call for naming a pushed changeset, if necessary + :raises RuntimeError: If there was an error (or the operation was cancelled) + """ + count = ctypes.c_ulonglong() + if not core.BNCollaborationPullDatabase(ctypes.cast(database.handle, core.BNDatabaseHandle), file_._handle, count, util.wrap_conflict_handler(conflict_handler), None, util.wrap_progress(progress), None, util.wrap_name_changeset(name_changeset), None): + raise RuntimeError(util._last_error()) + + +def merge_database(database: Database, conflict_handler: 'util.ConflictHandlerType', progress: 'util.ProgressFuncType' = util.nop): + """ + Merge all leaf snapshots in a database down to a single leaf snapshot. + + :param database: Database to merge + :param conflict_handler: Function to call for progress updates + :param progress: Function to call to resolve snapshot conflicts + :raises RuntimeError: If there was an error (or the operation was cancelled) + """ + if not core.BNCollaborationMergeDatabase(ctypes.cast(database.handle, core.BNDatabaseHandle), util.wrap_conflict_handler(conflict_handler), None, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + +def push_database(database: Database, file_: 'file.RemoteFile', progress: 'util.ProgressFuncType' = util.nop): + """ + Push locally added snapshots to the remote + + :param database: Database to push + :param file_: Remote File to push to + :param progress: Function to call for progress updates + :raises RuntimeError: If there was an error (or the operation was cancelled) + """ + count = ctypes.c_ulonglong() + if not core.BNCollaborationPushDatabase(ctypes.cast(database.handle, core.BNDatabaseHandle), file_._handle, count, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + +def dump_database(database: Database): + """ + Print debug information about a database to stdout + + :param database: Database to dump + :raises RuntimeError: If there was an error + """ + if not core.BNCollaborationDumpDatabase(ctypes.cast(database.handle, core.BNDatabaseHandle)): + raise RuntimeError(util._last_error()) + + +def ignore_snapshot(database: Database, snapshot: Snapshot): + """ + Ignore a snapshot from database syncing operations + + :param database: Parent database + :param snapshot: Snapshot to ignore + :raises RuntimeError: If there was an error + """ + if not core.BNCollaborationIgnoreSnapshot(ctypes.cast(database.handle, core.BNDatabaseHandle), ctypes.cast(snapshot.handle, core.BNSnapshotHandle)): + raise RuntimeError(util._last_error()) + + +def is_snapshot_ignored(database: Database, snapshot: Snapshot) -> bool: + """ + Test if a snapshot is ignored from the database + + :param database: Parent database + :param snapshot: Snapshot to test + :return: True if snapshot should be ignored + :raises RuntimeError: If there was an error + """ + return core.BNCollaborationIsSnapshotIgnored(ctypes.cast(database.handle, core.BNDatabaseHandle), ctypes.cast(snapshot.handle, core.BNSnapshotHandle)) + + +def get_snapshot_author(database: Database, snapshot: Snapshot) -> Optional[str]: + """ + Get the remote author of a local snapshot + + :param database: Parent database + :param snapshot: Snapshot to query + :return: Remote author, or None if one could not be determined + :raises RuntimeError: If there was an error + """ + value = ctypes.POINTER(ctypes.c_char_p)() + if not core.BNCollaborationGetSnapshotAuthor(ctypes.cast(database.handle, core.BNDatabaseHandle), ctypes.cast(snapshot.handle, core.BNSnapshotHandle), value): + raise RuntimeError(util._last_error()) + if value is None: + return None + return core.pyNativeStr(value) + + +def set_snapshot_author(database: Database, snapshot: Snapshot, author: str): + """ + Set the remote author of a local snapshot (does not upload) + + :param database: Parent database + :param snapshot: Snapshot to edit + :param author: Target author + :raises RuntimeError: If there was an error + """ + if not core.BNCollaborationSetSnapshotAuthor(ctypes.cast(database.handle, core.BNDatabaseHandle), ctypes.cast(snapshot.handle, core.BNSnapshotHandle), author): + raise RuntimeError(util._last_error()) + 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]} ") + 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]} ") + 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() + diff --git a/python/collaboration/file.py b/python/collaboration/file.py new file mode 100644 index 00000000..0cc3717d --- /dev/null +++ b/python/collaboration/file.py @@ -0,0 +1,460 @@ +import ctypes +import datetime +from typing import List, Optional, Union + +from .. import _binaryninjacore as core +from .. import enums +from . import databasesync +from . import folder as _folder +from . import project, remote, snapshot, util + + +from ..binaryview import BinaryView +from ..database import Database +from ..filemetadata import FileMetadata +from ..project import ProjectFile + + +class RemoteFile: + """ + Class representing a remote project file. It controls the various + snapshots and raw file contents associated with the analysis. + """ + def __init__(self, handle: core.BNRemoteFileHandle): + self._handle = ctypes.cast(handle, core.BNRemoteFileHandle) + + def __del__(self): + if core is not None: + core.BNFreeRemoteFile(self._handle) + + def __eq__(self, other): + if not isinstance(other, RemoteFile): + return False + return other.id == self.id + + def __str__(self): + path = self.name + parent = self.folder + while parent is not None: + path = parent.name + '/' + path + parent = parent.parent + return f'' + + def __repr__(self): + path = self.name + parent = self.folder + while parent is not None: + path = parent.name + '/' + path + parent = parent.parent + return f'' + + @staticmethod + def get_for_local_database(database: 'Database') -> Optional['RemoteFile']: + """ + Look up the remote File for a local database, or None if there is no matching + remote File found. + See :func:`get_for_bv` to load from a BinaryView. + + :param database: Local database + :return: Remote File object + :rtype: File or None + """ + remote = databasesync.get_remote_for_local_database(database) + if remote is None: + return None + if not remote.has_pulled_projects: + remote.pull_projects() + project = databasesync.get_remote_project_for_local_database(database) + if project is None: + return None + if not project.has_pulled_files: + project.pull_files() + return databasesync.get_remote_file_for_local_database(database) + + @staticmethod + def get_for_bv(bv: 'BinaryView') -> Optional['RemoteFile']: + """ + Look up the remote File for a local BinaryView, or None if there is no matching + remote File found. + + :param bv: Local BinaryView + :return: Remote File object + :rtype: File or None + """ + if not bv.file.has_database: + return None + return RemoteFile.get_for_local_database(bv.file.database) + + @property + def core_file(self) -> 'ProjectFile': + core_handle = core.BNRemoteFileGetCoreFile(self._handle) + if core_handle is None: + raise RuntimeError(util._last_error()) + return ProjectFile(handle=ctypes.cast(core_handle, ctypes.POINTER(core.BNProjectFile))) + + @property + def project(self) -> 'project.RemoteProject': + """ + Owning Project + + :return: Project object + """ + value = core.BNRemoteFileGetProject(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return project.RemoteProject(handle=value) + + @property + def remote(self) -> 'remote.Remote': + """ + Owning Remote + + :return: Remote object + """ + value = core.BNRemoteFileGetRemote(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return remote.Remote(handle=value) + + @property + def folder(self) -> Optional['_folder.RemoteFolder']: + """ + Parent folder, if one exists. None if this is in the root of the project. + + :return: Folder object or None + """ + if not self.project.has_pulled_folders: + self.project.pull_folders() + value = core.BNRemoteFileGetFolder(self._handle) + if value is None: + return None + return _folder.RemoteFolder(handle=value) + + @folder.setter + def folder(self, folder: Optional['_folder.RemoteFolder']): + """ + Set the parent folder of a file. + + :param folder: New parent folder, or None to move file to the root of the project. + """ + folder_handle = folder._handle if folder is not None else None + if not core.BNRemoteFileSetFolder(self._handle, folder_handle): + raise RuntimeError(util._last_error()) + + @property + def url(self) -> str: + """ + Web api endpoint URL + + :return: URL string + """ + return core.BNRemoteFileGetUrl(self._handle) + + @property + def chat_log_url(self) -> str: + """ + Chat log api endpoint URL + + :return: URL string + """ + return core.BNRemoteFileGetChatLogUrl(self._handle) + + @property + def id(self) -> str: + """ + Unique id + + :return: Id string + """ + return core.BNRemoteFileGetId(self._handle) + + @property + def type(self) -> enums.RemoteFileType: + """ + File Type + All files share the same properties, but files with different types may make different + uses of those properties, or not use some of them at all. + + :return: Type of file on server (enum) + """ + return enums.RemoteFileType(core.BNRemoteFileGetType(self._handle)) + + @property + def created(self) -> datetime.datetime: + """ + Created date of the file + + :return: Date object + """ + return datetime.datetime.utcfromtimestamp(core.BNRemoteFileGetCreated(self._handle)) + + @property + def last_modified(self) -> datetime.datetime: + """ + Last modified date of the file + + :return: Date object + """ + return datetime.datetime.utcfromtimestamp(core.BNRemoteFileGetLastModified(self._handle)) + + @property + def last_snapshot(self) -> datetime.datetime: + """ + Date of last snapshot in the file + + :return: Date object + """ + return datetime.datetime.utcfromtimestamp(core.BNRemoteFileGetLastSnapshot(self._handle)) + + @property + def last_snapshot_by(self) -> str: + """ + Username of user who pushed the last snapshot in the file + + :return: Username string + """ + return core.BNRemoteFileGetLastSnapshotBy(self._handle) + + @property + def hash(self) -> str: + """ + Hash of file contents (no algorithm guaranteed) + + :return: Hash string + """ + return core.BNRemoteFileGetHash(self._handle) + + @property + def name(self) -> str: + """ + Displayed name of file + + :return: Name string + """ + return core.BNRemoteFileGetName(self._handle) + + @name.setter + def name(self, value: str): + """ + Set the display name of the file. You will need to push the file to update the remote version. + + :param value: New name + """ + if not core.BNRemoteFileSetName(self._handle, value): + raise RuntimeError(util._last_error()) + + @property + def description(self) -> str: + """ + Description of the file + + :return: Description string + """ + return core.BNRemoteFileGetDescription(self._handle) + + @description.setter + def description(self, value: str): + """ + Set the description of the file. You will need to push the file to update the remote version. + + :param description: New description + """ + if not core.BNRemoteFileSetDescription(self._handle, value): + raise RuntimeError(util._last_error()) + + @property + def size(self) -> int: + """ + Size of raw content of file, in bytes + + :return: Size in bytes + """ + return core.BNRemoteFileGetSize(self._handle) + + @property + def default_path(self) -> str: + """ + Get the default filepath for a remote File. This is based off the Setting for + collaboration.directory, the file's id, the file's project's id, and the file's + remote's id. + + :return: Default file path + :rtype: str + """ + return databasesync.default_file_path(self) + + @property + def has_pulled_snapshots(self) -> bool: + """ + If the file has pulled the snapshots yet + + :return: True if they have been pulled + """ + return core.BNRemoteFileHasPulledSnapshots(self._handle) + + @property + def snapshots(self) -> List['snapshot.CollabSnapshot']: + """ + Get the list of snapshots in this file. + + .. note:: If snapshots have not been pulled, they will be pulled upon calling this. + + :return: List of Snapshot objects + :raises: RuntimeError if there was an error pulling snapshots + """ + if not self.has_pulled_snapshots: + self.pull_snapshots() + + count = ctypes.c_size_t() + value = core.BNRemoteFileGetSnapshots(self._handle, count) + if value is None: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(snapshot.CollabSnapshot(value[i])) + return result + + def get_snapshot_by_id(self, id: str) -> Optional['snapshot.CollabSnapshot']: + """ + Get a specific Snapshot in the File by its id + + .. note:: If snapshots have not been pulled, they will be pulled upon calling this. + + :param id: Id of Snapshot + :return: Snapshot object, if one with that id exists. Else, None + :raises: RuntimeError if there was an error pulling snapshots + """ + if not self.has_pulled_snapshots: + self.pull_snapshots() + + value = core.BNRemoteFileGetSnapshotById(self._handle, id) + if value is None: + return None + return snapshot.CollabSnapshot(value) + + def pull_snapshots(self, progress: 'util.ProgressFuncType' = util.nop): + """ + Pull the list of Snapshots from the Remote. + + :param progress: Function to call for progress updates + :raises: RuntimeError if there was an error pulling snapshots + """ + if not core.BNRemoteFilePullSnapshots(self._handle, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + def create_snapshot(self, name: str, contents: bytes, analysis_cache_contents: bytes, file: bytes, parent_ids: List[str], progress: 'util.ProgressFuncType' = util.nop) -> 'snapshot.CollabSnapshot': + """ + Create a new snapshot on the remote (and pull it) + + :param name: Snapshot name + :param contents: Snapshot contents + :param analysis_cache_contents: Contents of analysis cache of snapshot + :param file: New file contents (if contents changed) + :param parent_ids: List of ids of parent snapshots (or empty if this is a root snapshot) + :param progress: Function to call on progress updates + :return: Reference to the created snapshot + :raises: RuntimeError if there was an error + """ + array = (ctypes.c_char_p * len(parent_ids))() + for i in range(len(parent_ids)): + array[i] = parent_ids[i] + + value = core.BNRemoteFileCreateSnapshot(self._handle, name, contents, len(contents), analysis_cache_contents, len(analysis_cache_contents), file, len(file), array, len(parent_ids), util.wrap_progress(progress), None) + if value is None: + raise RuntimeError(util._last_error()) + return snapshot.CollabSnapshot(value) + + def delete_snapshot(self, snapshot: 'snapshot.CollabSnapshot'): + """ + Delete a snapshot from the remote + + :param snapshot: Snapshot to delete + :raises: RuntimeError if there was an error + """ + if not core.BNRemoteFileDeleteSnapshot(self._handle, snapshot._handle): + raise RuntimeError(util._last_error()) + + def download(self, progress: 'util.ProgressFuncType' = util.nop) -> bytes: + """ + Download the contents of a remote file + + :param progress: Function to call on progress updates + :return: Contents of the file + :raises: RuntimeError if there was an error + """ + data = (ctypes.POINTER(ctypes.c_ubyte))() + size = ctypes.c_size_t() + value = core.BNRemoteFileDownload(self._handle, util.wrap_progress(progress), None, data, size) + if not value: + raise RuntimeError(util._last_error()) + return bytes(ctypes.cast(data, ctypes.POINTER(ctypes.c_uint8 * size.value)).contents) + + def download_to_bndb(self, path: Optional[str] = None, progress: 'util.ProgressFuncType' = util.nop) -> FileMetadata: + """ + Download a remote file and save it to a bndb at the given path. + This calls databasesync.download_file and self.sync to fully prepare the bndb. + + :param path: Path to new bndb to create + :param progress: Function to call on progress updates + :return: Constructed FileMetadata object + :raises: RuntimeError if there was an error + """ + if path is None: + path = self.default_path + file = databasesync.download_file(self, path, util.split_progress(progress, 0, [0.5, 0.5])) + self.sync( + file.database, lambda conflicts: False, util.split_progress(progress, 1, [0.5, 0.5])) + return file + + def sync(self, bv_or_db: Union['BinaryView', 'Database'], conflict_handler: 'util.ConflictHandlerType', progress: 'util.ProgressFuncType' = util.nop, name_changeset: 'util.NameChangesetFuncType' = util.nop): + """ + Completely sync a file, pushing/pulling/merging/applying changes + + :param bv_or_db: Binary view or database to sync with + :param conflict_handler: Function to call to resolve snapshot conflicts + :param name_changeset: Function to call for naming a pushed changeset, if necessary + :param progress: Function to call for progress updates + :raises RuntimeError: If there was an error (or the operation was cancelled) + """ + if isinstance(bv_or_db, BinaryView): + if not bv_or_db.file.has_database: + raise RuntimeError("Cannot sync non-database view") + db = bv_or_db.file.database + else: + db = bv_or_db + databasesync.sync_database(db, self, conflict_handler, progress, name_changeset) + + def pull(self, bv_or_db: Union['BinaryView', 'Database'], conflict_handler: 'util.ConflictHandlerType', progress: 'util.ProgressFuncType' = util.nop, name_changeset: 'util.NameChangesetFuncType' = util.nop): + """ + Pull updated snapshots from the remote. Merge local changes with remote changes and + potentially create a new snapshot for unsaved changes, named via name_changeset. + + :param bv_or_db: Binary view or database to sync with + :param conflict_handler: Function to call to resolve snapshot conflicts + :param name_changeset: Function to call for naming a pushed changeset, if necessary + :param progress: Function to call for progress updates + :raises RuntimeError: If there was an error (or the operation was cancelled) + """ + if isinstance(bv_or_db, BinaryView): + if not bv_or_db.file.has_database: + raise RuntimeError("Cannot pull non-database view") + db = bv_or_db.file.database + else: + db = bv_or_db + databasesync.pull_database(db, self, conflict_handler, progress, name_changeset) + + def push(self, bv_or_db: Union['BinaryView', 'Database'], progress: 'util.ProgressFuncType' = util.nop): + """ + Push locally added snapshots to the remote + + :param bv_or_db: Binary view or database to sync with + :param progress: Function to call for progress updates + :raises RuntimeError: If there was an error (or the operation was cancelled) + """ + if isinstance(bv_or_db, BinaryView): + if not bv_or_db.file.has_database: + raise RuntimeError("Cannot pull non-database view") + db = bv_or_db.file.database + else: + db = bv_or_db + databasesync.push_database(db, self, progress) diff --git a/python/collaboration/folder.py b/python/collaboration/folder.py new file mode 100644 index 00000000..cedbbe6d --- /dev/null +++ b/python/collaboration/folder.py @@ -0,0 +1,164 @@ + +import ctypes +from typing import Optional + +from .. import _binaryninjacore as core +from .. import project as core_project +from . import project, remote, util + + +class RemoteFolder: + """ + Class representing a remote folder in a project. + """ + def __init__(self, handle: core.BNRemoteFolderHandle): + self._handle = ctypes.cast(handle, core.BNRemoteFolderHandle) + + def __del__(self): + if core is not None: + core.BNFreeRemoteFolder(self._handle) + + def __eq__(self, other): + if not isinstance(other, RemoteFolder): + return False + return self.id == other.id + + def __repr__(self): + path = self.name + parent = self.parent + while parent is not None: + path = parent.name + '/' + path + parent = parent.parent + return f'' + + def __str__(self): + path = self.name + parent = self.parent + while parent is not None: + path = parent.name + '/' + path + parent = parent.parent + return f'' + + @property + def core_folder(self) -> 'core_project.ProjectFolder': + core_handle = core.BNRemoteFolderGetCoreFolder(self._handle) + if core_handle is None: + raise RuntimeError(util._last_error()) + return core_project.ProjectFolder(handle=ctypes.cast(core_handle, ctypes.POINTER(core.BNProjectFolder))) + + @property + def project(self) -> 'project.RemoteProject': + """ + Owning Project. + + :return: Project object + """ + value = core.BNRemoteFolderGetProject(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return project.RemoteProject(handle=value) + + @property + def remote(self) -> 'remote.Remote': + """ + Owning Remote + + :return: Remote object + """ + value = core.BNRemoteFolderGetRemote(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return remote.Remote(handle=value) + + @property + def parent(self) -> Optional['RemoteFolder']: + """ + Parent folder, if one exists. None if this is in the root of a project. + + :return: Parent folder object or None + """ + if not self.project.has_pulled_folders: + self.project.pull_folders() + parent = ctypes.POINTER(core.BNRemoteFolder)() + if not core.BNRemoteFolderGetParent(self._handle, parent): + raise RuntimeError(util._last_error()) + if not parent: + return None + return RemoteFolder(handle=parent) + + @parent.setter + def parent(self, parent: Optional['RemoteFolder']): + """ + Set the parent folder. You will need to push the folder to update the remote version. + + :param parent: New parent folder + :raises RuntimeError: If there was an error + """ + if not core.BNRemoteFolderSetParent(self._handle, parent._handle if parent is not None else None): + raise RuntimeError(util._last_error()) + + @property + def url(self) -> str: + """ + Web api endpoint URL + + :return: URL string + """ + return core.BNRemoteFolderGetUrl(self._handle) + + @property + def id(self) -> str: + """ + Unique id + + :return: Id string + """ + return core.BNRemoteFolderGetId(self._handle) + + @property + def parent_id(self) -> Optional[str]: + """ + Unique id of parent folder, if there is a parent. None, otherwise + + :return: Id string or None + """ + parent_id = ctypes.c_char_p() + if not core.BNRemoteFolderGetParentId(self._handle, parent_id): + return None + return core.pyNativeStr(parent_id.value) + + @property + def name(self) -> str: + """ + Displayed name of folder + + :return: Name string + """ + return core.BNRemoteFolderGetName(self._handle) + + @name.setter + def name(self, name: str): + """ + Set the display name of the folder. You will need to push the folder to update the remote version. + + :param name: New name + """ + core.BNRemoteFolderSetName(self._handle, name) + + @property + def description(self) -> str: + """ + Description of the folder + + :return: Description string + """ + return core.BNRemoteFolderGetDescription(self._handle) + + @description.setter + def description(self, description: str): + """ + Set the description of the folder. You will need to push the folder to update the remote version. + + :param description: New description + """ + core.BNRemoteFolderSetDescription(self._handle, description) diff --git a/python/collaboration/group.py b/python/collaboration/group.py new file mode 100644 index 00000000..d1fbd9d0 --- /dev/null +++ b/python/collaboration/group.py @@ -0,0 +1,112 @@ +import ctypes +from typing import List, Tuple + +from .. import _binaryninjacore as core +from . import remote, util + + +class Group: + """ + Class representing a remote Group + """ + def __init__(self, handle: core.BNCollaborationGroupHandle): + self._handle = ctypes.cast(handle, core.BNCollaborationGroupHandle) + + def __del__(self): + core.BNFreeCollaborationGroup(self._handle) + + def __eq__(self, other): + if not isinstance(other, Group): + return False + return self.id == other.id + + @property + def remote(self) -> 'remote.Remote': + """ + Owning Remote + + :return: Remote object + """ + value = core.BNCollaborationGroupGetRemote(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return remote.Remote(handle=value) + + @property + def url(self) -> str: + """ + Web api endpoint url + + :return: URL string + """ + return core.BNCollaborationGroupGetUrl(self._handle) + + @property + def id(self) -> int: + """ + Unique id + + :return: Id number + """ + return core.BNCollaborationGroupGetId(self._handle) + + @property + def name(self) -> str: + """ + Group name + + :return: Name string + """ + return core.BNCollaborationGroupGetName(self._handle) + + @name.setter + def name(self, name: str): + """ + Set group name + You will need to push the group to update the Remote. + + :param name: New group name + """ + core.BNCollaborationGroupSetName(self._handle, name) + + @property + def users(self) -> List[Tuple[str, str]]: + """ + Get list of users in the group + + :return: List of (userid, username) pairs + """ + count = ctypes.c_size_t() + user_ids = ctypes.POINTER(ctypes.c_char_p)() + usernames = ctypes.POINTER(ctypes.c_char_p)() + if not core.BNCollaborationGroupGetUsers(self._handle, user_ids, usernames, count): + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append((core.pyNativeStr(user_ids[i]), core.pyNativeStr(usernames[i]))) + core.BNFreeStringList(user_ids, count.value) + core.BNFreeStringList(usernames, count.value) + return result + + @users.setter + def users(self, usernames: List[str]): + """ + Set the list of users in a group by their usernames. + You will need to push the group to update the Remote. + + :param usernames: Usernames of new group members + """ + array = (ctypes.c_char_p * len(usernames))() + for i in range(len(usernames)): + array[i] = core.cstr(usernames[i]) + if not core.BNCollaborationGroupSetUsernames(self._handle, array, len(usernames)): + raise RuntimeError(util._last_error()) + + def contains_user(self, username: str) -> bool: + """ + Test if a group has a user with the given username + + :param username: Username of user to check membership + :return: If the user is in the group + """ + return core.BNCollaborationGroupContainsUser(self._handle, username) diff --git a/python/collaboration/merge.py b/python/collaboration/merge.py new file mode 100644 index 00000000..af06d602 --- /dev/null +++ b/python/collaboration/merge.py @@ -0,0 +1,240 @@ +import abc +import ctypes +import json +import sys +import traceback +from typing import Dict, Optional + +from .. import _binaryninjacore as core +from ..enums import MergeConflictDataType +from . import util + +from ..database import Database, Snapshot +from ..filemetadata import FileMetadata + +OptionalStringDict = Optional[Dict[str, object]] + + +class MergeConflict: + """ + Structure representing an individual merge conflict + """ + def __init__(self, handle: core.BNAnalysisMergeConflictHandle): + """ + FFI constructor + + :param handle: FFI handle for internal use + """ + self._handle = ctypes.cast(handle, core.BNAnalysisMergeConflictHandle) + + def __del__(self): + core.BNFreeAnalysisMergeConflict(self._handle) + + @property + def database(self) -> Database: + """ + Database backing all snapshots in the merge conflict + + :return: Database object + """ + result = core.BNAnalysisMergeConflictGetDatabase(self._handle) + return Database(handle=ctypes.cast(result, ctypes.POINTER(core.BNDatabase))) + + @property + def base_snapshot(self) -> Optional[Snapshot]: + """ + Snapshot which is the parent of the two being merged + + :return: Snapshot object + """ + result = core.BNAnalysisMergeConflictGetBaseSnapshot(self._handle) + if result is None: + return None + return Snapshot(handle=ctypes.cast(result, ctypes.POINTER(core.BNSnapshot))) + + @property + def first_snapshot(self) -> Optional[Snapshot]: + """ + First snapshot being merged + + :return: Snapshot object + """ + result = core.BNAnalysisMergeConflictGetFirstSnapshot(self._handle) + if result is None: + return None + return Snapshot(handle=ctypes.cast(result, ctypes.POINTER(core.BNSnapshot))) + + @property + def second_snapshot(self) -> Optional[Snapshot]: + """ + Second snapshot being merged + + :return: Snapshot object + """ + result = core.BNAnalysisMergeConflictGetSecondSnapshot(self._handle) + if result is None: + return None + return Snapshot(handle=ctypes.cast(result, ctypes.POINTER(core.BNSnapshot))) + + @property + def base_file(self) -> Optional[FileMetadata]: + """ + FileMetadata with contents of file for base snapshot + This function is slow! Only use it if you really need it. + + :return: FileMetadata object + """ + result = core.BNAnalysisMergeConflictGetBaseFile(self._handle) + if result is None: + return None + lazy = util.LazyT(handle=result) + file = FileMetadata(handle=ctypes.cast(lazy.get(ctypes.POINTER(core.BNFileMetadata)), ctypes.POINTER(core.BNFileMetadata))) + core.BNCollaborationFreeLazyT(result) + return file + + @property + def first_file(self) -> Optional[FileMetadata]: + """ + FileMetadata with contents of file for first snapshot + This function is slow! Only use it if you really need it. + + :return: FileMetadata object + """ + result = core.BNAnalysisMergeConflictGetFirstFile(self._handle) + if result is None: + return None + lazy = util.LazyT(handle=result) + file = FileMetadata(handle=ctypes.cast(lazy.get(ctypes.POINTER(core.BNFileMetadata)), ctypes.POINTER(core.BNFileMetadata))) + core.BNCollaborationFreeLazyT(result) + return file + + @property + def second_file(self) -> Optional[FileMetadata]: + """ + FileMetadata with contents of file for second snapshot + This function is slow! Only use it if you really need it. + + :return: FileMetadata object + """ + result = core.BNAnalysisMergeConflictGetSecondFile(self._handle) + if result is None: + return None + lazy = util.LazyT(handle=result) + file = FileMetadata(handle=ctypes.cast(lazy.get(ctypes.POINTER(core.BNFileMetadata)), ctypes.POINTER(core.BNFileMetadata))) + core.BNCollaborationFreeLazyT(result) + return file + + @property + def base(self) -> OptionalStringDict: + """ + Json object for conflicting data in the base snapshot + + :return: Python dictionary from parsed json + """ + result = core.BNAnalysisMergeConflictGetBase(self._handle) + if result is None: + return None + return json.loads(result) + + @property + def first(self) -> OptionalStringDict: + """ + Json object for conflicting data in the first snapshot + + :return: Python dictionary from parsed json + """ + result = core.BNAnalysisMergeConflictGetFirst(self._handle) + if result is None: + return None + return json.loads(result) + + @property + def second(self) -> OptionalStringDict: + """ + Json object for conflicting data in the second snapshot + + :return: Python dictionary from parsed json + """ + result = core.BNAnalysisMergeConflictGetSecond(self._handle) + if result is None: + return None + return json.loads(result) + + @property + def data_type(self) -> 'MergeConflictDataType': + """ + Type of data in the conflict, Text/Json/Binary + + :return: Conflict data type + """ + return MergeConflictDataType(core.BNAnalysisMergeConflictGetDataType(self._handle)) + + @property + def type(self) -> str: + """ + String representing the type name of the data, not the same as data_type. + This is like "typeName" or "tag" depending on what object the conflict represents. + + :return: Type name + """ + return core.BNAnalysisMergeConflictGetType(self._handle) + + @property + def key(self) -> str: + """ + Lookup key for the merge conflict, ideally a tree path that contains the name of the conflict + and all the recursive children leading up to this conflict. + + :return: Key name + """ + return core.BNAnalysisMergeConflictGetKey(self._handle) + + def success(self, value: OptionalStringDict) -> bool: + """ + Call this when you've resolved the conflict to save the result + + :param value: Resolved value + :return: True if successful + """ + if value is None: + printed = None + else: + printed = json.dumps(value) + return core.BNAnalysisMergeConflictSuccess(self._handle, printed) + + def get_path_item(self, path_key: str) -> Optional[object]: + """ + Get item in the merge conflict's path for a given key. + + :param path_key: Key for path item + :return: Path item, or an None if not found + """ + value = core.BNAnalysisMergeConflictGetPathItem(self._handle, path_key) + if value is None: + return None + return ctypes.py_object(value) + + +class ConflictHandler: + """ + Helper class that resolves conflicts + """ + def _handle(self, ctxt: ctypes.c_void_p, keys: ctypes.POINTER(ctypes.c_char_p), conflicts: ctypes.POINTER(core.BNAnalysisMergeConflictHandle), count: ctypes.c_ulonglong) -> bool: + try: + py_conflicts = {} + for i in range(count.value): + py_conflicts[core.pyNativeStr(keys[i])] = MergeConflict(handle=conflicts[i]) + return self.handle(py_conflicts) + except: + traceback.print_exc(file=sys.stderr) + return False + + @abc.abstractmethod + def handle(self, conflicts: Dict[str, MergeConflict]) -> bool: + """ + Handle any merge conflicts by calling their success() function with a merged value + + :param conflicts: Map of conflict id to conflict structure + :return: True if all conflicts were successfully merged + """ + raise NotImplementedError("Not implemented") diff --git a/python/collaboration/permission.py b/python/collaboration/permission.py new file mode 100644 index 00000000..32e79b29 --- /dev/null +++ b/python/collaboration/permission.py @@ -0,0 +1,159 @@ +import ctypes +from typing import Optional + +from .. import _binaryninjacore as core +from . import project, remote, util +from ..enums import CollaborationPermissionLevel + + +class Permission: + """ + Class representing a permission grant for a user or group on a project. + """ + def __init__(self, handle: core.BNCollaborationPermissionHandle): + self._handle = ctypes.cast(handle, core.BNCollaborationPermissionHandle) + + def __del__(self): + if core is not None: + core.BNFreeCollaborationPermission(self._handle) + + def __eq__(self, other): + if not isinstance(other, Permission): + return False + return other.id == self.id + + @property + def remote(self) -> 'remote.Remote': + """ + Owning Remote + + :return: Remote object + """ + value = core.BNCollaborationPermissionGetRemote(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return remote.Remote(handle=value) + + @property + def project(self) -> 'project.RemoteProject': + """ + Owning Project + + :return: Project object + """ + value = core.BNCollaborationPermissionGetProject(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return project.RemoteProject(handle=value) + + @property + def url(self) -> str: + """ + Web api endpoint url + + :return: URL string + """ + return core.BNCollaborationPermissionGetUrl(self._handle) + + @property + def id(self) -> str: + """ + Unique id + + :return: Id string + """ + return core.BNCollaborationPermissionGetId(self._handle) + + @property + def level(self) -> CollaborationPermissionLevel: + """ + Level of permission + + :return: Permission level + """ + return CollaborationPermissionLevel(core.BNCollaborationPermissionGetLevel(self._handle)) + + @level.setter + def level(self, value: CollaborationPermissionLevel): + """ + Change the level of the permission + You will need to push the group to update the Remote. + + :param value: New value + """ + core.BNCollaborationPermissionSetLevel(self._handle, value) + + @property + def group_id(self) -> Optional[int]: + """ + Id of affected group + + :return: Group id, if this is a group permission. Else, None + """ + result = core.BNCollaborationPermissionGetGroupId(self._handle) + if result == 0: + return None + return result + + @property + def group_name(self) -> Optional[str]: + """ + Name of affected group + + :return: Group name, if this is a group permission. Else, None + """ + result = core.BNCollaborationPermissionGetGroupName(self._handle) + if result == "": + return None + return result + + @property + def user_id(self) -> Optional[str]: + """ + Id of affected user + + :return: User id, if this is a user permission. Else, None + """ + result = core.BNCollaborationPermissionGetUserId(self._handle) + if result == "": + return None + return result + + @property + def username(self) -> Optional[str]: + """ + Name of affected user + + :return: User name, if this is a user permission. Else, None + """ + result = core.BNCollaborationPermissionGetUsername(self._handle) + if result == "": + return None + return result + + @property + def can_view(self) -> bool: + """ + If the permission grants the affect user/group the ability to read files in the project + + :return: True if permission granted + """ + return core.BNCollaborationPermissionCanView(self._handle) + + @property + def can_edit(self) -> bool: + """ + If the permission grants the affect user/group the ability to edit files in the project + + :return: True if permission granted + """ + return core.BNCollaborationPermissionCanEdit(self._handle) + + @property + def can_admin(self) -> bool: + """ + If the permission grants the affect user/group the ability to administer the project + + :return: True if permission granted + """ + return core.BNCollaborationPermissionCanAdmin(self._handle) diff --git a/python/collaboration/project.py b/python/collaboration/project.py new file mode 100644 index 00000000..b94ecda6 --- /dev/null +++ b/python/collaboration/project.py @@ -0,0 +1,826 @@ +import ctypes +import datetime +import tempfile +from os import PathLike +from pathlib import Path +from typing import Dict, List, Optional, Union + +from .. import _binaryninjacore as core +from .. import enums, load, log_info +from ..binaryview import BinaryView, ProgressFuncType +from ..database import Database +from ..filemetadata import FileMetadata +from ..project import Project +from . import databasesync, file, folder, permission, remote, util + + +def _nop(*args, **kwargs): + """ + Function that just returns True, used as default for callbacks + + :return: True + """ + return True + + +class RemoteProject: + """ + Class representing a remote project + """ + def __init__(self, handle: core.BNRemoteProjectHandle): + self._handle = ctypes.cast(handle, core.BNRemoteProjectHandle) + + def __del__(self): + if self._handle is not None: + core.BNFreeRemoteProject(self._handle) + + def __eq__(self, other): + if not isinstance(other, RemoteProject): + return False + return other.id == self.id + + def __str__(self): + return f'' + + def __repr__(self): + return f'' + + @property + def is_open(self) -> bool: + """ + Determine if the project is open (it needs to be opened before you can access its files) + + :return: True if open + """ + return core.BNRemoteProjectIsOpen(self._handle) + + def open(self, progress: ProgressFuncType = _nop) -> bool: + """ + Open the project, allowing various file and folder based apis to work, as well as + connecting a core Project (see :py:func:`core_project`). + + :param progress: Function to call for progress updates + :raises: RuntimeError if there was an error opening the Project + """ + if self.is_open: + return True + return core.BNRemoteProjectOpen(self._handle, util.wrap_progress(progress), None) + + def close(self): + """ + Close the project and stop all background operations (e.g. file uploads) + """ + core.BNRemoteProjectClose(self._handle) + + @staticmethod + def get_for_local_database(database: 'Database') -> Optional['RemoteProject']: + """ + Get the Remote Project for a Database + + :param database: BN database, potentially with collaboration metadata + :return: Remote project from one of the connected remotes, or None if not found + or if projects are not pulled + :raises RuntimeError: If there was an error + """ + remote = databasesync.get_remote_for_local_database(database) + if remote is None: + return None + if not remote.has_pulled_projects: + remote.pull_projects() + return databasesync.get_remote_project_for_local_database(database) + + @staticmethod + def get_for_bv(bv: 'BinaryView') -> Optional['RemoteProject']: + """ + Get the Remote Project for a BinaryView + + :param bv: BinaryView, potentially with collaboration metadata + :return: Remote project from one of the connected remotes, or None if not found + or if projects are not pulled + :raises RuntimeError: If there was an error + """ + if not bv.file.has_database: + return None + db = bv.file.database + if db is None: + return None + return RemoteProject.get_for_local_database(db) + + @property + def core_project(self) -> 'Project': + """ + Get the core :py:class:`binaryninja.Project` object for this Remote Project. + + .. note:: If the project has not been opened, it will be opened upon calling this. + + :return: Project instance + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + core_handle = core.BNRemoteProjectGetCoreProject(self._handle) + if core_handle is None: + raise RuntimeError(util._last_error()) + return Project(handle=ctypes.cast(core_handle, ctypes.POINTER(core.BNProject))) + + @property + def remote(self) -> 'remote.Remote': + """ + Owning Remote + + :return: Remote object + """ + value = core.BNRemoteProjectGetRemote(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return remote.Remote(handle=value) + + @property + def url(self): + """ + Web api endpoint URL + + :return: URL string + """ + return core.BNRemoteProjectGetUrl(self._handle) + + @property + def id(self): + """ + Unique id + + :return: Id string + """ + return core.BNRemoteProjectGetId(self._handle) + + @property + def created(self) -> datetime.datetime: + """ + Created date of the project + + :return: Date object + """ + return datetime.datetime.utcfromtimestamp(core.BNRemoteProjectGetCreated(self._handle)) + + @property + def last_modified(self) -> datetime.datetime: + """ + Last modified date of the project + + :return: Date object + """ + return datetime.datetime.utcfromtimestamp(core.BNRemoteProjectGetLastModified(self._handle)) + + @property + def name(self): + """ + Displayed name of project + + :return: Name string + """ + return core.BNRemoteProjectGetName(self._handle) + + @property + def description(self): + """ + Description of the project + + :return: Description string + """ + return core.BNRemoteProjectGetDescription(self._handle) + + @name.setter + def name(self, value): + """ + Set the display name of the project. You will need to push the project to update the remote version. + + :param value: New name + """ + if not core.BNRemoteProjectSetName(self._handle, value): + raise RuntimeError(util._last_error()) + + @description.setter + def description(self, value): + """ + Set the description of the project. You will need to push the project to update the remote version. + + :param description: New description + """ + if not core.BNRemoteProjectSetDescription(self._handle, value): + raise RuntimeError(util._last_error()) + + @property + def received_file_count(self) -> int: + """ + Get the number of files in a project (without needing to pull them first) + + :return: Number of files + """ + return core.BNRemoteProjectGetReceivedFileCount(self._handle) + + @property + def received_folder_count(self) -> int: + """ + Get the number of folders in a project (without needing to pull them first) + + :return: Number of folders + """ + return core.BNRemoteProjectGetReceivedFolderCount(self._handle) + + @property + def default_path(self) -> str: + """ + Get the default directory path for a remote Project. This is based off the Setting for + collaboration.directory, the project's id, and the project's remote's id. + + :return: Default project path + :rtype: str + """ + return databasesync.default_project_path(self) + + @property + def has_pulled_files(self): + """ + If the project has pulled the files yet + + :return: True if they have been pulled + """ + return core.BNRemoteProjectHasPulledFiles(self._handle) + + @property + def has_pulled_folders(self): + """ + If the project has pulled the folders yet + + :return: True if they have been pulled + """ + return core.BNRemoteProjectHasPulledFolders(self._handle) + + @property + def has_pulled_group_permissions(self): + """ + If the project has pulled the group permissions yet + + :return: True if they have been pulled + """ + return core.BNRemoteProjectHasPulledGroupPermissions(self._handle) + + @property + def has_pulled_user_permissions(self): + """ + If the project has pulled the user permissions yet + + :return: True if they have been pulled + """ + return core.BNRemoteProjectHasPulledUserPermissions(self._handle) + + @property + def is_admin(self): + """ + If the currently logged in user is an administrator of the project (and can edit + permissions and such for the project). + + :return: True if the user is an admin + """ + return core.BNRemoteProjectIsAdmin(self._handle) + + @property + def files(self) -> List['file.RemoteFile']: + """ + Get the list of files in this project. + + .. note:: If the project has not been opened, it will be opened upon calling this. + + .. note:: If folders have not been pulled, they will be pulled upon calling this. + + .. note:: If files have not been pulled, they will be pulled upon calling this. + + :return: List of File objects + :raises: RuntimeError if there was an error pulling files + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if not self.has_pulled_files: + self.pull_files() + + count = ctypes.c_size_t() + value = core.BNRemoteProjectGetFiles(self._handle, count) + if value is None: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(file.RemoteFile(value[i])) + return result + + def get_file_by_id(self, id: str) -> Optional['file.RemoteFile']: + """ + Get a specific File in the Project by its id + + .. note:: If the project has not been opened, it will be opened upon calling this. + + .. note:: If files have not been pulled, they will be pulled upon calling this. + + :param id: Id of File + :return: File object, if one with that id exists. Else, None + :raises: RuntimeError if there was an error pulling files + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if not self.has_pulled_files: + self.pull_files() + + value = core.BNRemoteProjectGetFileById(self._handle, id) + if value is None: + return None + return file.RemoteFile(value) + + def get_file_by_name(self, name: str) -> Optional['file.RemoteFile']: + """ + Get a specific File in the Project by its name + + .. note:: If the project has not been opened, it will be opened upon calling this. + + .. note:: If files have not been pulled, they will be pulled upon calling this. + + :param name: Name of File + :return: File object, if one with that name exists. Else, None + :raises: RuntimeError if there was an error pulling files + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if not self.has_pulled_files: + self.pull_files() + + value = core.BNRemoteProjectGetFileByName(self._handle, name) + if value is None: + return None + return file.RemoteFile(value) + + def pull_files(self, progress: 'util.ProgressFuncType' = util.nop): + """ + Pull the list of files from the Remote. + + .. note:: If the project has not been opened, it will be opened upon calling this. + + .. note:: If folders have not been pulled, they will be pulled upon calling this. + + :param progress: Function to call for progress updates + :raises: RuntimeError if there was an error pulling files + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if not self.has_pulled_folders: + self.pull_folders() + + if not core.BNRemoteProjectPullFiles(self._handle, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + def create_file(self, filename: str, contents: bytes, name: str, description: str, parent_folder: Optional['folder.RemoteFolder'] = None, + file_type: enums.RemoteFileType = enums.RemoteFileType.BinaryViewAnalysisFileType, progress: 'util.ProgressFuncType' = util.nop) -> 'file.RemoteFile': + """ + Create a new file on the remote (and pull it) + + .. note:: If the project has not been opened, it will be opened upon calling this. + + :param filename: File name + :param contents: File contents + :param name: Displayed file name + :param description: File description + :param parent_folder: Folder that will contain the file + :param file_type: Type of File to create + :param progress: Function to call on upload progress updates + :return: Reference to the created file + :raises: RuntimeError if there was an error + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + folder_handle = parent_folder._handle if parent_folder is not None else None + value = core.BNRemoteProjectCreateFile(self._handle, filename, contents, len(contents), name, description, folder_handle, file_type.value, util.wrap_progress(progress), None) + if value is None: + raise RuntimeError(util._last_error()) + return file.RemoteFile(value) + + def push_file(self, file: 'file.RemoteFile', extra_fields: Optional[Dict[str, str]] = None): + """ + Push an updated File object to the Remote + + .. note:: If the project has not been opened, it will be opened upon calling this. + + :param file: File object which has been updated + :param extra_fields: Extra HTTP fields to send with the update + :raises: RuntimeError if there was an error + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if extra_fields is None: + extra_fields = {} + extra_field_keys = (ctypes.c_char_p * len(extra_fields))() + extra_field_values = (ctypes.c_char_p * len(extra_fields))() + for (i, (key, value)) in enumerate(extra_fields.items()): + extra_field_keys[i] = core.cstr(key) + extra_field_values[i] = core.cstr(value) + if not core.BNRemoteProjectPushFile(self._handle, file._handle, extra_field_keys, extra_field_values, len(extra_fields)): + raise RuntimeError(util._last_error()) + + def delete_file(self, file: 'file.RemoteFile'): + """ + Delete a file from the remote + + .. note:: If the project has not been opened, it will be opened upon calling this. + + :param file: File to delete + :raises: RuntimeError if there was an error + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if not core.BNRemoteProjectDeleteFile(self._handle, file._handle): + raise RuntimeError(util._last_error()) + + @property + def folders(self) -> List['folder.RemoteFolder']: + """ + Get the list of folders in this project. + + .. note:: If the project has not been opened, it will be opened upon calling this. + + .. note:: If folders have not been pulled, they will be pulled upon calling this. + + :return: List of Folder objects + :raises: RuntimeError if there was an error pulling folders + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if not self.has_pulled_folders: + self.pull_folders() + + count = ctypes.c_size_t() + value = core.BNRemoteProjectGetFolders(self._handle, count) + if value is None: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(folder.RemoteFolder(value[i])) + return result + + def get_folder_by_id(self, id: str) -> Optional['folder.RemoteFolder']: + """ + Get a specific Folder in the Project by its id + + .. note:: If the project has not been opened, it will be opened upon calling this. + + .. note:: If folders have not been pulled, they will be pulled upon calling this. + + :param id: Id of Folder + :return: Folder object, if one with that id exists. Else, None + :raises: RuntimeError if there was an error pulling folders + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if not self.has_pulled_folders: + self.pull_folders() + + value = core.BNRemoteProjectGetFolderById(self._handle, id) + if value is None: + return None + return folder.RemoteFolder(value) + + def pull_folders(self, progress: 'util.ProgressFuncType' = util.nop): + """ + Pull the list of folders from the Remote. + + .. note:: If the project has not been opened, it will be opened upon calling this. + + :param progress: Function to call for progress updates + :raises: RuntimeError if there was an error pulling folders + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if not core.BNRemoteProjectPullFolders(self._handle, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + def create_folder(self, name: str, description: str, parent: Optional['folder.RemoteFolder'] = None, progress: 'util.ProgressFuncType' = util.nop) -> 'folder.RemoteFolder': + """ + Create a new folder on the remote (and pull it) + + .. note:: If the project has not been opened, it will be opened upon calling this. + + :param name: Displayed folder name + :param description: Folder description + :param parent: Parent folder (optional) + :param progress: Function to call on upload progress updates + :return: Reference to the created folder + :raises: RuntimeError if there was an error pulling folders + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + parent_handle = parent._handle if parent is not None else None + value = core.BNRemoteProjectCreateFolder(self._handle, name, description, parent_handle, util.wrap_progress(progress), None) + if value is None: + raise RuntimeError(util._last_error()) + return folder.RemoteFolder(value) + + def push_folder(self, folder: 'folder.RemoteFolder', extra_fields: Optional[Dict[str, str]] = None): + """ + Push an updated Folder object to the Remote + + .. note:: If the project has not been opened, it will be opened upon calling this. + + :param folder: Folder object which has been updated + :param extra_fields: Extra HTTP fields to send with the update + :raises: RuntimeError if there was an error + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if extra_fields is None: + extra_fields = {} + extra_field_keys = (ctypes.c_char_p * len(extra_fields))() + extra_field_values = (ctypes.c_char_p * len(extra_fields))() + for (i, (key, value)) in enumerate(extra_fields.items()): + extra_field_keys[i] = core.cstr(key) + extra_field_values[i] = core.cstr(value) + if not core.BNRemoteProjectPushFolder(self._handle, folder._handle, extra_field_keys, extra_field_values, len(extra_fields)): + raise RuntimeError(util._last_error()) + + def delete_folder(self, folder: 'folder.RemoteFolder'): + """ + Delete a folder from the remote + + .. note:: If the project has not been opened, it will be opened upon calling this. + + :param folder: Folder to delete + :raises: RuntimeError if there was an error + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if not core.BNRemoteProjectDeleteFolder(self._handle, folder._handle): + raise RuntimeError(util._last_error()) + + @property + def group_permissions(self) -> List['permission.Permission']: + """ + Get the list of group permissions in this project. + + .. note:: If group permissions have not been pulled, they will be pulled upon calling this. + + :return: List of Permission objects + :raises: RuntimeError if there was an error pulling group permissions + """ + if not self.has_pulled_group_permissions: + self.pull_group_permissions() + + count = ctypes.c_size_t() + value = core.BNRemoteProjectGetGroupPermissions(self._handle, count) + if value is None: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(permission.Permission(value[i])) + return result + + @property + def user_permissions(self) -> List['permission.Permission']: + """ + Get the list of user permissions in this project. + + .. note:: If user permissions have not been pulled, they will be pulled upon calling this. + + :return: List of Permission objects + :raises: RuntimeError if there was an error pulling user permissions + """ + if not self.has_pulled_user_permissions: + self.pull_user_permissions() + + count = ctypes.c_size_t() + value = core.BNRemoteProjectGetUserPermissions(self._handle, count) + if value is None: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(permission.Permission(value[i])) + return result + + def get_permission_by_id(self, id: str) -> Optional['permission.Permission']: + """ + Get a specific permission in the Project by its id + + .. note:: If group or user permissions have not been pulled, they will be pulled upon calling this. + + :param id: Id of Permission + :return: Permission object, if one with that id exists. Else, None + :raises: RuntimeError if there was an error pulling permissions + """ + if not self.has_pulled_user_permissions: + self.pull_user_permissions() + if not self.has_pulled_group_permissions: + self.pull_group_permissions() + + value = core.BNRemoteProjectGetPermissionById(self._handle, id) + if value is None: + return None + return permission.Permission(value) + + def pull_group_permissions(self, progress: 'util.ProgressFuncType' = util.nop): + """ + Pull the list of group permissions from the Remote. + + :param progress: Function to call for progress updates + :raises: RuntimeError if there was an error pulling permissions + """ + if not core.BNRemoteProjectPullGroupPermissions(self._handle, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + def pull_user_permissions(self, progress: 'util.ProgressFuncType' = util.nop): + """ + Pull the list of user permissions from the Remote. + + :param progress: Function to call for progress updates + :raises: RuntimeError if there was an error pulling permissions + """ + if not core.BNRemoteProjectPullUserPermissions(self._handle, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + def create_group_permission(self, group_id: int, level: enums.CollaborationPermissionLevel, progress: 'util.ProgressFuncType' = util.nop) -> 'permission.Permission': + """ + Create a new group permission on the remote (and pull it) + + :param group_id: Group id + :param level: Permission level + :param progress: Function to call on upload progress updates + :return: Reference to the created permission + :raises: RuntimeError if there was an error pulling permissions + """ + value = core.BNRemoteProjectCreateGroupPermission(self._handle, group_id, level, util.wrap_progress(progress), None) + if value is None: + raise RuntimeError(util._last_error()) + return permission.Permission(value) + + def create_user_permission(self, user_id: str, level: enums.CollaborationPermissionLevel, progress: 'util.ProgressFuncType' = util.nop) -> 'permission.Permission': + """ + Create a new user permission on the remote (and pull it) + + :param user_id: User id + :param level: Permission level + :param progress: Function to call on upload progress updates + :return: Reference to the created permission + :raises: RuntimeError if there was an error pulling permissions + """ + value = core.BNRemoteProjectCreateUserPermission(self._handle, user_id, level, util.wrap_progress(progress), None) + if value is None: + raise RuntimeError(util._last_error()) + return permission.Permission(value) + + def push_permission(self, permission: 'permission.Permission', extra_fields: Optional[Dict[str, str]] = None): + """ + Push project permissions to the remote + + :param permission: Permission object which has been updated + :param extra_fields: Extra HTTP fields to send with the update + :raises: RuntimeError if there was an error + """ + if extra_fields is None: + extra_fields = {} + extra_field_keys = (ctypes.c_char_p * len(extra_fields))() + extra_field_values = (ctypes.c_char_p * len(extra_fields))() + for (i, (key, value)) in enumerate(extra_fields.items()): + extra_field_keys[i] = core.cstr(key) + extra_field_values[i] = core.cstr(value) + if not core.BNRemoteProjectPushPermission(self._handle, permission._handle, extra_field_keys, extra_field_values, len(extra_fields)): + raise RuntimeError(util._last_error()) + + def delete_permission(self, permission: 'permission.Permission'): + """ + Delete a permission from the remote + + :param permission: Permission to delete + :raises: RuntimeError if there was an error + """ + if not core.BNRemoteProjectDeletePermission(self._handle, permission._handle): + raise RuntimeError(util._last_error()) + + def can_user_view(self, username: str) -> bool: + """ + Determine if a user is in any of the view/edit/admin groups + + :param username: Username of user to check + :return: True if they are in any of those groups + :raises: RuntimeError if there was an error + """ + return core.BNRemoteProjectCanUserView(self._handle, username) + + def can_user_edit(self, username: str) -> bool: + """ + Determine if a user is in any of the edit/admin groups + + :param username: Username of user to check + :return: True if they are in any of those groups + :raises: RuntimeError if there was an error + """ + return core.BNRemoteProjectCanUserEdit(self._handle, username) + + def can_user_admin(self, username: str) -> bool: + """ + Determine if a user is in the admin group + + :param username: Username of user to check + :return: True if they are in any of those groups + :raises: RuntimeError if there was an error + """ + return core.BNRemoteProjectCanUserAdmin(self._handle, username) + + def upload_new_file( + self, + target: Union[str, PathLike, 'BinaryView', 'FileMetadata'], + parent_folder: Optional['folder.RemoteFolder'] = None, + progress: 'util.ProgressFuncType' = util.nop, + open_view_options = None) -> 'file.RemoteFile': + """ + Upload a file to the project, creating a new File and pulling it + + .. note:: If the project has not been opened, it will be opened upon calling this. + + :param target: Path to file on disk or BinaryView/FileMetadata object of + already-opened file + :param parent_folder: Parent folder to place the uploaded file in + :param progress: Function to call for progress updates + :return: Created File object + :raises: RuntimeError if there was an error + """ + if not self.open(): + raise RuntimeError("Failed to open project") + + if isinstance(target, FileMetadata): + if target.has_database: + return databasesync.upload_database(target, self, parent_folder=parent_folder, progress=progress) + else: + target = target.raw + + if isinstance(target, BinaryView): + maybe_bv = target + target = maybe_bv.file.original_filename + else: + # Convert PathLike to string + if isinstance(target, (Path,)): + target = target.resolve() + + target = str(target) + + # Argument is a path, try opening it: + try: + if open_view_options is None: + open_view_options = {} + maybe_bv = load( + target, progress_func=util.split_progress(progress, 0, [0.25, 0.75]), **open_view_options) + except Exception as e: + raise RuntimeError("Could not upload view: " + str(e)) + + with tempfile.TemporaryDirectory() as temp_dir: + with maybe_bv as bv: + # Can't open, can't upload + if not bv: + raise RuntimeError("Could not open file at path for uploading") + + # If it is backed by a database, just upload that + metadata = bv.file + if metadata.has_database: + uploaded = databasesync.upload_database( + metadata, self, parent_folder=parent_folder, progress=util.split_progress(progress, 1, [0.25, 0.75])) + return uploaded + + # Ported from remotebrowser.cpp (original comments): + # TODO: This is not efficient at all! + # No db exists, so create one before uploading so we always have a root snapshot + # on the server + # - Load file into memory + # - Make temp path for temp database + # - Make temp database with file + # - UploadDatabase copies the temp database and makes its own + # - Delete temp database + # - Now you don't have an empty remote file + db_path = Path(temp_dir) / Path(target).name + log_info(f'Saving temporary database at {db_path}') + # Save bndb first to create database + if not metadata.create_database( + str(db_path), util.split_progress(progress, 1, [0.25, 0.25, 0.25, 0.25])): + raise RuntimeError("Could not save database for temporary path") + + if not metadata.save_auto_snapshot( + util.split_progress(progress, 2, [0.25, 0.25, 0.25, 0.25])): + raise RuntimeError("Could not create initial snapshot for upload") + + metadata.filename = str(db_path) + uploaded = databasesync.upload_database( + metadata, self, parent_folder=parent_folder, progress=util.split_progress(progress, 3, [0.25, 0.25, 0.25, 0.25])) + return uploaded diff --git a/python/collaboration/remote.py b/python/collaboration/remote.py new file mode 100644 index 00000000..dbd72598 --- /dev/null +++ b/python/collaboration/remote.py @@ -0,0 +1,722 @@ +import ctypes +import json +from typing import Dict, List, Optional, Tuple + +import binaryninja +import binaryninja.enterprise as enterprise + +from .. import _binaryninjacore as core +from . import databasesync, group, project, user, util + + +class Remote: + """ + Class representing a connection to a Collaboration server + """ + def __init__(self, handle: core.BNRemoteHandle): + """ + Create a Remote object (but don't connect to it yet) + + :param name: Identifier for remote + :param address: Base address (HTTPS) for all api requests + :param handle: FFI handle for internal use + :raises: RuntimeError if there was an error + """ + + self._handle = ctypes.cast(handle, core.BNRemoteHandle) + + def __del__(self): + if core is not None: + core.BNFreeRemote(self._handle) + + def __eq__(self, other): + if not isinstance(other, Remote): + return False + if not self.has_loaded_metadata or not other.has_loaded_metadata: + # Don't pull metadata if we haven't yet + return self.address == other.address + return other.unique_id == self.unique_id + + def __str__(self): + return f'' + + def __repr__(self): + return f'' + + @staticmethod + def get_for_local_database(database: 'binaryninja.Database') -> Optional['Remote']: + """ + Get the Remote for a Database + + :param database: BN database, potentially with collaboration metadata + :return: Remote from one of the connected remotes, or None if not found + :rtype: Optional[Remote] + :raises RuntimeError: If there was an error + """ + return databasesync.get_remote_for_local_database(database) + + @staticmethod + def get_for_bv(bv: 'binaryninja.BinaryView') -> Optional['Remote']: + """ + Get the Remote for a Binary View + + :param bv: Binary view, potentially with collaboration metadata + :return: Remote from one of the connected remotes, or None if not found + :raises RuntimeError: If there was an error + """ + if not bv.file.has_database: + return None + db = bv.file.database + if db is None: + return None + return databasesync.get_remote_for_local_database(db) + + @property + def has_loaded_metadata(self): + """ + If the remote has pulled metadata like its id, etc + + :return: True if it has been pulled + """ + return core.BNRemoteHasLoadedMetadata(self._handle) + + @property + def unique_id(self) -> str: + """ + Unique id. If metadata has not been pulled, it will be pulled upon calling this. + + :return: Id string + :raises RuntimeError: If there was an error pulling metadata. + """ + if not self.has_loaded_metadata: + self.load_metadata() + return core.BNRemoteGetUniqueId(self._handle) + + @property + def name(self) -> str: + """ + Assigned name of the Remote + + :return: Name string + """ + return core.BNRemoteGetName(self._handle) + + @property + def address(self) -> str: + """ + Base address of the Remote + + :return: URL string + """ + return core.BNRemoteGetAddress(self._handle) + + @property + def is_connected(self) -> bool: + """ + If the Remote is connected (has `Remote.connect` been called) + + :return: True if connected + """ + return core.BNRemoteIsConnected(self._handle) + + @property + def username(self) -> str: + """ + Username used to connect to the remote + + :return: Username string + """ + return core.BNRemoteGetUsername(self._handle) + + @property + def token(self) -> str: + """ + Token used to connect to the remote + + :return: Token string + """ + return core.BNRemoteGetToken(self._handle) + + @property + def server_version(self) -> int: + """ + Version of software running on the server. If metadata has not been pulled, it will + be pulled upon calling this. + + :return: Server version number + :raises RuntimeError: If there was an error + """ + if not self.has_loaded_metadata: + self.load_metadata() + return core.BNRemoteGetServerVersion(self._handle) + + @property + def server_build_id(self) -> str: + """ + Build id of software running on the server. If metadata has not been pulled, it will + be pulled upon calling this. + + :return: Server build id string + :raises RuntimeError: If there was an error + """ + if not self.has_loaded_metadata: + self.load_metadata() + return core.BNRemoteGetServerBuildId(self._handle) + + @property + def auth_backends(self) -> List[Tuple[str, str]]: + """ + List of supported authentication backends on the server. + If metadata has not been pulled, it will be pulled upon calling this. + + :return: List of Backend id <=> backend display name tuples + :raises RuntimeError: If there was an error + """ + if not self.has_loaded_metadata: + self.load_metadata() + backend_ids = ctypes.POINTER(ctypes.c_char_p)() + backend_names = ctypes.POINTER(ctypes.c_char_p)() + count = ctypes.c_size_t() + if not core.BNRemoteGetAuthBackends(self._handle, backend_ids, backend_names, count): + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append((core.pyNativeStr(backend_ids[i]), core.pyNativeStr(backend_names[i]))) + core.BNFreeStringList(backend_ids, count.value) + core.BNFreeStringList(backend_names, count.value) + return result + + @property + def is_admin(self) -> bool: + """ + If the currently connected user is an administrator. + + .. note:: If users have not been pulled, they will attempt to be pulled upon calling this. + + :return: True if the user is an administrator + """ + + # This is the test by which the api knows it is an admin + if not self.has_pulled_users: + self.pull_users() + + return core.BNRemoteIsAdmin(self._handle) + + @property + def is_enterprise(self) -> bool: + """ + If this remote is the same as the Enterprise License server + + :return: True if the same + """ + if not self.has_loaded_metadata: + self.load_metadata() + return core.BNRemoteIsEnterprise(self._handle) + + def load_metadata(self): + """ + Load metadata from the Remote, including unique id and versions + + :raises RuntimeError: If there was an error + """ + if not core.BNRemoteLoadMetadata(self._handle): + raise RuntimeError(util._last_error()) + + def request_authentication_token(self, username: str, password: str) -> Optional[str]: + """ + Request an authentication token using a username and password. + + :param username: Username to authenticate with + :param password: Password of user + :return: Authentication token string, or None if there was an error + """ + return core.BNRemoteRequestAuthenticationToken(self._handle, username, password) + + def connect(self, username: Optional[str] = None, token: Optional[str] = None): + """ + Connect to a Remote, loading metadata and optionally acquiring a token. + + .. note:: If no username or token are provided, they will be looked up from the keychain, \ + likely saved there by Enterprise authentication. + + :param username: Optional username to connect with + :param token: Optional token to authenticate with + :raises RuntimeError: If the connection fails + """ + if not self.has_loaded_metadata: + self.load_metadata() + if username is None: + # Try logging in with defaults + if self.is_enterprise: + username = enterprise.username() + token = enterprise.token() + else: + # Load from default secrets provider + secrets = binaryninja.SecretsProvider[ + binaryninja.Settings().get_string("enterprise.secretsProvider")] + if not secrets.has_data(self.address): + raise RuntimeError("No username and token provided, and none found " + "in the default keychain.") + creds = json.loads(secrets.get_data(self.address)) + username = creds['username'] + token = creds['token'] + if username is None or token is None: + raise RuntimeError("Cannot connect without a username or token") + + if not core.BNRemoteConnect(self._handle, username, token): + raise RuntimeError(util._last_error()) + + def disconnect(self): + """ + Disconnect from the remote + + :raises RuntimeError: If there was somehow an error + """ + if not core.BNRemoteDisconnect(self._handle): + raise RuntimeError(util._last_error()) + + @property + def has_pulled_projects(self) -> bool: + """ + If the project has pulled the projects yet + + :return: True if they have been pulled + """ + return core.BNRemoteHasPulledProjects(self._handle) + + @property + def has_pulled_groups(self) -> bool: + """ + If the project has pulled the groups yet + + :return: True if they have been pulled + """ + return core.BNRemoteHasPulledGroups(self._handle) + + @property + def has_pulled_users(self) -> bool: + """ + If the project has pulled the users yet + + :return: True if they have been pulled + """ + return core.BNRemoteHasPulledUsers(self._handle) + + @property + def projects(self) -> List['project.RemoteProject']: + """ + Get the list of projects in this project. + + .. note:: If projects have not been pulled, they will be pulled upon calling this. + + :return: List of Project objects + :raises: RuntimeError if there was an error pulling projects + """ + if not self.has_pulled_projects: + self.pull_projects() + + count = ctypes.c_size_t() + value = core.BNRemoteGetProjects(self._handle, count) + if value is None: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(project.RemoteProject(value[i])) + return result + + def get_project_by_id(self, id: str) -> Optional['project.RemoteProject']: + """ + Get a specific project in the Remote by its id + + .. note:: If projects have not been pulled, they will be pulled upon calling this. + + :param id: Id of Project + :return: Project object, if one with that id exists. Else, None + :raises: RuntimeError if there was an error pulling projects + """ + if not self.has_pulled_projects: + self.pull_projects() + + value = core.BNRemoteGetProjectById(self._handle, id) + if value is None: + return None + return project.RemoteProject(value) + + def get_project_by_name(self, name: str) -> Optional['project.RemoteProject']: + """ + Get a specific project in the Remote by its name + + .. note:: If projects have not been pulled, they will be pulled upon calling this. + + :param name: Name of Project + :return: Project object, if one with that name exists. Else, None + :raises: RuntimeError if there was an error pulling projects + """ + if not self.has_pulled_projects: + self.pull_projects() + + value = core.BNRemoteGetProjectByName(self._handle, name) + if value is None: + return None + return project.RemoteProject(value) + + def pull_projects(self, progress: 'util.ProgressFuncType' = util.nop): + """ + Pull the list of projects from the Remote. + + :param progress: Function to call for progress updates + :raises: RuntimeError if there was an error pulling projects + """ + if not core.BNRemotePullProjects(self._handle, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + def create_project(self, name: str, description: str) -> 'project.RemoteProject': + """ + Create a new project on the remote (and pull it) + + :param name: Project name + :param description: Project description + :return: Reference to the created project + :raises: RuntimeError if there was an error + """ + value = core.BNRemoteCreateProject(self._handle, name, description) + if value is None: + raise RuntimeError(util._last_error()) + return project.RemoteProject(value) + + def push_project(self, project: 'project.RemoteProject', extra_fields: Optional[Dict[str, str]] = None): + """ + Push an updated Project object to the Remote + + :param project: Project object which has been updated + :param extra_fields: Extra HTTP fields to send with the update + :raises: RuntimeError if there was an error + """ + if extra_fields is None: + extra_fields = {} + extra_field_keys = (ctypes.c_char_p * len(extra_fields))() + extra_field_values = (ctypes.c_char_p * len(extra_fields))() + for (i, (key, value)) in enumerate(extra_fields.items()): + extra_field_keys[i] = core.cstr(key) + extra_field_values[i] = core.cstr(value) + if not core.BNRemotePushProject(self._handle, project._handle, extra_field_keys, extra_field_values, len(extra_fields)): + raise RuntimeError(util._last_error()) + + def delete_project(self, project: 'project.RemoteProject'): + """ + Delete a project from the remote + + :param project: Project to delete + :raises: RuntimeError if there was an error + """ + if not core.BNRemoteDeleteProject(self._handle, project._handle): + raise RuntimeError(util._last_error()) + + @property + def groups(self) -> List['group.Group']: + """ + Get the list of groups in this project. + + .. note:: If groups have not been pulled, they will be pulled upon calling this. + + .. note:: This function is only available to accounts with admin status on the Remote + + :return: List of Group objects + :raises: RuntimeError if there was an error pulling groups + """ + if not self.has_pulled_groups: + self.pull_groups() + + count = ctypes.c_size_t() + value = core.BNRemoteGetGroups(self._handle, count) + if value is None: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(group.Group(value[i])) + return result + + def get_group_by_id(self, id: int) -> Optional['group.Group']: + """ + Get a specific group in the Remote by its id + + .. note:: If groups have not been pulled, they will be pulled upon calling this. + + .. note:: This function is only available to accounts with admin status on the Remote + + :param id: Id of Group + :return: Group object, if one with that id exists. Else, None + :raises: RuntimeError if there was an error pulling groups + """ + if not self.has_pulled_groups: + self.pull_groups() + + value = core.BNRemoteGetGroupById(self._handle, id) + if value is None: + return None + return group.Group(value) + + def get_group_by_name(self, name: str) -> Optional['group.Group']: + """ + Get a specific group in the Remote by its name + + .. note:: If groups have not been pulled, they will be pulled upon calling this. + + .. note:: This function is only available to accounts with admin status on the Remote + + :param name: Name of Group + :return: Group object, if one with that name exists. Else, None + :raises: RuntimeError if there was an error pulling groups + """ + if not self.has_pulled_groups: + self.pull_groups() + + value = core.BNRemoteGetGroupByName(self._handle, name) + if value is None: + return None + return group.Group(value) + + def search_groups(self, prefix: str) -> List[Tuple[int, str]]: + """ + Search for groups in the Remote with a given prefix + + :param prefix: Prefix of name for groups + :return: List of group id <=> group name pairs + :raises: RuntimeError if there was an error + """ + count = ctypes.c_size_t() + group_ids = ctypes.POINTER(ctypes.c_uint64)() + group_names = ctypes.POINTER(ctypes.c_char_p)() + if not core.BNRemoteSearchGroups(self._handle, prefix, group_ids, group_names, count): + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append((group_ids[i], core.pyNativeStr(group_names[i]))) + core.BNCollaborationFreeIdList(group_ids, count.value) + core.BNFreeStringList(group_names, count.value) + return result + + def pull_groups(self, progress: 'util.ProgressFuncType' = util.nop): + """ + Pull the list of groups from the Remote. + + .. note:: This function is only available to accounts with admin status on the Remote + + :param progress: Function to call for progress updates + :raises: RuntimeError if there was an error pulling groups + """ + if not core.BNRemotePullGroups(self._handle, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + def create_group(self, name: str, usernames: List[str]) -> 'group.Group': + """ + Create a new group on the remote (and pull it) + + .. note:: This function is only available to accounts with admin status on the Remote + + :param name: Group name + :param usernames: List of usernames of users in the group + :return: Reference to the created group + :raises: RuntimeError if there was an error + """ + c_usernames = (ctypes.c_char_p * len(usernames))() + for (i, username) in enumerate(usernames): + c_usernames[i] = core.cstr(username) + + value = core.BNRemoteCreateGroup(self._handle, name, c_usernames, len(usernames)) + if value is None: + raise RuntimeError(util._last_error()) + return group.Group(value) + + def push_group(self, group: 'group.Group', extra_fields: Optional[Dict[str, str]] = None): + """ + Push an updated Group object to the Remote + + .. note:: This function is only available to accounts with admin status on the Remote + + :param group: Group object which has been updated + :param extra_fields: Extra HTTP fields to send with the update + :raises: RuntimeError if there was an error + """ + if extra_fields is None: + extra_fields = {} + extra_field_keys = (ctypes.c_char_p * len(extra_fields))() + extra_field_values = (ctypes.c_char_p * len(extra_fields))() + for (i, (key, value)) in enumerate(extra_fields.items()): + extra_field_keys[i] = core.cstr(key) + extra_field_values[i] = core.cstr(value) + if not core.BNRemotePushGroup(self._handle, group._handle, extra_field_keys, extra_field_values, len(extra_fields)): + raise RuntimeError(util._last_error()) + + def delete_group(self, group: 'group.Group'): + """ + Delete a group from the remote + + .. note:: This function is only available to accounts with admin status on the Remote + + :param group: Group to delete + :raises: RuntimeError if there was an error + """ + if not core.BNRemoteDeleteGroup(self._handle, group._handle): + raise RuntimeError(util._last_error()) + + @property + def users(self) -> List['user.User']: + """ + Get the list of users in this project. + + .. note:: If users have not been pulled, they will be pulled upon calling this. + + .. note:: This function is only available to accounts with admin status on the Remote + + :return: List of User objects + :raises: RuntimeError if there was an error pulling users + """ + if not self.has_pulled_users: + self.pull_users() + count = ctypes.c_size_t() + value = core.BNRemoteGetUsers(self._handle, count) + if value is None: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(user.User(value[i])) + return result + + def get_user_by_id(self, id: str) -> Optional['user.User']: + """ + Get a specific user in the Remote by their id + + .. note:: If users have not been pulled, they will be pulled upon calling this. + + .. note:: This function is only available to accounts with admin status on the Remote + + :param id: Id of User + :return: User object, if one with that id exists. Else, None + :raises: RuntimeError if there was an error pulling users + """ + if not self.has_pulled_users: + self.pull_users() + value = core.BNRemoteGetUserById(self._handle, id) + if value is None: + return None + return user.User(value) + + def get_user_by_username(self, username: str) -> Optional['user.User']: + """ + Get a specific user in the Remote by their username + + .. note:: If users have not been pulled, they will be pulled upon calling this. + + .. note:: This function is only available to accounts with admin status on the Remote + + :param username: Username of User + :return: User object, if one with that name exists. Else, None + :raises: RuntimeError if there was an error pulling users + """ + if not self.has_pulled_users: + self.pull_users() + value = core.BNRemoteGetUserByUsername(self._handle, username) + if value is None: + return None + return user.User(value) + + @property + def current_user(self) -> Optional['user.User']: + """ + Get the user object for the currently connected user (only if you are an admin) + + .. note:: If users have not been pulled, they will be pulled upon calling this. + + .. note:: This function is only available to accounts with admin status on the Remote + + :return: User object + :raises: RuntimeError if there was an error pulling users + """ + if not self.has_pulled_users: + self.pull_users() + value = core.BNRemoteGetCurrentUser(self._handle) + if value is None: + return None + return user.User(value) + + def search_users(self, prefix: str) -> List[Tuple[str, str]]: + """ + Search for users in the Remote with a given prefix + + :param prefix: Prefix of name for users + :return: List of user id <=> user name pairs + :raises: RuntimeError if there was an error + """ + count = ctypes.c_size_t() + user_ids = ctypes.POINTER(ctypes.c_char_p)() + usernames = ctypes.POINTER(ctypes.c_char_p)() + if not core.BNRemoteSearchUsers(self._handle, prefix, user_ids, usernames, count): + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append((core.pyNativeStr(user_ids[i]), core.pyNativeStr(usernames[i]))) + core.BNFreeStringList(user_ids, count.value) + core.BNFreeStringList(usernames, count.value) + return result + + def pull_users(self, progress: 'util.ProgressFuncType' = util.nop): + """ + Pull the list of users from the Remote. + + .. note:: This function is only available to accounts with admin status on the Remote. \ + Non-admin accounts attempting to call this function will pull an empty list of users. + + :param progress: Function to call for progress updates + :raises: RuntimeError if there was an error pulling users + """ + if not core.BNRemotePullUsers(self._handle, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + def create_user(self, username: str, email: str, is_active: bool, password: str, group_ids: List[int], user_permission_ids: List[int]) -> 'user.User': + """ + Create a new user on the remote (and pull it) + + .. note:: This function is only available to accounts with admin status on the Remote + + :param username: User username + :param email: User email + :param is_active: If the user is enabled + :param password: User password + :param group_ids: List of group ids for the user + :param user_permission_ids: List of permission ids for the user + :return: Reference to the created user + :raises: RuntimeError if there was an error + """ + group_ids_array = (ctypes.c_uint64 * len(group_ids))() + for i in range(len(group_ids)): + group_ids_array[i] = group_ids[i] + + user_permission_ids_array = (ctypes.c_uint64 * len(group_ids))() + for i in range(len(user_permission_ids)): + user_permission_ids_array[i] = user_permission_ids[i] + + value = core.BNRemoteCreateUser(self._handle, username, email, is_active, password, group_ids_array, len(group_ids), user_permission_ids_array, len(user_permission_ids)) + if value is None: + raise RuntimeError(util._last_error()) + return user.User(value) + + def push_user(self, user: 'user.User', extra_fields: Optional[Dict[str, str]] = None): + """ + Push an updated User object to the Remote + + .. note:: This function is only available to accounts with admin status on the Remote + + :param group: User object which has been updated + :param extra_fields: Extra HTTP fields to send with the update + :raises: RuntimeError if there was an error + """ + if extra_fields is None: + extra_fields = {} + extra_field_keys = (ctypes.c_char_p * len(extra_fields))() + extra_field_values = (ctypes.c_char_p * len(extra_fields))() + for (i, (key, value)) in enumerate(extra_fields.items()): + extra_field_keys[i] = core.cstr(key) + extra_field_values[i] = core.cstr(value) + if not core.BNRemotePushUser(self._handle, user._handle, extra_field_keys, extra_field_values, len(extra_fields)): + raise RuntimeError(util._last_error()) diff --git a/python/collaboration/snapshot.py b/python/collaboration/snapshot.py new file mode 100644 index 00000000..c704b1fb --- /dev/null +++ b/python/collaboration/snapshot.py @@ -0,0 +1,514 @@ +import ctypes +import datetime +from typing import List, Optional + +from .. import _binaryninjacore as core +from ..binaryview import BinaryView +from ..database import Snapshot +from . import databasesync, file, project, remote, util + +class CollabSnapshot: + """ + Class representing a remote Snapshot + """ + def __init__(self, handle: core.BNCollaborationSnapshotHandle): + self._handle = ctypes.cast(handle, core.BNCollaborationSnapshotHandle) + + def __del__(self): + if core is not None: + core.BNFreeCollaborationSnapshot(self._handle) + + def __eq__(self, other): + if not isinstance(other, CollabSnapshot): + return False + return other.id == self.id + + def __str__(self): + return f'' + + def __repr__(self): + return f'' + + @staticmethod + def get_for_local_snapshot(snapshot: 'Snapshot') -> Optional['CollabSnapshot']: + """ + Get the remote snapshot associated with a local snapshot (if it exists) + + :param snap: Local snapshot + :return: Remote snapshot if it exists, or None if not + :raises RuntimeError: If there was an error + """ + return databasesync.get_remote_snapshot_for_local(snapshot) + + @property + def file(self) -> 'file.RemoteFile': + """ + Owning File + + :return: File object + """ + value = core.BNCollaborationSnapshotGetFile(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return file.RemoteFile(handle=value) + + @property + def project(self) -> 'project.RemoteProject': + """ + Owning Project + + :return: Project object + """ + value = core.BNCollaborationSnapshotGetProject(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return project.RemoteProject(handle=value) + + @property + def remote(self) -> 'remote.Remote': + """ + Owning Remote + + :return: Remote object + """ + value = core.BNCollaborationSnapshotGetRemote(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return remote.Remote(handle=value) + + @property + def url(self) -> str: + """ + Web api endpoint url + + :return: + """ + return core.BNCollaborationSnapshotGetUrl(self._handle) + + @property + def id(self) -> str: + """ + Unique id + + :return: Id string + """ + return core.BNCollaborationSnapshotGetId(self._handle) + + @property + def name(self) -> str: + """ + Name of snapshot + + :return: Name string + """ + return core.BNCollaborationSnapshotGetName(self._handle) + + @property + def title(self) -> str: + """ + Get the title of a snapshot: the first line of its name + + :return: Snapshot title as described + """ + return core.BNCollaborationSnapshotGetTitle(self._handle) + + @property + def description(self) -> str: + """ + Get the description of a snapshot: the lines of its name after the first line + + :return: Snapshot description as described + """ + return core.BNCollaborationSnapshotGetDescription(self._handle) + + @property + def author(self) -> str: + """ + Get the user id of the author of a snapshot + + :return: Snapshot author user id + """ + return core.BNCollaborationSnapshotGetAuthor(self._handle) + + @property + def author_username(self) -> str: + """ + Get the username of the author of a snapshot, if possible (vs author which is user id) + + :return: Snapshot author username + """ + return core.BNCollaborationSnapshotGetAuthorUsername(self._handle) + + @property + def created(self) -> datetime.datetime: + """ + Created date of Snapshot + + :return: Created date + """ + return datetime.datetime.utcfromtimestamp(core.BNCollaborationSnapshotGetCreated(self._handle)) + + @property + def last_modified(self) -> datetime.datetime: + """ + Date of last modification to the snapshot + + :return: Last modified date + """ + return datetime.datetime.utcfromtimestamp(core.BNCollaborationSnapshotGetLastModified(self._handle)) + + @property + def hash(self) -> str: + """ + Hash of snapshot data (analysis and markup, etc) + No specific hash algorithm is guaranteed + + :return: Hash string + """ + return core.BNCollaborationSnapshotGetHash(self._handle) + + @property + def snapshot_file_hash(self) -> str: + """ + Hash of file contents in snapshot + No specific hash algorithm is guaranteed + + :return: Hash string + """ + return core.BNCollaborationSnapshotGetSnapshotFileHash(self._handle) + + @property + def has_pulled_undo_entires(self) -> bool: + """ + If the snapshot has pulled undo entries yet + + :return: True if they have been pulled + """ + return core.BNCollaborationSnapshotHasPulledUndoEntries(self._handle) + + @property + def is_finalized(self) -> bool: + """ + If the snapshot has been finalized on the server and is no longer editable + + :return: True if finalized + """ + return core.BNCollaborationSnapshotIsFinalized(self._handle) + + @property + def parent_ids(self) -> List[str]: + """ + List of ids of all remote parent Snapshots + + :return: List of id strings + :raises: RuntimeError if there was an error + """ + count = ctypes.c_size_t() + value = core.BNCollaborationSnapshotGetParentIds(self._handle, count) + if not value: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(value[i]) + return result + + @property + def child_ids(self) -> List[str]: + """ + List of ids of all remote child Snapshots + + :return: List of id strings + :raises: RuntimeError if there was an error + """ + count = ctypes.c_size_t() + value = core.BNCollaborationSnapshotGetChildIds(self._handle, count) + if not value: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(value[i]) + return result + + @property + def parents(self) -> List['CollabSnapshot']: + """ + List of all parent Snapshot objects + + :return: List of Snapshot objects + :raises: RuntimeError if there was an error + """ + count = ctypes.c_size_t() + value = core.BNCollaborationSnapshotGetParents(self._handle, count) + if not value: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(CollabSnapshot(handle=value[i])) + return result + + @property + def children(self) -> List['CollabSnapshot']: + """ + List of all child Snapshot objects + + :return: List of Snapshot objects + :raises: RuntimeError if there was an error + """ + count = ctypes.c_size_t() + value = core.BNCollaborationSnapshotGetChildren(self._handle, count) + if not value: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(CollabSnapshot(handle=value[i])) + return result + + @property + def undo_entries(self) -> List['UndoEntry']: + """ + Get the list of undo entries stored in this snapshot. + + .. note:: If undo entries have not been pulled, they will be pulled upon calling this. + + :return: List of UndoEntry objects + :raises: RuntimeError if there was an error pulling undo entries + """ + if not self.has_pulled_undo_entires: + self.pull_undo_entries() + + count = ctypes.c_size_t() + value = core.BNCollaborationSnapshotGetUndoEntries(self._handle, count) + if not value: + raise RuntimeError(util._last_error()) + result = [] + for i in range(count.value): + result.append(UndoEntry(handle=value[i])) + return result + + def get_undo_entry_by_id(self, id: int) -> Optional['UndoEntry']: + """ + Get a specific Undo Entry in the Snapshot by its id + + .. note:: If undo entries have not been pulled, they will be pulled upon calling this. + + :param id: Id of Undo Entry + :return: UndoEntry object, if one with that id exists. Else, None + :raises: RuntimeError if there was an error pulling undo entries + """ + if not self.has_pulled_undo_entires: + self.pull_undo_entries() + + value = core.BNCollaborationSnapshotGetUndoEntryById(self._handle, id) + if value is None: + return None + return UndoEntry(value) + + def pull_undo_entries(self, progress: 'util.ProgressFuncType' = util.nop): + """ + Pull the list of Undo Entries from the Remote. + + :param progress: Function to call for progress updates + :raises: RuntimeError if there was an error pulling undo entries + """ + if not core.BNCollaborationSnapshotPullUndoEntries(self._handle, util.wrap_progress(progress), None): + raise RuntimeError(util._last_error()) + + def create_undo_entry(self, parent: Optional[int], data: str) -> 'UndoEntry': + """ + Create a new Undo Entry in this snapshot. + + :param parent: Id of parent Undo Entry + :param data: Undo Entry contents + :return: Created Undo Entry + :raises: RuntimeError if there was an error + """ + value = core.BNCollaborationSnapshotCreateUndoEntry(self._handle, parent is not None, parent if parent is not None else 0, data) + if value is None: + raise RuntimeError(util._last_error()) + return UndoEntry(value) + + def finalize(self): + """ + Mark a snapshot as Finalized, committing it to the Remote, preventing future updates, + and allowing snapshots to be children of it. + + :raises: RuntimeError if there was an error + """ + if not core.BNCollaborationSnapshotFinalize(self._handle): + raise RuntimeError(util._last_error()) + + def download_snapshot_file(self, progress: 'util.ProgressFuncType' = util.nop) -> bytes: + """ + Download the contents of the file in the Snapshot. + + :param progress: Function to call for progress updates. Cancels if the function returns False. + :return: File contents data + :raises: RuntimeError if there was an error or the operation was cancelled + """ + data = ctypes.POINTER(ctypes.c_uint8)() + size = ctypes.c_size_t() + if not core.BNCollaborationSnapshotDownloadSnapshotFile(self._handle, util.wrap_progress(progress), None, data, size): + raise RuntimeError(util._last_error()) + return bytes(ctypes.cast(data, ctypes.POINTER(ctypes.c_uint8 * size.value)).contents) + + def download(self, progress: 'util.ProgressFuncType' = util.nop) -> bytes: + """ + Download the snapshot fields blob, compatible with KeyValueStore. + + :param progress: Function to call for progress updates. Cancels if the function returns False. + :return: Snapshot contents data + :raises: RuntimeError if there was an error or the operation was cancelled + """ + data = ctypes.POINTER(ctypes.c_uint8)() + size = ctypes.c_size_t() + if not core.BNCollaborationSnapshotDownload(self._handle, util.wrap_progress(progress), None, data, size): + raise RuntimeError(util._last_error()) + return bytes(ctypes.cast(data, ctypes.POINTER(ctypes.c_uint8 * size.value)).contents) + + def download_analysis_cache(self, progress: 'util.ProgressFuncType' = util.nop) -> bytes: + """ + Download the analysis cache fields blob, compatible with KeyValueStore. + + :param progress: Function to call for progress updates. Cancels if the function returns False. + :return: Snapshot analysis cache data + :raises: RuntimeError if there was an error or the operation was cancelled + """ + data = ctypes.POINTER(ctypes.c_uint8)() + size = ctypes.c_size_t() + if not core.BNCollaborationSnapshotDownloadAnalysisCache(self._handle, util.wrap_progress(progress), None, data, size): + raise RuntimeError(util._last_error()) + return bytes(ctypes.cast(data, ctypes.POINTER(ctypes.c_uint8 * size.value)).contents) + + def get_local_snapshot(self, bv: 'BinaryView') -> Optional['Snapshot']: + """ + Get the local snapshot associated with a remote snapshot (if it exists) + + :param bv: BinaryView with database to search + :return: Local snapshot, if one exists. Else, None + :raises: RuntimeError if there was an error + """ + if not bv.file.has_database: + return None + db = bv.file.database + if db is None: + return None + return databasesync.get_local_snapshot_for_remote(self, db) + + +class UndoEntry: + """ + Class representing a remote undo entry + """ + def __init__(self, handle: core.BNCollaborationUndoEntryHandle): + self._handle = ctypes.cast(handle, core.BNCollaborationUndoEntryHandle) + + def __del__(self): + if core is not None: + core.BNFreeCollaborationUndoEntry(self._handle) + + def __eq__(self, other): + if not isinstance(other, UndoEntry): + return False + return other.id == self.id + + @property + def snapshot(self) -> 'CollabSnapshot': + """ + Owning Snapshot + + :return: Snapshot object + """ + value = core.BNCollaborationUndoEntryGetSnapshot(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return CollabSnapshot(handle=value) + + @property + def file(self) -> 'file.RemoteFile': + """ + Owning File + + :return: File object + """ + value = core.BNCollaborationUndoEntryGetFile(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return file.RemoteFile(handle=value) + + @property + def project(self) -> 'project.RemoteProject': + """ + Owning Project + + :return: Project object + """ + value = core.BNCollaborationUndoEntryGetProject(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return project.RemoteProject(handle=value) + + @property + def remote(self) -> 'remote.Remote': + """ + Owning Remote + + :return: Remote object + """ + value = core.BNCollaborationUndoEntryGetRemote(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return remote.Remote(handle=value) + + @property + def url(self) -> str: + """ + Web api endpoint url + + :return: URL String + """ + return core.BNCollaborationUndoEntryGetUrl(self._handle) + + @property + def id(self) -> int: + """ + Unique id + + :return: Id number + """ + return core.BNCollaborationUndoEntryGetId(self._handle) + + @property + def parent_id(self) -> Optional[int]: + """ + Id of parent undo entry + + :return: Parent id number, if there is one, None otherwise + """ + id = ctypes.c_uint64() + if not core.BNCollaborationUndoEntryGetParentId(self._handle, id): + return None + return id.value + + @property + def data(self) -> str: + """ + Undo entry contents data + + :return: Data string + """ + data = ctypes.c_char_p() + if not core.BNCollaborationUndoEntryGetData(self._handle, data): + raise RuntimeError(util._last_error()) + return str(core.pyNativeStr(data.value)) + + @property + def parent(self) -> Optional['UndoEntry']: + """ + Parent Undo Entry object + + :return: Undo Entry object, if there is one, None otherwise + """ + value = core.BNCollaborationUndoEntryGetParent(self._handle) + if value is None: + return None + return UndoEntry(handle=value) diff --git a/python/collaboration/user.py b/python/collaboration/user.py new file mode 100644 index 00000000..1e8d4000 --- /dev/null +++ b/python/collaboration/user.py @@ -0,0 +1,119 @@ +import ctypes +from .. import _binaryninjacore as core +from . import remote, util + + +class User: + """ + Class representing a remote User + """ + def __init__(self, handle: core.BNCollaborationUserHandle): + self._handle = ctypes.cast(handle, core.BNCollaborationUserHandle) + + def __del__(self): + if core is not None: + core.BNFreeCollaborationUser(self._handle) + + def __eq__(self, other): + if not isinstance(other, User): + return False + return self.id == other.id + + @property + def remote(self) -> 'remote.Remote': + """ + Owning Remote + + :return: Remote object + """ + value = core.BNCollaborationUserGetRemote(self._handle) + if value is None: + raise RuntimeError(util._last_error()) + return remote.Remote(handle=value) + + @property + def url(self) -> str: + """ + Web api endpoint URL + + :return: URL string + """ + return core.BNCollaborationUserGetUrl(self._handle) + + @property + def id(self) -> str: + """ + Unique id + + :return: Id string + """ + return core.BNCollaborationUserGetId(self._handle) + + @property + def username(self) -> str: + """ + User's login username + + :return: Username string + """ + return core.BNCollaborationUserGetUsername(self._handle) + + @username.setter + def username(self, value: str): + """ + Set user's username. You will need to push the user to update the Remote + + :param value: New username + :raises RuntimeError: If there was an error + """ + if not core.BNCollaborationUserSetUsername(self._handle, value): + raise RuntimeError(util._last_error()) + + @property + def email(self) -> str: + """ + User's email address + + :return: Email string + """ + return core.BNCollaborationUserGetEmail(self._handle) + + @email.setter + def email(self, value: str): + """ + Set user's email. You will need to push the user to update the Remote + + :param value: New email address + :raises RuntimeError: If there was an error + """ + if not core.BNCollaborationUserSetEmail(self._handle, value): + raise RuntimeError(util._last_error()) + + @property + def last_login(self) -> str: + """ + String representing the last date the user logged in + + :return: Last login string + """ + return core.BNCollaborationUserGetLastLogin(self._handle) + + @property + def is_active(self) -> bool: + """ + If the user account is active and can log in + + :return: If account is active + """ + return core.BNCollaborationUserIsActive(self._handle) + + @is_active.setter + def is_active(self, value: bool): + """ + Enable/disable a user account. You will need to push the user to update the Remote + + :param value: New active value + :raises RuntimeError: If there was an error + """ + if not core.BNCollaborationUserSetIsActive(self._handle, value): + raise RuntimeError(util._last_error()) diff --git a/python/collaboration/util.py b/python/collaboration/util.py new file mode 100644 index 00000000..84750c25 --- /dev/null +++ b/python/collaboration/util.py @@ -0,0 +1,157 @@ +import ctypes +from typing import Callable, Dict, List, Optional, Union + +from .. import _binaryninjacore as core +from . import changeset, merge + +ProgressFuncType = Callable[[int, int], bool] +NameChangesetFuncType = Callable[['changeset.Changeset'], bool] +ConflictHandlerFuncType = Callable[[Dict[str, 'merge.MergeConflict']], bool] +ConflictHandlerType = Union['merge.ConflictHandler', ConflictHandlerFuncType] + + +def _last_error() -> str: + """ + Get last error from the api + + :return: Last error string + """ + return "Operation failed" + # TODO: keep track of last error in thread + #return core.BNCollaborationGetLastError() + + +def nop(*args, **kwargs): + """ + Function that just returns True, used as default for callbacks + + :return: True + """ + return True + + +def wrap_progress(progress_func: ProgressFuncType): + """ + Wraps a progress function in a ctypes function for passing to the FFI + + :param progress_func: Python progress function + :return: Wrapped ctypes function + """ + return ctypes.CFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.c_ulonglong, ctypes.c_ulonglong)( + lambda ctxt, cur, total: progress_func(cur, total)) + + +def wrap_name_changeset(name_changeset_func: NameChangesetFuncType): + """ + Wraps a changeset naming function in a ctypes function for passing to the FFI + + :param name_changeset_func: Python changeset naming function + :return: Wrapped ctypes function + """ + return ctypes.CFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, core.BNCollaborationChangesetHandle)( + lambda ctxt, cs: name_changeset_func(changeset.Changeset(handle=cs))) + + +def wrap_conflict_handler(handler: Union[ConflictHandlerFuncType, merge.ConflictHandler]): + """ + Wraps a conflict handler function in a ConflictHandler object so you can be lazy and just use a lambda + + :param handler: Python conflict handler function + :return: Wrapped ConflictHandler object + """ + + if isinstance(handler, merge.ConflictHandler): + handler_class = handler + else: + class LambdaConflictHandler(merge.ConflictHandler): + def handle(self, conflicts: Dict[str, 'merge.MergeConflict']) -> bool: + return handler(conflicts) + + handler_class = LambdaConflictHandler() + + return core.BNCollaborationAnalysisConflictHandler(handler_class._handle) + + +def split_progress(progress_func: Optional[ProgressFuncType], subpart: int, + subpart_weights: List[float]) -> ProgressFuncType: + """ + Split a single progress function into equally sized subparts. + This function takes the original progress function and returns a new function whose signature + is the same but whose output is shortened to correspond to the specified subparts. + + The length of a subpart is proportional to the sum of all the weights. + E.g. If subpart = 1 and subpartWeights = { 0.25, 0.5, 0.25 }, this will return a function that calls + progress_func and maps its progress to the range [0.25, 0.75] + + Internally this works by calling progress_func with total = 1000000 and doing math on the current value + + :param progress_func: Original progress function (usually updates a UI) + :param subpart: Index of subpart whose function to return, from 0 to (subpartWeights.size() - 1) + :param subpart_weights: Weights of subparts, described above + :return: A function that will call progress_func() within a modified progress region + """ + if not progress_func: + return lambda cur, total: True + subpart_sum = sum(subpart_weights) + if subpart_sum < 0.00001: + return lambda cur, total: True + + # Normalize weights and keep a running count of weights for the start + subpart_starts = [] + start = 0 + for i in range(len(subpart_weights)): + subpart_starts.append(start) + subpart_weights[i] /= subpart_sum + start += subpart_weights[i] + + def inner(cur: int, total: int) -> bool: + steps = 1000000 + subpart_size = steps * subpart_weights[subpart] + subpart_progress = float(cur) / float(total) * subpart_size + return progress_func(int(subpart_starts[subpart] * steps + subpart_progress), steps) + + return inner + + +class LazyT: + """ + Lazily loaded objects (but FFI) + Pretend this class is templated, because the C++ version is + """ + def __init__(self, ctor: Optional[Callable[[], object]] = None, handle=None): + """ + Create a new LazyT that will be initialized with the result of the given function, when it is first needed. + + :param ctor: Function to construct object + :param handle: FFI handle for internal use + """ + if handle is not None: + self._handle = handle + else: + self.ctor = ctor + self.value = None + self._handle = core.BNCollaborationLazyTCreate(ctypes.CFUNCTYPE(ctypes.c_void_p, ctypes.c_void_p)( + lambda ctxt: self._perform_deref()), None) + + def _perform_deref(self) -> ctypes.c_void_p: + if self.value is None: + self.value = self.ctor() + result = ctypes.cast(ctypes.py_object(self.value), ctypes.c_void_p) + return result + + def get(self, expected_type=object): + """ + Access the lazily loaded object. Will construct it if this is the first usage. + + :param expected_type: Expected type of result, ctypes will try to cast to it + :return: Result object + """ + result = core.BNCollaborationLazyTDereference(self._handle) + if result is None: + return None + if type == object: + result = ctypes.cast(result, ctypes.py_object) + return result + else: + result = ctypes.cast(result, expected_type) + return result diff --git a/python/generator.cpp b/python/generator.cpp index 4831e5ae..3fdf1fc1 100644 --- a/python/generator.cpp +++ b/python/generator.cpp @@ -294,6 +294,7 @@ int main(int argc, char* argv[]) // Create type objects fprintf(out, "# Type definitions\n"); + fprintf(out, "BNProgressFunction = ctypes.CFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.c_ulonglong, ctypes.c_ulonglong)\n"); for (auto& i : types) { string name; @@ -420,6 +421,8 @@ int main(int argc, char* argv[]) } fprintf(out, "\n# Function definitions\n"); + fprintf(out, "BNCollaborationAnalysisConflictHandler = ctypes.CFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.POINTER(ctypes.c_char_p), ctypes.POINTER(BNAnalysisMergeConflictHandle), ctypes.c_ulonglong)\n"); + fprintf(out, "BNCollaborationNameChangesetFunction = ctypes.CFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, BNCollaborationChangesetHandle)\n"); for (auto& i : funcs) { string name; -- cgit v1.3.1