1panel
This commit is contained in:
beabigegg
277
app/utils/api_auth.py
Normal file
277
app/utils/api_auth.py
Normal file
@@ -0,0 +1,277 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
API 認證服務
|
||||
用於與 PANJIT Auth API 整合認證
|
||||
|
||||
Author: PANJIT IT Team
|
||||
Created: 2025-10-01
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional, Dict, Any, Tuple
|
||||
from flask import current_app
|
||||
from .logger import get_logger
|
||||
from .exceptions import AuthenticationError
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class APIAuthService:
|
||||
"""API 認證服務"""
|
||||
|
||||
def __init__(self):
|
||||
self.config = current_app.config
|
||||
self.api_base_url = "https://pj-auth-api.vercel.app"
|
||||
self.login_endpoint = "/api/auth/login"
|
||||
self.logout_endpoint = "/api/auth/logout"
|
||||
self.timeout = 30 # 30 秒超時
|
||||
|
||||
def authenticate_user(self, username: str, password: str) -> Dict[str, Any]:
|
||||
"""
|
||||
透過 API 驗證使用者憑證
|
||||
|
||||
Args:
|
||||
username: 使用者帳號
|
||||
password: 密碼
|
||||
|
||||
Returns:
|
||||
Dict: 包含使用者資訊和 Token 的字典
|
||||
|
||||
Raises:
|
||||
AuthenticationError: 認證失敗時拋出
|
||||
"""
|
||||
try:
|
||||
login_url = f"{self.api_base_url}{self.login_endpoint}"
|
||||
|
||||
payload = {
|
||||
"username": username,
|
||||
"password": password
|
||||
}
|
||||
|
||||
headers = {
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
logger.info(f"正在透過 API 驗證使用者: {username}")
|
||||
|
||||
# 發送認證請求
|
||||
response = requests.post(
|
||||
login_url,
|
||||
json=payload,
|
||||
headers=headers,
|
||||
timeout=self.timeout
|
||||
)
|
||||
|
||||
# 解析回應
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
|
||||
if data.get('success'):
|
||||
logger.info(f"API 認證成功: {username}")
|
||||
return self._parse_auth_response(data)
|
||||
else:
|
||||
error_msg = data.get('error', '認證失敗')
|
||||
logger.warning(f"API 認證失敗: {username} - {error_msg}")
|
||||
raise AuthenticationError(f"認證失敗: {error_msg}")
|
||||
|
||||
elif response.status_code == 401:
|
||||
data = response.json()
|
||||
error_msg = data.get('error', '帳號或密碼錯誤')
|
||||
logger.warning(f"API 認證失敗 (401): {username} - {error_msg}")
|
||||
raise AuthenticationError("帳號或密碼錯誤")
|
||||
|
||||
else:
|
||||
logger.error(f"API 認證請求失敗: HTTP {response.status_code}")
|
||||
raise AuthenticationError(f"認證服務錯誤 (HTTP {response.status_code})")
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
logger.error(f"API 認證請求超時: {username}")
|
||||
raise AuthenticationError("認證服務回應超時，請稍後再試")
|
||||
|
||||
except requests.exceptions.ConnectionError:
|
||||
logger.error(f"API 認證連線錯誤: {username}")
|
||||
raise AuthenticationError("無法連接認證服務，請檢查網路連線")
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"API 認證請求錯誤: {username} - {str(e)}")
|
||||
raise AuthenticationError(f"認證服務錯誤: {str(e)}")
|
||||
|
||||
except json.JSONDecodeError:
|
||||
logger.error(f"API 認證回應格式錯誤: {username}")
|
||||
raise AuthenticationError("認證服務回應格式錯誤")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"API 認證未知錯誤: {username} - {str(e)}")
|
||||
raise AuthenticationError(f"認證過程發生錯誤: {str(e)}")
|
||||
|
||||
def _parse_auth_response(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
解析 API 認證回應
|
||||
|
||||
Args:
|
||||
data: API 回應資料
|
||||
|
||||
Returns:
|
||||
Dict: 標準化的使用者資訊
|
||||
"""
|
||||
try:
|
||||
auth_data = data.get('data', {})
|
||||
user_info = auth_data.get('userInfo', {})
|
||||
|
||||
# 解析 Token 過期時間
|
||||
expires_at = None
|
||||
issued_at = None
|
||||
|
||||
if 'expiresAt' in auth_data:
|
||||
try:
|
||||
expires_at = datetime.fromisoformat(auth_data['expiresAt'].replace('Z', '+00:00'))
|
||||
except (ValueError, AttributeError):
|
||||
logger.warning("無法解析 API Token 過期時間")
|
||||
|
||||
if 'issuedAt' in auth_data:
|
||||
try:
|
||||
issued_at = datetime.fromisoformat(auth_data['issuedAt'].replace('Z', '+00:00'))
|
||||
except (ValueError, AttributeError):
|
||||
logger.warning("無法解析 API Token 發行時間")
|
||||
|
||||
# 標準化使用者資訊 (方案 A: API name 是姓名+email 格式)
|
||||
api_name = user_info.get('name', '') # 例: "劉怡明 ymirliu@panjit.com.tw"
|
||||
api_email = user_info.get('email', '') # 例: "ymirliu@panjit.com.tw"
|
||||
|
||||
result = {
|
||||
# 基本使用者資訊 (方案 A: username 和 display_name 都用 API name)
|
||||
'username': api_name, # 姓名+email 格式
|
||||
'display_name': api_name, # 姓名+email 格式
|
||||
'email': api_email, # 純 email
|
||||
'department': user_info.get('jobTitle'), # 使用 jobTitle 作為部門
|
||||
'user_principal_name': api_email,
|
||||
|
||||
# API 特有資訊
|
||||
'api_user_id': user_info.get('id', ''), # Azure Object ID
|
||||
'job_title': user_info.get('jobTitle'),
|
||||
'office_location': user_info.get('officeLocation'),
|
||||
'business_phones': user_info.get('businessPhones', []),
|
||||
|
||||
# Token 資訊
|
||||
'api_access_token': auth_data.get('access_token', ''),
|
||||
'api_id_token': auth_data.get('id_token', ''),
|
||||
'api_token_type': auth_data.get('token_type', 'Bearer'),
|
||||
'api_expires_in': auth_data.get('expires_in', 0),
|
||||
'api_issued_at': issued_at,
|
||||
'api_expires_at': expires_at,
|
||||
|
||||
# 完整的 API 回應 (用於記錄)
|
||||
'full_api_response': data,
|
||||
'api_user_info': user_info
|
||||
}
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"解析 API 回應時發生錯誤: {str(e)}")
|
||||
raise AuthenticationError(f"解析認證回應時發生錯誤: {str(e)}")
|
||||
|
||||
def logout_user(self, access_token: str) -> bool:
|
||||
"""
|
||||
透過 API 登出使用者
|
||||
|
||||
Args:
|
||||
access_token: 使用者的 access token
|
||||
|
||||
Returns:
|
||||
bool: 登出是否成功
|
||||
"""
|
||||
try:
|
||||
logout_url = f"{self.api_base_url}{self.logout_endpoint}"
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {access_token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
response = requests.post(
|
||||
logout_url,
|
||||
headers=headers,
|
||||
timeout=self.timeout
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
if data.get('success'):
|
||||
logger.info("API 登出成功")
|
||||
return True
|
||||
|
||||
logger.warning(f"API 登出失敗: HTTP {response.status_code}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"API 登出時發生錯誤: {str(e)}")
|
||||
return False
|
||||
|
||||
def validate_token(self, access_token: str) -> bool:
|
||||
"""
|
||||
驗證 Token 是否有效
|
||||
|
||||
Args:
|
||||
access_token: 要驗證的 token
|
||||
|
||||
Returns:
|
||||
bool: Token 是否有效
|
||||
"""
|
||||
try:
|
||||
# 這裡可以實作 Token 驗證邏輯
|
||||
# 目前 API 沒有提供專門的驗證端點，可以考慮解析 JWT 或調用其他端點
|
||||
|
||||
# 簡單的檢查：Token 不能為空且格式看起來像 JWT
|
||||
if not access_token or len(access_token.split('.')) != 3:
|
||||
return False
|
||||
|
||||
# TODO: 實作更完整的 JWT 驗證邏輯
|
||||
# 可以解析 JWT payload 檢查過期時間等
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"驗證 Token 時發生錯誤: {str(e)}")
|
||||
return False
|
||||
|
||||
def test_connection(self) -> bool:
|
||||
"""
|
||||
測試 API 連線
|
||||
|
||||
Returns:
|
||||
bool: 連線是否正常
|
||||
"""
|
||||
try:
|
||||
# 嘗試連接 API 基礎端點
|
||||
response = requests.get(
|
||||
self.api_base_url,
|
||||
timeout=10
|
||||
)
|
||||
|
||||
return response.status_code in [200, 404] # 404 也算正常，表示能連接到伺服器
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"API 連線測試失敗: {str(e)}")
|
||||
return False
|
||||
|
||||
def calculate_internal_expiry(self, api_expires_at: Optional[datetime], extend_days: int = 3) -> datetime:
|
||||
"""
|
||||
計算內部 Token 過期時間
|
||||
|
||||
Args:
|
||||
api_expires_at: API Token 過期時間
|
||||
extend_days: 延長天數
|
||||
|
||||
Returns:
|
||||
datetime: 內部 Token 過期時間
|
||||
"""
|
||||
if api_expires_at:
|
||||
# 基於 API Token 過期時間延長
|
||||
return api_expires_at + timedelta(days=extend_days)
|
||||
else:
|
||||
# 如果沒有 API 過期時間，從現在開始計算
|
||||
return datetime.utcnow() + timedelta(days=extend_days)
|
||||
Reference in New Issue
Block a user