148 lines
5 KiB
Python
148 lines
5 KiB
Python
from pathlib import Path
|
|
from flask import Blueprint, request, session, redirect, flash
|
|
import os, re, sqlite3
|
|
from datetime import datetime, timezone
|
|
import auth
|
|
import config_utils
|
|
import sanitize
|
|
|
|
_PAGE = Path(__file__).parent.name
|
|
|
|
bp = Blueprint(_PAGE, __name__)
|
|
|
|
VALID_LEVELS = {'viewer': 1, 'administrator': 2, 'manager': 3}
|
|
|
|
|
|
@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 = config_utils.open_accounts_db()
|
|
con.execute('DELETE FROM sessions WHERE session_id=?', (sid,))
|
|
con.commit()
|
|
con.close()
|
|
flash('Session invalidated.', 'success')
|
|
except Exception:
|
|
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()
|
|
|
|
if not email:
|
|
flash('Email address is required.', 'error')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
if not re.match(r'^[^@\s]+@[^@\s]+\.[^@\s]+$', email):
|
|
flash('Email address does not appear to be valid.', 'error')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
if access_level not in VALID_LEVELS:
|
|
flash('Invalid access level.', 'error')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
if config_utils.get_account_by_email(email):
|
|
flash('An account with that email address already exists.', 'error')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
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}.', 'success')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
|
|
@bp.route('/action/accountmanage/accounts_edit', methods=['POST'])
|
|
@auth.require_level('manager')
|
|
def accounts_edit():
|
|
try:
|
|
row_index = int(request.form.get('row_index', ''))
|
|
except (ValueError, TypeError):
|
|
flash('Invalid request.', 'error')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
access_level = request.form.get('access_level', '').strip()
|
|
if access_level not in VALID_LEVELS:
|
|
flash('Invalid access level.', 'error')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
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]
|
|
if target.get('email_address', '').lower() == session.get('email_address', '').lower():
|
|
flash('You cannot change your own access level.', 'error')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
try:
|
|
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()
|
|
except Exception as exc:
|
|
flash(f'Could not update account: {exc}', 'error')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
flash('Account updated.', 'success')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
|
|
@bp.route('/action/accountmanage/accounts_delete', methods=['POST'])
|
|
@auth.require_level('manager')
|
|
def accounts_delete():
|
|
try:
|
|
row_index = int(request.form.get('row_index', ''))
|
|
except (ValueError, TypeError):
|
|
flash('Invalid request.', 'error')
|
|
return redirect(f'/{_PAGE}')
|
|
|
|
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_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}')
|
|
|
|
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 {target["email_address"]} has been removed.', 'success')
|
|
return redirect(f'/{_PAGE}')
|