325 lines
9.0 KiB
Python
325 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
JWT 認證 API
|
||
|
||
Author: PANJIT IT Team
|
||
Created: 2024-01-28
|
||
Modified: 2024-09-02
|
||
"""
|
||
|
||
from flask import Blueprint, request, jsonify, current_app
|
||
from flask_jwt_extended import (
|
||
create_access_token, create_refresh_token,
|
||
jwt_required, get_jwt_identity, get_jwt
|
||
)
|
||
from app.utils.ldap_auth import LDAPAuthService
|
||
from app.utils.decorators import validate_json, rate_limit
|
||
from app.utils.exceptions import AuthenticationError
|
||
from app.utils.logger import get_logger
|
||
from app.models.user import User
|
||
from app.models.log import SystemLog
|
||
|
||
auth_bp = Blueprint('auth', __name__, url_prefix='/auth')
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
@auth_bp.route('/login', methods=['POST'])
|
||
@rate_limit(max_requests=10, per_seconds=300) # 5分鐘內最多10次嘗試
|
||
@validate_json(['username', 'password'])
|
||
def login():
|
||
"""使用者登入"""
|
||
try:
|
||
data = request.get_json()
|
||
username = data['username'].strip()
|
||
password = data['password']
|
||
|
||
if not username or not password:
|
||
return jsonify({
|
||
'success': False,
|
||
'error': 'INVALID_INPUT',
|
||
'message': '帳號和密碼不能為空'
|
||
}), 400
|
||
|
||
# LDAP 認證
|
||
ldap_service = LDAPAuthService()
|
||
user_info = ldap_service.authenticate_user(username, password)
|
||
|
||
# 取得或建立使用者
|
||
user = User.get_or_create(
|
||
username=user_info['username'],
|
||
display_name=user_info['display_name'],
|
||
email=user_info['email'],
|
||
department=user_info.get('department')
|
||
)
|
||
|
||
# 更新登入時間
|
||
user.update_last_login()
|
||
|
||
# 創建 JWT tokens
|
||
access_token = create_access_token(
|
||
identity=user.username,
|
||
additional_claims={
|
||
'user_id': user.id,
|
||
'is_admin': user.is_admin,
|
||
'display_name': user.display_name,
|
||
'email': user.email
|
||
}
|
||
)
|
||
refresh_token = create_refresh_token(identity=user.username)
|
||
|
||
# 記錄登入日誌
|
||
SystemLog.info(
|
||
'auth.login',
|
||
f'User {username} logged in successfully',
|
||
user_id=user.id,
|
||
extra_data={
|
||
'ip_address': request.remote_addr,
|
||
'user_agent': request.headers.get('User-Agent')
|
||
}
|
||
)
|
||
|
||
logger.info(f"🔑 [JWT Created] User: {username}, UserID: {user.id}")
|
||
logger.info(f"User {username} logged in successfully")
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'access_token': access_token,
|
||
'refresh_token': refresh_token,
|
||
'user': user.to_dict()
|
||
},
|
||
'message': '登入成功'
|
||
})
|
||
|
||
except AuthenticationError as e:
|
||
# 記錄認證失敗
|
||
SystemLog.warning(
|
||
'auth.login_failed',
|
||
f'Authentication failed for user {username}: {str(e)}',
|
||
extra_data={
|
||
'username': username,
|
||
'ip_address': request.remote_addr,
|
||
'error': str(e)
|
||
}
|
||
)
|
||
|
||
logger.warning(f"Authentication failed for user {username}: {str(e)}")
|
||
|
||
return jsonify({
|
||
'success': False,
|
||
'error': 'INVALID_CREDENTIALS',
|
||
'message': str(e)
|
||
}), 401
|
||
|
||
except Exception as e:
|
||
logger.error(f"Login error: {str(e)}")
|
||
|
||
SystemLog.error(
|
||
'auth.login_error',
|
||
f'Login system error: {str(e)}',
|
||
extra_data={
|
||
'username': username,
|
||
'error': str(e)
|
||
}
|
||
)
|
||
|
||
return jsonify({
|
||
'success': False,
|
||
'error': 'SYSTEM_ERROR',
|
||
'message': '系統錯誤，請稍後再試'
|
||
}), 500
|
||
|
||
|
||
@auth_bp.route('/logout', methods=['POST'])
|
||
@jwt_required()
|
||
def logout():
|
||
"""使用者登出"""
|
||
try:
|
||
username = get_jwt_identity()
|
||
|
||
# 記錄登出日誌
|
||
SystemLog.info(
|
||
'auth.logout',
|
||
f'User {username} logged out'
|
||
)
|
||
|
||
logger.info(f"🚪 [JWT Logout] User: {username}")
|
||
logger.info(f"User {username} logged out")
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': '登出成功'
|
||
})
|
||
|
||
except Exception as e:
|
||
logger.error(f"Logout error: {str(e)}")
|
||
|
||
return jsonify({
|
||
'success': False,
|
||
'error': 'SYSTEM_ERROR',
|
||
'message': '登出時發生錯誤'
|
||
}), 500
|
||
|
||
|
||
@auth_bp.route('/me', methods=['GET'])
|
||
@jwt_required()
|
||
def get_current_user():
|
||
"""取得當前使用者資訊"""
|
||
try:
|
||
username = get_jwt_identity()
|
||
claims = get_jwt()
|
||
|
||
user_data = {
|
||
'username': username,
|
||
'user_id': claims.get('user_id'),
|
||
'is_admin': claims.get('is_admin'),
|
||
'display_name': claims.get('display_name'),
|
||
'email': claims.get('email')
|
||
}
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'user': user_data
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
logger.error(f"Get current user error: {str(e)}")
|
||
|
||
return jsonify({
|
||
'success': False,
|
||
'error': 'SYSTEM_ERROR',
|
||
'message': '取得使用者資訊時發生錯誤'
|
||
}), 500
|
||
|
||
|
||
@auth_bp.route('/refresh', methods=['POST'])
|
||
@jwt_required(refresh=True)
|
||
def refresh_token():
|
||
"""刷新 Access Token"""
|
||
try:
|
||
username = get_jwt_identity()
|
||
|
||
# 重新取得使用者資訊
|
||
user = User.query.filter_by(username=username).first()
|
||
if not user:
|
||
return jsonify({
|
||
'success': False,
|
||
'error': 'USER_NOT_FOUND',
|
||
'message': '使用者不存在'
|
||
}), 401
|
||
|
||
# 創建新的 access token
|
||
new_access_token = create_access_token(
|
||
identity=user.username,
|
||
additional_claims={
|
||
'user_id': user.id,
|
||
'is_admin': user.is_admin,
|
||
'display_name': user.display_name,
|
||
'email': user.email
|
||
}
|
||
)
|
||
|
||
logger.info(f"Token refreshed for user {user.username}")
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'access_token': new_access_token,
|
||
'user': user.to_dict()
|
||
},
|
||
'message': 'Token 已刷新'
|
||
})
|
||
|
||
except Exception as e:
|
||
logger.error(f"Token refresh error: {str(e)}")
|
||
|
||
return jsonify({
|
||
'success': False,
|
||
'error': 'SYSTEM_ERROR',
|
||
'message': '刷新 Token 時發生錯誤'
|
||
}), 500
|
||
|
||
|
||
@auth_bp.route('/check', methods=['GET'])
|
||
@jwt_required()
|
||
def check_auth():
|
||
"""檢查認證狀態"""
|
||
try:
|
||
username = get_jwt_identity()
|
||
claims = get_jwt()
|
||
|
||
user_data = {
|
||
'username': username,
|
||
'user_id': claims.get('user_id'),
|
||
'is_admin': claims.get('is_admin'),
|
||
'display_name': claims.get('display_name'),
|
||
'email': claims.get('email')
|
||
}
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'authenticated': True,
|
||
'data': {
|
||
'user': user_data
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
logger.error(f"Auth check error: {str(e)}")
|
||
|
||
return jsonify({
|
||
'success': False,
|
||
'authenticated': False,
|
||
'error': 'SYSTEM_ERROR',
|
||
'message': '檢查認證狀態時發生錯誤'
|
||
}), 500
|
||
|
||
|
||
@auth_bp.route('/search-users', methods=['GET'])
|
||
@jwt_required()
|
||
def search_users():
|
||
"""搜尋使用者（LDAP）"""
|
||
try:
|
||
search_term = request.args.get('q', '').strip()
|
||
limit = min(int(request.args.get('limit', 20)), 50)
|
||
|
||
if len(search_term) < 2:
|
||
return jsonify({
|
||
'success': False,
|
||
'error': 'INVALID_SEARCH_TERM',
|
||
'message': '搜尋關鍵字至少需要2個字元'
|
||
}), 400
|
||
|
||
ldap_service = LDAPAuthService()
|
||
users = ldap_service.search_users(search_term, limit)
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'users': users,
|
||
'count': len(users)
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
logger.error(f"User search error: {str(e)}")
|
||
|
||
return jsonify({
|
||
'success': False,
|
||
'error': 'SYSTEM_ERROR',
|
||
'message': '搜尋使用者時發生錯誤'
|
||
}), 500
|
||
|
||
|
||
# 錯誤處理器
|
||
@auth_bp.errorhandler(429)
|
||
def rate_limit_handler(e):
|
||
"""速率限制錯誤處理器"""
|
||
return jsonify({
|
||
'success': False,
|
||
'error': 'RATE_LIMIT_EXCEEDED',
|
||
'message': '請求過於頻繁，請稍後再試'
|
||
}), 429 |