555c86ce76
Backend (Django 4.2 + DRF): - Admin auth with SimpleJWT - Volcengine API client with HMAC-SHA256 signing - IAM user management (create/sync/import/disable/enable) - Billing query with pagination - Feishu webhook notifications (async) - APScheduler for periodic spending checks - AES-256 encrypted credential storage - API key auth for external system integration Frontend (Vue 3 + Element Plus): - Login page - Dashboard with stats overview - IAM user list with per-user threshold config - Billing view with spending progress bars - Alert history with type filtering - Settings page for global config and Volcengine account management Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
"""IAM 子账号管理服务"""
|
|
|
|
import logging
|
|
from .volcengine_client import get_iam_client, VolcengineAPIError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class IAMService:
|
|
"""封装火山引擎 IAM API 操作"""
|
|
|
|
def __init__(self, ak: str, sk: str):
|
|
self.client = get_iam_client(ak, sk)
|
|
|
|
def list_users(self, limit=100, offset=0) -> dict:
|
|
return self.client.call("ListUsers", {"Limit": str(limit), "Offset": str(offset)})
|
|
|
|
def get_user(self, username: str) -> dict:
|
|
return self.client.call("GetUser", {"UserName": username})
|
|
|
|
def create_user(self, username: str, display_name: str = "", email: str = "",
|
|
phone: str = "") -> dict:
|
|
params = {"UserName": username}
|
|
if display_name:
|
|
params["DisplayName"] = display_name
|
|
if email:
|
|
params["Email"] = email
|
|
if phone:
|
|
params["MobilePhone"] = phone
|
|
return self.client.call("CreateUser", params)
|
|
|
|
def create_login_profile(self, username: str, password: str,
|
|
login_allowed: bool = True, must_reset: bool = True) -> dict:
|
|
return self.client.call("CreateLoginProfile", {
|
|
"UserName": username,
|
|
"Password": password,
|
|
"LoginAllowed": str(login_allowed).lower(),
|
|
"PasswordResetRequired": str(must_reset).lower(),
|
|
})
|
|
|
|
def update_login_allowed(self, username: str, allowed: bool) -> dict:
|
|
return self.client.call("UpdateLoginProfile", {
|
|
"UserName": username,
|
|
"LoginAllowed": str(allowed).lower(),
|
|
})
|
|
|
|
def get_login_profile(self, username: str) -> dict:
|
|
return self.client.call("GetLoginProfile", {"UserName": username})
|
|
|
|
def list_access_keys(self, username: str) -> list:
|
|
resp = self.client.call("ListAccessKeys", {"UserName": username})
|
|
return resp.get("Result", {}).get("AccessKeyMetadata", [])
|
|
|
|
def update_access_key(self, ak_id: str, status: str, username: str = "") -> dict:
|
|
params = {"AccessKeyId": ak_id, "Status": status}
|
|
if username:
|
|
params["UserName"] = username
|
|
return self.client.call("UpdateAccessKey", params)
|
|
|
|
def create_access_key(self, username: str) -> dict:
|
|
return self.client.call("CreateAccessKey", {"UserName": username})
|
|
|
|
def attach_user_policy(self, username: str, policy_name: str,
|
|
policy_type: str = "System") -> dict:
|
|
return self.client.call("AttachUserPolicy", {
|
|
"UserName": username,
|
|
"PolicyName": policy_name,
|
|
"PolicyType": policy_type,
|
|
})
|
|
|
|
def detach_user_policy(self, username: str, policy_name: str,
|
|
policy_type: str = "System") -> dict:
|
|
return self.client.call("DetachUserPolicy", {
|
|
"UserName": username,
|
|
"PolicyName": policy_name,
|
|
"PolicyType": policy_type,
|
|
})
|
|
|
|
def list_attached_user_policies(self, username: str) -> dict:
|
|
return self.client.call("ListAttachedUserPolicies", {"UserName": username})
|
|
|
|
def disable_user(self, username: str):
|
|
"""完全停用用户：停控制台 + 停所有 AccessKey"""
|
|
errors = []
|
|
|
|
try:
|
|
self.update_login_allowed(username, False)
|
|
except VolcengineAPIError as e:
|
|
errors.append(f"停用控制台失败: {e}")
|
|
|
|
try:
|
|
keys = self.list_access_keys(username)
|
|
for key in keys:
|
|
if key.get("Status") == "active":
|
|
self.update_access_key(key["AccessKeyId"], "inactive", username)
|
|
except VolcengineAPIError as e:
|
|
errors.append(f"停用密钥失败: {e}")
|
|
|
|
if errors:
|
|
raise VolcengineAPIError("DisableUser", "PartialFailure", "; ".join(errors))
|
|
|
|
def enable_user(self, username: str):
|
|
"""恢复用户：恢复控制台 + 恢复所有 AccessKey"""
|
|
errors = []
|
|
|
|
try:
|
|
self.update_login_allowed(username, True)
|
|
except VolcengineAPIError as e:
|
|
errors.append(f"恢复控制台失败: {e}")
|
|
|
|
try:
|
|
keys = self.list_access_keys(username)
|
|
for key in keys:
|
|
if key.get("Status") == "inactive":
|
|
self.update_access_key(key["AccessKeyId"], "active", username)
|
|
except VolcengineAPIError as e:
|
|
errors.append(f"恢复密钥失败: {e}")
|
|
|
|
if errors:
|
|
raise VolcengineAPIError("EnableUser", "PartialFailure", "; ".join(errors))
|