Run a strategy programmatically

#!/usr/bin/env python3
"""Compile a public Turbine strategy and push it to your Locus runner.

Turbine supplies canonical Python for a public slug. The wallet owner signs the
exact code locally and sends it straight to the runner; Turbine is not involved
in the code push and never receives the private key.
"""

import hashlib
import json
import os
import re
import sys
import time
from urllib.parse import urlsplit, urlunsplit

import requests
from eth_account import Account
from eth_account.messages import encode_typed_data


DEFAULT_API_URL = "https://api.turbinefi.com"
MAX_CODE_BYTES = 2_000_000
ZERO_HASH = "0x" + "00" * 32
SLUG_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
REQUEST_TIMEOUT = (5, 30)

EIP712_DOMAIN = {
    "name": "Turbine Bot Studio",
    "version": "1",
    "chainId": 137,
}
EIP712_TYPES = {
    "BotCodeUpdate": [
        {"name": "codeHash", "type": "bytes32"},
        {"name": "envHash", "type": "bytes32"},
        {"name": "wallet", "type": "address"},
        {"name": "timestamp", "type": "uint256"},
        {"name": "intentId", "type": "string"},
        {"name": "authorizationNonce", "type": "bytes32"},
    ],
}


class PortablePushError(RuntimeError):
    """A safe-to-display error that never contains remote bodies or secrets."""


def _base_url(value, label, *, require_https=True):
    try:
        parsed = urlsplit(value.strip())
    except (AttributeError, ValueError) as exc:
        raise PortablePushError(f"{label} is not a valid URL") from exc
    if not parsed.netloc or parsed.username or parsed.password:
        raise PortablePushError(f"{label} is not a valid URL")
    if require_https and parsed.scheme != "https":
        raise PortablePushError(f"{label} must use HTTPS")
    if parsed.scheme not in ("http", "https") or parsed.query or parsed.fragment:
        raise PortablePushError(f"{label} is not a valid URL")
    return urlunsplit((parsed.scheme, parsed.netloc, parsed.path.rstrip("/"), "", ""))


def _request(session, method, url, **kwargs):
    try:
        return session.request(
            method,
            url,
            timeout=REQUEST_TIMEOUT,
            allow_redirects=False,
            **kwargs,
        )
    except requests.RequestException as exc:
        raise PortablePushError(f"{method} request failed") from exc


def _json_object(response, operation):
    try:
        value = response.json()
    except (ValueError, TypeError) as exc:
        raise PortablePushError(f"{operation} returned invalid JSON") from exc
    if not isinstance(value, dict):
        raise PortablePushError(f"{operation} returned invalid JSON")
    return value


def fetch_strategy_python(slug, turbine_api_url=DEFAULT_API_URL, session=None):
    """Return (source, sha256_hex) for one eligible public strategy slug."""
    if not isinstance(slug, str) or len(slug) > 160 or not SLUG_PATTERN.fullmatch(slug):
        raise PortablePushError("strategy slug must be lowercase letters, numbers, and hyphens")
    api_url = _base_url(turbine_api_url, "TURBINE_API_URL", require_https=False)
    session = session or requests.Session()
    response = _request(
        session,
        "GET",
        f"{api_url}/api/v1/strategies/{slug}/python",
        stream=True,
        headers={"Accept": "text/x-python"},
    )
    if response.status_code != 200:
        raise PortablePushError(f"strategy compile failed with HTTP {response.status_code}")
    if not response.headers.get("Content-Type", "").lower().startswith("text/x-python"):
        raise PortablePushError("strategy compile returned an unexpected content type")

    body = bytearray()
    for chunk in response.iter_content(chunk_size=64 * 1024):
        body.extend(chunk)
        if len(body) > MAX_CODE_BYTES:
            raise PortablePushError(f"compiled strategy is larger than {MAX_CODE_BYTES} bytes")
    digest = hashlib.sha256(body).hexdigest()
    expected = response.headers.get("X-Turbine-Code-SHA256", "").lower()
    if not re.fullmatch(r"[0-9a-f]{64}", expected) or expected != digest:
        raise PortablePushError("compiled strategy hash did not match the response header")
    try:
        source = bytes(body).decode("utf-8")
    except UnicodeDecodeError as exc:
        raise PortablePushError("compiled strategy was not valid UTF-8") from exc
    return source, digest


def push_strategy(
    slug,
    runner_url,
    private_key,
    turbine_api_url=DEFAULT_API_URL,
    session=None,
    now=time.time,
):
    """Compile, locally sign, and directly push a public strategy to Locus."""
    runner_url = _base_url(runner_url, "LOCUS_RUNNER_URL")
    session = session or requests.Session()
    try:
        account = Account.from_key(private_key)
    except (TypeError, ValueError) as exc:
        raise PortablePushError("WALLET_PRIVATE_KEY is invalid") from exc

    code, digest = fetch_strategy_python(slug, turbine_api_url, session)
    auth_response = _request(
        session,
        "GET",
        f"{runner_url}/auth-info",
        headers={"Accept": "application/json"},
    )
    if auth_response.status_code != 200:
        raise PortablePushError(f"runner wallet check failed with HTTP {auth_response.status_code}")
    authorized_wallet = _json_object(auth_response, "runner wallet check").get("authorized_wallet")
    if not isinstance(authorized_wallet, str) or authorized_wallet.lower() != account.address.lower():
        raise PortablePushError("runner is authorized for a different wallet")

    timestamp = int(now())
    signable = encode_typed_data(
        EIP712_DOMAIN,
        EIP712_TYPES,
        {
            "codeHash": bytes.fromhex(digest),
            "envHash": bytes(32),
            "wallet": account.address,
            "timestamp": timestamp,
            "intentId": "",
            "authorizationNonce": bytes(32),
        },
    )
    signature = Account.sign_message(signable, private_key=private_key).signature.hex()
    payload = {
        "code": code,
        "codeHash": "0x" + digest,
        "env": {},
        "envHash": ZERO_HASH,
        "signature": signature,
        "wallet": account.address,
        "timestamp": timestamp,
        "intentId": "",
        "authorizationNonce": ZERO_HASH,
    }
    push_response = _request(
        session,
        "POST",
        f"{runner_url}/code",
        headers={"Content-Type": "application/json", "Accept": "application/json"},
        json=payload,
    )
    if push_response.status_code != 200:
        raise PortablePushError(f"runner code push failed with HTTP {push_response.status_code}")
    _json_object(push_response, "runner code push")
    return {
        "slug": slug,
        "runner": runner_url,
        "signer": account.address,
        "code_sha256": digest,
    }


def main():
    try:
        result = push_strategy(
            slug=os.environ["TURBINE_STRATEGY_SLUG"],
            runner_url=os.environ["LOCUS_RUNNER_URL"],
            private_key=os.environ["WALLET_PRIVATE_KEY"],
            turbine_api_url=os.environ.get("TURBINE_API_URL", DEFAULT_API_URL),
        )
    except KeyError as exc:
        print(f"Missing required environment variable: {exc.args[0]}", file=sys.stderr)
        return 2
    except PortablePushError as exc:
        print(f"Push failed: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())