add superuser class to obscure superuser_id for admin in the ui

This commit is contained in:
dni ⚡
2022-12-07 11:00:48 +01:00
parent 02e82a8fdf
commit c189ec49e5
5 changed files with 29 additions and 11 deletions

View File

@@ -63,6 +63,7 @@ class User(BaseModel):
wallets: List[Wallet] = [] wallets: List[Wallet] = []
password: Optional[str] = None password: Optional[str] = None
admin: bool = False admin: bool = False
super_user: bool = False
@property @property
def wallet_ids(self) -> List[str]: def wallet_ids(self) -> List[str]:

View File

@@ -6,7 +6,7 @@ from lnbits.settings import readonly_variables, settings
from lnbits.tasks import internal_invoice_queue from lnbits.tasks import internal_invoice_queue
from . import db from . import db
from .models import AdminSettings, UpdateSettings from .models import SuperSettings, AdminSettings, UpdateSettings
async def update_wallet_balance(wallet_id: str, amount: int): async def update_wallet_balance(wallet_id: str, amount: int):
@@ -25,13 +25,24 @@ async def update_wallet_balance(wallet_id: str, amount: int):
return payment return payment
async def get_admin_settings() -> Optional[AdminSettings]: async def get_super_settings() -> Optional[SuperSettings]:
row = await db.fetchone("SELECT * FROM admin.settings") row = await db.fetchone("SELECT * FROM admin.settings")
if not row: if not row:
return None return None
return AdminSettings( return SuperSettings(**row)
lnbits_allowed_funding_sources=settings.lnbits_allowed_funding_sources, **row
async def get_admin_settings(is_super_user: bool = False) -> Optional[AdminSettings]:
sets = await get_super_settings()
if not sets:
return None
row_dict = dict(sets)
row_dict.pop("super_user")
admin_settings = AdminSettings(
super_user=is_super_user, lnbits_allowed_funding_sources=settings.lnbits_allowed_funding_sources, **row_dict
) )
return admin_settings
async def delete_admin_settings(): async def delete_admin_settings():
@@ -46,6 +57,7 @@ async def update_admin_settings(data: UpdateSettings):
def get_q_and_values(data): def get_q_and_values(data):
keys = [] keys = []
values = [] values = []
# exclude from api updates
data.pop("lnbits_allowed_funding_sources") data.pop("lnbits_allowed_funding_sources")
data.pop("super_user") data.pop("super_user")
for key, value in data.items(): for key, value in data.items():

View File

@@ -79,6 +79,10 @@ class UpdateSettings(BaseModel, extra=Extra.forbid):
boltz_url: str = Query(None) boltz_url: str = Query(None)
class SuperSettings(UpdateSettings):
super_user: str
class AdminSettings(UpdateSettings): class AdminSettings(UpdateSettings):
super_user: bool
lnbits_allowed_funding_sources: Optional[List[str]] lnbits_allowed_funding_sources: Optional[List[str]]
super_user: Optional[bool]

View File

@@ -29,9 +29,10 @@ async def api_restart_server() -> dict[str, str]:
@admin_ext.get("/api/v1/settings/") @admin_ext.get("/api/v1/settings/")
async def api_get_settings(user: User = Depends(check_admin)) -> Optional[AdminSettings]: async def api_get_settings(
admin_settings = await get_admin_settings() user: User = Depends(check_admin) #type: ignore
admin_settings.super_user = user.super_user ) -> Optional[AdminSettings]:
admin_settings = await get_admin_settings(user.super_user)
return admin_settings return admin_settings

View File

@@ -207,10 +207,10 @@ async def check_admin_settings():
# if not imported here, circular import error # if not imported here, circular import error
from lnbits.extensions.admin.crud import ( from lnbits.extensions.admin.crud import (
create_admin_settings, create_admin_settings,
get_admin_settings, get_super_settings,
) )
sets = await get_admin_settings() sets = await get_super_settings()
if not sets: if not sets:
# create new settings if table is empty # create new settings if table is empty
logger.warning( logger.warning(
@@ -218,7 +218,7 @@ async def check_admin_settings():
) )
await create_admin_settings() await create_admin_settings()
logger.warning("initialized admin.settings from enviroment variables.") logger.warning("initialized admin.settings from enviroment variables.")
sets = await get_admin_settings() sets = await get_super_settings()
if sets: if sets:
for key, value in sets.dict().items(): for key, value in sets.dict().items():