Updated tests

# Keys and certificates generated by the samples and testing clients

**/test_*_certificate_folder/

network_app_samples/network_app_provider_sample/aef_server_certificate/

ci_cd_test/testing_samples/aef_server_certificate/

ci_cd_test/testing_samples/testing_provider_sample/aef_servers.ready

#!/bin/sh

# Runs inside the SDK test container (see Dockerfile and test_sdk.sh):

#   1. Provider: onboards and publishes the test API.

#   2. Invoker: onboards, discovers the API, gets the security context and offboards.

#   3. Provider: offboards. It always runs, even if a previous step failed, to leave OpenCAPIF clean.

#   1. Provider: onboards and publishes the test API, exposed by three AEFs: one per security method

#      (OAUTH, PKI and PSK).

#   2. Provider: starts the three AEF servers in the background.

#   3. Invoker: onboards, discovers the API, gets the security context and an access token, invokes the

#      API at each AEF with its security method, and offboards.

#   4. Provider: stops the AEF servers and offboards. It always runs, even if a previous step failed,

#      to leave OpenCAPIF clean.

# Exits with 1 if any step fails, so the failure reaches test_sdk.sh.

#

# SDK_TEST_SECURITY_METHODS limits the security methods checked by the invoker (default: OAUTH,PKI,PSK).

# PSK requires OpenCAPIF 5.1.0 or later.

set -u

SAMPLES_DIR="$(cd "$(dirname "$0")" && pwd)/testing_samples"

AEF_SERVERS_READY="$SAMPLES_DIR/testing_provider_sample/aef_servers.ready"

AEF_SERVERS_TIMEOUT=60

AEF_SERVERS_PID=""

status=0

# Runs a step and records whether it failed: step "<name>" <command> [args...]

  (cd "$SAMPLES_DIR/testing_invoker_sample" && python network_app_invoker.py)

}

# The invoker test needs the API published by the provider

# Starts the AEF servers of the provider in the background and waits until they are listening

start_aef_servers() {

  rm -f "$AEF_SERVERS_READY"

  # Started directly (not through provider()), so that $! is the Python process and stop_aef_servers stops it.

  # Unbuffered output, so the server logs appear in order with the invoker ones.

  (cd "$SAMPLES_DIR/testing_provider_sample" && exec python -u testing_provider.py serve) &

  AEF_SERVERS_PID=$!

  waited=0

  while [ ! -f "$AEF_SERVERS_READY" ]; do

    if ! kill -0 "$AEF_SERVERS_PID" 2>/dev/null; then

      echo "The AEF servers exited before being ready"

      AEF_SERVERS_PID=""

      return 1

    fi

    if [ "$waited" -ge "$AEF_SERVERS_TIMEOUT" ]; then

      echo "The AEF servers were not ready after ${AEF_SERVERS_TIMEOUT}s"

      return 1

    fi

    sleep 1

    waited=$((waited + 1))

  done

}

stop_aef_servers() {

  if [ -n "$AEF_SERVERS_PID" ]; then

    kill "$AEF_SERVERS_PID" 2>/dev/null

    wait "$AEF_SERVERS_PID" 2>/dev/null

    AEF_SERVERS_PID=""

  fi

}

# The invoker test needs the API published by the provider and its AEF servers running

if step "provider onboarding and API publishing" provider onboard publish; then

  if step "provider AEF servers start" start_aef_servers; then

    step "invoker flow" invoker || true

  fi

  stop_aef_servers

fi

step "provider offboarding" provider offboard || true

echo "🚀 Running container: $CONTAINER_NAME in network: $NETWORK_NAME..."

rc=0

# SDK_TEST_SECURITY_METHODS (e.g. OAUTH,PKI) limits the security methods checked by the invoker test

docker run \

  --rm \

  --name "$CONTAINER_NAME" \

  --network "$NETWORK_NAME" \

  ${SDK_TEST_SECURITY_METHODS:+-e SDK_TEST_SECURITY_METHODS="$SDK_TEST_SECURITY_METHODS"} \

  "$IMAGE_NAME" || rc=$?

if [ "$rc" -ne 0 ]; then

    },

    "security_context": {

      "pref_security_context": ["OAUTH","PKI","PSK"]

    },

    "api_invocation": {

      "ca_cert": "../aef_server_certificate/aef_server.crt"

    }

  }

}

import json

import logging

import os

import sys

import traceback

import requests

from opencapif_sdk import capif_invoker_connector, service_discoverer

# Path to the OpenCAPIF SDK configuration file

capif_sdk_config_path = "./capif_sdk_config_sample.json"

# API published by the provider test, exposed by one AEF per security method (OAUTH, PKI and PSK)

API_NAME = "6G-resilience"

# Security methods to check (all by default). PSK requires OpenCAPIF 5.1.0 or later.

SECURITY_METHODS = [

    method.strip().upper()

    for method in os.getenv("SDK_TEST_SECURITY_METHODS", "OAUTH,PKI,PSK").split(",") if method.strip()

]

def invoke_with_each_security_method(discoverer, security_response):

    """Invokes the API at the AEF of each security method and checks that the AEF authenticated the invoker."""

    invoker_id = discoverer.invoker_capif_details["api_invoker_id"]

    selected = {

        security_info["aefId"]: security_info["selSecurityMethod"]

        for security_info in security_response["securityInfo"]

    }

    for method in SECURITY_METHODS:

        aef_ids = [aef_id for aef_id, selected_method in selected.items() if selected_method == method]

        if not aef_ids:

            raise RuntimeError(f"CAPIF did not select {method} for any AEF of the API: {selected}")

        for aef_id in aef_ids:

            response = discoverer.invoke_api(API_NAME, "GET", resource="RESILIENCE_MANAGEMENT", aef_id=aef_id)

            answer = response.json()

            expected = {"aefId": aef_id, "securityMethod": method, "invoker": invoker_id}

            received = {key: answer.get(key) for key in expected}

            if received != expected:

                raise RuntimeError(f"Unexpected answer of AEF {aef_id} with {method}: {answer} (expected {expected})")

            print(f"✅ API INVOKED WITH {method}: {response.request.method} {response.url} -> {answer}")

            check_rejected_without_credentials(response.url, method, discoverer.aef_ca_cert)

def check_rejected_without_credentials(url, method, ca_cert):

    """The AEF must reject a request without the credentials of its security method."""

    try:

        response = requests.get(url, verify=ca_cert, timeout=10)

        rejected = response.status_code == 401  # OAUTH: no access token

    except requests.ConnectionError:

        rejected = True  # PKI: no client certificate; PSK: no pre-shared key (TLS handshake fails)

    if not rejected:

        raise RuntimeError(f"The {method} AEF accepted a request without credentials: {url}")

    print(f"✅ The {method} AEF rejects a request without credentials")

def run_invoker_flow():

    # Discover APIs and obtain the security context

    print("✅ ACCESS TOKEN:")

    print(discoverer.token)

    invoke_with_each_security_method(discoverer, security_response)

    print("🎉 Script finished successfully.")