lnbits-exts/castle
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases 2 Packages Wiki Activity Actions

Adds account permissioning system

Browse source 
Adds an account permissioning system to allow granular control over account access.

Introduces the ability to grant users specific permissions (read, submit_expense, manage) on individual accounts.  This includes support for hierarchical permission inheritance, where permissions on parent accounts cascade to child accounts.

Adds new API endpoints for managing account permissions, including granting, listing, and revoking permissions.

Integrates permission checks into existing endpoints, such as creating expense entries, to ensure that users only have access to the accounts they are authorized to use.

Fixes #33 - Implements role based access control
This commit is contained in:
padreug 2025-11-07 17:55:59 +01:00
parent 7f9cecefa1
commit 92c1649f3b
4 changed files with 617 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

221
crud.py
View file

@ -7,19 +7,24 @@ from lnbits.helpers import urlsafe_short_hash
from .models import (
Account,
AccountPermission,
AccountType,
AssertionStatus,
BalanceAssertion,
CastleSettings,
CreateAccount,
CreateAccountPermission,
CreateBalanceAssertion,
CreateEntryLine,
CreateJournalEntry,
CreateUserEquityStatus,
EntryLine,
JournalEntry,
PermissionType,
StoredUserWalletSettings,
UserBalance,
UserCastleSettings,
UserEquityStatus,
UserWalletSettings,
)
@ -1062,3 +1067,219 @@ async def get_all_equity_eligible_users() -> list["UserEquityStatus"]:
)
return [UserEquityStatus(**row) for row in rows]
# ===== ACCOUNT PERMISSION OPERATIONS =====
async def create_account_permission(
data: "CreateAccountPermission", granted_by: str
) -> "AccountPermission":
"""Create a new account permission"""
from .models import AccountPermission
permission_id = urlsafe_short_hash()
permission = AccountPermission(
id=permission_id,
user_id=data.user_id,
account_id=data.account_id,
permission_type=data.permission_type,
granted_by=granted_by,
granted_at=datetime.now(),
expires_at=data.expires_at,
notes=data.notes,
)
await db.execute(
"""
INSERT INTO account_permissions (
id, user_id, account_id, permission_type, granted_by,
granted_at, expires_at, notes
)
VALUES (
:id, :user_id, :account_id, :permission_type, :granted_by,
:granted_at, :expires_at, :notes
)
""",
{
"id": permission.id,
"user_id": permission.user_id,
"account_id": permission.account_id,
"permission_type": permission.permission_type.value,
"granted_by": permission.granted_by,
"granted_at": permission.granted_at,
"expires_at": permission.expires_at,
"notes": permission.notes,
},
)
return permission
async def get_account_permission(permission_id: str) -> Optional["AccountPermission"]:
"""Get account permission by ID"""
from .models import AccountPermission, PermissionType
row = await db.fetchone(
"SELECT * FROM account_permissions WHERE id = :id",
{"id": permission_id},
)
if not row:
return None
return AccountPermission(
id=row["id"],
user_id=row["user_id"],
account_id=row["account_id"],
permission_type=PermissionType(row["permission_type"]),
granted_by=row["granted_by"],
granted_at=row["granted_at"],
expires_at=row["expires_at"],
notes=row["notes"],
)
async def get_user_permissions(
user_id: str, permission_type: Optional["PermissionType"] = None
) -> list["AccountPermission"]:
"""Get all permissions for a specific user"""
from .models import AccountPermission, PermissionType
if permission_type:
rows = await db.fetchall(
"""
SELECT * FROM account_permissions
WHERE user_id = :user_id
AND permission_type = :permission_type
AND (expires_at IS NULL OR expires_at > :now)
ORDER BY granted_at DESC
""",
{
"user_id": user_id,
"permission_type": permission_type.value,
"now": datetime.now(),
},
)
else:
rows = await db.fetchall(
"""
SELECT * FROM account_permissions
WHERE user_id = :user_id
AND (expires_at IS NULL OR expires_at > :now)
ORDER BY granted_at DESC
""",
{"user_id": user_id, "now": datetime.now()},
)
return [
AccountPermission(
id=row["id"],
user_id=row["user_id"],
account_id=row["account_id"],
permission_type=PermissionType(row["permission_type"]),
granted_by=row["granted_by"],
granted_at=row["granted_at"],
expires_at=row["expires_at"],
notes=row["notes"],
)
for row in rows
]
async def get_account_permissions(account_id: str) -> list["AccountPermission"]:
"""Get all permissions for a specific account"""
from .models import AccountPermission, PermissionType
rows = await db.fetchall(
"""
SELECT * FROM account_permissions
WHERE account_id = :account_id
AND (expires_at IS NULL OR expires_at > :now)
ORDER BY granted_at DESC
""",
{"account_id": account_id, "now": datetime.now()},
)
return [
AccountPermission(
id=row["id"],
user_id=row["user_id"],
account_id=row["account_id"],
permission_type=PermissionType(row["permission_type"]),
granted_by=row["granted_by"],
granted_at=row["granted_at"],
expires_at=row["expires_at"],
notes=row["notes"],
)
for row in rows
]
async def delete_account_permission(permission_id: str) -> None:
"""Delete (revoke) an account permission"""
await db.execute(
"DELETE FROM account_permissions WHERE id = :id",
{"id": permission_id},
)
async def check_user_has_permission(
user_id: str, account_id: str, permission_type: "PermissionType"
) -> bool:
"""Check if user has a specific permission on an account (direct permission only, no inheritance)"""
row = await db.fetchone(
"""
SELECT id FROM account_permissions
WHERE user_id = :user_id
AND account_id = :account_id
AND permission_type = :permission_type
AND (expires_at IS NULL OR expires_at > :now)
""",
{
"user_id": user_id,
"account_id": account_id,
"permission_type": permission_type.value,
"now": datetime.now(),
},
)
return row is not None
async def get_user_permissions_with_inheritance(
user_id: str, account_name: str, permission_type: "PermissionType"
) -> list[tuple["AccountPermission", Optional[str]]]:
"""
Get all permissions for a user on an account, including inherited permissions from parent accounts.
Returns list of tuples: (permission, parent_account_name or None)
Example:
If user has permission on "Expenses:Food", they also have permission on "Expenses:Food:Groceries"
Returns: [(permission_on_food, "Expenses:Food")]
"""
from .models import AccountPermission, PermissionType
# Get all user's permissions of this type
user_permissions = await get_user_permissions(user_id, permission_type)
# Find which permissions apply to this account (direct or inherited)
applicable_permissions = []
for perm in user_permissions:
# Get the account for this permission
account = await get_account(perm.account_id)
if not account:
continue
# Check if this account is a parent of the target account
# Parent accounts are indicated by hierarchical names (colon-separated)
# e.g., "Expenses:Food" is parent of "Expenses:Food:Groceries"
if account_name == account.name:
# Direct permission
applicable_permissions.append((perm, None))
elif account_name.startswith(account.name + ":"):
# Inherited permission from parent account
applicable_permissions.append((perm, account.name))
return applicable_permissions

61
migrations.py
View file

@ -391,3 +391,64 @@ async def m010_user_equity_status(db):
WHERE is_equity_eligible = TRUE;
"""
)
async def m011_account_permissions(db):
"""
Create account_permissions table for granular account access control.
Allows admins to grant specific permissions (read, submit_expense, manage) to users for specific accounts.
Supports hierarchical permission inheritance (permissions on parent accounts cascade to children).
"""
await db.execute(
f"""
CREATE TABLE account_permissions (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
account_id TEXT NOT NULL,
permission_type TEXT NOT NULL,
granted_by TEXT NOT NULL,
granted_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
expires_at TIMESTAMP,
notes TEXT,
FOREIGN KEY (account_id) REFERENCES accounts (id)
);
"""
)
# Index for looking up permissions by user
await db.execute(
"""
CREATE INDEX idx_account_permissions_user_id ON account_permissions (user_id);
"""
)
# Index for looking up permissions by account
await db.execute(
"""
CREATE INDEX idx_account_permissions_account_id ON account_permissions (account_id);
"""
)
# Composite index for checking specific user+account permissions
await db.execute(
"""
CREATE INDEX idx_account_permissions_user_account
ON account_permissions (user_id, account_id);
"""
)
# Index for finding permissions by type
await db.execute(
"""
CREATE INDEX idx_account_permissions_type ON account_permissions (permission_type);
"""
)
# Index for finding expired permissions
await db.execute(
"""
CREATE INDEX idx_account_permissions_expires
ON account_permissions (expires_at)
WHERE expires_at IS NOT NULL;
"""
)

45
models.py
View file

@ -273,3 +273,48 @@ class UserInfo(BaseModel):
user_id: str
is_equity_eligible: bool
equity_account_name: Optional[str] = None
class PermissionType(str, Enum):
"""Types of permissions for account access"""
READ = "read" # Can view account and its balance
SUBMIT_EXPENSE = "submit_expense" # Can submit expenses to this account
MANAGE = "manage" # Can modify account (admin level)
class AccountPermission(BaseModel):
"""Defines which accounts a user can access"""
id: str # Unique permission ID
user_id: str # User's wallet ID (from invoice key)
account_id: str # Account ID from accounts table
permission_type: PermissionType
granted_by: str # Admin user ID who granted permission
granted_at: datetime
expires_at: Optional[datetime] = None # Optional expiration
notes: Optional[str] = None # Admin notes about this permission
class CreateAccountPermission(BaseModel):
"""Create account permission"""
user_id: str
account_id: str
permission_type: PermissionType
expires_at: Optional[datetime] = None
notes: Optional[str] = None
class AccountWithPermissions(BaseModel):
"""Account with user-specific permission metadata"""
id: str
name: str
account_type: AccountType
description: Optional[str] = None
user_id: Optional[str] = None
created_at: datetime
# Only included when filter_by_user=true
user_permissions: Optional[list[PermissionType]] = None
inherited_from: Optional[str] = None # Parent account ID if inherited
# Hierarchical structure
parent_account: Optional[str] = None # Parent account name
level: Optional[int] = None # Depth in hierarchy (0 = top level)
has_children: Optional[bool] = None # Whether this account has sub-accounts

293
views_api.py
View file

@ -16,14 +16,18 @@ from .crud import (
approve_manual_payment_request,
check_balance_assertion,
create_account,
create_account_permission,
create_balance_assertion,
create_journal_entry,
create_manual_payment_request,
db,
delete_account_permission,
delete_balance_assertion,
get_account,
get_account_balance,
get_account_by_name,
get_account_permission,
get_account_permissions,
get_account_transactions,
get_all_accounts,
get_all_journal_entries,
@ -38,15 +42,20 @@ from .crud import (
get_or_create_user_account,
get_user_balance,
get_user_manual_payment_requests,
get_user_permissions,
get_user_permissions_with_inheritance,
reject_manual_payment_request,
)
from .models import (
Account,
AccountPermission,
AccountType,
AccountWithPermissions,
AssertionStatus,
BalanceAssertion,
CastleSettings,
CreateAccount,
CreateAccountPermission,
CreateBalanceAssertion,
CreateEntryLine,
CreateJournalEntry,
@ -57,6 +66,7 @@ from .models import (
JournalEntryFlag,
ManualPaymentRequest,
PayUser,
PermissionType,
ReceivableEntry,
RecordPayment,
RevenueEntry,
@ -120,9 +130,84 @@ async def api_get_currencies() -> list[str]:
@castle_api_router.get("/api/v1/accounts")
async def api_get_accounts() -> list[Account]:
"""Get all accounts in the chart of accounts"""
return await get_all_accounts()
async def api_get_accounts(
filter_by_user: bool = False,
wallet: WalletTypeInfo = Depends(require_invoice_key),
) -> list[Account] | list[AccountWithPermissions]:
"""
Get all accounts in the chart of accounts.
- filter_by_user: If true, only return accounts the user has permissions for
- Returns AccountWithPermissions objects when filter_by_user=true, otherwise Account objects
"""
all_accounts = await get_all_accounts()
if not filter_by_user:
# Return all accounts without filtering
return all_accounts
# Filter by user permissions
user_id = wallet.wallet.user
user_permissions = await get_user_permissions(user_id)
# Get set of account IDs the user has any permission on
permitted_account_ids = {perm.account_id for perm in user_permissions}
# Build list of accounts with permission metadata
accounts_with_permissions = []
for account in all_accounts:
# Check if user has direct permission on this account
account_perms = [
perm for perm in user_permissions if perm.account_id == account.id
]
# Check if user has inherited permission from parent account
inherited_perms = await get_user_permissions_with_inheritance(
user_id, account.name, PermissionType.READ
)
# Determine if account should be included
has_access = bool(account_perms) or bool(inherited_perms)
if has_access:
# Parse hierarchical account name to get parent and level
parts = account.name.split(":")
level = len(parts) - 1
parent_account = ":".join(parts[:-1]) if level > 0 else None
# Determine inherited_from (which parent account gave access)
inherited_from = None
if inherited_perms and not account_perms:
# Permission is inherited, use the parent account name
_, parent_name = inherited_perms[0]
inherited_from = parent_name
# Collect permission types for this account
permission_types = [perm.permission_type for perm in account_perms]
# Check if account has children
has_children = any(
a.name.startswith(account.name + ":") for a in all_accounts
)
accounts_with_permissions.append(
AccountWithPermissions(
id=account.id,
name=account.name,
account_type=account.account_type,
description=account.description,
user_id=account.user_id,
created_at=account.created_at,
user_permissions=permission_types if permission_types else None,
inherited_from=inherited_from,
parent_account=parent_account,
level=level,
has_children=has_children,
)
)
return accounts_with_permissions
@castle_api_router.post("/api/v1/accounts", status_code=HTTPStatus.CREATED)
@ -285,6 +370,19 @@ async def api_create_expense_entry(
detail=f"Expense account '{data.expense_account}' not found",
)
# Validate user has permission to submit expenses to this account
from .crud import get_user_permissions_with_inheritance
submit_perms = await get_user_permissions_with_inheritance(
wallet.wallet.user, expense_account.name, PermissionType.SUBMIT_EXPENSE
)
if not submit_perms:
raise HTTPException(
status_code=HTTPStatus.FORBIDDEN,
detail=f"You do not have permission to submit expenses to account '{expense_account.name}'. Please contact an administrator to request access.",
)
# Get or create user-specific account
if data.is_equity:
# Validate equity eligibility
@ -1898,3 +1996,192 @@ async def api_list_equity_eligible_users(
from .crud import get_all_equity_eligible_users
return await get_all_equity_eligible_users()
# ===== ACCOUNT PERMISSION ADMIN ENDPOINTS =====
@castle_api_router.post("/api/v1/admin/permissions", status_code=HTTPStatus.CREATED)
async def api_grant_permission(
data: CreateAccountPermission,
wallet: WalletTypeInfo = Depends(require_admin_key),
) -> AccountPermission:
"""Grant account permission to a user (admin only)"""
# Validate that account exists
account = await get_account(data.account_id)
if not account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND,
detail=f"Account with ID '{data.account_id}' not found",
)
return await create_account_permission(data, wallet.wallet.user)
@castle_api_router.get("/api/v1/admin/permissions")
async def api_list_permissions(
user_id: str | None = None,
account_id: str | None = None,
wallet: WalletTypeInfo = Depends(require_admin_key),
) -> list[AccountPermission]:
"""
List account permissions (admin only).
Can filter by user_id or account_id.
"""
if user_id:
return await get_user_permissions(user_id)
elif account_id:
return await get_account_permissions(account_id)
else:
# Get all permissions (get all users' permissions)
# This is a bit inefficient but works for admin overview
all_accounts = await get_all_accounts()
all_permissions = []
for account in all_accounts:
account_perms = await get_account_permissions(account.id)
all_permissions.extend(account_perms)
# Deduplicate by permission ID
seen_ids = set()
unique_permissions = []
for perm in all_permissions:
if perm.id not in seen_ids:
seen_ids.add(perm.id)
unique_permissions.append(perm)
return unique_permissions
@castle_api_router.delete("/api/v1/admin/permissions/{permission_id}")
async def api_revoke_permission(
permission_id: str,
wallet: WalletTypeInfo = Depends(require_admin_key),
) -> dict:
"""Revoke (delete) an account permission (admin only)"""
# Verify permission exists
permission = await get_account_permission(permission_id)
if not permission:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND,
detail=f"Permission with ID '{permission_id}' not found",
)
await delete_account_permission(permission_id)
return {
"success": True,
"message": f"Permission {permission_id} revoked successfully",
}
@castle_api_router.post("/api/v1/admin/permissions/bulk", status_code=HTTPStatus.CREATED)
async def api_bulk_grant_permissions(
permissions: list[CreateAccountPermission],
wallet: WalletTypeInfo = Depends(require_admin_key),
) -> list[AccountPermission]:
"""Grant multiple account permissions at once (admin only)"""
created_permissions = []
for perm_data in permissions:
# Validate that account exists
account = await get_account(perm_data.account_id)
if not account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND,
detail=f"Account with ID '{perm_data.account_id}' not found",
)
perm = await create_account_permission(perm_data, wallet.wallet.user)
created_permissions.append(perm)
return created_permissions
# ===== USER PERMISSION ENDPOINTS =====
@castle_api_router.get("/api/v1/users/me/permissions")
async def api_get_user_permissions(
wallet: WalletTypeInfo = Depends(require_invoice_key),
) -> list[AccountPermission]:
"""Get current user's account permissions"""
return await get_user_permissions(wallet.wallet.user)
# ===== ACCOUNT HIERARCHY ENDPOINT =====
@castle_api_router.get("/api/v1/accounts/hierarchy")
async def api_get_account_hierarchy(
root_account: str | None = None,
wallet: WalletTypeInfo = Depends(require_invoice_key),
) -> list[AccountWithPermissions]:
"""
Get hierarchical account structure with user permissions.
Optionally filter by root account (e.g., "Expenses" to get all expense sub-accounts).
"""
all_accounts = await get_all_accounts()
user_id = wallet.wallet.user
user_permissions = await get_user_permissions(user_id)
# Filter by root account if specified
if root_account:
all_accounts = [
acc for acc in all_accounts
if acc.name == root_account or acc.name.startswith(root_account + ":")
]
# Build hierarchy with permission metadata
accounts_with_hierarchy = []
for account in all_accounts:
# Check if user has direct permission on this account
account_perms = [
perm for perm in user_permissions if perm.account_id == account.id
]
# Check if user has inherited permission from parent account
inherited_perms = await get_user_permissions_with_inheritance(
user_id, account.name, PermissionType.READ
)
# Parse hierarchical account name to get parent and level
parts = account.name.split(":")
level = len(parts) - 1
parent_account = ":".join(parts[:-1]) if level > 0 else None
# Determine inherited_from (which parent account gave access)
inherited_from = None
if inherited_perms and not account_perms:
# Permission is inherited, use the parent account name
_, parent_name = inherited_perms[0]
inherited_from = parent_name
# Collect permission types for this account
permission_types = [perm.permission_type for perm in account_perms]
# Check if account has children
has_children = any(
a.name.startswith(account.name + ":") for a in all_accounts
)
accounts_with_hierarchy.append(
AccountWithPermissions(
id=account.id,
name=account.name,
account_type=account.account_type,
description=account.description,
user_id=account.user_id,
created_at=account.created_at,
user_permissions=permission_types if permission_types else None,
inherited_from=inherited_from,
parent_account=parent_account,
level=level,
has_children=has_children,
)
)
# Sort by hierarchical name for natural ordering
accounts_with_hierarchy.sort(key=lambda a: a.name)
return accounts_with_hierarchy