Skip to content

fix(python-sdk): add robust input validation and timeout bounding to REST requests #43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mozluk wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(python-sdk): add robust input validation and timeout bounding to REST requests #43

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 113 additions & 0 deletions common/validate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
"""Bounds and format checks for the values these examples sign.

Every script here signs a payload and submits it without any confirmation step, so a
mistyped literal becomes a real order, a real leverage change, or a real transfer. The
checks below are cheap local guards against the mistakes that are expensive and
irreversible rather than merely wrong: an extra zero on an amount, a leverage value
copied from a different venue's scale, an address with a typo that still base58-decodes.

Amounts are handled as ``Decimal`` from a *string*, never as ``float``. A USDC amount
has six decimals, and binary floating point cannot represent most decimal fractions
exactly: ``0.1 + 0.2`` is ``0.30000000000000004``, and ``int(round(4200.69 * 1_000_000))``
depends on how the literal happened to round. The API accepts amounts as strings for
this reason, and signatures are computed over the exact string, so keeping the value in
decimal form all the way through is both safer and required for the signature to match
what the user intended.
"""

from decimal import Decimal, InvalidOperation

import base58

# USDC on Solana has six decimal places. A value with more precision than the token
# supports would be silently truncated somewhere downstream.
USDC_DECIMALS = 6
USDC_SCALE = Decimal(10) ** USDC_DECIMALS

# Pacifica's documented leverage range. Checked so a value from another venue's scale
# (or a stray extra digit) is rejected locally instead of being signed and sent.
MIN_LEVERAGE = 1
MAX_LEVERAGE = 50

PUBLIC_KEY_BYTES = 32


def check_amount(value, *, name="amount", minimum="0", maximum=None):
"""Validate a decimal amount given as a string and return it as ``Decimal``.

``value`` must be a string (or ``Decimal``); passing a ``float`` is rejected rather
than coerced, because by the time a float reaches this function the precision loss
has already happened and there is nothing useful left to check.
"""
if isinstance(value, float):
raise TypeError(
f"{name} was passed as a float ({value!r}); pass it as a string such as "
f'"{value:.6f}" so the decimal value is exact.'
)
try:
amount = Decimal(str(value))
except (InvalidOperation, ValueError) as exc:
raise ValueError(f"{name}={value!r} is not a valid decimal number.") from exc

if not amount.is_finite():
raise ValueError(f"{name}={value!r} is not a finite number.")
if amount <= Decimal(minimum):
raise ValueError(f"{name} must be greater than {minimum}; got {amount}.")
if maximum is not None and amount > Decimal(maximum):
raise ValueError(
f"{name}={amount} exceeds the configured ceiling of {maximum}. Raise the "
"ceiling deliberately if this is intended."
)
if -amount.as_tuple().exponent > USDC_DECIMALS:
raise ValueError(
f"{name}={amount} has more than {USDC_DECIMALS} decimal places, which is "
"more precision than the token supports."
)
return amount


def to_base_units(amount):
"""Convert a validated ``Decimal`` amount to integer base units (6 decimals).

The multiplication is exact because both operands are ``Decimal``. ``check_amount``
has already rejected anything with more than six decimal places, so the result has
no fractional part to round away - and this asserts that rather than assuming it.
"""
scaled = Decimal(amount) * USDC_SCALE
if scaled != scaled.to_integral_value():
raise ValueError(f"amount {amount} does not convert to whole base units.")
return int(scaled)


def check_leverage(value):
"""Validate a leverage setting and return it as ``int``."""
if isinstance(value, bool) or not isinstance(value, int):
raise TypeError(f"leverage must be an integer; got {value!r}.")
if not MIN_LEVERAGE <= value <= MAX_LEVERAGE:
raise ValueError(
f"leverage={value} is outside the supported range "
f"{MIN_LEVERAGE}-{MAX_LEVERAGE}x."
)
return value


def check_address(value, *, name="address"):
"""Validate that a value is a base58 32-byte Solana address and return it.

A transfer destination cannot be recovered if it is wrong, and a typo in base58
often still decodes - just to different bytes - so the length check is the part that
catches real mistakes.
"""
if not isinstance(value, str) or not value.strip():
raise ValueError(f"{name} must be a non-empty base58 string.")
value = value.strip()
try:
decoded = base58.b58decode(value)
except Exception as exc:
raise ValueError(f"{name} is not valid base58.") from exc
if len(decoded) != PUBLIC_KEY_BYTES:
raise ValueError(
f"{name} decodes to {len(decoded)} bytes; a Solana address is "
f"{PUBLIC_KEY_BYTES} bytes."
)
return value
279 changes: 140 additions & 139 deletions rest/api_agent_keys.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,139 +1,140 @@
"""
This example shows how to bind an api agent key (also called agent wallet)
to an account and use the api agent key to sign on behalf of the account
to create a market order.
"""

import time
import uuid

import requests
from solders.keypair import Keypair

from common.constants import REST_URL
from common.utils import sign_message


BIND_AGENT_WALLET_API_URL = f"{REST_URL}/agent/bind"
MARKET_ORDER_API_URL = f"{REST_URL}/orders/create_market"
PRIVATE_KEY = "" # e.g. "2Z2Wn4kN5ZNhZzuFTQSyTiN4ixX8U6ew5wPDJbHngZaC3zF3uWNj4dQ63cnGfXpw1cESZPCqvoZE7VURyuj9kf8b"


def main():
# Generate account based on private key
keypair = Keypair.from_base58_string(PRIVATE_KEY)
public_key = str(keypair.pubkey())

# Generate a new agent wallet
agent_wallet_private_key = Keypair()
agent_wallet_public_key = str(agent_wallet_private_key.pubkey())

# ---------------------------------------------------------------
# Bind agent wallet
# ---------------------------------------------------------------

# Scaffold the signature header
timestamp = int(time.time() * 1_000)

signature_header = {
"timestamp": timestamp,
"expiry_window": 5_000,
"type": "bind_agent_wallet",
}

# Construct the signature payload
signature_payload = {
"agent_wallet": agent_wallet_public_key,
}

# Use the helper function to sign the message
message, signature = sign_message(signature_header, signature_payload, keypair)

# Construct the request reusing the payload and constructing common request fields
request_header = {
"account": public_key,
"signature": signature,
"timestamp": signature_header["timestamp"],
"expiry_window": signature_header["expiry_window"],
}

# Send the request
headers = {"Content-Type": "application/json"}

request = {
**request_header,
**signature_payload,
}

response = requests.post(BIND_AGENT_WALLET_API_URL, json=request, headers=headers)
print(f"Status Code: {response.status_code}")
print(f"Response: {response.text}")
print(f"Request: {request}")

# Print details for debugging
print("\nDebug Info:")
print(f"Address: {public_key}")
print(f"Agent Wallet: {agent_wallet_public_key}")
print(f"Message: {message}")
print(f"Signature: {signature}")
print("\n")

# ---------------------------------------------------------------
# Create market order
# ---------------------------------------------------------------

# Scaffold the signature header
timestamp = int(time.time() * 1_000)

signature_header = {
"timestamp": timestamp,
"expiry_window": 5_000,
"type": "create_market_order",
}

# Construct the signature payload
signature_payload = {
"symbol": "BTC",
"reduce_only": False,
"amount": "0.1",
"side": "bid",
"slippage_percent": "0.5",
"client_order_id": str(uuid.uuid4()),
}

# Use the helper function to sign the message, with the agent wallet's private key
message, signature = sign_message(
signature_header, signature_payload, agent_wallet_private_key
)

# Construct the request reusing the payload and constructing common request fields
request_header = {
"account": public_key,
"agent_wallet": agent_wallet_public_key, # use the agent wallet's public key
"signature": signature,
"timestamp": signature_header["timestamp"],
"expiry_window": signature_header["expiry_window"],
}

# Send the request
headers = {"Content-Type": "application/json"}

request = {
**request_header,
**signature_payload,
}

response = requests.post(MARKET_ORDER_API_URL, json=request, headers=headers)
print(f"Status Code: {response.status_code}")
print(f"Response: {response.text}")
print(f"Request: {request}")

# Print details for debugging
print("\nDebug Info:")
print(f"Address: {public_key}")
print(f"Message: {message}")
print(f"Signature: {signature}")


if __name__ == "__main__":
main()
"""
This example shows how to bind an api agent key (also called agent wallet)
to an account and use the api agent key to sign on behalf of the account
to create a market order.
"""

import time
import uuid

import requests
from solders.keypair import Keypair

from common.constants import REQUEST_TIMEOUT, REST_URL
from common.utils import sign_message
from common.env import load_private_key


BIND_AGENT_WALLET_API_URL = f"{REST_URL}/agent/bind"
MARKET_ORDER_API_URL = f"{REST_URL}/orders/create_market"
PRIVATE_KEY = load_private_key("PACIFICA_PRIVATE_KEY")


def main():
# Generate account based on private key
keypair = Keypair.from_base58_string(PRIVATE_KEY)
public_key = str(keypair.pubkey())

# Generate a new agent wallet
agent_wallet_private_key = Keypair()
agent_wallet_public_key = str(agent_wallet_private_key.pubkey())

# ---------------------------------------------------------------
# Bind agent wallet
# ---------------------------------------------------------------

# Scaffold the signature header
timestamp = int(time.time() * 1_000)

signature_header = {
"timestamp": timestamp,
"expiry_window": 5_000,
"type": "bind_agent_wallet",
}

# Construct the signature payload
signature_payload = {
"agent_wallet": agent_wallet_public_key,
}

# Use the helper function to sign the message
message, signature = sign_message(signature_header, signature_payload, keypair)

# Construct the request reusing the payload and constructing common request fields
request_header = {
"account": public_key,
"signature": signature,
"timestamp": signature_header["timestamp"],
"expiry_window": signature_header["expiry_window"],
}

# Send the request
headers = {"Content-Type": "application/json"}

request = {
**request_header,
**signature_payload,
}

response = requests.post(BIND_AGENT_WALLET_API_URL, json=request, headers=headers, timeout=REQUEST_TIMEOUT)
print(f"Status Code: {response.status_code}")
print(f"Response: {response.text}")
print(f"Request: {request}")

# Print details for debugging
print("\nDebug Info:")
print(f"Address: {public_key}")
print(f"Agent Wallet: {agent_wallet_public_key}")
print(f"Message: {message}")
print(f"Signature: {signature}")
print("\n")

# ---------------------------------------------------------------
# Create market order
# ---------------------------------------------------------------

# Scaffold the signature header
timestamp = int(time.time() * 1_000)

signature_header = {
"timestamp": timestamp,
"expiry_window": 5_000,
"type": "create_market_order",
}

# Construct the signature payload
signature_payload = {
"symbol": "BTC",
"reduce_only": False,
"amount": "0.1",
"side": "bid",
"slippage_percent": "0.5",
"client_order_id": str(uuid.uuid4()),
}

# Use the helper function to sign the message, with the agent wallet's private key
message, signature = sign_message(
signature_header, signature_payload, agent_wallet_private_key
)

# Construct the request reusing the payload and constructing common request fields
request_header = {
"account": public_key,
"agent_wallet": agent_wallet_public_key, # use the agent wallet's public key
"signature": signature,
"timestamp": signature_header["timestamp"],
"expiry_window": signature_header["expiry_window"],
}

# Send the request
headers = {"Content-Type": "application/json"}

request = {
**request_header,
**signature_payload,
}

response = requests.post(MARKET_ORDER_API_URL, json=request, headers=headers, timeout=REQUEST_TIMEOUT)
print(f"Status Code: {response.status_code}")
print(f"Response: {response.text}")
print(f"Request: {request}")

# Print details for debugging
print("\nDebug Info:")
print(f"Address: {public_key}")
print(f"Message: {message}")
print(f"Signature: {signature}")


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