219 lines
7.1 KiB
Python
219 lines
7.1 KiB
Python
import json
|
|
import re
|
|
from http import HTTPStatus
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from lnbits.core.crud import get_user, get_wallet
|
|
from lnbits.core.models import SimpleStatus, WalletTypeInfo
|
|
from lnbits.decorators import (
|
|
check_admin,
|
|
require_admin_key,
|
|
require_invoice_key,
|
|
)
|
|
|
|
from .crud import (
|
|
create_pay_link,
|
|
delete_lnurlp_settings,
|
|
delete_pay_link,
|
|
get_or_create_lnurlp_settings,
|
|
get_pay_link,
|
|
get_pay_link_by_username,
|
|
get_pay_links,
|
|
update_lnurlp_settings,
|
|
update_pay_link,
|
|
)
|
|
from .helpers import parse_nostr_private_key
|
|
from .models import CreatePayLinkData, LnurlpSettings, PayLink
|
|
|
|
lnurlp_api_router = APIRouter()
|
|
|
|
|
|
@lnurlp_api_router.get("/api/v1/links", status_code=HTTPStatus.OK)
|
|
async def api_links(
|
|
key_info: WalletTypeInfo = Depends(require_invoice_key),
|
|
all_wallets: bool = Query(False),
|
|
) -> list[PayLink]:
|
|
wallet_ids = [key_info.wallet.id]
|
|
if all_wallets:
|
|
user = await get_user(key_info.wallet.user)
|
|
wallet_ids = user.wallet_ids if user else []
|
|
|
|
links = await get_pay_links(wallet_ids)
|
|
return links
|
|
|
|
|
|
@lnurlp_api_router.get("/api/v1/links/{link_id}", status_code=HTTPStatus.OK)
|
|
async def api_link_retrieve(
|
|
link_id: str, key_info: WalletTypeInfo = Depends(require_invoice_key)
|
|
) -> PayLink:
|
|
link = await get_pay_link(link_id)
|
|
|
|
if not link:
|
|
raise HTTPException(
|
|
detail="Pay link does not exist.", status_code=HTTPStatus.NOT_FOUND
|
|
)
|
|
|
|
link_wallet = await get_wallet(link.wallet)
|
|
|
|
# admins are allowed to read paylinks beloging to regular users
|
|
user = await get_user(key_info.wallet.user)
|
|
admin_user = user.admin if user else False
|
|
if not admin_user and link_wallet and link_wallet.user != key_info.wallet.user:
|
|
raise HTTPException(
|
|
detail="Not your pay link.", status_code=HTTPStatus.FORBIDDEN
|
|
)
|
|
return link
|
|
|
|
|
|
async def check_username_exists(username: str):
|
|
prev_link = await get_pay_link_by_username(username)
|
|
if prev_link:
|
|
raise HTTPException(
|
|
detail="Username already taken.",
|
|
status_code=HTTPStatus.CONFLICT,
|
|
)
|
|
|
|
|
|
@lnurlp_api_router.post("/api/v1/links", status_code=HTTPStatus.CREATED)
|
|
@lnurlp_api_router.put("/api/v1/links/{link_id}", status_code=HTTPStatus.OK)
|
|
async def api_link_create_or_update(
|
|
data: CreatePayLinkData,
|
|
link_id: str | None = None,
|
|
key_info: WalletTypeInfo = Depends(require_admin_key),
|
|
) -> PayLink:
|
|
if data.min > data.max:
|
|
raise HTTPException(
|
|
detail="Min is greater than max.", status_code=HTTPStatus.BAD_REQUEST
|
|
)
|
|
|
|
if not data.currency:
|
|
if round(data.min) != data.min or round(data.max) != data.max or data.min < 1:
|
|
raise HTTPException(
|
|
detail="Must use full satoshis.", status_code=HTTPStatus.BAD_REQUEST
|
|
)
|
|
|
|
if data.webhook_headers:
|
|
try:
|
|
json.loads(data.webhook_headers)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
detail="Invalid JSON in webhook_headers.",
|
|
status_code=HTTPStatus.BAD_REQUEST,
|
|
) from exc
|
|
|
|
if data.webhook_body:
|
|
try:
|
|
json.loads(data.webhook_body)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
detail="Invalid JSON in webhook_body.",
|
|
status_code=HTTPStatus.BAD_REQUEST,
|
|
) from exc
|
|
|
|
# database only allows int4 entries for min and max. For fiat currencies,
|
|
# we multiply by data.fiat_base_multiplier (usually 100) to save the value in cents.
|
|
if data.currency and data.fiat_base_multiplier:
|
|
data.min *= data.fiat_base_multiplier
|
|
data.max *= data.fiat_base_multiplier
|
|
|
|
if (
|
|
data.success_url
|
|
and data.success_url != ""
|
|
and not data.success_url.startswith("https://")
|
|
):
|
|
raise HTTPException(
|
|
detail="Success URL must be secure https://...",
|
|
status_code=HTTPStatus.BAD_REQUEST,
|
|
)
|
|
|
|
if data.username and not re.match("^[a-z0-9-_.]{1,210}$", data.username):
|
|
raise HTTPException(
|
|
detail=f"Invalid username: {data.username}. "
|
|
"Only letters a-z0-9-_. allowed, min 1 and max 210 characters!",
|
|
status_code=HTTPStatus.BAD_REQUEST,
|
|
)
|
|
|
|
# if wallet is not provided, use the wallet of the key
|
|
if not data.wallet:
|
|
data.wallet = key_info.wallet.id
|
|
|
|
new_wallet = await get_wallet(data.wallet)
|
|
if not new_wallet:
|
|
raise HTTPException(
|
|
detail="Wallet does not exist.", status_code=HTTPStatus.FORBIDDEN
|
|
)
|
|
|
|
# admins are allowed to create/edit paylinks beloging to regular users
|
|
user = await get_user(key_info.wallet.user)
|
|
admin_user = user.admin if user else False
|
|
if not admin_user and new_wallet.user != key_info.wallet.user:
|
|
raise HTTPException(
|
|
detail="Not your pay link.", status_code=HTTPStatus.FORBIDDEN
|
|
)
|
|
|
|
if link_id:
|
|
link = await get_pay_link(link_id)
|
|
|
|
if not link:
|
|
raise HTTPException(
|
|
detail="Pay link does not exist.", status_code=HTTPStatus.NOT_FOUND
|
|
)
|
|
|
|
if data.username and data.username != link.username:
|
|
await check_username_exists(data.username)
|
|
|
|
for k, v in data.dict().items():
|
|
setattr(link, k, v)
|
|
|
|
link = await update_pay_link(link)
|
|
else:
|
|
if data.username:
|
|
await check_username_exists(data.username)
|
|
|
|
link = await create_pay_link(data)
|
|
|
|
return link
|
|
|
|
|
|
@lnurlp_api_router.delete("/api/v1/links/{link_id}", status_code=HTTPStatus.OK)
|
|
async def api_link_delete(
|
|
link_id: str, key_info: WalletTypeInfo = Depends(require_admin_key)
|
|
) -> SimpleStatus:
|
|
link = await get_pay_link(link_id)
|
|
|
|
if not link:
|
|
raise HTTPException(
|
|
detail="Pay link does not exist.", status_code=HTTPStatus.NOT_FOUND
|
|
)
|
|
|
|
# admins are allowed to delete paylinks belonging to regular users
|
|
user = await get_user(key_info.wallet.user)
|
|
admin_user = user.admin if user else False
|
|
if not admin_user and link.wallet != key_info.wallet.id:
|
|
raise HTTPException(
|
|
detail="Not your pay link.", status_code=HTTPStatus.FORBIDDEN
|
|
)
|
|
|
|
await delete_pay_link(link_id)
|
|
return SimpleStatus(success=True, message="Deleted Pay link")
|
|
|
|
|
|
@lnurlp_api_router.get("/api/v1/settings", dependencies=[Depends(check_admin)])
|
|
async def api_get_or_create_settings() -> LnurlpSettings:
|
|
return await get_or_create_lnurlp_settings()
|
|
|
|
|
|
@lnurlp_api_router.put("/api/v1/settings", dependencies=[Depends(check_admin)])
|
|
async def api_update_settings(data: LnurlpSettings) -> LnurlpSettings:
|
|
try:
|
|
parse_nostr_private_key(data.nostr_private_key)
|
|
except Exception as exc:
|
|
raise HTTPException(
|
|
detail="Invalid Nostr private key.", status_code=HTTPStatus.BAD_REQUEST
|
|
) from exc
|
|
return await update_lnurlp_settings(data)
|
|
|
|
|
|
@lnurlp_api_router.delete("/api/v1/settings", dependencies=[Depends(check_admin)])
|
|
async def api_delete_settings() -> None:
|
|
await delete_lnurlp_settings()
|