如何为自研Python CLI工具建立AKS连接并实现令牌认证？

你不用局限于直接调用REST API——官方的kubernetes Python客户端库是更高效、更易维护的方案，比自己手写REST请求靠谱得多。另外也可以通过Python调用kubectl命令行，但客户端库显然是自研CLI的首选。

首选方案：使用官方Kubernetes Python客户端库 #

步骤1：安装依赖 #

pip install kubernetes

步骤2：配置AKS认证（满足令牌安全需求） #

提供三种适配不同场景的认证方式，都支持令牌验证：

方式A：本地kubeconfig（开发/测试场景）

先通过Azure CLI拉取AKS的kubeconfig到本地：

az aks get-credentials --resource-group <你的资源组名称> --name <AKS集群名称>

客户端库会自动读取默认路径（~/.kube/config）的配置，直接初始化即可：

from kubernetes import client, config

# 加载kubeconfig配置
config.load_kube_config()

# 初始化核心API客户端
v1 = client.CoreV1Api()

方式B：Azure AD令牌（生产/自动化场景）

适合无本地配置文件的情况，通过Azure身份库动态获取令牌：
先安装Azure身份依赖：

pip install azure-identity

然后编写认证代码：

from kubernetes import client
from azure.identity import DefaultAzureCredential

# 替换为你的AKS集群API地址（可通过`az aks show --resource-group <资源组> --name <集群名> --query "fqdn"`获取）
api_server = "https://<你的AKS集群FQDN>"

# 获取Azure AD令牌
credential = DefaultAzureCredential()
token = credential.get_token("https://management.azure.com/.default").token

# 配置客户端
configuration = client.Configuration()
configuration.host = api_server
configuration.verify_ssl = True
configuration.api_key = {"authorization": f"Bearer {token}"}

client.Configuration.set_default(configuration)
v1 = client.CoreV1Api()

方式C：K8s服务账号令牌（长期自动化操作）

在AKS中创建具备Pod删除权限的服务账号，绑定最小权限角色：

# 创建服务账号
kubectl create sa aks-cli-sa -n <目标命名空间>

# 创建仅允许删除Pod的角色
kubectl create role pod-deleter --verb=delete --resource=pods -n <目标命名空间>

# 绑定角色到服务账号
kubectl create rolebinding aks-cli-sa-binding --role=pod-deleter --serviceaccount=<目标命名空间>:aks-cli-sa -n <目标命名空间>

提取服务账号令牌后，在Python中配置：

from kubernetes import client

# 替换为你的服务账号令牌（可通过`kubectl get secret <sa-secret-name> -n <命名空间> -o jsonpath='{.data.token}' | base64 -d`获取）
sa_token = "<你的服务账号令牌>"
api_server = "https://<你的AKS集群FQDN>"

# 配置客户端
configuration = client.Configuration()
configuration.host = api_server
configuration.verify_ssl = True
configuration.api_key = {"authorization": f"Bearer {sa_token}"}

client.Configuration.set_default(configuration)
v1 = client.CoreV1Api()

步骤3：实现删除Pod操作 #

有了客户端后，删除Pod的代码非常简洁：

def delete_pod(namespace, pod_name):
    try:
        v1.delete_namespaced_pod(name=pod_name, namespace=namespace)
        print(f"Pod {pod_name} 在命名空间 {namespace} 中已成功删除")
    except client.ApiException as e:
        print(f"删除Pod失败: {e.reason}")

# 调用示例
delete_pod("default", "test-pod-xxxx")

备选方案：直接调用Kubernetes REST API #

如果坚持手写REST请求，流程如下：

import requests

api_server = "https://<你的AKS集群FQDN>"
token = "<你的认证令牌>"
namespace = "default"
pod_name = "test-pod-xxxx"

url = f"{api_server}/api/v1/namespaces/{namespace}/pods/{pod_name}"
headers = {"Authorization": f"Bearer {token}"}

response = requests.delete(url, headers=headers, verify=True)
if response.status_code in [200, 202]:
    print("Pod删除成功")
else:
    print(f"删除失败，状态码: {response.status_code}, 信息: {response.text}")

注意事项 #

• 生产环境不要硬编码令牌，建议通过环境变量或Azure Key Vault读取
• 遵循最小权限原则，只给操作所需的权限（比如仅允许删除指定命名空间的Pod）
• 若使用自签证书的AKS集群，需在配置中关闭SSL验证或指定证书路径

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