IT狗
acbc767247
- database.py: CryptContext = ['bcrypt', 'sha256_crypt'] with sha256_crypt deprecated - database.py: add verify_and_update_password() returning (ok, new_hash) - main.py: login uses verify_and_update; on successful verify transparently commits new bcrypt hash when old hash is sha256_crypt - requirements.txt: passlib[bcrypt]==1.7.4 + bcrypt==4.0.1 (compat with passlib 1.7.4) Migration: zero-touch. Admin's first login after this deploy will auto-upgrade the stored hash from $5$... to $2b$... Avoids the passlib sha256_crypt cross-version verify bug that bit us this morning. bcrypt is OpenSSL-native and stable across passlib versions.
86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
import os
|
|
from datetime import datetime, timezone
|
|
from sqlalchemy import create_engine, Column, Integer, String, Text, Float, Boolean, DateTime, Date, Time, JSON
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
from passlib.context import CryptContext
|
|
from jose import jwt
|
|
from typing import Optional, List
|
|
from pydantic import BaseModel
|
|
import secret
|
|
|
|
# Database setup
|
|
DATABASE_URL = "sqlite:///./data/aars.db"
|
|
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
Base = declarative_base()
|
|
|
|
# bcrypt is the primary scheme (cross-version stable, OpenSSL-native).
|
|
# sha256_crypt is kept as deprecated fallback so legacy hashes still
|
|
# verify; on first successful login the password is re-hashed to bcrypt
|
|
# via verify_and_update().
|
|
pwd_context = CryptContext(schemes=["bcrypt", "sha256_crypt"], deprecated=["sha256_crypt"])
|
|
|
|
# JWT Settings
|
|
SECRET_KEY = secret.JWT_SECRET if hasattr(secret, 'JWT_SECRET') else "aars-dev-secret-change-in-production"
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_HOURS = 24
|
|
|
|
|
|
def create_access_token(data: dict) -> str:
|
|
expire = datetime.now(timezone.utc).timestamp() + (ACCESS_TOKEN_EXPIRE_HOURS * 3600)
|
|
to_encode = data.copy()
|
|
to_encode.update({"exp": expire})
|
|
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
|
|
def verify_and_update_password(plain_password: str, hashed_password: str) -> tuple[bool, str | None]:
|
|
"""Verify password and return new hash if rehash is needed.
|
|
|
|
Returns (ok, new_hash_or_none). Caller is responsible for committing
|
|
the new hash to the DB when new_hash is not None.
|
|
"""
|
|
return pwd_context.verify_and_update(plain_password, hashed_password)
|
|
|
|
|
|
def get_password_hash(password: str) -> str:
|
|
return pwd_context.hash(password)
|
|
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def init_db():
|
|
"""Initialize database with tables and default admin user."""
|
|
Base.metadata.create_all(bind=engine)
|
|
db = SessionLocal()
|
|
try:
|
|
# Check if admin exists
|
|
from models import User
|
|
admin = db.query(User).filter(User.email == "admin@aars.hk").first()
|
|
if not admin:
|
|
# Create default admin
|
|
admin = User(
|
|
email="admin@aars.hk",
|
|
password_hash=get_password_hash("admin123"),
|
|
name="系統管理員",
|
|
role="admin",
|
|
is_active=True
|
|
)
|
|
db.add(admin)
|
|
db.commit()
|
|
print("✅ Default admin user created: admin@aars.hk / admin123")
|
|
else:
|
|
print("✅ Admin user already exists")
|
|
finally:
|
|
db.close()
|