scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-10-14 04:08:06 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Pay top-up challenges with the Stripe Link wallet client

Browse source
This commit is contained in:
Alex Schapiro 2026-09-01 19:03:36 +00:00
parent a071022182
commit 517de86806
4 changed files with 245 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
skills/managed-pentesting-with-strix/SKILL.md
View file

@ -96,7 +96,7 @@ Non-Enterprise scans consume org credits. Enterprise engagements are plan-includ
strix cloud credits
```
When the balance is too low, buy credits with `strix cloud billing topup` (`billing:write`, admin token). The server answers the first request with **HTTP 402 and a machine-payment challenge** (Stripe Machine Payments Protocol). The CLI pays the challenge with the `mppx` client when Node.js is available — the user approves the spend in their agent wallet, for example the [Link Agent Wallet](https://link.com/agents). The response returns the receipt (`credits_granted`, `duplicate`, `reference`) and the new balance.
When the balance is too low, buy credits with `strix cloud billing topup` (`billing:write`, admin token). The server answers the first request with **HTTP 402 and a machine-payment challenge** (Stripe Machine Payments Protocol). The CLI pays the challenge with the Stripe Link wallet client when Node.js is available — the user approves the spend in the [Link app](https://link.com/agents). The response returns the receipt (`credits_granted`, `duplicate`, `reference`) and the new balance.
A default-tier source-only code review currently starts at 60 credits. Source uploads are not free: they launch an ordinary `code_review` and use the same deterministic scope estimator. The service checks the full balance before launch, reserves credits atomically only after validation succeeds, and does not create or charge a rejected scan. Retests and Enterprise scans are exempt.
@ -105,7 +105,7 @@ strix cloud billing topup --credits 20 --yes # explicit approval; skips the TT
strix cloud billing topup --credits 20 --no-pay # print the 402 challenge without paying
```
The default payment path is the Stripe agent wallet. Tell the user to set it up one time at [link.com/agents](https://link.com/agents). After setup, the user approves each payment in the Link app, and no keys or variables are necessary.
The default payment path is the Stripe Link wallet. When no wallet is connected, an interactive `strix cloud billing topup` starts the Link sign-in for the user and prints the verification link. The user approves the connection one time in the Link app, and then approves each payment there. No keys or variables are necessary. In a non-interactive process, the command stops and tells the user to connect the wallet at [link.com/agents](https://link.com/agents) or to use the hosted checkout link.
In a non-interactive agent or CI process, payment never proceeds unless the command includes `--yes`. Show the challenge or estimated spend to the user and obtain approval before adding it. `--no-pay` always stops after printing the challenge.

195
strix/interface/cloud/billing.py
View file

@ -30,6 +30,12 @@ _MAX_WALLET_DETAIL_CHARS = 2_000
# platform. This version is also old enough to remain installable in npm
# environments that apply a short package-publication safety window.
_MPPX_PACKAGE = "mppx@0.8.17"
# Stripe's own wallet client. It runs the complete challenge flow: it creates a
# spend request, waits for the person to approve it in the Link app, and retries
# the payment with the approved credential.
_LINK_CLI_PACKAGE = "@stripe/link-cli@0.13.1"
_LINK_CLI_CLIENT_NAME = "Strix CLI"
_LINK_LOGIN_TIMEOUT_S = 300
_NPM_REGISTRY = "https://registry.npmjs.org"
_WALLET_ENV_NAMES = frozenset(
{
@ -58,6 +64,8 @@ _WALLET_ENV_NAMES = frozenset(
"TMPDIR",
"USERPROFILE",
"XDG_CONFIG_HOME",
"XDG_DATA_HOME",
"XDG_STATE_HOME",
"all_proxy",
"http_proxy",
"https_proxy",
@ -74,7 +82,7 @@ class _WalletClientResult:
upstream_responses: tuple[WalletUpstreamResponse, ...]
def run_topup( # noqa: PLR0911, PLR0912
def run_topup( # noqa: PLR0911, PLR0912, PLR0915
console: Console,
args: argparse.Namespace,
body: dict[str, Any],
@ -137,17 +145,16 @@ def run_topup( # noqa: PLR0911, PLR0912
payment_method = getattr(args, "payment_method", None) or os.environ.get(
"MPPX_STRIPE_PAYMENT_METHOD"
)
if not payment_method and (
not as_json
and not os.environ.get("MPPX_ACCOUNT")
and not os.environ.get("MPPX_STRIPE_SECRET_KEY")
):
console.print(
"[dim]Tip: payments need a wallet. Set up a Stripe agent wallet at "
"https://link.com/agents, and the user approves each payment in the Link app. "
"If the user does not want a wallet, run "
"`strix cloud billing subscribe --plan strix_top_up` for a hosted checkout link.[/]"
)
use_link_wallet = payment_method is None and not _mppx_wallet_configured()
if use_link_wallet:
setup_error = _prepare_link_wallet(console, npx, as_json=as_json)
if setup_error is not None:
emit(
console,
{"error": setup_error, "challenge": challenge},
as_json=as_json,
)
return http.EXIT_PAYMENT
try:
wallet_result = _run_wallet_client(
npx,
@ -155,6 +162,7 @@ def run_topup( # noqa: PLR0911, PLR0912
body,
token=token,
payment_method=payment_method,
use_link_wallet=use_link_wallet,
capture_output=as_json,
)
except KeyboardInterrupt:
@ -268,9 +276,10 @@ def _run_wallet_client(
*,
token: str | None,
payment_method: str | None,
use_link_wallet: bool,
capture_output: bool,
) -> _WalletClientResult:
"""Run mppx through the loopback bridge without exposing the API token to it."""
"""Run the wallet through the loopback bridge without exposing the API token."""
upstream_url = f"{http.app_url()}/api/v1/billing/topup"
body_json = json.dumps(body)
wallet_env = _wallet_environment()
@ -281,6 +290,7 @@ def _run_wallet_client(
global_config = wallet_root / "global.npmrc"
user_config.touch(mode=0o600)
global_config.touch(mode=0o600)
npx_prefix = _npx_prefix(npx, wallet_root)
with wallet_payment_bridge(
upstream_url=upstream_url,
api_token=http.api_token(token),
@ -289,22 +299,33 @@ def _run_wallet_client(
timeout=getattr(args, "timeout", None),
response_observer=upstream_responses.append,
) as wallet_url:
command = [
npx,
"--yes",
f"--registry={_NPM_REGISTRY}",
"--ignore-scripts",
f"--userconfig={user_config}",
f"--globalconfig={global_config}",
f"--cache={wallet_root / 'npm-cache'}",
_MPPX_PACKAGE,
wallet_url,
"--fail",
"-J",
body_json,
]
if payment_method:
command += ["-M", f"paymentMethod={payment_method}"]
if use_link_wallet:
command = [
*npx_prefix,
_LINK_CLI_PACKAGE,
"mpp",
"pay",
wallet_url,
"--method",
"POST",
"--data",
body_json,
"--context",
_payment_context(body),
"--format",
"json",
]
else:
command = [
*npx_prefix,
_MPPX_PACKAGE,
wallet_url,
"--fail",
"-J",
body_json,
]
if payment_method:
command += ["-M", f"paymentMethod={payment_method}"]
process = subprocess.run( # noqa: S603
command,
check=False,
@ -316,12 +337,126 @@ def _run_wallet_client(
return _WalletClientResult(process=process, upstream_responses=tuple(upstream_responses))
def _npx_prefix(npx: str, wallet_root: Path) -> list[str]:
"""Install the wallet client from a fixed registry without lifecycle scripts."""
return [
npx,
"--yes",
f"--registry={_NPM_REGISTRY}",
"--ignore-scripts",
f"--userconfig={wallet_root / 'user.npmrc'}",
f"--globalconfig={wallet_root / 'global.npmrc'}",
f"--cache={wallet_root / 'npm-cache'}",
]
def _payment_context(body: dict[str, Any]) -> str:
"""Describe the purchase for the person who approves it in the Link app."""
credits_requested = body.get("credits")
return (
f"Strix scan credits. The Strix command line interface asks to buy "
f"{credits_requested} scan credit(s) for the selected Strix workspace on "
"app.strix.ai. Strix spends the credits on managed penetration test scans "
"that the user starts."
)
def _mppx_wallet_configured() -> bool:
"""Report whether the person already configured the mppx wallet client."""
return bool(os.environ.get("MPPX_ACCOUNT") or os.environ.get("MPPX_STRIPE_SECRET_KEY"))
def _run_link_cli(
npx: str,
arguments: list[str],
*,
capture_output: bool,
timeout: float | None = None,
) -> subprocess.CompletedProcess[str]:
"""Run one Stripe Link wallet command in an isolated npm environment."""
with tempfile.TemporaryDirectory(prefix="strix-wallet-") as wallet_cwd:
wallet_root = Path(wallet_cwd)
(wallet_root / "user.npmrc").touch(mode=0o600)
(wallet_root / "global.npmrc").touch(mode=0o600)
return subprocess.run( # noqa: S603
[*_npx_prefix(npx, wallet_root), _LINK_CLI_PACKAGE, *arguments],
check=False,
capture_output=capture_output,
text=True,
env=_wallet_environment(),
cwd=wallet_root,
timeout=timeout,
)
def _link_wallet_authenticated(npx: str) -> bool:
"""Report whether a Link wallet is already connected to this machine."""
try:
result = _run_link_cli(
npx,
["auth", "status", "--format", "json"],
capture_output=True,
timeout=_LINK_LOGIN_TIMEOUT_S,
)
except (OSError, subprocess.SubprocessError):
return False
try:
payload = json.loads(result.stdout or "null")
except (TypeError, ValueError):
return False
if isinstance(payload, list):
payload = payload[0] if payload else None
return bool(isinstance(payload, dict) and payload.get("authenticated"))
def _prepare_link_wallet(console: Console, npx: str, *, as_json: bool) -> str | None:
"""Connect a Link wallet when none is present. Return an error message on failure."""
if _link_wallet_authenticated(npx):
return None
manual_setup = (
"Payment needs a Stripe Link wallet. Run `strix cloud billing topup` in an "
"interactive terminal to connect one, or set up the wallet at "
"https://link.com/agents. For a browser checkout instead, run "
"`strix cloud billing subscribe --plan strix_top_up`."
)
if as_json or not (sys.stdin.isatty() and sys.stdout.isatty()):
return manual_setup
console.print(
"[yellow]No Stripe Link wallet is connected.[/] Strix starts the Link sign-in now. "
"Approve the connection in the Link app, then Strix continues the payment. "
"The user approves every payment in the Link app."
)
try:
_run_link_cli(
npx,
[
"auth",
"login",
"--client-name",
_LINK_CLI_CLIENT_NAME,
"--interval",
"3",
"--timeout",
str(_LINK_LOGIN_TIMEOUT_S),
],
capture_output=False,
timeout=_LINK_LOGIN_TIMEOUT_S + 30,
)
except (OSError, subprocess.SubprocessError):
return manual_setup
if _link_wallet_authenticated(npx):
return None
return manual_setup
def _wallet_environment() -> dict[str, str]:
"""Pass only platform essentials and explicit wallet variables to npm/mppx."""
environment = {
name: value
for name, value in os.environ.items()
if name in _WALLET_ENV_NAMES or name.startswith("MPPX_")
if name in _WALLET_ENV_NAMES or name.startswith(("LINK_", "MPPX_"))
}
for name in ("NO_PROXY", "no_proxy"):
entries = [entry.strip() for entry in environment.get(name, "").split(",") if entry.strip()]

5
strix/interface/cloud/runner.py
View file

@ -678,8 +678,9 @@ def _build_parser(group: str, verb_label: str, cmd: Cmd) -> argparse.ArgumentPar
default=None,
metavar="PM_ID",
help=(
"Stripe payment method for the card payment. "
"Defaults to MPPX_STRIPE_PAYMENT_METHOD."
"Pay with the mppx wallet client and this Stripe payment method "
"instead of the Stripe Link wallet. Defaults to "
"MPPX_STRIPE_PAYMENT_METHOD."
),
)
if cmd.path == "/scans" and cmd.method == "POST":

75
tests/test_cloud_wallet.py Normal file
View file

@ -0,0 +1,75 @@
"""Tests for the Stripe Link wallet setup path of `strix cloud billing topup`."""
from __future__ import annotations
import subprocess
from typing import TYPE_CHECKING, Any
from rich.console import Console
from strix.interface.cloud import billing
if TYPE_CHECKING:
import pytest
_MIN_LINK_CONTEXT_CHARS = 100
def _completed(stdout: str) -> subprocess.CompletedProcess[str]:
return subprocess.CompletedProcess(args=["link-cli"], returncode=0, stdout=stdout, stderr="")
def test_payment_context_is_long_enough_for_link_approval() -> None:
context = billing._payment_context({"credits": 5})
assert len(context) >= _MIN_LINK_CONTEXT_CHARS
assert "5" in context
def test_mppx_wallet_configured_follows_environment(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("MPPX_ACCOUNT", raising=False)
monkeypatch.delenv("MPPX_STRIPE_SECRET_KEY", raising=False)
assert billing._mppx_wallet_configured() is False
monkeypatch.setenv("MPPX_ACCOUNT", "agent")
assert billing._mppx_wallet_configured() is True
def test_link_wallet_authenticated_reads_status_list(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
billing,
"_run_link_cli",
lambda *_args, **_kwargs: _completed('[{"authenticated": true}]'),
)
assert billing._link_wallet_authenticated("npx") is True
def test_link_wallet_authenticated_handles_unusable_output(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(billing, "_run_link_cli", lambda *_args, **_kwargs: _completed("not json"))
assert billing._link_wallet_authenticated("npx") is False
def test_link_wallet_authenticated_handles_launch_failure(
monkeypatch: pytest.MonkeyPatch,
) -> None:
def explode(*_args: Any, **_kwargs: Any) -> subprocess.CompletedProcess[str]:
raise OSError
monkeypatch.setattr(billing, "_run_link_cli", explode)
assert billing._link_wallet_authenticated("npx") is False
def test_prepare_link_wallet_skips_login_when_connected(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(billing, "_link_wallet_authenticated", lambda _npx: True)
assert billing._prepare_link_wallet(Console(), "npx", as_json=True) is None
def test_prepare_link_wallet_explains_setup_without_a_terminal(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(billing, "_link_wallet_authenticated", lambda _npx: False)
message = billing._prepare_link_wallet(Console(), "npx", as_json=True)
assert message is not None
assert "https://link.com/agents" in message