使用Python连接VMware REST API时遇认证请求错误的解决

问题现象 #

运行Python代码连接vCenter REST API时，出现以下错误提示：

输入vCenter主机名：[myIP]
输入连接MyIP的用户名：[user] user
输入user@vsphere.local的密码：

• MyIP: 认证或请求错误

原代码 #

default_vc = "MyIP"
domain = "vsphere.local"

import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
requests.packages.urllib3.util.ssl_.DEFAULT_CIPHERS = 'ALL:@SECLEVEL=1'
session = requests.Session()
session.verify = False

import getpass

def get_session_info(default_vc, domain):
    default_vc = default_vc
    vc = input("Ingrese el nombre de host del vCenter: [" + default_vc + "] ")
    if not vc:
        vc = default_vc
    myuser = getpass.getuser()
    username = input("Ingrese el nombre de usuario para conectarse a " + vc + ": [" + myuser + '] ')
    if not username:
        username = myuser
    username = username + "@" + domain
    password = getpass.getpass("Ingrese la contraseña para el usuario " + username + ": ")
    while not password:
        print("Error: Debe ingresar una contraseña")
        password = getpass.getpass("Ingrese la contraseña para el usuario " + username + ": ")
    return vc, username, password

def get_vc_session(vc, domain, username, password):
    session.post('https://' + vc + '.' + domain + '/rest/com/vmware/cis/session', auth=(username, password))
    return session

import json

import getpass

def get_host(vc, domain, session):
    request_json_response = session.get('https://10.71.34.51/' + vc + '.' + domain + 'rest/vcenter/host')
    if request_json_response.status_code == 200:
        print(vc + ": Autenticación exitosa")
    else:
        print(vc + ": Error en la autenticación o en la petición")
        exit(1)
    host_list = json.loads(request_json_response.text)
    host_list_value = host_list["value"]
    return host_list_value

vc, username, password = get_session_info(default_vc, domain)
vc_session = get_vc_session(vc, domain, username, password)

host_list = get_host(vc, domain, vc_session)

for item in host_list:
    for item_key, item_value in item.items():
        print(item_key + ":", end="")
        print(str(item_value) + " ", end="\t")
    print()

问题分析 #

1. API地址硬编码与拼接错误：get_host函数中硬编码了固定IP10.71.34.51，且URL拼接格式错误，正确的vCenter REST API地址应为https://{vc}/rest/vcenter/host，无需额外拼接域名后缀（如果输入的vc是IP地址）。
2. 会话认证未做校验：get_vc_session函数调用session.post创建会话时，未检查响应状态码，即使认证失败也会返回会话，导致后续请求无有效权限。
3. 认证URL拼接错误：get_vc_session中把vc和domain拼接成vc.domain，如果输入的vc是IP地址，会形成无效的访问地址（如10.xx.xx.xx.vsphere.local）。

修复后的代码 #

default_vc = "MyIP"
domain = "vsphere.local"

import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
requests.packages.urllib3.util.ssl_.DEFAULT_CIPHERS = 'ALL:@SECLEVEL=1'
session = requests.Session()
session.verify = False

import getpass
import json

def get_session_info(default_vc, domain):
    vc = input("输入vCenter主机名: [" + default_vc + "] ")
    if not vc:
        vc = default_vc
    myuser = getpass.getuser()
    username = input("输入连接" + vc + "的用户名: [" + myuser + '] ')
    if not username:
        username = myuser
    username = username + "@" + domain
    password = getpass.getpass("输入" + username + "的密码: ")
    while not password:
        print("错误：必须输入密码")
        password = getpass.getpass("输入" + username + "的密码: ")
    return vc, username, password

def get_vc_session(vc, username, password):
    # 直接使用输入的vc地址，不拼接域名（如果输入的是IP/已解析的主机名）
    auth_url = f'https://{vc}/rest/com/vmware/cis/session'
    response = session.post(auth_url, auth=(username, password))
    # 检查认证是否成功
    if response.status_code != 200:
        print(f"{vc}: 会话创建失败，状态码: {response.status_code}")
        exit(1)
    return session

def get_host(vc, session):
    host_url = f'https://{vc}/rest/vcenter/host'
    request_json_response = session.get(host_url)
    if request_json_response.status_code == 200:
        print(f"{vc}: 认证成功")
    else:
        print(f"{vc}: 认证或请求错误，状态码: {request_json_response.status_code}")
        exit(1)
    host_list = json.loads(request_json_response.text)
    host_list_value = host_list["value"]
    return host_list_value

# 主逻辑
vc, username, password = get_session_info(default_vc, domain)
vc_session = get_vc_session(vc, username, password)

host_list = get_host(vc, vc_session)

# 打印主机列表
for item in host_list:
    for item_key, item_value in item.items():
        print(f"{item_key}: {item_value}", end="\t")
    print()

修复说明 #

• 移除了URL中的硬编码IP，改为使用用户输入的vc地址动态拼接API路径
• 修正了认证URL的拼接逻辑，不再错误添加域名后缀（若vc是IP/已解析主机名，直接使用即可）
• 增加了会话创建时的状态校验，认证失败直接退出并提示状态码
• 将提示文本改为中文，更符合使用习惯

内容的提问来源于stack exchange，提问作者Mercedes Aguilar