Back to skill

Security audit

Katbot Trading

Security checks for vulnerabilities and agentic risk

Overview

This live-trading skill is mostly coherent, but it handles high-value trading credentials too broadly and under-discloses where they are sent.

Review this carefully before installing. Only use it if you trust Katbot.ai and understand that an agent trading private key may be transmitted to the API. Do not use custom KATBOT_BASE_URL values unless you control and trust the endpoint, prefer testnet/paper mode first, and require explicit confirmation before any live trade or position close.

Vulnerability Patterns
  • Insecure DependenciesIntroduces malicious components through unsafe dependency sources
  • Insecure Skill Coding PracticesFinds exploitable flaws such as hardcoded secrets or command injection
  • Skill Instruction HijackingAlters the agent's session goals or safety constraints when the skill loads
  • Agent Memory PoisoningWrites attacker-controlled rules into memory that affect later sessions
  • Remote Payload Retrieval and ExecutionFetches external code whose behavior can change after review
Findings (6)

T09 · Insecure Skill Coding Practices

Error
Location
tools/katbot_client.py:247
Finding
Hyperliquid Agent Private Key Is Transmitted to Excessive and Configurable API Endpoints<![CDATA[ ## Vulnerability Details **File Location**: `tools/katbot_client.py:247-258`, with affected calls throughout `tools/katbot_client.py` **Vulnerability Type**: Excessive credential transmission and failure to enforce least privilege **Risk Level**: Critical ### Vulnerable Code ```python def _auth(token: str, agent_key: str = None) -> dict: """Build auth headers with optional agent private key. CRITICAL: ALWAYS include X-Agent-Private-Key for Hyperliquid portfolio calls. The API requires this header for all Hyperliquid portfolio endpoints. """ headers = {"Authorization": f"Bearer {token}"} # Always include agent key if available - required for Hyperliquid portfolios if agent_key: headers["X-Agent-Private-Key"] = agent_key elif AGENT_PRIVATE_KEY: headers["X-Agent-Private-Key"] = AGENT_PRIVATE_KEY return headers ``` The helper is then used for operations that do not require a trading key, including: ```python def list_portfolios(token: str) -> list: """List all portfolios for the authenticated user.""" r = requests.get(f"{BASE_URL}/portfolio", headers=_auth(token)) r.raise_for_status() return r.json() ``` ```python def list_agents(token: str) -> list: """List all agents owned by the authenticated user.""" r = requests.get(f"{BASE_URL}/agents", headers=_auth(token)) r.raise_for_status() return r.json() ``` ```python def get_user(token: str) -> dict: """Get current authenticated user details and subscription info.""" r = requests.get(f"{BASE_URL}/user", headers=_auth(token)) r.raise_for_status() return r.json() ``` Market-intelligence calls use the same helper: ```python r = requests.get(f"{BASE_URL}/market-intelligence/trending", params=params, headers=_auth(token)) ``` The destination is configurable: ```python BASE_URL = os.getenv("KATBOT_BASE_URL") ... if not BASE_URL: BASE_URL = os.getenv("KATBOT_BASE_URL", "https://api.katb ...[truncated 2060 chars]
Remediation
<![CDATA[ ## Remediation Suggestions 1. Make the default authentication helper return only the bearer token: ```python def _auth(token: str) -> dict: return {"Authorization": f"Bearer {token}"} ``` 2. Introduce a separate helper for the small set of operations that genuinely require the trading key: ```python def _trading_auth(token: str, agent_key: str) -> dict: return { "Authorization": f"Bearer {token}", "X-Agent-Private-Key": agent_key, } ``` 3. Require each credential-bearing call site to explicitly request the trading key. 4. Do not send the key to market-intelligence, user, plan, agent-management, research-listing, polling, or other read-only endpoints. 5. Restrict credential-bearing requests to an allowlist containing the expected HTTPS origin. 6. Reject plaintext HTTP base URLs. 7. Disable cross-origin redirects for credential-bearing calls or verify every redirect target before following it. 8. Present user consent at the point where the key will actually be transmitted, rather than relying only on documentation. 9. Prefer local transaction signing so the remote API never receives the private key. ]]>

T09 · Insecure Skill Coding Practices

Error
Location
tools/katbot_onboard.py:61
Finding
Onboarding Signs Unvalidated Messages Supplied by an Arbitrary Remote Server<![CDATA[ ## Vulnerability Details **File Location**: `tools/katbot_onboard.py:61-82` **Vulnerability Type**: Unvalidated SIWE message signing and signature phishing **Risk Level**: High ### Vulnerable Code ```python def siwe_login(base_url: str, private_key: str, chain_id: int) -> tuple[str, str, str]: """Authenticate with SIWE. Returns (access_token, refresh_token, wallet_address).""" account = Account.from_key(private_key) address = account.address print(f" Wallet address : {cyan(address)}") print(f" Authenticating with {base_url} ...") # Step 1: Get nonce r = requests.get(f"{base_url}/get-nonce/{address}?chain_id={chain_id}", timeout=15) r.raise_for_status() message_text = r.json()["message"] # Step 2: Sign signable = encode_defunct(text=message_text) signed = Account.sign_message(signable, private_key) signature = signed.signature.hex() # Step 3: Login r = requests.post( f"{base_url}/login", json={"address": address, "signature": signature, "chain_id": chain_id}, timeout=15, ) r.raise_for_status() ``` The remote origin is directly configurable: ```python parser.add_argument( "--base-url", default=DEFAULT_BASE_URL, help=f"Katbot API base URL (default: {DEFAULT_BASE_URL})" ) ``` The core client repeats the same pattern in `tools/katbot_client.py:187-197`. ### Technical Analysis The script treats arbitrary text returned by `/get-nonce/...` as a valid SIWE authentication message and signs it with the user’s MetaMask private key. It does not parse or validate the message’s: - Domain. - URI. - Wallet address. - Chain ID. - Nonce. - Issued-at timestamp. - Expiration time. - Statement or requested resources. Because `--base-url` accepts an arbitrary destination, a hostile endpoint can provide attacker-chosen text for the wallet to sign. Although the private key itself is not transmitted, an attacker receives a valid signature over content the attac ...[truncated 1270 chars]
Remediation
<![CDATA[ ## Remediation Suggestions 1. Parse the response as a standards-compliant SIWE message before signing. 2. Verify that the SIWE domain and URI exactly match the trusted HTTPS Katbot origin. 3. Verify that the message address equals the address derived from the local private key. 4. Verify the requested chain ID, nonce, issuance time, and expiration time. 5. Reject unexpected statements, resources, or malformed fields. 6. Restrict onboarding to `https://api.katbot.ai` by default. 7. If custom origins are required for development, require an explicit insecure-development flag and a separate confirmation displaying the complete message and origin. 8. Reject non-HTTPS URLs outside a clearly identified localhost development mode. 9. Apply the same validation to the fallback authentication implementation in `katbot_client.py`. ]]>

T09 · Insecure Skill Coding Practices

Warning
Location
tools/katbot_onboard.py:116
Finding
Credential Files Are Created Non-Atomically Before Restrictive Permissions Are Applied<![CDATA[ ## Vulnerability Details **File Location**: `tools/katbot_onboard.py:116-137` **Vulnerability Type**: Insecure sensitive-file creation and symlink handling **Risk Level**: Medium ### Vulnerable Code ```python def save_identity(identity_dir: str, config: dict, agent_private_key: str, jwt_token: str, refresh_token: str = ""): """Write katbot_config.json and katbot_token.json to the identity directory.""" os.makedirs(identity_dir, exist_ok=True) config_path = os.path.join(identity_dir, "katbot_config.json") with open(config_path, "w") as f: json.dump(config, f, indent=2) # Store agent private key in a local secrets file (not committed to git) secrets_path = os.path.join(identity_dir, "katbot_secrets.json") with open(secrets_path, "w") as f: json.dump({"agent_private_key": agent_private_key}, f, indent=2) os.chmod(secrets_path, 0o600) # Save JWT and refresh tokens for reuse token_path = os.path.join(identity_dir, "katbot_token.json") with open(token_path, "w") as f: json.dump({"access_token": jwt_token, "refresh_token": refresh_token}, f, indent=2) os.chmod(token_path, 0o600) ``` Token refresh in `tools/katbot_client.py:162-170` uses the same pattern: ```python os.makedirs(IDENTITY_DIR, exist_ok=True) with open(TOKEN_FILE, "w") as f: json.dump({"access_token": new_access, "refresh_token": new_refresh}, f, indent=2) try: os.chmod(TOKEN_FILE, 0o600) except Exception: pass ``` ### Technical Analysis The identity directory is created without explicitly requesting mode `0700`. The secret and token files are opened using the process umask and are only changed to `0600` after their full contents have been written and the file has been closed. This creates several weaknesses: - The file may initially be more permissive than intended under an unsafe umask. - `open(..., "w")` follows symbolic links. - Existing files are truncated in place. - There is no verification that t ...[truncated 1145 chars]
Remediation
<![CDATA[ ## Remediation Suggestions 1. Create the identity directory with mode `0700` and verify its owner: ```python os.makedirs(identity_dir, mode=0o700, exist_ok=True) os.chmod(identity_dir, 0o700) ``` 2. Reject identity directories and destination files that are symbolic links. 3. Create temporary files using `os.open()` with `O_CREAT | O_EXCL | O_NOFOLLOW` and mode `0600`. 4. Write and flush the complete JSON document, call `fsync()`, and atomically replace the final path. 5. Verify owner and mode after replacement. 6. Treat a failed permission operation as a fatal security error rather than silently continuing. 7. Apply the secure writer consistently to onboarding, authentication, and token refresh. 8. Consider using an operating-system credential store instead of plaintext JSON private-key storage. ]]>

T08 · Insecure Dependencies

Warning
Location
requirements.txt:1
Finding
Unpinned Dependencies Are Automatically Installed into the Active Python Environment<![CDATA[ ## Vulnerability Details **File Location**: `requirements.txt:1-2` and `tools/ensure_env.sh:47-51` **Vulnerability Type**: Dependency supply-chain exposure **Risk Level**: Medium ### Vulnerable Code `requirements.txt` contains open-ended minimum constraints: ```text eth-account>=0.8.0 requests>=2.31.0 ``` The installer runs pip in the active Python environment: ```bash python3 -m pip install --quiet -r "$REQUIREMENTS" # ── Write version stamp on success ──────────────────────────────────────────── echo "$CURRENT_VERSION" > "$STAMP_FILE" echo "✅ Dependencies installed for katbot-trading@${CURRENT_VERSION}." ``` The Skill documentation requires this script to run before every tool invocation, with installation triggered when the local stamp is missing or differs from the version in `SKILL.md`. ### Technical Analysis The requirements do not pin exact versions and provide no package hashes. Any future release satisfying the broad constraints may be installed. The code also does not create or enforce an isolated virtual environment, so packages can be added to or upgraded in the interpreter environment used by other tools. Security therefore depends on: - The current state of the configured package index. - All future releases of the direct dependencies. - Their transitive dependency resolution. - Local pip configuration and index overrides. - The integrity of mutable package metadata. The repeated pre-tool installation design increases exposure because a stamp change or deletion can trigger installation during routine use. ### Attack Path 1. A dependency or transitive dependency publishes a compromised release satisfying the minimum version constraint. 2. Alternatively, the process is configured to use a malicious package index or mirror. 3. The installed-version stamp is absent, deleted, or changed by a Skill upgrade. 4. `ensure_env.sh` invokes pip. 5. Pip resolves and installs the compromised package into the active environment. 6. Maliciou ...[truncated 612 chars]
Remediation
<![CDATA[ ## Remediation Suggestions 1. Pin exact, reviewed versions for direct and transitive dependencies. 2. Generate and enforce cryptographic hashes, such as with a hash-locked requirements file. 3. Install dependencies into a dedicated virtual environment owned by the Skill. 4. Enforce a trusted package index and disable untrusted extra indexes. 5. Review dependency updates before changing the lock file. 6. Do not perform package installation as a routine side effect of every tool invocation. 7. Store installation state inside the isolated environment rather than in a mutable project stamp alone. 8. Run package vulnerability and provenance checks in the release process. ]]>

T09 · Insecure Skill Coding Practices

Error
Location
tools/katbot_client.py:624
Finding
Live-Trading Safety and User-Consent Requirements Are Not Enforced by Execution Code<![CDATA[ ## Vulnerability Details **File Location**: `tools/katbot_onboard.py:259-264` and `tools/katbot_client.py:624-640` **Vulnerability Type**: Missing code-level authorization and unsafe financial defaults **Risk Level**: High ### Vulnerable Code Onboarding defaults to mainnet when the user presses Enter: ```python p_name = input(" Portfolio name (e.g. my-hl-mainnet): ").strip() or "my-hl-mainnet" testnet_input = input(" Use testnet? [y/N]: ").strip().lower() is_testnet = testnet_input in ("y", "yes") balance_input = input(" Initial balance in USD (e.g. 1000): ").strip() initial_balance = float(balance_input) if balance_input else 1000.0 ``` Trade execution performs no confirmation or portfolio safety validation: ```python def execute_recommendation(token: str, portfolio_id: int, rec_id: int, execute_onchain: bool = False, user_master_address: str = None) -> dict: """Execute an existing recommendation by ID.""" payload = {"recommendation_id": rec_id} if execute_onchain is not None: payload["execute_onchain"] = execute_onchain if AGENT_PRIVATE_KEY: payload["agent_private_key"] = AGENT_PRIVATE_KEY if user_master_address: payload["user_master_address"] = user_master_address r = requests.post(f"{BASE_URL}/portfolio/{portfolio_id}/execute", json=payload, headers=_auth(token)) r.raise_for_status() return r.json() ``` Position closure has the same issue: ```python def close_position(token: str, portfolio_id: int, symbol: str, user_master_address: str = None, reason: str = "API position closure", execute_onchain: bool = False) -> dict: payload = { "symbol": symbol, "reason": reason, "execute_onchain": execute_onchain, } if user_master_address: payload["user_master_address"] = user_master_address if AGENT_PRIVATE_KEY: ...[truncated 2180 chars]
Remediation
<![CDATA[ ## Remediation Suggestions 1. Change onboarding to default to testnet: ```python testnet_input = input(" Use testnet? [Y/n]: ").strip().lower() is_testnet = testnet_input not in ("n", "no") ``` 2. Require a typed, explicit acknowledgement before creating a mainnet portfolio. 3. Fetch and verify portfolio state inside the execution function. 4. Block mainnet execution unless `builder_fee_approved` is explicitly true. 5. Require a short-lived, single-use confirmation token bound to: - Portfolio ID. - Recommendation ID. - Symbol and direction. - Quantity or notional amount. - Leverage. - Entry, stop-loss, and take-profit parameters. 6. Enforce user-configured maximum notional value and leverage limits. 7. Separate read-only API functionality from trade-execution functionality. 8. Do not rely on Skill prose as the only authorization boundary. 9. Record an auditable confirmation event before any live execution. ]]>

T09 · Insecure Skill Coding Practices

Note
Location
tools/katbot_trigger_setup.py:222
Finding
Suggested Cron Job Logs Sensitive Trading Output to a Predictable Temporary File<![CDATA[ ## Vulnerability Details **File Location**: `tools/katbot_trigger_setup.py:222-229` **Vulnerability Type**: Unsafe temporary-file logging in a persistent scheduled workflow **Risk Level**: Low ### Vulnerable Code ```python tools_dir = _TOOLS_DIR base_dir = str(pathlib.Path(tools_dir).parent) cron_line = ( f"*/10 * * * * bash {tools_dir}/ensure_env.sh {base_dir} " f"&& PYTHONPATH={tools_dir} python3 {tools_dir}/katbot_signal_trigger.py" f" >> /tmp/katbot_trigger.log 2>&1" ) ``` The recommendation workflow prints the complete remote result: ```python print("\n=== Recommendation Result ===") print(json.dumps(rec_result, indent=2, default=str)) ``` ### Technical Analysis The setup wizard does not install a cron job automatically; it prints a line for the user to add manually. The scheduled monitoring mechanism is consistent with the declared 10-minute signal-monitoring function and is therefore not, by itself, an unauthorized persistence mechanism. However, the recommended command redirects all standard output and errors into the fixed path `/tmp/katbot_trigger.log`. Shell redirection follows symbolic links, and the resulting file mode depends on the cron process’s umask. The workflow can print signal details, token selections, recommendation results, portfolio identifiers, and operational errors. These records may therefore become visible to other local users or be redirected through a pre-existing symbolic link. ### Attack Path 1. A local attacker creates `/tmp/katbot_trigger.log` as a symbolic link to another file writable by the victim. 2. The user installs the suggested cron line. 3. Cron invokes the trigger and shell redirection follows the symbolic link. 4. Workflow output is appended to the target file. 5. Alternatively, if the file is created with permissive permissions, another local user reads its trading and portfolio contents. ### Impact Assessment Potential consequences include: - Disclosure of trading signals and rec ...[truncated 325 chars]
Remediation
<![CDATA[ ## Remediation Suggestions 1. Store logs under a private application directory owned by the user rather than `/tmp`. 2. Create the log directory with mode `0700` and log files with mode `0600`. 3. Reject symbolic links and verify file ownership before writing. 4. Use a logging handler that creates files securely instead of shell redirection. 5. Configure rotation and retention limits. 6. Redact portfolio details, recommendation bodies, tokens, and credential-bearing errors. 7. Quote all generated command paths with a robust shell-quoting function if a cron line continues to be printed. 8. Clearly state that scheduling is optional and provide removal instructions for the cron entry. ]]>
Vulnerability Patterns
  • Taint TrackingDirect Taint Flow, Variable-Mediated Taint Flow, Credential Exfiltration Chain
  • Prompt InjectionInstruction Override, Hidden Instructions, Exfiltration Commands
  • Data ExfiltrationExternal Transmission, Env Variable Harvesting, File System Enumeration
  • Privilege EscalationExcessive Permissions, Sudo/Root Execution, Credential Access
  • Supply ChainUnpinned Dependencies, External Script Fetching, Obfuscated Code
Findings (153)

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
if not refresh_token:
        return None
    try:
        r = requests.post(
            f"{BASE_URL}/refresh",
            json={"refresh_token": refresh_token},
            timeout=15,
Confidence
96% confidence
Finding
Refresh requests send the refresh token to BASE_URL, which may come from environment or a local env file. If an attacker controls that endpoint, they can steal a long-lived credential and mint fresh access tokens.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
address = account.address

    # Step 1: Get nonce
    r = requests.get(f"{BASE_URL}/get-nonce/{address}?chain_id={CHAIN_ID}")
    r.raise_for_status()
    message_text = r.json()["message"]
Confidence
97% confidence
Finding
BASE_URL is fully controllable via environment or local env file and is used for SIWE nonce retrieval before signing. If an attacker can set BASE_URL to a hostile server, the client will fetch an attacker-supplied message, sign it with WALLET_PRIVATE_KEY, and then transmit the resulting signature and later bearer tokens to that host, enabling credential theft or misuse.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
signature = signed.signature.hex()

    # Step 3: Login
    r = requests.post(f"{BASE_URL}/login", json={"address": address, "signature": signature, "chain_id": CHAIN_ID})
    r.raise_for_status()
    token_data = r.json()
Confidence
98% confidence
Finding
The login request sends the wallet address and freshly generated signature to a server selected by untrusted BASE_URL. In this skill context, that signature is sensitive because it was produced from the user's wallet key and can be harvested by a malicious endpoint during authentication.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
def list_portfolios(token: str) -> list:
    """List all portfolios for the authenticated user."""
    r = requests.get(f"{BASE_URL}/portfolio", headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
95% confidence
Finding
Authenticated API calls use BASE_URL from environment/local file without origin restrictions. A maliciously redirected endpoint would receive the bearer token in Authorization headers and potentially the agent private key header, exposing control over trading-related resources.

Tainted flow: 'payload' from os.getenv (line 505, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
if arbitrum_rpc_url is not None:
        payload["arbitrum_rpc_url"] = arbitrum_rpc_url

    r = requests.post(f"{BASE_URL}/portfolio", json=payload, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
96% confidence
Finding
create_portfolio may include agent_private_key and other sensitive configuration in a JSON payload sent to a BASE_URL that is externally controllable. In a live trading client, sending private keys to an attacker-controlled API would directly compromise trading authority.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
if window is not None:
        params["window"] = window

    r = requests.get(
        f"{BASE_URL}/portfolio/{portfolio_id}",
        params=params,
        headers=_auth(token, agent_key)
Confidence
96% confidence
Finding
get_portfolio sends Authorization and potentially X-Agent-Private-Key headers to a host chosen by BASE_URL. Because the code explicitly injects the Hyperliquid agent private key header for these calls, SSRF-like endpoint redirection becomes credential exfiltration with direct trading impact.

Tainted flow: 'payload' from os.getenv (line 505, credential/environment) → requests.put (network output)

Critical
Category
Data Flow
Content
if max_history_messages is not None:
        payload["max_history_messages"] = max_history_messages

    r = requests.put(f"{BASE_URL}/portfolio/{portfolio_id}",
                     json=payload, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
93% confidence
Finding
Portfolio update requests, including token selections and configuration, are sent with bearer authentication to an untrusted-configurable BASE_URL. This can leak account context and permit unauthorized state changes if the endpoint is attacker-controlled or if users are tricked into using a rogue server.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
Returns:
        List of symbol strings (e.g., ["BTC", "ETH", "SOL"])
    """
    r = requests.get(f"{BASE_URL}/portfolio/{portfolio_id}/tokens", headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
94% confidence
Finding
Listing portfolio tokens appears low risk functionally, but it still transmits bearer authentication to an untrusted endpoint if BASE_URL is manipulated. In this skill, the cumulative effect across many such calls is broad token exfiltration surface.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
Returns:
        Dict with portfolio_id, chain_id, is_testnet, network_name
    """
    r = requests.get(f"{BASE_URL}/portfolio/{portfolio_id}/chain-info", headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
94% confidence
Finding
Chain-info retrieval sends authenticated traffic to a configurable host. Even read-only endpoints are dangerous here because they leak access tokens and account metadata usable for later abuse.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
Dict with timeseries list, portfolio_id, granularity, window, limit
    """
    key = agent_private_key or AGENT_PRIVATE_KEY
    r = requests.get(
        f"{BASE_URL}/portfolio/{portfolio_id}/timeseries",
        params={"granularity": granularity, "limit": limit, "window": window},
        headers=_auth(token, key)
Confidence
97% confidence
Finding
Timeseries retrieval may send the agent private key via header to a BASE_URL-controlled destination. In a trading tool, exfiltration of that key is highly dangerous because it can enable unauthorized account actions beyond simple read access.

Tainted flow: 'payload' from os.getenv (line 505, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
Dict with status, result, and portfolio_id
    """
    payload = {"action": action, "signature": signature, "nonce": nonce}
    r = requests.post(f"{BASE_URL}/portfolio/{portfolio_id}/approve-builder-fee",
                      json=payload, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
90% confidence
Finding
approve_builder_fee forwards signed trading-related approval data to BASE_URL without host trust enforcement. While the signature is expected to be transmitted, an attacker-controlled endpoint can collect approval artifacts and authenticated context for replay or abuse depending on backend protections.

Tainted flow: 'payload' from os.getenv (line 505, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
"""
    key = agent_private_key or AGENT_PRIVATE_KEY
    payload = {"agent_private_key": key, "is_testnet": is_testnet}
    r = requests.post(f"{BASE_URL}/portfolio/validate-hyperliquid",
                      json=payload, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
99% confidence
Finding
validate_hyperliquid directly places agent_private_key in the request JSON body and sends it to BASE_URL. Because BASE_URL is externally configurable, this is a straightforward secret exfiltration path with immediate compromise of live trading credentials.

Tainted flow: 'payload' from os.getenv (line 505, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
payload["agent_id"] = agent_id
    if AGENT_PRIVATE_KEY:
        payload["agent_private_key"] = AGENT_PRIVATE_KEY
    r = requests.post(f"{BASE_URL}/agent/recommendation/message", json=payload, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
97% confidence
Finding
request_recommendation includes AGENT_PRIVATE_KEY in the payload when available and posts to a configurable BASE_URL. This combines arbitrary endpoint control with direct private key exfiltration from a live trading skill.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
"""
    deadline = time.time() + max_wait
    while time.time() < deadline:
        r = requests.get(f"{BASE_URL}/agent/recommendation/poll/{ticket_id}", headers=_auth(token))
        r.raise_for_status()
        data = r.json()
        if data.get("done") or data.get("status") in ("COMPLETED", "complete", "FAILED"):
Confidence
93% confidence
Finding
Recommendation polling repeatedly sends bearer auth to the configurable backend. Besides token leakage, a rogue server can feed manipulated recommendation status and content that may influence automated execution.

Tainted flow: 'payload' from os.getenv (line 505, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
if key:
        payload["agent_private_key"] = key

    r = requests.post(f"{BASE_URL}/agent/recommendation/response",
                      json=payload, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
97% confidence
Finding
submit_recommendation_response may transmit agent_private_key to an endpoint selected by BASE_URL. In this context, recommendation analysis is adjacent to execution, so theft of the key or bearer token can lead to unauthorized portfolio operations.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
"""
    deadline = time.time() + max_wait
    while time.time() < deadline:
        r = requests.get(f"{BASE_URL}/agent/recommendation/response/poll/{ticket_id}",
                         headers=_auth(token))
        r.raise_for_status()
        data = r.json()
Confidence
93% confidence
Finding
Recommendation-response polling trusts and queries a configurable backend with bearer auth. This exposes credentials and allows spoofed analysis to shape trading decisions.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
def get_recommendations(token: str, portfolio_id: int) -> list:
    """Get existing recommendations for a portfolio."""
    r = requests.get(f"{BASE_URL}/portfolio/{portfolio_id}/recommendation", headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
94% confidence
Finding
Fetching recommendations sends an Authorization bearer token to a potentially attacker-controlled BASE_URL. This is a real exposure because the client broadly trusts endpoint configuration while operating on trading accounts.

Tainted flow: 'payload' from os.getenv (line 505, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
payload["agent_private_key"] = AGENT_PRIVATE_KEY
    if user_master_address:
        payload["user_master_address"] = user_master_address
    r = requests.post(f"{BASE_URL}/portfolio/{portfolio_id}/execute",
                      json=payload, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
98% confidence
Finding
execute_recommendation can transmit agent_private_key and wallet address to a configurable endpoint while initiating trade execution. This is especially dangerous in a live trading skill because compromise yields both secret material and execution context.

Tainted flow: 'payload' from os.getenv (line 505, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
payload["user_master_address"] = user_master_address
    if AGENT_PRIVATE_KEY:
        payload["agent_private_key"] = AGENT_PRIVATE_KEY
    r = requests.post(f"{BASE_URL}/portfolio/{portfolio_id}/close-position",
                      json=payload, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
98% confidence
Finding
close_position posts trade-closing instructions and may include agent_private_key to a host determined by BASE_URL. A malicious endpoint could steal credentials and manipulate or spoof destructive trading operations.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
key = agent_private_key or AGENT_PRIVATE_KEY
    if key:
        params["agent_private_key"] = key
    r = requests.get(f"{BASE_URL}/portfolio/{portfolio_id}/trade",
                     params=params, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
98% confidence
Finding
list_trades adds agent_private_key into URL query parameters when present, and sends the request to configurable BASE_URL. Query-string transmission is especially bad because secrets may be logged by proxies, history, and server logs in addition to endpoint compromise.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
if user_master_address:
        params["user_master_address"] = user_master_address
    key = agent_private_key or AGENT_PRIVATE_KEY
    r = requests.get(f"{BASE_URL}/portfolio/{portfolio_id}/events",
                     params=params, headers=_auth(token, key))
    r.raise_for_status()
    return r.json()
Confidence
96% confidence
Finding
get_position_events sends Authorization and potentially X-Agent-Private-Key to a host chosen through BASE_URL. This leaks both account access and trading authority to any attacker who can influence configuration.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
Returns:
        List of AgentInfo dicts (id, name, max_history_messages, avatar_url, etc.)
    """
    r = requests.get(f"{BASE_URL}/agents", headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
94% confidence
Finding
Agent listing transmits bearer authentication to a configurable endpoint. Even though the operation is administrative, the core flaw remains token leakage to untrusted infrastructure.

Tainted flow: 'payload' from os.getenv (line 505, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
AgentInfo dict with id, name, avatar_url, etc.
    """
    payload = {"name": name, "max_history_messages": max_history_messages}
    r = requests.post(f"{BASE_URL}/agents", json=payload, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
93% confidence
Finding
Creating agents sends authenticated administrative data to BASE_URL without trust restrictions. In this skill, the broad administrative surface increases the value of stolen tokens and the blast radius of a rogue endpoint.

Tainted flow: 'BASE_URL' from os.getenv (line 83, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
Returns:
        AgentInfo dict
    """
    r = requests.get(f"{BASE_URL}/agents/{agent_id}", headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
94% confidence
Finding
get_agent exposes authenticated account metadata to a configurable host. This is part of a repeated pattern of secret-bearing requests to untrusted destinations.

Tainted flow: 'payload' from os.getenv (line 505, credential/environment) → requests.put (network output)

Critical
Category
Data Flow
Content
payload["max_history_messages"] = max_history_messages
    if avatar_seed is not None:
        payload["avatar_seed"] = avatar_seed
    r = requests.put(f"{BASE_URL}/agents/{agent_id}", json=payload, headers=_auth(token))
    r.raise_for_status()
    return r.json()
Confidence
93% confidence
Finding
Updating agents sends authenticated state-changing traffic to a BASE_URL that may be attacker-controlled. This can leak tokens and enable unauthorized administrative actions through deceptive endpoints.

Static analysis

No suspicious patterns detected.