diff --git a/app_utils/orchestrator_utils.py b/app_utils/orchestrator_utils.py
index acfb151ed0458ad76915d4dfeb0226bd3bb5f23d..c16f764c12f8d1155ed5f77792e8f57be2101fd7 100644
--- a/app_utils/orchestrator_utils.py
+++ b/app_utils/orchestrator_utils.py
@@ -1,31 +1,39 @@
# SPDX-License-Identifier: Apache-2.0
#orchestrator_utils.py
+
from __future__ import print_function
-from app_utils import catalogue_utils, security_utils, xml_utils, editor_utils
-import xmltodict
-from pprint import pprint
-import time
+
+import json
+import re
+import datetime
+import clouditor_orchestrator_client.models
+
+from app_utils import catalogue_utils, security_utils
+from clouditor_orchestrator_client.model.metric_implementation import MetricImplementation
+from clouditor_orchestrator_client.model.metric_configuration import MetricConfiguration
+from clouditor_orchestrator_client.paths.v1_orchestrator_metrics_implementation_metric_id_implementation.put import \
+ ApiResponseFor200 as MetricImplementationResponse200
+from clouditor_orchestrator_client.paths. \
+ v1_orchestrator_cloud_services_cloud_service_id_metric_configurations_metric_id.put import \
+ ApiResponseFor200 as MetricConfigurationResponse200
+from clouditor_orchestrator_client.apis.tags import orchestrator_api
+from clouditor_orchestrator_client.rest import ApiException
+from clouditor_orchestrator_client_legacy.rest import ApiException as ApiExceptionLegacy
import clouditor_orchestrator_client_legacy
-from clouditor_orchestrator_client_legacy.api import orchestrator_api
-from clouditor_orchestrator_client_legacy.rest import ApiException
-from pprint import pprint
-from keycloak import KeycloakOpenID
from config import *
-# Defining the host is optional and defaults to http://localhost
-# See configuration.py for a list of all supported configuration parameters.
def get_metrics_from_clouditor(request=None):
token = security_utils.get_access_token(request)
orchestrator_configuration = clouditor_orchestrator_client_legacy.Configuration(host=ORCHESTRATOR_API_URL)
with clouditor_orchestrator_client_legacy.ApiClient(orchestrator_configuration, header_name="Authorization",
- header_value=f"Bearer {token}") as orchstrator_api_client:
- orchestrator_api_instance = orchestrator_api.OrchestratorApi(orchstrator_api_client)
+ header_value=f"Bearer {token}") as orchestrator_api_client_legacy:
+ orchestrator_api_instance = clouditor_orchestrator_client_legacy.OrchestratorApi(orchestrator_api_client_legacy)
try:
api_response = orchestrator_api_instance.orchestrator_list_metrics()
- return (api_response)
- except ApiException as e:
+ return api_response
+ except ApiExceptionLegacy as e:
print("Exception when calling OrchestratorApi->orchestrator_list_metrics: %s\n" % e)
@@ -33,8 +41,8 @@ def get_metric_from_clouditor(metric_id, request=None):
token = security_utils.get_access_token(request)
orchestrator_configuration = clouditor_orchestrator_client_legacy.Configuration(host=ORCHESTRATOR_API_URL)
with clouditor_orchestrator_client_legacy.ApiClient(orchestrator_configuration, header_name="Authorization",
- header_value=f"Bearer {token}") as orchestrator_api_client:
- orchestrator_api_instance = orchestrator_api.OrchestratorApi(orchestrator_api_client)
+ header_value=f"Bearer {token}") as orchestrator_api_client_legacy:
+ orchestrator_api_instance = clouditor_orchestrator_client_legacy.OrchestratorApi(orchestrator_api_client_legacy)
try:
api_response = orchestrator_api_instance.orchestrator_get_metric(metric_id)
return api_response
@@ -44,39 +52,66 @@ def get_metric_from_clouditor(metric_id, request=None):
def get_metric_implementation_from_clouditor(metric_id, request=None):
token = security_utils.get_access_token(request)
- orchestrator_configuration = clouditor_orchestrator_client_legacy.Configuration(host=ORCHESTRATOR_API_URL)
- with clouditor_orchestrator_client_legacy.ApiClient(orchestrator_configuration, header_name="Authorization",
- header_value=f"Bearer {token}") as orchestrator_api_client:
+ orchestrator_configuration = clouditor_orchestrator_client.Configuration(host=ORCHESTRATOR_API_URL)
+ '''
+ with clouditor_orchestrator_client.ApiClient(orchestrator_configuration, header_name="Authorization",
+ header_value=f"Bearer {token}") as orchestrator_api_client:
orchestrator_api_instance = orchestrator_api.OrchestratorApi(orchestrator_api_client)
+ '''
+ with clouditor_orchestrator_client_legacy.ApiClient(orchestrator_configuration, header_name="Authorization",
+ header_value=f"Bearer {token}") as orchestrator_api_client_legacy:
+ orchestrator_api_instance = clouditor_orchestrator_client_legacy.OrchestratorApi(orchestrator_api_client_legacy)
+
try:
api_response = orchestrator_api_instance.orchestrator_get_metric_implementation(metric_id)
return api_response
except ApiException as e:
print("Exception when calling OrchestratorApi->orchestrator_get_metric_implementation: %s\n" % e)
+ return None
def update_metric_implementation_to_clouditor(metric_id, rego_code, request=None):
- ########### update metric implementation ############
token = security_utils.get_access_token(request)
- orchestrator_configuration = clouditor_orchestrator_client_legacy.Configuration(host=ORCHESTRATOR_API_URL)
- with clouditor_orchestrator_client_legacy.ApiClient(orchestrator_configuration, header_name="Authorization",
- header_value=f"Bearer {token}") as orchestrator_api_client:
- orchestrator_api_instance = orchestrator_api.OrchestratorApi(orchestrator_api_client)
+ orchestrator_configuration = clouditor_orchestrator_client.Configuration(host=ORCHESTRATOR_API_URL)
+
+ with clouditor_orchestrator_client.ApiClient(orchestrator_configuration, header_name="Authorization",
+ header_value=f"Bearer {token}") as api_client:
+ api_instance = orchestrator_api.OrchestratorApi(api_client)
+ '''
+ with clouditor_orchestrator_client_legacy.ApiClient(orchestrator_configuration, header_name="Authorization",
+ header_value=f"Bearer {token}") as orchestrator_api_client_legacy:
+ api_instance = clouditor_orchestrator_client_legacy.OrchestratorApi(orchestrator_api_client_legacy)
+ '''
+
+ path_params = {
+ 'implementation.metric_id': metric_id,
+ }
+ metric_implementation = MetricImplementation(
+ metricId=metric_id,
+ lang="LANGUAGE_REGO",
+ code=rego_code,
+ )
+
try:
- metric_implementation = clouditor_orchestrator_client_legacy.models.MetricImplementation(metric_id = metric_id, lang='REGO', code=rego_code)
- api_response = orchestrator_api_instance.orchestrator_update_metric_implementation(metric_id,
- metric_implementation)
- return (api_response)
+ api_response = api_instance.orchestrator_update_metric_implementation(metric_implementation,
+ path_params=path_params)
+ return api_response
except ApiException as e:
- print("Exception when calling OrchestratorApi->orchestrator_update_metric_implementation: %s\n" % e)
+ return str(e)
def get_metric_configuration_from_clouditor(service_id, metric_id, request=None):
token = security_utils.get_access_token(request)
- orchestrator_configuration = clouditor_orchestrator_client_legacy.Configuration(host=ORCHESTRATOR_API_URL)
- with clouditor_orchestrator_client_legacy.ApiClient(orchestrator_configuration, header_name="Authorization",
- header_value=f"Bearer {token}") as orchestrator_api_client:
+ orchestrator_configuration = clouditor_orchestrator_client.Configuration(host=ORCHESTRATOR_API_URL)
+ '''
+ with clouditor_orchestrator_client.ApiClient(orchestrator_configuration, header_name="Authorization",
+ header_value=f"Bearer {token}") as orchestrator_api_client:
orchestrator_api_instance = orchestrator_api.OrchestratorApi(orchestrator_api_client)
+ '''
+ with clouditor_orchestrator_client_legacy.ApiClient(orchestrator_configuration, header_name="Authorization",
+ header_value=f"Bearer {token}") as orchestrator_api_client_legacy:
+ orchestrator_api_instance = clouditor_orchestrator_client_legacy.OrchestratorApi(orchestrator_api_client_legacy)
+
try:
api_response = orchestrator_api_instance.orchestrator_get_metric_configuration(service_id, metric_id)
return api_response
@@ -84,18 +119,30 @@ def get_metric_configuration_from_clouditor(service_id, metric_id, request=None)
print("Exception when calling OrchestratorApi->orchestrator_get_metric_configuration: %s\n" % e)
-def update_metric_configuration_to_clouditor(service_id, metric_id, metric_configuration, request=None):
+def update_metric_configuration_to_clouditor(service_id, metric_id, data, request=None):
token = security_utils.get_access_token(request)
- orchestrator_configuration = clouditor_orchestrator_client_legacy.Configuration(host=ORCHESTRATOR_API_URL)
- with clouditor_orchestrator_client_legacy.ApiClient(orchestrator_configuration, header_name="Authorization",
- header_value=f"Bearer {token}") as orchestrator_api_client:
- orchestrator_api_instance = orchestrator_api.OrchestratorApi(orchestrator_api_client)
+ orchestrator_configuration = clouditor_orchestrator_client.Configuration(host=ORCHESTRATOR_API_URL)
+
+ with clouditor_orchestrator_client.ApiClient(orchestrator_configuration, header_name="Authorization",
+ header_value=f"Bearer {token}") as api_client:
+ api_instance = orchestrator_api.OrchestratorApi(api_client)
+ path_params = {
+ 'cloudServiceId': service_id,
+ 'metricId': metric_id,
+ }
+ metric_configuration = MetricConfiguration(
+ operator=data['operator'],
+ target_value=data['target_value'],
+ is_default=True,
+ metric_id=metric_id,
+ cloud_service_id=service_id
+ )
try:
- api_response = orchestrator_api_instance.orchestrator_update_metric_configuration(service_id, metric_id,
- metric_configuration)
+ api_response = api_instance.orchestrator_update_metric_configuration(path_params=path_params,
+ body=metric_configuration)
return api_response
except ApiException as e:
- print("Exception when calling OrchestratorApi->orchestrator_update_metric_configuration: %s\n" % e)
+ return str(e)
def get_metric_configuration_from_obligation(statement_info):
@@ -106,169 +153,176 @@ def get_metric_configuration_from_obligation(statement_info):
operator (str): the operator of the obligation (e.g. <=)
target_value (str): the target value of the obligation (e.g. 40)
"""
- s = statement_info
- param = s[s.find("{")+1:s.find("}")].split()[0]
- option = s[s.find("{")+1:s.find("}")].split()[1]
-
- operator = param.split('=',1)[1]
- target_value = option.split('=',1)[1]
- return operator, target_value
+ operator = re.findall(r"(?<==).*?(?= option)", statement_info)[0]
+ target_value = re.findall(r"(?<=option=).*?(?=})", statement_info)[0]
+ return operator, target_value
def create_metric_configuration(obligation):
""" Create the 'data' part of the rego rule (metric configuration) from obligation info
Args: i
- metric_name (str): unique metric name to retrieve metric from the cataolgue (e.g. BackupEnabled)
+ metric_name (str): unique metric name to retrieve metric from the catalogue (e.g. BackupEnabled)
Returns:
metric_configuration (dict): dict containing the metric configuration parameters
"""
-
+
operator, target_value = get_metric_configuration_from_obligation(obligation['@statementInfo'])
- metric_configuration = { "operator" : operator,
- "target_value" : target_value
- }
+
+ if operator == 'na':
+ operator = ""
+
+ if target_value == 'na':
+ target_value = ""
+
+ metric_configuration = {"operator": operator,
+ "target_value": target_value
+ }
return metric_configuration
-
def create_metric_implementation(metric_name):
""" Create the 'rego' part for a metric (metric implementation)
- Args:
- metric_name (str): unique metric name to retrieve metric from the cataolgue (e.g. BackupEnabled)
+ Args:
+ metric_name (str): unique metric name to retrieve metric from the catalogue (e.g. BackupEnabled)
Returns:
-
+
"""
- #TODO: if a metric is associated to multiple targetResourceType, create 2 obligations (split into two oblig)
+ CLOUDITOR_PACKAGE = 'package clouditor.metrics.'
+ CLOUDITOR_IMPORT = 'import data.clouditor.'
+ CLOUDITOR_COMPLIANT = 'compliant{'
+ TLS_METRIC = 'TLSVersion'
+ IS_IN_VALUE = 'isIn'
+ COMPARE_VALUE = 'compare'
+ IS_IN_DATA = '(data.target_value, version)'
+ COMPARE_DATA = '(data.operator, data.target_value, SECURITYFEATURE)'
+ TARGET_TYPES = 'target_types := ['
+ target_types = ''
+
metric = catalogue_utils.get_metric_by_name_from_catalogue(metric_name)
-
- head = ['package clouditor','default compliant = false','default applicable = false']
+ COMPARE_DATA = COMPARE_DATA.replace('SECURITYFEATURE',metric['securityFeature'])
+
+ clouditor_package = CLOUDITOR_PACKAGE + re.sub('(?<!^)(?=[A-Z])', '_', metric_name).lower()
+
+ if metric['name'] == TLS_METRIC:
+ if type(metric['targetValue']) is list:
+ clouditor_import = CLOUDITOR_IMPORT + IS_IN_VALUE
+ clouditor_compliant = CLOUDITOR_COMPLIANT + IS_IN_VALUE + IS_IN_DATA + '}'
+ else:
+ clouditor_import = CLOUDITOR_IMPORT + COMPARE_VALUE
+ clouditor_compliant = CLOUDITOR_COMPLIANT + COMPARE_VALUE + COMPARE_DATA + '}'
+ else:
+ if metric['operator'] == 'in':
+ clouditor_import = CLOUDITOR_IMPORT + IS_IN_VALUE
+ clouditor_compliant = CLOUDITOR_COMPLIANT + IS_IN_VALUE + IS_IN_DATA + '}'
+ else:
+ clouditor_import = CLOUDITOR_IMPORT + COMPARE_VALUE
+ clouditor_compliant = CLOUDITOR_COMPLIANT + COMPARE_VALUE + COMPARE_DATA + '}'
+
+ #patch for inserting TargetResourceType in the rego code
+ target_types = TARGET_TYPES + '"{0}"]'.format(metric['targetResourceType'])
+
+ if type(metric['targetResourceType']) is list:
+ for idx, resource in enumerate(metric['targetResourceType']):
+ if idx == len(metric['targetResourceType']) - 1:
+ target_types = target_types + '"{0}"]'.format(resource)
+ else:
+ target_types = target_types + '"{0}", '.format(resource)
+ target_types = TARGET_TYPES + target_types
+
+ head = [clouditor_package, clouditor_import, 'default compliant = false', 'default applicable = false']
line1 = "\n".join(head)
- #lines2 = ['target_types := ["ObjectStorage", "FileStorage"]']
- line2 = "target_types := [\"" + "\",\"".join(metric['targetResourceType'].split(",")) + "\"]"
+
+ # TODO: this field will need to be updated
line3 = "applicable {isIn(input.type, target_types)}"
- line4 = "compliant {isIn(data.target_value, input." + metric_name + ")}"#input.httpEndpoint.transportEncryption.tlsVersion
- line5 = "compliant {compare(data.operator, data.target_value, input." + metric_name + ")}"#input.httpEndpoint.transportEncryption.tlsVersion
- lines = [line1, line2, line3, line4, line5]
+ lines = [line1, target_types, line3, clouditor_compliant]
metric_implementation = "\n".join(lines)
- return metric_implementation
+ return metric, metric_implementation
+def save_data_on_error(metric, metric_implementation=None, metric_configuration=None, implementation_details=None,
+ configuration_details=None):
+ DATA_ERROR_PATH = DATA_PATH + 'rego_generation_error'
+ if not os.path.exists(DATA_ERROR_PATH):
+ os.mkdir(DATA_ERROR_PATH)
-def map_reo(reoid,service_id):
- #read reo from cnl store
+ current_date = datetime.datetime.now()
+ current_date = current_date.strftime("%d-%m-%Y-%H:%M:%S")
+ data_directory_path = os.path.join(DATA_ERROR_PATH, metric['name'] + "_" + current_date + '/')
+ os.mkdir(data_directory_path)
+
+ with open(data_directory_path + 'metric.json', 'w') as metric_file:
+ json.dump(metric, metric_file, indent=4)
+
+ if metric_implementation is not None:
+ with open(data_directory_path + 'metric_implementation.txt', 'w') as metric_implementation_file:
+ metric_implementation_file.write(metric_implementation)
+
+ with open(data_directory_path + 'metric_implementation_problems.txt', 'w') as metric_implementation_details:
+ metric_implementation_details.write(implementation_details)
+
+ if metric_configuration is not None:
+ with open(data_directory_path + 'metric_configuration.json', 'w') as metric_configuration_file:
+ json.dump(metric_configuration, metric_configuration_file, indent=4)
+
+ with open(data_directory_path + 'metric_configuration_problems.txt', 'w') as metric_configuration_details:
+ metric_configuration_details.write(configuration_details)
+
+
+def map_reo(service_id, reo_dict):
+ metric_implementation_error = False
+ metric_configuration_error = False
+
+ obl_list = []
try:
- reo_xml = editor_utils.get_reo(reoid)
- #reo xml 2 dict
- reo_dict = xmltodict.parse(reo_xml)
- except TypeError as e:
- return 'Cannot find requested reo id'
- #read obligations
- resp_configuration = []
- resp_implementation = []
-
- try:#no obligations
- obl_list = reo_dict['dsa']['obligations']['obligation']
+ if type(reo_dict['dsa']['obligations']['obligation']) is dict:
+ obl_list.append(reo_dict['dsa']['obligations']['obligation'])
+ else:
+ obl_list = reo_dict['dsa']['obligations']['obligation']
except KeyError as e:
- print('Mapping applied to an object without obligations')
- #TODO:return an error status to the editor
- return "Trying to map a REO with no obligations"
-
- try:#many obligations
- #for each obligation
- for obligation in obl_list:
-
- metric_id = (obligation['@metricId'])
- print('CREATING REGO RULE FOR METRIC {0}'.format(metric_id))
- #create metric configuration (operator and target value)
- metric_configuration = create_metric_configuration(obligation)
- #create metric implementation (other fields)
- metric_implementation = create_metric_implementation(metric_id)
- #print('-----')
- #print('Metric implementation:\n{0}'.format(metric_implementation))
- #print('-----')
- #print('Metric configuration:\n{0}'.format(metric_configuration))
- #print('-----')
- #TODO:replace request None
- #TODO: check if response id ok for each obligation print('-----')
- #resp_implementation.append(update_metric_implementation_to_clouditor(metric_id,metric_implementation, request=None))
- resp_configuration.append(update_metric_configuration_to_clouditor(service_id, metric_id, metric_configuration, request=None))
- return 'success'
- except TypeError as e:
- metric_id = reo_dict['dsa']['obligations']['obligation']['@metricId']
- print('CREATING REGO RULE FOR METRIC {0}'.format(metric_id))
- #single obligation
- obligation = reo_dict['dsa']['obligations']['obligation']
- metric_configuration = create_metric_configuration(obligation)
- metric_implementation = create_metric_implementation(metric_id)
- #resp_implementation.append(update_metric_implementation_to_clouditor(metric_id,metric_implementation, request=None))
- resp_configuration.append(update_metric_configuration_to_clouditor(service_id, metric_id, metric_configuration, request=None))
- #print('-----')
- #print (resp_implementation)
- #print('-----')
- #print(resp_configuration)
- #print('-----')
- #TODO: check if response from orchestrator is ok and return success or not to the editor
- return 'success'
-
-if __name__ == '__main__':
- #dsl_data('AntimalwareScanFrequencyQ1')
- #map_reo('DSA-b1f8558d-c29a-458f-ba77-bea3ecad89a2')
- #metric = catalogue_utils.get_metric_by_name_from_catalogue('BackupEnabled')
- #pprint(metric)
- #metric_implementation = create_metric_implementation('BackupEnabled')
- #print(metric_implementation)
- #metric_id = 'AntiMalwareEnabled'
- #metric_id = 'TLSVersion'
- #metric_id = 'ActivityLoggingEnable'
- #metric_id = 'BackupMonitoringPolicyCheckQ1'
- metric_id = 'NetworkAccessPolicy01'
- service_id = '00000000-0000-0000-000000000000'
-
-
- #TEST GET METRICS
- #print(get_metrics_from_clouditor())
-
- #TEST GET METRICS
- #print(get_metric_configuration_from_clouditor(service_id,metric_id))
- #print(get_metric_configuration_from_clouditor(service_id,'NumberOfThreatsFound'))
- #print(get_metric_configuration_from_clouditor(service_id,'MalwareProtectionOutput'))
-
-
- #TEST MAP_REO
- #reo_id_0obl = 'DSA-c07a5704-1a47-4ba8-807f-81b1e23357ff'
- #reo_id_1obl = 'DSA-3b8f1375-3614-4fa6-bfc7-82ee5c4fdb0b'
- #reo_id_many = 'DSA-e6094009-5867-4e3c-8ec4-e2fad8592bc7'
- #reo_id = 'nonesiste'
- reo_id = 'DSA-e89039e0-4674-425b-97a2-6e4b60e6236b'
- resp = map_reo(reo_id, service_id)
- print(resp)
-
- #TEST UPDATE METRIC CONFIGURATION
- #metric_config = {'is_default': False, 'operator': '==', 'target_value': ['TLS1.2', 'TLS1.3']}
- #print('-------------------------')
- #resp = get_metric_configuration_from_clouditor(service_id,metric_id)
- #print('Metric configuration before:\n{0}'.format(resp))
- #print('-------------------------')
- #resp = update_metric_configuration_to_clouditor(service_id, metric_id, metric_config)
- #print('Metric configuration after:\n{0}'.format(resp))
- #print('-------------------------')
-
- #TEST UPDATE METRIC IMPLEMENTATION
- #metric_id = 'EventMonitoringPolicyCheckQ1'
- #rego_code = create_metric_implementation(metric_id)
- #resp = update_metric_implementation_to_clouditor(metric_id, rego_code)
- #print(resp)
-
+ print("'Mapping applied to an object without obligations: ' %s\n" % e)
+ # return an error status to the editor
+ return "ERROR: Trying to map a REO with no obligations"
- #print('Current metric implementation')
- #resp = get_metric_implementation_from_clouditor(metric_id)
- #print(resp)
+ config_error_list = []
+ impl_error_list = []
+ for obligation in obl_list:
+ metric_id = (obligation['@metricId'])
+ #print('CREATING REGO RULE FOR METRIC {0}'.format(metric_id))
- #print('Metric implementation')
- #print(rego_code)
+ # create metric implementation (other fields)
+ metric, metric_implementation = create_metric_implementation(metric_id)
- #print('Updated metric implementation')
- #resp = update_metric_implementation_to_clouditor(metric_id,metric_implementation)
- #print(resp)
+ # create metric configuration (operator and target value)
+ metric_configuration = create_metric_configuration(obligation)
+
+ #print('Sending metric implementation to Orchestrator: {0}'.format(metric_id))
+ resp_implementation = update_metric_implementation_to_clouditor(metric_id, metric_implementation,
+ request=None)
+
+ if type(resp_implementation) is MetricImplementationResponse200:
+ #print('Successfully update metric implementation for: {0}'.format(metric_id))
+ resp_configuration = update_metric_configuration_to_clouditor(service_id, metric_id,
+ metric_configuration, request=None)
+ if type(resp_configuration) is not MetricConfigurationResponse200:
+ metric_configuration_error = True
+ print('ERROR: metric configuration: {0}'.format(metric_id))
+ config_error_list.append(metric_id)
+ save_data_on_error(metric, metric_configuration=metric_configuration,
+ configuration_details=resp_configuration)
+ else:
+ metric_implementation_error = True
+ print('ERROR: metric implementation: {0}'.format(metric_id))
+ impl_error_list.append(metric_id)
+ save_data_on_error(metric, metric_implementation=metric_implementation,
+ implementation_details=resp_implementation)
+
+ if metric_configuration_error and metric_implementation_error:
+ return 'WARNING: Unable to update some configurations ({0}) and implementations ({1}) to orchestrator'.format(config_error_list,impl_error_list)
+ elif metric_configuration_error and not metric_implementation_error:
+ return 'WARNING: Unable to update some configurations to orchestrator ({0})'.format(config_error_list)
+ elif metric_implementation_error and not metric_configuration_error:
+ return 'WARNING: Unable to update metric implementation to orchestrator ({0})'.format(impl_error_list)
+ else:
+ return 'Success'