TSP Python bindings

We use PyO3 to generate Python bindings for the TSP SDK. We recommend using uv to manage your Python dependencies, but it is also possible to use pyenv with maturin manually.

To add TSP as a dependency to your uv project, use the following command:

uv add git+https://github.com/openwallet-foundation-labs/tsp#subdirectory=tsp_python

Example usage

Here's an example showing how you can use the Python bindings to create an identity, resolve someone else's identity, and send and receive TSP messages:

import tsp_python as tsp
import random
import requests

store = tsp.SecureStore()

# Create identity
print("Creating identity")
username = "alice" + str(random.randint(0, 999999))
did = "did:web:did.teaspoon.world:endpoint:" + username
transport_url = "https://p.teaspoon.world/endpoint/" + did
alice = tsp.OwnedVid.bind(did, transport_url)

# Publish DID (this is non-standard and dependents on the implementation of the DID support server)
print("Publishing DID: " + did)
response = requests.post(
    "https://did.teaspoon.world/add-vid",
    data=alice.json(),
    headers={"Content-type": "application/json"},
)
if not response.ok:
    raise Exception(f"Could not publish DID (status code: {response.status_code})")

# Save DID in wallet
store.add_private_vid(alice)

# Resolve other party (may fail if endpoint bob does not exist)
bob_did = "did:web:did.teaspoon.world:endpoint:bob"
print("Resolving DID: " + bob_did)
bob_endpoint = store.verify_vid(bob_did)
print("> Bob's endpoint:", bob_endpoint)

# Send a message
print("Sending message...")
store.send(alice.identifier(), bob_did, b"hi bob")

# Receive messages
print("Listening for messages...")
while True:
    response = store.receive(alice.identifier())
    print("Received: " + str(response))
    if isinstance(response, tsp.GenericMessage):
        print("> " + str(response.message))

Creating identities

The OwnedVid class is used to create and manage identities. It supports did:web, did:webvh, and did:peer identifiers:

class OwnedVid:
    """Class for managing VIDs for which we own the private keys"""

    @staticmethod
    def new_did_peer(url: str) -> OwnedVid:
        """Create a `did:peer` for a particular end-point"""
        ...

    @staticmethod
    def new_did_webvh(did_name: str, transport: str) -> tuple[OwnedVid, str]:
        """Create a `did:webvh` for a name and a transport URL"""
        ...

    @staticmethod
    def bind(did: str, transport_url: str) -> OwnedVid:
        """Create a `did:web` by binding a DID to a transport URL"""
        ...

    def json(self) -> str:
        """Get a JSON representation of the VID"""
        ...

    def identifier(self) -> str:
        """Get the DID"""
        ...

    def endpoint(self) -> str:
        """Get the transport URL"""
        ...

To store a created identity, use the add_private_vid method of your SecureStore.

Secure Store

The SecureStore class exposes the functionality from the Rust AsyncSecureStore, including methods to open and seal messages, and to send and receive messages. The Python SecureStore is connected to a secure storage wallet to store the state, the location of which can be configured during initialization of the SecureStore:

class SecureStore:
    inner: tsp_python.Store

    def __init__(self, wallet_url="sqlite://wallet.sqlite", wallet_password="unsecure"):
        self.inner = tsp_python.Store(wallet_url, wallet_password)

Opening and sealing TSP messages

The open_message and seal_message functions of a SecureStore can be used to open and seal TSP messages:

    def seal_message(
        self,
        sender: str,
        receiver: str,
        message: bytes,
        nonconfidential_data: bytes | None = None,
    ) -> tuple[str, bytes]:
        """
        Seal a TSP message.

        The message is encrypted, encoded, and signed using the key material
        of the sender and receiver, specified by their VIDs.
        """
        with Wallet(self):
            return self.inner.seal_message(
                sender, receiver, message, nonconfidential_data
            )

    def open_message(self, message: bytes):
        """Decode an encrypted `message`"""
        with Wallet(self):
            flat_message = self.inner.open_message(message)
            return ReceivedTspMessage.from_flat(flat_message)

Sending and receiving TSP messages

The send and receive functions of a SecureStore can be used to send and receive TSP messages using the appropriate transport:

    def send(
        self,
        sender: str,
        receiver: str,
        message: bytes,
        nonconfidential_data: bytes | None = None,
    ):
        """
        Send a TSP message given earlier resolved VIDs

        Encodes, encrypts, signs, and sends a TSP message
        """
        with Wallet(self):
            self.inner.send(sender, receiver, message, nonconfidential_data)

    def receive(self, vid: str):
        """Receive a single TSP messages for the private VID identified by `vid`, using the appropriate transport mechanism for it."""
        with Wallet(self):
            message = self.inner.receive(vid)
            return None if message is None else ReceivedTspMessage.from_flat(message)

Managing VIDs

The SecureStore provides a number of functions to manage VIDs. When adding a VID to the wallet (using add_private_vid for your own VIDs, or verify_vid for VIDs from others), it is also possible to set aliases for them, which makes it easy to identify VIDs stored in the wallet.

    def verify_vid(self, did: str, alias: str | None = None) -> str:
        """Resolve DID document, verify it, add vid to the wallet, and its return endpoint"""
        with Wallet(self):
            return self.inner.verify_vid(did, alias)

    def add_private_vid(
        self,
        vid: OwnedVid,
        alias: str | None = None,
        metadata: dict[Any, Any] | None = None,
    ):
        """Adds a private `vid` to the wallet"""
        with Wallet(self):
            self.inner.add_private_vid(vid, alias, metadata)

    def forget_vid(self, vid: str):
        """Remove a `vid` from the wallet"""
        with Wallet(self):
            self.inner.forget_vid(vid)

    def resolve_alias(self, alias: str) -> str | None:
        """Resolve alias to its corresponding DID (if it exists in the wallet)"""
        with Wallet(self):
            return self.inner.resolve_alias(alias)