使用 API 接口批量配置锐捷交换机

jrui 47 阅读 网络

设备型号:

SF2920U_1.0(1)B8P1 Release(12192322)

一、项目背景

由于需要对大量交换机执行相同配置操作,逐台登录 Web 页面进行修改效率较低。

通过分析 Web 管理页面与后台 API 的交互过程,可以实现:

  • 批量配置 VLAN

  • 批量修改端口参数

  • 自动保存配置

  • 批量处理多台设备

从而大幅提高运维效率。

二、Web 页面操作流程

人工配置流程如下:

登录设备
    ↓
进入端口配置
    ↓
修改 VLAN
    ↓
修改端口参数
    ↓
保存配置

通过抓包分析发现,上述操作均通过后台 API 实现。

三、接口分析

1、登录接口

请求地址

GET /api/login

认证方式

请求头中包含:

Authorization: Basic xxxxxxxxx

认证方式为 HTTP Basic Authentication。

示例:

admin:password
        ↓
Base64编码
        ↓
Authorization: Basic YWRtaW46cGFzc3dvcmQ=

Apifox 验证

登录

返回结果

{"user":"admin","token":"1664245936"}

说明:

  • 登录成功后返回 Token

  • Token 有效期约 10 分钟

  • 后续接口通过 Cookie 完成认证

2、VLAN 配置接口

用于修改端口 VLAN 参数。

请求地址

POST /api/l2/vlan/set

请求头

Content-Type: application/json;charset=UTF-8
Accept: application/json, text/plain, */*
Cookie: access_token=xxxxxxxx

请求参数(Body)

{
  "portList": "1",
  "portType": "trunk",
  "allowedVlan": "1-20,50-70,100,110,120,200-220,300-360",
  "nativeVlan": "356"
}

参数说明

参数

说明

portList

端口号

portType

端口模式

allowedVlan

Trunk允许通过的VLAN

nativeVlan

Native VLAN

Apifox 验证

VLAN配置

返回码

返回值

说明

0

配置成功

-4

认证失败

-25

参数校验失败

3、端口配置接口

用于修改端口基础属性。

请求地址

POST /api/port/set

请求头

Content-Type: application/json;charset=UTF-8
Accept: application/json, text/plain, */*
Cookie: access_token=xxxxxxxx

请求参数(Body)

{
    "ifx": 1,
    "admin": 1,
    "duplex": 2,
    "speed": 0,
    "rstp": 1,
    "lldp": 1,
    "stormCtl": {
        "unicast": 0,
        "broadcast": 0,
        "mutcast": 0
    },
    "poe": 1,
    "portIsolated": 0,
    "dhcpTrust": {
        "glben": 1,
        "porten": 0
    }
}

Apifox 验证

image-20260615141159255

端口配置

返回码

返回值

说明

0

配置成功

-4

认证失败

-25

参数校验失败

4、保存配置接口

用于将运行配置写入启动配置,确保设备重启后配置不丢失。

请求地址

POST /api/cfg/save

请求头

Content-Type: application/json;charset=UTF-8
Accept: application/json, text/plain, */*
Cookie: access_token=xxxxxxxx

Apifox 验证

保存配置

返回结果

{
    "status": 0
}

表示配置保存成功。

四、认证机制

认证流程如下:

/api/login
      ↓
获取 Token
      ↓
生成 access_token
      ↓
写入 Cookie
      ↓
调用 VLAN 接口
      ↓
调用端口接口
      ↓
保存配置

注意事项:

  • Token 约 10 分钟过期

  • 配置接口依赖 Cookie 完成认证

  • Cookie 失效后需重新登录

  • 未认证会返回 status=-4

五、自动化实现思路

整体执行逻辑如下:

读取设备列表
      ↓
登录设备
      ↓
获取认证信息
      ↓
配置 VLAN
      ↓
配置端口
      ↓
保存配置
      ↓
记录结果
      ↓
处理下一台设备

六、Apifox 验证流程

建议先通过 Apifox 验证接口逻辑。

执行顺序:

GET  /api/login
        ↓
POST /api/l2/vlan/set
        ↓
POST /api/port/set
        ↓
POST /api/cfg/save

所有接口返回:

{
    "status": 0
}

即表示执行成功。

七、Python 自动化脚本

验证完成后,即可通过 Python 实现批量下发。

功能:

  • 自动登录设备

  • 自动获取认证信息

  • 自动配置 VLAN

  • 自动配置端口参数

  • 自动保存配置

  • 自动记录日志

  • 支持多线程并发

  • 支持失败自动重试

import requests
import logging
import time
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
​
DEVICE_FILE = "devices.txt"
LOG_FILE = "batch_config_debug.log"
​
# 请求超时时间。设备下发 VLAN 时可能需要十几秒,不建议改得太小。
TIMEOUT = 180
​
# 并发线程数。设备数量较多时可适当调大;现场网络不稳定时建议保持 5 或更低。
MAX_WORKERS = 5
​
# 每个接口失败后的重试次数。
RETRY = 3
​
# 重要:这里填浏览器 Network 里抓到的完整 Authorization
# 格式必须是:Basic + 一个空格 + 后面的认证字符串。
# 不要填中文占位符,否则 requests 会因为请求头编码失败而直接报错。
AUTHORIZATION = "Basic aFhTRUpVVmxWYVYxWlZNYUZoVFJVcFZWbXhXWVZZeFdsWk5ZVVpvVkZKVmNGWldiWGhYV1ZaWmVGZHNXazVaVjFKMFlWYzBObGRIYUhWYVdGSXpZak5LY2xGRVJYbE5kejA5"
​
# VLAN 下发参数。这里控制端口类型、允许 VLAN 和 native VLAN。
# 修改多个端口时,portList 可按设备 Web 前端支持的格式填写,例如 "1,2" 或 "1-4"。
VLAN_PAYLOAD = {
    "portList": "1",
    "portType": "trunk",
    "allowedVlan": "1-20,50-70,100,110,120,200-220,300-360",
    "nativeVlan": "356"
}
​
# 端口基础参数。ifx 是端口号,需要和 VLAN_PAYLOAD["portList"] 对应。
# 例如只改 1 号口就保持 ifx=1;改其他单端口时同步调整 ifx。
PORT_PAYLOAD = {
    "ifx": 1,
    "admin": 1,
    "duplex": 2,
    "speed": 0,
    "rstp": 1,
    "lldp": 1,
    "stormCtl": {
        "unicast": 0,
        "broadcast": 0,
        "mutcast": 0
    },
    "poe": 1,
    "portIsolated": 0,
    "dhcpTrust": {
        "glben": 1,
        "porten": 0
    }
}
​
logging.basicConfig(
    filename=LOG_FILE,
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(message)s",
    encoding="utf-8"
)
​
​
def load_devices():
    # 从 devices.txt 读取设备列表;空行和 # 开头的注释会自动跳过。
    with open(DEVICE_FILE, "r", encoding="utf-8") as f:
        return [
            line.strip()
            for line in f
            if line.strip() and not line.strip().startswith("#")
        ]
​
​
def base_headers(ip, json_type=False):
    # 设备 Web API 需要 Referer、X-Lang 和 Authorization。
    # json_type=True 时用于 POST JSON 请求,会额外带 Content-Type。
    headers = {
        "Accept": "application/json, text/plain, */*",
        "X-Lang": "zh",
        "Referer": f"http://{ip}/",
        "Authorization": AUTHORIZATION
    }
​
    if json_type:
        headers["Content-Type"] = "application/json;charset=UTF-8"
​
    return headers
​
​
def log_response(ip, name, r):
    # 详细记录请求和响应,方便排查接口路径、参数和设备返回码。
    # Authorization 会脱敏写入日志,避免把认证信息明文留在日志里。
    logging.info(f"{ip} [{name}] URL: {r.request.method} {r.url}")
    request_headers = dict(r.request.headers)
    if "Authorization" in request_headers:
        request_headers["Authorization"] = "[REDACTED]"
    logging.info(f"{ip} [{name}] Request Headers: {request_headers}")
​
    if r.request.body:
        logging.info(f"{ip} [{name}] Request Body: {r.request.body}")
​
    logging.info(f"{ip} [{name}] Status Code: {r.status_code}")
    logging.info(f"{ip} [{name}] Response Headers: {dict(r.headers)}")
    logging.info(f"{ip} [{name}] Response Body: {r.text[:1000]}")
​
​
def check_response(r, ip, name):
    # 所有接口统一检查 HTTP 状态和业务 status。
    log_response(ip, name, r)
​
    if r.status_code != 200:
        raise Exception(f"{name} HTTP失败:{r.status_code}")
​
    try:
        data = r.json()
    except Exception:
        raise Exception(f"{name} 返回不是JSON:{r.text[:300]}")
​
    status = data.get("status")
​
    # 部分设备登录成功不返回 status=0,而是返回 {"user": "...", "token": "..."}。
    if name == "登录" and "token" in data:
        return data
​
    if status != 0:
        raise Exception(f"{name} 业务失败:status={status},返回={data}")
​
    return data
​
​
def retry_request(ip, name, func):
    # 对登录、配置、保存等步骤统一做重试。
    last_error = None
​
    for i in range(1, RETRY + 1):
        try:
            logging.info(f"{ip} [{name}] 开始执行,第 {i} 次")
            result = func()
            logging.info(f"{ip} [{name}] 执行成功")
            return result
        except Exception as e:
            last_error = e
            logging.warning(f"{ip} [{name}] 第 {i} 次失败:{e}")
            time.sleep(1)
​
    raise Exception(f"{name} 重试 {RETRY} 次后失败:{last_error}")
​
​
def login(session, ip):
    # 登录接口。成功后同一个 session 会继续用于后续配置请求。
    url = f"http://{ip}/api/login"
​
    def do():
        r = session.get(
            url,
            headers=base_headers(ip),
            timeout=TIMEOUT
        )
        return check_response(r, ip, "登录")
​
    return retry_request(ip, "登录", do)
​
​
def post_json(session, ip, path, payload, name):
    # 通用 JSON POST 方法,VLAN 配置和端口配置都走这里。
    url = f"http://{ip}{path}"
​
    def do():
        logging.info(f"{ip} [{name}] 准备请求 Payload: {payload}")
​
        r = session.post(
            url,
            json=payload,
            headers=base_headers(ip, json_type=True),
            timeout=TIMEOUT
        )
​
        return check_response(r, ip, name)
​
    return retry_request(ip, name, do)
​
​
def save_config(session, ip):
    # 保存配置到设备,避免重启后配置丢失。
    url = f"http://{ip}/api/cfg/save"
​
    def do():
        r = session.post(
            url,
            headers=base_headers(ip),
            timeout=TIMEOUT
        )
        return check_response(r, ip, "保存配置")
​
    return retry_request(ip, "保存配置", do)
​
​
def process_device(ip):
    # 单台设备的完整处理流程:登录 -> VLAN 配置 -> 端口配置 -> 保存。
    session = requests.Session()
​
    # 禁止读取系统代理,避免走 127.0.0.1:7890
    session.trust_env = False
​
    logging.info("=" * 80)
    logging.info(f"{ip} 开始处理")
​
    try:
        login(session, ip)
​
        vlan_result = post_json(
            session,
            ip,
            "/api/l2/vlan/set",
            VLAN_PAYLOAD,
            "VLAN配置"
        )
​
        port_result = post_json(
            session,
            ip,
            "/api/port/set",
            PORT_PAYLOAD,
            "端口配置"
        )
​
        save_result = save_config(session, ip)
​
        logging.info(f"{ip} VLAN配置返回:{vlan_result}")
        logging.info(f"{ip} 端口配置返回:{port_result}")
        logging.info(f"{ip} 保存配置返回:{save_result}")
        logging.info(f"{ip} 最终结果:配置成功")
​
        return f"{ip} 配置成功"
​
    except Exception as e:
        logging.error(f"{ip} 最终结果:配置失败:{e}")
        return f"{ip} 配置失败:{e}"
​
​
def main():
    devices = load_devices()
​
    print(f"共发现 {len(devices)} 台设备,开始执行...")
​
    # 多线程并发处理 devices.txt 中的设备。
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = [executor.submit(process_device, ip) for ip in devices]
​
        for future in as_completed(futures):
            print(future.result())
​
    print(f"执行完成,日志文件:{LOG_FILE}")
​
​
if __name__ == "__main__":
    main()
​

八、使用说明

1、修改认证信息

编辑脚本中的:

AUTHORIZATION = "Basic xxxxxxxxx"

替换为浏览器抓包获取的认证字符串。

2、创建设备列表

在脚本目录下创建:

devices.txt

示例:

172.18.13.18
172.18.13.19
172.18.13.20

每行一个设备 IP。

3、安装依赖

pip install requests

4、运行脚本

python batch_vlan.py

5、查看日志

脚本执行完成后会生成:

batch_config_debug.log

用于查看:

  • 登录状态

  • 请求参数

  • 返回结果

  • 错误原因

九、执行流程示意图

设备列表
    ↓
登录设备
    ↓
获取认证信息
    ↓
VLAN配置
    ↓
端口配置
    ↓
保存配置
    ↓
记录日志
    ↓
输出结果

至此即可实现交换机的批量自动化配置。