Add support for authenticating via desktop app

.github/workflows/validate.yml

@@ -19,7 +19,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
-        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v3

src/onepassword/__init__.py

@@ -1,6 +1,6 @@
 # Code generated by op-codegen - DO NO EDIT MANUALLY
 
-from .client import Client
+from .client import Client, DesktopAuth
 from .defaults import DEFAULT_INTEGRATION_NAME, DEFAULT_INTEGRATION_VERSION
 from .types import *  # noqa F403
 from .errors import *  # noqa F403
@@ -20,6 +20,7 @@
     "Vaults",
     "DEFAULT_INTEGRATION_NAME",
     "DEFAULT_INTEGRATION_VERSION",
+    "DesktopAuth",
 ]
 
 for name, obj in inspect.getmembers(sys.modules["onepassword.types"]):
@@ -30,7 +31,7 @@
             and inspect.getmodule(obj) == sys.modules["onepassword.types"]
         )
         or isinstance(obj, int)
         or type(obj) == typing._LiteralGenericAlias
     ):
         __all__.append(name)
 

src/onepassword/client.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 import weakref
-from .core import _init_client, _release_client
-from .defaults import new_default_config
+from .core import Core, UniffiCore
+from .desktop_core import DesktopCore
+from .defaults import new_default_config, DesktopAuth
 from .secrets import Secrets
 from .items import Items
 from .vaults import Vaults
@@ -16,23 +17,28 @@
 
     @classmethod
     async def authenticate(
-        cls, auth: str, integration_name: str, integration_version: str
+        cls, auth: str | DesktopAuth, integration_name: str, integration_version: str
     ) -> Client:
         config = new_default_config(
-            auth=auth or "",
+            auth=auth,
             integration_name=integration_name,
             integration_version=integration_version,
         )
 
-        client_id = int(await _init_client(config))
+        if isinstance(auth, str):
+            core = UniffiCore()
+        elif isinstance(auth, DesktopAuth):
+            core = DesktopCore()
+
+        client_id = int(await core.init_client(config))
 
         authenticated_client = cls()
 
-        authenticated_client.secrets = Secrets(client_id)
-        authenticated_client.items = Items(client_id)
-        authenticated_client.vaults = Vaults(client_id)
+        authenticated_client.secrets = Secrets(client_id, core)
+        authenticated_client.items = Items(client_id, core)
+        authenticated_client.vaults = Vaults(client_id, core)
         authenticated_client._finalizer = weakref.finalize(
-            cls, _release_client, client_id
+            cls, core.release_client, client_id
         )
 
         return authenticated_client

src/onepassword/core.py

@@ -1,59 +1,72 @@
 import json
 import platform
-
+from typing import Protocol
 from onepassword.errors import raise_typed_exception
 
 # In empirical tests, we determined that maximum message size that can cross the FFI boundary
 # is ~128MB. Past this limit, FFI will throw an error and the program will crash.
 # We set the limit to 50MB to be safe and consistent with the other SDKs (where this limit is 64MB), to be reconsidered upon further testing
 MESSAGE_LIMIT = 50 * 1024 * 1024
 
-machine_arch = platform.machine().lower()
-
-if machine_arch in ["x86_64", "amd64"]:
-    import onepassword.lib.x86_64.op_uniffi_core as core
-elif machine_arch in ["aarch64", "arm64"]:
-    import onepassword.lib.aarch64.op_uniffi_core as core
-else:
-    raise ImportError(
-        f"Your machine's architecture is not currently supported: {machine_arch}"
-    )
-
-
-# InitClient creates a client instance in the current core module and returns its unique ID.
-async def _init_client(client_config):
-    try:
-        return await core.init_client(json.dumps(client_config))
-    except Exception as e:
-        raise_typed_exception(e)
-
-
-# Invoke calls specified business logic from the SDK core.
-async def _invoke(invoke_config):
-    serialized_config = json.dumps(invoke_config)
-    if len(serialized_config.encode()) > MESSAGE_LIMIT:
-        raise ValueError(
-            f"message size exceeds the limit of {MESSAGE_LIMIT} bytes, please contact 1Password at [email protected] or https://developer.1password.com/joinslack if you need help."
-        )
-    try:
-        return await core.invoke(serialized_config)
-    except Exception as e:
-        raise_typed_exception(e)
-
-
-# Invoke calls specified business logic from the SDK core.
-def _invoke_sync(invoke_config):
-    serialized_config = json.dumps(invoke_config)
-    if len(serialized_config.encode()) > MESSAGE_LIMIT:
-        raise ValueError(
-            f"message size exceeds the limit of {MESSAGE_LIMIT} bytes, please contact 1Password at [email protected] or https://developer.1password.com/joinslack if you need help."
-        )
-    try:
-        return core.invoke_sync(serialized_config)
-    except Exception as e:
-        raise_typed_exception(e)
-
-
-# ReleaseClient releases memory in the SDK core associated with the given client ID.
-def _release_client(client_id):
-    return core.release_client(json.dumps(client_id))
+class Core(Protocol):
+    async def init_client(self, client_config: dict) -> str: ...
+    async def invoke(self, invoke_config: dict) -> str: ...
+    def invoke_sync(self, invoke_config: dict) -> str: ...
+    def release_client(self, client_id: int) -> None: ...
+
+class UniffiCore:
+    def __init__(self):
+        machine_arch = platform.machine().lower()
+
+        if machine_arch in ["x86_64", "amd64"]:
+            import onepassword.lib.x86_64.op_uniffi_core as core
+        elif machine_arch in ["aarch64", "arm64"]:
+            import onepassword.lib.aarch64.op_uniffi_core as core
+        else:
+            raise ImportError(
+                f"Your machine's architecture is not currently supported: {machine_arch}"
+            )
+
+        self.core = core
+
+    async def init_client(self, client_config: dict):
+        """Creates a client instance in the current core module and returns its unique ID."""
+        try:
+            return await self.core.init_client(json.dumps(client_config))
+        except Exception as e:
+            raise_typed_exception(e)
+
+    async def invoke(self, invoke_config: dict):
+        """Invoke business logic asynchronously."""
+        serialized_config = json.dumps(invoke_config)
+        if len(serialized_config.encode()) > MESSAGE_LIMIT:
+            raise ValueError(
+                f"message size exceeds the limit of {MESSAGE_LIMIT} bytes, "
+                "please contact 1Password at [email protected] or "
+                "https://developer.1password.com/joinslack if you need help."
+            )
+        try:
+            return await self.core.invoke(serialized_config)
+        except Exception as e:
+            raise_typed_exception(e)
+
+    def invoke_sync(self, invoke_config: dict):
+        """Invoke business logic synchronously."""
+        serialized_config = json.dumps(invoke_config)
+        if len(serialized_config.encode()) > MESSAGE_LIMIT:
+            raise ValueError(
+                f"message size exceeds the limit of {MESSAGE_LIMIT} bytes, "
+                "please contact 1Password at [email protected] or "
+                "https://developer.1password.com/joinslack if you need help."
+            )
+        try:
+            return self.core.invoke_sync(serialized_config)
+        except Exception as e:
+            raise_typed_exception(e)
+
+    def release_client(self, client_id: int):
+        """Releases memory in the SDK core associated with the given client ID."""
+        try:
+            return self.core.release_client(json.dumps(client_id))
+        except Exception as e:
+            raise_typed_exception(e)

src/onepassword/defaults.py

@@ -9,11 +9,19 @@
 DEFAULT_REQUEST_LIBRARY_VERSION = "0.11.24"
 DEFAULT_OS_VERSION = "0.0.0"
 
+class DesktopAuth:
+    def __init__(self, account_name: str):
+        """
+        Initialize a DesktopAuth instance.
+
+        Args:
+            account_name (str): The name of the account.
+        """
+        self.account_name = account_name
 
 # Generates a configuration dictionary with the user's parameters
-def new_default_config(auth, integration_name, integration_version):
+def new_default_config(auth: DesktopAuth | str, integration_name, integration_version):
     client_config_dict = {
-        "serviceAccountToken": auth,
         "programmingLanguage": SDK_LANGUAGE,
         "sdkVersion": SDK_VERSION,
         "integrationName": integration_name,
@@ -24,4 +32,9 @@ def new_default_config(auth, integration_name, integration_version):
         "osVersion": DEFAULT_OS_VERSION,
         "architecture": platform.machine(),
     }
+    if isinstance(auth, str):
+        client_config_dict["serviceAccountToken"] = auth
+    elif isinstance(auth, DesktopAuth):
+        client_config_dict["accountName"] = auth.account_name
+
     return client_config_dict

src/onepassword/desktop_core.py

@@ -0,0 +1,98 @@
+import ctypes
+import json
+import os
+import platform
+import sys
+from ctypes import c_uint8, c_size_t, c_int32, POINTER, byref, c_void_p
+from .core import UniffiCore
+
+
+def find_1password_lib_path():
+    host_os = platform.system().lower()  # "darwin", "linux", "windows"
+
+    core = UniffiCore()
+    if core is None:
+        raise RuntimeError("failed to get ExtismCore")
+
+    locations_raw = core.invoke({
+        "invocation": {
+            "parameters": {
+                "methodName": "GetDesktopAppIPCClientLocations",
+                "serializedParams": {"host_os": host_os},
+            }
+        }
+    })
+
+    try:
+        locations = json.loads(locations_raw)
+    except Exception as e:
+        raise RuntimeError(f"failed to parse core response: {e}")
+
+    for lib_path in locations:
+        if os.path.exists(lib_path):
+            return lib_path
+
+    raise FileNotFoundError("1Password desktop application not found")
+
+class DesktopCore:
+    def __init__(self):
+        # Determine the path to the desktop app.
+        path = find_1password_lib_path()
+
+        self.lib = ctypes.CDLL(path)
+
+        # Bind the Rust-exported functions
+        self.send_message = self.lib.op_sdk_ipc_send_message
+        self.send_message.argtypes = [
+            POINTER(c_uint8),             # msg_ptr
+            c_size_t,                     # msg_len
+            POINTER(POINTER(c_uint8)),    # out_buf
+            POINTER(c_size_t),            # out_len
+            POINTER(c_size_t),            # out_cap
+        ]
+        self.send_message.restype = c_int32
+
+        self.free_message = self.lib.op_sdk_ipc_free_response
+        self.free_message.argtypes = [POINTER(c_uint8), c_size_t, c_size_t]
+        self.free_message.restype = None
+
+    def call_shared_library(self, payload: bytes) -> bytes:
+        out_buf = POINTER(c_uint8)()
+        out_len = c_size_t()
+        out_cap = c_size_t()
+
+        ret = self.send_message(
+            (ctypes.cast(payload, POINTER(c_uint8))),
+            len(payload),
+            byref(out_buf),
+            byref(out_len),
+            byref(out_cap),
+        )
+
+        if ret != 0:
+            raise RuntimeError(f"send_message failed with code {ret}")
+
+        # Copy bytes into Python
+        data = ctypes.string_at(out_buf, out_len.value)
+
+        # Free memory via Rust's exported function
+        self.free_message(out_buf, out_len, out_cap)
+
+        return data
+
+    def init_client(self, config: dict) -> int:
+        payload = json.dumps(config).encode("utf-8")
+        resp = self.call_shared_library(payload)
+        return json.loads(resp)
+
+    def invoke(self, invoke_config: dict) -> str:
+        payload = json.dumps(invoke_config).encode("utf-8")
+        resp = self.call_shared_library(payload)
+        return resp.decode("utf-8")
+
+    def release_client(self, client_id: int):
+        payload = json.dumps(client_id).encode("utf-8")
+        try:
+            self.call_shared_library(payload)
+        except Exception as e:
+            print(f"failed to release client: {e}")