599 lines
23 KiB
Python
599 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
from io import BytesIO
|
|
import time
|
|
import database
|
|
from app_config import get_config
|
|
from app_logger import get_logger
|
|
from db.utils import get_cst_now, get_cst_now_str
|
|
from flask import Blueprint, jsonify, request, send_file, session
|
|
from flask_login import current_user, login_required, login_user
|
|
from services.accounts_service import load_user_accounts
|
|
from services.models import User
|
|
from services.social_login import (
|
|
BIND_TOKEN_TTL_SECONDS,
|
|
PROVIDER_LABELS,
|
|
SocialLoginError,
|
|
SpaceProfile,
|
|
admin_social_config_out,
|
|
encrypt_social_appkey,
|
|
fetch_social_login_url,
|
|
fetch_space_profile,
|
|
new_bind_token,
|
|
normalize_social_endpoint,
|
|
parse_providers,
|
|
poll_social_scan,
|
|
provider_label,
|
|
public_social_config,
|
|
)
|
|
|
|
logger = get_logger("app")
|
|
|
|
api_social_bp = Blueprint("api_social", __name__)
|
|
|
|
|
|
def _get_json_payload() -> dict:
|
|
data = request.get_json(silent=True)
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
def _social_error(error: SocialLoginError):
|
|
return jsonify({"error": error.message}), error.status_code
|
|
|
|
|
|
def _allowed_redirect_hosts() -> set[str]:
|
|
hosts: set[str] = set()
|
|
host = request.headers.get("X-Forwarded-Host") or request.headers.get("Host") or ""
|
|
if host:
|
|
hosts.add(host.lower())
|
|
hostname = host.split(":", 1)[0].lower()
|
|
if hostname:
|
|
hosts.add(hostname)
|
|
return hosts
|
|
|
|
|
|
def _create_pending_from_profile(profile: SpaceProfile) -> dict:
|
|
expires_at = (get_cst_now() + timedelta(seconds=BIND_TOKEN_TTL_SECONDS)).strftime("%Y-%m-%d %H:%M:%S")
|
|
return database.create_social_pending_bind(
|
|
token=new_bind_token(),
|
|
provider=profile.provider,
|
|
social_uid=profile.social_uid,
|
|
nickname=profile.nickname,
|
|
avatar_url=profile.avatar_url,
|
|
expires_at=expires_at,
|
|
)
|
|
|
|
|
|
def _login_user_id(user_id: int) -> None:
|
|
user_obj = User(user_id)
|
|
login_user(user_obj)
|
|
load_user_accounts(user_id)
|
|
|
|
|
|
def _login_admin_id(admin_id: int) -> dict | None:
|
|
admin = database.get_admin_by_id(int(admin_id))
|
|
if not admin:
|
|
return None
|
|
session.pop("admin_id", None)
|
|
session.pop("admin_username", None)
|
|
session["admin_id"] = admin["id"]
|
|
session["admin_username"] = admin["username"]
|
|
session["admin_reauth_until"] = time.time() + int(get_config().ADMIN_REAUTH_WINDOW_SECONDS)
|
|
session.permanent = True
|
|
session.modified = True
|
|
return admin
|
|
|
|
|
|
def _binding_row(provider: str, binding: dict | None) -> dict:
|
|
return {
|
|
"provider": provider,
|
|
"provider_label": provider_label(provider),
|
|
"bound": bool(binding),
|
|
"nickname": (binding or {}).get("nickname") or "",
|
|
"avatar_url": (binding or {}).get("avatar_url") or "",
|
|
"last_login_at": (binding or {}).get("last_login_at"),
|
|
"created_at": (binding or {}).get("created_at"),
|
|
}
|
|
|
|
|
|
def _admin_binding_row(provider: str, binding: dict | None) -> dict:
|
|
return {
|
|
"provider": provider,
|
|
"provider_label": provider_label(provider),
|
|
"bound": bool(binding),
|
|
"nickname": (binding or {}).get("nickname") or "",
|
|
"avatar_url": (binding or {}).get("avatar_url") or "",
|
|
"last_login_at": (binding or {}).get("last_login_at"),
|
|
"created_at": (binding or {}).get("created_at"),
|
|
}
|
|
|
|
|
|
@api_social_bp.route("/api/auth/social/config", methods=["GET"])
|
|
def social_public_config():
|
|
return jsonify(public_social_config(database.get_system_config()))
|
|
|
|
|
|
@api_social_bp.route("/api/auth/social/login-url", methods=["POST"])
|
|
def social_login_url():
|
|
data = _get_json_payload()
|
|
provider = str(data.get("provider") or "").strip().lower()
|
|
mode = "bind" if str(data.get("mode") or "").strip().lower() == "bind" else "login"
|
|
redirect_uri = str(data.get("redirect_uri") or "").strip()
|
|
|
|
try:
|
|
result = fetch_social_login_url(
|
|
database.get_system_config(),
|
|
provider=provider,
|
|
mode=mode,
|
|
redirect_uri=redirect_uri,
|
|
allowed_hosts=_allowed_redirect_hosts(),
|
|
)
|
|
except SocialLoginError as error:
|
|
logger.warning(f"[social/login-url] provider={provider or '-'} mode={mode} failed: {error.message}")
|
|
return _social_error(error)
|
|
return jsonify(result)
|
|
|
|
|
|
@api_social_bp.route("/api/auth/social/poll", methods=["POST"])
|
|
def social_poll():
|
|
data = _get_json_payload()
|
|
provider = str(data.get("provider") or "").strip().lower()
|
|
state = str(data.get("state") or "").strip()
|
|
try:
|
|
result = poll_social_scan(database.get_system_config(), provider=provider, state=state)
|
|
except SocialLoginError as error:
|
|
logger.warning(f"[social/poll] provider={provider or '-'} failed: {error.message}")
|
|
return _social_error(error)
|
|
return jsonify(result)
|
|
|
|
|
|
@api_social_bp.route("/api/auth/social/callback", methods=["POST"])
|
|
def social_callback():
|
|
data = _get_json_payload()
|
|
provider = str(data.get("provider") or data.get("type") or "").strip().lower()
|
|
code = str(data.get("code") or "").strip()
|
|
mode = "bind" if str(data.get("mode") or "").strip().lower() == "bind" else "login"
|
|
|
|
try:
|
|
profile = fetch_space_profile(database.get_system_config(), provider=provider, code=code)
|
|
except SocialLoginError as error:
|
|
logger.warning(f"[social/callback] provider={provider or '-'} mode={mode} failed: {error.message}")
|
|
return _social_error(error)
|
|
|
|
binding = database.find_social_login_binding(profile.provider, profile.social_uid)
|
|
|
|
if binding:
|
|
if mode == "bind":
|
|
current_id = int(getattr(current_user, "id", 0) or 0)
|
|
if not current_id or int(binding.get("user_id") or 0) != current_id:
|
|
return jsonify({"error": "该第三方账号已绑定其他用户"}), 409
|
|
|
|
user = database.get_user_by_id(int(binding["user_id"]))
|
|
if not user or user.get("status") != "approved":
|
|
return jsonify({"error": "绑定账号不可用"}), 401
|
|
database.update_social_login_binding_profile(
|
|
int(binding["id"]),
|
|
nickname=profile.nickname,
|
|
avatar_url=profile.avatar_url,
|
|
)
|
|
_login_user_id(int(user["id"]))
|
|
return jsonify(
|
|
{
|
|
"success": True,
|
|
"mode": mode,
|
|
"provider": profile.provider,
|
|
"provider_label": provider_label(profile.provider),
|
|
"bound": True,
|
|
"username": user.get("username") or "",
|
|
}
|
|
)
|
|
|
|
pending = _create_pending_from_profile(profile)
|
|
return jsonify(
|
|
{
|
|
"success": True,
|
|
"mode": mode,
|
|
"provider": profile.provider,
|
|
"provider_label": provider_label(profile.provider),
|
|
"requires_register": mode == "login",
|
|
"requires_existing_login": mode == "bind",
|
|
"bind_token": pending.get("token"),
|
|
"expires_in": BIND_TOKEN_TTL_SECONDS,
|
|
"nickname": pending.get("nickname") or "",
|
|
"avatar_url": pending.get("avatar_url") or "",
|
|
}
|
|
)
|
|
|
|
|
|
@api_social_bp.route("/api/user/social-bindings", methods=["GET"])
|
|
@login_required
|
|
def list_social_bindings():
|
|
cfg = database.get_system_config()
|
|
providers = parse_providers(cfg.get("social_login_providers")) or list(PROVIDER_LABELS.keys())
|
|
existing = {
|
|
item["provider"]: item
|
|
for item in database.list_social_login_bindings(int(current_user.id))
|
|
}
|
|
return jsonify({"items": [_binding_row(provider, existing.get(provider)) for provider in providers]})
|
|
|
|
|
|
@api_social_bp.route("/api/user/social-bindings", methods=["POST"])
|
|
@login_required
|
|
def bind_social_account():
|
|
data = _get_json_payload()
|
|
token = str(data.get("bind_token") or "").strip()
|
|
pending = database.get_social_pending_bind(token)
|
|
if not pending:
|
|
return jsonify({"error": "绑定凭证已过期，请重新授权"}), 404
|
|
|
|
provider = str(pending.get("provider") or "").strip().lower()
|
|
social_uid = str(pending.get("social_uid") or "").strip()
|
|
existing_identity = database.find_social_login_binding(provider, social_uid)
|
|
if existing_identity and int(existing_identity.get("user_id") or 0) != int(current_user.id):
|
|
return jsonify({"error": "该第三方账号已绑定其他用户"}), 409
|
|
|
|
existing_provider = database.find_user_social_login_binding(int(current_user.id), provider)
|
|
if existing_provider and str(existing_provider.get("social_uid") or "") != social_uid:
|
|
return jsonify({"error": f"当前账号已绑定{provider_label(provider)}"}), 409
|
|
|
|
binding = database.upsert_social_login_binding(
|
|
user_id=int(current_user.id),
|
|
provider=provider,
|
|
social_uid=social_uid,
|
|
nickname=pending.get("nickname") or "",
|
|
avatar_url=pending.get("avatar_url") or "",
|
|
)
|
|
if not binding:
|
|
return jsonify({"error": "该第三方账号已绑定其他用户"}), 409
|
|
|
|
database.delete_social_pending_bind(token)
|
|
return jsonify({"success": True, "item": _binding_row(provider, binding)})
|
|
|
|
|
|
@api_social_bp.route("/api/user/social-bindings/<provider>", methods=["DELETE"])
|
|
@login_required
|
|
def unbind_social_account(provider):
|
|
provider = str(provider or "").strip().lower()
|
|
if provider not in PROVIDER_LABELS:
|
|
return jsonify({"error": "不支持的登录方式"}), 400
|
|
if not database.delete_social_login_binding(int(current_user.id), provider):
|
|
return jsonify({"error": "绑定记录不存在"}), 404
|
|
return jsonify({"success": True})
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/social-login/config", methods=["GET"])
|
|
def admin_social_config():
|
|
from routes.decorators import admin_required
|
|
|
|
protected = admin_required(lambda: jsonify(admin_social_config_out(database.get_system_config())))
|
|
return protected()
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/admin-auth/social/login-url", methods=["POST"])
|
|
def admin_auth_social_login_url():
|
|
data = _get_json_payload()
|
|
provider = str(data.get("provider") or "").strip().lower()
|
|
redirect_uri = str(data.get("redirect_uri") or "").strip()
|
|
try:
|
|
result = fetch_social_login_url(
|
|
database.get_system_config(),
|
|
provider=provider,
|
|
mode="login",
|
|
redirect_uri=redirect_uri,
|
|
allowed_hosts=_allowed_redirect_hosts(),
|
|
)
|
|
except SocialLoginError as error:
|
|
logger.warning(f"[admin-auth/social/login-url] provider={provider or '-'} failed: {error.message}")
|
|
return _social_error(error)
|
|
return jsonify(result)
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/admin-auth/social/poll", methods=["POST"])
|
|
def admin_auth_social_poll():
|
|
data = _get_json_payload()
|
|
provider = str(data.get("provider") or "").strip().lower()
|
|
state = str(data.get("state") or "").strip()
|
|
try:
|
|
result = poll_social_scan(database.get_system_config(), provider=provider, state=state)
|
|
except SocialLoginError as error:
|
|
logger.warning(f"[admin-auth/social/poll] provider={provider or '-'} failed: {error.message}")
|
|
return _social_error(error)
|
|
return jsonify(result)
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/admin-auth/social/callback", methods=["POST"])
|
|
def admin_auth_social_callback():
|
|
data = _get_json_payload()
|
|
provider = str(data.get("provider") or data.get("type") or "").strip().lower()
|
|
code = str(data.get("code") or "").strip()
|
|
|
|
try:
|
|
profile = fetch_space_profile(database.get_system_config(), provider=provider, code=code)
|
|
except SocialLoginError as error:
|
|
logger.warning(f"[admin-auth/social/callback] provider={provider or '-'} failed: {error.message}")
|
|
return _social_error(error)
|
|
|
|
binding = database.find_admin_social_login_binding_by_identity(profile.provider, profile.social_uid)
|
|
if not binding:
|
|
return jsonify({"error": "该第三方账号未绑定管理员，请先使用账号密码登录后在设置中绑定"}), 404
|
|
|
|
admin = _login_admin_id(int(binding.get("admin_id") or 0))
|
|
if not admin:
|
|
return jsonify({"error": "绑定管理员账号不存在"}), 401
|
|
|
|
database.update_admin_social_login_binding_profile(
|
|
int(binding["id"]),
|
|
nickname=profile.nickname,
|
|
avatar_url=profile.avatar_url,
|
|
)
|
|
logger.info(f"[admin-auth/social/login] admin_id={admin['id']} provider={profile.provider}")
|
|
return jsonify(
|
|
{
|
|
"success": True,
|
|
"redirect": "/yuyx/admin",
|
|
"provider": profile.provider,
|
|
"provider_label": provider_label(profile.provider),
|
|
"username": admin.get("username") or "",
|
|
}
|
|
)
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/admin-auth/social/session", methods=["GET"])
|
|
def admin_auth_social_session():
|
|
admin_id = int(session.get("admin_id") or 0)
|
|
if not admin_id:
|
|
return jsonify({"authenticated": False}), 401
|
|
admin = database.get_admin_by_id(admin_id)
|
|
if not admin:
|
|
session.pop("admin_id", None)
|
|
session.pop("admin_username", None)
|
|
session.pop("admin_reauth_until", None)
|
|
session.modified = True
|
|
return jsonify({"authenticated": False}), 401
|
|
return jsonify({"authenticated": True, "username": admin.get("username") or ""})
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/admin-auth/social/qr", methods=["GET"])
|
|
def admin_auth_social_qr():
|
|
value = str(request.args.get("data") or "").strip()
|
|
if not value:
|
|
return jsonify({"error": "缺少二维码内容"}), 400
|
|
if len(value) > 2048:
|
|
return jsonify({"error": "二维码内容过长"}), 400
|
|
|
|
try:
|
|
import qrcode
|
|
except ImportError:
|
|
logger.error("[admin-auth/social/qr] qrcode package is not installed")
|
|
return jsonify({"error": "二维码组件未安装"}), 500
|
|
|
|
image = qrcode.make(value)
|
|
buffer = BytesIO()
|
|
image.save(buffer, format="PNG")
|
|
buffer.seek(0)
|
|
response = send_file(buffer, mimetype="image/png", max_age=0)
|
|
response.headers["Cache-Control"] = "no-store"
|
|
return response
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/admin/social-bindings", methods=["GET"])
|
|
def list_admin_social_bindings():
|
|
from routes.decorators import admin_required
|
|
|
|
@admin_required
|
|
def _inner():
|
|
cfg = database.get_system_config()
|
|
providers = parse_providers(cfg.get("social_login_providers")) or list(PROVIDER_LABELS.keys())
|
|
admin_id = int(session.get("admin_id") or 0)
|
|
existing = {
|
|
item["provider"]: item
|
|
for item in database.list_admin_social_login_bindings(admin_id)
|
|
}
|
|
public_cfg = public_social_config(cfg)
|
|
return jsonify(
|
|
{
|
|
"enabled": bool(public_cfg.get("enabled")),
|
|
"providers": providers,
|
|
"items": [_admin_binding_row(provider, existing.get(provider)) for provider in providers],
|
|
}
|
|
)
|
|
|
|
return _inner()
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/admin/social-login-url", methods=["POST"])
|
|
def admin_social_login_url():
|
|
from routes.decorators import admin_required
|
|
|
|
@admin_required
|
|
def _inner():
|
|
data = _get_json_payload()
|
|
provider = str(data.get("provider") or "").strip().lower()
|
|
redirect_uri = str(data.get("redirect_uri") or "").strip()
|
|
try:
|
|
result = fetch_social_login_url(
|
|
database.get_system_config(),
|
|
provider=provider,
|
|
mode="bind",
|
|
redirect_uri=redirect_uri,
|
|
allowed_hosts=_allowed_redirect_hosts(),
|
|
)
|
|
except SocialLoginError as error:
|
|
logger.warning(f"[admin/social/login-url] provider={provider or '-'} failed: {error.message}")
|
|
return _social_error(error)
|
|
return jsonify(result)
|
|
|
|
return _inner()
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/admin/social-poll", methods=["POST"])
|
|
def admin_social_poll():
|
|
from routes.decorators import admin_required
|
|
|
|
@admin_required
|
|
def _inner():
|
|
data = _get_json_payload()
|
|
provider = str(data.get("provider") or "").strip().lower()
|
|
state = str(data.get("state") or "").strip()
|
|
try:
|
|
result = poll_social_scan(database.get_system_config(), provider=provider, state=state)
|
|
except SocialLoginError as error:
|
|
logger.warning(f"[admin/social/poll] provider={provider or '-'} failed: {error.message}")
|
|
return _social_error(error)
|
|
return jsonify(result)
|
|
|
|
return _inner()
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/admin/social-bindings/<provider>/callback", methods=["POST"])
|
|
def bind_admin_social_callback(provider):
|
|
from routes.decorators import admin_required
|
|
|
|
@admin_required
|
|
def _inner():
|
|
data = _get_json_payload()
|
|
provider_value = str(data.get("provider") or provider or data.get("type") or "").strip().lower()
|
|
code = str(data.get("code") or "").strip()
|
|
admin_id = int(session.get("admin_id") or 0)
|
|
|
|
try:
|
|
profile = fetch_space_profile(database.get_system_config(), provider=provider_value, code=code)
|
|
except SocialLoginError as error:
|
|
logger.warning(f"[admin/social/callback] provider={provider_value or '-'} failed: {error.message}")
|
|
return _social_error(error)
|
|
|
|
existing_identity = database.find_admin_social_login_binding_by_identity(profile.provider, profile.social_uid)
|
|
if existing_identity and int(existing_identity.get("admin_id") or 0) != admin_id:
|
|
return jsonify({"error": "该第三方账号已绑定其他管理员"}), 409
|
|
|
|
existing_provider = database.find_admin_social_login_binding(admin_id, profile.provider)
|
|
if existing_provider and str(existing_provider.get("social_uid") or "") != profile.social_uid:
|
|
return jsonify({"error": f"当前管理员已绑定{provider_label(profile.provider)}"}), 409
|
|
|
|
binding = database.upsert_admin_social_login_binding(
|
|
admin_id=admin_id,
|
|
provider=profile.provider,
|
|
social_uid=profile.social_uid,
|
|
nickname=profile.nickname,
|
|
avatar_url=profile.avatar_url,
|
|
)
|
|
if not binding:
|
|
return jsonify({"error": "该第三方账号已绑定其他管理员"}), 409
|
|
|
|
logger.info(f"[admin/social/bind] admin_id={admin_id} provider={profile.provider}")
|
|
return jsonify({"success": True, "item": _admin_binding_row(profile.provider, binding)})
|
|
|
|
return _inner()
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/admin/social-bindings/<provider>", methods=["DELETE"])
|
|
def unbind_admin_social_account(provider):
|
|
from routes.decorators import admin_required
|
|
|
|
@admin_required
|
|
def _inner():
|
|
provider_value = str(provider or "").strip().lower()
|
|
if provider_value not in PROVIDER_LABELS:
|
|
return jsonify({"error": "不支持的登录方式"}), 400
|
|
admin_id = int(session.get("admin_id") or 0)
|
|
if not database.delete_admin_social_login_binding(admin_id, provider_value):
|
|
return jsonify({"error": "绑定记录不存在"}), 404
|
|
logger.info(f"[admin/social/unbind] admin_id={admin_id} provider={provider_value}")
|
|
return jsonify({"success": True})
|
|
|
|
return _inner()
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/social-login/test", methods=["POST"])
|
|
def test_admin_social_config():
|
|
from routes.decorators import admin_required
|
|
|
|
@admin_required
|
|
def _inner():
|
|
data = _get_json_payload()
|
|
provider = str(data.get("provider") or "wx").strip().lower()
|
|
appkey = str(data.get("social_login_appkey") or "").strip()
|
|
old_config = database.get_system_config() or {}
|
|
|
|
try:
|
|
temp_config = {
|
|
**old_config,
|
|
"social_login_enabled": 1,
|
|
"social_login_endpoint": normalize_social_endpoint(str(data.get("social_login_endpoint") or "")),
|
|
"social_login_appid": str(data.get("social_login_appid") or "").strip(),
|
|
"social_login_appkey": encrypt_social_appkey(appkey) if appkey else str(old_config.get("social_login_appkey") or ""),
|
|
"social_login_providers": ",".join(parse_providers(data.get("social_login_providers"))),
|
|
}
|
|
redirect_uri = str(data.get("redirect_uri") or "").strip()
|
|
result = fetch_social_login_url(
|
|
temp_config,
|
|
provider=provider,
|
|
mode="login",
|
|
redirect_uri=redirect_uri,
|
|
allowed_hosts=_allowed_redirect_hosts(),
|
|
)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 400
|
|
except SocialLoginError as error:
|
|
return _social_error(error)
|
|
|
|
return jsonify(
|
|
{
|
|
"success": True,
|
|
"provider": result.get("provider"),
|
|
"has_url": bool(result.get("url")),
|
|
"has_scan_url": bool(result.get("scan_url")),
|
|
}
|
|
)
|
|
|
|
return _inner()
|
|
|
|
|
|
@api_social_bp.route("/yuyx/api/social-login/config", methods=["POST"])
|
|
def update_admin_social_config():
|
|
from routes.decorators import admin_required
|
|
|
|
@admin_required
|
|
def _inner():
|
|
data = _get_json_payload()
|
|
enabled = 1 if data.get("social_login_enabled") in (1, True, "1", "true", "on") else 0
|
|
endpoint_raw = str(data.get("social_login_endpoint") or "").strip()
|
|
appid = str(data.get("social_login_appid") or "").strip()
|
|
appkey = str(data.get("social_login_appkey") or "").strip()
|
|
providers = parse_providers(data.get("social_login_providers"))
|
|
|
|
try:
|
|
endpoint = normalize_social_endpoint(endpoint_raw)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 400
|
|
|
|
old_config = database.get_system_config() or {}
|
|
old_key = str(old_config.get("social_login_appkey") or "")
|
|
|
|
if enabled:
|
|
if not providers:
|
|
return jsonify({"error": "启用聚合登录时至少选择一种登录方式"}), 400
|
|
if not appid:
|
|
return jsonify({"error": "启用聚合登录时必须填写 APPID"}), 400
|
|
if not appkey and not old_key:
|
|
return jsonify({"error": "启用聚合登录时必须填写 APPKEY"}), 400
|
|
|
|
encrypted_key = encrypt_social_appkey(appkey) if appkey else old_key
|
|
ok = database.update_system_config(
|
|
social_login_enabled=enabled,
|
|
social_login_endpoint=endpoint,
|
|
social_login_appid=appid,
|
|
social_login_appkey=encrypted_key,
|
|
social_login_providers=",".join(providers),
|
|
)
|
|
if not ok:
|
|
return jsonify({"error": "更新失败"}), 400
|
|
|
|
logger.info(f"[social/config] updated enabled={enabled} providers={','.join(providers)} at={get_cst_now_str()}")
|
|
return jsonify({"message": "聚合登录配置已更新", "config": admin_social_config_out(database.get_system_config())})
|
|
|
|
return _inner()
|