改用API驗證

app/models/sys_user.py

@@ -0,0 +1,297 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+系統使用者模型
+專門用於記錄帳號密碼和登入相關資訊
+
+Author: PANJIT IT Team
+Created: 2025-10-01
+"""
+
+import json
+from datetime import datetime, timedelta
+from typing import Optional, Dict, Any
+from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime, JSON, Enum as SQLEnum, BigInteger
+from werkzeug.security import generate_password_hash, check_password_hash
+from app import db
+from app.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class SysUser(db.Model):
+    """系統使用者模型 - 專門處理帳號密碼和登入記錄"""
+    __tablename__ = 'sys_user'
+
+    id = Column(BigInteger, primary_key=True)
+
+    # 帳號資訊
+    username = Column(String(255), nullable=False, unique=True, comment='登入帳號')
+    password_hash = Column(String(512), comment='密碼雜湊 (如果需要本地儲存)')
+    email = Column(String(255), nullable=False, unique=True, comment='電子郵件')
+    display_name = Column(String(255), comment='顯示名稱')
+
+    # API 認證資訊
+    api_user_id = Column(String(255), comment='API 回傳的使用者 ID')
+    api_access_token = Column(Text, comment='API 回傳的 access_token')
+    api_token_expires_at = Column(DateTime, comment='API Token 過期時間')
+
+    # 登入相關
+    auth_method = Column(SQLEnum('API', 'LDAP', name='sys_user_auth_method'),
+                        default='API', comment='認證方式')
+    last_login_at = Column(DateTime, comment='最後登入時間')
+    last_login_ip = Column(String(45), comment='最後登入 IP')
+    login_count = Column(Integer, default=0, comment='登入次數')
+    login_success_count = Column(Integer, default=0, comment='成功登入次數')
+    login_fail_count = Column(Integer, default=0, comment='失敗登入次數')
+
+    # 帳號狀態
+    is_active = Column(Boolean, default=True, comment='是否啟用')
+    is_locked = Column(Boolean, default=False, comment='是否鎖定')
+    locked_until = Column(DateTime, comment='鎖定至何時')
+
+    # 審計欄位
+    created_at = Column(DateTime, default=datetime.utcnow, comment='建立時間')
+    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, comment='更新時間')
+
+    def __repr__(self):
+        return f'<SysUser {self.username}>'
+
+    def to_dict(self) -> Dict[str, Any]:
+        """轉換為字典格式"""
+        return {
+            'id': self.id,
+            'username': self.username,
+            'email': self.email,
+            'display_name': self.display_name,
+            'api_user_id': self.api_user_id,
+            'auth_method': self.auth_method,
+            'last_login_at': self.last_login_at.isoformat() if self.last_login_at else None,
+            'login_count': self.login_count,
+            'login_success_count': self.login_success_count,
+            'login_fail_count': self.login_fail_count,
+            'is_active': self.is_active,
+            'is_locked': self.is_locked,
+            'api_token_expires_at': self.api_token_expires_at.isoformat() if self.api_token_expires_at else None,
+            'created_at': self.created_at.isoformat() if self.created_at else None
+        }
+
+    @classmethod
+    def get_or_create(cls, email: str, **kwargs) -> 'SysUser':
+        """
+        取得或建立系統使用者 (方案A: 使用 email 作為主要識別鍵)
+
+        Args:
+            email: 電子郵件 (主要識別鍵)
+            **kwargs: 其他欄位
+
+        Returns:
+            SysUser: 系統使用者實例
+        """
+        try:
+            # 使用 email 作為主要識別 (專門用於登入記錄)
+            sys_user = cls.query.filter_by(email=email).first()
+
+            if sys_user:
+                # 更新現有記錄
+                sys_user.username = kwargs.get('username', sys_user.username)  # API name (姓名+email)
+                sys_user.display_name = kwargs.get('display_name', sys_user.display_name)  # API name (姓名+email)
+                sys_user.api_user_id = kwargs.get('api_user_id', sys_user.api_user_id)  # Azure Object ID
+                sys_user.api_access_token = kwargs.get('api_access_token', sys_user.api_access_token)
+                sys_user.api_token_expires_at = kwargs.get('api_token_expires_at', sys_user.api_token_expires_at)
+                sys_user.auth_method = kwargs.get('auth_method', sys_user.auth_method)
+                sys_user.updated_at = datetime.utcnow()
+
+                logger.info(f"更新現有系統使用者: {email}")
+            else:
+                # 建立新記錄
+                sys_user = cls(
+                    username=kwargs.get('username', ''),      # API name (姓名+email 格式)
+                    email=email,                             # 純 email，主要識別鍵
+                    display_name=kwargs.get('display_name', ''),  # API name (姓名+email 格式)
+                    api_user_id=kwargs.get('api_user_id'),   # Azure Object ID
+                    api_access_token=kwargs.get('api_access_token'),
+                    api_token_expires_at=kwargs.get('api_token_expires_at'),
+                    auth_method=kwargs.get('auth_method', 'API'),
+                    login_count=0,
+                    login_success_count=0,
+                    login_fail_count=0
+                )
+                db.session.add(sys_user)
+                logger.info(f"建立新系統使用者: {email}")
+
+            db.session.commit()
+            return sys_user
+
+        except Exception as e:
+            db.session.rollback()
+            logger.error(f"取得或建立系統使用者失敗: {str(e)}")
+            raise
+
+    @classmethod
+    def get_by_email(cls, email: str) -> Optional['SysUser']:
+        """根據 email 查找系統使用者"""
+        return cls.query.filter_by(email=email).first()
+
+    def record_login_attempt(self, success: bool, ip_address: str = None, auth_method: str = None):
+        """
+        記錄登入嘗試
+
+        Args:
+            success: 是否成功
+            ip_address: IP 地址
+            auth_method: 認證方式
+        """
+        try:
+            self.login_count = (self.login_count or 0) + 1
+
+            if success:
+                self.login_success_count = (self.login_success_count or 0) + 1
+                self.last_login_at = datetime.utcnow()
+                self.last_login_ip = ip_address
+                if auth_method:
+                    self.auth_method = auth_method
+
+                # 成功登入時解除鎖定
+                if self.is_locked:
+                    self.is_locked = False
+                    self.locked_until = None
+
+            else:
+                self.login_fail_count = (self.login_fail_count or 0) + 1
+
+                # 檢查是否需要鎖定帳號 (連續失敗5次)
+                if self.login_fail_count >= 5:
+                    self.is_locked = True
+                    self.locked_until = datetime.utcnow() + timedelta(minutes=30)  # 鎖定30分鐘
+
+            self.updated_at = datetime.utcnow()
+            db.session.commit()
+
+        except Exception as e:
+            db.session.rollback()
+            logger.error(f"記錄登入嘗試失敗: {str(e)}")
+
+    def is_account_locked(self) -> bool:
+        """檢查帳號是否被鎖定"""
+        if not self.is_locked:
+            return False
+
+        # 檢查鎖定時間是否已過
+        if self.locked_until and datetime.utcnow() > self.locked_until:
+            self.is_locked = False
+            self.locked_until = None
+            db.session.commit()
+            return False
+
+        return True
+
+    def set_password(self, password: str):
+        """設置密碼雜湊 (如果需要本地儲存密碼)"""
+        self.password_hash = generate_password_hash(password)
+
+    def check_password(self, password: str) -> bool:
+        """檢查密碼 (如果有本地儲存密碼)"""
+        if not self.password_hash:
+            return False
+        return check_password_hash(self.password_hash, password)
+
+    def update_api_token(self, access_token: str, expires_at: datetime = None):
+        """更新 API Token"""
+        self.api_access_token = access_token
+        self.api_token_expires_at = expires_at
+        self.updated_at = datetime.utcnow()
+        db.session.commit()
+
+    def is_api_token_valid(self) -> bool:
+        """檢查 API Token 是否有效"""
+        if not self.api_access_token or not self.api_token_expires_at:
+            return False
+        return datetime.utcnow() < self.api_token_expires_at
+
+
+class LoginLog(db.Model):
+    """登入記錄模型"""
+    __tablename__ = 'login_logs'
+
+    id = Column(BigInteger, primary_key=True)
+
+    # 基本資訊
+    username = Column(String(255), nullable=False, comment='登入帳號')
+    auth_method = Column(SQLEnum('API', 'LDAP', name='login_log_auth_method'),
+                        nullable=False, comment='認證方式')
+
+    # 登入結果
+    login_success = Column(Boolean, nullable=False, comment='是否成功')
+    error_message = Column(Text, comment='錯誤訊息（失敗時）')
+
+    # 環境資訊
+    ip_address = Column(String(45), comment='IP 地址')
+    user_agent = Column(Text, comment='瀏覽器資訊')
+
+    # API 回應 (可選，用於除錯)
+    api_response_summary = Column(JSON, comment='API 回應摘要')
+
+    # 時間
+    login_at = Column(DateTime, default=datetime.utcnow, comment='登入時間')
+
+    def __repr__(self):
+        return f'<LoginLog {self.username}:{self.auth_method}:{self.login_success}>'
+
+    @classmethod
+    def create_log(cls, username: str, auth_method: str, login_success: bool,
+                   error_message: str = None, ip_address: str = None,
+                   user_agent: str = None, api_response_summary: Dict = None) -> 'LoginLog':
+        """
+        建立登入記錄
+
+        Args:
+            username: 使用者帳號
+            auth_method: 認證方式
+            login_success: 是否成功
+            error_message: 錯誤訊息
+            ip_address: IP 地址
+            user_agent: 瀏覽器資訊
+            api_response_summary: API 回應摘要
+
+        Returns:
+            LoginLog: 登入記錄
+        """
+        try:
+            log = cls(
+                username=username,
+                auth_method=auth_method,
+                login_success=login_success,
+                error_message=error_message,
+                ip_address=ip_address,
+                user_agent=user_agent,
+                api_response_summary=api_response_summary
+            )
+
+            db.session.add(log)
+            db.session.commit()
+            return log
+
+        except Exception as e:
+            db.session.rollback()
+            logger.error(f"建立登入記錄失敗: {str(e)}")
+            return None
+
+    @classmethod
+    def get_recent_failed_attempts(cls, username: str, minutes: int = 15) -> int:
+        """
+        取得最近失敗的登入嘗試次數
+
+        Args:
+            username: 使用者帳號
+            minutes: 時間範圍（分鐘）
+
+        Returns:
+            int: 失敗次數
+        """
+        since = datetime.utcnow() - timedelta(minutes=minutes)
+        return cls.query.filter(
+            cls.username == username,
+            cls.login_success == False,
+            cls.login_at >= since
+        ).count()