Add admin reset-password command

Features:
- Add admin reset-password command to reset user passwords
- Support SQLite, PostgreSQL, and MySQL databases
- Use scrypt hashing compatible with CodiMD
- Update README with reset-password usage examples
This commit is contained in:
2026-04-07 12:27:23 +08:00
parent 1df784ec7c
commit 8575d7631f
4 changed files with 273 additions and 4 deletions

View File

@@ -9,6 +9,45 @@ from typing import List, Dict, Any, Optional
from pathlib import Path
class PasswordHasher:
"""Password hashing compatible with CodiMD (scrypt-kdf)."""
@staticmethod
def hash_password(password: str) -> str:
"""Hash password using scrypt (compatible with CodiMD).
CodiMD uses scrypt-kdf with default parameters (N=16384, r=8, p=1).
Returns hex-encoded hash that CodiMD can verify.
"""
try:
import hashlib
import scrypt
import os
# Generate random salt (16 bytes)
salt = os.urandom(16)
# Scrypt parameters matching CodiMD defaults
N = 16384
r = 8
p = 1
# Derive key (64 bytes output)
derived_key = scrypt.hash(
password.encode(),
salt,
N,
r,
p,
64 # output length in bytes
)
# Return hex-encoded salt + hash (CodiMD format)
return salt.hex() + derived_key.hex()
except ImportError:
raise ImportError(
"scrypt required for password operations. "
"Install with: pip install scrypt\n"
"Or install with: pip install cli-anything-codimd[admin]"
)
class CodiMDDatabase:
"""Direct database access for CodiMD."""
@@ -148,3 +187,147 @@ class CodiMDDatabase:
notes = [dict(row) for row in cursor.fetchall()]
conn.close()
return notes
@staticmethod
def reset_password_sqlite(db_path: str, email: str, new_password: str) -> bool:
"""Reset user password in SQLite database."""
try:
import sqlite3
except ImportError:
raise ImportError("sqlite3 required for SQLite access")
# Hash the new password
hashed_password = PasswordHasher.hash_password(new_password)
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Check if user exists by email
cursor.execute("SELECT id, email FROM Users WHERE email = ?", (email,))
user = cursor.fetchone()
if not user:
conn.close()
raise ValueError(f"User not found: {email}")
user_id, user_email = user
# Update password
cursor.execute(
'UPDATE Users SET password = ? WHERE email = ?',
(hashed_password, email)
)
conn.commit()
conn.close()
return True
@staticmethod
def reset_password_sqlite_by_id(db_path: str, user_id: str, new_password: str) -> bool:
"""Reset user password in SQLite database by user ID."""
try:
import sqlite3
except ImportError:
raise ImportError("sqlite3 required for SQLite access")
# Hash the new password
hashed_password = PasswordHasher.hash_password(new_password)
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Check if user exists
cursor.execute("SELECT id, email FROM Users WHERE id = ?", (user_id,))
user = cursor.fetchone()
if not user:
conn.close()
raise ValueError(f"User not found: {user_id}")
# Update password
cursor.execute(
'UPDATE Users SET password = ? WHERE id = ?',
(hashed_password, user_id)
)
conn.commit()
conn.close()
return True
@staticmethod
def reset_password_postgresql(host: str, database: str, user: str, password: str,
port: int, target_email: str, new_password: str) -> bool:
"""Reset user password in PostgreSQL database."""
try:
import psycopg2
except ImportError:
raise ImportError("psycopg2 required for PostgreSQL access")
# Hash the new password
hashed_password = PasswordHasher.hash_password(new_password)
conn = psycopg2.connect(
host=host,
database=database,
user=user,
password=password,
port=port
)
cursor = conn.cursor()
# Check if user exists
cursor.execute('SELECT id, email FROM "Users" WHERE email = %s', (target_email,))
row = cursor.fetchone()
if not row:
conn.close()
raise ValueError(f"User not found: {target_email}")
# Update password
cursor.execute(
'UPDATE "Users" SET password = %s WHERE email = %s',
(hashed_password, target_email)
)
conn.commit()
conn.close()
return True
@staticmethod
def reset_password_mysql(host: str, database: str, user: str, password: str,
port: int, target_email: str, new_password: str) -> bool:
"""Reset user password in MySQL database."""
try:
import mysql.connector
except ImportError:
raise ImportError("mysql-connector-python required for MySQL access")
# Hash the new password
hashed_password = PasswordHasher.hash_password(new_password)
conn = mysql.connector.connect(
host=host,
database=database,
user=user,
password=password,
port=port
)
cursor = conn.cursor()
# Check if user exists
cursor.execute('SELECT id, email FROM Users WHERE email = %s', (target_email,))
row = cursor.fetchone()
if not row:
conn.close()
raise ValueError(f"User not found: {target_email}")
# Update password
cursor.execute(
'UPDATE Users SET password = %s WHERE email = %s',
(hashed_password, target_email)
)
conn.commit()
conn.close()
return True