120 lines
3.7 KiB
Python
120 lines
3.7 KiB
Python
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
import hashlib
|
|
import os
|
|
from app.core.database import get_connection
|
|
|
|
router = APIRouter(prefix="/api/users", tags=["users"])
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class ChangePasswordRequest(BaseModel):
|
|
username: str
|
|
current_password: str
|
|
new_password: str
|
|
|
|
|
|
class LoginResponse(BaseModel):
|
|
id: int
|
|
username: str
|
|
full_name: str | None
|
|
email: str | None
|
|
message: str
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""パスワードをハッシュ化"""
|
|
# 実運用ではより安全なハッシュアルゴリズムを使用してください
|
|
return hashlib.sha256(password.encode()).hexdigest()
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
async def login(request: LoginRequest):
|
|
"""ユーザーのログイン処理"""
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor()
|
|
|
|
# ユーザーが存在するか確認
|
|
cur.execute(
|
|
"SELECT id, username, password, full_name, email FROM users WHERE username = %s AND is_active = TRUE",
|
|
(request.username,)
|
|
)
|
|
user = cur.fetchone()
|
|
cur.close()
|
|
conn.close()
|
|
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="ユーザー名またはパスワードが正しくありません")
|
|
|
|
# パスワードの検証
|
|
hashed_password = hash_password(request.password)
|
|
if user["password"] != hashed_password:
|
|
raise HTTPException(status_code=401, detail="ユーザー名またはパスワードが正しくありません")
|
|
|
|
return LoginResponse(
|
|
id=user["id"],
|
|
username=user["username"],
|
|
full_name=user["full_name"],
|
|
email=user["email"],
|
|
message="ログインに成功しました"
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"ログイン処理エラー: {str(e)}")
|
|
|
|
|
|
@router.post("/register")
|
|
async def register(request: LoginRequest):
|
|
"""新規ユーザー登録（無効化済み）"""
|
|
raise HTTPException(status_code=403, detail="新規ユーザー登録は現在受け付けていません")
|
|
|
|
|
|
@router.post("/change-password")
|
|
async def change_password(request: ChangePasswordRequest):
|
|
"""パスワード変更"""
|
|
if len(request.new_password) < 8:
|
|
raise HTTPException(status_code=400, detail="新しいパスワードは8文字以上である必要があります")
|
|
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor()
|
|
|
|
cur.execute(
|
|
"SELECT id, password FROM users WHERE username = %s AND is_active = TRUE",
|
|
(request.username,)
|
|
)
|
|
user = cur.fetchone()
|
|
|
|
if not user:
|
|
cur.close()
|
|
conn.close()
|
|
raise HTTPException(status_code=401, detail="ユーザーが見つかりません")
|
|
|
|
if user["password"] != hash_password(request.current_password):
|
|
cur.close()
|
|
conn.close()
|
|
raise HTTPException(status_code=401, detail="現在のパスワードが正しくありません")
|
|
|
|
new_hashed = hash_password(request.new_password)
|
|
cur.execute(
|
|
"UPDATE users SET password = %s WHERE username = %s",
|
|
(new_hashed, request.username)
|
|
)
|
|
conn.commit()
|
|
cur.close()
|
|
conn.close()
|
|
|
|
return {"message": "パスワードを変更しました"}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"パスワード変更エラー: {str(e)}")
|