40564f06ec
Promoted from the door-portal scratch repo to its own repo for install via the aiolabs catalog. Authenticates a tapped Bolt Card via boltcards /verify, authorizes against per-door grants, and fires the door's local Home Assistant webhook to unlock a Z-Wave lock. Fails closed; logs every attempt. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
from http import HTTPStatus
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from lnbits.core.crud import get_user
|
|
from lnbits.core.models import WalletTypeInfo
|
|
from lnbits.decorators import require_admin_key, require_invoice_key
|
|
|
|
from .crud import (
|
|
create_door,
|
|
create_grant,
|
|
delete_door,
|
|
delete_grant,
|
|
get_door,
|
|
get_doors,
|
|
get_grant,
|
|
get_grants,
|
|
get_logs,
|
|
update_door,
|
|
)
|
|
from .models import AccessLog, CreateDoor, CreateGrant, Door, Grant
|
|
|
|
access_api_router = APIRouter()
|
|
|
|
|
|
async def _wallet_ids(key_info: WalletTypeInfo, all_wallets: bool) -> list[str]:
|
|
if not all_wallets:
|
|
return [key_info.wallet.id]
|
|
user = await get_user(key_info.wallet.user)
|
|
return user.wallet_ids if user else []
|
|
|
|
|
|
async def _owned_door(door_id: str, wallet_id: str) -> Door:
|
|
door = await get_door(door_id)
|
|
if not door:
|
|
raise HTTPException(HTTPStatus.NOT_FOUND, "Door does not exist.")
|
|
if door.wallet != wallet_id:
|
|
raise HTTPException(HTTPStatus.FORBIDDEN, "Not your door.")
|
|
return door
|
|
|
|
|
|
# ── Doors ──────────────────────────────────────────────────────────────────
|
|
@access_api_router.get("/api/v1/doors")
|
|
async def api_doors(
|
|
key_info: WalletTypeInfo = Depends(require_invoice_key), all_wallets: bool = False
|
|
) -> list[Door]:
|
|
return await get_doors(await _wallet_ids(key_info, all_wallets))
|
|
|
|
|
|
@access_api_router.post("/api/v1/doors", status_code=HTTPStatus.CREATED)
|
|
async def api_door_create(
|
|
data: CreateDoor, key_info: WalletTypeInfo = Depends(require_admin_key)
|
|
) -> Door:
|
|
return await create_door(key_info.wallet.id, data)
|
|
|
|
|
|
@access_api_router.put("/api/v1/doors/{door_id}")
|
|
async def api_door_update(
|
|
door_id: str,
|
|
data: CreateDoor,
|
|
key_info: WalletTypeInfo = Depends(require_admin_key),
|
|
) -> Door:
|
|
door = await _owned_door(door_id, key_info.wallet.id)
|
|
door.name = data.name
|
|
door.ha_webhook_url = data.ha_webhook_url
|
|
door.boltcards_base_url = data.boltcards_base_url
|
|
door.unlock_timeout_ms = data.unlock_timeout_ms
|
|
door.enabled = data.enabled
|
|
if data.controller_token:
|
|
door.controller_token = data.controller_token
|
|
return await update_door(door)
|
|
|
|
|
|
@access_api_router.delete("/api/v1/doors/{door_id}")
|
|
async def api_door_delete(
|
|
door_id: str, key_info: WalletTypeInfo = Depends(require_admin_key)
|
|
):
|
|
await _owned_door(door_id, key_info.wallet.id)
|
|
await delete_door(door_id)
|
|
return {"deleted": True}
|
|
|
|
|
|
# ── Grants ─────────────────────────────────────────────────────────────────
|
|
@access_api_router.get("/api/v1/grants")
|
|
async def api_grants(
|
|
key_info: WalletTypeInfo = Depends(require_invoice_key), all_wallets: bool = False
|
|
) -> list[Grant]:
|
|
doors = await get_doors(await _wallet_ids(key_info, all_wallets))
|
|
return await get_grants([d.id for d in doors])
|
|
|
|
|
|
@access_api_router.post("/api/v1/grants", status_code=HTTPStatus.CREATED)
|
|
async def api_grant_create(
|
|
data: CreateGrant, key_info: WalletTypeInfo = Depends(require_admin_key)
|
|
) -> Grant:
|
|
# Only allow granting on a door the caller owns.
|
|
await _owned_door(data.door_id, key_info.wallet.id)
|
|
return await create_grant(data)
|
|
|
|
|
|
@access_api_router.delete("/api/v1/grants/{grant_id}")
|
|
async def api_grant_delete(
|
|
grant_id: str, key_info: WalletTypeInfo = Depends(require_admin_key)
|
|
):
|
|
grant = await get_grant(grant_id)
|
|
if not grant:
|
|
raise HTTPException(HTTPStatus.NOT_FOUND, "Grant does not exist.")
|
|
await _owned_door(grant.door_id, key_info.wallet.id)
|
|
await delete_grant(grant_id)
|
|
return {"deleted": True}
|
|
|
|
|
|
# ── Access log ─────────────────────────────────────────────────────────────
|
|
@access_api_router.get("/api/v1/logs")
|
|
async def api_logs(
|
|
key_info: WalletTypeInfo = Depends(require_invoice_key), all_wallets: bool = False
|
|
) -> list[AccessLog]:
|
|
doors = await get_doors(await _wallet_ids(key_info, all_wallets))
|
|
return await get_logs([d.id for d in doors])
|