SSH、SNMP、NETCONF、SFTP

SSH

CE12800配置

#开启SSH服务 stelnet server enable ssh user renxinyu ssh user renxinyu authentication-type password ssh user renxinyu service-type stelnet #创建本地用户 aaa local-user renxinyu password cipher Huawei@123 local-user renxinyu level 3 local-user renxinyu service-type ssh quit #生成RSA密钥对 rsa local-key-pair create #VTY线路允许SSH接入 user-interface vty 0 4 authentication-mode aaa protocol inbound ssh quit

python脚本

""" 自动化运维脚本 针对华为设备 自动备份脚本 """ import paramiko import time #参数区域 DEVICES = {"192.168.77.2":("yeslab","Admin@456"), "192.168.66.2":("yeslab2","Admin@456")} IPs = list(DEVICES.keys()) ip_list = list(DEVICES.values()) print(ip_list) USERNAMEs = [one[0] for one in ip_list] PASSWORDs = [one[1] for one in ip_list] # print(USERNAMEs) #配置备份代码 def save_config(ip,config): with open(f"config/{ip}_conf.txt","w") as f: f.write(config) def run_backup(ip,username,password): ssh_port = paramiko.SSHClient()#使用Paramiko库连接网络设备并备份配置 ssh_port.set_missing_host_key_policy(paramiko.AutoAddPolicy)#跳过主机密钥验证 ssh_port.connect(hostname=ip, username=username, password=password)#输入IP，用户名，密码 print(f"正在备份{ip}设备...") shell = ssh_port.invoke_shell()#调用 invoke_shell() 方法创建一个交互式 shell 会话（类似于 SSH 终端） shell.send("screen-length 0 temporary\n")#防止--more--出现 shell.send("display cur\n") time.sleep(2)#暂停程序执行 2 秒，等待设备返回配置输出。 resp = shell.recv(99999).decode()#暂停程序执行 2 秒，等待设备返回配置输出。 config = resp.split("<HUAWEI>display cur\r\n")[1].split("<HUAWEI>")[0] save_config(ip,config) ssh_port.close() #主进程 if __name__ == "__main__": for index in range(len(IPs)): ip = IPs[index] username = USERNAMEs[index] password = PASSWORDs[index] run_backup(ip,username,password)

SNMP

CE12800（SSH基础上）

#创建用户创建组 snmp-agent usm-user v3 yeslab group dc-admin #设置认证 snmp-agent usm-user v3 yeslab authentication-mode sha Huawei@123 Huawei@123 #加密方式 snmp-agent usm-user v3 yeslab privacy-mode aes128 Huawei@123 Huawei@123 #设置源端口 snmp-agent trap source GE1/0/0 #自定义Mib视图并绑定对应iso snmp-agent mib-view included nt iso snmp-agent mib-view included rd iso snmp-agent mib-view included wt iso snmp-agent mib-view included iso-view iso #设置权限 snmp-agent group v3 dc-admin privacy read-view rd write-view wt notify-view nt

python脚本

device_info.py

IP = "192.168.66.2" USERNAME = "yeslab2" PASSWORD = "Admin@123"

ssh_monitor.py

import paramiko import time def connect(IP,USERNAME,PASSWORD): #创建SSH连接 ssh = paramiko.client.SSHClient() ssh.set_missing_host_key_policy(paramiko.client.AutoAddPolicy()) ssh.connect(hostname=IP, port=22,username=USERNAME, password=PASSWORD) print(f"{IP}设备登陆成功") cli = ssh.invoke_shell() cli.send("screen-length 0 temporary\n") time.sleep(0.5) cli.send("system-view immediately\n") time.sleep(0.5) return cli,ssh #将cli交互式终端和ssh实例返回 def monitor(cli): #每间隔2s从设备端接收数据，无限循环 while True: display = cli.recv(99999).decode() print(display) time.sleep(2) if __name__ == "__main__": monitor()

snmp.py

from pysnmp.hlapi import * from device_info import * from ssh_monitor import * import time def read_config_txt(cli): #读取snmp配置文件，并以每0.5s进行发送 with open("../snmp.txt") as f: snmp_config_list = f.readlines() for snmp_cmd in snmp_config_list: cli.send(snmp_cmd) time.sleep(0.5) if __name__ == "__main__": cli,ssh = connect(IP, USERNAME, PASSWORD) #将IP等信息传递到connect建立ssh连接 read_config_txt(cli) #将cli交互式终端传递到snmp配置函数中 transport = UdpTransportTarget((IP,161)) #建立snmp连接的IP和端口 snmp_channel = getCmd(SnmpEngine(), #发送GET请求，读取单个OID；snmpengine，snmp引擎，相当于初始化pysnmp，创建snmp会话环境 UsmUserData(userName="yeslab", #用户数据 authKey="Huawei@123", privKey="Huawei@123", authProtocol=usmHMACSHAAuthProtocol, privProtocol=usmAesCfb128Protocol), transport, #传递的IP+端口 ContextData(), #snmpv3上下文，默认即可，一般为空 ObjectType(ObjectIdentity("SNMPv2-MIB", "sysName",0))) errorIndication,errorStatus,errorIndex,varBinds = next(snmp_channel) #读取sysname.0即设备主机名，获得返回结果逐条返回（需要运行一次返回一条） # for msg in varBinds: #打印SNMPv2-MIB::sysName.0 = HUAWEI HUAWEI # print(msg) # print(str(msg).split("=")[1].strip()) # monitor(cli) #打印配置 print("errorIndication:", errorIndication) #查看 SNMP 请求是否成功，以及返回了什么 OID 数据。 print("errorStatus:", errorStatus) print("errorIndex:", errorIndex) for msg in varBinds: print("msg:",msg) ssh.close()

NETCONF

CE12800(SSH基础上)

system-view immediately # ====== 1. 创建NETCONF专用用户 ====== aaa local-user netconf password irreversible-cipher Huawei@123 local-user netconf service-type ssh local-user netconf level 3 ← 至少level 3，建议直接15 quit # ====== 2. 配置SSH用户认证方式和服务类型 ====== ssh user netconf ssh user netconf authentication-type password ssh user netconf service-type snetconf ← 关键！必须是snetconf，不是stelnet # ====== 3. 开启SNETCONF服务（22端口） ====== snetconf server enable # ====== 4. 开启830端口（NETCONF专用端口） ====== netconf protocol inbound ssh port 830 quit # ====== 5. 配置VTY用户线 ====== user-interface vty 0 4 authentication-mode aaa protocol inbound ssh user privilege level 3 quit

python脚本

from ncclient import manager from ncclient import operations import time #设备参数 ip = ('192.168.66.2') netconf_port = '830' netconf_user = 'yeslab2' netconf_password = 'Admin@456' #通告NetConf进行设备配置 def huawei_connect(host, port, username, password): #定义连接函数 return manager.connect(host=host, port=port, username=username, password=password, hostkey_verify=False, #关闭SSH host key校验 device_params={'name': 'huawei'}, #连接的是华为VRP设备 allow_agent=False, #不从本机 SSH Agent 获取密钥。 look_for_keys=False) #不查找本地SSH私钥 CREATE_INTERFACE = CREATE_INTERFACE = '''<config> <ethernet xmlns="http://www.huawei.com/netconf/vrp" content-version="1.0" format-version="1.0"> <ethernetIfs> <ethernetIf operation="merge"> <ifName>GE1/0/2</ifName> <l2Enable>disable</l2Enable> </ethernetIf> </ethernetIfs> </ethernet> <ifm xmlns="http://www.huawei.com/netconf/vrp" content-version="1.0" format-version="1.0"> <interfaces> <interface operation="merge"> <ifName>GE1/0/2</ifName> <ifDescr>Config by NETCONF</ifDescr> <ifmAm4> <am4CfgAddrs> <am4CfgAddr operation="create"> <ifIpAddr>192.168.2.1</ifIpAddr> <subnetMask>255.255.255.0</subnetMask> <addrType>main</addrType> </am4CfgAddr> </am4CfgAddrs> </ifmAm4> </interface> </interfaces> </ifm> </config>''' m = huawei_connect(ip,netconf_port,netconf_user,netconf_password) #建立连接 m.edit_config(target="running",config=CREATE_INTERFACE)#修改设备running-config，直接修改当前运行配置，把XML配置发送给设备

SFTP

CE12800（基于SSH）

sftp server enable ssh user yeslab service-type all ssh user yeslab sftp-directory cfcard:

python脚本

""" 自动化运维脚本 上传下载 """ import paramiko #SSH远程连接使用 import time #时间库 import os #创建文件夹 def check_folder(ip,path): #创建以path+IP的文件夹 path += ip if not os.path.isdir(path): #检查文件夹是否存在，如果不存在创建文件夹，os.path.isdir()检查路径是否存在，存在返回True，不存在返回Flase，if not取反 os.mkdir(path) print(f"{path}目录已经建立") else: print(f"{ip}文件夹已存在") return path #返回path路径 #time.localtime() def create_filename(time_struct): #下载下来的文件改名存放 str_year = str(time_struct.tm_year) + "_" str_mon = str(time_struct.tm_mon) + "_" str_mday = str(time_struct.tm_mday) + "_" str_hour = str(time_struct.tm_hour) + "_" str_min = str(time_struct.tm_min) + "_" str_sec = str(time_struct.tm_sec) + ".cfg" filename = str_year + str_mon + str_mday + str_hour + str_min + str_sec return filename ip = "192.168.77.2"#ip地址 port = 22#端口号 my_sock = (ip,port)#将IP和端口打包成元组，作为连接参数 backup_path = "config/" folder = check_folder(ip,backup_path) file_name = create_filename(time.localtime()) local_file = folder + "/" + file_name #本地文件位置 device_file = "/vrpcfg.cfg" #远端文件位置 sftp_proc = paramiko.Transport(sock = my_sock)#创建Transport对象，建立底层SSH传输通道 sftp_proc.connect(username="yeslab",password="Admin@456")#用用户名密码进行身份验证，建立加密通道 sftp_con = paramiko.SFTPClient.from_transport(sftp_proc)#在已经建立的SSH通道上开启一个SFTP子会话 sftp_con.get(remotepath=device_file,localpath = local_file) #下载 #sftp_con.put(localpath="config/192.168.77.2/2026_5_19_21_33_51.cfg",remotepath="/test.cfg")#上传 sftp_proc.close()