Andrew Miller
5410b283e2
The HTTP Session was hardcoded with no injection point and followed redirects by default, so a server-side caller pointing the client at an attacker-influenced base_url (a tenant's prime_endpoint) had no way to attach an SSRF policy, and a public endpoint could 302-redirect the request to an internal address (e.g. the cloud metadata service). - Client/DragonchainSDK now accept an optional `session` so callers can inject a Session whose transport adapter refuses internal IPs. Default stays unguarded for trusted/CLI use — the guard belongs in the server. - Requests are sent with allow_redirects=False; Prime never legitimately redirects, and a 3xx now surfaces to the caller instead of being followed.
159 lines
4.5 KiB
Python
159 lines
4.5 KiB
Python
"""Python SDK for interacting with Dragonchain (Prime) nodes.
|
|
|
|
Example — fire and forget (most use cases)::
|
|
|
|
from prime_sdk import DragonchainSDK, TransactionCreateRequest
|
|
|
|
client = DragonchainSDK(
|
|
"your-public-id",
|
|
"your-auth-key-id",
|
|
"your-auth-key",
|
|
"https://your-dragonchain-endpoint.com",
|
|
)
|
|
|
|
resp = client.transaction.create(
|
|
TransactionCreateRequest(
|
|
txn_type="my-transaction-type",
|
|
payload='{"message": "Hello Dragonchain"}',
|
|
)
|
|
)
|
|
print("Transaction ID:", resp.transaction_id)
|
|
# Done — no need to wait for a block.
|
|
|
|
Example — wait for block inclusion (interchain / Daria)::
|
|
|
|
import time
|
|
|
|
resp = client.transaction.create(
|
|
TransactionCreateRequest(
|
|
txn_type="my-transaction-type",
|
|
payload='{"message": "Hello Dragonchain"}',
|
|
)
|
|
)
|
|
|
|
while True:
|
|
txn = client.transaction.get(resp.transaction_id)
|
|
if txn.header.block_id:
|
|
print("Block ID:", txn.header.block_id)
|
|
break
|
|
time.sleep(2)
|
|
"""
|
|
|
|
from .block import BlockClient
|
|
from .client import CONTENT_TYPE_JSON, Client
|
|
from .contract import ContractClient
|
|
from .errors import DragonchainAPIError, DragonchainError
|
|
from .models import (
|
|
Block,
|
|
BlockHeader,
|
|
BlockProof,
|
|
GrpcConnectionInfo,
|
|
InterchainTrace,
|
|
InterchainTransaction,
|
|
ListResponse,
|
|
ListTransactionsResponse,
|
|
SmartContract,
|
|
SmartContractCreateRequest,
|
|
SmartContractExecutionInfo,
|
|
SmartContractUpdateRequest,
|
|
SuccessResponse,
|
|
SystemStatus,
|
|
Transaction,
|
|
TransactionBulkRequest,
|
|
TransactionBulkResponse,
|
|
TransactionCreateRequest,
|
|
TransactionCreateResponse,
|
|
TransactionHeader,
|
|
TransactionListResponse,
|
|
TransactionProof,
|
|
TransactionType,
|
|
TransactionTypeCreateRequest,
|
|
TransactionTypeCreateResponse,
|
|
VerificationBlock,
|
|
)
|
|
from .system import SystemClient
|
|
from .transaction import TransactionClient
|
|
from .transaction_type import TransactionTypeClient
|
|
|
|
__version__ = "0.2.0"
|
|
|
|
|
|
class DragonchainSDK:
|
|
"""Main SDK client for interacting with Dragonchain nodes.
|
|
|
|
Provides access to all API endpoints through specialized client instances:
|
|
``system``, ``transaction``, ``transaction_type``, ``contract``, ``block``.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
public_id: str,
|
|
auth_key_id: str,
|
|
auth_key: str,
|
|
base_url: str,
|
|
session: "Optional[requests.Session]" = None,
|
|
):
|
|
"""Create a new Dragonchain SDK client.
|
|
|
|
Args:
|
|
public_id: Your Dragonchain public ID.
|
|
auth_key_id: Your authentication key ID.
|
|
auth_key: Your authentication key.
|
|
base_url: Base URL of your node (e.g. "https://chains.dragonchain.com").
|
|
session: Optional pre-configured ``requests.Session``. Inject one
|
|
with an SSRF-guarded transport adapter when ``base_url`` is
|
|
attacker-influenced (a tenant's prime_endpoint); the default
|
|
session is unguarded and intended for trusted/CLI use.
|
|
"""
|
|
self._client = Client(public_id, auth_key_id, auth_key, base_url, session=session)
|
|
self.transaction = TransactionClient(self._client)
|
|
self.transaction_type = TransactionTypeClient(self._client)
|
|
self.contract = ContractClient(self._client)
|
|
self.block = BlockClient(self._client)
|
|
self.system = SystemClient(self._client)
|
|
|
|
def get_client(self) -> Client:
|
|
"""Return the underlying HTTP client (advanced use)."""
|
|
return self._client
|
|
|
|
|
|
__all__ = [
|
|
"DragonchainSDK",
|
|
"Client",
|
|
"CONTENT_TYPE_JSON",
|
|
"DragonchainError",
|
|
"DragonchainAPIError",
|
|
"BlockClient",
|
|
"ContractClient",
|
|
"SystemClient",
|
|
"TransactionClient",
|
|
"TransactionTypeClient",
|
|
# models
|
|
"Block",
|
|
"BlockHeader",
|
|
"BlockProof",
|
|
"GrpcConnectionInfo",
|
|
"InterchainTrace",
|
|
"InterchainTransaction",
|
|
"ListResponse",
|
|
"ListTransactionsResponse",
|
|
"SmartContract",
|
|
"SmartContractCreateRequest",
|
|
"SmartContractExecutionInfo",
|
|
"SmartContractUpdateRequest",
|
|
"SuccessResponse",
|
|
"SystemStatus",
|
|
"Transaction",
|
|
"TransactionBulkRequest",
|
|
"TransactionBulkResponse",
|
|
"TransactionCreateRequest",
|
|
"TransactionCreateResponse",
|
|
"TransactionHeader",
|
|
"TransactionListResponse",
|
|
"TransactionProof",
|
|
"TransactionType",
|
|
"TransactionTypeCreateRequest",
|
|
"TransactionTypeCreateResponse",
|
|
"VerificationBlock",
|
|
]
|