diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cd2d713..b9e029c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -108,7 +108,7 @@ jobs: - name: Run unit tests with coverage run: | pytest tests/test_types.py tests/test_daemon_client.py \ - tests/test_spec.py tests/test_common_api.py \ + tests/test_spec.py tests/test_common_api.py tests/test_itn.py \ -v --cov=mina_sdk --cov-report=term-missing --cov-report=xml - name: Upload coverage diff --git a/CHANGELOG.md b/CHANGELOG.md index 63301e4..7929ab1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 epoch data, dates, ledger hashes, coinbase, fee transfers and user commands; kind, source, receiver, amount, fee and memo of a sent command. - `AccountNotFoundError`, a `ValueError`, from `get_account`. +- `execute_query`, for custom GraphQL documents, as in the other SDKs. +- ITN client, `mina_sdk.itn` (extra `itn`, which adds `cryptography`): + `ItnClient` for the daemon's ITN GraphQL server (`--itn-graphql-port`), + with ed25519 request signing (`ItnKey`), the `auth` handshake, sequence + numbers (one request at a time, a new auth after HTTP 412) and no repeat + after a transport error. It covers every operation of + `spec/itn-operations.graphql`, as the Rust, Go and JS SDKs do. +- `spec/` is mina-sdk-spec v0.1.1. - Integration tests of the new methods. - `DaemonConnectionError` exception (replaces shadowed `ConnectionError`) - `Currency.__rmul__` for `3 * Currency(10)` support diff --git a/README.md b/README.md index 4740c15..c3effba 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,7 @@ The Mina SDKs have the same API, defined in [mina-sdk-spec](https://github.com/o1-labs/mina-sdk-spec). `spec/` is a copy of it at the tag in `spec/VERSION`. `tests/test_spec.py` checks that this SDK's queries are the specification's documents, and CI checks that `spec/` -is the tag's copy. The ITN client of the specification is not in this SDK yet. +is the tag's copy. ### Queries @@ -89,6 +89,7 @@ is the tag's copy. The ITN client of the specification is not in this SDK yet. | `get_tracked_accounts()` | `list[TrackedAccount]` | Accounts in the daemon's keystore | | `get_snark_pool()` | `list[CompletedWork]` | Completed snark work | | `get_fork_config()` | `Any` | The daemon's fork configuration (JSON) | +| `execute_query(query, variables=None, query_name="custom")` | `dict` | Run a custom GraphQL query | ### Mutations @@ -101,6 +102,53 @@ is the tag's copy. The ITN client of the specification is not in this SDK yet. | `set_snark_worker(public_key)` | `str \| None` | Set/unset SNARK worker | | `set_snark_work_fee(fee)` | `str` | Set SNARK work fee | +### ITN server (`mina_sdk.itn`) + +A daemon started with `ITN_FEATURES=1`, `--itn-graphql-port` and `--itn-keys` +serves a second GraphQL API, which load testing tools use. `ItnClient` signs +each request with an ed25519 `ItnKey` whose public half must be in +`--itn-keys`, and it handles the daemon's sequence numbers (a new `auth` +after a daemon restart, HTTP 412). It needs the `itn` extra: + +```bash +pip install 'mina-sdk[itn]' +``` + +```python +from mina_sdk.itn import ItnClient, ItnKey + +key = ItnKey.from_base64(open("itn_sk").read()) +print("start the daemon with --itn-keys", key.public_key_base64()) + +with ItnClient("http://127.0.0.1:3086/graphql", key) as itn: + logs = itn.internal_logs(0) + if logs: + itn.flush_internal_logs(logs[-1].id) +``` + +| Method | GraphQL | +|--------|---------| +| `auth()` | `auth` (server UUID, sequence number, peer ID, block producer) | +| `slots_won()` | `slotsWon` | +| `internal_logs(start)` / `flush_internal_logs(end)` | `internalLogs` / `flushInternalLogs` | +| `schedule_payments(PaymentsDetails)` | `schedulePayments` | +| `schedule_zkapp_commands(ZkappCommandsDetails)` | `scheduleZkappCommands` | +| `stop_scheduled_transactions(handle)` | `stopScheduledTransactions` | +| `update_gating(GatingUpdate)` | `updateGating` | +| `stop_daemon(delay_seconds=None, clean_config=None)` | `stopDaemon` | +| `set_zkapp_command_limit(limit)` | `zkAppCommandLimit` | +| `execute_query(query, variables, query_name)` | any document, sequenced and signed | + +Requests of one client are sent one at a time, because the daemon accepts +only its exact next sequence number; the client is safe to share between +threads. A sequenced request is never repeated after a transport error, +because the daemon may already have run it. Errors: `ItnUnauthorizedError` +(401), `ItnSequencingError` (412 after a new auth), `InvalidItnKeyError`, and +`DaemonConnectionError` whose `__cause__` is an `ItnHttpError` with the +status. The documents are those of `spec/itn-operations.graphql`. Integration +tests: set `MINA_ITN_URI` and `MINA_ITN_KEY` and run +`pytest tests/test_itn_integration.py`. + ### Currency ```python diff --git a/pyproject.toml b/pyproject.toml index b1f5343..094e009 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,10 @@ dependencies = [ [project.optional-dependencies] archive = ["asyncpg>=0.29"] +# The ITN client (mina_sdk.itn) signs requests with ed25519. +itn = ["cryptography>=42"] dev = [ + "cryptography>=42", "pytest>=8.0", "pytest-asyncio>=0.24", "pytest-cov>=6.0", @@ -86,6 +89,7 @@ ignore = [ "examples/*" = ["T201"] # The documents are copied from spec/operations.graphql and are not wrapped. "src/mina_sdk/daemon/queries.py" = ["E501"] +"src/mina_sdk/itn/queries.py" = ["E501"] [tool.ruff.format] quote-style = "double" diff --git a/spec/ITN.md b/spec/ITN.md index 98d82a1..99df212 100644 --- a/spec/ITN.md +++ b/spec/ITN.md @@ -3,11 +3,12 @@ A daemon started with `ITN_FEATURES=1`, `--itn-graphql-port ` and `--itn-keys ` serves a second GraphQL API, the ITN server (schema `Mina_graphql.schema_itn`, see -[`schema/itn.graphql`](schema/itn.graphql)). Load testing tools use it to +[`schema/itn.graphql`](https://github.com/o1-labs/mina-sdk-spec/blob/main/schema/itn.graphql)). Load testing tools use it to schedule payments and zkApp commands, read internal logs, change connection gating and stop the daemon. -The Mina SDKs have the same ITN client. This document and +The Rust, Go and JS SDKs have the same ITN client; `mina-sdk-python` has no +ITN client yet (– in the Python column). This document and [`itn-operations.graphql`](itn-operations.graphql) define it. The rules of [`SPEC.md`](SPEC.md) apply: each method sends one named operation, and nullable variables are always sent, as null when the caller omits them. @@ -43,21 +44,21 @@ A client must: ## Methods -| Operation | Rust | Go | JS | Arguments | Returns | -|:--|:--|:--|:--|:--|:--| -| `Auth` | `auth` | `Auth` | `auth` | – | server UUID, sequence number, libp2p port, peer ID, block producer flag | -| `SlotsWon` | `slots_won` | `SlotsWon` | `slotsWon` | – | list of global slots | -| `InternalLogs` | `internal_logs` | `InternalLogs` | `internalLogs` | start log ID | list of logs (`id`, `timestamp`, `message`, `metadata` (`item`, `value`), `process?`) | -| `FlushInternalLogs` | `flush_internal_logs` | `FlushInternalLogs` | `flushInternalLogs` | end log ID | string | -| `SchedulePayments` | `schedule_payments` | `SchedulePayments` | `schedulePayments` | `PaymentsDetails` | handle | -| `ScheduleZkappCommands` | `schedule_zkapp_commands` | `ScheduleZkappCommands` | `scheduleZkappCommands` | `ZkappCommandsDetails` | handle | -| `StopScheduledTransactions` | `stop_scheduled_transactions` | `StopScheduledTransactions` | `stopScheduledTransactions` | handle | string | -| `UpdateGating` | `update_gating` | `UpdateGating` | `updateGating` | `GatingUpdate` | string | -| `StopDaemon` | `stop_daemon` | `StopDaemon` | `stopDaemon` | [delay seconds], [clean config] | string | -| `ZkappCommandLimit` | `set_zkapp_command_limit` | `SetZkappCommandLimit` | `setZkappCommandLimit` | limit or null | limit now in force | +| Operation | Rust | Go | JS | Python | Arguments | Returns | +|:--|:--|:--|:--|:--|:--|:--| +| `Auth` | `auth` | `Auth` | `auth` | – | – | server UUID, sequence number, libp2p port, peer ID, block producer flag | +| `SlotsWon` | `slots_won` | `SlotsWon` | `slotsWon` | – | – | list of global slots | +| `InternalLogs` | `internal_logs` | `InternalLogs` | `internalLogs` | – | start log ID | list of logs (`id`, `timestamp`, `message`, `metadata` (`item`, `value`), `process?`) | +| `FlushInternalLogs` | `flush_internal_logs` | `FlushInternalLogs` | `flushInternalLogs` | – | end log ID | string | +| `SchedulePayments` | `schedule_payments` | `SchedulePayments` | `schedulePayments` | – | `PaymentsDetails` | handle | +| `ScheduleZkappCommands` | `schedule_zkapp_commands` | `ScheduleZkappCommands` | `scheduleZkappCommands` | – | `ZkappCommandsDetails` | handle | +| `StopScheduledTransactions` | `stop_scheduled_transactions` | `StopScheduledTransactions` | `stopScheduledTransactions` | – | handle | string | +| `UpdateGating` | `update_gating` | `UpdateGating` | `updateGating` | – | `GatingUpdate` | string | +| `StopDaemon` | `stop_daemon` | `StopDaemon` | `stopDaemon` | – | [delay seconds], [clean config] | string | +| `ZkappCommandLimit` | `set_zkapp_command_limit` | `SetZkappCommandLimit` | `setZkappCommandLimit` | – | limit or null | limit now in force | The input types `PaymentsDetails`, `ZkappCommandsDetails` and `GatingUpdate` -have the fields of [`schema/itn.graphql`](schema/itn.graphql). Every SDK also +have the fields of [`schema/itn.graphql`](https://github.com/o1-labs/mina-sdk-spec/blob/main/schema/itn.graphql). Every SDK also has a custom-query method that signs and sequences any document. ## Notes diff --git a/spec/SPEC.md b/spec/SPEC.md index 5d62576..f8e464d 100644 --- a/spec/SPEC.md +++ b/spec/SPEC.md @@ -1,6 +1,7 @@ # Mina SDK common API -The Mina SDKs (`mina-sdk-rust`, `mina-sdk-go`, `mina-sdk-js`) have the same +The Mina SDKs (`mina-sdk-rust`, `mina-sdk-go`, `mina-sdk-js`, +`mina-sdk-python`) have the same client API for the daemon's public GraphQL server (default port 3085). This document and [`operations.graphql`](operations.graphql) define it. [`ITN.md`](ITN.md) defines the client of the daemon's ITN server. @@ -15,33 +16,34 @@ repository first, make a release, then update every SDK with Names follow each language's style. Arguments in brackets are optional. -| Operation | Rust | Go | JS | Arguments | Returns | -|:--|:--|:--|:--|:--|:--| -| `SyncStatus` | `get_sync_status` | `GetSyncStatus` | `getSyncStatus` | – | `SyncStatus` | -| `DaemonStatus` | `get_daemon_status` | `GetDaemonStatus` | `getDaemonStatus` | – | `DaemonStatus` | -| `DaemonMetrics` | `get_daemon_metrics` | `GetDaemonMetrics` | `getDaemonMetrics` | – | `DaemonMetrics` | -| `NetworkId` | `get_network_id` | `GetNetworkID` | `getNetworkId` | – | string | -| `Account` | `get_account` | `GetAccount` | `getAccount` | public key, [token ID] | `AccountData` | -| `BestChain` | `get_best_chain` | `GetBestChain` | `getBestChain` | [max length] | list of `BlockInfo` | -| `GenesisBlock` | `get_genesis_block` | `GetGenesisBlock` | `getGenesisBlock` | – | `BlockInfo` | -| `Block` | `get_block` | `GetBlock` | `getBlock` | state hash **or** height | `BlockInfo` | -| `Peers` | `get_peers` | `GetPeers` | `getPeers` | – | list of `PeerInfo` | -| `PooledUserCommands` | `get_pooled_user_commands` | `GetPooledUserCommands` | `getPooledUserCommands` | [public key] | list of `PooledUserCommand` | -| `PooledZkappCommands` | `get_pooled_zkapp_commands` | `GetPooledZkappCommands` | `getPooledZkappCommands` | [public key] | list of `ZkappCommandResult` | -| `TransactionStatus` | `get_transaction_status` | `GetTransactionStatus` | `getTransactionStatus` | payment ID **or** zkApp transaction ID | `TransactionStatus` | -| `GenesisConstants` | `get_genesis_constants` | `GetGenesisConstants` | `getGenesisConstants` | – | `GenesisConstants` | -| `TrackedAccounts` | `get_tracked_accounts` | `GetTrackedAccounts` | `getTrackedAccounts` | – | list of `TrackedAccount` | -| `SnarkPool` | `get_snark_pool` | `GetSnarkPool` | `getSnarkPool` | – | list of `CompletedWork` | -| `ForkConfig` | `get_fork_config` | `GetForkConfig` | `getForkConfig` | – | JSON value | -| `SendPayment` | `send_payment` | `SendPayment` | `sendPayment` | payment, [signature] | `SubmittedCommand` | -| `SendDelegation` | `send_delegation` | `SendDelegation` | `sendDelegation` | delegation, [signature] | `SubmittedCommand` | -| `SendZkapp` | `send_zkapp` | `SendZkapp` | `sendZkapp` | zkApp command (JSON) | `ZkappCommandResult` | -| `UnlockAccount` | `unlock_account` | `UnlockAccount` | `unlockAccount` | public key, password | public key | -| `SetSnarkWorker` | `set_snark_worker` | `SetSnarkWorker` | `setSnarkWorker` | [public key] | previous worker or null | -| `SetSnarkWorkFee` | `set_snark_work_fee` | `SetSnarkWorkFee` | `setSnarkWorkFee` | fee | previous fee | - -Every SDK also has a custom-query method (`execute_query`, `Request`, -`executeQuery`) that uses the same transport, retries and errors. +| Operation | Rust | Go | JS | Python | Arguments | Returns | +|:--|:--|:--|:--|:--|:--|:--| +| `SyncStatus` | `get_sync_status` | `GetSyncStatus` | `getSyncStatus` | `get_sync_status` | – | `SyncStatus` | +| `DaemonStatus` | `get_daemon_status` | `GetDaemonStatus` | `getDaemonStatus` | `get_daemon_status` | – | `DaemonStatus` | +| `DaemonMetrics` | `get_daemon_metrics` | `GetDaemonMetrics` | `getDaemonMetrics` | `get_daemon_metrics` | – | `DaemonMetrics` | +| `NetworkId` | `get_network_id` | `GetNetworkID` | `getNetworkId` | `get_network_id` | – | string | +| `Account` | `get_account` | `GetAccount` | `getAccount` | `get_account` | public key, [token ID] | `AccountData` | +| `BestChain` | `get_best_chain` | `GetBestChain` | `getBestChain` | `get_best_chain` | [max length] | list of `BlockInfo` | +| `GenesisBlock` | `get_genesis_block` | `GetGenesisBlock` | `getGenesisBlock` | `get_genesis_block` | – | `BlockInfo` | +| `Block` | `get_block` | `GetBlock` | `getBlock` | `get_block` | state hash **or** height | `BlockInfo` | +| `Peers` | `get_peers` | `GetPeers` | `getPeers` | `get_peers` | – | list of `PeerInfo` | +| `PooledUserCommands` | `get_pooled_user_commands` | `GetPooledUserCommands` | `getPooledUserCommands` | `get_pooled_user_commands` | [public key] | list of `PooledUserCommand` | +| `PooledZkappCommands` | `get_pooled_zkapp_commands` | `GetPooledZkappCommands` | `getPooledZkappCommands` | `get_pooled_zkapp_commands` | [public key] | list of `ZkappCommandResult` | +| `TransactionStatus` | `get_transaction_status` | `GetTransactionStatus` | `getTransactionStatus` | `get_transaction_status` | payment ID **or** zkApp transaction ID | `TransactionStatus` | +| `GenesisConstants` | `get_genesis_constants` | `GetGenesisConstants` | `getGenesisConstants` | `get_genesis_constants` | – | `GenesisConstants` | +| `TrackedAccounts` | `get_tracked_accounts` | `GetTrackedAccounts` | `getTrackedAccounts` | `get_tracked_accounts` | – | list of `TrackedAccount` | +| `SnarkPool` | `get_snark_pool` | `GetSnarkPool` | `getSnarkPool` | `get_snark_pool` | – | list of `CompletedWork` | +| `ForkConfig` | `get_fork_config` | `GetForkConfig` | `getForkConfig` | `get_fork_config` | – | JSON value | +| `SendPayment` | `send_payment` | `SendPayment` | `sendPayment` | `send_payment` | payment, [signature] | `SubmittedCommand` | +| `SendDelegation` | `send_delegation` | `SendDelegation` | `sendDelegation` | `send_delegation` | delegation, [signature] | `SubmittedCommand` | +| `SendZkapp` | `send_zkapp` | `SendZkapp` | `sendZkapp` | `send_zkapp` | zkApp command (JSON) | `ZkappCommandResult` | +| `UnlockAccount` | `unlock_account` | `UnlockAccount` | `unlockAccount` | `unlock_account` | public key, password | public key | +| `SetSnarkWorker` | `set_snark_worker` | `SetSnarkWorker` | `setSnarkWorker` | `set_snark_worker` | [public key] | previous worker or null | +| `SetSnarkWorkFee` | `set_snark_work_fee` | `SetSnarkWorkFee` | `setSnarkWorkFee` | `set_snark_work_fee` | fee | previous fee | + +Every SDK also has a custom-query method (Rust and Python `execute_query`, +Go `Request`, JS `executeQuery`) that uses the same transport, retries and +errors. ## Rules @@ -65,14 +67,18 @@ Every SDK also has a custom-query method (`execute_query`, `Request`, ## Result types -Field names are shown in camelCase; Rust uses snake_case and Go PascalCase. +Field names are shown in camelCase; Rust and Python use snake_case and Go +PascalCase. Python names the daemon's `from` field `from_`, because `from` +is a keyword. `?` marks a field that can be null or absent. The types are the union of what the SDKs returned before this API, so no caller loses a field. Where an SDK already had a field or a type under another name, it keeps that name (for example Go `CreatorPK` and Rust `creator_pk` for `creatorPublicKey`, and the JS type `Block` for the result -of `getBlock`), and adds the fields of this document to it. +of `getBlock`), and adds the fields of this document to it. Python's +`PooledUserCommand` also answers the dictionary keys of its earlier +dictionary result. **`DaemonStatus`**: `syncStatus`, `blockchainLength?`, `highestBlockLengthReceived?`, `highestUnvalidatedBlockLengthReceived?`, diff --git a/spec/VERSION b/spec/VERSION index b82608c..8308b63 100644 --- a/spec/VERSION +++ b/spec/VERSION @@ -1 +1 @@ -v0.1.0 +v0.1.1 diff --git a/src/mina_sdk/daemon/client.py b/src/mina_sdk/daemon/client.py index d46f3da..f250aff 100644 --- a/src/mina_sdk/daemon/client.py +++ b/src/mina_sdk/daemon/client.py @@ -190,6 +190,18 @@ def _request( f"Failed to execute {query_name} after {self._retries} attempts: {last_error}" ) + def execute_query( + self, query: str, variables: dict[str, Any] | None = None, query_name: str = "custom" + ) -> dict[str, Any]: + """Run a custom GraphQL document, with the client's transport, retries + and errors. Returns the ``data`` field of the response. + + Raises: + GraphQLError: If the response contains GraphQL-level errors (not retried). + DaemonConnectionError: If all retry attempts fail due to network errors. + """ + return self._request(query, variables=variables, query_name=query_name) + # -- Queries -- def get_sync_status(self) -> str: diff --git a/src/mina_sdk/itn/__init__.py b/src/mina_sdk/itn/__init__.py new file mode 100644 index 0000000..f78c439 --- /dev/null +++ b/src/mina_sdk/itn/__init__.py @@ -0,0 +1,47 @@ +"""Client for the daemon's ITN GraphQL server (``ITN_FEATURES=1``, +``--itn-graphql-port``, ``--itn-keys``), whose requests are signed with an +ed25519 key. It needs the ``itn`` extra: ``pip install mina-sdk[itn]``. +""" + +try: + import cryptography # noqa: F401 +except ImportError as e: # pragma: no cover + raise ImportError( + "mina_sdk.itn needs the 'cryptography' package: pip install 'mina-sdk[itn]'" + ) from e + +from mina_sdk.itn import queries +from mina_sdk.itn.client import ItnClient +from mina_sdk.itn.errors import ( + InvalidItnKeyError, + ItnHttpError, + ItnSequencingError, + ItnUnauthorizedError, +) +from mina_sdk.itn.key import ItnKey +from mina_sdk.itn.types import ( + GatingUpdate, + ItnAuth, + ItnLog, + LogMetadatum, + NetworkPeer, + PaymentsDetails, + ZkappCommandsDetails, +) + +__all__ = [ + "GatingUpdate", + "InvalidItnKeyError", + "ItnAuth", + "ItnClient", + "ItnHttpError", + "ItnKey", + "ItnLog", + "ItnSequencingError", + "ItnUnauthorizedError", + "LogMetadatum", + "NetworkPeer", + "PaymentsDetails", + "ZkappCommandsDetails", + "queries", +] diff --git a/src/mina_sdk/itn/client.py b/src/mina_sdk/itn/client.py new file mode 100644 index 0000000..1005365 --- /dev/null +++ b/src/mina_sdk/itn/client.py @@ -0,0 +1,334 @@ +"""Client for the daemon's ITN GraphQL server. + +A daemon started with ``ITN_FEATURES=1``, ``--itn-graphql-port `` and +``--itn-keys `` serves a second GraphQL API +(``Mina_graphql.schema_itn``). Load testing tools use it to schedule payments +and zkApp commands, read internal logs, change connection gating and stop the +daemon. Every request must be signed with an ed25519 key whose public half is +in ``--itn-keys``. + +Authentication protocol (see ``spec/ITN.md``): the daemon reads the +``Authorization`` header. + +- ``Signature ``: the signature covers the request body. The daemon + accepts this form for the ``Auth`` operation only. +- ``Signature ; Sequencing ``: the signature covers the + big-endian u16 ``n``, then the server UUID, then the body. ``n`` must be + exactly the daemon's sequence number for this public key, which starts at + 0, goes up by one (wrapping) after each accepted sequenced request, and is + shared by all clients that sign with the same key. + +A bad signature or an unknown key gives HTTP 401; a wrong UUID (the daemon +restarted) or sequence number gives HTTP 412. ``ItnClient`` runs auth to learn +the UUID and the sequence number, sends sequenced requests one at a time, and +on a 412 runs auth again and repeats the request once. It never repeats a +sequenced request after a transport error, because the daemon may have run it. +""" + +from __future__ import annotations + +import json +import logging +import struct +import threading +import time +from typing import Any + +import httpx + +from mina_sdk.daemon.client import DaemonConnectionError, GraphQLError +from mina_sdk.itn import queries +from mina_sdk.itn.errors import ItnHttpError, ItnSequencingError, ItnUnauthorizedError +from mina_sdk.itn.key import ItnKey +from mina_sdk.itn.types import ( + GatingUpdate, + ItnAuth, + ItnLog, + LogMetadatum, + PaymentsDetails, + ZkappCommandsDetails, +) + +logger = logging.getLogger(__name__) + + +class ItnClient: + """Client for one daemon's ITN GraphQL server. + + It is safe to use from several threads; share one client per daemon, + because it holds the daemon's session. + + Args: + graphql_uri: The ITN endpoint, for example ``http://127.0.0.1:3086/graphql``. + key: The key whose public half is in the daemon's ``--itn-keys``. + retries: Number of attempts of the auth handshake (must be >= 1). + Sequenced requests are never repeated after a transport error. + retry_delay: Seconds between auth attempts. + timeout: HTTP request timeout in seconds. + + Example:: + + key = ItnKey.from_base64(open("itn_sk").read()) + with ItnClient("http://127.0.0.1:3086/graphql", key) as itn: + logs = itn.internal_logs(0) + """ + + def __init__( + self, + graphql_uri: str, + key: ItnKey, + retries: int = 3, + retry_delay: float = 5.0, + timeout: float = 30.0, + ): + if not graphql_uri: + raise ValueError("graphql_uri must not be empty") + if retries < 1: + raise ValueError(f"retries must be >= 1, got {retries}") + if retry_delay < 0: + raise ValueError(f"retry_delay must be >= 0, got {retry_delay}") + if timeout <= 0: + raise ValueError(f"timeout must be > 0, got {timeout}") + self._uri = graphql_uri + self._key = key + self._retries = retries + self._retry_delay = retry_delay + self._client = httpx.Client(timeout=timeout) + # Serializes requests: the daemon accepts only its exact next number. + self._lock = threading.Lock() + # Server UUID and the next sequence number of the key; None until the + # first handshake and after an error that leaves it unknown. + self._session: tuple[str, int] | None = None + self._last_auth: ItnAuth | None = None + + @property + def graphql_uri(self) -> str: + return self._uri + + def public_key_base64(self) -> str: + """The client key's public half, as it must appear in ``--itn-keys``.""" + return self._key.public_key_base64() + + def close(self) -> None: + """Release resources held by the HTTP client.""" + self._client.close() + + def __enter__(self) -> ItnClient: + return self + + def __exit__(self, *args: Any) -> None: + self.close() + + @property + def last_auth(self) -> ItnAuth | None: + """The answer of the latest auth handshake, or ``None`` before the first + one. The client runs auth again by itself after a daemon restart (HTTP + 412), so a caller that keeps the node's peer ID or libp2p port can + refresh them from here after a request.""" + return self._last_auth + + # -- Transport -- + + def _post(self, body: bytes, authorization: str) -> httpx.Response: + return self._client.post( + self._uri, + content=body, + headers={"Content-Type": "application/json", "Authorization": authorization}, + ) + + @staticmethod + def _decode(resp: httpx.Response, query_name: str) -> dict[str, Any]: + try: + parsed = resp.json() + except ValueError as e: + raise DaemonConnectionError(f"Invalid JSON response from {query_name}: {e}") from e + if parsed.get("errors"): + raise GraphQLError(parsed["errors"], query_name) + return parsed.get("data") or {} + + def _handshake(self) -> ItnAuth: + """Run auth with an unsequenced signature and store the session. The + caller holds the lock.""" + name = "itn_auth" + body = json.dumps({"query": queries.AUTH}).encode() + authorization = f"Signature {self._key.public_key_base64()} {self._key.sign_base64(body)}" + last_error: Exception | None = None + for attempt in range(1, self._retries + 1): + try: + resp = self._post(body, authorization) + except httpx.HTTPError as e: + last_error = e + logger.warning("ITN auth error (attempt %d/%d): %s", attempt, self._retries, e) + else: + if resp.status_code == 401: + raise ItnUnauthorizedError(name) + if resp.status_code >= 300: + last_error = ItnHttpError(resp.status_code, resp.text) + else: + auth = _parse_auth(self._decode(resp, name)) + self._session = (auth.server_uuid, auth.signer_sequence_number) + self._last_auth = auth + return auth + if attempt < self._retries: + time.sleep(self._retry_delay) + raise DaemonConnectionError( + f"Failed to execute {name} after {self._retries} attempts: {last_error}" + ) from last_error + + def execute_query( + self, query: str, variables: dict[str, Any] | None = None, query_name: str = "itn_custom" + ) -> dict[str, Any]: + """Send one sequenced, signed request and return its ``data`` field. + + Raises: + GraphQLError: The response has GraphQL errors. + ItnUnauthorizedError: HTTP 401. + ItnSequencingError: HTTP 412 again after a new auth. + DaemonConnectionError: A transport error or another HTTP error status + (its ``__cause__`` is an ``ItnHttpError``). It is not repeated. + """ + payload: dict[str, Any] = {"query": query} + if variables is not None: + payload["variables"] = variables + body = json.dumps(payload).encode() + + with self._lock: + # A second pass only follows a 412: the daemon restarted, or another + # client with the same key used our sequence number. + for _ in range(2): + if self._session is None: + auth = self._handshake() + uuid, seq = auth.server_uuid, auth.signer_sequence_number + else: + uuid, seq = self._session + message = struct.pack(">H", seq) + uuid.encode() + body + authorization = ( + f"Signature {self._key.public_key_base64()} " + f"{self._key.sign_base64(message)} ; Sequencing {uuid} {seq}" + ) + try: + resp = self._post(body, authorization) + except httpx.HTTPError as e: + # Unknown whether the daemon counted it: start over next time. + self._session = None + raise DaemonConnectionError(f"Failed to execute {query_name}: {e}") from e + if resp.status_code == 412: + self._session = None + continue + if resp.status_code == 401: + raise ItnUnauthorizedError(query_name) + if resp.status_code >= 300: + self._session = None + error = ItnHttpError(resp.status_code, resp.text) + raise DaemonConnectionError( + f"Failed to execute {query_name}: {error}" + ) from error + # The signature was accepted, so the daemon has moved to the next + # number, whatever the GraphQL result is. + self._session = (uuid, (seq + 1) % 65536) + return self._decode(resp, query_name) + raise ItnSequencingError(query_name) + + # -- Operations -- + + def auth(self) -> ItnAuth: + """Run the auth handshake and return the node's answer. Other methods + run it when needed, so calling it first is optional.""" + with self._lock: + return self._handshake() + + def slots_won(self) -> list[int]: + """The global slots the node's block producer keys won in the current epoch.""" + data = self.execute_query(queries.SLOTS_WON, None, "itn_slots_won") + return [int(s) for s in data["slotsWon"]] + + def internal_logs(self, start_log_id: int) -> list[ItnLog]: + """The internal logs with an ID of at least ``start_log_id``.""" + data = self.execute_query( + queries.INTERNAL_LOGS, {"startLogId": start_log_id}, "itn_internal_logs" + ) + return [ + ItnLog( + id=int(log["id"]), + timestamp=log["timestamp"], + message=log["message"], + metadata=[LogMetadatum(item=m["item"], value=m["value"]) for m in log["metadata"]], + process=log.get("process"), + ) + for log in data["internalLogs"] + ] + + def flush_internal_logs(self, end_log_id: int) -> str: + """Drop the internal logs up to and including ``end_log_id``.""" + data = self.execute_query( + queries.FLUSH_INTERNAL_LOGS, {"endLogId": end_log_id}, "itn_flush_internal_logs" + ) + return data["flushInternalLogs"] + + def schedule_payments(self, details: PaymentsDetails) -> str: + """Start sending payments; returns the handle for ``stop_scheduled_transactions``.""" + data = self.execute_query( + queries.SCHEDULE_PAYMENTS, {"input": details.to_variables()}, "itn_schedule_payments" + ) + return data["schedulePayments"] + + def schedule_zkapp_commands(self, details: ZkappCommandsDetails) -> str: + """Start sending zkApp commands; returns the handle for + ``stop_scheduled_transactions``.""" + data = self.execute_query( + queries.SCHEDULE_ZKAPP_COMMANDS, + {"input": details.to_variables()}, + "itn_schedule_zkapp_commands", + ) + return data["scheduleZkappCommands"] + + def stop_scheduled_transactions(self, handle: str) -> str: + """Stop the transactions of a schedule handle.""" + data = self.execute_query( + queries.STOP_SCHEDULED_TRANSACTIONS, + {"handle": handle}, + "itn_stop_scheduled_transactions", + ) + return data["stopScheduledTransactions"] + + def update_gating(self, update: GatingUpdate) -> str: + """Change the node's connection gating.""" + data = self.execute_query( + queries.UPDATE_GATING, {"input": update.to_variables()}, "itn_update_gating" + ) + return data["updateGating"] + + def stop_daemon( + self, delay_seconds: int | None = None, clean_config: bool | None = None + ) -> str: + """Stop the daemon after ``delay_seconds`` (the daemon's minimum is 5; + ``None`` for its default), deleting its configuration directory if + ``clean_config`` is true.""" + data = self.execute_query( + queries.STOP_DAEMON, + {"delaySeconds": delay_seconds, "cleanConfig": clean_config}, + "itn_stop_daemon", + ) + return data["stopDaemon"] + + def set_zkapp_command_limit(self, limit: int | None) -> int | None: + """Set the block producer's limit of zkApp commands per block; ``None`` + removes the limit. Returns the limit now in force.""" + data = self.execute_query( + queries.ZKAPP_COMMAND_LIMIT, {"limit": limit}, "itn_set_zkapp_command_limit" + ) + return data["zkAppCommandLimit"] + + +def _parse_auth(data: dict[str, Any]) -> ItnAuth: + auth = data.get("auth") + if not auth or not auth.get("serverUuid"): + raise ValueError("itn_auth: missing field auth.serverUuid") + # UInt16 values come as strings ("8302") or numbers. + return ItnAuth( + server_uuid=auth["serverUuid"], + signer_sequence_number=int(auth["signerSequenceNumber"]), + libp2p_port=int(auth["libp2pPort"]), + peer_id=auth.get("peerId"), + is_block_producer=bool(auth.get("isBlockProducer")), + ) diff --git a/src/mina_sdk/itn/errors.py b/src/mina_sdk/itn/errors.py new file mode 100644 index 0000000..ac0b4fe --- /dev/null +++ b/src/mina_sdk/itn/errors.py @@ -0,0 +1,45 @@ +"""Errors of the ITN client.""" + +from __future__ import annotations + + +class ItnUnauthorizedError(Exception): + """The ITN server rejected the request signature (HTTP 401): the signature + is wrong, or the key's public half is not in the daemon's ``--itn-keys``.""" + + def __init__(self, query_name: str): + self.query_name = query_name + super().__init__( + f"ITN server rejected the signature of {query_name} (HTTP 401); " + "is the public key in --itn-keys?" + ) + + +class ItnSequencingError(Exception): + """The ITN server still rejected the sequence information (HTTP 412) after + a new auth handshake.""" + + def __init__(self, query_name: str): + self.query_name = query_name + super().__init__( + f"ITN server rejected the sequence number of {query_name} again " + "after a new auth (HTTP 412)" + ) + + +class InvalidItnKeyError(ValueError): + """An ITN key cannot be decoded.""" + + def __init__(self, reason: str): + super().__init__(f"invalid ITN key: {reason}") + + +class ItnHttpError(Exception): + """The ``__cause__`` of a ``DaemonConnectionError`` when the daemon answered + with an HTTP error status other than 401 and 412. Read ``status_code`` to + tell a node that is unwell (5xx) from a request that no node would accept.""" + + def __init__(self, status_code: int, body: str): + self.status_code = status_code + self.body = body + super().__init__(f"HTTP {status_code}: {body[:200]}") diff --git a/src/mina_sdk/itn/key.py b/src/mina_sdk/itn/key.py new file mode 100644 index 0000000..3a513a0 --- /dev/null +++ b/src/mina_sdk/itn/key.py @@ -0,0 +1,69 @@ +"""The ed25519 key that signs requests to a daemon's ITN server.""" + +from __future__ import annotations + +import base64 +import binascii + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.serialization import ( + Encoding, + NoEncryption, + PrivateFormat, + PublicFormat, +) + +from mina_sdk.itn.errors import InvalidItnKeyError + +SEED_SIZE = 32 + + +class ItnKey: + """An ed25519 key that signs requests to a daemon's ITN GraphQL server. + + The daemon accepts a request only when the key's public half is in its + ``--itn-keys`` list. Keys are exchanged as standard base64 with padding: + the private key as its 32-byte seed (the format of mina-perf-testing's + orchestrator key and ``fetcher_sk``), and the public key as the 32-byte + value for ``--itn-keys``. + """ + + def __init__(self, seed: bytes): + if len(seed) != SEED_SIZE: + raise InvalidItnKeyError(f"expected {SEED_SIZE} bytes, got {len(seed)}") + self._seed = bytes(seed) + self._sk = Ed25519PrivateKey.from_private_bytes(self._seed) + + @classmethod + def from_base64(cls, seed_b64: str) -> ItnKey: + """Load a key from the base64 encoding of its 32-byte seed. Surrounding + white space is ignored.""" + try: + seed = base64.b64decode(seed_b64.strip(), validate=True) + except (binascii.Error, ValueError) as e: + raise InvalidItnKeyError(f"not base64: {e}") from e + return cls(seed) + + @classmethod + def generate(cls) -> ItnKey: + """Create a new random key.""" + sk = Ed25519PrivateKey.generate() + return cls(sk.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption())) + + def to_base64(self) -> str: + """The base64 encoding of the key's 32-byte seed; the inverse of + ``from_base64``.""" + return base64.b64encode(self._seed).decode() + + def public_key_base64(self) -> str: + """The base64 public key, as the daemon expects it in ``--itn-keys``.""" + raw = self._sk.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) + return base64.b64encode(raw).decode() + + def sign_base64(self, message: bytes) -> str: + """The base64 ed25519 signature of ``message``.""" + return base64.b64encode(self._sk.sign(message)).decode() + + def __repr__(self) -> str: + # The public key only, so that a key never ends up in a log. + return f"ItnKey(public={self.public_key_base64()})" diff --git a/src/mina_sdk/itn/queries.py b/src/mina_sdk/itn/queries.py new file mode 100644 index 0000000..9ac0c43 --- /dev/null +++ b/src/mina_sdk/itn/queries.py @@ -0,0 +1,76 @@ +"""GraphQL documents for the daemon's ITN server. + +These are the documents of ``spec/itn-operations.graphql`` (a copy of +o1-labs/mina-sdk-spec, whose CI validates them against the daemon's +``schema_itn``); ``tests/test_spec.py`` checks that they stay identical. Use +them with ``ItnClient.execute_query`` for custom selections. +""" + +AUTH = """query Auth { + auth { + serverUuid + signerSequenceNumber + libp2pPort + peerId + isBlockProducer + } +}""" + +SLOTS_WON = """query SlotsWon { + slotsWon +}""" + +INTERNAL_LOGS = """query InternalLogs($startLogId: Int!) { + internalLogs(startLogId: $startLogId) { + id + timestamp + message + metadata { + item + value + } + process + } +}""" + +FLUSH_INTERNAL_LOGS = """mutation FlushInternalLogs($endLogId: Int!) { + flushInternalLogs(endLogId: $endLogId) +}""" + +SCHEDULE_PAYMENTS = """mutation SchedulePayments($input: PaymentsDetails!) { + schedulePayments(input: $input) +}""" + +SCHEDULE_ZKAPP_COMMANDS = """mutation ScheduleZkappCommands($input: ZkappCommandsDetails!) { + scheduleZkappCommands(input: $input) +}""" + +STOP_SCHEDULED_TRANSACTIONS = """mutation StopScheduledTransactions($handle: String!) { + stopScheduledTransactions(handle: $handle) +}""" + +UPDATE_GATING = """mutation UpdateGating($input: GatingUpdate!) { + updateGating(input: $input) +}""" + +STOP_DAEMON = """mutation StopDaemon($delaySeconds: Int, $cleanConfig: Boolean) { + stopDaemon(delaySeconds: $delaySeconds, cleanConfig: $cleanConfig) +}""" + +ZKAPP_COMMAND_LIMIT = """mutation ZkappCommandLimit($limit: Int) { + zkAppCommandLimit(limit: $limit) +}""" + +# Every ITN document of the specification, for the conformance test. +ALL_ITN_DOCUMENTS = [ + AUTH, + SLOTS_WON, + INTERNAL_LOGS, + FLUSH_INTERNAL_LOGS, + SCHEDULE_PAYMENTS, + SCHEDULE_ZKAPP_COMMANDS, + STOP_SCHEDULED_TRANSACTIONS, + UPDATE_GATING, + STOP_DAEMON, + ZKAPP_COMMAND_LIMIT, +] diff --git a/src/mina_sdk/itn/types.py b/src/mina_sdk/itn/types.py new file mode 100644 index 0000000..05cebe3 --- /dev/null +++ b/src/mina_sdk/itn/types.py @@ -0,0 +1,201 @@ +"""Inputs and results of the ITN client.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from mina_sdk.types import Currency + + +@dataclass(frozen=True) +class ItnAuth: + """The result of the auth query. + + Attributes: + server_uuid: UUID of the ITN GraphQL server; new after each daemon restart. + signer_sequence_number: The number the next sequenced request of this + key must carry. + libp2p_port: The node's libp2p port. + peer_id: The node's libp2p peer ID; ``None`` if the daemon sent none. + is_block_producer: Whether the node produces blocks. + """ + + server_uuid: str + signer_sequence_number: int + libp2p_port: int + peer_id: str | None + is_block_producer: bool + + +@dataclass(frozen=True) +class LogMetadatum: + """One (item, value) pair of an internal log; the value is any JSON value.""" + + item: str + value: Any + + +@dataclass(frozen=True) +class ItnLog: + """One entry of the daemon's internal log. + + Attributes: + id: ID of the log; IDs increase. + process: Process that sent the log if it is not the daemon (prover or + verifier); ``None`` for the daemon. + """ + + id: int + timestamp: str + message: str + metadata: list[LogMetadatum] + process: str | None = None + + +@dataclass(frozen=True) +class PaymentsDetails: + """The input of ``schedule_payments``. + + Attributes: + duration_min: Length of the scheduler run, in minutes. + tps: Frequency of transactions, per second. + memo_prefix: The memo, up to 32 characters. + amount: Amount of each payment. + receiver: Public key of the receiver of the payments. + senders: Base58 private keys of the accounts to send from. + """ + + duration_min: int + tps: float + memo_prefix: str + max_fee: Currency + min_fee: Currency + amount: Currency + receiver: str + senders: list[str] = field(default_factory=list) + + def to_variables(self) -> dict[str, Any]: + return { + "durationMin": self.duration_min, + "tps": self.tps, + "memoPrefix": self.memo_prefix, + "maxFee": self.max_fee.to_nanomina_str(), + "minFee": self.min_fee.to_nanomina_str(), + "amount": self.amount.to_nanomina_str(), + "receiver": self.receiver, + "senders": list(self.senders), + } + + +@dataclass(frozen=True) +class ZkappCommandsDetails: + """The input of ``schedule_zkapp_commands``. + + Attributes: + max_account_updates: Each generated zkApp transaction has + 2*max_account_updates+2 account updates (including balancing and + fee payer). ``None`` sends null, and the daemon uses its default. + max_cost: Generate max-cost zkApp commands. + account_queue_size: Size of the queue of recently used accounts. + deployment_fee: Fee of the initial deployment of zkApp accounts. + init_balance: Initial balance of the zkApp accounts deployed for the test. + no_precondition: Disable the precondition in account updates. + duration_min: Length of the scheduler run, in minutes. + tps: Frequency of transactions, per second. + num_new_accounts: Number of zkApp accounts the scheduler creates during + the test. + num_zkapps_to_deploy: Number of zkApp accounts deployed at the start of + the test. + fee_payers: Base58 private keys of the fee payers, which also create + the accounts. + non_default_token: Load a custom (non-default) owned token. Only an + unreleased daemon branch has this field; released daemons ignore + it. It is sent only when set. + """ + + account_queue_size: int + deployment_fee: Currency + max_fee: Currency + min_fee: Currency + init_balance: Currency + max_new_zkapp_balance: Currency + min_new_zkapp_balance: Currency + max_balance_change: Currency + min_balance_change: Currency + memo_prefix: str + duration_min: int + tps: float + num_new_accounts: int + num_zkapps_to_deploy: int + fee_payers: list[str] = field(default_factory=list) + max_account_updates: int | None = None + max_cost: bool = False + no_precondition: bool = False + non_default_token: bool | None = None + + def to_variables(self) -> dict[str, Any]: + v: dict[str, Any] = { + "maxAccountUpdates": self.max_account_updates, + "maxCost": self.max_cost, + "accountQueueSize": self.account_queue_size, + "deploymentFee": self.deployment_fee.to_nanomina_str(), + "maxFee": self.max_fee.to_nanomina_str(), + "minFee": self.min_fee.to_nanomina_str(), + "initBalance": self.init_balance.to_nanomina_str(), + "maxNewZkappBalance": self.max_new_zkapp_balance.to_nanomina_str(), + "minNewZkappBalance": self.min_new_zkapp_balance.to_nanomina_str(), + "maxBalanceChange": self.max_balance_change.to_nanomina_str(), + "minBalanceChange": self.min_balance_change.to_nanomina_str(), + "noPrecondition": self.no_precondition, + "memoPrefix": self.memo_prefix, + "durationMin": self.duration_min, + "tps": self.tps, + "numNewAccounts": self.num_new_accounts, + "numZkappsToDeploy": self.num_zkapps_to_deploy, + "feePayers": list(self.fee_payers), + } + if self.non_default_token is not None: + v["nonDefaultToken"] = self.non_default_token + return v + + +@dataclass(frozen=True) +class NetworkPeer: + """A peer in a ``GatingUpdate``: IP address, libp2p port and base58 peer ID.""" + + host: str + libp2p_port: int + peer_id: str + + def to_variables(self) -> dict[str, Any]: + return {"host": self.host, "libp2pPort": self.libp2p_port, "peerId": self.peer_id} + + +@dataclass(frozen=True) +class GatingUpdate: + """The input of ``update_gating``. + + Attributes: + added_peers: Peers to connect to. + clean_added_peers: Reset the added peers, including the seeds, to an + empty list. + isolate: Allow connections only from trusted peers. + banned_peers: Peers that may never connect, unless they are also trusted. + trusted_peers: Peers that may always connect. + """ + + added_peers: list[NetworkPeer] = field(default_factory=list) + clean_added_peers: bool = False + isolate: bool = False + banned_peers: list[NetworkPeer] = field(default_factory=list) + trusted_peers: list[NetworkPeer] = field(default_factory=list) + + def to_variables(self) -> dict[str, Any]: + return { + "addedPeers": [p.to_variables() for p in self.added_peers], + "cleanAddedPeers": self.clean_added_peers, + "isolate": self.isolate, + "bannedPeers": [p.to_variables() for p in self.banned_peers], + "trustedPeers": [p.to_variables() for p in self.trusted_peers], + } diff --git a/tests/test_common_api.py b/tests/test_common_api.py index d615205..e7ffa39 100644 --- a/tests/test_common_api.py +++ b/tests/test_common_api.py @@ -399,6 +399,14 @@ def test_send_zkapp_and_unlock_account(client): assert _sent(route)["variables"] == {"input": {"publicKey": "B62qacc", "password": "pw"}} +@respx.mock +def test_execute_query(client): + route = _mock({"syncStatus": "SYNCED"}) + data = client.execute_query("query S { syncStatus }", {"x": None}, "sync") + assert data == {"syncStatus": "SYNCED"} + assert _sent(route) == {"query": "query S { syncStatus }", "variables": {"x": None}} + + @respx.mock def test_transaction_status_and_small_queries(client): route = _mock({"transactionStatus": "INCLUDED"}) diff --git a/tests/test_itn.py b/tests/test_itn.py new file mode 100644 index 0000000..cf99766 --- /dev/null +++ b/tests/test_itn.py @@ -0,0 +1,274 @@ +"""Tests of the ITN client against a mock ITN server that checks signatures +and sequence numbers as the daemon does.""" + +from __future__ import annotations + +import base64 +import json +import struct + +import httpx +import pytest +import respx +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + +from mina_sdk import Currency, DaemonConnectionError, GraphQLError +from mina_sdk.itn import ( + GatingUpdate, + InvalidItnKeyError, + ItnClient, + ItnHttpError, + ItnKey, + ItnSequencingError, + ItnUnauthorizedError, + NetworkPeer, + PaymentsDetails, + queries, +) + +URL = "http://localhost:3086/graphql" +SEED = base64.b64encode(bytes(range(32))).decode() + + +def _verify(public_b64: str, sig_b64: str, message: bytes) -> bool: + key = Ed25519PublicKey.from_public_bytes(base64.b64decode(public_b64)) + try: + key.verify(base64.b64decode(sig_b64), message) + except InvalidSignature: + return False + return True + + +class MockItnServer: + """Answers like the daemon: auth with an unsequenced signature, everything + else with the exact next sequence number of the key.""" + + def __init__(self, key: ItnKey, seq: int = 0, data: dict | None = None): + self.public = key.public_key_base64() + self.uuid = "uuid-1" + self.seq = seq + self.data = data or {} + self.requests: list[dict] = [] + self.status_override: list[int] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + body = request.content + payload = json.loads(body) + self.requests.append(payload) + header = request.headers["Authorization"] + sig_part, _, seq_part = header.partition(" ; ") + # status_override answers sequenced requests only. + if seq_part and self.status_override: + return httpx.Response(self.status_override.pop(0), text="override") + _, pk, sig = sig_part.split(" ") + if pk != self.public: + return httpx.Response(401, text="unknown key") + if payload["query"] == queries.AUTH and not seq_part: + if not _verify(pk, sig, body): + return httpx.Response(401, text="bad signature") + return httpx.Response( + 200, + json={ + "data": { + "auth": { + "serverUuid": self.uuid, + "signerSequenceNumber": str(self.seq), + "libp2pPort": "8302", + "peerId": "12D3peer", + "isBlockProducer": True, + } + } + }, + ) + if not seq_part: + return httpx.Response(200, json={"errors": [{"message": "Missing sequence"}]}) + _, uuid, n = seq_part.split(" ") + if not _verify(pk, sig, struct.pack(">H", int(n)) + uuid.encode() + body): + return httpx.Response(401, text="bad signature") + if uuid != self.uuid or int(n) != self.seq: + return httpx.Response(412, text="stale") + self.seq = (self.seq + 1) % 65536 + return httpx.Response(200, json={"data": self.data}) + + +@pytest.fixture +def key(): + return ItnKey.from_base64(SEED) + + +def _client(key): + return ItnClient(URL, key, retries=1, retry_delay=0.0, timeout=5.0) + + +def test_key_round_trip_and_repr(key): + assert key.to_base64() == SEED + assert key.public_key_base64() in repr(key) + assert SEED not in repr(key) + assert ItnKey.generate().to_base64() != SEED + with pytest.raises(InvalidItnKeyError): + ItnKey.from_base64("not base64!") + with pytest.raises(InvalidItnKeyError): + ItnKey.from_base64(base64.b64encode(b"short").decode()) + + +@respx.mock +def test_auth_then_sequenced_requests(key): + server = MockItnServer(key, seq=7, data={"slotsWon": [1, 2]}) + respx.post(URL).mock(side_effect=server) + with _client(key) as itn: + auth = itn.auth() + assert (auth.server_uuid, auth.signer_sequence_number) == ("uuid-1", 7) + assert (auth.libp2p_port, auth.peer_id, auth.is_block_producer) == (8302, "12D3peer", True) + assert itn.slots_won() == [1, 2] + assert itn.slots_won() == [1, 2] + assert server.seq == 9 + assert [r["query"] for r in server.requests][1].startswith("query SlotsWon") + + +@respx.mock +def test_first_request_runs_auth_itself(key): + server = MockItnServer(key, data={"flushInternalLogs": "ok"}) + respx.post(URL).mock(side_effect=server) + with _client(key) as itn: + assert itn.last_auth is None + assert itn.flush_internal_logs(5) == "ok" + assert itn.last_auth is not None + assert server.requests[1]["variables"] == {"endLogId": 5} + + +@respx.mock +def test_sequence_number_wraps(key): + server = MockItnServer(key, seq=65535, data={"updateGating": "ok"}) + respx.post(URL).mock(side_effect=server) + with _client(key) as itn: + itn.update_gating(GatingUpdate()) + itn.update_gating(GatingUpdate()) + assert server.seq == 1 + + +@respx.mock +def test_412_runs_auth_again_and_repeats_once(key): + server = MockItnServer(key, data={"stopScheduledTransactions": "stopped"}) + respx.post(URL).mock(side_effect=server) + with _client(key) as itn: + itn.auth() + server.uuid = "uuid-2" # the daemon restarted + assert itn.stop_scheduled_transactions("h") == "stopped" + assert itn.last_auth is not None + assert itn.last_auth.server_uuid == "uuid-2" + + # 412, a new auth, 412 again. + server.status_override = [412, 412] + with pytest.raises(ItnSequencingError): + itn.stop_scheduled_transactions("h") + + +@respx.mock +def test_401_is_not_repeated(key): + server = MockItnServer(ItnKey.generate()) + route = respx.post(URL).mock(side_effect=server) + with _client(key) as itn, pytest.raises(ItnUnauthorizedError): + itn.slots_won() + assert route.call_count == 1 + + +@respx.mock +def test_transport_error_is_not_repeated(key): + server = MockItnServer(key, data={"slotsWon": []}) + calls = [] + + def flaky(request): + calls.append(request) + if len(calls) == 2: + raise httpx.ConnectError("reset") + return server(request) + + respx.post(URL).mock(side_effect=flaky) + with _client(key) as itn: + with pytest.raises(DaemonConnectionError): + itn.slots_won() + assert len(calls) == 2 + # The next request starts with a new auth. + assert itn.slots_won() == [] + assert server.requests[-2]["query"] == queries.AUTH + + +@respx.mock +def test_http_error_status_is_the_cause(key): + server = MockItnServer(key) + respx.post(URL).mock(side_effect=server) + with _client(key) as itn: + itn.auth() + server.status_override = [503] + with pytest.raises(DaemonConnectionError) as e: + itn.slots_won() + assert isinstance(e.value.__cause__, ItnHttpError) + assert e.value.__cause__.status_code == 503 + + +@respx.mock +def test_graphql_errors(key): + server = MockItnServer(key) + respx.post(URL).mock(side_effect=server) + with _client(key) as itn: + itn.auth() + server.data = None # type: ignore[assignment] + respx.post(URL).mock( + side_effect=lambda r: ( + server(r), + httpx.Response(200, json={"errors": [{"message": "boom"}]}), + )[1] + ) + with pytest.raises(GraphQLError): + itn.slots_won() + + +@respx.mock +def test_variables_of_the_operations(key): + server = MockItnServer( + key, + data={ + "internalLogs": [ + { + "id": 3, + "timestamp": "t", + "message": "m", + "metadata": [{"item": "k", "value": "{}"}], + "process": None, + } + ], + "schedulePayments": "h1", + "stopDaemon": "stopping", + "zkAppCommandLimit": None, + "updateGating": "ok", + }, + ) + respx.post(URL).mock(side_effect=server) + with _client(key) as itn: + (log,) = itn.internal_logs(2) + assert (log.id, log.metadata[0].item, log.process) == (3, "k", None) + details = PaymentsDetails( + duration_min=1, + tps=0.5, + memo_prefix="m", + max_fee=Currency.from_nanomina(20), + min_fee=Currency.from_nanomina(10), + amount=Currency.from_nanomina(1), + receiver="B62qr", + ) + assert itn.schedule_payments(details) == "h1" + assert itn.stop_daemon() == "stopping" + assert itn.set_zkapp_command_limit(None) is None + itn.update_gating( + GatingUpdate(trusted_peers=[NetworkPeer(host="1.2.3.4", libp2p_port=1, peer_id="p")]) + ) + variables = [r.get("variables") for r in server.requests[1:]] + assert variables[0] == {"startLogId": 2} + assert variables[1]["input"]["senders"] == [] + assert variables[1]["input"]["maxFee"] == "20" + assert variables[2] == {"delaySeconds": None, "cleanConfig": None} + assert variables[3] == {"limit": None} + assert variables[4]["input"]["trustedPeers"] == [ + {"host": "1.2.3.4", "libp2pPort": 1, "peerId": "p"} + ] diff --git a/tests/test_itn_integration.py b/tests/test_itn_integration.py new file mode 100644 index 0000000..ce8a250 --- /dev/null +++ b/tests/test_itn_integration.py @@ -0,0 +1,76 @@ +"""Integration tests against a running daemon's ITN GraphQL server. + +They need MINA_ITN_URI (e.g. http://127.0.0.1:3086/graphql) and MINA_ITN_KEY +(a base64 ed25519 seed whose public key is in --itn-keys), and are skipped +otherwise. None of them stops the daemon or sends transactions. +""" + +from __future__ import annotations + +import contextlib +import os + +import pytest + +from mina_sdk import GraphQLError +from mina_sdk.itn import GatingUpdate, ItnClient, ItnKey, ItnUnauthorizedError + +URI = os.environ.get("MINA_ITN_URI", "") +SEED = os.environ.get("MINA_ITN_KEY", "") + +pytestmark = pytest.mark.skipif(not (URI and SEED), reason="MINA_ITN_URI and MINA_ITN_KEY not set") + + +def _client(key: ItnKey | None = None) -> ItnClient: + return ItnClient(URI, key or ItnKey.from_base64(SEED), retries=2, retry_delay=1.0) + + +def test_auth(): + with _client() as itn: + auth = itn.auth() + assert auth.server_uuid + assert auth.libp2p_port > 0 + + +def test_unknown_key_is_rejected(): + with _client(ItnKey.generate()) as itn, pytest.raises(ItnUnauthorizedError): + itn.auth() + + +def test_sequenced_requests(): + with _client() as itn: + for _ in range(3): + itn.internal_logs(0) + assert itn.set_zkapp_command_limit(7) == 7 + assert itn.set_zkapp_command_limit(None) is None + + +def test_stale_sequence_number_is_recovered(): + # Two clients with the same key share the daemon's sequence number: each + # one's next request gets a 412, runs auth again and succeeds. + with _client() as a, _client() as b: + a.internal_logs(0) + b.internal_logs(0) + a.internal_logs(0) + b.internal_logs(0) + + +def test_internal_logs_and_flush(): + with _client() as itn: + logs = itn.internal_logs(0) + if logs: + assert logs[0].id <= logs[-1].id + itn.flush_internal_logs(logs[0].id) + + +def test_slots_won(): + # A block producer answers with its slots, or with a GraphQL error while + # its VRF evaluation runs; anything else is a transport or schema problem. + with _client() as itn, contextlib.suppress(GraphQLError): + assert all(isinstance(s, int) for s in itn.slots_won()) + + +def test_update_gating_empty(): + # An empty update with isolate=False leaves the node's gating open. + with _client() as itn: + assert isinstance(itn.update_gating(GatingUpdate()), str) diff --git a/tests/test_spec.py b/tests/test_spec.py index cb1f267..9ee042a 100644 --- a/tests/test_spec.py +++ b/tests/test_spec.py @@ -1,8 +1,8 @@ """Conformance to the specification in spec/ (a copy of o1-labs/mina-sdk-spec -at the tag in spec/VERSION): the query strings of mina_sdk.daemon.queries are -exactly the documents of spec/operations.graphql, up to white space, and every -document has one. mina-sdk-spec's CI validates the documents against the -daemon's schema. +at the tag in spec/VERSION): the query strings of mina_sdk.daemon.queries and +mina_sdk.itn.queries are exactly the documents of spec/operations.graphql and +spec/itn-operations.graphql, up to white space, and every document has one. +mina-sdk-spec's CI validates the documents against the daemon's schemas. """ from __future__ import annotations @@ -11,6 +11,7 @@ from pathlib import Path from mina_sdk.daemon import queries +from mina_sdk.itn import queries as itn_queries ROOT = Path(__file__).resolve().parent.parent @@ -58,6 +59,10 @@ def test_daemon_queries_are_the_spec_documents(): _assert_documents_are_the_spec("spec/operations.graphql", queries.ALL_DOCUMENTS) +def test_itn_queries_are_the_spec_documents(): + _assert_documents_are_the_spec("spec/itn-operations.graphql", itn_queries.ALL_ITN_DOCUMENTS) + + def test_checker_catches_a_changed_document(): changed = [d.replace("workIds", "workIdz") for d in queries.ALL_DOCUMENTS] try: