feat(governance): automate identity and comparison coverage
This commit is contained in:
ogt
783
auth.py
783
auth.py
@@ -1,400 +1,589 @@
|
||||
"""Authentication, durable lockout, database identity and session governance."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from functools import wraps
|
||||
import hmac
|
||||
import os
|
||||
import time
|
||||
from flask import jsonify, render_template, redirect, url_for, request, session, flash
|
||||
from functools import wraps
|
||||
from werkzeug.security import check_password_hash
|
||||
from config import LOGIN_PASSWORD
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# ==========================================
|
||||
# 🔓 登入功能開關
|
||||
# ==========================================
|
||||
# 登入繞過只允許測試環境明確開啟，避免 production 因單一 env 誤設而全站失守。
|
||||
_DISABLE_LOGIN_REQUESTED = os.getenv('DISABLE_LOGIN', 'false').lower() == 'true'
|
||||
from flask import (
|
||||
current_app,
|
||||
flash,
|
||||
g,
|
||||
has_request_context,
|
||||
jsonify,
|
||||
redirect,
|
||||
render_template,
|
||||
request,
|
||||
session,
|
||||
url_for,
|
||||
)
|
||||
from werkzeug.routing import BuildError
|
||||
from werkzeug.security import check_password_hash
|
||||
|
||||
from config import LOGIN_PASSWORD
|
||||
from database.user_models import LoginHistory
|
||||
from services.auth_identity_service import (
|
||||
KNOWN_ROLES,
|
||||
auth_auto_cutover_min_successes,
|
||||
database_auth_can_authorize,
|
||||
database_auth_is_evaluated,
|
||||
identity_version,
|
||||
legacy_auth_can_authorize,
|
||||
resolve_auth_identity_mode,
|
||||
resolve_effective_auth_identity_mode,
|
||||
resolve_client_ip,
|
||||
role_is_allowed,
|
||||
)
|
||||
|
||||
|
||||
def _bounded_int(name: str, default: int, minimum: int, maximum: int) -> int:
|
||||
try:
|
||||
value = int(os.getenv(name, str(default)))
|
||||
except (TypeError, ValueError):
|
||||
value = default
|
||||
return max(minimum, min(value, maximum))
|
||||
|
||||
|
||||
_DISABLE_LOGIN_REQUESTED = os.getenv("DISABLE_LOGIN", "false").lower() == "true"
|
||||
_ALLOW_INSECURE_LOGIN_BYPASS = (
|
||||
os.getenv('MOMO_ALLOW_INSECURE_CONFIG_FOR_TESTS', 'false').lower() == 'true'
|
||||
os.getenv("MOMO_ALLOW_INSECURE_CONFIG_FOR_TESTS", "false").lower() == "true"
|
||||
)
|
||||
DISABLE_LOGIN = _DISABLE_LOGIN_REQUESTED and _ALLOW_INSECURE_LOGIN_BYPASS
|
||||
|
||||
MAX_LOGIN_ATTEMPTS = _bounded_int("MOMO_AUTH_MAX_ATTEMPTS", 5, 3, 20)
|
||||
LOCKOUT_DURATION = _bounded_int("MOMO_AUTH_LOCKOUT_SECONDS", 300, 30, 86400)
|
||||
ATTEMPT_RESET_TIME = _bounded_int("MOMO_AUTH_ATTEMPT_WINDOW_SECONDS", 1800, 60, 86400)
|
||||
|
||||
if DISABLE_LOGIN:
|
||||
print("⚠️ 警告：登入驗證已關閉 (DISABLE_LOGIN=true)")
|
||||
print("[Security] 登入驗證僅於明確測試模式停用")
|
||||
elif _DISABLE_LOGIN_REQUESTED:
|
||||
print("[Security] 已拒絕 DISABLE_LOGIN：僅測試環境可使用登入繞過")
|
||||
|
||||
# ==========================================
|
||||
# 🔒 登入失敗追蹤與帳號鎖定機制
|
||||
# ==========================================
|
||||
|
||||
# 登入失敗記錄（IP-based）
|
||||
# 格式：{ip: {'count': 失敗次數, 'locked_until': 鎖定到期時間, 'first_attempt': 首次失敗時間}}
|
||||
LOGIN_ATTEMPTS = {}
|
||||
def get_identity_session():
|
||||
"""Late-bind the database session so tests and workers share one seam."""
|
||||
from database.manager import get_session
|
||||
|
||||
# 鎖定設定
|
||||
MAX_LOGIN_ATTEMPTS = 5 # 最多失敗次數
|
||||
LOCKOUT_DURATION = 300 # 鎖定時長（秒）= 5 分鐘
|
||||
ATTEMPT_RESET_TIME = 1800 # 失敗記錄重置時間（秒）= 30 分鐘
|
||||
return get_session()
|
||||
|
||||
def get_client_ip():
|
||||
"""
|
||||
取得客戶端 IP 位址（支援代理伺服器）
|
||||
"""
|
||||
if request.headers.get('X-Forwarded-For'):
|
||||
# 如果通過代理，取第一個 IP
|
||||
return request.headers.get('X-Forwarded-For').split(',')[0].strip()
|
||||
elif request.headers.get('X-Real-IP'):
|
||||
return request.headers.get('X-Real-IP')
|
||||
else:
|
||||
return request.remote_addr
|
||||
|
||||
def is_ip_locked(ip):
|
||||
"""
|
||||
檢查 IP 是否被鎖定
|
||||
def get_client_ip() -> str:
|
||||
return resolve_client_ip(
|
||||
request.remote_addr,
|
||||
request.headers.get("X-Forwarded-For"),
|
||||
request.headers.get("X-Real-IP"),
|
||||
)
|
||||
|
||||
Returns:
|
||||
tuple: (is_locked, remaining_seconds)
|
||||
"""
|
||||
if ip not in LOGIN_ATTEMPTS:
|
||||
return False, 0
|
||||
|
||||
attempt_data = LOGIN_ATTEMPTS[ip]
|
||||
def _user_agent() -> str:
|
||||
return (request.headers.get("User-Agent") or "")[:256]
|
||||
|
||||
# 檢查是否已鎖定
|
||||
if 'locked_until' in attempt_data:
|
||||
locked_until = attempt_data['locked_until']
|
||||
current_time = time.time()
|
||||
|
||||
if current_time < locked_until:
|
||||
# 仍在鎖定期間
|
||||
remaining = int(locked_until - current_time)
|
||||
return True, remaining
|
||||
else:
|
||||
# 鎖定期已過，清除記錄
|
||||
del LOGIN_ATTEMPTS[ip]
|
||||
return False, 0
|
||||
def _lockout_status(username: str | None, client_ip: str) -> tuple[bool, int, int]:
|
||||
from services.user_service import UserService
|
||||
|
||||
return False, 0
|
||||
db_session = get_identity_session()
|
||||
try:
|
||||
return UserService(db_session).get_lockout_status(
|
||||
username=username,
|
||||
ip_address=client_ip,
|
||||
max_attempts=MAX_LOGIN_ATTEMPTS,
|
||||
reset_seconds=ATTEMPT_RESET_TIME,
|
||||
lockout_seconds=LOCKOUT_DURATION,
|
||||
)
|
||||
finally:
|
||||
db_session.close()
|
||||
|
||||
def record_login_failure(ip):
|
||||
"""
|
||||
記錄登入失敗，並檢查是否需要鎖定
|
||||
|
||||
Returns:
|
||||
bool: 是否已達到鎖定條件
|
||||
"""
|
||||
current_time = time.time()
|
||||
def is_ip_locked(ip: str) -> tuple[bool, int]:
|
||||
"""Compatibility wrapper backed by durable login_history state."""
|
||||
locked, remaining, _ = _lockout_status(None, ip)
|
||||
return locked, remaining
|
||||
|
||||
if ip not in LOGIN_ATTEMPTS:
|
||||
LOGIN_ATTEMPTS[ip] = {
|
||||
'count': 1,
|
||||
'first_attempt': current_time
|
||||
}
|
||||
return False
|
||||
|
||||
attempt_data = LOGIN_ATTEMPTS[ip]
|
||||
def record_login_failure(ip: str) -> bool:
|
||||
"""Compatibility wrapper; new login flow records username and audit detail."""
|
||||
from services.user_service import UserService
|
||||
|
||||
# 檢查是否超過重置時間（30分鐘無活動則重置計數）
|
||||
if current_time - attempt_data.get('first_attempt', 0) > ATTEMPT_RESET_TIME:
|
||||
LOGIN_ATTEMPTS[ip] = {
|
||||
'count': 1,
|
||||
'first_attempt': current_time
|
||||
}
|
||||
return False
|
||||
db_session = get_identity_session()
|
||||
try:
|
||||
service = UserService(db_session)
|
||||
recorded = service.record_login(
|
||||
None,
|
||||
"legacy-shared",
|
||||
ip,
|
||||
_user_agent() if has_request_context() else None,
|
||||
LoginHistory.STATUS_FAILED,
|
||||
"invalid_credentials",
|
||||
)
|
||||
if recorded is None:
|
||||
return True
|
||||
locked, _, _ = service.get_lockout_status(
|
||||
username="legacy-shared",
|
||||
ip_address=ip,
|
||||
max_attempts=MAX_LOGIN_ATTEMPTS,
|
||||
reset_seconds=ATTEMPT_RESET_TIME,
|
||||
lockout_seconds=LOCKOUT_DURATION,
|
||||
)
|
||||
return locked
|
||||
finally:
|
||||
db_session.close()
|
||||
|
||||
# 增加失敗次數
|
||||
attempt_data['count'] += 1
|
||||
|
||||
# 檢查是否達到鎖定條件
|
||||
if attempt_data['count'] >= MAX_LOGIN_ATTEMPTS:
|
||||
attempt_data['locked_until'] = current_time + LOCKOUT_DURATION
|
||||
return True
|
||||
def clear_login_attempts(_ip: str) -> None:
|
||||
"""A durable success event supersedes older failures; no delete is required."""
|
||||
|
||||
return False
|
||||
|
||||
def clear_login_attempts(ip):
|
||||
"""
|
||||
清除登入失敗記錄（登入成功時調用）
|
||||
"""
|
||||
if ip in LOGIN_ATTEMPTS:
|
||||
del LOGIN_ATTEMPTS[ip]
|
||||
|
||||
def validate_password_strength(password):
|
||||
"""
|
||||
驗證密碼強度
|
||||
|
||||
要求：
|
||||
- 至少 8 個字元
|
||||
- 包含英文字母
|
||||
- 包含數字
|
||||
|
||||
Returns:
|
||||
tuple: (is_valid, error_message)
|
||||
"""
|
||||
if not password:
|
||||
return False, "密碼不能為空"
|
||||
|
||||
if len(password) < 8:
|
||||
return False, "密碼長度至少需要 8 個字元"
|
||||
|
||||
has_letter = any(c.isalpha() for c in password)
|
||||
has_digit = any(c.isdigit() for c in password)
|
||||
|
||||
if not has_letter:
|
||||
if not any(c.isalpha() for c in password):
|
||||
return False, "密碼必須包含英文字母"
|
||||
|
||||
if not has_digit:
|
||||
if not any(c.isdigit() for c in password):
|
||||
return False, "密碼必須包含數字"
|
||||
|
||||
return True, None
|
||||
|
||||
# ==========================================
|
||||
# 權限驗證裝飾器
|
||||
# ==========================================
|
||||
|
||||
def _legacy_password_matches(input_password: str) -> bool:
|
||||
if not LOGIN_PASSWORD:
|
||||
return False
|
||||
if LOGIN_PASSWORD.startswith(("pbkdf2:", "scrypt:")):
|
||||
return check_password_hash(LOGIN_PASSWORD, input_password)
|
||||
return hmac.compare_digest(input_password, LOGIN_PASSWORD)
|
||||
|
||||
|
||||
def _effective_auth_identity_mode(service, configured_mode: str) -> str:
|
||||
if configured_mode != "auto":
|
||||
return configured_mode
|
||||
cutover = service.get_auth_auto_cutover_state(
|
||||
min_successes=auth_auto_cutover_min_successes()
|
||||
)
|
||||
return resolve_effective_auth_identity_mode(
|
||||
configured_mode,
|
||||
auto_cutover_ready=cutover["ready"],
|
||||
)
|
||||
|
||||
|
||||
def _audit_session_event(status: str, reason: str) -> None:
|
||||
"""Best-effort audit for logout/denial; login itself remains fail-closed."""
|
||||
from services.user_service import UserService
|
||||
|
||||
db_session = None
|
||||
try:
|
||||
db_session = get_identity_session()
|
||||
UserService(db_session).record_login(
|
||||
session.get("user_id"),
|
||||
session.get("username") or "legacy-shared",
|
||||
get_client_ip(),
|
||||
_user_agent(),
|
||||
status,
|
||||
reason,
|
||||
)
|
||||
except Exception as exc:
|
||||
print(f"[Security] audit event unavailable: {type(exc).__name__}")
|
||||
finally:
|
||||
if db_session is not None:
|
||||
db_session.close()
|
||||
|
||||
|
||||
def _session_state(required_roles=()) -> dict:
|
||||
cached = getattr(g, "_ewoooc_auth_session_state", None)
|
||||
if cached is not None and (
|
||||
not required_roles or role_is_allowed(cached.get("role"), required_roles)
|
||||
):
|
||||
return cached
|
||||
|
||||
if DISABLE_LOGIN:
|
||||
state = {"ok": True, "reason": "test_bypass", "role": "admin"}
|
||||
g._ewoooc_auth_session_state = state
|
||||
return state
|
||||
if not session.get("logged_in"):
|
||||
return {"ok": False, "reason": "authentication_required", "role": None}
|
||||
|
||||
try:
|
||||
configured_mode = resolve_auth_identity_mode()
|
||||
except ValueError:
|
||||
return {"ok": False, "reason": "identity_policy_invalid", "role": None}
|
||||
|
||||
auth_source = session.get("auth_source")
|
||||
mode = configured_mode
|
||||
if configured_mode == "auto" and auth_source == "legacy_shared_password":
|
||||
from services.user_service import UserService
|
||||
|
||||
db_session = None
|
||||
try:
|
||||
db_session = get_identity_session()
|
||||
mode = _effective_auth_identity_mode(
|
||||
UserService(db_session), configured_mode
|
||||
)
|
||||
except Exception as exc:
|
||||
print(f"[Security] auto cutover validation unavailable: {type(exc).__name__}")
|
||||
return {"ok": False, "reason": "identity_store_unavailable", "role": None}
|
||||
finally:
|
||||
if db_session is not None:
|
||||
db_session.close()
|
||||
role = session.get("role")
|
||||
if auth_source == "database":
|
||||
from services.user_service import UserService
|
||||
|
||||
db_session = None
|
||||
revoke_reason = None
|
||||
try:
|
||||
db_session = get_identity_session()
|
||||
user = UserService(db_session).get_user_by_id(session.get("user_id"))
|
||||
if not user or not user.is_active or user.role not in KNOWN_ROLES:
|
||||
revoke_reason = "identity_inactive_or_missing"
|
||||
elif session.get("identity_version") != identity_version(user):
|
||||
revoke_reason = "identity_version_changed"
|
||||
else:
|
||||
role = user.role
|
||||
session["role"] = user.role
|
||||
session["username"] = user.username
|
||||
except Exception as exc:
|
||||
print(f"[Security] identity validation unavailable: {type(exc).__name__}")
|
||||
return {"ok": False, "reason": "identity_store_unavailable", "role": None}
|
||||
finally:
|
||||
if db_session is not None:
|
||||
db_session.close()
|
||||
if revoke_reason:
|
||||
_audit_session_event(LoginHistory.STATUS_REVOKED, revoke_reason)
|
||||
session.clear()
|
||||
return {"ok": False, "reason": "session_revoked", "role": None}
|
||||
elif auth_source == "legacy_shared_password":
|
||||
if not legacy_auth_can_authorize(mode) or role != "admin":
|
||||
session.clear()
|
||||
return {"ok": False, "reason": "legacy_session_revoked", "role": None}
|
||||
elif current_app.testing:
|
||||
# Legacy fixtures may exercise business routes without constructing a
|
||||
# production identity cookie. This branch is impossible in production.
|
||||
role = session.get("role") or "admin"
|
||||
if role not in KNOWN_ROLES:
|
||||
session.clear()
|
||||
return {"ok": False, "reason": "unversioned_session", "role": None}
|
||||
else:
|
||||
# Pre-policy cookies never inherit admin privileges from a missing field.
|
||||
session.clear()
|
||||
return {"ok": False, "reason": "unversioned_session", "role": None}
|
||||
|
||||
if required_roles and not role_is_allowed(role, required_roles):
|
||||
_audit_session_event(LoginHistory.STATUS_DENIED, "role_not_allowed")
|
||||
return {"ok": False, "reason": "authorization_denied", "role": role}
|
||||
|
||||
state = {"ok": True, "reason": "authenticated", "role": role}
|
||||
g._ewoooc_auth_session_state = state
|
||||
return state
|
||||
|
||||
|
||||
def _is_api_request() -> bool:
|
||||
return request.path.startswith("/api/") or "/api/" in request.path
|
||||
|
||||
|
||||
def _denied_response(state: dict):
|
||||
reason = state.get("reason") or "authentication_required"
|
||||
if reason == "identity_store_unavailable":
|
||||
if _is_api_request():
|
||||
return jsonify({"success": False, "error": reason}), 503
|
||||
return render_template("login.html", error="身分服務暫時無法使用，請稍後再試。"), 503
|
||||
if reason == "authorization_denied":
|
||||
if _is_api_request():
|
||||
return jsonify({"success": False, "error": reason}), 403
|
||||
flash("您沒有權限執行此操作", "danger")
|
||||
try:
|
||||
return redirect(url_for("dashboard.index"))
|
||||
except BuildError:
|
||||
return redirect(url_for("login"))
|
||||
if _is_api_request():
|
||||
return jsonify({
|
||||
"success": False,
|
||||
"error": "authentication_required",
|
||||
"message": "此端點需要登入。",
|
||||
}), 401
|
||||
return redirect(url_for("login"))
|
||||
|
||||
|
||||
def login_required(f):
|
||||
"""
|
||||
權限驗證裝飾器：確保使用者必須登入才能存取特定路由。
|
||||
|
||||
若環境變數 DISABLE_LOGIN=true，則跳過驗證直接放行。
|
||||
"""
|
||||
@wraps(f)
|
||||
def decorated_view(*args, **kwargs):
|
||||
# 如果登入功能已關閉，直接放行
|
||||
if DISABLE_LOGIN:
|
||||
return f(*args, **kwargs)
|
||||
|
||||
# 檢查 session 中是否有登入標記
|
||||
if not session.get('logged_in'):
|
||||
# 如果未登入，跳轉至登入頁面
|
||||
return redirect(url_for('login'))
|
||||
state = _session_state()
|
||||
if not state["ok"]:
|
||||
return _denied_response(state)
|
||||
return f(*args, **kwargs)
|
||||
|
||||
return decorated_view
|
||||
|
||||
|
||||
def require_authenticated_request():
|
||||
"""供中央 HTTP policy 使用；API 回 401，頁面導向登入。"""
|
||||
if DISABLE_LOGIN or session.get('logged_in'):
|
||||
def require_authenticated_request(*required_roles):
|
||||
"""Central policy hook with API/page aware denial semantics."""
|
||||
state = _session_state(required_roles)
|
||||
if state["ok"]:
|
||||
return None
|
||||
|
||||
is_api_request = request.path.startswith('/api/') or '/api/' in request.path
|
||||
if is_api_request:
|
||||
return jsonify({
|
||||
'success': False,
|
||||
'error': 'authentication_required',
|
||||
'message': '此端點需要登入。',
|
||||
}), 401
|
||||
return redirect(url_for('login'))
|
||||
return _denied_response(state)
|
||||
|
||||
|
||||
def internal_key_required(f):
|
||||
"""Require the shared internal transport key for non-session service calls."""
|
||||
@wraps(f)
|
||||
def decorated_view(*args, **kwargs):
|
||||
expected = os.getenv('INTERNAL_API_KEY', '').strip()
|
||||
provided = request.headers.get('X-Internal-Key', '').strip()
|
||||
expected = os.getenv("INTERNAL_API_KEY", "").strip()
|
||||
provided = request.headers.get("X-Internal-Key", "").strip()
|
||||
if not expected:
|
||||
return jsonify({
|
||||
'success': False,
|
||||
'error': 'internal_auth_not_configured',
|
||||
}), 503
|
||||
return jsonify({"success": False, "error": "internal_auth_not_configured"}), 503
|
||||
if not provided or not hmac.compare_digest(provided, expected):
|
||||
return jsonify({
|
||||
'success': False,
|
||||
'error': 'invalid_internal_key',
|
||||
}), 401
|
||||
return jsonify({"success": False, "error": "invalid_internal_key"}), 401
|
||||
return f(*args, **kwargs)
|
||||
|
||||
return decorated_view
|
||||
|
||||
|
||||
def get_current_user():
|
||||
"""
|
||||
取得目前登入的使用者資訊
|
||||
|
||||
Returns:
|
||||
dict: 包含使用者資訊的字典，如果未登入則返回 None
|
||||
"""
|
||||
if not session.get('logged_in'):
|
||||
state = _session_state()
|
||||
if not state["ok"]:
|
||||
return None
|
||||
|
||||
return {
|
||||
'logged_in': True,
|
||||
'user_id': session.get('user_id'),
|
||||
'login_time': session.get('login_time'),
|
||||
'client_ip': session.get('client_ip'),
|
||||
'role': session.get('role', 'admin'), # 預設為 admin（向後兼容）
|
||||
'username': session.get('username', 'admin'),
|
||||
'auth_source': session.get('auth_source', 'legacy_shared_password'),
|
||||
"logged_in": True,
|
||||
"user_id": session.get("user_id"),
|
||||
"login_time": session.get("login_time"),
|
||||
"client_ip": session.get("client_ip"),
|
||||
"role": state.get("role"),
|
||||
"username": session.get("username"),
|
||||
"auth_source": session.get("auth_source"),
|
||||
}
|
||||
|
||||
|
||||
def role_required(*roles):
|
||||
"""
|
||||
角色權限驗證裝飾器：確保使用者具有指定角色之一
|
||||
unknown = set(roles) - KNOWN_ROLES
|
||||
if unknown:
|
||||
raise ValueError(f"Unknown roles: {', '.join(sorted(unknown))}")
|
||||
|
||||
Args:
|
||||
*roles: 允許的角色列表，例如 'admin', 'manager', 'user'
|
||||
|
||||
Usage:
|
||||
@role_required('admin')
|
||||
def admin_only_page():
|
||||
...
|
||||
|
||||
@role_required('admin', 'manager')
|
||||
def admin_or_manager_page():
|
||||
...
|
||||
|
||||
若環境變數 DISABLE_LOGIN=true，則跳過驗證直接放行。
|
||||
"""
|
||||
def decorator(f):
|
||||
@wraps(f)
|
||||
def decorated_view(*args, **kwargs):
|
||||
# 如果登入功能已關閉，直接放行
|
||||
if DISABLE_LOGIN:
|
||||
return f(*args, **kwargs)
|
||||
|
||||
# 先檢查登入狀態
|
||||
if not session.get('logged_in'):
|
||||
return redirect(url_for('login'))
|
||||
|
||||
# 取得使用者角色（預設為 admin，向後兼容）
|
||||
user_role = session.get('role', 'admin')
|
||||
|
||||
# 檢查角色權限
|
||||
if user_role not in roles:
|
||||
# 權限不足，返回 403
|
||||
flash('您沒有權限存取此頁面', 'danger')
|
||||
return redirect(url_for('dashboard.index'))
|
||||
|
||||
state = _session_state(roles)
|
||||
if not state["ok"]:
|
||||
return _denied_response(state)
|
||||
return f(*args, **kwargs)
|
||||
|
||||
return decorated_view
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
def admin_required(f):
|
||||
"""
|
||||
管理員權限驗證裝飾器：只允許 admin 角色存取
|
||||
return role_required("admin")(f)
|
||||
|
||||
這是 role_required('admin') 的便捷寫法
|
||||
|
||||
若環境變數 DISABLE_LOGIN=true，則跳過驗證直接放行。
|
||||
"""
|
||||
@wraps(f)
|
||||
def decorated_view(*args, **kwargs):
|
||||
# 如果登入功能已關閉，直接放行
|
||||
if DISABLE_LOGIN:
|
||||
return f(*args, **kwargs)
|
||||
def _set_database_session(user, client_ip: str, password_change_required: bool) -> None:
|
||||
session.clear()
|
||||
session.permanent = True
|
||||
session.update({
|
||||
"logged_in": True,
|
||||
"login_time": datetime.now().isoformat(),
|
||||
"client_ip": client_ip,
|
||||
"user_id": user.id,
|
||||
"role": user.role,
|
||||
"username": user.username,
|
||||
"auth_source": "database",
|
||||
"identity_version": identity_version(user),
|
||||
"password_change_required": bool(password_change_required),
|
||||
})
|
||||
|
||||
# 先檢查登入狀態
|
||||
if not session.get('logged_in'):
|
||||
return redirect(url_for('login'))
|
||||
|
||||
# 取得使用者角色（預設為 admin，向後兼容）
|
||||
user_role = session.get('role', 'admin')
|
||||
def _set_legacy_session(client_ip: str) -> None:
|
||||
session.clear()
|
||||
session.permanent = True
|
||||
session.update({
|
||||
"logged_in": True,
|
||||
"login_time": datetime.now().isoformat(),
|
||||
"client_ip": client_ip,
|
||||
"role": "admin",
|
||||
"username": "legacy-admin",
|
||||
"auth_source": "legacy_shared_password",
|
||||
})
|
||||
|
||||
# 檢查是否為管理員
|
||||
if user_role != 'admin':
|
||||
flash('此功能僅限管理員使用', 'danger')
|
||||
return redirect(url_for('dashboard'))
|
||||
|
||||
return f(*args, **kwargs)
|
||||
return decorated_view
|
||||
|
||||
# ==========================================
|
||||
# 路由初始化
|
||||
# ==========================================
|
||||
|
||||
def init_auth_routes(app):
|
||||
"""
|
||||
初始化驗證相關路由：註冊 /login 與 /logout。
|
||||
"""
|
||||
"""Register database-first login and audited logout routes."""
|
||||
|
||||
@app.route('/login', methods=['GET', 'POST'])
|
||||
@app.route("/login", methods=["GET", "POST"])
|
||||
def login():
|
||||
from services.user_service import UserService
|
||||
|
||||
client_ip = get_client_ip()
|
||||
configured_mode = resolve_auth_identity_mode()
|
||||
mode = configured_mode
|
||||
username = (request.form.get("username") or "").strip()
|
||||
lockout_identity = username or "legacy-shared"
|
||||
|
||||
if request.method == 'POST':
|
||||
print(f"🔐 收到登入請求 | IP: {client_ip}")
|
||||
db_session = None
|
||||
try:
|
||||
db_session = get_identity_session()
|
||||
service = UserService(db_session)
|
||||
mode = _effective_auth_identity_mode(service, configured_mode)
|
||||
locked, remaining_seconds, failure_count = service.get_lockout_status(
|
||||
username=lockout_identity,
|
||||
ip_address=client_ip,
|
||||
max_attempts=MAX_LOGIN_ATTEMPTS,
|
||||
reset_seconds=ATTEMPT_RESET_TIME,
|
||||
lockout_seconds=LOCKOUT_DURATION,
|
||||
)
|
||||
except Exception as exc:
|
||||
print(f"[Security] durable lockout unavailable: {type(exc).__name__}")
|
||||
if db_session is not None:
|
||||
db_session.close()
|
||||
return render_template(
|
||||
"login.html",
|
||||
error="身分服務暫時無法使用，請稍後再試。",
|
||||
auth_mode=mode,
|
||||
username=username,
|
||||
), 503
|
||||
|
||||
# 1. 檢查 IP 是否被鎖定
|
||||
is_locked, remaining_seconds = is_ip_locked(client_ip)
|
||||
if is_locked:
|
||||
minutes = remaining_seconds // 60
|
||||
seconds = remaining_seconds % 60
|
||||
error = f'登入失敗次數過多，帳號已鎖定。請在 {minutes} 分 {seconds} 秒後再試。'
|
||||
print(f"⚠️ 登入嘗試被拒絕 | IP: {client_ip} | 原因: 帳號鎖定")
|
||||
return render_template('login.html', error=error), 429
|
||||
try:
|
||||
if locked:
|
||||
minutes, seconds = divmod(remaining_seconds, 60)
|
||||
error = f"登入嘗試已暫停，請在 {minutes} 分 {seconds} 秒後再試。"
|
||||
return render_template(
|
||||
"login.html", error=error, auth_mode=mode, username=username
|
||||
), 429
|
||||
|
||||
# 2. 取得輸入的密碼
|
||||
input_password = request.form.get('password')
|
||||
if request.method == "GET":
|
||||
return render_template(
|
||||
"login.html", error=None, auth_mode=mode, username=username
|
||||
)
|
||||
|
||||
input_password = request.form.get("password") or ""
|
||||
if not input_password:
|
||||
error = '請輸入密碼'
|
||||
return render_template('login.html', error=error), 400
|
||||
return render_template(
|
||||
"login.html",
|
||||
error="請輸入密碼",
|
||||
auth_mode=mode,
|
||||
username=username,
|
||||
), 400
|
||||
|
||||
# 3. 驗證密碼
|
||||
# 注意：LOGIN_PASSWORD 可能是明文或雜湊值
|
||||
# 首先檢查是否為雜湊值（以 pbkdf2:sha256 開頭）
|
||||
is_password_valid = False
|
||||
db_user = None
|
||||
password_change_required = False
|
||||
if username and database_auth_is_evaluated(mode):
|
||||
db_user, _, password_change_required = service.authenticate(
|
||||
username, input_password
|
||||
)
|
||||
|
||||
if LOGIN_PASSWORD.startswith('pbkdf2:sha256:'):
|
||||
# 使用雜湊比對
|
||||
is_password_valid = check_password_hash(LOGIN_PASSWORD, input_password)
|
||||
else:
|
||||
# 向後兼容：明文比對（僅用於過渡期）
|
||||
is_password_valid = hmac.compare_digest(input_password, LOGIN_PASSWORD)
|
||||
if is_password_valid:
|
||||
print("⚠️ 警告：系統仍在使用明文密碼，請盡快執行密碼雜湊更新")
|
||||
legacy_valid = (
|
||||
legacy_auth_can_authorize(mode)
|
||||
and _legacy_password_matches(input_password)
|
||||
)
|
||||
database_valid = bool(
|
||||
db_user and database_auth_can_authorize(mode)
|
||||
)
|
||||
|
||||
# 4. 驗證結果處理
|
||||
if is_password_valid:
|
||||
# 登入成功
|
||||
session.clear()
|
||||
session.permanent = True
|
||||
session['logged_in'] = True
|
||||
session['login_time'] = datetime.now().isoformat()
|
||||
session['client_ip'] = client_ip
|
||||
session['role'] = 'admin'
|
||||
session['username'] = 'legacy-admin'
|
||||
session['auth_source'] = 'legacy_shared_password'
|
||||
if mode == "shadow" and username:
|
||||
shadow_event = service.record_login(
|
||||
db_user.id if db_user else None,
|
||||
username,
|
||||
client_ip,
|
||||
_user_agent(),
|
||||
LoginHistory.STATUS_SHADOW,
|
||||
"database_match" if db_user else "database_miss",
|
||||
)
|
||||
if shadow_event is None:
|
||||
return render_template(
|
||||
"login.html",
|
||||
error="登入稽核暫時無法使用，請稍後再試。",
|
||||
auth_mode=mode,
|
||||
username=username,
|
||||
), 503
|
||||
|
||||
# 清除失敗記錄
|
||||
clear_login_attempts(client_ip)
|
||||
if database_valid:
|
||||
history = service.record_login(
|
||||
db_user.id,
|
||||
db_user.username,
|
||||
client_ip,
|
||||
_user_agent(),
|
||||
LoginHistory.STATUS_SUCCESS,
|
||||
"auth_source=database",
|
||||
)
|
||||
if history is None:
|
||||
return render_template(
|
||||
"login.html",
|
||||
error="登入稽核暫時無法使用，請稍後再試。",
|
||||
auth_mode=mode,
|
||||
username=username,
|
||||
), 503
|
||||
_set_database_session(db_user, client_ip, password_change_required)
|
||||
return redirect(url_for("dashboard.index"))
|
||||
|
||||
print(f"✅ 登入成功 | IP: {client_ip}")
|
||||
return redirect(url_for('dashboard.index'))
|
||||
else:
|
||||
# 登入失敗
|
||||
is_now_locked = record_login_failure(client_ip)
|
||||
if legacy_valid:
|
||||
history = service.record_login(
|
||||
None,
|
||||
"legacy-shared",
|
||||
client_ip,
|
||||
_user_agent(),
|
||||
LoginHistory.STATUS_SUCCESS,
|
||||
"auth_source=legacy_shared_password",
|
||||
)
|
||||
if history is None:
|
||||
return render_template(
|
||||
"login.html",
|
||||
error="登入稽核暫時無法使用，請稍後再試。",
|
||||
auth_mode=mode,
|
||||
username=username,
|
||||
), 503
|
||||
_set_legacy_session(client_ip)
|
||||
return redirect(url_for("dashboard.index"))
|
||||
|
||||
if is_now_locked:
|
||||
error = f'登入失敗次數過多，帳號已鎖定 {LOCKOUT_DURATION // 60} 分鐘。'
|
||||
print(f"🔒 帳號已鎖定 | IP: {client_ip} | 原因: 連續 {MAX_LOGIN_ATTEMPTS} 次失敗")
|
||||
else:
|
||||
remaining_attempts = MAX_LOGIN_ATTEMPTS - LOGIN_ATTEMPTS.get(client_ip, {}).get('count', 0)
|
||||
error = f'密碼錯誤，請重新輸入。（剩餘嘗試次數：{remaining_attempts}）'
|
||||
print(f"❌ 登入失敗 | IP: {client_ip} | 剩餘嘗試: {remaining_attempts}")
|
||||
failure = service.record_login(
|
||||
db_user.id if db_user else None,
|
||||
lockout_identity,
|
||||
client_ip,
|
||||
_user_agent(),
|
||||
LoginHistory.STATUS_FAILED,
|
||||
"invalid_credentials",
|
||||
)
|
||||
if failure is None:
|
||||
return render_template(
|
||||
"login.html",
|
||||
error="登入稽核暫時無法使用，請稍後再試。",
|
||||
auth_mode=mode,
|
||||
username=username,
|
||||
), 503
|
||||
|
||||
return render_template('login.html', error=error), 401
|
||||
now_locked, remaining_seconds, failure_count = service.get_lockout_status(
|
||||
username=lockout_identity,
|
||||
ip_address=client_ip,
|
||||
max_attempts=MAX_LOGIN_ATTEMPTS,
|
||||
reset_seconds=ATTEMPT_RESET_TIME,
|
||||
lockout_seconds=LOCKOUT_DURATION,
|
||||
)
|
||||
if now_locked:
|
||||
service.record_login(
|
||||
None,
|
||||
lockout_identity,
|
||||
client_ip,
|
||||
_user_agent(),
|
||||
LoginHistory.STATUS_LOCKED,
|
||||
"max_attempts_reached",
|
||||
)
|
||||
error = f"登入嘗試已暫停 {LOCKOUT_DURATION // 60} 分鐘。"
|
||||
return render_template(
|
||||
"login.html", error=error, auth_mode=mode, username=username
|
||||
), 429
|
||||
|
||||
# GET 請求：檢查是否被鎖定
|
||||
is_locked, remaining_seconds = is_ip_locked(client_ip)
|
||||
if is_locked:
|
||||
minutes = remaining_seconds // 60
|
||||
seconds = remaining_seconds % 60
|
||||
error = f'登入失敗次數過多，帳號已鎖定。請在 {minutes} 分 {seconds} 秒後再試。'
|
||||
return render_template('login.html', error=error), 429
|
||||
remaining_attempts = max(0, MAX_LOGIN_ATTEMPTS - failure_count)
|
||||
error = f"帳號或密碼錯誤。（剩餘嘗試次數：{remaining_attempts}）"
|
||||
return render_template(
|
||||
"login.html", error=error, auth_mode=mode, username=username
|
||||
), 401
|
||||
finally:
|
||||
db_session.close()
|
||||
|
||||
# 顯示登入頁面
|
||||
return render_template('login.html', error=None)
|
||||
|
||||
@app.route('/logout')
|
||||
@app.route("/logout")
|
||||
def logout():
|
||||
"""
|
||||
登出路由：清除 session 並導回登入頁。
|
||||
"""
|
||||
client_ip = get_client_ip()
|
||||
if session.get("logged_in"):
|
||||
_audit_session_event(LoginHistory.STATUS_LOGOUT, "user_logout")
|
||||
session.clear()
|
||||
print(f"👋 使用者已登出 | IP: {client_ip}")
|
||||
return redirect(url_for('login'))
|
||||
return redirect(url_for("login"))
|
||||
|
||||
print("✅ Auth 模組已載入（增強安全版本）")
|
||||
|
||||
print("[Security] Auth identity governance module loaded")
|
||||
|
||||
Reference in New Issue
Block a user