Metadata-Version: 2.4
Name: aplanesdk
Version: 0.39.1
Summary: Python SDK for APlane transaction signing
Author: APlane Project LLC
License-Expression: MIT
Project-URL: Homepage, https://github.com/aplane-algo/aplanesdk
Project-URL: Documentation, https://github.com/aplane-algo/aplanesdk/tree/main/python
Project-URL: Repository, https://github.com/aplane-algo/aplanesdk
Keywords: algorand,blockchain,signing,post-quantum,falcon
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Security :: Cryptography
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: py-algorand-sdk>=2.11.0
Requires-Dist: paramiko>=3.0.0
Requires-Dist: pyyaml>=6.0
Requires-Dist: requests>=2.25.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0; extra == "dev"
Requires-Dist: black>=23.0; extra == "dev"
Requires-Dist: mypy>=1.0; extra == "dev"
Dynamic: license-file

# APlane Python SDK

Python SDK for signing Algorand transactions via apsigner.

## Versioning

SDK packages are published only when the SDK changes. SDK versions track
compatible APlane release tags and may skip product release numbers.

The native-Falcon SDK release line requires an APlane signer release after
`v0.35.0`; see [`docs/COMPATIBILITY.md`](../docs/COMPATIBILITY.md).

## Installation

```bash
pip install aplanesdk
```

The published package is `aplanesdk` on PyPI.

For the optional AlgoKit adapter, install the current AlgoKit Utils Python v5
beta in the same environment. The adapter is tested against 5.0.0b5:

```bash
pip install --pre 'algokit-utils==5.0.0b5'
```

Or install from source:

```bash
cd python
pip install -e .
```

## Quick Start

```python
from aplanesdk import SignerClient, send_raw_transaction
from algosdk import transaction
from algosdk.v2client import algod

# Connect to signer (reads endpoints.yaml and the enrolled SSH key from data dir)
client = SignerClient.from_env()

# Build transaction with algosdk
algod_client = algod.AlgodClient("", "https://testnet-api.4160.nodely.dev")
params = algod_client.suggested_params()

txn = transaction.PaymentTxn(
    sender="SENDER_ADDRESS",
    sp=params,
    receiver="RECEIVER_ADDRESS",
    amt=1000000  # 1 ALGO
)

# Sign via apsigner (waits for operator approval)
signed = client.sign_transaction(txn)

# Submit to network (signed is ready to use, no processing needed)
txid = send_raw_transaction(algod_client, signed)
print(f"Submitted: {txid}")
```

## Connection Methods

Data-directory connections use `endpoints.yaml`. SSH endpoints create a
managed tunnel; HTTPS and loopback HTTP endpoints connect directly.
Managed SSH keys and host trust live under the APlane client data directory;
the SDK does not use the operating-system user's personal SSH directory.

### Environment-Based Connection (Recommended)

Load configuration from a data directory. The directory is required — pass
`data_dir` or set the `APCLIENT_DATA` environment variable:

```python
# Set environment variable
# export APCLIENT_DATA=~/aplane/apclient

client = SignerClient.from_env()

# Or pass directly
client = SignerClient.from_env(data_dir="~/aplane/apclient")
```

Data directory structure (installer default: `~/aplane/apclient`):
```
<data_dir>/
  config.yaml          # Non-routing client settings
  endpoints.yaml       # Signer and cosigner routing
  .ssh/
    id_ed25519         # SSH private key: the client's credential
    known_hosts        # Trusted server host keys
```

Example `endpoints.yaml`:
```yaml
schema_version: 2
default: primary
endpoints:
  primary:
    role: signer
    url: ssh://localhost:1127
    identity_file: .ssh/id_ed25519
    known_hosts_path: .ssh/known_hosts
```

### Direct SSH Connection

Connect explicitly via SSH tunnel, authenticated by the enrolled client key:

```python
client = SignerClient.connect_ssh(
    host="signer.example.com",
    ssh_key_path="~/aplane/apclient/.ssh/id_ed25519",
    known_hosts_path="~/aplane/apclient/.ssh/known_hosts",
    ssh_port=1127,                # default: 1127
    timeout=30,                   # optional explicit shorter request timeout
    ssh_setup_timeout=60,         # TCP dial + SSH authentication only
)
```

SSH setup uses its own 60-second deadline, independent of HTTP request and
operator approval timeouts. The deadline is cleared after authentication.
`request_enrollment()` and `request_enrollment_from_env()` expose the same
setup-only deadline as `setup_timeout`.

**Note**: SSH uses the fixed non-secret username `aplane` and public-key
authentication only; the enrolled key is the client's only credential. Keys
are enrolled via the `request-enrollment` operator-approved flow.

The SSH tunnel is established automatically. Remember to close when done:

```python
client.close()
```

Or use as a context manager:

```python
with SignerClient.connect_ssh(
    host="...",
    ssh_key_path="~/aplane/apclient/.ssh/id_ed25519",
    known_hosts_path="~/aplane/apclient/.ssh/known_hosts",
) as client:
    signed = client.sign_transaction(txn)
# Tunnel closed automatically
```

## Authentication

The client's SSH key is its credential. A new key is enrolled once: the
request is queued for the operator to approve later in `apadmin`
(`pending=True`, the normal outcome) or the key is already enrolled, and
nothing is stored on the client. Connect once the operator has approved.
`request_enrollment_from_env(endpoint="cosigner.qa", label="ci-runner")`
selects a named endpoint; without an alias it selects the default signer:

```python
from aplanesdk import request_enrollment_from_env

result = request_enrollment_from_env(label="ci-runner")
if result.pending:
    print(f"key {result.fingerprint} queued; have the operator approve it in apadmin")
```

`request_enrollment(host, ssh_key_path, known_hosts_path=..., label=...)` is
the explicit form. The same enrollment can be done from `apshell` with
`request-enrollment`.

## API Reference

### SignerClient

#### `health() -> bool`

Check if signer is reachable.

```python
if client.health():
    print("Signer is online")
```

#### `get_status() -> StatusResponse`

Fetch authenticated signer status. This works while the signer is locked.

```python
status = client.get_status()
print(status.state, status.keyset_revision, status.warnings)
```

`keyset_revision` is process-local and useful for deciding when to refresh
`list_keys(refresh=True)`; it is not durable across apsigner restarts.
`approval_wait_seconds` is used by the SDK to size `/sign` deadlines. Display
non-empty `warnings` to operators; they report persistent health conditions
that require attention.

#### `list_keys() -> List[KeyInfo]`

List available signing keys.

```python
keys = client.list_keys()
for key in keys:
    print(f"{key.address} [{key.key_type}]")
```

Returns list of `KeyInfo`:
- `address`: Algorand address
- `key_type`: "ed25519", "aplane.falcon1024.v1", "aplane.htlc.v1", etc.
- `logic_sig_resources`: independent program-byte, argument-byte, and
  maximum-opcode-cost demand by authorization path.
- `is_generic_lsig`: True if no cryptographic signature needed
- `signing_args`: List of `SigningArg` for LogicSigs (name, arg_type, description)

The SDK exposes bounded inventory and ordinary spend signing only. It does not
build, partially sign, or complete contract-admin rekey transactions; use the
APlane `aprekey` workflow for those operations.

**Discovering required arguments for generic LogicSigs:**

```python
key_info = client.get_key_info(hashlock_address)
if key_info.signing_args:
    for arg in key_info.signing_args:
        print(f"{arg.name}: {arg.arg_type} - {arg.description}")
```

#### `sign_transaction(txn, auth_address=None, lsig_args=None) -> str`

Sign a single transaction. Returns a base64-encoded string ready for submission.

The server automatically handles fee pooling for large LogicSigs (e.g., Falcon-1024) by adding dummy transactions as needed.

```python
# Basic signing (uses txn.sender as auth_address)
signed = client.sign_transaction(txn)

# Rekeyed account (different auth key)
signed = client.sign_transaction(txn, auth_address="SIGNER_KEY_ADDRESS")

# Generic LogicSig with runtime args (e.g., HTLC)
signed = client.sign_transaction(
    txn,
    auth_address="HASHLOCK_ADDRESS",
    lsig_args={"preimage": b"secret_value"}
)

# Submit directly (no processing needed)
txid = send_raw_transaction(algod_client, signed)
```

#### `sign_transactions(txns, auth_addresses=None, lsig_args_map=None) -> str`

Sign multiple transactions as a group. Returns a base64-encoded string of concatenated signed transactions, ready for submission.

**Important**: Do NOT pre-assign group IDs. The server computes the group ID after adding any required dummy transactions for large LogicSigs.

```python
# Build transactions (do NOT call assign_group_id)
txn1 = transaction.PaymentTxn(sender=addr1, sp=params, receiver=addr2, amt=100000)
txn2 = transaction.PaymentTxn(sender=addr2, sp=params, receiver=addr1, amt=100000)

# Sign group (server handles grouping and dummies)
signed = client.sign_transactions([txn1, txn2])

# Submit directly (no processing needed)
txid = algod_client.send_raw_transaction(signed)
```

#### `sign_transactions_list(txns, auth_addresses=None, lsig_args_map=None) -> List[str]`

Like `sign_transactions()` but returns individual base64-encoded transactions instead of concatenated. Useful when you need to inspect transactions individually.

```python
signed_list = client.sign_transactions_list([txn1, txn2])
# signed_list is List[str], each element is a base64-encoded signed transaction
```

#### `sign_requests(sign_entries, request_id=None) -> GroupSignResponse`

Send one or more raw `/sign` request entries. Use this when an integration
already owns transaction encoding and wants APlane's native response shape.

```python
response = client.sign_requests(
    [{
        "txn_bytes_hex": "5458...",
        "auth_address": "SIGNER_KEY_ADDRESS",
        "txn_sender": "SENDER_ADDRESS",  # advisory display hint only
    }],
    request_id="app-owned-request-id",
)
```

### AlgoKit Utils Adapter

For AlgoKit Utils Python v5 transaction composers, use the adapter
account. It connects AlgoKit clients to APlane's transaction signing functions
and presents the `addr` + `signer(txn_group, indexes_to_sign)` shape.

The minimal repository example is `examples/algokit_self_send.py`. From a
checkout, run it as a module so it imports the local SDK source:

```bash
cd ~/aplanesdk/python
export APCLIENT_DATA=~/aplane/apclient
export APLANE_ADDRESS=SENDER_ADDRESS
python -m examples.algokit_self_send
```

The example builds a transaction with AlgoKit, signs it through the APlane
adapter, then submits the signed blobs with AlgoKit's algod client:

```python
from algokit_utils import AlgoAmount, AlgorandClient, PaymentParams
from aplanesdk import SignerClient
from aplanesdk.algokit import create_apsigner_account

sender = "SENDER_ADDRESS"
algorand = AlgorandClient.testnet()

with SignerClient.from_env() as signer:
    auth = algorand.client.algod.account_information(sender).auth_addr or sender
    account = create_apsigner_account(signer, sender, auth_address=auth)
    txn = algorand.create_transaction.payment(
        PaymentParams(
            sender=sender,
            signer=account,
            receiver=sender,
            amount=AlgoAmount(micro_algo=0),
            validity_window=1000,
        )
    )
    signed = account.signer([txn], [0])
    tx_id = algorand.client.algod.send_raw_transaction(signed).tx_id
```

Use `create_transaction.*` when APlane must own final signing and any
APlane-managed group expansion. `algorand.send.*` owns the composer send path
and signs inside that path.

The Python AlgoKit signer is synchronous. `ApsignerAccount` tracks one active
signing request at a time; overlapping calls on the same account raise
`RuntimeError`. Use separate account objects for concurrent signing. For
asyncio applications, run the AlgoKit call site in a worker thread, for example
with `asyncio.to_thread(...)`. If an application needs to own request IDs, pass
`new_request_id`, a callable that returns a fresh ID for each sign call.

Signing calls discover `/status.approval_wait_seconds` and use that value
plus 30 seconds of slack for the request timeout. If discovery fails or an older
signer omits the field, signing falls back to 6 minutes. An explicit shorter
timeout still wins; SDK `/sign` calls include a `request_id` and send a
best-effort `/sign/cancel` when the HTTP request times out or disconnects.
High-level signing methods accept an optional keyword-only `request_id`.
AlgoKit adapter callers can call `account.cancel()` from another thread to
cancel the in-flight adapter request.

#### `cancel_sign_request(request_id) -> CancelSignResponse`

Ask apsigner to cancel a live synchronous `/sign` request by request ID.
Successful responses are idempotent for client behavior and return state
`"canceled"` or `"not_found"`.

Python high-level signing generates a request ID by default. Interactive
applications can pass an application-owned ID, then call
`cancel_sign_request()` with the same value from another thread:

```python
request_id = "wallet-ui-approval-123"
signed = client.sign_transaction(txn, request_id=request_id)
# elsewhere, if the user aborts while approval is pending:
client.cancel_sign_request(request_id)
```

For AlgoKit adapter signing, call `cancel()` on the account:

```python
account.cancel()
```

#### `close()`

Close the client and SSH tunnel (if any).

```python
client.close()
```

## Supported Key Types

| Key Type | Description | Notes |
|----------|-------------|-------|
| `ed25519` | Native Algorand keys | Standard signing |
| `aplane.falcon1024.v1` | Post-quantum LogicSig | Signature in LogicSig.Args[0] |
| `aplane.ed25519.v1` | Ed25519 DSA LogicSig | Library-visible plain DSA account |
| `aplane.witness-falcon1024.v1` | Witness key | Cosigner-custodied policy signature key; not a spending account |
| `aplane.falcon1024-cosigner1024.v1` | Guarded account | Requires user and cosigner component signatures |
| `aplane.corridor.v1` | Bounded Corridor account | `bounded1` contract; `bounded-cosigner1` spend flow |
| `aplane.falcon1024-allowlist.v1` | Bounded allowlist | Inline allowlist; `bounded1` signing flow |
| `aplane.falcon1024-allowlist.v2` | Bounded allowlist | Merkle allowlist; `bounded1` signing flow |
| `aplane.falcon1024-timelock.v1` | Bounded timelock | Round-gated `bounded1` signing flow |
| `aplane.falcon1024-allowlist-alock.v1` | Rekey-locked bounded allowlist | Ordinary spending uses `bounded1`; admin rekey is outside SDK scope |
| `aplane.htlc.v1` | Hash-locked funds | Requires `preimage` arg (check `signing_args`) |

The server assembles the complete signed transaction - the SDK returns a base64 string ready for submission.

## Cosigner And Guarded Accounts

Witness keys enrolled as cosigners are public policy-signature selectors, not Algorand
spending accounts. Do not use them as senders, receivers, auth addresses, or
rekey targets. Guarded account keys must be signed through the guarded flow.

Low-level endpoint wrappers are available:

```python
from aplanesdk import (
    AssemblyRequest,
    AssemblyTarget,
    ComponentRequest,
    COMPONENT_TARGET_KIND_COSIGNER,
    COMPONENT_TARGET_KIND_USER,
)

user_part = user_client.request_components(ComponentRequest(
    group_bytes_hex=["5458..."],
    targets=[{
        "target_index": 0,
        "kind": COMPONENT_TARGET_KIND_USER,
        "auth_address": "GUARDED_ACCOUNT_ADDRESS",
    }],
))

cosigner_part = cosigner_client.request_components(ComponentRequest(
    group_bytes_hex=["5458..."],
    targets=[{
        "target_index": 0,
        "kind": COMPONENT_TARGET_KIND_COSIGNER,
        "component_key": "COSIGNER_COMPONENT_SELECTOR",
    }],
))

assembled = user_client.request_assemble(AssemblyRequest(
    group_bytes_hex=["5458..."],
    targets=[AssemblyTarget(
        target_index=0,
        kind=ASSEMBLY_TARGET_KIND_GUARDED,
        auth_address="GUARDED_ACCOUNT_ADDRESS",
        user_signature=user_part.components[0]["signature"],
        cosigner_signature=cosigner_part.components[0]["signature"],
    )],
))
```

For the common explicit two-client flow, use `sign_guarded_group`. The direct
helper does not perform inventory discovery, so pass the reviewed spend-path
resource profile returned by `list_keys()`:

```python
result = sign_guarded_group(
    user_client=user_client,
    cosigner_client=cosigner_client,
    cosigner_component_key="COSIGNER_COMPONENT_SELECTOR",
    group_bytes_hex=["5458..."],
    guarded_targets=[
        GuardedSignTarget(
            target_index=0,
            guarded_account="GUARDED_ACCOUNT_ADDRESS",
            logic_sig_resources=reviewed_spend_resources,
        ),
    ],
)
signed_group = result.signed_group
```

`assemble_group()` remains the local multi-party concatenation helper; it is
not the same operation as server-side guarded assembly.

### Bounded Cosigner Accounts

Corridor uses the bounded contract `bounded1` with the distinct
`bounded-cosigner1` online signing flow. The contract identifies the LogicSig
rules; the flow identifies the user-first multi-endpoint choreography. The
prepared helper detects that flow from signer inventory and routes it
automatically:

```python
result = sign_prepared_guarded_group(
    user_client=user_client,
    cosigner_resolver=cosigner_resolver,
    prepared_group=prepared_group,
)
signed_group = result.signed_group
```

The SDK first freezes the complete canonical group through `/plan`; the user
signer then approves those bytes through `request_components()` with
`kind="bounded-base"`. It then signs ordinary positions, and only after
the user side is complete does it request cosigner signatures over the same
bytes and call `request_assemble()`. Before
signing anything, the SDK compares the
signer-produced plan with the caller's prepared group: only reported fee
pooling and group-ID assignment are accepted, and appended positions must be
canonical budget dummies. The returned group must use canonical transaction
encoding and a group ID recomputed from the presented membership. The SDK also
verifies ordinary signed positions and every assembled transaction against the
frozen transaction bytes.

`request_components()` sends best-effort `/sign/cancel` when its
approval-bearing HTTP request times out or disconnects.
`request_assemble()` does not open an approval request and is not a
cancellation handle.

The signer planner owns fee selection, authorization-resource sizing, and any
reported group mutations for both guarded flows.

Applications that own orchestration can call `request_components()` and
`request_assemble()` directly. Cosigner authorization is spend-only in
this contract; bounded contract-admin rekeys remain an external `aprekey`
ceremony and are not completed by the SDK.

## Error Handling

### Signing Exceptions

```python
from aplanesdk import (
    SignerError,
    AuthenticationError,
    SigningRejectedError,
    SignerUnavailableError,
    KeyNotFoundError
)

try:
    signed = client.sign_transaction(txn)
except AuthenticationError:
    print("Connection is not authenticated by an enrolled SSH key")
except SigningRejectedError:
    print("Operator rejected the request")
except SignerUnavailableError:
    print("Signer not reachable or locked")
except KeyNotFoundError:
    print("Key not found in signer")
except SignerError as e:
    print(f"Signing failed: {e}")
```

### Submission Exceptions

`send_raw_transaction()` wraps verbose algod errors into clean exceptions:

```python
from aplanesdk import (
    send_raw_transaction,
    TransactionRejectedError,
    LogicSigRejectedError,
    InsufficientFundsError,
    InvalidTransactionError
)

try:
    txid = send_raw_transaction(algod_client, signed)
except LogicSigRejectedError as e:
    print(f"LogicSig failed: {e.reason}")  # e.txid also available
except InsufficientFundsError as e:
    print(f"Not enough funds: {e.reason}")
except InvalidTransactionError as e:
    print(f"Invalid transaction: {e.reason}")
except TransactionRejectedError as e:
    print(f"Rejected: {e.reason}")
```

## Example: Complete Workflow

```python
#!/usr/bin/env python3
from aplanesdk import SignerClient, SignerError, send_raw_transaction
from algosdk import transaction
from algosdk.v2client import algod

def main():
    # Connect via SSH, authenticated by the enrolled client key
    with SignerClient.connect_ssh(
        host="signer.example.com",
        ssh_key_path="~/aplane/apclient/.ssh/id_ed25519",
        known_hosts_path="~/aplane/apclient/.ssh/known_hosts",
    ) as client:

        # List keys
        keys = client.list_keys()
        sender = keys[0].address
        print(f"Using: {sender}")

        # Build transaction
        algod_client = algod.AlgodClient("", "https://testnet-api.4160.nodely.dev")
        params = algod_client.suggested_params()

        txn = transaction.PaymentTxn(
            sender=sender,
            sp=params,
            receiver=sender,
            amt=0
        )

        # Sign (will wait for operator approval)
        try:
            signed = client.sign_transaction(txn)
            print("Signed!")

            # Submit directly (no processing needed)
            txid = send_raw_transaction(algod_client, signed)
            print(f"TxID: {txid}")

            # Wait for confirmation
            result = transaction.wait_for_confirmation(algod_client, txid, 4)
            print(f"Confirmed in round {result['confirmed-round']}")

        except SignerError as e:
            print(f"Failed: {e}")

if __name__ == "__main__":
    main()
```

## LogicSig Resource Planning

The signer plans LogicSig program bytes, argument bytes, and opcode cost as
independent consensus resources. Under v42, excess program bytes are paid by
the group fee; dummies are added only when argument or opcode capacity
requires them.

### How It Works (Server-Side)

1. Server selects the authorization-path resource profile.
2. Server solves argument and opcode capacity, adding canonical dummies only
   when required.
3. Server adds the v42 program-byte fee contribution.
4. Server computes the final group ID and signs all transactions.
5. SDK returns the complete signed group ready for submission.

### Example: Falcon-1024 Key

```python
params = algod_client.suggested_params()
txn = transaction.PaymentTxn(sender=falcon_addr, sp=params, receiver=receiver, amt=1000000)

# Server automatically adds dummies - just sign and submit
signed = client.sign_transaction(txn)
txid = send_raw_transaction(algod_client, signed)
```

### Fee Impact

| Key Type | LogicSig Size | Dummies Needed | Extra Fee |
|----------|---------------|----------------|-----------|
| Ed25519 | 0 | 0 | 0 |
| Falcon-1024 | ~3035 | 3 | ~3000 uA |

The extra fee covers the dummy transactions required for post-quantum security.

## License

MIT

## Project

This SDK is part of the APlane project:

- Repository: https://github.com/aplane-algo/aplanesdk
- SDK path: `python`

APlane is an open-source project stewarded by the APlane Project.

See the repository [README](https://github.com/aplane-algo/aplanesdk/blob/main/README.md) for project overview and alpha-status guidance, and [DISCLAIMER.md](https://github.com/aplane-algo/aplanesdk/blob/main/DISCLAIMER.md) for risk, liability, and usage information.
