我已经创建了保存我的实时VPN数据的对象列表。我在数据库中也有VPN数据存储。在
我希望能够将实时数据对象列表与数据库进行比较,如果数据匹配,则执行“某些操作”
如果一个对象是anyconnect,用户名将与数据库匹配;如果一个对象是站点到站点,则其对等端将与BD匹配
我需要检查一下物品清单,找到匹配的
基本上,我所做的是将实时数据与数据库数据进行比较,如果在数据库中找到实时数据,我将更新数据库记录,说明服务已“启动”;如果找不到,则服务将“关闭”
有谁能为我指出实现这一目标的正确方向吗?在
编辑:
实时数据列表
>>> for i in active_vpns:
... print('username: {} peer_ip: {}'.format(i.username,i.peer_ip))
username: remote_vpn.user1: 1.2.3.4
username: peer_ip: 1.1.1.1
username: peer_ip: 2.2.2.2
username: peer_ip: 3.3.3.3
数据库数据列表
^{pr2}$现在在这个例子中,我想将peer_ip4.4.4.4设置为Down and remote_vpn.user2其他的服务都准备好了
完整代码:
#!/usr/bin/env python
from django_setup import setup
setup()
import re
import ipaddress
from netmiko import ConnectHandler
from monitoring.models import ThirdPartyService
from datetime import datetime
class VPNData(object):
def __init__(self, service_name='', username='', vpn_peer_ip='', duration='', data_transmit='', data_receive='', timestamp=''):
self.service_name = service_name
self.username = username
self.vpn_peer_ip = vpn_peer_ip
self.duration = duration
self.data_transmit = data_transmit
self.data_receive = data_receive
self.timestamp = timestamp
def __repr__(self):
return '{} {}'.format(self.__class__.__name__, self.username)
def to_megabytes(bytes, bsize=1024):
r = float(bytes)
for i in range(2):
r = r / bsize
return round(r)
#creds for logging on to devices
username = 'monitoring'
password = '*******'
device_ip = '10.10.10.10'
# firewall conenction details
firewall = {
'device_type': 'cisco_asa',
'ip': device_ip,
'username': username,
'password': password,
'port' : 22, # optional, defaults to 22
'secret': '', # optional, defaults to ''
'verbose': False, # optional, defaults to False
}
# connect to firewall
conn_fw = ConnectHandler(**firewall)
# get anyconnect sessions
anyconnect_connections = conn_fw.send_command('show vpn-sessiondb anyconnect')
# get site to site vpn sessions
s2s_connections = conn_fw.send_command('show vpn-sessiondb l2l')
# disconnect ssh session
conn_fw.disconnect()
active_vpns =[]
# create list of anyconnect sessions
anyconnect_sessions = re.findall(r'(?s)(?<=Username :)(.*?)(?=Security Grp)', anyconnect_connections, flags=re.S)
# process anyconnect sessions
for ac_sess in anyconnect_sessions:
# create object for data
vpn = VPNData()
# get username
username = re.findall(r'(?s).*?(?=Index)', ac_sess, flags=re.S)
vpn.username = username[0].strip()
# get peer ip
peer_ip = re.findall(r'(?s)(?<=Public IP : )(.*?)(?=Protocol)', ac_sess, flags=re.S)
vpn.peer_ip = peer_ip[0].strip()
# get duration
duration = re.findall(r'(?s)(?<=Duration : )(.*?)(?=Inactivity)', ac_sess, flags=re.S)
vpn.duration = duration[0].strip()
# transmit bytes
data_transmit = re.findall(r'(?s)(?<=Bytes Tx : )(.*?)(?=Bytes Rx)', ac_sess, flags=re.S)
vpn.data_transmit = data_transmit[0].strip()
# receive bytes
data_receive = re.findall(r'(?s)(?<=Bytes Rx : )(.*?)(?=Group Policy)', ac_sess, flags=re.S)
vpn.data_receive = data_receive[0].strip()
# Add VPN to List
active_vpns.append(vpn)
# Add extra string to get the last connection
s2s_connections += 'Connection'
# create list of sie to site vpn sessions
s2s_sessions = re.findall(r'(?s)(?<=Connection :)(.*?)(?=Connection)', s2s_connections, flags=re.S)
# process site to site sessions
for s2s_sess in s2s_sessions:
# create object for data
vpn = VPNData()
# get peer ip
peer_ip = re.findall(r'(?s).*?(?=Index)', s2s_sess, flags=re.S)
vpn.peer_ip = peer_ip[0].strip()
# get duration
duration = re.findall(r'(?s)(?<=Duration : ).*', s2s_sess, flags=re.S)
vpn.duration = duration[0].strip()
# transmit bytes
data_transmit = re.findall(r'(?s)(?<=Bytes Tx : )(.*?)(?=Bytes Rx)', s2s_sess, flags=re.S)
vpn.data_transmit = data_transmit[0].strip()
# receive bytes
data_receive = re.findall(r'(?s)(?<=Bytes Rx : )(.*?)(?=Login Time)', s2s_sess, flags=re.S)
vpn.data_receive = data_receive[0].strip()
active_vpns.append(vpn)
# get list of VPNs from DB
vpn_services = ThirdPartyService.objects.all()
for service in vpn_services:
vpn_data = ThirdPartyService.objects.update_or_create(
defaults={
'service_name' : 'anyconnect',
'username' : username,
'vpn_peer_ip' : peer_ip,
'duration' : duration,
'data_transmit' : to_megabytes(data_transmit),
'data_receive' : to_megabytes(data_receive),
'timestamp' : datetime.now().strftime("%Y-%m-%d %H:%M:%S")
},
service_name = 'anyconnect',
username = username
)
print('{0} Script Completed'.format(datetime.now().strftime('%d-%m-%Y %H:%M:%S')))
我将从DB条目构建一个
dict
或set
,然后用它来测试活动连接。在比如:
相关问题 更多 >
编程相关推荐