1 from jinja2 import Template
7 BASE_DIR = os.path.dirname(os.path.dirname(__file__))
9 HEADERS = {'Content-Type': 'application/json'}
10 AUTH = requests.auth.HTTPBasicAuth('healthcheck', 'zb!XztG34')
13 def get_tosca_policy(policy):
14 pol = json.loads(policy)
16 'tosca_definitions_version': 'tosca_simple_yaml_1_1_0',
17 'topology_template': {
21 return json.dumps(tosca_policy)
23 def gen_policy(template_dir, gen_dir, filename, jinja_args):
24 with open(os.path.join(template_dir, filename), 'r') as file:
25 contents = file.read()
26 tm = Template(contents)
27 gen = tm.render(jinja_args)
28 tosca_policy = get_tosca_policy(gen)
29 with open(os.path.join(gen_dir, filename), 'w') as file:
30 file.write(tosca_policy)
32 def create_and_push_policies(policy_dir):
33 for filename in os.listdir(policy_dir):
34 if filename.endswith('.json'):
35 with open(os.path.join(policy_dir, filename), 'r') as file:
36 data = json.loads(file.read())
37 metadata = create_policy(data)
41 def delete_policies(policy_dir):
42 for filename in os.listdir(policy_dir):
43 if filename.endswith('.json'):
44 with open(os.path.join(policy_dir, filename), 'r') as file:
45 data = json.loads(file.read())
46 policy_id = list(data['topology_template']['policies'][0].keys())[0]
47 undeploy_policy(policy_id)
48 metadata = delete_policy(data)
50 def create_policy(data):
51 policy = data['topology_template']['policies'][0]
52 content = policy[list(policy.keys())[0]]
53 policy_type = content['type']
54 type_version = content['type_version']
55 policy_url = "https://policy-api:6969"
56 path = '/policy/api/v1/policytypes/{}/versions/{}/policies'.format(policy_type, type_version)
57 url = policy_url + path
59 response = requests.post(url, headers=HEADERS, auth=AUTH, data=json.dumps(data), verify=False)
60 except Exception as e:
63 if response.status_code == 200:
64 print('Policy {} created'.format(content['metadata']['policy-id']))
65 return content['metadata']
67 print(response.content)
70 def push_policy(metadata):
71 data = {'policies': [metadata]}
72 policy_url = "https://policy-pap:6969"
73 path = '/policy/pap/v1/pdps/policies'
74 url = policy_url + path
76 response = requests.post(url, headers=HEADERS, auth=AUTH, data=json.dumps(data), verify=False)
77 except Exception as e:
79 print("Cannot push policy {}".format(metadata['policy-id']))
80 if response.status_code == 200:
81 print("Policy {} pushed".format(metadata['policy-id']))
83 print(response.content)
85 def undeploy_policy(policy_id):
86 policy_url = "https://policy-pap:6969"
87 path = '/policy/pap/v1/pdps/policies/{}'.format(policy_id)
88 url = policy_url + path
90 response = requests.delete(url, headers=HEADERS, auth=AUTH, verify=False)
91 except Exception as e:
93 print("Cannot undeploy policy {}".format(policy_id))
94 if response.status_code == 200:
95 print("Policy {} undeployed".format(policy_id))
97 print(response.content)
99 def delete_policy(data):
100 policy = data['topology_template']['policies'][0]
101 content = policy[list(policy.keys())[0]]
102 policy_type = content['type']
103 type_version = content['type_version']
104 policy_id = content['metadata']['policy-id']
105 version = content['version']
106 policy_url = "https://policy-api:6969"
107 path = '/policy/api/v1/policytypes/{}/versions/{}/policies/{}/versions/{}'.format(policy_type, type_version, policy_id, version)
108 url = policy_url + path
110 response = requests.delete(url, headers=HEADERS, auth=AUTH, data=json.dumps(data), verify=False)
111 except Exception as e:
114 if response.status_code == 200:
115 print('Policy {} deleted'.format(content['metadata']['policy-id']))
116 return content['metadata']
118 print(response.content)
121 def generate_nssi_policies(jinja_args):
122 template_dir = BASE_DIR + 'nssi_policies'
123 gen_dir = BASE_DIR + 'gen_nssi_policies'
125 if not os.path.exists(gen_dir):
128 for filename in os.listdir(template_dir):
129 if filename.endswith('.json'):
130 gen_policy(template_dir, gen_dir, filename, jinja_args)
132 def generate_nsi_policies(jinja_args):
133 template_dir = BASE_DIR + 'nsi_policies'
134 gen_dir = BASE_DIR + 'gen_nsi_policies'
136 if not os.path.exists(gen_dir):
139 for filename in os.listdir(template_dir):
140 if filename.endswith('.json'):
141 gen_policy(template_dir, gen_dir, filename, jinja_args)
143 def create_policy_types(policy_dir):
144 for filename in os.listdir(policy_dir):
145 if filename.endswith('.json'):
146 with open(os.path.join(policy_dir, filename), 'r') as file:
147 data = json.loads(file.read())
148 create_policy_type(data)
150 def create_policy_type(data):
151 policy_url = "https://policy-api:6969"
152 path = '/policy/api/v1/policytypes'
153 url = policy_url + path
155 response = requests.post(url, headers=HEADERS, auth=AUTH, data=json.dumps(data), verify=False)
156 except Exception as e:
159 if response.status_code == 200:
160 print('Policy type created')
162 print(response.content)
168 if action == "generate_nssi_policies":
170 'service_name': sys.argv[2],
172 'attribute': sys.argv[4]
174 generate_nssi_policies(jinja_args)
176 elif action == "create_and_push_policies":
177 policy_dir = sys.argv[2]
178 create_and_push_policies(policy_dir)
180 elif action == "delete_policies":
181 policy_dir = sys.argv[2]
182 delete_policies(policy_dir)
184 elif action == "generate_nsi_policies":
186 'service_name': sys.argv[2]
188 generate_nsi_policies(jinja_args)
190 elif action == "create_policy_types":
191 policy_dir = sys.argv[2]
192 create_policy_types(policy_dir)