Kuboard 模拟登录与能力集成实践

Kuboard 是一款高效的 Kubernetes 可视化管理工具。但在自动化运维场景下，由于其部分管理功能（如 SSO 接入、集群导入）未完全开放标准 API，或者 API 文档不够完善，直接集成往往面临权限校验瓶颈。本文记录一种通过模拟浏览器登录获取 KuboardToken，进而实现自动化集群管理的务实方案。

登录流程剖析

Kuboard 的登录逻辑本质上是一个标准表单提交。关键步骤如下：

  1. 获取 Request ID：访问 /sso/auth 路由获取服务器生成的 req 参数。
  2. 表单认证：向 /sso/auth/default?req={req} 提交用户名（login）和密码 JSON 串（password）。
  3. 获取授权码：成功后，系统会重定向并携带认证 code。
  4. 置换 Token：访问 /callback 接口，从 Set-Cookie 响应头中提取核心凭据 KuboardToken。

Python 模拟实现

以下代码演示了如何利用 http.client（或 requests）串联上述流程。

import urllib.parse
import http.client
import logging

# 配置参数
config = {
    "host": "10.10.1.111",
    "port": 8089,
    "login": "admin",
    "password": '{"password":"Kuboard123","passcode":""}',
    "client_id": "kuboard-sso",
    "redirect_uri": "/callback",
    "scope": "openid profile email groups",
    "connector_id": "default",
    "state": "/"
}

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
global_code = None

def send_request(method, url, headers=None, body=None):
    conn = http.client.HTTPConnection(config["host"], config["port"])
    conn.request(method, url, body, headers)
    response = conn.getresponse()

    if response.status in [303, 302]:
        location = response.getheader("Location")
        parsed_url = urllib.parse.urlparse(location)
        query_params = urllib.parse.parse_qs(parsed_url.query)
        if 'code' in query_params:
            global global_code
            global_code = query_params['code'][0]
        return location
    return response

# 1. 获取初始 req
def get_req():
    url = f"/sso/auth?access_type=offline&client_id={config['client_id']}&redirect_uri={urllib.parse.quote(config['redirect_uri'])}&response_type=code&scope={urllib.parse.quote(config['scope'])}&state={urllib.parse.quote(config['state'])}&connector_id={config['connector_id']}"
    resp = send_request("GET", url)
    return resp.split("req=")[-1] if isinstance(resp, str) else None

# 2. 模拟表单登录
def login(req):
    url = f"/sso/auth/default?req={req}"
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    data = {"login": config['login'], "password": config['password']}
    send_request("POST", url, headers, urllib.parse.urlencode(data))

# 3. 授权确认
def submit_auth(req):
    url = f"/sso/approval?req={req}"
    return send_request("GET", url)

# 4. 提取 Token
def get_kuboard_token():
    if not global_code: return None
    url = f"/callback?code={global_code}&state=%2F"
    resp = send_request("GET", url)
    if isinstance(resp, http.client.HTTPResponse):
        cookies = resp.getheader("Set-Cookie")
        for cookie in cookies.split(";"):
            if "KuboardToken" in cookie:
                return cookie.split("=")[-1]
    return None

def main():
    req = get_req()
    if req:
        login(req)
        if submit_auth(req):
            token = get_kuboard_token()
            if token: print(f"Successfully obtained Token: {token}")

if __name__ == "__main__":
    main()

应用场景：自动化添加集群

获取 KuboardToken 后，可以通过 REST 接口直接操作。以下为添加 Kubernetes 集群的示例（需根据实际抓包调整 JSON 结构）：

# 获取 Token
TOKEN=$(python3 login.py)

# 调用 API 添加集群
curl -X POST 'http://10.10.1.111:8089/kuboard-api/cluster/my-cluster/kind/KubernetesCluster' \
  -H 'Content-Type: application/json' \
  -b "KuboardToken=$TOKEN" \
  --data-raw '{
    "kind": "KubernetesCluster",
    "metadata": {"name": "my-cluster"},
    "spec": {
      "connectionType": "token",
      "kubeserver": "https://10.10.1.111:6443",
      "kubetoken": "YOUR_K8S_SERVICE_ACCOUNT_TOKEN"
    }
  }'

技术注意事项

  1. 会话持久性：KuboardToken 通常有过期时间，建议在自动化流水线中即用即取。
  2. 重定向处理：Kuboard 的 SSO 涉及多次 302/303 重定向，必须确保代码能准确捕获 Location 头部的 code 字段。
  3. 安全风险：模拟登录涉及账号密码明文处理，建议通过环境变量或 KMS 获取敏感信息。

总结

通过模拟登录绕过前端交互限制，是打通运维工具链“最后一公里”的常见手段。此方案已稳定运行于多套 K8s 环境的自动化接入任务中。