Session & Authentication

import binascii
 
def user_sign(key, msg: bytes) -> bytes:
    """
    User signing: sign hex-encoded message.
    Used for CreateSession.
    """
    payload = binascii.hexlify(msg)
    return key.sign(payload)
 
def session_sign(key, msg: bytes) -> bytes:
    """
    Session signing: sign raw bytes.
    Used for PlaceOrder, CancelOrder, etc.
    """
    return key.sign(msg)

import requests
from google.protobuf.internal import encoder, decoder
import schema_pb2
 
API_URL = "https://zo-devnet.n1.xyz"
 
def get_varint_bytes(value):
    """Encode an integer as a protobuf varint."""
    return encoder._VarintBytes(value)
 
def read_varint(buffer, offset=0):
    """Decode a protobuf varint from a buffer."""
    return decoder._DecodeVarint32(buffer, offset)
 
def execute_action(action, signing_key, sign_func):
    """
    Serialize, sign, and send an Action to the API.
    Returns the parsed Receipt.
    """
    # Serialize the action
    payload = action.SerializeToString()
    length_prefix = get_varint_bytes(len(payload))
    message = length_prefix + payload
    
    # Sign the message
    signature = sign_func(signing_key, message)
    
    # Send to API
    resp = requests.post(
        f"{API_URL}/action",
        data=message + signature,
        headers={"Content-Type": "application/octet-stream"}
    )
    
    if resp.status_code != 200:
        raise Exception(f"API error: {resp.status_code} {resp.text}")
    
    # Parse the receipt
    msg_len, pos = read_varint(resp.content, 0)
    receipt = schema_pb2.Receipt()
    receipt.ParseFromString(resp.content[pos:pos + msg_len])
    
    return receipt

import json
import requests
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from base58 import b58encode
 
import schema_pb2
 
API_URL = "https://zo-devnet.n1.xyz"
 
def create_session():
    # Load user wallet key
    with open("id.json", "r") as f:
        key_data = json.load(f)
    
    user_signkey = Ed25519PrivateKey.from_private_bytes(bytes(key_data[:32]))
    user_pubkey = user_signkey.public_key().public_bytes_raw()
    print(f"User pubkey: {b58encode(user_pubkey).decode()}")
    
    # Generate session keypair
    session_signkey = Ed25519PrivateKey.generate()
    session_pubkey = session_signkey.public_key().public_bytes_raw()
    print(f"Session pubkey: {b58encode(session_pubkey).decode()}")
    
    # Get server timestamp
    server_time = int(requests.get(f"{API_URL}/timestamp").json())
    print(f"Server time: {server_time}")
    
    # Build CreateSession action
    action = schema_pb2.Action()
    action.current_timestamp = server_time
    action.nonce = 0
    action.create_session.user_pubkey = user_pubkey
    action.create_session.session_pubkey = session_pubkey
    action.create_session.expiry_timestamp = server_time + 3600  # 1 hour
    
    # Execute with user signing
    receipt = execute_action(action, user_signkey, user_sign)
    
    if receipt.HasField("err"):
        error_name = schema_pb2.Error.Name(receipt.err)
        raise Exception(f"CreateSession failed: {error_name}")
    
    session_id = receipt.create_session_result.session_id
    print(f"Session created! ID: {session_id}")
    
    return session_id, session_signkey
 
if __name__ == "__main__":
    session_id, session_key = create_session()

import json
import requests
import binascii
from google.protobuf.internal import encoder, decoder
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from base58 import b58encode
 
import schema_pb2
 
API_URL = "https://zo-devnet.n1.xyz"
 
def get_varint_bytes(value):
    return encoder._VarintBytes(value)
 
def read_varint(buffer, offset=0):
    return decoder._DecodeVarint32(buffer, offset)
 
def user_sign(key, msg):
    return key.sign(binascii.hexlify(msg))
 
def session_sign(key, msg):
    return key.sign(msg)
 
def execute_action(action, signing_key, sign_func):
    payload = action.SerializeToString()
    length_prefix = get_varint_bytes(len(payload))
    message = length_prefix + payload
    signature = sign_func(signing_key, message)
    
    resp = requests.post(
        f"{API_URL}/action",
        data=message + signature,
        headers={"Content-Type": "application/octet-stream"}
    )
    resp.raise_for_status()
    
    msg_len, pos = read_varint(resp.content, 0)
    receipt = schema_pb2.Receipt()
    receipt.ParseFromString(resp.content[pos:pos + msg_len])
    return receipt
 
def main():
    # Load keys and create session
    with open("id.json", "r") as f:
        key_data = json.load(f)
    
    user_signkey = Ed25519PrivateKey.from_private_bytes(bytes(key_data[:32]))
    session_signkey = Ed25519PrivateKey.generate()
    
    server_time = int(requests.get(f"{API_URL}/timestamp").json())
    
    # Create session
    action = schema_pb2.Action()
    action.current_timestamp = server_time
    action.nonce = 0
    action.create_session.user_pubkey = user_signkey.public_key().public_bytes_raw()
    action.create_session.session_pubkey = session_signkey.public_key().public_bytes_raw()
    action.create_session.expiry_timestamp = server_time + 3600
    
    receipt = execute_action(action, user_signkey, user_sign)
    
    if receipt.HasField("err"):
        raise Exception(f"Error: {schema_pb2.Error.Name(receipt.err)}")
    
    session_id = receipt.create_session_result.session_id
    print(f"Session ID: {session_id}")
    
    # Ready to trade! See Trading Operations guide.
 
if __name__ == "__main__":
    main()