server/scripts/diagnose_passwords.py

#!/usr/bin/env python3
"""
Diagnostic script to check user passwords and provide options to fix them.
"""
import sys
import os

sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
from app.core.security import get_password_hash
from app.models.user import User

def is_bcrypt_hash(password_hash: str) -> bool:
    """Check if a string is a valid bcrypt hash"""
    if not password_hash:
        return False
    return password_hash.startswith(('$2a$', '$2b$', '$2y$'))

def diagnose_and_fix():
    print(f"Connecting to database: {settings.DATABASE_URL}")
    engine = create_engine(settings.DATABASE_URL)
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    db = SessionLocal()
    
    try:
        users = db.query(User).all()
        print(f"\nFound {len(users)} users in database\n")
        
        for user in users:
            print(f"User: {user.email}")
            print(f"  Hash starts with: {user.hashed_password[:20]}...")
            print(f"  Hash length: {len(user.hashed_password)} characters")
            print(f"  Hash byte length: {len(user.hashed_password.encode('utf-8'))} bytes")
            print(f"  Is valid bcrypt: {is_bcrypt_hash(user.hashed_password)}")
            
            if not is_bcrypt_hash(user.hashed_password):
                print(f"  ⚠️  NEEDS FIXING")
                # Set a temporary password that users will need to reset
                temp_password = "TempPassword123!"
                print(f"  Setting temporary password: {temp_password}")
                user.hashed_password = get_password_hash(temp_password)
            print()
        
        db.commit()
        print("✅ All users have been fixed with temporary passwords if needed")
        print("Users with invalid hashes now have password: TempPassword123!")
        
    except Exception as e:
        print(f"Error: {e}")
        import traceback
        traceback.print_exc()
        db.rollback()
    finally:
        db.close()

if __name__ == "__main__":
    diagnose_and_fix()