OCF176: initial framework and logic for api filtering (fe9999a6) · Commits · OCF / capif

services/TS29222_CAPIF_Discover_Service_API/service_apis/core/discoveredapis.py

+38 −0

Original line number	Diff line number	Diff line

		import json
		import requests

		from flask import current_app

		@@ -117,6 +118,43 @@ class DiscoverApisOperations(Resource):
		if len(json_docs) == 0:
		return not_found_error(detail="API Invoker " + api_invoker_id + " has no API Published that accomplish filter conditions", cause="No API Published accomplish filter conditions")

		# Apply visibility control filtering
		# try:
		# visibility_control_url = "http://helper:8080/visibility-control/v1/decision/invokers/{}/discoverable-apis".format(api_invoker_id)
		# visibility_payload = {
		# "serviceAPIDescriptions": json_docs
		# }

		# current_app.logger.debug("Calling visibility control for invoker: " + api_invoker_id)
		# visibility_response = requests.post(
		# visibility_control_url,
		# json=visibility_payload,
		# headers={"Content-Type": "application/json"},
		# timeout=10
		# )

		# if visibility_response.status_code == 200:
		# filtered_data = visibility_response.json()
		# json_docs = filtered_data.get("serviceAPIDescriptions", [])
		# current_app.logger.debug(f"Visibility control filtered {len(json_docs)} APIs for invoker {api_invoker_id}")
		# else:
		# current_app.logger.warning(f"Visibility control returned status {visibility_response.status_code}: {visibility_response.text}")
		# # Fallback: return no APIs if visibility control fails
		# json_docs = []

		# except requests.exceptions.RequestException as e:
		# current_app.logger.error(f"Failed to call visibility control: {str(e)}")
		# # Fallback: return all APIs if visibility control is unreachable (graceful degradation)
		# current_app.logger.warning(f"Visibility control unreachable for invoker {api_invoker_id}, returning all discovered APIs")
		# except Exception as e:
		# current_app.logger.error(f"Unexpected error in visibility control integration: {str(e)}")
		# # Fallback: return all APIs if there's an error (graceful degradation)
		# current_app.logger.warning(f"Error filtering APIs for invoker {api_invoker_id}, returning all discovered APIs")

		# # Check again after filtering
		# if len(json_docs) == 0:
		# return not_found_error(detail="API Invoker " + api_invoker_id + " has no visible APIs after applying visibility rules", cause="No APIs visible after visibility filtering")

		apis_discovered = DiscoveredAPIs(service_api_descriptions=json_docs)
		res = make_response(object=serialize_clean_camel_case(apis_discovered), status=200)
		return res

services/helper/helper_service/services/visibility_control/controllers/decision_controller.py

+22 −2

Original line number	Diff line number	Diff line
		@@ -4,18 +4,38 @@ from typing import Tuple
		from typing import Union

		from visibility_control.models.discovered_apis import DiscoveredAPIs # noqa: E501
		from visibility_control.models.discovery_request import DiscoveryRequest # noqa: E501
		from visibility_control.models.error import Error # noqa: E501
		from visibility_control import util
		from visibility_control.core import visibility_control_core


		def decision_invokers_api_invoker_id_discoverable_apis_get(api_invoker_id): # noqa: E501
		def decision_invokers_api_invoker_id_discoverable_apis_post(api_invoker_id, body=None): # noqa: E501
		"""Get discoverable APIs filter for an invoker (global scope)

		Returns a filtered list of APIs for the API Invoker. # noqa: E501

		:param api_invoker_id: CAPIF API Invoker identifier
		:type api_invoker_id: str
		:param body: List of all discovered APIs to be filtered
		:type body: DiscoveryRequest

		:rtype: Union[DiscoveredAPIs, Tuple[DiscoveredAPIs, int], Tuple[DiscoveredAPIs, int, Dict[str, str]]
		"""
		return 'do some magic!'
		# Handle body parameter - it may come as None from Connexion
		if body is None:
		if connexion.request.is_json:
		body = connexion.request.get_json()
		else:
		return {'code': 400, 'message': 'JSON body required'}, 400

		if body is None or body.get('serviceAPIDescriptions') is None:
		return {'code': 400, 'message': 'Missing serviceAPIDescriptions in request body'}, 400

		try:
		apis_list = body.get('serviceAPIDescriptions', [])
		# result = visibility_control_core.get_discoverable_apis(api_invoker_id, apis_list)
		result = apis_list # for testing
		return {"serviceAPIDescriptions": result}, 200
		except Exception as e:
		return {"error": str(e)}, 400

services/helper/helper_service/services/visibility_control/core/visibility_control_core.py

+167 −1

Original line number	Diff line number	Diff line
		@@ -268,3 +268,169 @@ def update_rule_patch(rule_id, body):
		# Return the fully updated object (excluding Mongo's internal _id)
		updated_rule = col.find_one({"ruleId": rule_id}, {"_id": 0})
		return updated_rule, 200


		def get_discoverable_apis(api_invoker_id, all_apis):
		"""
		Filter APIs based on visibility rules for a specific invoker.

		:param api_invoker_id: The ID of the API invoker
		:param all_apis: List of all discovered APIs (as dict objects)
		:return: List of APIs that the invoker is allowed to see
		"""
		db = get_mongo()
		rules_col = db.get_col_by_name("visibility_rules")

		# Get all active rules
		now = datetime.now(timezone.utc)
		active_rules = list(rules_col.find({
		"enabled": True,
		"$or": [
		{"startsAt": {"$lte": now}, "endsAt": {"$gte": now}},
		{"startsAt": {"$lte": now}, "endsAt": None},
		{"startsAt": None, "endsAt": {"$gte": now}},
		{"startsAt": None, "endsAt": None}
		]
		}, {"_id": 0}))

		if not active_rules:
		# No rules = default ALLOW (all APIs visible)
		return all_apis

		has_allow_rules = any(
		rule.get('default_access', rule.get('defaultAccess', 'ALLOW')) == 'ALLOW'
		for rule in active_rules
		)

		# Filter APIs based on rules
		discoverable_apis = []
		for api in all_apis:
		if _invoker_allowed_by_rule(
		api_invoker_id,
		api,
		active_rules,
		default_allow_no_match=not has_allow_rules
		):
		discoverable_apis.append(api)

		return discoverable_apis


		def _invoker_allowed_by_rule(api_invoker_id, api, rules, default_allow_no_match=True):
		"""
		Check if an invoker is allowed to see an API based on the rules.

		:param api_invoker_id: The invoker ID
		:param api: The API description (dict)
		:param rules: List of active rules
		:param default_allow_no_match: Whether to allow APIs that do not match any rule
		:return: True if allowed, False otherwise
		"""
		# Sort rules by specificity (most specific first)
		sorted_rules = sorted(rules, key=lambda r: _rule_specificity(r), reverse=True)

		for rule in sorted_rules:
		if _rule_matches_invoker(rule, api_invoker_id) and _rule_matches_api(rule, api):
		default_access = rule.get('default_access', rule.get('defaultAccess', 'ALLOW'))
		return default_access == 'ALLOW'

		return default_allow_no_match


		def _rule_matches_api(rule, api):
		"""
		Check if a rule matches an API based on provider selector.

		:param rule: The visibility rule
		:param api: The API description (dict)
		:return: True if the rule applies to this API
		"""
		provider_selector = rule.get('providerSelector', {})

		if not provider_selector:
		return True # No selector = matches all APIs

		# Check apiProviderId against apiProvName
		if 'apiProviderId' in provider_selector:
		api_provider_ids = provider_selector['apiProviderId']
		api_provider_id = api.get('apiProvName')
		if api_provider_id and api_provider_id not in api_provider_ids:
		return False

		# Check userName against apiProvName or provider username if present
		if 'userName' in provider_selector:
		user_names = provider_selector['userName']
		api_user_name = api.get('apiProvName')
		if api_user_name and api_user_name not in user_names:
		return False

		# Check apiName
		if 'apiName' in provider_selector:
		api_names = provider_selector['apiName']
		api_name = api.get('apiName')
		if api_name and api_name not in api_names:
		return False

		# Check apiId
		if 'apiId' in provider_selector:
		api_ids = provider_selector['apiId']
		api_id = api.get('apiId')
		if api_id and api_id not in api_ids:
		return False

		# Check aefId in nested profiles
		if 'aefId' in provider_selector:
		aef_ids = provider_selector['aefId']
		aef_profiles = api.get('aefProfiles', [])
		aef_match = False
		for profile in aef_profiles:
		if profile.get('aefId') in aef_ids:
		aef_match = True
		break
		if not aef_match:
		return False

		return True


		def _rule_matches_invoker(rule, api_invoker_id):
		"""
		Check if a rule matches an invoker based on invoker selector.

		:param rule: The visibility rule
		:param api_invoker_id: The invoker ID
		:return: True if the rule applies to this invoker
		"""
		invoker_selector = rule.get('invokerSelector', {})

		if not invoker_selector:
		return True # No selector = matches all invokers

		# Check invokerId
		if 'invokerId' in invoker_selector:
		invoker_ids = invoker_selector['invokerId']
		if api_invoker_id not in invoker_ids:
		return False

		# Add other selectors as needed (e.g., invokerName, etc.)

		return True


		def _rule_specificity(rule):
		"""
		Calculate rule specificity for ordering.
		Higher specificity = more specific selectors = higher priority.

		:param rule: The visibility rule
		:return: Specificity score
		"""
		specificity = 0
		provider_selector = rule.get('providerSelector', {})

		# Each selector adds to specificity
		for selector in ['apiProviderId', 'userName', 'apiName', 'aefId', 'apiId']:
		if selector in provider_selector:
		specificity += len(provider_selector[selector])

		return specificity
		No newline at end of file