-
Notifications
You must be signed in to change notification settings - Fork 5
/
custom-networks.py
185 lines (157 loc) · 7.47 KB
/
custom-networks.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
#!/usr/bin/python3
# Cisco Cyber Vision V4.0
# Network Management
# Version 1.0 - 2023-08-03
import csv
import argparse
import json
import io
import cvconfig
import api
import sys
def main():
parser = argparse.ArgumentParser(prog="custom_networks.py",
description="Custom Network Management")
# Options parsing
parser.add_argument("--token", dest="token", help="Use this token")
parser.add_argument("--center-ip", dest="center_ip",
help="Specified the center FQDN or IPv4 address"
" (default:'cybervision')")
parser.add_argument("--center-port", dest="center_port",
help="Specified the center port (default: %d)" % cvconfig.center_port,
default=cvconfig.center_port)
parser.add_argument("--proxy", dest="proxy",
help="Specified the proxy to use (default: %s)"%cvconfig.proxy,
default=cvconfig.proxy)
parser.add_argument("--encoding", dest="csv_encoding",
help="CSV file encoding, default is %s" % cvconfig.csv_encoding)
parser.add_argument("--delimiter", dest="csv_delimiter",
help="CSV file delimiter, default is %s" % cvconfig.csv_delimiter)
parser.add_argument("--filename", dest="filename", help="Use this filename, default is custom-networks.csv",
default="custom-networks.csv")
# Main Command Parsing
command_group = parser.add_mutually_exclusive_group()
command_group.add_argument("--import",
help="Create custom networks from a CSV file\n",
action="store_true", default=False, dest="create")
command_group.add_argument("--export",
help="Export all custom networks into a CSV file\n",
action="store_true", default=False, dest="command_export")
command_group.add_argument("--delete",
help="Delete all custom networks",
action="store_true", default=False, dest="delete")
args = parser.parse_args()
# Handle Cybervision configuration
token = set_conf(args.token, cvconfig.token)
center_ip = set_conf(args.center_ip, cvconfig.center_ip)
center_port = set_conf(args.center_port, cvconfig.center_port)
proxy = set_conf(args.proxy, cvconfig.proxy)
csv_encoding = set_conf(args.csv_encoding, cvconfig.csv_encoding)
csv_delimiter = set_conf(args.csv_delimiter, cvconfig.csv_delimiter)
if not token or not center_ip:
print("TOKEN and CENTER_IP are mandatory, check cvconfig.py or us --token/--center-ip")
if args.filename:
csv_file = args.filename
if args.create:
if not csv_file:
print("Missing required parameter: --csv-file path_to_csv_file")
print_csv_file_format()
return
create_networks(center_ip=center_ip, center_port=center_port, token=token, proxy=proxy, csv_file=csv_file,
csv_delimiter=csv_delimiter)
return
if args.delete:
delete_networks(center_ip=center_ip, center_port=center_port, token=token, proxy=proxy)
return
if args.command_export:
response = get_network_data(center_ip=center_ip, center_port=center_port, token=token, proxy=proxy)
csv_data = convert_json_to_csv(response, csv_delimiter=csv_delimiter)
with open(csv_file, "w", encoding=csv_encoding) as file:
file.write(csv_data)
print(csv_data)
return
parser.print_help()
def print_csv_file_format():
print("Valid CSV file format example")
print("name,type,ip_range,vlan_id,duplicated,split_devices_per_sensor")
print("Network3,IT Internal,172.16.0.0/16,2003,False,True")
print("Network2,IT Internal,192.168.1.0/24,1001,True,False")
print("Note: In case name has a comma, then, provide the name in double quotes")
def create_networks(center_ip, center_port, token, proxy, csv_file, csv_delimiter):
networks = read_csv_file(csv_file, csv_delimiter)
if networks:
with api.APISession(center_ip, center_port, token, proxy) as session:
response = api.post_route(session, '/api/3.0/networks/', json=networks)
if response.status_code != 200:
print_response(response)
print('ERROR: Failed to create custom networks')
else:
print_response(response)
print('INFO: Successfully created custom networks')
else:
print("INFO: CSV file is empty")
def delete_networks(center_ip, center_port, token, proxy):
network_data = get_network_data(center_ip=center_ip, center_port=center_port, token=token, proxy=proxy)
if network_data:
custom_network_ids = [entry["id"] for entry in network_data]
with api.APISession(center_ip, center_port, token, proxy) as session:
response = api.delete_route(session, '/api/3.0/networks', json=custom_network_ids)
if response.status_code != 200:
print_response(response)
print('ERROR: Failed to delete custom networks')
else:
print_response(response)
print('INFO: Successfully deleted custom networks')
else:
print('INFO: Nothing to delete')
def get_network_data(center_ip, center_port, token, proxy):
with api.APISession(center_ip, center_port, token, proxy) as session:
return api.get_route(session, '/api/3.0/networks')
def read_csv_file(csv_file, csv_delimiter):
networks = []
try:
with open(csv_file, newline='') as csvfile:
reader = csv.DictReader(csvfile, delimiter=csv_delimiter)
for row in reader:
vlan_id = int(row['vlan_id']) if row['vlan_id'].strip() else None
network = {
"name": row['name'],
"type": row['type'],
"ipRange": row['ip_range'],
"vlanId": vlan_id,
"duplicated": row['duplicated'].lower() == 'true',
"splitDevicesPerSensor": row['split_devices_per_sensor'].lower() == 'true'
}
networks.append(network)
except FileNotFoundError:
print("Error: File '{csv_file}' not found.")
sys.exit(1)
except BaseException as ex:
print("Error: File '{csv_file}' is not well formatted.")
print(ex)
print_csv_file_format()
sys.exit(1)
return networks
def print_response(response):
print(f'Response Status :{response.status_code}')
try:
print(f'Response Body: {json.dumps(response.json(), indent=2)}')
except BaseException:
print(f'Response Body: {response.content.decode()}')
def convert_json_to_csv(json_data, csv_delimiter):
csv_output = io.StringIO()
# name,type,ip_range,vlan_id,duplicated,split_devices_per_sensor
fieldnames = ['id', 'name', 'type', 'ipRange',
'vlanId', 'duplicated', 'splitDevicesPerSensor']
headers = ['id', 'name', 'type', 'ip_range', 'vlan_id', 'duplicated', 'split_devices_per_sensor']
writer = csv.DictWriter(csv_output, fieldnames=fieldnames, delimiter=csv_delimiter)
writer.writerow(dict(zip(fieldnames, headers)))
writer.writerows(json_data)
csv_output.seek(0)
return csv_output.getvalue()
def set_conf(arg, conf):
if arg and arg != conf:
return arg
return conf
if __name__ == "__main__":
main()