Skip to content
Merged
117 changes: 117 additions & 0 deletions examples/client/client_set_operator_with.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,117 @@
"""
Set up a client whose operator signs through a callback instead of a private key.

`Client.set_operator_with(account_id, public_key, signer)` lets the operator key
stay in an HSM or a cloud KMS. The client then uses the signer to pay for every
transaction and query it executes.

This example simulates the HSM with a callback that wraps OPERATOR_KEY.

Usage:
uv run examples/client/client_set_operator_with.py
python examples/client/client_set_operator_with.py
"""

import os
import sys

from dotenv import load_dotenv

from hiero_sdk_python import (
AccountCreateTransaction,
AccountId,
AccountInfoQuery,
Client,
Hbar,
Network,
PrivateKey,
ResponseCode,
Signer,
)


load_dotenv()

NETWORK_NAME = os.getenv("NETWORK", "testnet").lower()
OPERATOR_ID = os.getenv("OPERATOR_ID")
OPERATOR_KEY = os.getenv("OPERATOR_KEY")


def simulated_hsm(private_key: PrivateKey) -> Signer:
"""
Return a signer callback standing in for an HSM or KMS.

A real signer would send body_bytes to the HSM and return the raw signature.
"""

def signer(body_bytes: bytes) -> bytes:
print(f" HSM asked to sign {len(body_bytes)} bytes")
return private_key.sign(body_bytes)

return signer


def setup_client() -> Client:
"""Create a client whose operator signs through the simulated HSM."""
if not OPERATOR_ID or not OPERATOR_KEY:
raise ValueError("OPERATOR_ID and OPERATOR_KEY environment variables are required")

operator_id = AccountId.from_string(OPERATOR_ID)
operator_key = PrivateKey.from_string(OPERATOR_KEY)

# Only the public key and the callback reach the client
client = Client(Network(NETWORK_NAME)).set_operator_with(
operator_id, operator_key.public_key(), simulated_hsm(operator_key)
)

print(f"Network: {client.network.network}")
print(f"Client initialized with signer-based operator {client.operator_account_id}")
print(f"Operator private key held by the client: {client.operator_private_key}")
return client


def create_account(client: Client) -> AccountId:
"""Create an account; the operator pays and signs through the HSM."""
print("Creating an account...")
receipt = (
AccountCreateTransaction()
.set_key_without_alias(PrivateKey.generate_ed25519().public_key())
.set_initial_balance(Hbar(1))
.execute(client)
)
if receipt.status != ResponseCode.SUCCESS:
raise RuntimeError(f"Account creation failed with status: {ResponseCode(receipt.status).name}")

print(f"Created account {receipt.account_id}")
return receipt.account_id


def query_account_info(client: Client, account_id: AccountId) -> None:
"""Run a paid query; the query payment is signed through the HSM too."""
print("Querying account info...")
info = AccountInfoQuery().set_account_id(account_id).execute(client)
print(f"Account {info.account_id} has a balance of {info.balance}")


def main():
"""
1. Set up a client with set_operator_with() and a simulated HSM signer.

2. Execute a transaction paid for by the signer-based operator.
3. Execute a paid query, whose payment is signed by the same signer.
"""
try:
client = setup_client()

account_id = create_account(client)
query_account_info(client, account_id)

client.close()

except Exception as exc:
print(f"Error: {exc}")
sys.exit(1)


if __name__ == "__main__":
main()
105 changes: 105 additions & 0 deletions examples/transaction/transaction_sign_with_external_signer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,105 @@
"""
Sign a transaction with an external signer instead of a private key.

`sign_with(public_key, signer)` hands each transaction body to a callback and
attaches the signature it returns. This is how keys held in an HSM or a cloud KMS
sign transactions: the private key never enters the SDK.

This example simulates the HSM with a callback that wraps a locally generated key.

Usage:
uv run examples/transaction/transaction_sign_with_external_signer.py
python examples/transaction/transaction_sign_with_external_signer.py
"""

import sys

from hiero_sdk_python import (
AccountCreateTransaction,
AccountId,
Client,
Hbar,
PrivateKey,
ResponseCode,
Signer,
TransferTransaction,
)


def setup_client() -> Client:
"""Initialize the client from the OPERATOR_ID and OPERATOR_KEY environment variables."""
client = Client.from_env()
print(f"Network: {client.network.network}")
print(f"Client initialized with operator {client.operator_account_id}")
return client


def simulated_hsm(private_key: PrivateKey) -> Signer:
"""
Return a signer callback standing in for an HSM or KMS.

A real signer would send body_bytes to the HSM and return the raw signature:
64 bytes for Ed25519, or the 64-byte r || s over keccak256(body_bytes) for ECDSA.
"""

def signer(body_bytes: bytes) -> bytes:
print(f" HSM asked to sign {len(body_bytes)} bytes")
return private_key.sign(body_bytes)

return signer


def create_account(client: Client, key: PrivateKey) -> AccountId:
"""Create an account controlled by the HSM key."""
receipt = (
AccountCreateTransaction().set_key_without_alias(key.public_key()).set_initial_balance(Hbar(2)).execute(client)
)
if receipt.status != ResponseCode.SUCCESS:
raise RuntimeError(f"Account creation failed with status: {ResponseCode(receipt.status).name}")

print(f"Created account {receipt.account_id} with an ECDSA key held by the simulated HSM")
return receipt.account_id


def transfer_with_external_signer(client: Client, account_id: AccountId, key: PrivateKey) -> None:
"""Transfer HBAR out of the HSM account, signing through the callback."""
tx = (
TransferTransaction()
.add_hbar_transfer(account_id, -Hbar(1).to_tinybars())
.add_hbar_transfer(client.operator_account_id, Hbar(1).to_tinybars())
.freeze_with(client)
)

# The signer is called once per transaction body, i.e. once per node
tx.sign_with(key.public_key(), simulated_hsm(key))
print(f"Transaction signed by the HSM key: {tx.is_signed_by(key.public_key())}")

# The operator pays the fee, so execute() adds the operator signature
receipt = tx.execute(client)
if receipt.status != ResponseCode.SUCCESS:
raise RuntimeError(f"Transfer failed with status: {ResponseCode(receipt.status).name}")

print(f"Transfer succeeded: {receipt.status == ResponseCode.SUCCESS}")


def main():
"""
1. Set up a client with an operator.

2. Create an account whose key lives in a (simulated) HSM.
3. Transfer HBAR from that account, signing with sign_with() instead of sign().
"""
try:
client = setup_client()
hsm_key = PrivateKey.generate_ecdsa()

account_id = create_account(client, hsm_key)
transfer_with_external_signer(client, account_id, hsm_key)

except Exception as exc:
print(f"Error: {exc}")
sys.exit(1)


if __name__ == "__main__":
main()
2 changes: 2 additions & 0 deletions src/hiero_sdk_python/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@
from .crypto.evm_address import EvmAddress
from .crypto.private_key import PrivateKey
from .crypto.public_key import PublicKey
from .crypto.signer import Signer

# Duration
from .Duration import Duration
Expand Down Expand Up @@ -193,6 +194,7 @@
# Crypto
"PrivateKey",
"PublicKey",
"Signer",
"EvmAddress",
# Tokens
"TokenCreateTransaction",
Expand Down
76 changes: 69 additions & 7 deletions src/hiero_sdk_python/client/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@

from hiero_sdk_python.account.account_id import AccountId
from hiero_sdk_python.crypto.private_key import PrivateKey
from hiero_sdk_python.crypto.public_key import PublicKey
from hiero_sdk_python.crypto.signer import Signer
from hiero_sdk_python.hapi.mirror import (
consensus_service_pb2_grpc as mirror_consensus_grpc,
)
Expand All @@ -35,10 +37,17 @@


class Operator(NamedTuple):
"""A named tuple for the operator's account ID and private key."""
"""
A named tuple for the operator's account ID and signing credentials.

private_key is None when the operator was set with Client.set_operator_with().
public_key and signer are always set on an Operator returned by Client.operator.
"""

account_id: AccountId
private_key: PrivateKey
private_key: PrivateKey | None
public_key: PublicKey | None = None
signer: Signer | None = None


class Client:
Expand All @@ -51,6 +60,8 @@ def __init__(self, network: Network = None) -> None:
"""
self.operator_account_id: AccountId = None
self.operator_private_key: PrivateKey = None
self._operator_public_key: PublicKey | None = None
self._operator_signer: Signer | None = None

if network is None:
network = Network()
Expand Down Expand Up @@ -186,21 +197,72 @@ def for_network(cls, network_map: dict[str, AccountId], network_name: str | None
nodes = [_Node(account_id, address, None) for address, account_id in network_map.items()]
return cls(Network(network=network_name, nodes=nodes))

def set_operator(self, account_id: AccountId, private_key: PrivateKey) -> None:
"""Sets the operator credentials (account ID and private key)."""
self.operator_account_id = account_id
def set_operator(self, account_id: AccountId, private_key: PrivateKey) -> Client:
"""
Sets the operator credentials (account ID and private key).

Args:
account_id (AccountId): The operator account, which pays for transactions and queries.
private_key (PrivateKey): The operator account's private key.

Returns:
Client: This client instance for fluent chaining.
"""
self.set_operator_with(account_id, private_key.public_key(), private_key.sign)
self.operator_private_key = private_key
Comment thread
danielmarv marked this conversation as resolved.
return self

def set_operator_with(self, account_id: AccountId, public_key: PublicKey, signer: Signer) -> Client:
"""
Sets the operator with a signer callback instead of a private key.

Use this when the operator key is held outside the process, e.g. in an HSM or KMS.
The signer receives the canonical body bytes and must return the raw signature.
After this call, operator_private_key is None.

Args:
account_id (AccountId): The operator account, which pays for transactions and queries.
public_key (PublicKey): The public key matching the signer.
signer (Signer): The signing callback.

Returns:
Client: This client instance for fluent chaining.
"""
self.operator_account_id = account_id
self.operator_private_key = None
self._operator_public_key = public_key
self._operator_signer = signer
return self

@property
def operator_public_key(self) -> PublicKey | None:
"""The operator's public key, or None if no operator signer is set."""
if self.operator_private_key is not None:
return self.operator_private_key.public_key()
return self._operator_public_key

@property
def operator_signer(self) -> Signer | None:
"""The operator's signing callback, or None if no operator signer is set."""
if self.operator_private_key is not None:
return self.operator_private_key.sign
return self._operator_signer

@property
def operator(self) -> Operator | None:
"""
Returns an Operator namedtuple if both account ID and private key are set,
Returns an Operator namedtuple if both the account ID and a signer are set,
otherwise None.
"""
if self.operator_account_id and self.operator_private_key:
public_key = self.operator_public_key
signer = self.operator_signer

if self.operator_account_id and public_key is not None and signer is not None:
return Operator(
account_id=self.operator_account_id,
private_key=self.operator_private_key,
public_key=public_key,
signer=signer,
)
return None

Expand Down
Loading
Loading