[feat] Add stripe payments (#3184)
This commit is contained in:
Vlad Stan
dni ⚡
parent
5c9511ccfe
commit
695e9b6471
51 changed files with 2014 additions and 175 deletions
|
|
@ -8,10 +8,12 @@ from pytest_mock.plugin import MockerFixture
|
|||
from lnbits import bolt11
|
||||
from lnbits.core.models import CreateInvoice, Payment
|
||||
from lnbits.core.views.payment_api import api_payment
|
||||
from lnbits.fiat.base import FiatInvoiceResponse
|
||||
from lnbits.settings import Settings
|
||||
|
||||
from ..helpers import (
|
||||
get_random_invoice_data,
|
||||
get_random_string,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -155,6 +157,60 @@ async def test_create_invoice_fiat_amount(client, inkey_headers_to):
|
|||
assert extra["fiat_rate"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_fiat_invoice(
|
||||
client, inkey_headers_to, settings: Settings, mocker: MockerFixture
|
||||
):
|
||||
data = await get_random_invoice_data()
|
||||
data["unit"] = "EUR"
|
||||
data["fiat_provider"] = "stripe"
|
||||
|
||||
settings.stripe_enabled = True
|
||||
settings.stripe_api_secret_key = "mock_sk_test_4eC39HqLyjWDarjtT1zdp7dc"
|
||||
|
||||
fiat_payment_request = "https://stripe.com/pay/session_123"
|
||||
fiat_mock_response = FiatInvoiceResponse(
|
||||
ok=True,
|
||||
checking_id=f"session_123_{get_random_string(10)}",
|
||||
payment_request=fiat_payment_request,
|
||||
)
|
||||
|
||||
mocker.patch(
|
||||
"lnbits.fiat.StripeWallet.create_invoice",
|
||||
AsyncMock(return_value=fiat_mock_response),
|
||||
)
|
||||
mocker.patch(
|
||||
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
||||
AsyncMock(return_value=1000), # 1 BTC = 100 000 EUR, so 1 EUR = 1000 sats
|
||||
)
|
||||
response = await client.post(
|
||||
"/api/v1/payments", json=data, headers=inkey_headers_to
|
||||
)
|
||||
assert response.status_code == 201
|
||||
invoice = response.json()
|
||||
decode = bolt11.decode(invoice["bolt11"])
|
||||
assert decode.amount_msat == 10_000_000
|
||||
assert decode.payment_hash
|
||||
assert invoice["fiat_provider"] == "stripe"
|
||||
assert invoice["status"] == "pending"
|
||||
assert invoice["extra"]["fiat_checking_id"]
|
||||
assert invoice["extra"]["fiat_payment_request"] == fiat_payment_request
|
||||
|
||||
response = await client.get(
|
||||
f"/api/v1/payments/{decode.payment_hash}", headers=inkey_headers_to
|
||||
)
|
||||
assert response.is_success
|
||||
data = response.json()
|
||||
assert data["status"] == "pending"
|
||||
invoice = data["details"]
|
||||
|
||||
assert invoice["fiat_provider"] == "stripe"
|
||||
assert invoice["status"] == "pending"
|
||||
assert invoice["amount"] == 10_000_000
|
||||
assert invoice["extra"]["fiat_checking_id"]
|
||||
assert invoice["extra"]["fiat_payment_request"] == fiat_payment_request
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize("currency", ("msat", "RRR"))
|
||||
async def test_create_invoice_validates_used_currency(
|
||||
|
|
|
|||
|
|
@ -19,10 +19,10 @@ from lnbits.core.crud import (
|
|||
from lnbits.core.models import Account, CreateInvoice, PaymentState, User
|
||||
from lnbits.core.models.users import UpdateSuperuserPassword
|
||||
from lnbits.core.services import create_user_account, update_wallet_balance
|
||||
from lnbits.core.services.payments import create_wallet_invoice
|
||||
from lnbits.core.views.auth_api import first_install
|
||||
from lnbits.core.views.payment_api import _api_payments_create_invoice
|
||||
from lnbits.db import DB_TYPE, SQLITE, Database
|
||||
from lnbits.settings import AuthMethods, Settings
|
||||
from lnbits.settings import AuthMethods, FiatProviderLimits, Settings
|
||||
from lnbits.settings import settings as lnbits_settings
|
||||
from lnbits.wallets.fake import FakeWallet
|
||||
from tests.helpers import (
|
||||
|
|
@ -255,7 +255,7 @@ async def adminkey_headers_to(to_wallet):
|
|||
async def invoice(to_wallet):
|
||||
data = await get_random_invoice_data()
|
||||
invoice_data = CreateInvoice(**data)
|
||||
invoice = await _api_payments_create_invoice(invoice_data, to_wallet)
|
||||
invoice = await create_wallet_invoice(to_wallet.id, invoice_data)
|
||||
yield invoice
|
||||
del invoice
|
||||
|
||||
|
|
@ -312,3 +312,4 @@ def _settings_cleanup(settings: Settings):
|
|||
settings.lnbits_admin_users = []
|
||||
settings.lnbits_max_outgoing_payment_amount_sats = 10_000_000_100
|
||||
settings.lnbits_max_incoming_payment_amount_sats = 10_000_000_200
|
||||
settings.stripe_limits = FiatProviderLimits()
|
||||
|
|
|
|||
432
tests/unit/test_fiat_providers.py
Normal file
432
tests/unit/test_fiat_providers.py
Normal file
|
|
@ -0,0 +1,432 @@
|
|||
import hashlib
|
||||
import hmac
|
||||
import time
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from pytest_mock.plugin import MockerFixture
|
||||
|
||||
from lnbits.core.crud.payments import get_payments
|
||||
from lnbits.core.crud.users import get_user
|
||||
from lnbits.core.crud.wallets import create_wallet
|
||||
from lnbits.core.models.payments import CreateInvoice, PaymentState
|
||||
from lnbits.core.models.users import User
|
||||
from lnbits.core.models.wallets import Wallet
|
||||
from lnbits.core.services import payments
|
||||
from lnbits.core.services.fiat_providers import check_stripe_signature
|
||||
from lnbits.core.services.users import create_user_account
|
||||
from lnbits.fiat.base import FiatInvoiceResponse, FiatPaymentStatus
|
||||
from lnbits.settings import Settings
|
||||
from tests.helpers import get_random_string
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_wallet_fiat_invoice_missing_provider():
|
||||
invoice_data = CreateInvoice(
|
||||
unit="USD", amount=1.0, memo="Test", fiat_provider=None
|
||||
)
|
||||
with pytest.raises(ValueError, match="Fiat provider is required"):
|
||||
await payments.create_fiat_invoice("wallet_id", invoice_data)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_wallet_fiat_invoice_provider_not_enabled(settings: Settings):
|
||||
settings.stripe_enabled = False
|
||||
invoice_data = CreateInvoice(
|
||||
unit="USD", amount=1.0, memo="Test", fiat_provider="notarealprovider"
|
||||
)
|
||||
with pytest.raises(
|
||||
ValueError, match="Fiat provider 'notarealprovider' is not enabled"
|
||||
):
|
||||
await payments.create_fiat_invoice("wallet_id", invoice_data)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_wallet_fiat_invoice_with_sat_unit(settings: Settings):
|
||||
settings.stripe_enabled = True
|
||||
invoice_data = CreateInvoice(
|
||||
unit="sat", amount=1.0, memo="Test", fiat_provider="stripe"
|
||||
)
|
||||
with pytest.raises(ValueError, match="Fiat provider cannot be used with satoshis"):
|
||||
await payments.create_fiat_invoice("wallet_id", invoice_data)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_wallet_fiat_invoice_allowed_users(
|
||||
to_user: User, settings: Settings
|
||||
):
|
||||
|
||||
settings.stripe_enabled = False
|
||||
settings.stripe_limits.allowed_users = []
|
||||
user = await get_user(to_user.id)
|
||||
assert user
|
||||
assert user.fiat_providers == []
|
||||
|
||||
settings.stripe_enabled = True
|
||||
user = await get_user(to_user.id)
|
||||
assert user
|
||||
assert user.fiat_providers == ["stripe"]
|
||||
|
||||
settings.stripe_limits.allowed_users = ["some_other_user_id"]
|
||||
user = await get_user(to_user.id)
|
||||
assert user
|
||||
assert user.fiat_providers == []
|
||||
|
||||
settings.stripe_limits.allowed_users.append(to_user.id)
|
||||
user = await get_user(to_user.id)
|
||||
assert user
|
||||
assert user.fiat_providers == ["stripe"]
|
||||
|
||||
settings.stripe_enabled = False
|
||||
user = await get_user(to_user.id)
|
||||
assert user
|
||||
assert user.fiat_providers == []
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_wallet_fiat_invoice_fiat_limits_fail(
|
||||
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
|
||||
):
|
||||
|
||||
settings.stripe_enabled = True
|
||||
settings.stripe_limits.service_min_amount_sats = 0
|
||||
settings.stripe_limits.service_max_amount_sats = 105
|
||||
settings.stripe_limits.service_faucet_wallet_id = None
|
||||
invoice_data = CreateInvoice(
|
||||
unit="USD", amount=1.0, memo="Test", fiat_provider="stripe"
|
||||
)
|
||||
|
||||
mocker.patch(
|
||||
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
||||
AsyncMock(return_value=1000), # 1 BTC = 100 000 USD, so 1 USD = 1000 sats
|
||||
)
|
||||
with pytest.raises(ValueError, match="Maximum amount is 105 sats for 'stripe'."):
|
||||
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
||||
|
||||
settings.stripe_limits.service_min_amount_sats = 1001
|
||||
settings.stripe_limits.service_max_amount_sats = 10000
|
||||
|
||||
with pytest.raises(ValueError, match="Minimum amount is 1001 sats for 'stripe'."):
|
||||
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
||||
|
||||
settings.stripe_limits.service_min_amount_sats = 10
|
||||
settings.stripe_limits.service_max_amount_sats = 10000
|
||||
settings.stripe_limits.service_max_fee_sats = 100
|
||||
|
||||
with pytest.raises(
|
||||
ValueError, match="Fiat provider 'stripe' service fee wallet missing."
|
||||
):
|
||||
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
||||
|
||||
settings.stripe_limits.service_fee_wallet_id = "not_a_real_wallet_id"
|
||||
|
||||
with pytest.raises(
|
||||
ValueError, match="Fiat provider 'stripe' service fee wallet not found."
|
||||
):
|
||||
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
||||
|
||||
settings.stripe_limits.service_fee_wallet_id = to_wallet.id
|
||||
settings.stripe_limits.service_faucet_wallet_id = "not_a_real_wallet_id"
|
||||
|
||||
with pytest.raises(
|
||||
ValueError, match="Fiat provider 'stripe' faucet wallet not found."
|
||||
):
|
||||
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
||||
|
||||
user = await create_user_account()
|
||||
wallet = await create_wallet(user_id=user.id)
|
||||
settings.stripe_limits.service_faucet_wallet_id = wallet.id
|
||||
|
||||
with pytest.raises(
|
||||
ValueError, match="The amount exceeds the 'stripe'faucet wallet balance."
|
||||
):
|
||||
await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_wallet_fiat_provider_fails(
|
||||
settings: Settings, mocker: MockerFixture
|
||||
):
|
||||
settings.stripe_enabled = True
|
||||
settings.stripe_api_secret_key = "mock_sk_test_4eC39HqLyjWDarjtT1zdp7dc"
|
||||
invoice_data = CreateInvoice(
|
||||
unit="USD", amount=2.0, memo="Test", fiat_provider="stripe"
|
||||
)
|
||||
|
||||
fiat_mock_response = FiatInvoiceResponse(
|
||||
ok=False,
|
||||
error_message="Failed to create invoice",
|
||||
)
|
||||
|
||||
mocker.patch(
|
||||
"lnbits.fiat.StripeWallet.create_invoice",
|
||||
AsyncMock(return_value=fiat_mock_response),
|
||||
)
|
||||
mocker.patch(
|
||||
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
||||
AsyncMock(return_value=1000), # 1 BTC = 100 000 USD, so 1 USD = 1000 sats
|
||||
)
|
||||
|
||||
user = await create_user_account()
|
||||
wallet = await create_wallet(user_id=user.id)
|
||||
with pytest.raises(ValueError, match="Cannot create payment request for 'stripe'."):
|
||||
await payments.create_fiat_invoice(wallet.id, invoice_data)
|
||||
|
||||
wallet_payments = await get_payments(wallet_id=wallet.id)
|
||||
assert len(wallet_payments) == 1
|
||||
assert wallet_payments[0].status == PaymentState.FAILED
|
||||
assert wallet_payments[0].amount == 2000000
|
||||
assert wallet_payments[0].fee == 0
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_wallet_fiat_invoice_success(
|
||||
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
|
||||
):
|
||||
settings.stripe_enabled = True
|
||||
settings.stripe_api_secret_key = "mock_sk_test_4eC39HqLyjWDarjtT1zdp7dc"
|
||||
settings.stripe_limits.service_min_amount_sats = 0
|
||||
settings.stripe_limits.service_max_amount_sats = 0
|
||||
settings.stripe_limits.service_faucet_wallet_id = None
|
||||
|
||||
invoice_data = CreateInvoice(
|
||||
unit="USD", amount=1.0, memo="Test", fiat_provider="stripe"
|
||||
)
|
||||
fiat_mock_response = FiatInvoiceResponse(
|
||||
ok=True,
|
||||
checking_id=f"session_123_{get_random_string(10)}",
|
||||
payment_request="https://stripe.com/pay/session_123",
|
||||
)
|
||||
|
||||
mocker.patch(
|
||||
"lnbits.fiat.StripeWallet.create_invoice",
|
||||
AsyncMock(return_value=fiat_mock_response),
|
||||
)
|
||||
mocker.patch(
|
||||
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
||||
AsyncMock(return_value=1000), # 1 BTC = 100 000 USD, so 1 USD = 1000 sats
|
||||
)
|
||||
payment = await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
||||
assert payment.status == PaymentState.PENDING
|
||||
assert payment.amount == 1000_000
|
||||
assert payment.fiat_provider == "stripe"
|
||||
assert payment.extra.get("fiat_checking_id") == fiat_mock_response.checking_id
|
||||
assert (
|
||||
payment.extra.get("fiat_payment_request")
|
||||
== "https://stripe.com/pay/session_123"
|
||||
)
|
||||
assert payment.checking_id.startswith("fiat_stripe_")
|
||||
assert payment.fee <= 0
|
||||
|
||||
status = await payment.check_status()
|
||||
assert status.success is False
|
||||
assert status.pending is True
|
||||
|
||||
fiat_mock_status = FiatPaymentStatus(paid=True, fee=123)
|
||||
mocker.patch(
|
||||
"lnbits.fiat.StripeWallet.get_invoice_status",
|
||||
AsyncMock(return_value=fiat_mock_status),
|
||||
)
|
||||
status = await payment.check_status()
|
||||
assert status.paid is True
|
||||
assert status.success is True
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_fiat_service_fee(settings: Settings):
|
||||
# settings.stripe_limits.service_min_amount_sats = 0
|
||||
amount_msats = 100_000
|
||||
fee = payments.service_fee_fiat(amount_msats, "no_such_fiat_provider")
|
||||
assert fee == 0
|
||||
|
||||
settings.stripe_limits.service_fee_wallet_id = None
|
||||
fee = payments.service_fee_fiat(amount_msats, "stripe")
|
||||
assert fee == 0
|
||||
|
||||
settings.stripe_limits.service_fee_wallet_id = "wallet_id"
|
||||
fee = payments.service_fee_fiat(amount_msats, "stripe")
|
||||
assert fee == 0
|
||||
|
||||
settings.stripe_limits.service_max_fee_sats = 5
|
||||
settings.stripe_limits.service_fee_percent = 20
|
||||
fee = payments.service_fee_fiat(amount_msats, "stripe")
|
||||
assert fee == 5000
|
||||
|
||||
fee = payments.service_fee_fiat(-amount_msats, "stripe")
|
||||
assert fee == 5000
|
||||
|
||||
settings.stripe_limits.service_max_fee_sats = 5
|
||||
settings.stripe_limits.service_fee_percent = 3
|
||||
fee = payments.service_fee_fiat(amount_msats, "stripe")
|
||||
assert fee == 3000
|
||||
|
||||
fee = payments.service_fee_fiat(-amount_msats, "stripe")
|
||||
assert fee == 3000
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_handle_fiat_payment_confirmation(
|
||||
to_wallet: Wallet, settings: Settings, mocker: MockerFixture
|
||||
):
|
||||
user = await create_user_account()
|
||||
service_fee_wallet = await create_wallet(user_id=user.id)
|
||||
faucet_wallet = await create_wallet(user_id=user.id)
|
||||
await payments.update_wallet_balance(wallet=faucet_wallet, amount=100_000_000)
|
||||
|
||||
settings.stripe_api_secret_key = "mock_sk_test_4eC39HqLyjWDarjtT1zdp7dc"
|
||||
invoice_data = CreateInvoice(
|
||||
unit="USD", amount=1.0, memo="Test", fiat_provider="stripe"
|
||||
)
|
||||
|
||||
settings.stripe_enabled = True
|
||||
settings.stripe_limits.service_min_amount_sats = 0
|
||||
settings.stripe_limits.service_max_amount_sats = 0
|
||||
|
||||
settings.stripe_limits.service_fee_percent = 20
|
||||
settings.stripe_limits.service_fee_wallet_id = service_fee_wallet.id
|
||||
settings.stripe_limits.service_faucet_wallet_id = faucet_wallet.id
|
||||
|
||||
fiat_mock_response = FiatInvoiceResponse(
|
||||
ok=True,
|
||||
checking_id=f"session_1000_{get_random_string(10)}",
|
||||
payment_request="https://stripe.com/pay/session_1000",
|
||||
)
|
||||
|
||||
mocker.patch(
|
||||
"lnbits.fiat.StripeWallet.create_invoice",
|
||||
AsyncMock(return_value=fiat_mock_response),
|
||||
)
|
||||
mocker.patch(
|
||||
"lnbits.utils.exchange_rates.get_fiat_rate_satoshis",
|
||||
AsyncMock(return_value=10000), # 1 BTC = 100 000 USD, so 1 USD = 1000 sats
|
||||
)
|
||||
payment = await payments.create_fiat_invoice(to_wallet.id, invoice_data)
|
||||
assert payment.status == PaymentState.PENDING
|
||||
assert payment.amount == 10_000_000
|
||||
|
||||
await payments.handle_fiat_payment_confirmation(payment)
|
||||
# await asyncio.sleep(1) # Simulate async delay
|
||||
|
||||
service_fee_payments = await get_payments(wallet_id=service_fee_wallet.id)
|
||||
assert len(service_fee_payments) == 1
|
||||
assert service_fee_payments[0].amount == 2_000_000
|
||||
assert service_fee_payments[0].fee == 0
|
||||
assert service_fee_payments[0].status == PaymentState.SUCCESS
|
||||
assert service_fee_payments[0].fiat_provider is None
|
||||
|
||||
faucet_wallet_payments = await get_payments(wallet_id=faucet_wallet.id)
|
||||
|
||||
# Background tasks may create more payments, so we check for at least 2
|
||||
# One for the service fee, one for the top-up)
|
||||
assert len(faucet_wallet_payments) >= 2
|
||||
faucet_payment = next(
|
||||
(p for p in faucet_wallet_payments if p.payment_hash == payment.payment_hash),
|
||||
None,
|
||||
)
|
||||
assert faucet_payment
|
||||
assert faucet_payment.amount == -10_000_000
|
||||
assert faucet_payment.fee == 0
|
||||
assert faucet_payment.status == PaymentState.SUCCESS
|
||||
assert faucet_payment.fiat_provider is None
|
||||
assert (
|
||||
faucet_payment.extra.get("fiat_checking_id") == fiat_mock_response.checking_id
|
||||
)
|
||||
assert (
|
||||
faucet_payment.extra.get("fiat_payment_request")
|
||||
== fiat_mock_response.payment_request
|
||||
)
|
||||
assert faucet_payment.checking_id.startswith("internal_fiat_stripe_")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload", [b'{"id": "evt_test"}', b"{}", b""])
|
||||
def test_check_stripe_signature_success(payload):
|
||||
secret = "whsec_testsecret"
|
||||
sig_header, _, _ = _make_stripe_sig_header(payload, secret)
|
||||
# Should not raise
|
||||
check_stripe_signature(payload, sig_header, secret)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload", [b'{"id": "evt_test"}'])
|
||||
def test_check_stripe_signature_missing_header(payload):
|
||||
secret = "whsec_testsecret"
|
||||
with pytest.raises(ValueError, match="Stripe-Signature header is missing"):
|
||||
check_stripe_signature(payload, None, secret)
|
||||
|
||||
|
||||
def test_check_stripe_signature_missing_secret():
|
||||
payload = b'{"id": "evt_test"}'
|
||||
sig_header, _, _ = _make_stripe_sig_header(payload, "whsec_testsecret")
|
||||
with pytest.raises(ValueError, match="Stripe webhook cannot be verified"):
|
||||
check_stripe_signature(payload, sig_header, None)
|
||||
|
||||
|
||||
def test_check_stripe_signature_invalid_signature():
|
||||
payload = b'{"id": "evt_test"}'
|
||||
secret = "whsec_testsecret"
|
||||
_, timestamp, _ = _make_stripe_sig_header(payload, secret)
|
||||
# Tamper with signature
|
||||
bad_sig_header = f"t={timestamp},v1=deadbeef"
|
||||
with pytest.raises(ValueError, match="Stripe signature verification failed"):
|
||||
check_stripe_signature(payload, bad_sig_header, secret)
|
||||
|
||||
|
||||
def test_check_stripe_signature_old_timestamp():
|
||||
payload = b'{"id": "evt_test"}'
|
||||
secret = "whsec_testsecret"
|
||||
old_timestamp = int(time.time()) - 10000 # way outside default tolerance
|
||||
sig_header, _, _ = _make_stripe_sig_header(payload, secret, timestamp=old_timestamp)
|
||||
with pytest.raises(ValueError, match="Timestamp outside tolerance"):
|
||||
check_stripe_signature(payload, sig_header, secret)
|
||||
|
||||
|
||||
def test_check_stripe_signature_future_timestamp():
|
||||
payload = b'{"id": "evt_test"}'
|
||||
secret = "whsec_testsecret"
|
||||
future_timestamp = int(time.time()) + 10000
|
||||
sig_header, _, _ = _make_stripe_sig_header(
|
||||
payload, secret, timestamp=future_timestamp
|
||||
)
|
||||
with pytest.raises(ValueError, match="Timestamp outside tolerance"):
|
||||
check_stripe_signature(payload, sig_header, secret)
|
||||
|
||||
|
||||
def test_check_stripe_signature_malformed_header():
|
||||
payload = b'{"id": "evt_test"}'
|
||||
secret = "whsec_testsecret"
|
||||
# Missing v1 part
|
||||
bad_header = "t=1234567890"
|
||||
with pytest.raises(Exception): # noqa: B017
|
||||
check_stripe_signature(payload, bad_header, secret)
|
||||
# Missing t part
|
||||
bad_header2 = "v1=abcdef"
|
||||
with pytest.raises(Exception): # noqa: B017
|
||||
check_stripe_signature(payload, bad_header2, secret)
|
||||
# Not split by =
|
||||
bad_header3 = "t1234567890,v1abcdef"
|
||||
with pytest.raises(Exception): # noqa: B017
|
||||
check_stripe_signature(payload, bad_header3, secret)
|
||||
|
||||
|
||||
def test_check_stripe_signature_non_utf8_payload():
|
||||
secret = "whsec_testsecret"
|
||||
payload = b"\xff\xfe\xfd" # not valid utf-8
|
||||
timestamp = int(time.time())
|
||||
# This will raise UnicodeDecodeError inside check_stripe_signature
|
||||
signed_payload = f"{timestamp}." + payload.decode(errors="ignore")
|
||||
signature = hmac.new(
|
||||
secret.encode(), signed_payload.encode(), hashlib.sha256
|
||||
).hexdigest()
|
||||
sig_header = f"t={timestamp},v1={signature}"
|
||||
with pytest.raises(UnicodeDecodeError):
|
||||
check_stripe_signature(payload, sig_header, secret)
|
||||
|
||||
|
||||
# Helper to generate a valid Stripe signature header
|
||||
def _make_stripe_sig_header(payload, secret, timestamp=None):
|
||||
if timestamp is None:
|
||||
timestamp = int(time.time())
|
||||
signed_payload = f"{timestamp}.{payload.decode()}"
|
||||
signature = hmac.new(
|
||||
secret.encode(), signed_payload.encode(), hashlib.sha256
|
||||
).hexdigest()
|
||||
return f"t={timestamp},v1={signature}", timestamp, signature
|
||||
Loading…
Add table
Add a link
Reference in a new issue