JumpServer 开启 MFA 后通过 API 获取 Bearer Token 的技术实践

在企业级堡垒机 JumpServer 的实际应用中，多因素认证(MFA)是提升系统安全性的重要手段。然而当管理员需要批量操作或自动化运维时，如何在开启 MFA 的情况下通过 API 获取有效的 Bearer Token 就成为了一个技术挑战。本文将深入探讨这一技术实现方案。

认证流程解析

JumpServer 的 API 认证流程在开启 MFA 后分为两个关键阶段：

1. 基础认证阶段：首先需要通过 /api/v1/authentication/auth 接口提交用户名和密码进行初步验证
2. MFA 验证阶段：基础认证通过后，系统会返回 MFA 挑战信息，需要通过 /api/v1/authentication/mfa/challenge/ 接口完成二次验证

详细实现步骤

1. 基础认证请求

首先向认证接口发送 POST 请求，携带用户名和密码：

import requests

auth_url = "http://jumpserver.example.com/api/v1/authentication/auth"
credentials = {
    "username": "your_username",
    "password": "your_password"
}

response = requests.post(auth_url, json=credentials)
auth_result = response.json()

2. 处理 MFA 挑战

基础认证成功后，响应中会包含 MFA 挑战信息：

if response.status_code == 200 and auth_result.get("mfa_required"):
    challenge_id = auth_result["challenge"]["id"]
    mfa_type = auth_result["challenge"]["type"]  # 通常是 'otp'
    
    # 获取用户输入的 MFA 验证码
    mfa_code = input("请输入MFA验证码: ")
    
    # 提交 MFA 验证
    mfa_url = "http://jumpserver.example.com/api/v1/authentication/mfa/challenge/"
    mfa_data = {
        "challenge_id": challenge_id,
        "code": mfa_code
    }
    
    mfa_response = requests.post(mfa_url, json=mfa_data)
    mfa_result = mfa_response.json()
    
    if mfa_response.status_code == 200:
        token = mfa_result["token"]
        print(f"成功获取 Bearer Token: {token}")

自动化处理建议

对于需要完全自动化的场景，可以考虑以下方案：

1. 使用应用密码：部分 MFA 系统支持生成应用专用密码
2. 预共享密钥：在安全环境下预先配置好 MFA 种子
3. 硬件令牌集成：使用支持 API 调用的硬件安全模块

安全注意事项

1. 永远不要在代码中硬编码凭据信息
2. 使用环境变量或密钥管理系统存储敏感信息
3. 为自动化账号设置最小必要权限
4. 定期轮换 API Token
5. 监控 API 调用日志，及时发现异常行为

完整示例代码

import requests
import os
from getpass import getpass

def get_jumpserver_token():
    base_url = os.getenv("JUMPSERVER_URL")
    username = os.getenv("JUMPSERVER_USER")
    password = os.getenv("JUMPSERVER_PASS") or getpass("请输入密码: ")
    
    # 第一阶段：基础认证
    auth_url = f"{base_url}/api/v1/authentication/auth"
    response = requests.post(auth_url, json={"username": username, "password": password})
    
    if response.status_code != 200:
        raise Exception(f"认证失败: {response.text}")
    
    data = response.json()
    
    # 第二阶段：MFA验证
    if data.get("mfa_required"):
        mfa_code = input("请输入MFA验证码: ")
        challenge_id = data["challenge"]["id"]
        
        mfa_url = f"{base_url}/api/v1/authentication/mfa/challenge/"
        mfa_response = requests.post(mfa_url, json={
            "challenge_id": challenge_id,
            "code": mfa_code
        })
        
        if mfa_response.status_code != 200:
            raise Exception(f"MFA验证失败: {mfa_response.text}")
        
        return mfa_response.json()["token"]
    
    return data["token"]

# 使用示例
if __name__ == "__main__":
    token = get_jumpserver_token()
    print(f"获取到的Token: {token}")

通过以上方法，即使在 JumpServer 开启 MFA 的情况下，也能安全可靠地获取 API 访问所需的 Bearer Token，为自动化运维工作提供支持。