-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathug_write.py
More file actions
156 lines (134 loc) · 6.65 KB
/
Copy pathug_write.py
File metadata and controls
156 lines (134 loc) · 6.65 KB
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
#!/usr/bin/env python3
'''Чтение правил межсетевого экрана из файла и загрузка их в NGFW Usergate'''
import os
import xlwt
import keyring
import xmlrpc.client
from dotenv import load_dotenv
from modules.ug_client import UsergateClient
from modules.xlsread import ExcelToJson
from modules.fw_rules import FirewallRules
from modules.zones import Zones
from modules.addr_list import AddressList
from modules.services import Services
# def zone_add(zones, item, zone_list, newrule, all_zones_name):
# '''zone_list - src_zones or dst_zones'''
# for zone in item.get(zone_list):
# if zone == 'any':
# newrule[zone_list] = []
# elif all_zones_name.get(zone):
# newrule[zone_list].append(zones.get_by_key(all_zones_name, zone))
# else:
# newrule[zone_list].append(zones.create_zone(all_zones_name, zone))
# all_zones_name = zones.get_all_zones('name')
# return newrule, all_zones_name
# def addr_list_add(addr_lists, item, addr_list, newrule, all_addr_lists_name):
# '''addr_list - src_ips or dst_ips'''
# create_items = False
# for ip_list in item.get(addr_list):
# ip_list_name = ip_list.get('name')
# new_ip_list = ['list_id']
# if ip_list_name == 'any' or ip_list_name == []:
# newrule[addr_list] = []
# elif all_addr_lists_name.get(ip_list_name):
# new_ip_list.append(addr_lists.get_by_key(all_addr_lists_name, ip_list_name))
# newrule[addr_list].append(new_ip_list)
# else:
# new_ip_list_item = {'type': 'network', 'name': ip_list_name}
# new_ip_list.append(addr_lists.create_list(all_addr_lists_name, new_ip_list_item))
# newrule[addr_list].append(new_ip_list)
# all_addr_lists_name = addr_lists.get_all_address_lists('name')
# create_items = True
# return newrule, all_addr_lists_name, create_items
def main():
load_dotenv()
# UGUSER = os.getenv('TESTUGUSER')
# # UGPASS = os.getenv('TESTUGPASS')
# UGPASS = keyring.get_password('usergate-api', UGUSER)
# UGSERVER = os.getenv('TESTUGSERVER')
# UGUSER = os.getenv('TESTUGUSER_6')
# UGPASS = os.getenv('TESTUGPASS_6')
# UGSERVER = os.getenv('TESTUGSERVER_6')
UGUSER = os.getenv('UGMFBUSER')
UGPASS = os.getenv('UGMFBPASS')
UGSERVER = os.getenv('UGMFB')
with UsergateClient(
host=UGSERVER,
username=UGUSER,
password=UGPASS
) as client:
# Version detect:
result = client.server.v2.core.license.info(client.auth_token)
version = result.get('version')
if '6.1.9' in version:
version_619 = True
else:
version_619 = False
e2j = ExcelToJson('all_staff3_normalized_2.xls') # Input file
rule_manager = FirewallRules(client)
zone_manager = Zones(client)
addr_list_manager = AddressList(client)
services_manager = Services(client)
all_zones_name = zone_manager.get_all_zones('name') # Кэш зон для поиска по name
all_addr_lists_name = addr_list_manager.get_all_address_lists('name') # Кэш списков адресов для поиска по name
all_services = services_manager.get_all_services('name')
newrules = e2j.convert_values()
for item in newrules:
newrule = {}
newrule['name'] = item.get('name')
newrule['action'] = item.get('action')[0]
newrule['src_zones'] = []
newrule['dst_zones'] = []
newrule['src_ips'] = []
newrule['dst_ips'] = []
newrule['services'] = []
newrule['log'] = True
newrule['log_session_start'] = True
for zone_list in ['src_zones', 'dst_zones']:
newrule, all_zones_name = Zones.zone_add(zone_manager, item, zone_list, newrule, all_zones_name)
for addr_list_type in ['src_ips', 'dst_ips']:
new_addr_list, all_addr_lists_name, create_items = AddressList.addr_list_add(addr_list_manager, item, addr_list_type, all_addr_lists_name)
newrule[addr_list_type] = new_addr_list
if create_items == True:
for ip_list in item[addr_list_type]:
ip_list_name = ip_list.get('name')
ip_list_id = all_addr_lists_name.get(ip_list_name).get('id')
new_values = ip_list.get('items')
for value in new_values:
new_list_value = {'value': value}
addr_list_manager.create_list_item(ip_list_id, new_list_value)
if item.get('enabled')[0] == 'enabled':
newrule['enabled'] = True
else:
newrule['enabled'] = False
for service_list in item['services']:
service_list_name = service_list.get('name')
new_service_list = ['service']
if service_list == []:
newrule['services'] = []
elif all_services.get(service_list_name):
service_id = all_services.get(service_list_name).get('id')
new_service_list.append(service_id)
if version_619 == True:
newrule['services'].append(service_id)
else:
newrule['services'].append(new_service_list)
else:
new_protocol_list = []
item_protocol_list = item.get('services')
for protocol_list in item_protocol_list:
new_protocol_list_name = protocol_list.get('name')
new_protocol_list_item = {}
for item in protocol_list.get('protocols'):
proto_list = item.split(':')
new_protocol_list_item['proto'] = proto_list[0]
new_protocol_list_item['port'] = proto_list[1].strip()
new_protocol_list.append(new_protocol_list_item)
new_service_item = {'name': new_protocol_list_name, 'protocols': new_protocol_list}
services_manager.create_service(all_services, new_service_item)
all_services = services_manager.get_all_services('name')
# print(new_service_item)
rule_manager.create_rule(newrule)
print(newrule)
if __name__ == '__main__':
main()