server/scripts/fix_user_passwords.py

#!/usr/bin/env python3
"""
Script to fix user passwords that are stored in plain text.
This will rehash all passwords that are not properly bcrypt hashed.
"""
import sys
import os

# Add the current directory to sys.path to make imports work
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from sqlalchemy import create_engine, text
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
    # Bcrypt hashes start with $2a$, $2b$, or $2y$
    return password_hash.startswith(('$2a$', '$2b$', '$2y$'))

def fix_user_passwords():
    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"Found {len(users)} users in database")
        
        fixed_count = 0
        for user in users:
            if not is_bcrypt_hash(user.hashed_password):
                print(f"User {user.email} has invalid hash format. Rehashing...")
                # Assume the current value is the plain text password
                # Bcrypt has a 72 byte limit, so truncate if necessary
                plain_password = user.hashed_password[:72]
                user.hashed_password = get_password_hash(plain_password)
                fixed_count += 1
            else:
                print(f"User {user.email} has valid bcrypt hash")
        
        if fixed_count > 0:
            db.commit()
            print(f"\nSuccessfully rehashed {fixed_count} user password(s)")
        else:
            print("\nNo passwords needed to be fixed")
            
    except Exception as e:
        print(f"Error: {e}")
        db.rollback()
    finally:
        db.close()

if __name__ == "__main__":
    fix_user_passwords()