590 lines
20 KiB
Python
590 lines
20 KiB
Python
"""Authentication, durable lockout, database identity and session governance."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from functools import wraps
|
||
import hmac
|
||
import os
|
||
|
||
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"
|
||
)
|
||
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("[Security] 登入驗證僅於明確測試模式停用")
|
||
elif _DISABLE_LOGIN_REQUESTED:
|
||
print("[Security] 已拒絕 DISABLE_LOGIN：僅測試環境可使用登入繞過")
|
||
|
||
|
||
def get_identity_session():
|
||
"""Late-bind the database session so tests and workers share one seam."""
|
||
from database.manager import get_session
|
||
|
||
return get_session()
|
||
|
||
|
||
def get_client_ip() -> str:
|
||
return resolve_client_ip(
|
||
request.remote_addr,
|
||
request.headers.get("X-Forwarded-For"),
|
||
request.headers.get("X-Real-IP"),
|
||
)
|
||
|
||
|
||
def _user_agent() -> str:
|
||
return (request.headers.get("User-Agent") or "")[:256]
|
||
|
||
|
||
def _lockout_status(username: str | None, client_ip: str) -> tuple[bool, int, int]:
|
||
from services.user_service import UserService
|
||
|
||
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 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
|
||
|
||
|
||
def record_login_failure(ip: str) -> bool:
|
||
"""Compatibility wrapper; new login flow records username and audit detail."""
|
||
from services.user_service import UserService
|
||
|
||
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()
|
||
|
||
|
||
def clear_login_attempts(_ip: str) -> None:
|
||
"""A durable success event supersedes older failures; no delete is required."""
|
||
|
||
|
||
def validate_password_strength(password):
|
||
if not password:
|
||
return False, "密碼不能為空"
|
||
if len(password) < 8:
|
||
return False, "密碼長度至少需要 8 個字元"
|
||
if not any(c.isalpha() for c in password):
|
||
return False, "密碼必須包含英文字母"
|
||
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):
|
||
@wraps(f)
|
||
def decorated_view(*args, **kwargs):
|
||
state = _session_state()
|
||
if not state["ok"]:
|
||
return _denied_response(state)
|
||
return f(*args, **kwargs)
|
||
|
||
return decorated_view
|
||
|
||
|
||
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
|
||
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()
|
||
if not expected:
|
||
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 f(*args, **kwargs)
|
||
|
||
return decorated_view
|
||
|
||
|
||
def get_current_user():
|
||
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": 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))}")
|
||
|
||
def decorator(f):
|
||
@wraps(f)
|
||
def decorated_view(*args, **kwargs):
|
||
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):
|
||
return role_required("admin")(f)
|
||
|
||
|
||
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),
|
||
})
|
||
|
||
|
||
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",
|
||
})
|
||
|
||
|
||
def init_auth_routes(app):
|
||
"""Register database-first login and audited logout routes."""
|
||
|
||
@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"
|
||
|
||
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
|
||
|
||
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
|
||
|
||
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:
|
||
return render_template(
|
||
"login.html",
|
||
error="請輸入密碼",
|
||
auth_mode=mode,
|
||
username=username,
|
||
), 400
|
||
|
||
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
|
||
)
|
||
|
||
legacy_valid = (
|
||
legacy_auth_can_authorize(mode)
|
||
and _legacy_password_matches(input_password)
|
||
)
|
||
database_valid = bool(
|
||
db_user and database_auth_can_authorize(mode)
|
||
)
|
||
|
||
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
|
||
|
||
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"))
|
||
|
||
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"))
|
||
|
||
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
|
||
|
||
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
|
||
|
||
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()
|
||
|
||
@app.route("/logout")
|
||
def logout():
|
||
if session.get("logged_in"):
|
||
_audit_session_event(LoginHistory.STATUS_LOGOUT, "user_logout")
|
||
session.clear()
|
||
return redirect(url_for("login"))
|
||
|
||
|
||
print("[Security] Auth identity governance module loaded")
|