#!/usr/bin/env python3 """LOSURIA command-line entry point. The CLI deliberately has two trust boundaries: * ``api read`` is a bounded, read-only API client. It never accepts a URL, private key, or arbitrary method from the command line and redacts sensitive response fields before printing. * ``api doctor`` probes the same read-only allowlist but prints only HTTP status codes, making credential/origin problems diagnosable without exposing response data. * ``api build`` calls only fixed, non-mutating Solana builders from a local JSON request file. It never accepts a relay route, a signature, or a broadcast payload and never retries the POST. * ``contract inspect`` validates an owner-selected EVM call and prints the exact chain-bound relay envelope shape. It does not sign or broadcast anything. * ``bot`` delegates to the existing generated bot, which keeps owner signing, fee checks, nonce handling, and relay policy in one implementation. No private key is stored by this program. The existing bot reads its signing material only from the explicit environment variable required by the selected command. """ from __future__ import annotations import argparse import getpass import hmac import json import os import re import runpy import stat import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any CLI_VERSION = "0.6.1" DEFAULT_API_BASE = "https://app.losuria.com" API_KEY_RE = re.compile(r"lsk_[0-9a-fA-F]{64}") HEX_ADDRESS = re.compile(r"^0x[0-9a-fA-F]{40}$") HEX_DATA = re.compile(r"^0x[0-9a-fA-F]+$") EVM_CHAINS = {"ethereum": 1, "base": 8453, "monad": 143} API_BUILD_ROUTES = { "solana-swap": "/api/v1/solana/build", "solana-launch": "/api/v1/solana/launch/build", "solana-launch-pool": "/api/v1/solana/launch/pool-build", "solana-conditional": "/api/v1/solana/conditional/build", "solana-conditional-cancel": "/api/v1/solana/conditional/cancel/build", } READ_ROUTES = frozenset( { "/api/v1/status", "/api/v1/orders", "/api/v1/limits", "/api/v1/dca", "/api/v1/copy", "/api/v1/fills", "/api/v1/grids", "/api/v1/intel/usage", "/api/v1/intel/entitlement", } ) BUILD_REQUEST_FORBIDDEN_FIELDS = frozenset( { "signature", "owner_signature", "mint_signature", "signed_transaction", "raw_transaction", "wire", } ) BUILD_REQUEST_SECRET_FIELDS = BUILD_REQUEST_FORBIDDEN_FIELDS | frozenset( {"api_key", "authorization", "bearer", "private_key", "seed", "secret"} ) OWNER_INJECTED_FIELDS = frozenset({"owner_eoa", "creator"}) MAX_BUILD_RESPONSE_BYTES = 1_048_576 SENSITIVE_LABELS = ( "api_key", "authorization", "bearer", "private", "secret", "seed", "signature", "token", ) EMBEDDED_SECRET_PATTERNS = ( re.compile(r"(?i)\blsk_[0-9a-f]{64}\b"), re.compile(r"(?i)\bbearer\s+[a-z0-9._~+/=-]+"), ) def fail(message: str) -> "NoReturn": raise SystemExit(f"losuria: {message}") def validate_target(raw: str) -> str: if not HEX_ADDRESS.fullmatch(raw): fail("target must be a 20-byte 0x-prefixed EVM address") return raw def validate_calldata(raw: str) -> tuple[str, str]: if not HEX_DATA.fullmatch(raw): fail("data must be 0x-prefixed hexadecimal calldata") payload = raw[2:] if len(payload) < 8 or len(payload) % 2: fail("data must contain a four-byte selector and complete bytes") return raw, "0x" + payload[:8] def call_risk(selector: str) -> dict[str, Any]: """Describe the risk boundary without pretending to decode unknown ABI.""" known = { "0x095ea7b3": "ERC-20 approve(address,uint256)", "0xa22cb465": "ERC-721/1155 setApprovalForAll(address,bool)", "0x23b872dd": "ERC-20 transferFrom(address,address,uint256)", "0xd505accf": "ERC-2612 permit(address,address,uint256,uint256,uint8,bytes32,bytes32)", } label = known.get(selector.lower()) warnings = [ "ABI is not available in this offline command; treat calldata as unknown until locally decoded", "owner must inspect target, value, gas and the exact calldata before signing", "relay must simulate and bind the target code identity before submission", ] if label and ("approve" in label.lower() or "permit" in label.lower()): warnings.insert(0, "approval-like call can grant token authority; verify spender and allowance scope") return { "abi": "known-selector" if label else "unknown", "selector_name": label, "warnings": warnings, "phishing_policy": "fail-closed on target/code identity or simulation mismatch", } def validate_value(raw: str) -> int: try: value = int(raw, 0) except ValueError: fail("value-wei must be a non-negative integer") if value < 0: fail("value-wei must be a non-negative integer") return value def validate_gas(raw: str) -> int: try: gas = int(raw, 0) except ValueError: fail("gas must be an integer from 21000 to 8000000") if not 21_000 <= gas <= 8_000_000: fail("gas must be an integer from 21000 to 8000000") return gas def validate_chain(raw: str) -> tuple[str, int]: chain = raw.strip().lower() chain_id = EVM_CHAINS.get(chain) if chain_id is None: fail("chain must be one of: ethereum, base, monad") return chain, chain_id def validate_positive_wei(raw: str, label: str) -> int: value = validate_value(raw) if value <= 0: fail(f"{label} must be a positive integer") return value def validate_ttl(raw: str) -> int: try: ttl = int(raw, 10) except ValueError: fail("ttl-seconds must be an integer from 30 to 300") if not 30 <= ttl <= 300: fail("ttl-seconds must be an integer from 30 to 300") return ttl def _local_config_dir() -> Path: configured = os.environ.get("XDG_CONFIG_HOME", "").strip() base = Path(configured).expanduser() if configured else Path.home() / ".config" if not base.is_absolute(): fail("XDG_CONFIG_HOME must be an absolute path") return base / "losuria" def _api_key_path() -> Path: return _local_config_dir() / "api-key" def _owned_by_current_user(path: Path, owner_label: str) -> None: getuid = getattr(os, "getuid", None) if not callable(getuid): return try: owner = path.lstat().st_uid except OSError as exc: fail(f"cannot inspect the local API credential {owner_label}: {exc}") if owner != getuid(): fail(f"local API credential {owner_label} is not owned by the current user") def _ensure_private_config_dir() -> Path: directory = _local_config_dir() try: directory.mkdir(parents=True, mode=0o700, exist_ok=True) metadata = directory.lstat() except OSError as exc: fail(f"cannot create the local API credential store: {exc}") if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): fail("the local API credential store is not a directory") _owned_by_current_user(directory, "directory") if metadata.st_mode & 0o077: try: directory.chmod(0o700) metadata = directory.lstat() except OSError as exc: fail(f"local API credential store permissions are too broad: {exc}") if metadata.st_mode & 0o077: fail("local API credential store must be private to the current user") return directory def _read_stored_api_key() -> str | None: path = _api_key_path() try: metadata = path.lstat() except FileNotFoundError: return None except OSError as exc: fail(f"cannot inspect the local API credential: {exc}") if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): fail("the local API credential must be a regular private file") _owned_by_current_user(path, "file") if metadata.st_mode & 0o077: fail("local API credential permissions are too broad; run `losuria api configure`") try: value = path.read_text(encoding="utf-8").strip() except OSError as exc: fail(f"cannot read the local API credential: {exc}") if not API_KEY_RE.fullmatch(value): fail("the local API credential is malformed; run `losuria api configure`") return value def _write_stored_api_key(key: str) -> Path: directory = _ensure_private_config_dir() path = directory / "api-key" temporary = directory / f".api-key.{os.getpid()}.tmp" flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW descriptor = None try: descriptor = os.open(temporary, flags, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream: descriptor = None stream.write(key + "\n") stream.flush() os.fsync(stream.fileno()) os.replace(temporary, path) path.chmod(0o600) except OSError as exc: if descriptor is not None: os.close(descriptor) try: temporary.unlink() except FileNotFoundError: pass except OSError: pass fail(f"could not save the local API credential: {exc}") return path def _api_key_from_sources() -> tuple[str | None, str]: raw = os.environ.get("LOSURIA_API_KEY", "").strip() if raw: if not API_KEY_RE.fullmatch(raw): fail("LOSURIA_API_KEY is malformed; use `losuria api configure` or a complete lsk_ credential") return raw, "process-environment" stored = _read_stored_api_key() if stored is None: return None, "not-configured" return stored, "local-protected-file" def _load_api_key() -> tuple[str, str]: key, source = _api_key_from_sources() if key is None: fail( "no API credential configured; run `losuria api configure` " "and enter it at the hidden prompt" ) return key, source def _api_base() -> str: base = os.environ.get("LOSURIA_API_BASE", DEFAULT_API_BASE).rstrip("/") if not re.fullmatch(r"https://[^/]+", base): fail("LOSURIA_API_BASE must be a single HTTPS origin") return base def api_credentials() -> tuple[str, str]: key, _source = _load_api_key() return _api_base(), key def disallowed_request_fields( value: Any, field_names: frozenset[str], path: str = "", ) -> list[str]: found: list[str] = [] if isinstance(value, dict): for key, item in value.items(): key_text = str(key) key_path = f"{path}.{key_text}" if path else key_text if key_text.lower() in field_names: found.append(key_path) found.extend(disallowed_request_fields(item, field_names, key_path)) elif isinstance(value, list): for index, item in enumerate(value): found.extend(disallowed_request_fields(item, field_names, f"{path}[{index}]")) return found def load_json_object(path_value: str) -> dict[str, Any]: path = Path(path_value).expanduser() try: with path.open("r", encoding="utf-8") as stream: payload = json.load(stream) except (OSError, json.JSONDecodeError) as exc: fail(f"request file is not valid JSON: {exc}") if not isinstance(payload, dict): fail("request file must contain one JSON object") forbidden = sorted( disallowed_request_fields(payload, BUILD_REQUEST_SECRET_FIELDS) ) if forbidden: fail( "build requests must not contain secret, signing, or transaction fields: " + ", ".join(forbidden) ) injected = sorted(disallowed_request_fields(payload, OWNER_INJECTED_FIELDS)) if injected: fail( "owner identity is injected from the bound API key; remove: " + ", ".join(injected) ) return payload def read_bounded_json(response: Any) -> Any: body = response.read(MAX_BUILD_RESPONSE_BYTES + 1) if len(body) > MAX_BUILD_RESPONSE_BYTES: fail("API builder response exceeded the 1 MiB safety limit") try: return json.loads(body) except json.JSONDecodeError as exc: fail(f"API builder returned invalid JSON: {exc}") def build_api(args: argparse.Namespace) -> int: """Call one fixed, non-mutating API builder and print only its result. Builders create short-lived unsigned messages/manifests. They never relay, sign, or broadcast. A POST is intentionally attempted once: retrying a builder could create a second short-lived capability and would teach users the wrong retry contract for the subsequent mutation lane. """ route = API_BUILD_ROUTES[args.builder] request_payload = load_json_object(args.request_file) base, key = api_credentials() request = urllib.request.Request( base + route, data=json.dumps(request_payload, separators=(",", ":")).encode("utf-8"), headers={ "Authorization": f"Bearer {key}", "Accept": "application/json", "Content-Type": "application/json", }, method="POST", ) try: with urllib.request.urlopen(request, timeout=30) as response: status = int(response.status) body = read_bounded_json(response) except urllib.error.HTTPError as exc: status = int(exc.code) body = read_bounded_json(exc) except (urllib.error.URLError, TimeoutError) as exc: fail(f"API builder request failed: {exc}") print( json.dumps( {"builder": args.builder, "route": route, "status": status, "response": redact(body)}, indent=2, sort_keys=True, ) ) return 0 if 200 <= status < 300 else 1 def contract_manifest(args: argparse.Namespace) -> int: chain, chain_id = validate_chain(args.chain) target = validate_target(args.target) data, selector = validate_calldata(args.data) value = validate_value(args.value_wei) gas = validate_gas(args.gas) risk = call_risk(selector) document = { "schema": "losuria.contract-manifest/v1", "chain": chain, "chain_id": chain_id, "target": target, "calldata": data, "selector": selector, "calldata_bytes": (len(data) - 2) // 2, "value_wei": str(value), "gas": gas, "execution": { "mode": "owner-signed-relay", "route": "POST /api/v1/relay", "owner_signature_required": True, "chain_identity_required": True, "broadcast": False, "fee_gate": "LOSURIA server and immutable execution contracts", }, "preview": { "status": "unknown_abi_offline", "target": target, "selector": selector, "calldata": "not decoded offline", "target_codehash": { "status": "required_at_sign_and_relay", "observed": None, "on_change": "fail_closed", }, "simulation": { "status": "required_before_owner_signature", "performed": False, "on_failure": "do not sign or relay", }, }, "limits": { "value_wei": str(value), "gas_limit": gas, "worst_case_fee": "unknown_until_fresh_chain_quote", "unknown_components": "gas_price, network_fee, protocol_fee and finality risk require a fresh route quote", }, "risk": risk, "receipt": { "status": "not_broadcast", "expected_states": ["submitted", "included", "finalized", "reverted", "lost"], "fee": "enforced_at_relay_then_verified_from_receipt", }, } print(json.dumps(document, indent=2, sort_keys=True)) return 0 def bot_plan(args: argparse.Namespace) -> int: """Create a bounded, offline draft for one owner-signed bot order.""" chain, chain_id = validate_chain(args.chain) token = validate_target(args.token) amount = validate_positive_wei(args.amount_wei, "amount-wei") min_out = validate_positive_wei(args.min_out, "min-out") gas = validate_gas(args.gas) ttl = validate_ttl(args.ttl_seconds) issued_at_ms = int(time.time() * 1000) document = { "schema": "losuria.bot-plan/v1", "chain": chain, "chain_id": chain_id, "token": token, "amount_wei": str(amount), "min_out": str(min_out), "gas_limit": gas, "issued_at_ms": issued_at_ms, "expires_at_ms": issued_at_ms + ttl * 1000, "execution": { "mode": "owner-signed-relay", "route": "POST /api/v1/relay", "owner_signature_required": True, "broadcast": False, "single_order_scope": True, "fee_gate": "LOSURIA relay and immutable execution contract", "value_ceiling_wei": str(amount), "gas_ceiling": gas, }, "next_local_step": "pass this bounded draft to the local signing flow", } print(json.dumps(document, indent=2, sort_keys=True)) return 0 def is_sensitive(label: str) -> bool: lowered = label.lower() return any(part in lowered for part in SENSITIVE_LABELS) def redact(value: Any, label: str = "") -> Any: if is_sensitive(label): return "[REDACTED]" if isinstance(value, dict): return {str(key): redact(item, str(key)) for key, item in value.items()} if isinstance(value, list): return [redact(item, label) for item in value] if isinstance(value, str): for pattern in EMBEDDED_SECRET_PATTERNS: value = pattern.sub("[REDACTED]", value) return value def read_api(args: argparse.Namespace) -> int: route = args.route if route not in READ_ROUTES or ".." in route or not route.startswith("/api/v1/"): fail("route is not in the bounded read-only API allowlist") if "?" in route or "#" in route: fail("route must not contain query parameters or fragments") base, key = api_credentials() url = base + route request = urllib.request.Request( url, headers={"Authorization": f"Bearer {key}", "Accept": "application/json"}, method="GET", ) try: with urllib.request.urlopen(request, timeout=15) as response: payload = json.load(response) except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: fail(f"read-only API request failed: {exc}") print(json.dumps(redact(payload), indent=2, sort_keys=True)) return 0 def doctor_api(args: argparse.Namespace) -> int: """Probe every allowlisted GET route without reading or printing bodies.""" allow_prompt = not getattr(args, "no_prompt", False) key, key_source = _api_key_from_sources() if key is None: if allow_prompt and sys.stdin.isatty() and sys.stderr.isatty(): print("No LOSURIA API credential is configured. API access is opt-in.") print("Configure it now; both entries below are hidden and stored only in your user config.") _configure_api_interactively() key, key_source = _api_key_from_sources() if key is None: fail("API credential setup did not produce a credential; run `losuria api configure`") else: print( json.dumps( { "schema": "losuria.api-doctor/v1", "configured": False, "status": "not_configured", "next_command": "losuria api configure", "message": "API access is opt-in; no network request was made.", "routes": {}, }, indent=2, sort_keys=True, ) ) return 2 base = _api_base() statuses: dict[str, int | str] = {} for route in sorted(READ_ROUTES): request = urllib.request.Request( base + route, headers={"Authorization": f"Bearer {key}", "Accept": "application/json"}, method="GET", ) try: with urllib.request.urlopen(request, timeout=15) as response: statuses[route] = int(response.status) except urllib.error.HTTPError as exc: # The status is the diagnostic; the error body may contain user or # implementation data and is deliberately never consumed. statuses[route] = int(exc.code) except (urllib.error.URLError, TimeoutError) as exc: statuses[route] = type(exc).__name__ print( json.dumps( { "schema": "losuria.api-doctor/v1", "configured": True, "source": key_source, "base": base, "routes": statuses, }, indent=2, sort_keys=True, ) ) return 0 if all(isinstance(status, int) and 200 <= status < 300 for status in statuses.values()) else 1 def _local_state_dir() -> Path: configured = os.environ.get("XDG_STATE_HOME", "").strip() base = Path(configured).expanduser() if configured else Path.home() / ".local" / "state" return base / "losuria" def _configure_api_interactively() -> Path: """Collect and persist one API credential without exposing it to the shell.""" if not sys.stdin.isatty() and not sys.stderr.isatty(): fail("API credential setup requires an interactive terminal; never pass the key as an argument") try: first = getpass.getpass("LOSURIA API key (input hidden): ").strip() if not API_KEY_RE.fullmatch(first): fail("API credential format is invalid; expected lsk_ followed by 64 hexadecimal characters") second = getpass.getpass("Repeat API key (input hidden): ").strip() except (EOFError, KeyboardInterrupt): fail("API credential setup cancelled; nothing was changed") if not hmac.compare_digest(first, second): fail("API credentials did not match; nothing was changed") return _write_stored_api_key(first) def configure_api(args: argparse.Namespace) -> int: """Opt in to local API-key configuration through a hidden terminal prompt.""" del args _configure_api_interactively() print("LOSURIA API credential saved in protected local configuration (mode 0600).") print("Run `losuria api doctor` to verify the connection.") return 0 def show_config(args: argparse.Namespace) -> int: """Show safe local configuration without reading or printing credentials.""" del args state_dir = _local_state_dir() base = os.environ.get("LOSURIA_API_BASE", DEFAULT_API_BASE).rstrip("/") environment_key = os.environ.get("LOSURIA_API_KEY", "").strip() if environment_key: key_configured = bool(API_KEY_RE.fullmatch(environment_key)) key_source = "process-environment" if key_configured else "invalid-process-environment" else: stored_key = _read_stored_api_key() key_configured = stored_key is not None key_source = "local-protected-file" if key_configured else "not-configured" installed_version = None version_file = state_dir / "installed-version" try: installed_version = version_file.read_text(encoding="utf-8").strip() or None except OSError: pass print( json.dumps( { "schema": "losuria.cli-config/v1", "cli_version": CLI_VERSION, "api_origin": base if re.fullmatch(r"https://[^/]+", base) else "[INVALID]", "api_key": { "configured": key_configured, "source": key_source, }, "state_dir": str(state_dir), "installed_version": installed_version, "secrets_persisted_by_cli": key_source == "local-protected-file", }, indent=2, sort_keys=True, ) ) return 0 def delegate_bot(args: argparse.Namespace) -> int: # A source checkout keeps the generated bot under frontend/generated. A # verified release keeps the exact same artifact beside the CLI. Trying # both paths lets one command work in both forms without a second bot # implementation or an unverified download at runtime. candidates = ( Path(__file__).resolve().with_name("losuria_bot.py"), Path(__file__).resolve().parents[1] / "frontend" / "generated" / "losuria_bot.py", ) bot = next((candidate for candidate in candidates if candidate.is_file()), None) if bot is None: fail("verified bot entry point is missing") # The bundled bot retains its own signing and relay implementation. When # the user explicitly configured the CLI, make that credential available # to the delegated process without putting it in argv or shell history. if not os.environ.get("LOSURIA_API_KEY", "").strip(): stored = _read_stored_api_key() if stored: os.environ["LOSURIA_API_KEY"] = stored sys.argv = [str(bot), *args.bot_args] runpy.run_path(str(bot), run_name="__main__") return 0 def parser() -> argparse.ArgumentParser: root = argparse.ArgumentParser( prog="losuria", description=( "Owner-controlled LOSURIA API, relay and contract tooling.\n\n" "The CLI starts without an API credential. Configure one only when\n" "you opt in to API access: losuria api configure" ), ) root.add_argument("--version", action="version", version=CLI_VERSION) commands = root.add_subparsers(dest="command", required=True) api = commands.add_parser("api", help="bounded API operations") api_commands = api.add_subparsers(dest="api_command", required=True) read = api_commands.add_parser("read", help="read one allowlisted API route") read.add_argument("route", help="for example /api/v1/status") read.set_defaults(handler=read_api) doctor = api_commands.add_parser( "doctor", help="check the allowlisted API routes; offer secure setup when no key exists", ) doctor.add_argument( "--no-prompt", action="store_true", help="report missing setup without opening the hidden interactive prompt", ) doctor.set_defaults(handler=doctor_api) api_commands.add_parser( "configure", aliases=("setup",), help="securely configure the API credential through a hidden terminal prompt", ).set_defaults(handler=configure_api) build = api_commands.add_parser( "build", help="call one fixed unsigned builder; never signs, relays, or broadcasts", ) build.add_argument("builder", choices=sorted(API_BUILD_ROUTES)) build.add_argument( "--request-file", required=True, help="local JSON object containing only the selected builder's public inputs", ) build.set_defaults(handler=build_api) contract = commands.add_parser("contract", help="local contract-call tooling") contract_commands = contract.add_subparsers(dest="contract_command", required=True) for name, help_text in ( ("inspect", "validate a chain-bound, relay-ready call manifest"), ("build", "build the same chain-bound offline call manifest"), ): contract_parser = contract_commands.add_parser(name, help=help_text) contract_parser.add_argument("--chain", required=True, choices=sorted(EVM_CHAINS)) contract_parser.add_argument("target") contract_parser.add_argument("data") contract_parser.add_argument("--value-wei", default="0") contract_parser.add_argument("--gas", default="200000") contract_parser.set_defaults(handler=contract_manifest) bot_plan_parser = commands.add_parser( "bot-plan", help="create an offline bounded draft for one owner-signed bot order", ) bot_plan_parser.add_argument("--chain", required=True, choices=sorted(EVM_CHAINS)) bot_plan_parser.add_argument("--token", required=True) bot_plan_parser.add_argument("--amount-wei", required=True) bot_plan_parser.add_argument("--min-out", required=True) bot_plan_parser.add_argument("--gas", default="200000") bot_plan_parser.add_argument("--ttl-seconds", default="120") bot_plan_parser.set_defaults(handler=bot_plan) config = commands.add_parser( "config", help="show safe local configuration; secrets are never printed", ) config.set_defaults(handler=show_config) config_commands = config.add_subparsers(dest="config_command") config_commands.add_parser("show", help="show safe local configuration").set_defaults( handler=show_config ) bot = commands.add_parser("bot", help="run the owner-signing LOSURIA bot") bot.add_argument("bot_args", nargs=argparse.REMAINDER) bot.set_defaults(handler=delegate_bot) return root def main(argv: list[str] | None = None) -> int: cli_parser = parser() # A bare `losuria` is the supported entry point after installation. It # opens the command surface without inventing an interactive signer or a # second execution path; explicit owner-signing remains required. if argv is not None and len(argv) == 0: cli_parser.print_help() return 0 if argv is None and len(sys.argv) == 1: cli_parser.print_help() return 0 args = cli_parser.parse_args(argv) return int(args.handler(args)) if __name__ == "__main__": raise SystemExit(main())