feat: multi-admin support — named logins instead of one shared password
Matches Marzban's multi-admin (WIP there) and closes a real gap vs both. New 'admins' table (username + PBKDF2-SHA256 password hash, 200k iterations, random salt per account, stdlib hashlib/hmac only — no new dependency), admin_sessions now tracks which admin is logged in. Existing installs aren't broken: on first run, if no admins exist yet, a default 'admin' account is seeded from the current ADMIN_PANEL_PASSWORD — old password keeps working under username 'admin', pre-filled on the login screen. Admin management lives in Settings: list, add (username + password, min 8 chars), remove. Can't delete the last remaining admin or your own currently-logged-in account. Sidebar now shows who's logged in. Verified end-to-end: bootstrap, correct/wrong/nonexistent login, session->admin resolution, last-admin-delete protection, duplicate username rejection, add/remove round trip, and that identical passwords hash to different values (unique salt) but both verify. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
81cbc4e391
commit
9e6e314c94
3 changed files with 210 additions and 13 deletions
102
db.py
102
db.py
|
|
@ -1,3 +1,5 @@
|
|||
import hashlib
|
||||
import hmac
|
||||
import os
|
||||
import sqlite3
|
||||
import secrets
|
||||
|
|
@ -52,6 +54,13 @@ CREATE TABLE IF NOT EXISTS admin_sessions (
|
|||
expires_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS admins (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
username TEXT UNIQUE NOT NULL,
|
||||
password_hash TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS payments (
|
||||
id TEXT PRIMARY KEY,
|
||||
tg_id INTEGER NOT NULL,
|
||||
|
|
@ -111,6 +120,10 @@ _NEW_USER_COLUMNS = {
|
|||
"hwid_limit": "INTEGER",
|
||||
}
|
||||
|
||||
_NEW_ADMIN_SESSION_COLUMNS = {
|
||||
"admin_id": "INTEGER",
|
||||
}
|
||||
|
||||
|
||||
def _migrate():
|
||||
with get_conn() as conn:
|
||||
|
|
@ -123,6 +136,10 @@ def _migrate():
|
|||
for name, decl in _NEW_USER_COLUMNS.items():
|
||||
if name not in ucols:
|
||||
conn.execute(f"ALTER TABLE users ADD COLUMN {name} {decl}")
|
||||
scols = {r["name"] for r in conn.execute("PRAGMA table_info(admin_sessions)").fetchall()}
|
||||
for name, decl in _NEW_ADMIN_SESSION_COLUMNS.items():
|
||||
if name not in scols:
|
||||
conn.execute(f"ALTER TABLE admin_sessions ADD COLUMN {name} {decl}")
|
||||
if needs_sort_order_backfill:
|
||||
rows = conn.execute(
|
||||
"SELECT code FROM nodes ORDER BY (code='de1') DESC, created_at ASC"
|
||||
|
|
@ -154,6 +171,7 @@ def init_db():
|
|||
conn.executescript(SCHEMA)
|
||||
_migrate()
|
||||
_seed_local_node()
|
||||
_seed_default_admin()
|
||||
for suffix in ("", "-wal", "-shm"):
|
||||
path = DB_PATH + suffix
|
||||
if os.path.exists(path):
|
||||
|
|
@ -177,6 +195,36 @@ def _seed_local_node():
|
|||
)
|
||||
|
||||
|
||||
def _hash_password(password: str, salt: bytes | None = None) -> str:
|
||||
if salt is None:
|
||||
salt = os.urandom(16)
|
||||
dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 200_000)
|
||||
return salt.hex() + "$" + dk.hex()
|
||||
|
||||
|
||||
def _verify_password(password: str, stored: str) -> bool:
|
||||
try:
|
||||
salt_hex, hash_hex = stored.split("$")
|
||||
except ValueError:
|
||||
return False
|
||||
salt = bytes.fromhex(salt_hex)
|
||||
dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 200_000)
|
||||
return hmac.compare_digest(dk.hex(), hash_hex)
|
||||
|
||||
|
||||
def _seed_default_admin():
|
||||
from config import ADMIN_PANEL_PASSWORD
|
||||
|
||||
with get_conn() as conn:
|
||||
row = conn.execute("SELECT 1 FROM admins LIMIT 1").fetchone()
|
||||
if row or not ADMIN_PANEL_PASSWORD:
|
||||
return
|
||||
conn.execute(
|
||||
"INSERT INTO admins (username, password_hash, created_at) VALUES (?,?,?)",
|
||||
("admin", _hash_password(ADMIN_PANEL_PASSWORD), now_iso()),
|
||||
)
|
||||
|
||||
|
||||
def list_nodes(enabled_only: bool = False):
|
||||
q = "SELECT * FROM nodes"
|
||||
if enabled_only:
|
||||
|
|
@ -356,13 +404,13 @@ def redeem_gift_code(code: str, tg_id: int):
|
|||
return dict(row), None
|
||||
|
||||
|
||||
def create_admin_session(hours: int = 168):
|
||||
def create_admin_session(admin_id: int | None = None, hours: int = 168):
|
||||
token = secrets.token_urlsafe(32)
|
||||
expires = datetime.datetime.utcnow() + datetime.timedelta(hours=hours)
|
||||
with get_conn() as conn:
|
||||
conn.execute(
|
||||
"INSERT INTO admin_sessions (token, created_at, expires_at) VALUES (?,?,?)",
|
||||
(token, now_iso(), expires.isoformat()),
|
||||
"INSERT INTO admin_sessions (token, admin_id, created_at, expires_at) VALUES (?,?,?,?)",
|
||||
(token, admin_id, now_iso(), expires.isoformat()),
|
||||
)
|
||||
return token
|
||||
|
||||
|
|
@ -377,6 +425,18 @@ def validate_admin_session(token: str) -> bool:
|
|||
return row is not None
|
||||
|
||||
|
||||
def get_session_admin(token: str):
|
||||
if not token:
|
||||
return None
|
||||
with get_conn() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT a.id, a.username FROM admin_sessions s "
|
||||
"JOIN admins a ON a.id = s.admin_id "
|
||||
"WHERE s.token=? AND s.expires_at>?", (token, now_iso())
|
||||
).fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
def delete_admin_session(token: str):
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM admin_sessions WHERE token=?", (token,))
|
||||
|
|
@ -387,6 +447,42 @@ def delete_expired_admin_sessions():
|
|||
conn.execute("DELETE FROM admin_sessions WHERE expires_at<=?", (now_iso(),))
|
||||
|
||||
|
||||
def verify_admin_login(username: str, password: str):
|
||||
with get_conn() as conn:
|
||||
row = conn.execute("SELECT * FROM admins WHERE username=?", (username,)).fetchone()
|
||||
if not row or not _verify_password(password, row["password_hash"]):
|
||||
return None
|
||||
return dict(row)
|
||||
|
||||
|
||||
def list_admins():
|
||||
with get_conn() as conn:
|
||||
rows = conn.execute("SELECT id, username, created_at FROM admins ORDER BY created_at ASC").fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def create_admin(username: str, password: str):
|
||||
with get_conn() as conn:
|
||||
existing = conn.execute("SELECT 1 FROM admins WHERE username=?", (username,)).fetchone()
|
||||
if existing:
|
||||
raise ValueError("username already taken")
|
||||
conn.execute(
|
||||
"INSERT INTO admins (username, password_hash, created_at) VALUES (?,?,?)",
|
||||
(username, _hash_password(password), now_iso()),
|
||||
)
|
||||
row = conn.execute("SELECT id, username, created_at FROM admins WHERE username=?", (username,)).fetchone()
|
||||
return dict(row)
|
||||
|
||||
|
||||
def delete_admin(admin_id: int):
|
||||
with get_conn() as conn:
|
||||
count = conn.execute("SELECT COUNT(*) c FROM admins").fetchone()["c"]
|
||||
if count <= 1:
|
||||
raise ValueError("cannot delete the last remaining admin")
|
||||
conn.execute("DELETE FROM admins WHERE id=?", (admin_id,))
|
||||
conn.execute("DELETE FROM admin_sessions WHERE admin_id=?", (admin_id,))
|
||||
|
||||
|
||||
def list_all_subscriptions(limit: int = 200):
|
||||
with get_conn() as conn:
|
||||
rows = conn.execute(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue