661 lines
21 KiB
Python
661 lines
21 KiB
Python
"""
|
||
用戶服務模組
|
||
|
||
提供：
|
||
- 用戶 CRUD 操作
|
||
- 用戶認證
|
||
- 登入歷史記錄
|
||
"""
|
||
|
||
from datetime import datetime, timedelta
|
||
from sqlalchemy import func, or_
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy.exc import IntegrityError
|
||
|
||
from database.user_models import User, LoginHistory
|
||
from services.password_service import (
|
||
hash_password,
|
||
verify_password,
|
||
validate_password_complexity,
|
||
is_password_expired,
|
||
)
|
||
|
||
|
||
class UserService:
|
||
"""用戶服務類"""
|
||
|
||
def __init__(self, db_session: Session):
|
||
"""
|
||
初始化用戶服務
|
||
|
||
Args:
|
||
db_session: SQLAlchemy Session 物件
|
||
"""
|
||
self.db = db_session
|
||
|
||
def _add_identity_audit(self, user, status, audit_context=None, change=None):
|
||
"""Add a no-secret audit event to the caller's current transaction."""
|
||
context = audit_context or {}
|
||
try:
|
||
actor_id = int(context.get('actor_id'))
|
||
actor = f"actor_id={actor_id}"
|
||
except (TypeError, ValueError):
|
||
actor = "actor_id=system"
|
||
|
||
reason_parts = [actor]
|
||
if change:
|
||
reason_parts.append(f"change={change}")
|
||
self.db.add(LoginHistory(
|
||
user_id=user.id,
|
||
username_attempted=(user.username or '')[:50] or None,
|
||
login_time=datetime.now(),
|
||
ip_address=str(context.get('ip_address') or '')[:45] or None,
|
||
user_agent=str(context.get('user_agent') or '')[:256] or None,
|
||
status=status[:20],
|
||
failure_reason=';'.join(reason_parts)[:100],
|
||
))
|
||
|
||
# ==========================================
|
||
# 用戶 CRUD 操作
|
||
# ==========================================
|
||
|
||
def create_user(
|
||
self,
|
||
username,
|
||
password,
|
||
role='user',
|
||
email=None,
|
||
display_name=None,
|
||
created_by=None,
|
||
audit_context=None,
|
||
):
|
||
"""
|
||
建立新用戶
|
||
|
||
Args:
|
||
username: 帳號名稱
|
||
password: 明文密碼
|
||
role: 角色 (admin/manager/user)
|
||
email: 電子郵件 (可選)
|
||
display_name: 顯示名稱 (可選)
|
||
created_by: 建立者 ID (可選)
|
||
|
||
Returns:
|
||
tuple: (user, error_message)
|
||
- user: User 物件（成功時）或 None（失敗時）
|
||
- error_message: 錯誤訊息（成功時為 None）
|
||
"""
|
||
username = (username or '').strip()
|
||
if not username:
|
||
return None, "帳號名稱不能為空"
|
||
|
||
# 驗證密碼複雜度
|
||
is_valid, error = validate_password_complexity(password)
|
||
if not is_valid:
|
||
return None, f"密碼不符合要求：{error}"
|
||
|
||
# 驗證角色
|
||
if role not in User.ROLES:
|
||
return None, f"無效的角色：{role}"
|
||
|
||
# 檢查帳號是否已存在
|
||
existing = self.get_user_by_username(username)
|
||
if existing:
|
||
return None, f"帳號 '{username}' 已存在"
|
||
|
||
# 檢查 email 是否已存在
|
||
if email:
|
||
existing_email = self.db.query(User).filter(User.email == email).first()
|
||
if existing_email:
|
||
return None, f"電子郵件 '{email}' 已被使用"
|
||
|
||
try:
|
||
user = User(
|
||
username=username,
|
||
password_hash=hash_password(password),
|
||
role=role,
|
||
email=email,
|
||
display_name=display_name or username,
|
||
is_active=True,
|
||
password_changed_at=datetime.now(),
|
||
created_at=datetime.now(),
|
||
created_by=created_by
|
||
)
|
||
self.db.add(user)
|
||
self.db.flush()
|
||
self._add_identity_audit(
|
||
user,
|
||
LoginHistory.STATUS_USER_CREATED,
|
||
audit_context,
|
||
change=f"role:{role}",
|
||
)
|
||
self.db.commit()
|
||
self.db.refresh(user)
|
||
|
||
# 自動套用角色預設權限（admin 以外的角色）
|
||
if role != 'admin':
|
||
try:
|
||
from services.permission_service import PermissionService
|
||
PermissionService.init_user_permissions_by_role(
|
||
user.id,
|
||
role,
|
||
created_by,
|
||
audit_context=audit_context,
|
||
)
|
||
except Exception as perm_err:
|
||
print(f"⚠️ 初始化用戶權限失敗: {perm_err}")
|
||
|
||
return user, None
|
||
|
||
except IntegrityError as e:
|
||
self.db.rollback()
|
||
return None, f"資料庫錯誤：{str(e)}"
|
||
except Exception as e:
|
||
self.db.rollback()
|
||
return None, f"建立用戶失敗：{str(e)}"
|
||
|
||
def get_user_by_id(self, user_id):
|
||
"""
|
||
根據 ID 取得用戶
|
||
|
||
Args:
|
||
user_id: 用戶 ID
|
||
|
||
Returns:
|
||
User 或 None
|
||
"""
|
||
return self.db.query(User).filter(User.id == user_id).first()
|
||
|
||
def get_user_by_username(self, username):
|
||
"""
|
||
根據帳號名稱取得用戶
|
||
|
||
Args:
|
||
username: 帳號名稱
|
||
|
||
Returns:
|
||
User 或 None
|
||
"""
|
||
normalized = (username or '').strip().lower()
|
||
if not normalized:
|
||
return None
|
||
return self.db.query(User).filter(func.lower(User.username) == normalized).first()
|
||
|
||
def get_all_users(self, include_inactive=False):
|
||
"""
|
||
取得所有用戶
|
||
|
||
Args:
|
||
include_inactive: 是否包含停用的用戶
|
||
|
||
Returns:
|
||
list[User]
|
||
"""
|
||
query = self.db.query(User)
|
||
if not include_inactive:
|
||
query = query.filter(User.is_active == True)
|
||
return query.order_by(User.username).all()
|
||
|
||
def update_user(self, user_id, audit_context=None, **kwargs):
|
||
"""
|
||
更新用戶資料
|
||
|
||
Args:
|
||
user_id: 用戶 ID
|
||
**kwargs: 要更新的欄位
|
||
|
||
Returns:
|
||
tuple: (success, error_message)
|
||
"""
|
||
user = self.get_user_by_id(user_id)
|
||
if not user:
|
||
return False, "用戶不存在"
|
||
|
||
allowed_fields = ['email', 'display_name', 'role', 'is_active']
|
||
|
||
requested = {key: value for key, value in kwargs.items() if key in allowed_fields}
|
||
if 'role' in requested and requested['role'] not in User.ROLES:
|
||
return False, f"無效的角色：{requested['role']}"
|
||
if 'is_active' in requested and not isinstance(requested['is_active'], bool):
|
||
return False, "is_active 必須是布林值"
|
||
if requested.get('email'):
|
||
existing = self.db.query(User).filter(
|
||
User.email == requested['email'],
|
||
User.id != user_id,
|
||
).first()
|
||
if existing:
|
||
return False, f"電子郵件 '{requested['email']}' 已被使用"
|
||
|
||
changed_fields = sorted(
|
||
key for key, value in requested.items()
|
||
if getattr(user, key) != value
|
||
)
|
||
if not changed_fields:
|
||
return True, None
|
||
|
||
try:
|
||
for key in changed_fields:
|
||
setattr(user, key, requested[key])
|
||
user.updated_at = datetime.now()
|
||
self._add_identity_audit(
|
||
user,
|
||
LoginHistory.STATUS_IDENTITY_CHANGED,
|
||
audit_context,
|
||
change=f"fields:{','.join(changed_fields)}",
|
||
)
|
||
self.db.commit()
|
||
return True, None
|
||
|
||
except Exception as e:
|
||
self.db.rollback()
|
||
return False, f"更新失敗：{str(e)}"
|
||
|
||
def change_password(self, user_id, old_password, new_password, audit_context=None):
|
||
"""
|
||
變更密碼
|
||
|
||
Args:
|
||
user_id: 用戶 ID
|
||
old_password: 舊密碼
|
||
new_password: 新密碼
|
||
|
||
Returns:
|
||
tuple: (success, error_message)
|
||
"""
|
||
user = self.get_user_by_id(user_id)
|
||
if not user:
|
||
return False, "用戶不存在"
|
||
|
||
# 驗證舊密碼
|
||
if not verify_password(user.password_hash, old_password):
|
||
return False, "舊密碼不正確"
|
||
|
||
# 驗證新密碼複雜度
|
||
is_valid, error = validate_password_complexity(new_password)
|
||
if not is_valid:
|
||
return False, f"新密碼不符合要求：{error}"
|
||
|
||
# 檢查新密碼是否與舊密碼相同
|
||
if verify_password(user.password_hash, new_password):
|
||
return False, "新密碼不能與舊密碼相同"
|
||
|
||
try:
|
||
user.password_hash = hash_password(new_password)
|
||
user.password_changed_at = datetime.now()
|
||
user.updated_at = datetime.now()
|
||
self._add_identity_audit(
|
||
user,
|
||
LoginHistory.STATUS_PASSWORD_CHANGED,
|
||
audit_context,
|
||
change="self_service",
|
||
)
|
||
self.db.commit()
|
||
return True, None
|
||
|
||
except Exception as e:
|
||
self.db.rollback()
|
||
return False, f"變更密碼失敗：{str(e)}"
|
||
|
||
def reset_password(self, user_id, new_password, admin_id=None, audit_context=None):
|
||
"""
|
||
重設密碼（管理員功能）
|
||
|
||
Args:
|
||
user_id: 用戶 ID
|
||
new_password: 新密碼
|
||
admin_id: 執行重設的管理員 ID
|
||
|
||
Returns:
|
||
tuple: (success, error_message)
|
||
"""
|
||
user = self.get_user_by_id(user_id)
|
||
if not user:
|
||
return False, "用戶不存在"
|
||
|
||
# 驗證新密碼複雜度
|
||
is_valid, error = validate_password_complexity(new_password)
|
||
if not is_valid:
|
||
return False, f"新密碼不符合要求：{error}"
|
||
|
||
try:
|
||
user.password_hash = hash_password(new_password)
|
||
user.password_changed_at = datetime.now()
|
||
user.updated_at = datetime.now()
|
||
event_context = dict(audit_context or {})
|
||
if admin_id is not None:
|
||
event_context.setdefault('actor_id', admin_id)
|
||
self._add_identity_audit(
|
||
user,
|
||
LoginHistory.STATUS_PASSWORD_CHANGED,
|
||
event_context,
|
||
change="admin_reset",
|
||
)
|
||
self.db.commit()
|
||
|
||
print(f"🔐 密碼重設 | User: {user.username} | Admin ID: {admin_id}")
|
||
return True, None
|
||
|
||
except Exception as e:
|
||
self.db.rollback()
|
||
return False, f"重設密碼失敗：{str(e)}"
|
||
|
||
def delete_user(self, user_id, audit_context=None):
|
||
"""
|
||
刪除用戶（實際是停用）
|
||
|
||
Args:
|
||
user_id: 用戶 ID
|
||
|
||
Returns:
|
||
tuple: (success, error_message)
|
||
"""
|
||
user = self.get_user_by_id(user_id)
|
||
if not user:
|
||
return False, "用戶不存在"
|
||
|
||
# 不允許刪除最後一個 admin
|
||
if user.role == User.ROLE_ADMIN:
|
||
admin_count = self.db.query(User).filter(
|
||
User.role == User.ROLE_ADMIN,
|
||
User.is_active == True
|
||
).count()
|
||
if admin_count <= 1:
|
||
return False, "無法刪除：系統必須保留至少一個管理員帳號"
|
||
|
||
try:
|
||
user.is_active = False
|
||
user.updated_at = datetime.now()
|
||
self._add_identity_audit(
|
||
user,
|
||
LoginHistory.STATUS_USER_DISABLED,
|
||
audit_context,
|
||
change="active:false",
|
||
)
|
||
self.db.commit()
|
||
return True, None
|
||
|
||
except Exception as e:
|
||
self.db.rollback()
|
||
return False, f"刪除用戶失敗：{str(e)}"
|
||
|
||
# ==========================================
|
||
# 用戶認證
|
||
# ==========================================
|
||
|
||
def authenticate(self, username, password):
|
||
"""
|
||
驗證用戶登入
|
||
|
||
Args:
|
||
username: 帳號名稱
|
||
password: 密碼
|
||
|
||
Returns:
|
||
tuple: (user, error_message, need_password_change)
|
||
- user: User 物件（成功時）或 None（失敗時）
|
||
- error_message: 錯誤訊息（成功時為 None）
|
||
- need_password_change: 是否需要變更密碼
|
||
"""
|
||
user = self.get_user_by_username(username)
|
||
|
||
if not user:
|
||
return None, "帳號或密碼錯誤", False
|
||
|
||
if not user.is_active:
|
||
return None, "帳號已停用，請聯繫管理員", False
|
||
|
||
if not verify_password(user.password_hash, password):
|
||
return None, "帳號或密碼錯誤", False
|
||
|
||
# 檢查密碼是否過期
|
||
is_expired, days_until = is_password_expired(user.password_changed_at)
|
||
|
||
return user, None, is_expired
|
||
|
||
# ==========================================
|
||
# 登入歷史記錄
|
||
# ==========================================
|
||
|
||
def record_login(self, user_id, username_attempted, ip_address, user_agent, status, failure_reason=None):
|
||
"""
|
||
記錄登入歷史
|
||
|
||
Args:
|
||
user_id: 用戶 ID (可為 None，用於記錄失敗的登入嘗試)
|
||
username_attempted: 嘗試登入的帳號名稱
|
||
ip_address: IP 位址
|
||
user_agent: User-Agent 字串
|
||
status: 狀態 (success/failed/locked)
|
||
failure_reason: 失敗原因
|
||
|
||
Returns:
|
||
LoginHistory 物件
|
||
"""
|
||
try:
|
||
history = LoginHistory(
|
||
user_id=user_id,
|
||
username_attempted=(username_attempted or '')[:50] or None,
|
||
login_time=datetime.now(),
|
||
ip_address=(ip_address or '')[:45] or None,
|
||
user_agent=user_agent[:256] if user_agent else None, # 限制長度
|
||
status=(status or '')[:20],
|
||
failure_reason=(failure_reason or '')[:100] or None,
|
||
)
|
||
self.db.add(history)
|
||
self.db.commit()
|
||
return history
|
||
|
||
except Exception as e:
|
||
self.db.rollback()
|
||
print(f"⚠️ 記錄登入歷史失敗: {e}")
|
||
return None
|
||
|
||
def get_lockout_status(
|
||
self,
|
||
username=None,
|
||
ip_address=None,
|
||
*,
|
||
max_attempts=5,
|
||
reset_seconds=1800,
|
||
lockout_seconds=300,
|
||
now=None,
|
||
):
|
||
"""Return a durable lockout decision backed by login_history."""
|
||
current_time = now or datetime.now()
|
||
username = (username or '').strip().lower()
|
||
ip_address = (ip_address or '').strip()
|
||
scope = []
|
||
if username:
|
||
scope.append(func.lower(LoginHistory.username_attempted) == username)
|
||
if ip_address:
|
||
scope.append(LoginHistory.ip_address == ip_address)
|
||
if not scope:
|
||
return False, 0, 0
|
||
|
||
scope_filter = or_(*scope)
|
||
since = current_time - timedelta(seconds=max(1, int(reset_seconds)))
|
||
|
||
latest_success = self.db.query(func.max(LoginHistory.login_time)).filter(
|
||
scope_filter,
|
||
LoginHistory.status == LoginHistory.STATUS_SUCCESS,
|
||
).scalar()
|
||
if latest_success and latest_success > since:
|
||
since = latest_success
|
||
|
||
latest_lock = self.db.query(func.max(LoginHistory.login_time)).filter(
|
||
scope_filter,
|
||
LoginHistory.status == LoginHistory.STATUS_LOCKED,
|
||
LoginHistory.login_time >= since,
|
||
).scalar()
|
||
if latest_lock:
|
||
lock_expires_at = latest_lock + timedelta(seconds=max(1, int(lockout_seconds)))
|
||
if current_time < lock_expires_at:
|
||
remaining = max(1, int((lock_expires_at - current_time).total_seconds()))
|
||
return True, remaining, int(max_attempts)
|
||
since = max(since, lock_expires_at)
|
||
|
||
failures = self.db.query(LoginHistory).filter(
|
||
scope_filter,
|
||
LoginHistory.status == LoginHistory.STATUS_FAILED,
|
||
LoginHistory.login_time >= since,
|
||
).order_by(LoginHistory.login_time.desc()).all()
|
||
failure_count = len(failures)
|
||
if failure_count < int(max_attempts):
|
||
return False, 0, failure_count
|
||
|
||
locked_until = failures[0].login_time + timedelta(seconds=max(1, int(lockout_seconds)))
|
||
if current_time < locked_until:
|
||
remaining = max(1, int((locked_until - current_time).total_seconds()))
|
||
return True, remaining, failure_count
|
||
return False, 0, 0
|
||
|
||
def get_identity_inventory(self):
|
||
"""Return non-secret identity and audit counts for runtime governance."""
|
||
role_rows = self.db.query(
|
||
User.role,
|
||
User.is_active,
|
||
func.count(User.id),
|
||
).group_by(User.role, User.is_active).all()
|
||
mutation_statuses = (
|
||
LoginHistory.STATUS_IDENTITY_CHANGED,
|
||
LoginHistory.STATUS_PASSWORD_CHANGED,
|
||
LoginHistory.STATUS_USER_CREATED,
|
||
LoginHistory.STATUS_USER_DISABLED,
|
||
LoginHistory.STATUS_PERMISSION_CHANGED,
|
||
)
|
||
mutation_rows = self.db.query(
|
||
LoginHistory.status,
|
||
func.count(LoginHistory.id),
|
||
).filter(
|
||
LoginHistory.status.in_(mutation_statuses),
|
||
).group_by(LoginHistory.status).all()
|
||
return {
|
||
'users_total': self.db.query(func.count(User.id)).scalar() or 0,
|
||
'active_admin_count': self.db.query(func.count(User.id)).filter(
|
||
User.role == User.ROLE_ADMIN,
|
||
User.is_active.is_(True),
|
||
).scalar() or 0,
|
||
'users_by_role': [
|
||
{
|
||
'role': role or 'unassigned',
|
||
'is_active': bool(is_active),
|
||
'count': count,
|
||
}
|
||
for role, is_active, count in role_rows
|
||
],
|
||
'login_history_events': self.db.query(func.count(LoginHistory.id)).scalar() or 0,
|
||
'identity_mutation_events': sum(count for _, count in mutation_rows),
|
||
'identity_mutation_events_by_status': {
|
||
status: count for status, count in mutation_rows
|
||
},
|
||
}
|
||
|
||
def get_auth_auto_cutover_state(self, *, min_successes=2):
|
||
"""Return the durable admin-login gate used by automatic cutover."""
|
||
threshold = max(2, min(int(min_successes), 10))
|
||
event_count, distinct_admins = self.db.query(
|
||
func.count(LoginHistory.id),
|
||
func.count(func.distinct(LoginHistory.user_id)),
|
||
).join(
|
||
User,
|
||
LoginHistory.user_id == User.id,
|
||
).filter(
|
||
LoginHistory.status == LoginHistory.STATUS_SUCCESS,
|
||
LoginHistory.failure_reason == 'auth_source=database',
|
||
User.role == User.ROLE_ADMIN,
|
||
User.is_active.is_(True),
|
||
).one()
|
||
event_count = int(event_count or 0)
|
||
distinct_admins = int(distinct_admins or 0)
|
||
return {
|
||
'policy': 'durable_admin_login_receipt_threshold',
|
||
'minimum_successes': threshold,
|
||
'database_admin_successes': event_count,
|
||
'distinct_database_admins': distinct_admins,
|
||
'ready': event_count >= threshold and distinct_admins >= 1,
|
||
'manual_review_required': False,
|
||
}
|
||
|
||
def get_login_history(self, user_id=None, limit=100):
|
||
"""
|
||
取得登入歷史記錄
|
||
|
||
Args:
|
||
user_id: 用戶 ID (可選，None 表示取得所有用戶)
|
||
limit: 最大筆數
|
||
|
||
Returns:
|
||
list[LoginHistory]
|
||
"""
|
||
query = self.db.query(LoginHistory)
|
||
if user_id:
|
||
query = query.filter(LoginHistory.user_id == user_id)
|
||
return query.order_by(LoginHistory.login_time.desc()).limit(limit).all()
|
||
|
||
def get_recent_failed_logins(self, ip_address=None, minutes=30):
|
||
"""
|
||
取得最近的登入失敗記錄
|
||
|
||
Args:
|
||
ip_address: IP 位址 (可選)
|
||
minutes: 時間範圍（分鐘）
|
||
|
||
Returns:
|
||
list[LoginHistory]
|
||
"""
|
||
from datetime import timedelta
|
||
|
||
since = datetime.now() - timedelta(minutes=minutes)
|
||
query = self.db.query(LoginHistory).filter(
|
||
LoginHistory.login_time >= since,
|
||
LoginHistory.status == LoginHistory.STATUS_FAILED
|
||
)
|
||
|
||
if ip_address:
|
||
query = query.filter(LoginHistory.ip_address == ip_address)
|
||
|
||
return query.order_by(LoginHistory.login_time.desc()).all()
|
||
|
||
|
||
def create_initial_admin(db_session, password=None):
|
||
"""
|
||
建立初始管理員帳號
|
||
|
||
Args:
|
||
db_session: SQLAlchemy Session
|
||
password: 初始密碼（若不傳，從 INITIAL_ADMIN_PASSWORD 環境變數讀取）
|
||
|
||
Returns:
|
||
tuple: (success, message)
|
||
"""
|
||
import os
|
||
if password is None:
|
||
password = os.getenv('INITIAL_ADMIN_PASSWORD')
|
||
if not password:
|
||
raise RuntimeError(
|
||
"INITIAL_ADMIN_PASSWORD 環境變數未設定，無法建立初始管理員。"
|
||
"請在 .env 或 Docker 環境設定此值。"
|
||
)
|
||
service = UserService(db_session)
|
||
|
||
# 檢查是否已有管理員
|
||
existing_admin = db_session.query(User).filter(User.role == User.ROLE_ADMIN).first()
|
||
if existing_admin:
|
||
return False, f"已存在管理員帳號：{existing_admin.username}"
|
||
|
||
# 建立預設管理員
|
||
user, error = service.create_user(
|
||
username='admin',
|
||
password=password,
|
||
role=User.ROLE_ADMIN,
|
||
display_name='系統管理員',
|
||
)
|
||
|
||
if error:
|
||
return False, f"建立管理員失敗：{error}"
|
||
|
||
return True, f"管理員帳號 'admin' 建立成功"
|
||
|
||
|
||
print("✅ User service 已載入")
|