Add docstrings

frida/__init__.py

@@ -43,6 +43,10 @@ OperationCancelledError = _frida.OperationCancelledError
 
 
 def query_system_parameters() -> Dict[str, Any]:
+    """
+    Returns a dictionary of information about the host system
+    """
+
     return get_local_device().query_system_parameters()
 
 
@@ -55,53 +59,114 @@ def spawn(
     stdio: Optional[str] = None,
     **kwargs: Any,
 ) -> int:
+    """
+    Spawn a process into an attachable state
+    """
+
     return get_local_device().spawn(program=program, argv=argv, envp=envp, env=env, cwd=cwd, stdio=stdio, **kwargs)
 
 
 def resume(target: core.ProcessTarget) -> None:
+    """
+    Resume a process from the attachable state
+    :param target: the PID or name of the process
+    """
+
     get_local_device().resume(target)
 
 
 def kill(target: core.ProcessTarget) -> None:
+    """
+    Kill a process
+    :param target: the PID or name of the process
+    """
+
     get_local_device().kill(target)
 
 
-def attach(target: Union[int, str], realm: Optional[str] = None, persist_timeout: Optional[int] = None) -> core.Session:
+def attach(
+    target: core.ProcessTarget, realm: Optional[str] = None, persist_timeout: Optional[int] = None
+) -> core.Session:
+    """
+    Attach to a process
+    :param target: the PID or name of the process
+    """
+
     return get_local_device().attach(target, realm=realm, persist_timeout=persist_timeout)
 
 
-def inject_library_file(target: Union[int, str], path: str, entrypoint: str, data: str) -> int:
+def inject_library_file(target: core.ProcessTarget, path: str, entrypoint: str, data: str) -> int:
+    """
+    Inject a library file to a process.
+    :param target: the PID or name of the process
+    """
+
     return get_local_device().inject_library_file(target, path, entrypoint, data)
 
 
-def inject_library_blob(target: Union[int, str], blob: bytes, entrypoint: str, data: str) -> int:
+def inject_library_blob(target: core.ProcessTarget, blob: bytes, entrypoint: str, data: str) -> int:
+    """
+    Inject a library blob to a process
+    :param target: the PID or name of the process
+    """
+
     return get_local_device().inject_library_blob(target, blob, entrypoint, data)
 
 
 def get_local_device() -> core.Device:
-    return get_device_matching(lambda d: d.type == "local", timeout=0)
+    """
+    Get the local device
+    """
+
+    return get_device_manager().get_local_device()
 
 
 def get_remote_device() -> core.Device:
-    return get_device_matching(lambda d: d.type == "remote", timeout=0)
+    """
+    Get the first remote device in the devices list
+    """
+
+    return get_device_manager().get_remote_device()
 
 
 def get_usb_device(timeout: int = 0) -> core.Device:
-    return get_device_matching(lambda d: d.type == "usb", timeout)
+    """
+    Get the first device connected over USB in the devices list
+    """
+
+    return get_device_manager().get_usb_device(timeout)
 
 
 def get_device(id: Optional[str], timeout: int = 0) -> core.Device:
+    """
+    Get a device by its id
+    """
+
     return get_device_manager().get_device(id, timeout)
 
 
 def get_device_matching(predicate: Callable[[core.Device], bool], timeout: int = 0) -> core.Device:
+    """
+    Get device matching predicate.
+    :param predicate: a function to filter the devices
+    :param timeout: operation timeout in seconds
+    """
+
     return get_device_manager().get_device_matching(predicate, timeout)
 
 
 def enumerate_devices() -> List[core.Device]:
+    """
+    Enumerate all the devices from the device manager
+    """
+
     return get_device_manager().enumerate_devices()
 
 
 @core.cancellable
 def shutdown() -> None:
+    """
+    Shutdown the main device manager
+    """
+
     get_device_manager()._impl.close()

frida/core.py

@@ -17,6 +17,10 @@ ProcessTarget = Union[int, str]
 
 
 def get_device_manager() -> "DeviceManager":
+    """
+    Get or create a singleton DeviceManager that let you manage all the devices
+    """
+
     global _device_manager
     if _device_manager is None:
         _device_manager = DeviceManager(_frida.DeviceManager())
@@ -46,6 +50,10 @@ def cancellable(f: Callable[..., R]) -> Callable[..., R]:
 
 
 class IOStream:
+    """
+    Frida's own implementation of an input/output stream
+    """
+
     def __init__(self, impl) -> None:
         self._impl = impl
 
@@ -54,26 +62,50 @@ class IOStream:
 
     @property
     def is_closed(self) -> bool:
+        """
+        Query whether the stream is closed
+        """
+
         return self._impl.is_closed()
 
     @cancellable
     def close(self) -> None:
+        """
+        Close the stream.
+        """
+
         self._impl.close()
 
     @cancellable
     def read(self, count: int) -> bytes:
+        """
+        Read up to the specified number of bytes from the stream
+        """
+
         return self._impl.read(count)
 
     @cancellable
     def read_all(self, count: int) -> bytes:
+        """
+        Read exactly the specified number of bytes from the stream
+        """
+
         return self._impl.read_all(count)
 
     @cancellable
     def write(self, data: bytes) -> int:
+        """
+        Write as much as possible of the provided data to the stream
+        """
+
         return self._impl.write(data)
 
     @cancellable
     def write_all(self, data: bytes) -> None:
+        """
+        Write all of the provided data to the stream
+        """
+
         self._impl.write_all(data)
 
 
@@ -83,10 +115,20 @@ class PortalMembership:
 
     @cancellable
     def terminate(self) -> None:
+        """
+        Terminate the membership
+        """
+
         self._impl.terminate()
 
 
 class ScriptExports:
+    """
+    Proxy object that expose all the RPC exports of a script as attributes on this class
+
+    A method named exampleMethod in a script will be called with instance.example_method on this object
+    """
+
     def __init__(self, script: "Script") -> None:
         self._script = script
 
@@ -124,21 +166,41 @@ class Script:
 
     @property
     def is_destroyed(self) -> bool:
+        """
+        Query whether the script has been destroyed
+        """
+
         return self._impl.is_destroyed()
 
     @cancellable
     def load(self) -> None:
+        """
+        Load the script.
+        """
+
         self._impl.load()
 
     @cancellable
     def unload(self) -> None:
+        """
+        Unload the script
+        """
+
         self._impl.unload()
 
     @cancellable
     def eternalize(self) -> None:
+        """
+        Eternalize the script
+        """
+
         self._impl.eternalize()
 
     def post(self, message: Any, data: Optional[str] = None) -> None:
+        """
+        Post a JSON-encoded message to the script
+        """
+
         raw_message = json.dumps(message)
         kwargs = {"data": data}
         _filter_missing_kwargs(kwargs)
@@ -146,33 +208,65 @@ class Script:
 
     @cancellable
     def enable_debugger(self, port: Optional[int] = None) -> None:
+        """
+        Enable the Node.js compatible script debugger
+        """
+
         kwargs = {"port": port}
         _filter_missing_kwargs(kwargs)
         self._impl.enable_debugger(**kwargs)
 
     @cancellable
     def disable_debugger(self) -> None:
+        """
+        Disable the Node.js compatible script debugger
+        """
+
         self._impl.disable_debugger()
 
     def on(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Add a signal handler
+        """
+
         if signal == "message":
             self._on_message_callbacks.append(callback)
         else:
             self._impl.on(signal, callback)
 
     def off(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Remove a signal handler
+        """
+
         if signal == "message":
             self._on_message_callbacks.remove(callback)
         else:
             self._impl.off(signal, callback)
 
     def get_log_handler(self) -> Callable[[str, str], None]:
+        """
+        Get the method that handles the script logs
+        """
+
         return self._log_handler
 
     def set_log_handler(self, handler: Callable[[str, str], None]) -> None:
+        """
+        Set the method that handles the script logs
+        :param handler: a callable that accepts two parameters:
+                        1. the log level name
+                        2. the log message
+        """
+
         self._log_handler = handler
 
     def default_log_handler(self, level: str, text: str) -> None:
+        """
+        The default implementation of the log handler, prints the message to stdout
+        or stderr, depending on the level
+        """
+
         if level == "info":
             print(text, file=sys.stdout)
         else:
@@ -290,28 +384,52 @@ class Session:
 
     @property
     def is_detached(self) -> bool:
+        """
+        Query whether the session is detached
+        """
+
         return self._impl.is_detached()
 
     @cancellable
     def detach(self) -> None:
+        """
+        Detach session from the process
+        """
+
         self._impl.detach()
 
     @cancellable
     def resume(self) -> None:
+        """
+        Resume session after network error
+        """
+
         self._impl.resume()
 
     @cancellable
     def enable_child_gating(self) -> None:
+        """
+        Enable child gating
+        """
+
         self._impl.enable_child_gating()
 
     @cancellable
     def disable_child_gating(self) -> None:
+        """
+        Disable child gating
+        """
+
         self._impl.disable_child_gating()
 
     @cancellable
     def create_script(
         self, source: str, name: Optional[str] = None, snapshot: Optional[bytes] = None, runtime: Optional[str] = None
     ) -> Script:
+        """
+        Create a new script
+        """
+
         kwargs = {"name": name, "snapshot": snapshot, "runtime": runtime}
         _filter_missing_kwargs(kwargs)
         return Script(self._impl.create_script(source, **kwargs))
@@ -320,12 +438,20 @@ class Session:
     def create_script_from_bytes(
         self, data: bytes, name: Optional[str] = None, snapshot: Optional[bytes] = None, runtime: Optional[str] = None
     ) -> Script:
+        """
+        Create a new script from bytecode
+        """
+
         kwargs = {"name": name, "snapshot": snapshot, "runtime": runtime}
         _filter_missing_kwargs(kwargs)
         return Script(self._impl.create_script_from_bytes(data, **kwargs))
 
     @cancellable
     def compile_script(self, source: str, name: Optional[str] = None, runtime: Optional[str] = None) -> bytes:
+        """
+        Compile script source code to bytecode
+        """
+
         kwargs = {"name": name, "runtime": runtime}
         _filter_missing_kwargs(kwargs)
         return self._impl.compile_script(source, **kwargs)
@@ -340,6 +466,10 @@ class Session:
     def setup_peer_connection(
         self, stun_server: Optional[str] = None, relays: Optional[Sequence[_frida.Relay]] = None
     ) -> None:
+        """
+        Set up a peer connection with the target process
+        """
+
         kwargs = {"stun_server": stun_server, "relays": relays}
         _filter_missing_kwargs(kwargs)
         self._impl.setup_peer_connection(**kwargs)
@@ -352,14 +482,26 @@ class Session:
         token: Optional[str] = None,
         acl: Union[None, List[str], Tuple[str]] = None,
     ) -> PortalMembership:
+        """
+        Join a portal
+        """
+
         kwargs: Dict[str, Any] = {"certificate": certificate, "token": token, "acl": acl}
         _filter_missing_kwargs(kwargs)
         return PortalMembership(self._impl.join_portal(address, **kwargs))
 
     def on(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Add a signal handler
+        """
+
         self._impl.on(signal, callback)
 
     def off(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Remove a signal handler
+        """
+
         self._impl.off(signal, callback)
 
 
@@ -372,21 +514,37 @@ class Bus:
 
     @cancellable
     def attach(self) -> None:
+        """
+        Attach to the bus
+        """
+
         self._impl.attach()
 
     def post(self, message: Any, data: Optional[Union[str, bytes]] = None) -> None:
+        """
+        Post a JSON-encoded message to the bus
+        """
+
         raw_message = json.dumps(message)
         kwargs = {"data": data}
         _filter_missing_kwargs(kwargs)
         self._impl.post(raw_message, **kwargs)
 
     def on(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Add a signal handler
+        """
+
         if signal == "message":
             self._on_message_callbacks.append(callback)
         else:
             self._impl.on(signal, callback)
 
     def off(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Remove a signal handler
+        """
+
         if signal == "message":
             self._on_message_callbacks.remove(callback)
         else:
@@ -403,6 +561,9 @@ class Bus:
 
 
 class Device:
+    """
+    Represents a device that Frida connects to
+    """
 
     def __init__(self, device) -> None:
         self.id = device.id
@@ -418,14 +579,26 @@ class Device:
 
     @property
     def is_lost(self) -> bool:
+        """
+        Query whether the device has been lost
+        """
+
         return self._impl.is_lost()
 
     @cancellable
     def query_system_parameters(self) -> Dict[str, Any]:
+        """
+        Returns a dictionary of information about the host system
+        """
+
         return self._impl.query_system_parameters()
 
     @cancellable
     def get_frontmost_application(self, scope: Optional[str] = None) -> Optional[_frida.Application]:
+        """
+        Get details about the frontmost application
+        """
+
         kwargs = {"scope": scope}
         _filter_missing_kwargs(kwargs)
         return self._impl.get_frontmost_application(**kwargs)
@@ -434,6 +607,10 @@ class Device:
     def enumerate_applications(
         self, identifiers: Optional[Sequence[str]] = None, scope: Optional[str] = None
     ) -> List[_frida.Application]:
+        """
+        Enumerate applications
+        """
+
         kwargs = {"identifiers": identifiers, "scope": scope}
         _filter_missing_kwargs(kwargs)
         return self._impl.enumerate_applications(**kwargs)
@@ -442,12 +619,21 @@ class Device:
     def enumerate_processes(
         self, pids: Optional[Sequence[int]] = None, scope: Optional[str] = None
     ) -> List[_frida.Process]:
+        """
+        Enumerate processes
+        """
+
         kwargs = {"pids": pids, "scope": scope}
         _filter_missing_kwargs(kwargs)
         return self._impl.enumerate_processes(**kwargs)
 
     @cancellable
     def get_process(self, process_name: str) -> _frida.Process:
+        """
+        Get the process with the given name
+        :raises ProcessNotFoundError: if the process was not found or there were more than one process with the given name
+        """
+
         process_name_lc = process_name.lower()
         matching = [
             process
@@ -464,18 +650,34 @@ class Device:
 
     @cancellable
     def enable_spawn_gating(self) -> None:
+        """
+        Enable spawn gating
+        """
+
         self._impl.enable_spawn_gating()
 
     @cancellable
     def disable_spawn_gating(self) -> None:
+        """
+        Disable spawn gating
+        """
+
         self._impl.disable_spawn_gating()
 
     @cancellable
     def enumerate_pending_spawn(self) -> List[_frida.Spawn]:
+        """
+        Enumerate pending spawn
+        """
+
         return self._impl.enumerate_pending_spawn()
 
     @cancellable
     def enumerate_pending_children(self) -> List[_frida.Child]:
+        """
+        Enumerate pending children
+        """
+
         return self._impl.enumerate_pending_children()
 
     @cancellable
@@ -489,6 +691,10 @@ class Device:
         stdio: Optional[str] = None,
         **kwargs: Any,
     ) -> int:
+        """
+        Spawn a process into an attachable state
+        """
+
         if not isinstance(program, str):
             argv = program
             if isinstance(argv[0], bytes):
@@ -504,14 +710,28 @@ class Device:
 
     @cancellable
     def input(self, target: ProcessTarget, data: bytes) -> None:
+        """
+        Input data on stdin of a spawned process
+        :param target: the PID or name of the process
+        """
+
         self._impl.input(self._pid_of(target), data)
 
     @cancellable
     def resume(self, target: ProcessTarget) -> None:
+        """
+        Resume a process from the attachable state
+        :param target: the PID or name of the process
+        """
+
         self._impl.resume(self._pid_of(target))
 
     @cancellable
     def kill(self, target: ProcessTarget) -> None:
+        """
+        Kill a process
+        :param target: the PID or name of the process
+        """
         self._impl.kill(self._pid_of(target))
 
     @cancellable
@@ -521,30 +741,61 @@ class Device:
         realm: Optional[str] = None,
         persist_timeout: Optional[int] = None,
     ) -> Session:
+        """
+        Attach to a process
+        :param target: the PID or name of the process
+        """
+
         kwargs = {"realm": realm, "persist_timeout": persist_timeout}
         _filter_missing_kwargs(kwargs)
         return Session(self._impl.attach(self._pid_of(target), **kwargs))
 
     @cancellable
     def inject_library_file(self, target: ProcessTarget, path: str, entrypoint: str, data: str) -> int:
+        """
+        Inject a library file to a process
+        :param target: the PID or name of the process
+        """
+
         return self._impl.inject_library_file(self._pid_of(target), path, entrypoint, data)
 
     @cancellable
     def inject_library_blob(self, target: ProcessTarget, blob: bytes, entrypoint: str, data: str) -> int:
+        """
+        Inject a library blob to a process
+        :param target: the PID or name of the process
+        """
+
         return self._impl.inject_library_blob(self._pid_of(target), blob, entrypoint, data)
 
     @cancellable
     def open_channel(self, address: str) -> IOStream:
+        """
+        Open a device-specific communication channel
+        """
+
         return IOStream(self._impl.open_channel(address))
 
     @cancellable
     def get_bus(self) -> Bus:
+        """
+        Get the message bus of the device
+        """
+
         return Bus(self._impl.get_bus())
 
     def on(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Add a signal handler
+        """
+
         self._impl.on(signal, callback)
 
     def off(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Remove a signal handler
+        """
+
         self._impl.off(signal, callback)
 
     def _pid_of(self, target: ProcessTarget) -> int:
@@ -562,19 +813,41 @@ class DeviceManager:
         return repr(self._impl)
 
     def get_local_device(self) -> Device:
+        """
+        Get the local device
+        """
+
         return self.get_device_matching(lambda d: d.type == "local", timeout=0)
 
     def get_remote_device(self) -> Device:
+        """
+        Get the first remote device in the devices list
+        """
+
         return self.get_device_matching(lambda d: d.type == "remote", timeout=0)
 
     def get_usb_device(self, timeout: int = 0) -> Device:
+        """
+        Get the first device connected over USB in the devices list
+        """
+
         return self.get_device_matching(lambda d: d.type == "usb", timeout)
 
     def get_device(self, id: Optional[str], timeout: int = 0) -> Device:
+        """
+        Get a device by its id
+        """
+
         return self.get_device_matching(lambda d: d.id == id, timeout)
 
     @cancellable
     def get_device_matching(self, predicate: Callable[[Device], bool], timeout: int = 0) -> Device:
+        """
+        Get device matching predicate
+        :param predicate: a function to filter the devices
+        :param timeout: operation timeout in seconds
+        """
+
         if timeout < 0:
             raw_timeout = -1
         elif timeout == 0:
@@ -585,6 +858,10 @@ class DeviceManager:
 
     @cancellable
     def enumerate_devices(self) -> List[Device]:
+        """
+        Enumerate devices
+        """
+
         return [Device(device) for device in self._impl.enumerate_devices()]
 
     @cancellable
@@ -596,6 +873,10 @@ class DeviceManager:
         token: Optional[str] = None,
         keepalive_interval: Optional[int] = None,
     ) -> Device:
+        """
+        Add a remote device
+        """
+
         kwargs: Dict[str, Any] = {
             "certificate": certificate,
             "origin": origin,
@@ -607,16 +888,32 @@ class DeviceManager:
 
     @cancellable
     def remove_remote_device(self, address: str) -> None:
+        """
+        Remove a remote device
+        """
+
         self._impl.remove_remote_device(address=address)
 
     def on(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Add a signal handler
+        """
+
         self._impl.on(signal, callback)
 
     def off(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Remove a signal handler
+        """
+
         self._impl.off(signal, callback)
 
 
 class RPCException(Exception):
+    """
+    Wraps remote errors from the script RPC
+    """
+
     def __str__(self) -> str:
         return str(self.args[2]) if len(self.args) >= 3 else str(self.args[0])
 
@@ -673,40 +970,79 @@ class PortalService:
 
     @cancellable
     def start(self) -> None:
+        """
+        Start listening for incoming connections
+        :raises InvalidOperationError: if the service isn't stopped
+        :raises AddressInUseError: if the given address is already in use
+        """
+
         self._impl.start()
 
     @cancellable
     def stop(self) -> None:
+        """
+        Stop listening for incoming connections, and kick any connected clients
+        :raises InvalidOperationError: if the service is already stopped
+        """
+
         self._impl.stop()
 
     def post(self, connection_id: int, message: Any, data: Optional[Union[str, bytes]] = None) -> None:
+        """
+        Post a message to a specific control channel.
+        """
+
         raw_message = json.dumps(message)
         kwargs = {"data": data}
         _filter_missing_kwargs(kwargs)
         self._impl.post(connection_id, raw_message, **kwargs)
 
     def narrowcast(self, tag: str, message: Any, data: Optional[Union[str, bytes]] = None) -> None:
+        """
+        Post a message to control channels with a specific tag
+        """
+
         raw_message = json.dumps(message)
         kwargs = {"data": data}
         _filter_missing_kwargs(kwargs)
         self._impl.narrowcast(tag, raw_message, **kwargs)
 
     def broadcast(self, message: Any, data: Optional[Union[str, bytes]] = None) -> None:
+        """
+        Broadcast a message to all control channels
+        """
+
         raw_message = json.dumps(message)
         kwargs = {"data": data}
         _filter_missing_kwargs(kwargs)
         self._impl.broadcast(raw_message, **kwargs)
 
     def enumerate_tags(self, connection_id: int) -> List[str]:
+        """
+        Enumerate tags of a specific connection
+        """
+
         return self._impl.enumerate_tags(connection_id)
 
     def tag(self, connection_id: int, tag: str) -> None:
+        """
+        Tag a specific control channel
+        """
+
         self._impl.tag(connection_id, tag)
 
     def untag(self, connection_id: int, tag: str) -> None:
+        """
+        Untag a specific control channel
+        """
+
         self._impl.untag(connection_id, tag)
 
     def on(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Add a signal handler
+        """
+
         if signal == "authenticated":
             self._on_authenticated_callbacks.append(callback)
         elif signal == "message":
@@ -715,6 +1051,10 @@ class PortalService:
             self._impl.on(signal, callback)
 
     def off(self, signal: str, callback: Callable[..., Any]) -> None:
+        """
+        Remove a signal handler
+        """
+
         if signal == "authenticated":
             self._on_authenticated_callbacks.remove(callback)
         elif signal == "message":
@@ -818,9 +1158,18 @@ class Cancellable:
 
     @property
     def is_cancelled(self) -> bool:
+        """
+        Query whether cancellable has been cancelled
+        """
+
         return self._impl.is_cancelled()
 
     def raise_if_cancelled(self) -> None:
+        """
+        Raise an exception if cancelled
+        :raises OperationCancelledError:
+        """
+
         self._impl.raise_if_cancelled()
 
     def get_pollfd(self) -> CancellablePollFD:
@@ -829,7 +1178,7 @@ class Cancellable:
     @classmethod
     def get_current(cls) -> _frida.Cancellable:
         """
-        Get the top cancellable from the stack.
+        Get the top cancellable from the stack
         """
 
         return _Cancellable.get_current()
@@ -846,16 +1195,33 @@ class Cancellable:
         self._impl.pop_current()
 
     def connect(self, callback: Callable[..., Any]) -> int:
+        """
+        Register notification callback
+        :returns: the created handler id
+        """
+
         return self._impl.connect(callback)
 
     def disconnect(self, handler_id: int) -> None:
+        """
+        Unregister notification callback.
+        """
+
         self._impl.disconnect(handler_id)
 
     def cancel(self) -> None:
+        """
+        Set cancellable to cancelled
+        """
+
         self._impl.cancel()
 
 
 def make_auth_callback(callback: Callable[[str], Any]) -> Callable[[Any], str]:
+    """
+    Wraps authenticated callbacks with JSON marshaling
+    """
+
     def authenticate(token: str) -> str:
         session_info = callback(token)
         return json.dumps(session_info)