feat: access extension — NFC door access via boltcards SUN + Home Assistant

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>
This commit is contained in:
Patrick Mulligan 2026-08-07 22:47:41 +02:00
commit bf4994cbe4
22 changed files with 986 additions and 0 deletions

70
README.md Normal file
View file

@ -0,0 +1,70 @@
# Access — LNbits extension
NFC **door access** for LNbits. A tapped Bolt Card (NTAG424) is authenticated
via the `boltcards` SUN check, authorized against per-door **grants**, and — if
allowed — the door's local **Home Assistant** webhook is fired to unlock a
Z-Wave lock. Every tap is logged.
Runs on the **on-prem LNbits** on the door's LAN, so the unlock path stays local
and the door does not depend on the internet.
## Flow
```text
Pi + PN532 reader
→ POST /access/api/v1/check (X-Controller-Token: <door token>)
{ "doorId": "...", "external_id": "...", "p": "...", "c": "..." }
1. authenticate → boltcards /verify (valid, non-replayed SUN?)
2. authorize → active grant for this external_id on this door?
3. actuate → POST door.ha_webhook_url (→ lock.unlock → Z-Wave)
4. log → access.logs (allow/deny + reason)
→ { "allow": true|false, "reason": "..." }
```
The reader never unlocks directly; this extension owns the allow/deny decision
and the Home Assistant call. Fails **closed** at every step.
## Data model
- **doors** — a resource: `name`, `controller_token` (reader secret),
`ha_webhook_url` (local HA webhook), `boltcards_base_url` (blank = this
instance), `unlock_timeout_ms`, `enabled`.
- **grants**`external_id → door` permission, `enabled`/`expires_at`.
- **logs** — every allow/deny with a reason.
## Admin API (wallet admin key)
- `GET/POST /access/api/v1/doors`, `PUT/DELETE /access/api/v1/doors/{id}`
- `GET/POST /access/api/v1/grants`, `DELETE /access/api/v1/grants/{id}`
- `GET /access/api/v1/logs`
## Reader API (controller token)
```http
POST /access/api/v1/check
X-Controller-Token: <door.controller_token>
Content-Type: application/json
{ "doorId": "front-door", "external_id": "abc123", "p": "<32-hex>", "c": "<16-hex>" }
```
## Setup
1. Install on the on-prem LNbits (same instance as `boltcards`).
2. Create a **door**, set its Home Assistant webhook URL, copy its
**controller token** to the reader config.
3. Issue Bolt Cards in `boltcards` as usual; add a **grant** (the card's
`external_id` → the door).
4. Point the door reader (Pi) at `/access/api/v1/check` with the token.
## Notes
- Card authentication delegates to `boltcards` over loopback HTTP via its
**`/api/v1/verify/{external_id}`** endpoint (aiolabs fork ≥ **1.1.1-aio.2**) —
a side-effect-light SUN check that returns `{"authenticated": true, …}`
without `/scan`'s spend semantics (no withdrawRequest, no daily-limit, no
"hit"), while still advancing the SUN counter for replay protection.
- Requires boltcards **1.1.1-aio.2+** installed on the same LNbits.
- Empty `ha_webhook_url` = authorize + log only (bring-up mode; no unlock).
- Keep this on the LAN; use a strong per-door controller token; never expose
Home Assistant publicly.

24
__init__.py Normal file
View file

@ -0,0 +1,24 @@
from fastapi import APIRouter
from .crud import db
from .views import access_generic_router
from .views_api import access_api_router
from .views_reader import access_reader_router
access_static_files = [
{
"path": "/access/static",
"name": "access_static",
}
]
access_ext: APIRouter = APIRouter(prefix="/access", tags=["access"])
access_ext.include_router(access_generic_router)
access_ext.include_router(access_api_router)
access_ext.include_router(access_reader_router)
__all__ = [
"access_ext",
"access_static_files",
"db",
]

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

13
config.json Normal file
View file

@ -0,0 +1,13 @@
{
"name": "Access",
"short_description": "NFC door access — authorize Bolt Card taps and unlock via Home Assistant",
"tile": "/access/static/image/access.png",
"version": "0.0.1",
"min_lnbits_version": "1.3.0",
"contributors": [
{
"name": "aiolabs",
"uri": "https://git.atitlan.io/aiolabs"
}
]
}

160
crud.py Normal file
View file

@ -0,0 +1,160 @@
import secrets
from lnbits.db import Database
from lnbits.helpers import urlsafe_short_hash
from .models import AccessLog, CreateDoor, CreateGrant, Door, Grant
db = Database("ext_access")
# ── Doors ──────────────────────────────────────────────────────────────────
async def create_door(wallet_id: str, data: CreateDoor) -> Door:
door_id = urlsafe_short_hash()
controller_token = data.controller_token or secrets.token_urlsafe(24)
await db.execute(
"""
INSERT INTO access.doors (
id, wallet, name, controller_token, ha_webhook_url,
boltcards_base_url, unlock_timeout_ms, enabled
)
VALUES (
:id, :wallet, :name, :controller_token, :ha_webhook_url,
:boltcards_base_url, :unlock_timeout_ms, :enabled
)
""",
{
"id": door_id,
"wallet": wallet_id,
"name": data.name,
"controller_token": controller_token,
"ha_webhook_url": data.ha_webhook_url,
"boltcards_base_url": data.boltcards_base_url,
"unlock_timeout_ms": data.unlock_timeout_ms,
"enabled": data.enabled,
},
)
door = await get_door(door_id)
assert door, "Newly created door couldn't be retrieved"
return door
async def update_door(door: Door) -> Door:
await db.update("access.doors", door)
return door
async def get_door(door_id: str) -> Door | None:
return await db.fetchone(
"SELECT * FROM access.doors WHERE id = :id", {"id": door_id}, Door
)
async def get_door_by_id_or_name(ident: str) -> Door | None:
"""The reader's `doorId` may be the door id or its human name."""
return await db.fetchone(
"SELECT * FROM access.doors WHERE id = :ident OR name = :ident",
{"ident": ident},
Door,
)
async def get_doors(wallet_ids: list[str]) -> list[Door]:
if not wallet_ids:
return []
q = ",".join(f"'{w}'" for w in wallet_ids)
return await db.fetchall(
f"SELECT * FROM access.doors WHERE wallet IN ({q}) ORDER BY name", model=Door
)
async def delete_door(door_id: str) -> None:
await db.execute("DELETE FROM access.doors WHERE id = :id", {"id": door_id})
await db.execute(
"DELETE FROM access.grants WHERE door_id = :id", {"id": door_id}
)
# ── Grants ─────────────────────────────────────────────────────────────────
async def create_grant(data: CreateGrant) -> Grant:
grant_id = urlsafe_short_hash()
await db.execute(
"""
INSERT INTO access.grants (id, door_id, external_id, label, enabled, expires_at)
VALUES (:id, :door_id, :external_id, :label, :enabled, :expires_at)
""",
{
"id": grant_id,
"door_id": data.door_id,
"external_id": data.external_id.lower(),
"label": data.label,
"enabled": data.enabled,
"expires_at": data.expires_at,
},
)
grant = await get_grant(grant_id)
assert grant, "Newly created grant couldn't be retrieved"
return grant
async def get_grant(grant_id: str) -> Grant | None:
return await db.fetchone(
"SELECT * FROM access.grants WHERE id = :id", {"id": grant_id}, Grant
)
async def get_grants(door_ids: list[str]) -> list[Grant]:
if not door_ids:
return []
q = ",".join(f"'{d}'" for d in door_ids)
return await db.fetchall(
f"SELECT * FROM access.grants WHERE door_id IN ({q}) ORDER BY time DESC",
model=Grant,
)
async def get_active_grant(door_id: str, external_id: str) -> Grant | None:
"""The permission decision: an enabled, unexpired grant for this card+door."""
return await db.fetchone(
"""
SELECT * FROM access.grants
WHERE door_id = :door_id AND external_id = :external_id AND enabled = true
""",
{"door_id": door_id, "external_id": external_id.lower()},
Grant,
)
async def delete_grant(grant_id: str) -> None:
await db.execute("DELETE FROM access.grants WHERE id = :id", {"id": grant_id})
# ── Access log ─────────────────────────────────────────────────────────────
async def record_log(
door_id: str, external_id: str, decision: str, reason: str, ip: str = ""
) -> None:
await db.execute(
"""
INSERT INTO access.logs (id, door_id, external_id, decision, reason, ip)
VALUES (:id, :door_id, :external_id, :decision, :reason, :ip)
""",
{
"id": urlsafe_short_hash(),
"door_id": door_id,
"external_id": external_id,
"decision": decision,
"reason": reason,
"ip": ip,
},
)
async def get_logs(door_ids: list[str], limit: int = 200) -> list[AccessLog]:
if not door_ids:
return []
q = ",".join(f"'{d}'" for d in door_ids)
return await db.fetchall(
f"SELECT * FROM access.logs WHERE door_id IN ({q}) "
f"ORDER BY time DESC LIMIT {int(limit)}",
model=AccessLog,
)

9
manifest.json Normal file
View file

@ -0,0 +1,9 @@
{
"repos": [
{
"id": "access",
"organisation": "aiolabs",
"repository": "access"
}
]
}

45
migrations.py Normal file
View file

@ -0,0 +1,45 @@
async def m001_initial(db):
"""Doors (resources), grants (card→door permissions), and an access log."""
await db.execute(
f"""
CREATE TABLE access.doors (
id TEXT PRIMARY KEY UNIQUE,
wallet TEXT NOT NULL,
name TEXT NOT NULL,
controller_token TEXT NOT NULL,
ha_webhook_url TEXT NOT NULL DEFAULT '',
boltcards_base_url TEXT NOT NULL DEFAULT '',
unlock_timeout_ms INT NOT NULL DEFAULT 2500,
enabled BOOL NOT NULL DEFAULT true,
time TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
);
"""
)
await db.execute(
f"""
CREATE TABLE access.grants (
id TEXT PRIMARY KEY UNIQUE,
door_id TEXT NOT NULL,
external_id TEXT NOT NULL,
label TEXT NOT NULL DEFAULT '',
enabled BOOL NOT NULL DEFAULT true,
expires_at TIMESTAMP,
time TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
);
"""
)
await db.execute(
f"""
CREATE TABLE access.logs (
id TEXT PRIMARY KEY UNIQUE,
door_id TEXT NOT NULL,
external_id TEXT NOT NULL DEFAULT '',
decision TEXT NOT NULL,
reason TEXT NOT NULL DEFAULT '',
ip TEXT NOT NULL DEFAULT '',
time TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
);
"""
)

75
models.py Normal file
View file

@ -0,0 +1,75 @@
from datetime import datetime
from pydantic import BaseModel
class Door(BaseModel):
"""A physical door/resource that a tap can open."""
id: str
wallet: str
name: str
# Secret the reader (Pi) must present in X-Controller-Token to hit /check.
controller_token: str
# Local Home Assistant webhook that fires lock.unlock (LAN-only). When empty,
# /check still authenticates + authorizes and logs, but performs no unlock
# (useful for bring-up / bench testing before the lock is wired).
ha_webhook_url: str
# boltcards API base used to SUN-verify the tapped card, e.g.
# http://localhost:5000/boltcards/api/v1 (same on-prem LNbits). Falls back to
# the instance base URL when empty (see services.verify_card).
boltcards_base_url: str
unlock_timeout_ms: int
enabled: bool
time: datetime
class CreateDoor(BaseModel):
name: str
ha_webhook_url: str = ""
boltcards_base_url: str = ""
# Generated server-side when left blank.
controller_token: str = ""
unlock_timeout_ms: int = 2500
enabled: bool = True
class Grant(BaseModel):
"""A card's permission to open a specific door."""
id: str
door_id: str
# The Bolt Card's boltcards `external_id` (the authenticated identity).
external_id: str
label: str
enabled: bool
# Optional hard expiry; null = no expiry.
expires_at: datetime | None
time: datetime
class CreateGrant(BaseModel):
door_id: str
external_id: str
label: str = ""
expires_at: datetime | None = None
enabled: bool = True
class AccessLog(BaseModel):
id: str
door_id: str
external_id: str
decision: str # "allow" | "deny"
reason: str
ip: str
time: datetime
class CheckRequest(BaseModel):
"""What the door reader (Pi) POSTs to /access/api/v1/check."""
doorId: str
external_id: str
p: str
c: str

56
services.py Normal file
View file

@ -0,0 +1,56 @@
import httpx
from lnbits.settings import settings
from .models import Door
def _boltcards_base(door: Door) -> str:
"""Where to SUN-verify the tapped card.
Defaults to this same on-prem LNbits instance's boltcards extension. The
`access` extension runs in the same process/host, so the loopback base URL
is reachable and keeps verification on the LAN.
"""
if door.boltcards_base_url:
return door.boltcards_base_url.rstrip("/")
return f"{settings.lnbits_baseurl.rstrip('/')}/boltcards/api/v1"
async def verify_card(door: Door, external_id: str, p: str, c: str) -> bool:
"""Authenticate the tap by delegating the NTAG424 SUN check to boltcards.
Calls the boltcards `/verify` endpoint (aiolabs fork 1.1.1-aio.2): a
side-effect-light SUN check that confirms a genuine, non-replayed tap and
returns `{"authenticated": true, ...}` WITHOUT `/scan`'s spend semantics
(no withdrawRequest, no daily-limit, no "hit"). It still advances the SUN
counter server-side, so a captured p/c can't be replayed.
"""
url = f"{_boltcards_base(door)}/verify/{external_id}"
try:
async with httpx.AsyncClient() as client:
resp = await client.get(url, params={"p": p, "c": c}, timeout=5.0)
data = resp.json()
except Exception:
return False
return isinstance(data, dict) and data.get("authenticated") is True
async def trigger_unlock(door: Door, external_id: str) -> bool:
"""Fire the door's local Home Assistant webhook (→ lock.unlock → Z-Wave).
Returns True only on a 2xx. Empty webhook URL means "authorize + log only"
(bring-up mode) and is treated as a successful no-op unlock.
"""
if not door.ha_webhook_url:
return True
timeout = max(door.unlock_timeout_ms, 250) / 1000.0
try:
async with httpx.AsyncClient() as client:
resp = await client.post(
door.ha_webhook_url,
json={"doorId": door.id, "doorName": door.name, "externalId": external_id},
timeout=timeout,
)
return resp.is_success
except Exception:
return False

BIN
static/image/access.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 199 B

141
static/js/index.js Normal file
View file

@ -0,0 +1,141 @@
// LNbits `access` extension admin app.
// Manages doors (resources), grants (card → door permissions), and shows the
// access log. The reader hits /access/api/v1/check with a door controller token;
// this UI is admin-only (wallet admin key).
window.app = Vue.createApp({
el: '#vue',
mixins: [windowMixin],
data() {
return {
doors: [],
grants: [],
logs: [],
doorCols: [
{ name: 'name', label: 'Name', field: 'name' },
{ name: 'id', label: 'ID', field: 'id' },
{ name: 'enabled', label: 'State', field: 'enabled' },
{ name: 'actions', label: '', field: 'actions' }
],
grantCols: [
{ name: 'door', label: 'Door', field: 'door_id' },
{ name: 'external_id', label: 'Card', field: 'external_id' },
{ name: 'label', label: 'Label', field: 'label' },
{ name: 'actions', label: '', field: 'actions' }
],
logCols: [
{ name: 'time', label: 'Time', field: 'time' },
{ name: 'door', label: 'Door', field: 'door_id' },
{ name: 'decision', label: 'Decision', field: 'decision' },
{ name: 'reason', label: 'Reason', field: 'reason' }
],
doorDialog: { show: false, data: { unlock_timeout_ms: 2500 } },
grantDialog: { show: false, data: {} }
}
},
computed: {
adminkey() {
return this.g.user.wallets[0].adminkey
},
inkey() {
return this.g.user.wallets[0].inkey
},
doorOptions() {
return this.doors.map(d => ({ label: d.name, value: d.id }))
}
},
methods: {
doorName(id) {
const d = this.doors.find(x => x.id === id)
return d ? d.name : id
},
async loadAll() {
await Promise.all([this.loadDoors(), this.loadGrants(), this.loadLogs()])
},
async loadDoors() {
const { data } = await LNbits.api.request(
'GET',
'/access/api/v1/doors',
this.inkey
)
this.doors = data
},
async loadGrants() {
const { data } = await LNbits.api.request(
'GET',
'/access/api/v1/grants',
this.inkey
)
this.grants = data
},
async loadLogs() {
const { data } = await LNbits.api.request(
'GET',
'/access/api/v1/logs',
this.inkey
)
this.logs = data
},
async createDoor() {
try {
await LNbits.api.request(
'POST',
'/access/api/v1/doors',
this.adminkey,
this.doorDialog.data
)
this.doorDialog = { show: false, data: { unlock_timeout_ms: 2500 } }
await this.loadDoors()
} catch (e) {
LNbits.utils.notifyApiError(e)
}
},
async deleteDoor(id) {
try {
await LNbits.api.request(
'DELETE',
`/access/api/v1/doors/${id}`,
this.adminkey
)
await this.loadAll()
} catch (e) {
LNbits.utils.notifyApiError(e)
}
},
showToken(door) {
this.$q.dialog({
title: `${door.name} — controller token`,
message: `Set this as X-Controller-Token on the reader for doorId "${door.id}":<br><br><code>${door.controller_token}</code>`,
html: true
})
},
async createGrant() {
try {
await LNbits.api.request(
'POST',
'/access/api/v1/grants',
this.adminkey,
this.grantDialog.data
)
this.grantDialog = { show: false, data: {} }
await this.loadGrants()
} catch (e) {
LNbits.utils.notifyApiError(e)
}
},
async deleteGrant(id) {
try {
await LNbits.api.request(
'DELETE',
`/access/api/v1/grants/${id}`,
this.adminkey
)
await this.loadGrants()
} catch (e) {
LNbits.utils.notifyApiError(e)
}
}
},
created() {
this.loadAll()
}
})

203
templates/access/index.html Normal file
View file

@ -0,0 +1,203 @@
{% extends "base.html" %} {% from "macros.jinja" import window_vars with context
%} {% block scripts %} {{ window_vars(user) }}
<script src="{{ static_url_for('access/static', 'js/index.js') }}"></script>
{% endblock %} {% block page %}
<div class="row q-col-gutter-md">
<div class="col-12 col-md-7 q-gutter-y-md">
<q-card>
<q-card-section class="row items-center justify-between">
<h5 class="text-subtitle1 q-my-none">Doors</h5>
<q-btn unelevated color="primary" @click="doorDialog.show = true"
>New door</q-btn
>
</q-card-section>
<q-card-section class="q-pa-none">
<q-table
flat
dense
:rows="doors"
row-key="id"
:columns="doorCols"
:pagination="{rowsPerPage: 0}"
hide-bottom
>
<template v-slot:body="props">
<q-tr :props="props">
<q-td>{% raw %}{{ props.row.name }}{% endraw %}</q-td>
<q-td class="text-caption"
>{% raw %}{{ props.row.id }}{% endraw %}</q-td
>
<q-td>
<q-badge :color="props.row.enabled ? 'green' : 'grey'"
>{% raw %}{{ props.row.enabled ? 'on' : 'off' }}{% endraw
%}</q-badge
>
</q-td>
<q-td>
<q-btn
flat
dense
size="sm"
icon="vpn_key"
@click="showToken(props.row)"
><q-tooltip>Controller token</q-tooltip></q-btn
>
<q-btn
flat
dense
size="sm"
color="negative"
icon="delete"
@click="deleteDoor(props.row.id)"
></q-btn>
</q-td>
</q-tr>
</template>
</q-table>
</q-card-section>
</q-card>
<q-card>
<q-card-section class="row items-center justify-between">
<h5 class="text-subtitle1 q-my-none">Grants (card → door)</h5>
<q-btn unelevated color="primary" @click="grantDialog.show = true"
>New grant</q-btn
>
</q-card-section>
<q-card-section class="q-pa-none">
<q-table
flat
dense
:rows="grants"
row-key="id"
:columns="grantCols"
:pagination="{rowsPerPage: 0}"
hide-bottom
>
<template v-slot:body="props">
<q-tr :props="props">
<q-td>{% raw %}{{ doorName(props.row.door_id) }}{% endraw %}</q-td>
<q-td class="text-caption"
>{% raw %}{{ props.row.external_id }}{% endraw %}</q-td
>
<q-td>{% raw %}{{ props.row.label }}{% endraw %}</q-td>
<q-td>
<q-btn
flat
dense
size="sm"
color="negative"
icon="delete"
@click="deleteGrant(props.row.id)"
></q-btn>
</q-td>
</q-tr>
</template>
</q-table>
</q-card-section>
</q-card>
</div>
<div class="col-12 col-md-5 q-gutter-y-md">
<q-card>
<q-card-section>
<h6 class="text-subtitle1 q-my-none">Access log</h6>
</q-card-section>
<q-card-section class="q-pa-none">
<q-table
flat
dense
:rows="logs"
row-key="id"
:columns="logCols"
:pagination="{rowsPerPage: 15}"
>
<template v-slot:body="props">
<q-tr :props="props">
<q-td class="text-caption"
>{% raw %}{{ props.row.time }}{% endraw %}</q-td
>
<q-td>{% raw %}{{ doorName(props.row.door_id) }}{% endraw %}</q-td>
<q-td>
<q-badge
:color="props.row.decision === 'allow' ? 'green' : 'red'"
>{% raw %}{{ props.row.decision }}{% endraw %}</q-badge
>
</q-td>
<q-td class="text-caption"
>{% raw %}{{ props.row.reason }}{% endraw %}</q-td
>
</q-tr>
</template>
</q-table>
</q-card-section>
</q-card>
</div>
<q-dialog v-model="doorDialog.show">
<q-card class="q-pa-lg" style="min-width: 420px">
<h6 class="q-mt-none">New door</h6>
<q-input v-model="doorDialog.data.name" label="Name" filled dense />
<q-input
v-model="doorDialog.data.ha_webhook_url"
label="Home Assistant webhook URL (local)"
filled
dense
class="q-mt-sm"
/>
<q-input
v-model="doorDialog.data.boltcards_base_url"
label="boltcards API base (blank = this instance)"
filled
dense
class="q-mt-sm"
/>
<q-input
v-model.number="doorDialog.data.unlock_timeout_ms"
type="number"
label="Unlock timeout (ms)"
filled
dense
class="q-mt-sm"
/>
<div class="row q-mt-md q-gutter-sm justify-end">
<q-btn flat v-close-popup>Cancel</q-btn>
<q-btn unelevated color="primary" @click="createDoor">Create</q-btn>
</div>
</q-card>
</q-dialog>
<q-dialog v-model="grantDialog.show">
<q-card class="q-pa-lg" style="min-width: 420px">
<h6 class="q-mt-none">New grant</h6>
<q-select
v-model="grantDialog.data.door_id"
:options="doorOptions"
label="Door"
filled
dense
emit-value
map-options
/>
<q-input
v-model="grantDialog.data.external_id"
label="Card external_id (from boltcards)"
filled
dense
class="q-mt-sm"
/>
<q-input
v-model="grantDialog.data.label"
label="Label (optional)"
filled
dense
class="q-mt-sm"
/>
<div class="row q-mt-md q-gutter-sm justify-end">
<q-btn flat v-close-popup>Cancel</q-btn>
<q-btn unelevated color="primary" @click="createGrant">Create</q-btn>
</div>
</q-card>
</q-dialog>
</div>
{% endblock %}

18
views.py Normal file
View file

@ -0,0 +1,18 @@
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse
from lnbits.core.models import User
from lnbits.decorators import check_user_exists
from lnbits.helpers import template_renderer
access_generic_router = APIRouter()
def access_renderer():
return template_renderer(["access/templates"])
@access_generic_router.get("/", response_class=HTMLResponse)
async def index(request: Request, user: User = Depends(check_user_exists)):
return access_renderer().TemplateResponse(
"access/index.html", {"request": request, "user": user.json()}
)

118
views_api.py Normal file
View file

@ -0,0 +1,118 @@
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])

54
views_reader.py Normal file
View file

@ -0,0 +1,54 @@
from fastapi import APIRouter, Request
from .crud import get_active_grant, get_door_by_id_or_name, record_log
from .models import CheckRequest
from .services import trigger_unlock, verify_card
access_reader_router = APIRouter()
def _client_ip(request: Request) -> str:
if "x-real-ip" in request.headers:
return request.headers["x-real-ip"]
if "x-forwarded-for" in request.headers:
return request.headers["x-forwarded-for"]
return request.client.host if request.client else ""
# The door reader (Pi + PN532) calls this. Fails closed at every step.
# POST /access/api/v1/check
# X-Controller-Token: <door.controller_token>
# { "doorId": "...", "external_id": "...", "p": "...", "c": "..." }
@access_reader_router.post("/api/v1/check")
async def check(data: CheckRequest, request: Request):
ip = _client_ip(request)
door = await get_door_by_id_or_name(data.doorId)
# Unknown/disabled door, or wrong controller token → deny (no card read logged
# against a real door we can't identify; log against the requested id).
if not door or not door.enabled:
await record_log(data.doorId, data.external_id, "deny", "door_unknown", ip)
return {"allow": False, "reason": "door_unknown"}
token = request.headers.get("x-controller-token", "")
if not token or token != door.controller_token:
await record_log(door.id, data.external_id, "deny", "bad_controller_token", ip)
return {"allow": False, "reason": "bad_controller_token"}
# 1) Authenticate the card (NTAG424 SUN via boltcards).
if not await verify_card(door, data.external_id, data.p, data.c):
await record_log(door.id, data.external_id, "deny", "card_invalid", ip)
return {"allow": False, "reason": "card_invalid"}
# 2) Authorize: does this card have an active grant on this door?
grant = await get_active_grant(door.id, data.external_id)
if not grant:
await record_log(door.id, data.external_id, "deny", "not_authorized", ip)
return {"allow": False, "reason": "not_authorized"}
# 3) Actuate: fire the Home Assistant unlock webhook.
unlocked = await trigger_unlock(door, data.external_id)
decision = "allow" if unlocked else "deny"
reason = "unlocked" if unlocked else "unlock_failed"
await record_log(door.id, data.external_id, decision, reason, ip)
return {"allow": unlocked, "reason": reason}