Implement CA root retrieve from vault, some changes on testing and adding some comments to code (1c9e176c) · Commits · OCF / capif

services/TS29222_CAPIF_Security_API/capif_security/core/servicesecurity.py

+76 −6

Original line number	Diff line number	Diff line
		@@ -105,13 +105,67 @@ class SecurityOperations(Resource):
		current_app.logger.error("Not found security context")
		return not_found_error(detail=security_context_not_found_detail, cause=api_invoker_no_context_cause)

		if not authentication_info:
		for security_info_obj in services_security_object['security_info']:
		if security_info_obj.get('sel_security_method') == "PKI":
		current_app.logger.debug("PKI security method selected")
		if authentication_info:
		# Read the CA certificate from the file
		with open("/usr/src/app/capif_security/ca.crt", "rb") as key_file:
		key_data = key_file.read()
		# Decode the certificate to a string
		key_data = key_data.decode('utf-8')
		# Add the CA certificate to the authentication_info
		security_info_obj['authentication_info'] = key_data
		else:
		# If authentication_info is not needed, remove the key_data
		del security_info_obj['authentication_info']
		if not authorization_info:
		for security_info_obj in services_security_object['security_info']:

		if authorization_info:
		security_info_obj['authorization_info'] = security_info_obj.get('authorization_info', "")
		else:
		# If authorization_info is not needed, remove the key_data
		del security_info_obj['authorization_info']

		elif security_info_obj.get('sel_security_method') == "PSK":
		current_app.logger.debug("PSK security method selected")
		if authentication_info:
		# Read the PSK from the file
		with open("/usr/src/app/capif_security/ca.crt", "rb") as key_file:
		key_data = key_file.read()
		# Decode the PSK to a string
		key_data = key_data.decode('utf-8')
		# Add the PSK to the authentication_info
		security_info_obj['authentication_info'] = key_data
		else:
		# If authentication_info is not needed, remove the key_data
		del security_info_obj['authentication_info']

		if authorization_info:
		security_info_obj['authorization_info'] = security_info_obj.get('authorization_info', "UNDER DEVELOPMENT")
		else:
		# If authorization_info is not needed, remove the key_data
		del security_info_obj['authorization_info']

		elif security_info_obj.get('sel_security_method') == "OAUTH":
		current_app.logger.debug("OAUTH security method selected, this request is not needed")

		if authentication_info:
		security_info_obj['authentication_info'] = security_info_obj.get('authentication_info', "")
		else:
		# If authentication_info is not needed, remove the key_data
		del security_info_obj['authentication_info']

		if authorization_info:
		security_info_obj['authorization_info'] = security_info_obj.get('authorization_info', "")
		else:
		# If authorization_info is not needed, remove the key_data
		del security_info_obj['authorization_info']

		else:
		current_app.logger.error("Bad format security method")
		return bad_request_error(detail="Bad format security method", cause="Bad format security method", invalid_params=[{"param": "securityMethod", "reason": "Bad format security method"}])


		properyly_json = json.dumps(
		services_security_object, default=json_util.default)
		my_service_security = dict_to_camel_case(
		@@ -176,6 +230,22 @@ class SecurityOperations(Resource):
		return not_found_error(detail=f"Service with interfaceDescription {json.dumps(clean_empty(service_instance.interface_details.to_dict()))} not found", cause="Not found Service")

		# We obtain the interface security methods
		# We need to go deeper here, because the interface description is an array
		# and we need to find the correct one according to preferred security method by invoker,
		# maybe Published API contains more than one interface description, and each one is related
		# with a different security method, then we need to get a complete list (interface and related security methods)
		# amd then we need to check if the preferred security method is compatible with the interface description
		# also the security methods inside interface description is not mandatory, in that case we use aefProfile.securityMethods
		# an also that aefProfile.securityMethods is not mandatory, only in cases described on TS 29222 - 8.2.4.2.4 Type: AefProfile -
		#
		# NOTE4:
		# For AEFs defined by 3GPP interacting with API invokers via CAPIF-2e, at least one of the "securityMethods" attribute
		# within this data type or the "securityMethods" attribute within the "interfaceDescriptions" attribute shall be present.
		# For AEFs defined by 3GPP interacting with API invokers via CAPIF-2, the "securityMethods" attribute is optional.
		# For AEFs not defined by 3GPP, the "securityMethods" attribute is optional.
		#
		# To achieve this, we need to setup at config which domains or IPs are CAPIF-2e or CAPIF-2, and then we need to check if the domain or IP of the service is in the list.

		security_methods = aef_profile["aef_profiles"][0]["interface_descriptions"][0]["security_methods"]

		current_app.logger.debug("Interface security methods: " + str(security_methods))

services/TS29222_CAPIF_Security_API/prepare_security.sh

+47 −6

Original line number	Diff line number	Diff line
		@@ -3,6 +3,9 @@
		VAULT_ADDR="http://$VAULT_HOSTNAME:$VAULT_PORT"
		VAULT_TOKEN=$VAULT_ACCESS_TOKEN

		CERTS_FOLDER="/usr/src/app/capif_security"
		# cd $CERTS_FOLDER

		# Maximum number of retry attempts
		MAX_RETRIES=30
		# Delay between retries (in seconds)
		@@ -10,6 +13,40 @@ RETRY_DELAY=10
		# Attempt counter
		ATTEMPT=0

		while [ $ATTEMPT -lt $MAX_RETRIES ]; do
		# Increment ATTEMPT using eval
		eval "ATTEMPT=\$((ATTEMPT + 1))"
		echo "Attempt $ATTEMPT of $MAX_RETRIES"

		# Make the request to Vault and store the response in a variable
		RESPONSE=$(curl -s -k --connect-timeout 5 --max-time 10 \
		--header "X-Vault-Token: $VAULT_TOKEN" \
		--request GET "$VAULT_ADDR/v1/secret/data/ca" \| jq -r '.data.data.ca')

		echo "$RESPONSE"

		# Check if the response is "null" or empty
		if [ -n "$RESPONSE" ] && [ "$RESPONSE" != "null" ]; then
		echo "$RESPONSE" > $CERTS_FOLDER/ca.crt
		openssl verify -CAfile $CERTS_FOLDER/ca.crt $CERTS_FOLDER/ca.crt
		echo "CA Root successfully saved."
		SUCCES_OPERATION=true
		break
		else
		echo "Invalid response ('null' or empty), retrying in $RETRY_DELAY seconds..."
		sleep $RETRY_DELAY
		fi
		done

		if [ "$SUCCES_OPERATION" = false ]; then
		echo "Error: Failed to retrieve ca root a valid response after $MAX_RETRIES attempts."
		exit 1 # Exit with failure
		fi

		# Setup inital value to ATTEMPT and SUCCESS_OPERATION
		ATTEMPT=0
		SUCCES_OPERATION=false

		while [ $ATTEMPT -lt $MAX_RETRIES ]; do
		# Increment ATTEMPT using eval
		eval "ATTEMPT=\$((ATTEMPT + 1))"
		@@ -24,16 +61,20 @@ while [ $ATTEMPT -lt $MAX_RETRIES ]; do

		# Check if the response is "null" or empty
		if [ -n "$RESPONSE" ] && [ "$RESPONSE" != "null" ]; then
		echo "$RESPONSE" > /usr/src/app/capif_security/server.key
		echo "$RESPONSE" > $CERTS_FOLDER/server.key
		echo "Public key successfully saved."
		gunicorn -k uvicorn.workers.UvicornH11Worker --bind 0.0.0.0:8080 \
		--chdir /usr/src/app/capif_security wsgi:app
		exit 0 # Exit successfully
		SUCCES_OPERATION=true
		break
		else
		echo "Invalid response ('null' or empty), retrying in $RETRY_DELAY seconds..."
		sleep $RETRY_DELAY
		fi
		done

		echo "Error: Failed to retrieve a valid response after $MAX_RETRIES attempts."
		if [ "$SUCCES_OPERATION" = false ]; then
		echo "Error: Failed to retrieve server key valid response after $MAX_RETRIES attempts."
		exit 1 # Exit with failure
		fi

		gunicorn -k uvicorn.workers.UvicornH11Worker --bind 0.0.0.0:8080 \
		--chdir $CERTS_FOLDER wsgi:app

services/run.sh

+3 −3

Original line number	Diff line number	Diff line
		@@ -4,7 +4,7 @@ source $(dirname "$(readlink -f "$0")")/variables.sh
		help() {
		echo "Usage: $1 <options>"
		echo " -c : Setup different hostname for capif"
		echo " -s : Run Mock server"
		echo " -s : Run Mock server. Default true"
		echo " -m : Run monitoring service"
		echo " -l : Set Log Level (default DEBUG). Select one of: [CRITICAL, FATAL, ERROR, WARNING, WARN, INFO, DEBUG, NOTSET]"
		echo " -r : Remove cached information on build"
		@@ -35,7 +35,7 @@ then
		fi

		# Read params
		while getopts ":c:l:mshrv:f:g:b:" opt; do
		while getopts ":c:l:ms:hrv:f:g:b:" opt; do
		case $opt in
		c)
		CAPIF_HOSTNAME="$OPTARG"
		@@ -44,7 +44,7 @@ while getopts ":c:l:mshrv:f:g:b:" opt; do
		MONITORING_STATE=true
		;;
		s)
		ROBOT_MOCK_SERVER=true
		ROBOT_MOCK_SERVER="$OPTARG"
		;;
		v)
		OCF_VERSION="$OPTARG"

services/variables.sh

+1 −0

Original line number	Diff line number	Diff line
		@@ -32,6 +32,7 @@ export LOG_LEVEL=DEBUG
		export CACHED_INFO=""
		export BUILD_DOCKER_IMAGES=true
		export REMOVE_IMAGES=false
		export ROBOT_MOCK_SERVER=true

		# Needed to avoid write permissions on bind volumes with prometheus and grafana
		export DUID=$(id -u)

tests/libraries/security_api/bodyRequests.py

+1 −46

Original line number	Diff line number	Diff line
		@@ -7,7 +7,7 @@ def create_service_security_default_body(
		authentication_info=None,
		authorization_info=None,
		grant_type=None,
		pref_security_methods=["PSK", "PKI", "OAUTH"],
		pref_security_methods=["OAUTH", "PKI", "PSK"],
		sel_security_method=None,
		request_websocket_uri=None,
		websocket_uri=None):
		@@ -32,51 +32,6 @@ def create_service_security_default_body(
		return data


		def create_service_security_body(notification_destination,
		supported_features,
		security_info=None,
		aef_id=None,
		api_id=None,
		authentication_info=None,
		authorization_info=None):
		data = {
		"notificationDestination": notification_destination,
		"supportedFeatures": supported_features,
		"securityInfo": [{
		"authenticationInfo": "authenticationInfo",
		"authorizationInfo": "authorizationInfo",
		"interfaceDetails": {
		"ipv4Addr": "127.0.0.1",
		"securityMethods": ["PSK"],
		"port": 5248
		},
		"prefSecurityMethods": ["PSK", "PKI", "OAUTH"],
		}
		],
		"websockNotifConfig": {
		"requestWebsocketUri": True,
		"websocketUri": "websocketUri"
		},
		"requestTestNotification": True
		}

		if aef_id is not None and api_id is not None:
		security_info = dict()
		if authentication_info is not None:
		security_info['authenticationInfo'] = authentication_info
		if authorization_info is not None:
		security_info['authorizationInfo'] = authorization_info
		data['securityInfo'].append({
		"authenticationInfo": "authenticationInfo",
		"authorizationInfo": "authorizationInfo",
		"prefSecurityMethods": ["PSK", "PKI", "OAUTH"],
		"aefId": aef_id,
		"apiId": api_id
		})

		return data


		def create_security_info(
		aef_id=None,
		interface_details=None,