mgrotke/linuxrouter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Development

Browse source
This commit is contained in:
Matthew Grotke 2026-06-10 14:23:47 -04:00
parent f5722f3c7b
commit d60bf15ce4
15 changed files with 367 additions and 285 deletions
Download patch file Download diff file Expand all files Collapse all files

114
docker/routlin-dash/app/pages/accountmanage/action.py
View file

@ -1,6 +1,6 @@
from pathlib import Path
from flask import Blueprint, request, session, redirect, flash
import json, os, re, sqlite3
import os, re, sqlite3
from datetime import datetime, timezone
import auth
import config_utils
@ -10,24 +10,31 @@ _PAGE = Path(__file__).parent.name
bp = Blueprint(_PAGE, __name__)
VALID_LEVELS = {'viewer', 'administrator', 'manager'}
VALID_LEVELS = {'viewer': 1, 'administrator': 2, 'manager': 3}
def _load_accounts():
@bp.route('/action/accountmanage/session_invalidate', methods=['POST'])
@auth.require_level('manager')
def session_invalidate():
sid = request.form.get('session_id', '').strip()
if not sid:
flash('Invalid request.', 'error')
return redirect(f'/{_PAGE}')
try:
with open(config_utils.ACCOUNTS_FILE) as f:
return json.load(f)
con = config_utils.open_accounts_db()
con.execute('DELETE FROM sessions WHERE session_id=?', (sid,))
con.commit()
con.close()
flash('Session invalidated.', 'success')
except Exception:
return {'accounts': []}
def _save_accounts(data):
with open(config_utils.ACCOUNTS_FILE, 'w') as f:
json.dump(data, f, indent=2)
flash('Failed to invalidate session.', 'error')
return redirect(f'/{_PAGE}')
@bp.route('/action/accountmanage/accounts_add', methods=['POST'])
@auth.require_level('manager')
def accounts_add():
import uuid as _uuid, time as _t
email = sanitize.email(request.form.get('email_address', ''))
access_level = request.form.get('access_level', '').strip()
@ -43,26 +50,24 @@ def accounts_add():
flash('Invalid access level.', 'error')
return redirect(f'/{_PAGE}')
data = _load_accounts()
accounts = data.get('accounts', [])
if any(a.get('email_address', '').lower() == email for a in accounts):
if config_utils.get_account_by_email(email):
flash('An account with that email address already exists.', 'error')
return redirect(f'/{_PAGE}')
now = datetime.now(tz=timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
accounts.append({
'email_address': email,
'access_level': access_level,
'account_created_utc': now,
'account_created_by': session.get('email_address', ''),
'hashed_password': '',
'timezone': '',
})
data['accounts'] = accounts
_save_accounts(data)
try:
con = config_utils.open_accounts_db()
con.execute(
'INSERT INTO accounts(account_id,email,access_level,created_ts,created_by) VALUES(?,?,?,?,?)',
(str(_uuid.uuid4()), email, VALID_LEVELS[access_level], int(_t.time()),
session.get('email_address', ''))
)
con.commit()
con.close()
except Exception as exc:
flash(f'Could not add account: {exc}', 'error')
return redirect(f'/{_PAGE}')
flash(f'Authorization added for {email}. User must complete account setup via the Create Account page.', 'success')
flash(f'Authorization added for {email}.', 'success')
return redirect(f'/{_PAGE}')
@ -80,9 +85,7 @@ def accounts_edit():
flash('Invalid access level.', 'error')
return redirect(f'/{_PAGE}')
data = _load_accounts()
accounts = data.get('accounts', [])
accounts = config_utils.list_accounts()
if row_index < 0 or row_index >= len(accounts):
flash('Account not found.', 'error')
return redirect(f'/{_PAGE}')
@ -92,29 +95,19 @@ def accounts_edit():
flash('You cannot change your own access level.', 'error')
return redirect(f'/{_PAGE}')
accounts[row_index]['access_level'] = access_level
data['accounts'] = accounts
_save_accounts(data)
flash('Account updated.', 'success')
return redirect(f'/{_PAGE}')
@bp.route('/action/accountmanage/session_invalidate', methods=['POST'])
@auth.require_level('manager')
def session_invalidate():
sid = request.form.get('session_id', '').strip()
if not sid:
flash('Invalid request.', 'error')
return redirect(f'/{_PAGE}')
try:
con = sqlite3.connect(config_utils.SESSIONS_DB, timeout=5)
con.execute('DELETE FROM sessions WHERE session_id=?', (sid,))
con = config_utils.open_accounts_db()
con.execute(
'UPDATE accounts SET access_level=? WHERE account_id=?',
(VALID_LEVELS[access_level], target['account_id'])
)
con.commit()
con.close()
flash('Session invalidated.', 'success')
except Exception:
flash('Failed to invalidate session.', 'error')
except Exception as exc:
flash(f'Could not update account: {exc}', 'error')
return redirect(f'/{_PAGE}')
flash('Account updated.', 'success')
return redirect(f'/{_PAGE}')
@ -127,26 +120,29 @@ def accounts_delete():
flash('Invalid request.', 'error')
return redirect(f'/{_PAGE}')
data = _load_accounts()
accounts = data.get('accounts', [])
accounts = config_utils.list_accounts()
if row_index < 0 or row_index >= len(accounts):
flash('Account not found.', 'error')
return redirect(f'/{_PAGE}')
target = accounts[row_index]
target = accounts[row_index]
target_email = target.get('email_address', '').lower()
current_email = session.get('email_address', '').lower()
initial_email = os.environ.get('INITIAL_MANAGER_EMAIL', '').strip().lower()
if target_email == current_email and target_email != initial_email:
flash('You cannot remove your own account.', 'error')
return redirect(f'/{_PAGE}')
removed_email = target.get('email_address', '')
accounts.pop(row_index)
data['accounts'] = accounts
_save_accounts(data)
try:
con = config_utils.open_accounts_db()
con.execute('DELETE FROM sessions WHERE account_id=?', (target['account_id'],))
con.execute('DELETE FROM accounts WHERE account_id=?', (target['account_id'],))
con.commit()
con.close()
except Exception as exc:
flash(f'Could not delete account: {exc}', 'error')
return redirect(f'/{_PAGE}')
flash(f'Account for {removed_email} has been removed.', 'success')
flash(f'Account for {target["email_address"]} has been removed.', 'success')
return redirect(f'/{_PAGE}')