Spaces:
Running
Running
File size: 2,857 Bytes
a7c9d43 0b9317a a7c9d43 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | import sqlite3
import streamlit_authenticator as stauth
DB_NAME = "users.db"
def init_db(admin_email, admin_password):
"""Initializes the database and creates default admin account if not existing."""
conn = sqlite3.connect(DB_NAME)
c = conn.cursor()
c.execute('''
CREATE TABLE IF NOT EXISTS users (
username TEXT PRIMARY KEY,
name TEXT,
email TEXT,
password TEXT,
status TEXT
)
''')
# Check if admin exists
c.execute('SELECT * FROM users WHERE username = ?', ('admin',))
if not c.fetchone():
hashed_password = stauth.Hasher([admin_password]).generate()[0]
c.execute(
'INSERT INTO users (username, name, email, password, status) VALUES (?, ?, ?, ?, ?)',
('admin', 'System Admin', admin_email, hashed_password, 'approved')
)
conn.commit()
conn.close()
def verify_password(stored_password, provided_password):
"""Verifies a provided password against the stored hashed password."""
return stauth.Hasher.check_password(stored_password, provided_password)
def register_user(username, name, email, password):
"""Registers a new user with status 'pending'."""
conn = sqlite3.connect(DB_NAME)
c = conn.cursor()
try:
hashed_password = stauth.Hasher([password]).generate()[0]
c.execute(
'INSERT INTO users (username, name, email, password, status) VALUES (?, ?, ?, ?, ?)',
(username, name, email, hashed_password, 'pending')
)
conn.commit()
return True, "Registration successful! Awaiting admin approval."
except sqlite3.IntegrityError:
return False, "Username already exists."
finally:
conn.close()
def get_pending_users():
"""Fetches all users waiting for approval."""
conn = sqlite3.connect(DB_NAME)
c = conn.cursor()
c.execute('SELECT username, name, email FROM users WHERE status = "pending"')
pending = c.fetchall()
conn.close()
return pending
def approve_user(username):
"""Approves a user's login access."""
conn = sqlite3.connect(DB_NAME)
c = conn.cursor()
c.execute('UPDATE users SET status = "approved" WHERE username = ?', (username,))
conn.commit()
conn.close()
def reject_user(username):
"""Rejects/Deletes a user's registration request."""
conn = sqlite3.connect(DB_NAME)
c = conn.cursor()
c.execute('DELETE FROM users WHERE username = ?', (username,))
conn.commit()
conn.close()
def check_user_status(username):
"""Returns status and hashed password for authentication."""
conn = sqlite3.connect(DB_NAME)
c = conn.cursor()
c.execute('SELECT password, status, name, email FROM users WHERE username = ?', (username,))
res = c.fetchone()
conn.close()
return res |