openstack/tacker
Code Issues Proposed changes

Add certificateRef check in OAuth 2.0 mTLS

Browse Source 
This patch adds a check "ParamsOauth2ClientCert.certificateRef"
when using OAuth 2.0 Mutual-TLS for authentication.
Also refactor the functions related to
SubscriptionAuthentication in common_script_utils.

Implements: blueprint support-oauth2-mtls
Implements: blueprint enhance-http-client
Change-Id: I9bcf4957ac8b173695e416fd56a0c3f3065e0c9f
This commit is contained in:
Ken Fujimoto
2023-06-15 00:48:41 +00:00
parent 9c7918f0b5
commit 2455d3f720
19 changed files with 592 additions and 503 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
tacker/sol_refactored/common/common_script_utils.py
View File

@@ -13,32 +13,29 @@
# License for the specific language governing permissions and limitations
# under the License.
import base64
import threading
from cryptography.hazmat.primitives import hashes
from cryptography import x509
from oslo_log import log as logging
from tacker.sol_refactored.api import api_version
from tacker.sol_refactored.api.schemas import common_types
from tacker.sol_refactored.api import validator
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import http_client
from tacker.sol_refactored.common import subscription_utils
from tacker.sol_refactored.common import vnfd_utils
from tacker.sol_refactored import objects
LOG = logging.getLogger(__name__)
# NOTE: The methods defined in this file are intended to be used from
# scripts (ex. mgmt_driver, UserData class method, coordinate script)
# which executed as a separate process by tacker process (i.e.
# conductor).
# Note that 'dict' is used instead of 'objects' since 'objects' is not
# used by scripts.
# Some methods intend to be used with tacker process commonly. Note
# that objects are dict compat.
CONF = config.CONF
TEST_NOTIFICATION_TIMEOUT = 20 # seconds
NOTIFY_TYPE_PM = 'PM'
NOTIFY_TYPE_FM = 'FM'
AUTH_TYPE_OAUTH2_CLIENT_CERT = 'OAUTH2_CLIENT_CERT'
AUTH_TYPE_OAUTH2_CLIENT_CREDENTIALS = 'OAUTH2_CLIENT_CREDENTIALS'
AUTH_TYPE_BASIC = 'BASIC'
def get_vnfd(vnfd_id, csar_dir):
vnfd = vnfd_utils.Vnfd(vnfd_id)
@@ -299,105 +296,87 @@ def apply_ext_managed_vls_from_inst(hot_dict, inst):
_apply_ext_managed_vls(hot_dict, mgd_vls)
def get_notification_auth_handle(obj_data):
auth_req = obj_data.get('authentication', None)
if auth_req:
auth = objects.SubscriptionAuthentication(
authType=auth_req['authType']
)
if AUTH_TYPE_OAUTH2_CLIENT_CERT in auth.authType:
param = obj_data.authentication.paramsOauth2ClientCert
ca_cert = CONF.v2_vnfm.notification_mtls_ca_cert_file
client_cert = CONF.v2_vnfm.notification_mtls_client_cert_file
return http_client.OAuth2MtlsAuthHandle(
None, param.tokenEndpoint,
param.clientId, ca_cert, client_cert)
elif AUTH_TYPE_OAUTH2_CLIENT_CREDENTIALS in auth.authType:
param = obj_data.authentication.paramsOauth2ClientCredentials
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.OAuth2AuthHandle(
None, param.tokenEndpoint, param.clientId,
param.clientPassword,
verify=verify)
elif AUTH_TYPE_BASIC in auth.authType:
param = obj_data.authentication.paramsBasic
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.BasicAuthHandle(
param.userName, param.password,
verify=verify)
def check_subsc_auth(auth_req, validation=True):
if validation:
auth_validator = validator.SolSchemaValidator(
common_types.SubscriptionAuthentication)
auth_validator.validate(auth_req)
auth_type = auth_req['authType']
if 'OAUTH2_CLIENT_CERT' in auth_type:
oauth2_mtls_req = auth_req.get('paramsOauth2ClientCert')
if oauth2_mtls_req is None:
msg = "paramsOauth2ClientCert must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
client_cert_file = CONF.v2_vnfm.notification_mtls_client_cert_file
cert_ref = oauth2_mtls_req.get('certificateRef', {})
if cert_ref.get('type') == 'x5t#S256':
hash_type = hashes.SHA256()
else:
raise sol_ex.AuthTypeNotFound(auth.authType)
else:
# support type is only "x5t#S256"(SHA-256)
msg = "certificateRef type is invalid."
raise sol_ex.InvalidSubscription(sol_detail=msg)
with open(client_cert_file, "rb") as f:
client_cert = x509.load_pem_x509_certificate(f.read())
cert_fingerprint = client_cert.fingerprint(hash_type)
fingerprint_value = (base64.urlsafe_b64encode(cert_fingerprint).
rstrip(b'=').decode('utf-8'))
if fingerprint_value != cert_ref.get('value'):
msg = "certificateRef value is incorrect."
raise sol_ex.InvalidSubscription(sol_detail=msg)
if 'OAUTH2_CLIENT_CREDENTIALS' in auth_type:
oauth2_req = auth_req.get('paramsOauth2ClientCredentials')
if oauth2_req is None:
msg = "paramsOauth2ClientCredentials must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
if 'BASIC' in auth_type:
basic_req = auth_req.get('paramsBasic')
if basic_req is None:
msg = "paramsBasic must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
def get_http_auth_handle(auth_req):
# NOTE: this method uses tacker configuration. script should set
# the following parameters before calling this method.
# ---
# from tacker.common import config
# args = ["--config-file", "/etc/tacker/tacker.conf"]
# config.init(args)
# ---
if auth_req is None:
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.NoAuthHandle(verify=verify)
def async_call(func):
def inner(*args, **kwargs):
th = threading.Thread(target=func, args=args,
kwargs=kwargs, daemon=True)
th.start()
return inner
@async_call
def send_notification(obj_data, notif_data, notify_type=None):
version = api_version.CURRENT_VERSION
auth_handle = subscription_utils.get_notification_auth_handle(obj_data)
connect_retries = (CONF.v2_vnfm.notify_connect_retries
if CONF.v2_vnfm.notify_connect_retries else None)
client = http_client.HttpClient(auth_handle,
version=version,
connect_retries=connect_retries)
if notify_type == NOTIFY_TYPE_PM:
version = api_version.CURRENT_PM_VERSION
auth_handle = get_notification_auth_handle(obj_data)
client = http_client.HttpClient(auth_handle,
version=version)
if notify_type == NOTIFY_TYPE_FM:
version = api_version.CURRENT_FM_VERSION
auth_handle = get_notification_auth_handle(obj_data)
client = http_client.HttpClient(auth_handle,
version=version)
url = obj_data.callbackUri
try:
resp, _ = client.do_request(
url, "POST", expected_status=[204], body=notif_data)
except sol_ex.SolException as ex:
# it may occur if test_notification was not executed.
LOG.exception(f"send_notification failed: {ex}")
if resp.status_code != 204:
LOG.error(f"send_notification failed: {resp.__dict__}")
def test_notification(obj_data, notify_type=None):
version = api_version.CURRENT_VERSION
auth_handle = subscription_utils.get_notification_auth_handle(obj_data)
if notify_type == NOTIFY_TYPE_PM:
version = api_version.CURRENT_PM_VERSION
auth_handle = get_notification_auth_handle(obj_data)
if notify_type == NOTIFY_TYPE_FM:
version = api_version.CURRENT_FM_VERSION
auth_handle = get_notification_auth_handle(obj_data)
client = http_client.HttpClient(auth_handle,
version=version,
timeout=TEST_NOTIFICATION_TIMEOUT)
url = obj_data.callbackUri
try:
resp, _ = client.do_request(url, "GET", expected_status=[204])
except sol_ex.SolException as e:
# any notify_type of error is considered. avoid 500 error.
raise sol_ex.TestNotificationFailed() from e
if resp.status_code != 204:
raise sol_ex.TestNotificationFailed()
# NOTE: auth_req is already validated.
auth_type = auth_req['authType']
# NOTE: if there are multiple auth_types, the following priority
# applied.
if 'OAUTH2_CLIENT_CERT' in auth_type:
oauth2_mtls_req = auth_req['paramsOauth2ClientCert']
ca_cert = CONF.v2_vnfm.notification_mtls_ca_cert_file
client_cert = CONF.v2_vnfm.notification_mtls_client_cert_file
return http_client.OAuth2MtlsAuthHandle(
None, oauth2_mtls_req['tokenEndpoint'],
oauth2_mtls_req['clientId'], ca_cert, client_cert)
elif 'OAUTH2_CLIENT_CREDENTIALS' in auth_type:
oauth2_req = auth_req.get('paramsOauth2ClientCredentials')
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.OAuth2AuthHandle(
None, oauth2_req.get('tokenEndpoint'),
oauth2_req.get('clientId'), oauth2_req.get('clientPassword'),
verify=verify)
elif 'BASIC' in auth_type:
basic_req = auth_req.get('paramsBasic')
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.BasicAuthHandle(
basic_req.get('userName'), basic_req.get('password'),
verify=verify)

4
tacker/sol_refactored/common/exceptions.py
View File

@@ -379,10 +379,6 @@ class OIDCAuthFailed(SolHttpError400):
" Detail: %(detail)s")
class AuthTypeNotFound(SolHttpError400):
message = _("AuthType parameter %(auth_type)s is not found")
class HelmOperationFailed(SolHttpError422):
title = 'Helm operation failed'
# detail set in the code

110
tacker/sol_refactored/common/subscription_utils.py
View File

@@ -13,11 +13,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import threading
from oslo_log import log as logging
from oslo_utils import timeutils
from oslo_utils import uuidutils
from tacker.sol_refactored.api import api_version
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import exceptions as sol_ex
@@ -30,6 +32,15 @@ LOG = logging.getLogger(__name__)
CONF = config.CONF
TEST_NOTIFICATION_TIMEOUT = 20 # seconds
NOTIFY_TYPE_PM = 'PM'
NOTIFY_TYPE_FM = 'FM'
# NOTE: The methods of first half are for LCM subscription only
# since this file was for LCM subscription originally.
# The methods of later half are common for LCM, PM and FM
# subscription.
def get_subsc(context, subsc_id):
subsc = objects.LccnSubscriptionV2.get_by_id(context, subsc_id)
@@ -46,16 +57,6 @@ def subsc_href(subsc_id, endpoint):
return "{}/vnflcm/v2/subscriptions/{}".format(endpoint, subsc_id)
def get_notification_auth_handle(subsc):
auth_req = subsc.get('authentication', None)
if auth_req:
return common_script_utils.get_notification_auth_handle(subsc)
else:
return http_client.NoAuthHandle()
# not reach here
def match_version(version, inst):
# - vnfSoftwareVersion 1
# - vnfdVersions 0..N
@@ -220,27 +221,31 @@ def make_delete_inst_notif_data(subsc, inst, endpoint):
return notif_data
def check_http_client_auth(auth_req):
# common methods for LCM, PM and FM subscription
def get_subsc_auth(auth_req):
# NOTE: assume auth_req already validated.
common_script_utils.check_subsc_auth(auth_req, validation=False)
auth = objects.SubscriptionAuthentication(
authType=auth_req['authType']
)
if 'OAUTH2_CLIENT_CERT' in auth.authType:
oauth2_mtls_req = auth_req.get('paramsOauth2ClientCert')
if oauth2_mtls_req is None:
msg = "paramsOauth2ClientCert must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
oauth2_mtls_req = auth_req['paramsOauth2ClientCert']
auth.paramsOauth2ClientCert = (
objects.SubscriptionAuthentication_ParamsOauth2ClientCert(
clientId=oauth2_mtls_req.get('clientId'),
certificateRef=oauth2_mtls_req.get('certificateRef'),
certificateRef=objects.ParamsOauth2ClientCert_CertificateRef(
type=oauth2_mtls_req['certificateRef']['type'],
value=oauth2_mtls_req['certificateRef']['value']
),
tokenEndpoint=oauth2_mtls_req.get('tokenEndpoint')
)
)
elif 'OAUTH2_CLIENT_CREDENTIALS' in auth.authType:
oauth2_req = auth_req.get('paramsOauth2ClientCredentials')
if oauth2_req is None:
msg = "paramsOauth2ClientCredentials must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
if 'OAUTH2_CLIENT_CREDENTIALS' in auth.authType:
oauth2_req = auth_req['paramsOauth2ClientCredentials']
auth.paramsOauth2ClientCredentials = (
objects.SubscriptionAuthentication_ParamsOauth2(
clientId=oauth2_req.get('clientId'),
@@ -248,17 +253,66 @@ def check_http_client_auth(auth_req):
tokenEndpoint=oauth2_req.get('tokenEndpoint')
)
)
elif 'BASIC' in auth.authType:
basic_req = auth_req.get('paramsBasic')
if basic_req is None:
msg = "ParamsBasic must be specified."
raise sol_ex.InvalidSubscription(sol_detail=msg)
if 'BASIC' in auth.authType:
basic_req = auth_req['paramsBasic']
auth.paramsBasic = (
objects.SubscriptionAuthentication_ParamsBasic(
userName=basic_req.get('userName'),
password=basic_req.get('password')
)
)
else:
raise sol_ex.AuthTypeNotFound(auth.authType)
return auth
def async_call(func):
def inner(*args, **kwargs):
th = threading.Thread(target=func, args=args,
kwargs=kwargs, daemon=True)
th.start()
return inner
@async_call
def send_notification(obj_data, notif_data, notify_type=None):
version = api_version.CURRENT_VERSION
if notify_type == NOTIFY_TYPE_PM:
version = api_version.CURRENT_PM_VERSION
elif notify_type == NOTIFY_TYPE_FM:
version = api_version.CURRENT_FM_VERSION
auth_handle = common_script_utils.get_http_auth_handle(
obj_data.get('authentication'))
connect_retries = (CONF.v2_vnfm.notify_connect_retries
if CONF.v2_vnfm.notify_connect_retries else None)
client = http_client.HttpClient(auth_handle,
version=version,
connect_retries=connect_retries)
url = obj_data.callbackUri
try:
resp, _ = client.do_request(
url, "POST", expected_status=[204], body=notif_data)
except sol_ex.SolException as ex:
# it may occur if test_notification was not executed.
LOG.exception(f"send_notification failed: {ex}")
def test_notification(obj_data, notify_type=None):
version = api_version.CURRENT_VERSION
if notify_type == NOTIFY_TYPE_PM:
version = api_version.CURRENT_PM_VERSION
elif notify_type == NOTIFY_TYPE_FM:
version = api_version.CURRENT_FM_VERSION
auth_handle = common_script_utils.get_http_auth_handle(
obj_data.get('authentication'))
client = http_client.HttpClient(auth_handle,
version=version,
timeout=TEST_NOTIFICATION_TIMEOUT)
url = obj_data.callbackUri
try:
resp, _ = client.do_request(url, "GET", expected_status=[204])
except sol_ex.SolException as e:
# any notify_type of error is considered. avoid 500 error.
raise sol_ex.TestNotificationFailed() from e

6
tacker/sol_refactored/conductor/vnfpm_driver_v2.py
View File

@@ -15,11 +15,11 @@
import datetime
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import pm_job_utils
from tacker.sol_refactored.common import pm_threshold_utils
from tacker.sol_refactored.common import subscription_utils as subsc_utils
from tacker.sol_refactored.nfvo import nfvo_client
from tacker.sol_refactored import objects
@@ -71,8 +71,8 @@ class VnfPmDriverV2():
notif_data = pm_threshold_utils.make_threshold_notif_data(
datetime_now, threshold_state,
self.endpoint, threshold)
common_script_utils.send_notification(
threshold, notif_data, common_script_utils.NOTIFY_TYPE_PM)
subsc_utils.send_notification(
threshold, notif_data, subsc_utils.NOTIFY_TYPE_PM)
def _store_report(self, context, report):
report = objects.PerformanceReportV2.from_dict(report)

7
tacker/sol_refactored/controller/vnffm_v1.py
View File

@@ -20,7 +20,6 @@ from tacker.sol_refactored.api import api_version
from tacker.sol_refactored.api.schemas import vnffm_v1 as schema
from tacker.sol_refactored.api import validator
from tacker.sol_refactored.api import wsgi as sol_wsgi
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import coordinate
from tacker.sol_refactored.common import exceptions as sol_ex
@@ -120,11 +119,11 @@ class VnfFmControllerV1(sol_wsgi.SolAPIController):
auth_req = body.get('authentication')
if auth_req:
subsc.authentication = subsc_utils.check_http_client_auth(auth_req)
subsc.authentication = subsc_utils.get_subsc_auth(auth_req)
if CONF.v2_nfvo.test_callback_uri:
common_script_utils.test_notification(
subsc, common_script_utils.NOTIFY_TYPE_FM)
subsc_utils.test_notification(
subsc, subsc_utils.NOTIFY_TYPE_FM)
subsc.create(context)

5
tacker/sol_refactored/controller/vnflcm_v2.py
View File

@@ -22,7 +22,6 @@ from tacker.sol_refactored.api.policies.vnflcm_v2 import POLICY_NAME
from tacker.sol_refactored.api.schemas import vnflcm_v2 as schema
from tacker.sol_refactored.api import validator
from tacker.sol_refactored.api import wsgi as sol_wsgi
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import coordinate
from tacker.sol_refactored.common import exceptions as sol_ex
@@ -476,10 +475,10 @@ class VnfLcmControllerV2(sol_wsgi.SolAPIController):
auth_req = body.get('authentication')
if auth_req:
subsc.authentication = subsc_utils.check_http_client_auth(auth_req)
subsc.authentication = subsc_utils.get_subsc_auth(auth_req)
if CONF.v2_nfvo.test_callback_uri:
common_script_utils.test_notification(subsc)
subsc_utils.test_notification(subsc)
subsc.create(context)

25
tacker/sol_refactored/controller/vnfpm_v2.py
View File

@@ -23,7 +23,6 @@ from tacker.sol_refactored.api import api_version
from tacker.sol_refactored.api.schemas import vnfpm_v2 as schema
from tacker.sol_refactored.api import validator
from tacker.sol_refactored.api import wsgi as sol_wsgi
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import coordinate
from tacker.sol_refactored.common import exceptions as sol_ex
@@ -173,7 +172,7 @@ class VnfPmControllerV2(sol_wsgi.SolAPIController):
# authentication
auth_req = body.get('authentication')
if auth_req:
pm_job.authentication = subsc_utils.check_http_client_auth(
pm_job.authentication = subsc_utils.get_subsc_auth(
auth_req)
# metadata
@@ -182,8 +181,8 @@ class VnfPmControllerV2(sol_wsgi.SolAPIController):
pm_job.metadata = metadata
if CONF.v2_nfvo.test_callback_uri:
common_script_utils.test_notification(
pm_job, common_script_utils.NOTIFY_TYPE_PM)
subsc_utils.test_notification(
pm_job, subsc_utils.NOTIFY_TYPE_PM)
try:
self.plugin.create_job(context=context, pm_job=pm_job)
@@ -235,12 +234,12 @@ class VnfPmControllerV2(sol_wsgi.SolAPIController):
if body.get("callbackUri"):
pm_job.callbackUri = body.get("callbackUri")
if body.get("authentication"):
pm_job.authentication = subsc_utils.check_http_client_auth(
pm_job.authentication = subsc_utils.get_subsc_auth(
body.get("authentication"))
if CONF.v2_nfvo.test_callback_uri:
common_script_utils.test_notification(
pm_job, common_script_utils.NOTIFY_TYPE_PM)
subsc_utils.test_notification(
pm_job, subsc_utils.NOTIFY_TYPE_PM)
with context.session.begin(subtransactions=True):
pm_job.update(context)
@@ -348,12 +347,12 @@ class VnfPmControllerV2(sol_wsgi.SolAPIController):
auth_req = body.get('authentication')
if auth_req:
threshold.authentication = subsc_utils.check_http_client_auth(
threshold.authentication = subsc_utils.get_subsc_auth(
auth_req)
if CONF.v2_nfvo.test_callback_uri:
common_script_utils.test_notification(
threshold, common_script_utils.NOTIFY_TYPE_PM)
subsc_utils.test_notification(
threshold, subsc_utils.NOTIFY_TYPE_PM)
try:
self.threshold_plugin.create_threshold(
@@ -410,10 +409,10 @@ class VnfPmControllerV2(sol_wsgi.SolAPIController):
if body.get("callbackUri"):
pm_threshold.callbackUri = body.get("callbackUri")
if CONF.v2_nfvo.test_callback_uri:
common_script_utils.test_notification(
pm_threshold, common_script_utils.NOTIFY_TYPE_PM)
subsc_utils.test_notification(
pm_threshold, subsc_utils.NOTIFY_TYPE_PM)
if body.get("authentication"):
pm_threshold.authentication = subsc_utils.check_http_client_auth(
pm_threshold.authentication = subsc_utils.get_subsc_auth(
body.get("authentication"))
with context.session.begin(subtransactions=True):

15
tacker/sol_refactored/nfvo/nfvo_client.py
View File

@@ -21,7 +21,6 @@ import zipfile
from oslo_log import log as logging
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import fm_alarm_utils as alarm_utils
from tacker.sol_refactored.common import fm_subscription_utils as fm_utils
@@ -186,7 +185,7 @@ class NfvoClient(object):
for subsc in subscs:
notif_data = subsc_utils.make_create_inst_notif_data(
subsc, inst, endpoint)
common_script_utils.send_notification(subsc, notif_data)
subsc_utils.send_notification(subsc, notif_data)
if self.is_local:
self.nfvo.recv_inst_create_notification(context, inst)
@@ -196,7 +195,7 @@ class NfvoClient(object):
for subsc in subscs:
notif_data = subsc_utils.make_delete_inst_notif_data(
subsc, inst, endpoint)
common_script_utils.send_notification(subsc, notif_data)
subsc_utils.send_notification(subsc, notif_data)
if self.is_local:
self.nfvo.recv_inst_delete_notification(context, inst)
@@ -209,7 +208,7 @@ class NfvoClient(object):
for subsc in subscs:
notif_data = lcmocc_utils.make_lcmocc_notif_data(
subsc, lcmocc, endpoint)
common_script_utils.send_notification(subsc, notif_data)
subsc_utils.send_notification(subsc, notif_data)
if self.is_local:
self.nfvo.recv_lcmocc_notification(context, lcmocc, inst)
@@ -219,8 +218,8 @@ class NfvoClient(object):
for subsc in subscs:
notif_data = alarm_utils.make_alarm_notif_data(
subsc, alarm, endpoint)
common_script_utils.send_notification(
subsc, notif_data, common_script_utils.NOTIFY_TYPE_FM)
subsc_utils.send_notification(
subsc, notif_data, subsc_utils.NOTIFY_TYPE_FM)
def send_pm_job_notification(self, report, pm_job, timestamp, endpoint):
report_object_instance_id = {entry.objectInstanceId
@@ -234,5 +233,5 @@ class NfvoClient(object):
notif_data = pm_job_utils.make_pm_notif_data(
instance_id, sub_instance_ids, report.id,
pm_job, timestamp, endpoint)
common_script_utils.send_notification(
pm_job, notif_data, common_script_utils.NOTIFY_TYPE_PM)
subsc_utils.send_notification(
pm_job, notif_data, subsc_utils.NOTIFY_TYPE_PM)

7
tacker/sol_refactored/objects/base.py
View File

@@ -460,4 +460,9 @@ class TackerPersistentObject(TackerObject):
return obj
TackerObjectDictCompat = ovoo_base.VersionedObjectDictCompat
class TackerObjectDictCompat(ovoo_base.VersionedObjectDictCompat):
def get(self, key, value=None):
# NOTE: VersionedObjectDictCompat requires default value if
# obj_attr is not set. This makes get return None if obj_attr is
# not set and default value is not specified.
return super().get(key, value)

4
tacker/sol_refactored/objects/common/subscription_authentication.py
View File

@@ -18,7 +18,7 @@ from tacker.sol_refactored.objects import fields
# NFV-SOL 013
# - v3.4.1 Table 8.3.4-1
# - v3.5.1 Table 8.3.4-1
@base.TackerObjectRegistry.register
class SubscriptionAuthentication(base.TackerObject,
base.TackerObjectDictCompat):
@@ -85,7 +85,7 @@ class SubscriptionAuthentication_ParamsOauth2ClientCert(
fields = {
'clientId': fields.StringField(nullable=False),
'cerficateRef': fields.ObjectField(
'certificateRef': fields.ObjectField(
'ParamsOauth2ClientCert_CertificateRef', nullable=False),
'tokenEndpoint': fields.UriField(nullable=False),
}

370
tacker/tests/unit/sol_refactored/common/test_common_script_utils.py
View File

@@ -15,24 +15,19 @@
import os
import requests
from unittest import mock
from oslo_config import cfg
from oslo_utils import uuidutils
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import fm_alarm_utils
from tacker.sol_refactored.common import http_client
from tacker.sol_refactored.common import pm_job_utils
from tacker.sol_refactored import objects
from tacker.tests import base
from tacker.tests.unit.sol_refactored.samples import fakes_for_fm
SAMPLE_VNFD_ID = "b1bb0ce7-ebca-4fa7-95ed-4840d7000000"
SAMPLE_FLAVOUR_ID = "simple"
CONF = config.CONF
class TestCommontScriptUtils(base.BaseTestCase):
@@ -579,275 +574,128 @@ class TestCommontScriptUtils(base.BaseTestCase):
self.assertNotIn(vl, top_hot['resources'])
self.assertNotIn(vl_subnet, top_hot['resources'])
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification(self, mock_resp):
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-1', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback')
notif_data_no_auth = objects.VnfLcmOperationOccurrenceNotificationV2(
id=uuidutils.generate_uuid()
)
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
def test_check_subsc_auth(self):
# Check OAUTH2_CLIENT_CERT
auth_req_1 = {
'authType': ['OAUTH2_CLIENT_CERT']
}
ex = self.assertRaises(sol_ex.InvalidSubscription,
common_script_utils.check_subsc_auth,
auth_req_1)
self.assertEqual("paramsOauth2ClientCert must be specified.",
ex.detail)
# execute no_auth
common_script_utils.send_notification(
subsc_no_auth, notif_data_no_auth)
# Check OAUTH2_CLIENT_CERT certificateRef type
cur_dir = os.path.dirname(__file__)
sample_cert = os.path.join(
cur_dir, "../samples/sample_cert", "notification_client_cert.pem")
CONF.v2_vnfm.notification_mtls_client_cert_file = sample_cert
subsc_basic_auth = objects.LccnSubscriptionV2(
id='sub-2', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=['BASIC'],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test', password='test')))
auth_req_2 = {
"authType": ["OAUTH2_CLIENT_CERT"],
"paramsOauth2ClientCert": {
"clientId": "test",
"certificateRef": {
"type": "x5t#256",
"value": "8Shbulz8zlFdKG-iMCUz5CCv0A7q0k6X7wL3NcZpshM"
},
"tokenEndpoint": "http://127.0.0.1/token"
}
}
ex = self.assertRaises(sol_ex.InvalidSubscription,
common_script_utils.check_subsc_auth,
auth_req_2)
self.assertEqual("certificateRef type is invalid.", ex.detail)
# execute basic_auth
common_script_utils.send_notification(
subsc_basic_auth, notif_data_no_auth)
# Check OAUTH2_CLIENT_CERT certificateRef value
auth_req_3 = {
"authType": ["OAUTH2_CLIENT_CERT"],
"paramsOauth2ClientCert": {
"clientId": "test",
"certificateRef": {
"type": "x5t#S256",
"value": "DHQ2bEQZcdk_OWNxYXor9yoWTV6EDhuz4JU3bkLn17S"
},
"tokenEndpoint": "http://127.0.0.1/token"
}
}
ex = self.assertRaises(sol_ex.InvalidSubscription,
common_script_utils.check_subsc_auth,
auth_req_3)
self.assertEqual("certificateRef value is incorrect.", ex.detail)
subsc_oauth2 = objects.LccnSubscriptionV2(
id='sub-3', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=['OAUTH2_CLIENT_CREDENTIALS'],
paramsOauth2ClientCredentials=(
objects.SubscriptionAuthentication_ParamsOauth2(
clientId='test', clientPassword='test',
tokenEndpoint='http://127.0.0.1/token'))))
# Check OAUTH2_CLIENT_CREDENTIALS
auth_req_4 = {
'authType': ['OAUTH2_CLIENT_CREDENTIALS']
}
ex = self.assertRaises(sol_ex.InvalidSubscription,
common_script_utils.check_subsc_auth,
auth_req_4)
self.assertEqual("paramsOauth2ClientCredentials must be specified.",
ex.detail)
# execute oauth2
common_script_utils.send_notification(subsc_oauth2, notif_data_no_auth)
# Check BASIC
auth_req_5 = {
'authType': ['BASIC'],
}
ex = self.assertRaises(sol_ex.InvalidSubscription,
common_script_utils.check_subsc_auth,
auth_req_5)
self.assertEqual("paramsBasic must be specified.", ex.detail)
subsc_oauth2_mtls = objects.LccnSubscriptionV2(
id='sub-4', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=["OAUTH2_CLIENT_CERT"],
paramsOauth2ClientCert=(
objects.SubscriptionAuthentication_ParamsOauth2ClientCert(
clientId='test',
certificateRef=objects.
ParamsOauth2ClientCert_CertificateRef(
type='x5t#256',
value='03c6e188d1fe5d3da8c9bc9a8dc531a2'
'b3ecf812b03aede9bec7ba1b410b6b64'
),
tokenEndpoint='http://127.0.0.1/token'
)
def test_get_http_auth_handle(self):
# Check NoAuth
no_auth_req = None
# execute NoAuth
result = common_script_utils.get_http_auth_handle(no_auth_req)
self.assertIsInstance(result, http_client.NoAuthHandle)
# Check OAUTH2_CLIENT_CERT
oauth2_mtls_req = objects.SubscriptionAuthentication(
authType=["OAUTH2_CLIENT_CERT"],
paramsOauth2ClientCert=(
objects.SubscriptionAuthentication_ParamsOauth2ClientCert(
clientId='test',
certificateRef=(
objects.ParamsOauth2ClientCert_CertificateRef(
type='x5t#S256',
value='8Shbulz8zlFdKG-iMCUz5CCv0A7q0k6X7wL3NcZpshM'
)
),
tokenEndpoint='http://127.0.0.1/token'
)
)
)
# execute oauth2 mtls
common_script_utils.send_notification(
subsc_oauth2_mtls, notif_data_no_auth)
# execute OAUTH2_CLIENT_CERT
result = common_script_utils.get_http_auth_handle(oauth2_mtls_req)
self.assertIsInstance(result, http_client.OAuth2MtlsAuthHandle)
cfg.CONF.set_override("notification_verify_cert", "True",
group="v2_vnfm")
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-5', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback')
notif_data_no_auth = objects.VnfLcmOperationOccurrenceNotificationV2(
id=uuidutils.generate_uuid()
)
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
# execute no_auth
common_script_utils.send_notification(
subsc_no_auth, notif_data_no_auth)
subsc_basic_auth = objects.LccnSubscriptionV2(
id='sub-6', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=['BASIC'],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test', password='test')
)
)
# execute basic_auth
common_script_utils.send_notification(
subsc_basic_auth, notif_data_no_auth)
subsc_oauth2 = objects.LccnSubscriptionV2(
id='sub-7', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=['OAUTH2_CLIENT_CREDENTIALS'],
paramsOauth2ClientCredentials=(
objects.SubscriptionAuthentication_ParamsOauth2(
clientId='test', clientPassword='test',
tokenEndpoint='http://127.0.0.1/token')
# Check OAUTH2_CLIENT_CREDENTIALS
oauth2_req = objects.SubscriptionAuthentication(
authType=['OAUTH2_CLIENT_CREDENTIALS'],
paramsOauth2ClientCredentials=(
objects.SubscriptionAuthentication_ParamsOauth2(
clientId='test', clientPassword='test',
tokenEndpoint='http://127.0.0.1/token'
)
)
)
# execute oauth2
common_script_utils.send_notification(subsc_oauth2, notif_data_no_auth)
# execute OAUTH2_CLIENT_CREDENTIALS
result = common_script_utils.get_http_auth_handle(oauth2_req)
self.assertIsInstance(result, http_client.OAuth2AuthHandle)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification_error_code(self, mock_resp):
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-1', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback')
notif_data_no_auth = objects.VnfLcmOperationOccurrenceNotificationV2(
id=uuidutils.generate_uuid()
)
resp_no_auth = requests.Response()
resp_no_auth.status_code = 200
mock_resp.return_value = (resp_no_auth, None)
# execute no_auth
common_script_utils.send_notification(
subsc_no_auth, notif_data_no_auth)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification_error(self, mock_resp):
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-1', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback')
notif_data_no_auth = objects.VnfLcmOperationOccurrenceNotificationV2(
id=uuidutils.generate_uuid()
)
resp_no_auth = Exception()
mock_resp.return_value = (resp_no_auth, None)
# execute no_auth
common_script_utils.send_notification(
subsc_no_auth, notif_data_no_auth)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_test_notification(self, mock_resp):
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-1', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback')
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
# execute no_auth
common_script_utils.test_notification(subsc_no_auth)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_test_notification_error_code(self, mock_resp):
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-1', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback')
resp_no_auth = requests.Response()
resp_no_auth.status_code = 200
mock_resp.return_value = (resp_no_auth, None)
# execute no_auth
self.assertRaises(sol_ex.TestNotificationFailed,
common_script_utils.test_notification, subsc_no_auth)
class mock_session():
def request(url, method, raise_exc=False, **kwargs):
resp = requests.Response()
resp.status_code = 400
resp.headers['Content-Type'] = 'application/zip'
return resp
@mock.patch.object(http_client.HttpClient, '_decode_body')
@mock.patch.object(http_client.NoAuthHandle, 'get_session')
def test_test_notification_error(self, mock_session, mock_decode_body):
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-1', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback')
mock_session.return_value = self.mock_session
mock_decode_body.return_value = None
self.assertRaises(sol_ex.TestNotificationFailed,
common_script_utils.test_notification, subsc_no_auth)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_test_notification_fm_subscription(self, mock_resp):
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
subsc_basic_auth = objects.FmSubscriptionV1.from_dict(
fakes_for_fm.fm_subsc_example)
subsc_basic_auth.authentication = objects.SubscriptionAuthentication(
authType=["BASIC"],
# Check BASIC
basic_auth_req = objects.SubscriptionAuthentication(
authType=['BASIC'],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test', password='test'))
common_script_utils.test_notification(
subsc_basic_auth, common_script_utils.NOTIFY_TYPE_FM)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_test_notification_pm_job(self, mock_do_request):
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_do_request.return_value = (resp_no_auth, None)
pm_job = objects.PmJobV2(
id='pm_job_1',
authentication=objects.SubscriptionAuthentication(
authType=["BASIC"],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test',
password='test'
)
),
callbackUri='http://127.0.0.1/callback'
userName='test',
password='test'
)
)
common_script_utils.test_notification(
pm_job, common_script_utils.NOTIFY_TYPE_PM)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification_fm_subscribtion(self, mock_resp):
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
subsc_basic_auth = objects.FmSubscriptionV1.from_dict(
fakes_for_fm.fm_subsc_example)
subsc_basic_auth.authentication = objects.SubscriptionAuthentication(
authType=["BASIC"],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test', password='test'))
alarm = objects.AlarmV1.from_dict(
fakes_for_fm.alarm_example)
notif_data = fm_alarm_utils.make_alarm_notif_data(
subsc_basic_auth, alarm, 'http://127.0.0.1:9890')
common_script_utils.send_notification(
subsc_basic_auth, notif_data)
self.assertEqual(1, mock_resp.call_count)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification_pm_job(self, mock_resp):
pm_job = objects.PmJobV2(
id='pm_job_1',
objectType='VNF',
authentication=objects.SubscriptionAuthentication(
authType=["BASIC"],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test',
password='test'
),
),
callbackUri='http://127.0.0.1/callback'
)
sub_instance_ids = ['1', '2', '3', '4']
notif_data = pm_job_utils.make_pm_notif_data('instance_id',
sub_instance_ids,
'report_id',
pm_job,
'2008-01-03 08:04:34',
'endpoint')
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
common_script_utils.send_notification(
pm_job, notif_data, common_script_utils.NOTIFY_TYPE_PM)
# execute BASIC
result = common_script_utils.get_http_auth_handle(basic_auth_req)
self.assertIsInstance(result, http_client.BasicAuthHandle)

226
tacker/tests/unit/sol_refactored/common/test_subscription_utils.py
View File

@@ -13,15 +13,26 @@
# License for the specific language governing permissions and limitations
# under the License.
import os
import requests
from unittest import mock
from oslo_utils import uuidutils
from tacker import context
from tacker.sol_refactored.api import api_version
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import fm_alarm_utils
from tacker.sol_refactored.common import http_client
from tacker.sol_refactored.common import pm_job_utils
from tacker.sol_refactored.common import subscription_utils as subsc_utils
from tacker.sol_refactored import objects
from tacker.tests import base
from tacker.tests.unit.sol_refactored.samples import fakes_for_fm
CONF = config.CONF
class TestSubscriptionUtils(base.BaseTestCase):
@@ -276,7 +287,7 @@ class TestSubscriptionUtils(base.BaseTestCase):
self.assertEqual('subsc-1', result.subscriptionId)
self.assertEqual('test-instance', result.vnfInstanceId)
def test_check_http_client_auth(self):
def test_get_subsc_auth(self):
auth_req_1 = {
'authType': ['BASIC'],
'paramsBasic': {
@@ -284,7 +295,7 @@ class TestSubscriptionUtils(base.BaseTestCase):
'password': 'test'
},
}
result = subsc_utils.check_http_client_auth(auth_req_1)
result = subsc_utils.get_subsc_auth(auth_req_1)
self.assertEqual(['BASIC'], result.authType)
auth_req_2 = {
@@ -296,43 +307,202 @@ class TestSubscriptionUtils(base.BaseTestCase):
'http://127.0.0.1/token'
}
}
result = subsc_utils.check_http_client_auth(auth_req_2)
result = subsc_utils.get_subsc_auth(auth_req_2)
self.assertEqual(['OAUTH2_CLIENT_CREDENTIALS'], result.authType)
cur_dir = os.path.dirname(__file__)
sample_cert = os.path.join(
cur_dir, "../samples/sample_cert", "notification_client_cert.pem")
CONF.v2_vnfm.notification_mtls_client_cert_file = sample_cert
auth_req_3 = {
'authType': ['OAUTH2_CLIENT_CERT'],
'paramsOauth2ClientCert': {
'clientId': 'test',
'certificateRef': {
'type': 'x5t#256',
'value': '03c6e188d1fe5d3da8c9bc9a8dc531a2'
'b3ecf812b03aede9bec7ba1b410b6b64'
'type': 'x5t#S256',
'value': '8Shbulz8zlFdKG-iMCUz5CCv0A7q0k6X7wL3NcZpshM'
},
'tokenEndpoint': 'http://127.0.0.1/token'
}
}
result = subsc_utils.check_http_client_auth(auth_req_3)
result = subsc_utils.get_subsc_auth(auth_req_3)
self.assertEqual(['OAUTH2_CLIENT_CERT'], result.authType)
def test_check_http_client_auth_error(self):
auth_req_1 = {
'authType': ['BASIC'],
'paramsBasic': None
}
self.assertRaises(sol_ex.InvalidSubscription,
subsc_utils.check_http_client_auth,
auth_req=auth_req_1)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification(self, mock_resp):
subsc_oauth2_mtls = objects.LccnSubscriptionV2(
id='sub-1', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=["OAUTH2_CLIENT_CERT"],
paramsOauth2ClientCert=(
objects.SubscriptionAuthentication_ParamsOauth2ClientCert(
clientId='test',
certificateRef=(
objects.ParamsOauth2ClientCert_CertificateRef(
type='x5t#S256',
value='8Shbulz8zlFdKG-iMCUz5C'
'Cv0A7q0k6X7wL3NcZpshM'
)
),
tokenEndpoint='http://127.0.0.1/token'
)
)
)
)
auth_req_2 = {
'authType': ['OAUTH2_CLIENT_CREDENTIALS'],
}
self.assertRaises(sol_ex.InvalidSubscription,
subsc_utils.check_http_client_auth,
auth_req=auth_req_2)
notif_data_no_auth = objects.VnfLcmOperationOccurrenceNotificationV2(
id=uuidutils.generate_uuid()
)
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
auth_req_3 = {
'authType': ['OAUTH2_CLIENT_CERT']
}
self.assertRaises(sol_ex.InvalidSubscription,
subsc_utils.check_http_client_auth,
auth_req=auth_req_3)
# execute oauth2 mtls
subsc_utils.send_notification(subsc_oauth2_mtls, notif_data_no_auth)
mock_resp.assert_called_once()
@mock.patch('tacker.sol_refactored.common.subscription_utils.LOG')
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification_error(self, mock_resp,
mock_log):
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-1', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback')
notif_data_no_auth = objects.VnfLcmOperationOccurrenceNotificationV2(
id=uuidutils.generate_uuid()
)
mock_resp.side_effect = sol_ex.SolException(
sol_status=400, sol_detail="unit test")
# execute no_auth
subsc_utils.send_notification(subsc_no_auth, notif_data_no_auth)
expected_message = "send_notification failed: unit test"
mock_log.exception.assert_called_with(expected_message)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification_fm_subscription(self, mock_resp):
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
subsc_basic_auth = objects.FmSubscriptionV1.from_dict(
fakes_for_fm.fm_subsc_example)
subsc_basic_auth.authentication = objects.SubscriptionAuthentication(
authType=["BASIC"],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test',
password='test'
)
)
alarm = objects.AlarmV1.from_dict(
fakes_for_fm.alarm_example)
notif_data = fm_alarm_utils.make_alarm_notif_data(
subsc_basic_auth, alarm, 'http://127.0.0.1:9890')
subsc_utils.send_notification(subsc_basic_auth, notif_data)
mock_resp.assert_called_once()
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification_pm_job(self, mock_resp):
pm_job = objects.PmJobV2(
id='pm_job_1',
objectType='VNF',
authentication=objects.SubscriptionAuthentication(
authType=["BASIC"],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test',
password='test'
)
),
callbackUri='http://127.0.0.1/callback'
)
sub_instance_ids = ['1', '2', '3', '4']
notif_data = pm_job_utils.make_pm_notif_data('instance_id',
sub_instance_ids,
'report_id',
pm_job,
'2008-01-03 08:04:34',
'endpoint')
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
subsc_utils.send_notification(
pm_job, notif_data, subsc_utils.NOTIFY_TYPE_PM)
mock_resp.assert_called_once()
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_test_notification(self, mock_resp):
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-1', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback')
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
# execute no_auth
subsc_utils.test_notification(subsc_no_auth)
mock_resp.assert_called_once()
class mock_session():
def request(url, method, raise_exc=False, **kwargs):
resp = requests.Response()
resp.status_code = 400
resp.headers['Content-Type'] = 'application/zip'
return resp
@mock.patch.object(http_client.HttpClient, '_decode_body')
@mock.patch.object(http_client.NoAuthHandle, 'get_session')
def test_test_notification_error(self, mock_session, mock_decode_body):
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-1', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback'
)
mock_session.return_value = self.mock_session
mock_decode_body.return_value = None
self.assertRaises(sol_ex.TestNotificationFailed,
subsc_utils.test_notification, subsc_no_auth)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_test_notification_fm_subscription(self, mock_resp):
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
subsc_basic_auth = objects.FmSubscriptionV1.from_dict(
fakes_for_fm.fm_subsc_example)
subsc_basic_auth.authentication = objects.SubscriptionAuthentication(
authType=["BASIC"],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test',
password='test'
)
)
subsc_utils.test_notification(subsc_basic_auth,
subsc_utils.NOTIFY_TYPE_FM)
mock_resp.assert_called_once()
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_test_notification_pm_job(self, mock_resp):
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
pm_job = objects.PmJobV2(
id='pm_job_1',
authentication=objects.SubscriptionAuthentication(
authType=["BASIC"],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test',
password='test'
)
),
callbackUri='http://127.0.0.1/callback'
)
subsc_utils.test_notification(pm_job, subsc_utils.NOTIFY_TYPE_PM)
mock_resp.assert_called_once()

4
tacker/tests/unit/sol_refactored/conductor/test_vnffm_driver_v1.py
View File

@@ -18,8 +18,8 @@ from unittest import mock
from oslo_utils import uuidutils
from tacker import context
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import fm_subscription_utils
from tacker.sol_refactored.common import subscription_utils as subsc_utils
from tacker.sol_refactored.common import vnf_instance_utils as inst_utils
from tacker.sol_refactored.conductor import vnffm_driver_v1
from tacker.sol_refactored import objects
@@ -35,7 +35,7 @@ class TestVnffmDriverV1(base.BaseTestCase):
self.driver = vnffm_driver_v1.VnfFmDriverV1()
self.context = context.get_admin_context()
@mock.patch.object(common_script_utils, 'send_notification')
@mock.patch.object(subsc_utils, 'send_notification')
@mock.patch.object(fm_subscription_utils, 'get_alarm_subscs')
@mock.patch.object(objects.base.TackerPersistentObject, 'create')
@mock.patch.object(objects.base.TackerPersistentObject, 'update')

4
tacker/tests/unit/sol_refactored/conductor/test_vnfpm_driver_v2.py
View File

@@ -18,9 +18,9 @@ from unittest import mock
from tacker.tests import base
from tacker import context
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import pm_job_utils
from tacker.sol_refactored.common import pm_threshold_utils
from tacker.sol_refactored.common import subscription_utils as subsc_utils
from tacker.sol_refactored.conductor.vnfpm_driver_v2 import VnfPmDriverV2
from tacker.sol_refactored.nfvo.nfvo_client import NfvoClient
from tacker.sol_refactored import objects
@@ -72,7 +72,7 @@ class TestVnfPmDriverV2(base.BaseTestCase):
VnfPmDriverV2().store_job_info(context=self.context,
report=report)
@mock.patch.object(common_script_utils, 'send_notification')
@mock.patch.object(subsc_utils, 'send_notification')
@mock.patch.object(objects.base.TackerPersistentObject, 'get_by_id')
@mock.patch.object(pm_threshold_utils,
'update_threshold_state_data')

33
tacker/tests/unit/sol_refactored/controller/test_vnffm_v1.py
View File

@@ -14,15 +14,18 @@
# under the License.
import os
import requests
from unittest import mock
from tacker import context
from tacker.sol_refactored.api import api_version
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import fm_alarm_utils as alarm_utils
from tacker.sol_refactored.common import fm_subscription_utils as subsc_utils
from tacker.sol_refactored.common import fm_subscription_utils\
as fm_subsc_utils
from tacker.sol_refactored.common import subscription_utils as subsc_utils
from tacker.sol_refactored.controller import vnffm_v1
from tacker.sol_refactored import objects
from tacker.tests import base
@@ -32,6 +35,8 @@ SAMPLE_INST_ID = 'c61314d0-f583-4ab3-a457-46426bce02d3'
SAMPLE_ALARM_ID = '78a39661-60a8-4824-b989-88c1b0c3534a'
SAMPLE_SUBSC_ID = '78a39661-60a8-4824-b989-88c1b0c3534a'
CONF = config.CONF
class TestVnffmV1(base.BaseTestCase):
@@ -110,8 +115,12 @@ class TestVnffmV1(base.BaseTestCase):
request=self.request, id=SAMPLE_ALARM_ID, body=body)
@mock.patch.object(objects.base.TackerPersistentObject, 'create')
@mock.patch.object(common_script_utils, 'test_notification')
@mock.patch.object(subsc_utils, 'test_notification')
def test_subscription_create(self, mock_test, mock_create):
cur_dir = os.path.dirname(__file__)
sample_cert = os.path.join(
cur_dir, "../samples/sample_cert", "notification_client_cert.pem")
CONF.v2_vnfm.notification_mtls_client_cert_file = sample_cert
body_1 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
@@ -129,9 +138,8 @@ class TestVnffmV1(base.BaseTestCase):
"paramsOauth2ClientCert": {
"clientId": "test",
"certificateRef": {
"type": "x5t#256",
"value": "03c6e188d1fe5d3da8c9bc9a8dc531a2"
"b3ecf812b03aede9bec7ba1b410b6b64"
"type": "x5t#S256",
"value": "8Shbulz8zlFdKG-iMCUz5CCv0A7q0k6X7wL3NcZpshM"
},
"tokenEndpoint": "https://127.0.0.1/token"
}
@@ -189,9 +197,8 @@ class TestVnffmV1(base.BaseTestCase):
"paramsOauth2ClientCert": {
"clientId": "test",
"certificateRef": {
"type": "x5t#256",
"value": "03c6e188d1fe5d3da8c9bc9a8dc531a2"
"b3ecf812b03aede9bec7ba1b410b6b64"
"type": "x5t#S256",
"value": "8Shbulz8zlFdKG-iMCUz5CCv0A7q0k6X7wL3NcZpshM"
},
"tokenEndpoint": "https://127.0.0.1/token"
}
@@ -215,7 +222,7 @@ class TestVnffmV1(base.BaseTestCase):
ex = self.assertRaises(sol_ex.InvalidSubscription,
self.controller.subscription_create, request=self.request,
body=body)
self.assertEqual("ParamsBasic must be specified.", ex.detail)
self.assertEqual("paramsBasic must be specified.", ex.detail)
body = {
"callbackUri": "http://127.0.0.1:6789/notification",
@@ -241,7 +248,7 @@ class TestVnffmV1(base.BaseTestCase):
self.assertEqual("paramsOauth2ClientCert must be specified.",
ex.detail)
@mock.patch.object(subsc_utils, 'get_subsc_all')
@mock.patch.object(fm_subsc_utils, 'get_subsc_all')
def test_subscription_list(self, mock_subsc):
request = requests.Request()
request.context = self.context
@@ -258,7 +265,7 @@ class TestVnffmV1(base.BaseTestCase):
result = self.controller.subscription_list(request)
self.assertEqual(200, result.status)
@mock.patch.object(subsc_utils, 'get_subsc')
@mock.patch.object(fm_subsc_utils, 'get_subsc')
def test_subscription_show(self, mock_subsc):
mock_subsc.return_value = objects.FmSubscriptionV1.from_dict(
fakes_for_fm.fm_subsc_example)
@@ -266,7 +273,7 @@ class TestVnffmV1(base.BaseTestCase):
request=self.request, id=SAMPLE_SUBSC_ID)
self.assertEqual(200, result.status)
@mock.patch.object(subsc_utils, 'get_subsc')
@mock.patch.object(fm_subsc_utils, 'get_subsc')
@mock.patch.object(objects.base.TackerPersistentObject, 'delete')
def test_subscription_delete(self, mock_delete, mock_subsc):
mock_subsc.return_value = objects.FmSubscriptionV1.from_dict(

20
tacker/tests/unit/sol_refactored/controller/test_vnflcm_v2.py
View File

@@ -16,6 +16,7 @@ import copy
from datetime import datetime
import ddt
from http import client as http_client
import os
import requests
from unittest import mock
@@ -27,7 +28,6 @@ from tacker import context
from tacker import policy
from tacker.sol_refactored.api import api_version
from tacker.sol_refactored.api.policies.vnflcm_v2 import POLICY_NAME
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import lcm_op_occ_utils as lcmocc_utils
@@ -1054,7 +1054,7 @@ class TestVnflcmV2(db_base.SqlTestCase):
ex = self.assertRaises(sol_ex.InvalidSubscription,
self.controller.subscription_create, request=self.request,
body=body_1)
self.assertEqual("ParamsBasic must be specified.", ex.detail)
self.assertEqual("paramsBasic must be specified.", ex.detail)
body_2 = {
"callbackUri": "http://127.0.0.1:6789/notification",
@@ -1080,8 +1080,12 @@ class TestVnflcmV2(db_base.SqlTestCase):
self.assertEqual("paramsOauth2ClientCert must be specified.",
ex.detail)
@mock.patch.object(common_script_utils, 'test_notification')
@mock.patch.object(subsc_utils, 'test_notification')
def test_subscription_create_201(self, mock_test):
cur_dir = os.path.dirname(__file__)
sample_cert = os.path.join(
cur_dir, "../samples/sample_cert", "notification_client_cert.pem")
CONF.v2_vnfm.notification_mtls_client_cert_file = sample_cert
body_1 = {
"callbackUri": "http://127.0.0.1:6789/notification",
"authentication": {
@@ -1099,9 +1103,8 @@ class TestVnflcmV2(db_base.SqlTestCase):
"paramsOauth2ClientCert": {
"clientId": "test",
"certificateRef": {
"type": "x5t#256",
"value": "03c6e188d1fe5d3da8c9bc9a8dc531a2"
"b3ecf812b03aede9bec7ba1b410b6b64"
"type": "x5t#S256",
"value": "8Shbulz8zlFdKG-iMCUz5CCv0A7q0k6X7wL3NcZpshM"
},
"tokenEndpoint": "https://127.0.0.1/token"
}
@@ -1156,9 +1159,8 @@ class TestVnflcmV2(db_base.SqlTestCase):
"paramsOauth2ClientCert": {
"clientId": "test",
"certificateRef": {
"type": "x5t#256",
"value": "03c6e188d1fe5d3da8c9bc9a8dc531a2"
"b3ecf812b03aede9bec7ba1b410b6b64"
"type": "x5t#S256",
"value": "8Shbulz8zlFdKG-iMCUz5CCv0A7q0k6X7wL3NcZpshM"
},
"tokenEndpoint": "https://127.0.0.1/token"
}

10
tacker/tests/unit/sol_refactored/controller/test_vnfpm_v2.py
View File

@@ -18,12 +18,12 @@ from unittest import mock
from tacker import context
from tacker.sol_refactored.api import api_version
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import pm_job_utils
from tacker.sol_refactored.common.prometheus_plugin import (
PrometheusPluginThreshold)
from tacker.sol_refactored.common import subscription_utils as subsc_utils
from tacker.sol_refactored.controller.vnflcm_view import BaseViewBuilder
from tacker.sol_refactored.controller.vnflcm_view import Pager
from tacker.sol_refactored.controller import vnfpm_v2
@@ -247,7 +247,7 @@ class TestVnfpmV2(base.BaseTestCase):
self.controller.create, request=self.request, body=body)
@mock.patch.object(objects.base.TackerPersistentObject, 'create')
@mock.patch.object(common_script_utils, 'test_notification')
@mock.patch.object(subsc_utils, 'test_notification')
@mock.patch.object(objects.base.TackerPersistentObject, 'get_by_id')
def test_create_201(self, mock_inst, mock_notifi, mock_create):
mock_inst.return_value = objects.VnfInstanceV2(
@@ -335,7 +335,7 @@ class TestVnfpmV2(base.BaseTestCase):
@mock.patch.object(objects.base.TackerPersistentObject, 'update')
@mock.patch.object(objects.base.TackerPersistentObject, 'get_by_id')
@mock.patch.object(common_script_utils, 'test_notification')
@mock.patch.object(subsc_utils, 'test_notification')
def test_update(self, mock_notifi, mock_pm, mock_update):
mock_notifi.return_value = None
mock_pm.return_value = objects.PmJobV2(id='pm_job_1')
@@ -486,7 +486,7 @@ class TestVnfpmV2(base.BaseTestCase):
@mock.patch.object(objects.base.TackerPersistentObject, 'create')
@mock.patch.object(PrometheusPluginThreshold, 'create_threshold')
@mock.patch.object(common_script_utils, 'test_notification')
@mock.patch.object(subsc_utils, 'test_notification')
@mock.patch.object(objects.base.TackerPersistentObject, 'get_by_id')
def test_create_pm_threshold(
self, mock_inst, mock_notifi, mock_create_threshold, mock_create):
@@ -584,7 +584,7 @@ class TestVnfpmV2(base.BaseTestCase):
@mock.patch.object(objects.base.TackerPersistentObject, 'update')
@mock.patch.object(objects.base.TackerPersistentObject, 'get_by_id')
@mock.patch.object(common_script_utils, 'test_notification')
@mock.patch.object(subsc_utils, 'test_notification')
def test_pm_threshold_update(self, mock_notifi, mock_pm, mock_update):
mock_notifi.return_value = None
mock_pm.return_value = objects.ThresholdV2(

10
tacker/tests/unit/sol_refactored/nfvo/test_nfvo_client.py
View File

@@ -23,10 +23,10 @@ from oslo_utils import uuidutils
from tacker import context
from tacker.sol_refactored.api import api_version
from tacker.sol_refactored.common import common_script_utils
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import http_client
from tacker.sol_refactored.common import pm_job_utils
from tacker.sol_refactored.common import subscription_utils as subsc_utils
from tacker.sol_refactored.common import vnfd_utils
from tacker.sol_refactored.nfvo import local_nfvo
from tacker.sol_refactored.nfvo import nfvo_client
@@ -465,7 +465,7 @@ class TestNfvoClient(base.BaseTestCase):
'vimAssets']['softwareImages'][0]['vimSoftwareImageId'])
@mock.patch.object(objects.base.TackerPersistentObject, 'get_all')
@mock.patch.object(common_script_utils, 'send_notification')
@mock.patch.object(subsc_utils, 'send_notification')
@mock.patch.object(local_nfvo.LocalNfvo, 'recv_inst_create_notification')
def test_send_inst_create_notification(
self, mock_recv, mock_send, mock_subscs):
@@ -479,7 +479,7 @@ class TestNfvoClient(base.BaseTestCase):
self.assertEqual(1, mock_send.call_count)
@mock.patch.object(objects.base.TackerPersistentObject, 'get_all')
@mock.patch.object(common_script_utils, 'send_notification')
@mock.patch.object(subsc_utils, 'send_notification')
@mock.patch.object(local_nfvo.LocalNfvo, 'recv_inst_delete_notification')
def test_send_inst_delete_notification(
self, mock_recv, mock_send, mock_subscs):
@@ -493,7 +493,7 @@ class TestNfvoClient(base.BaseTestCase):
self.assertEqual(1, mock_send.call_count)
@mock.patch.object(objects.base.TackerPersistentObject, 'get_all')
@mock.patch.object(common_script_utils, 'send_notification')
@mock.patch.object(subsc_utils, 'send_notification')
@mock.patch.object(local_nfvo.LocalNfvo, 'recv_lcmocc_notification')
def test_send_lcmocc_notification(self, mock_recv, mock_send, mock_subscs):
inst = objects.VnfInstanceV2(id='test-instance')
@@ -505,7 +505,7 @@ class TestNfvoClient(base.BaseTestCase):
self.assertEqual(1, mock_recv.call_count)
self.assertEqual(1, mock_send.call_count)
@mock.patch.object(common_script_utils, 'send_notification')
@mock.patch.object(subsc_utils, 'send_notification')
@mock.patch.object(pm_job_utils, 'make_pm_notif_data')
def test_send_pm_job_notification(self, mock_notif, mock_send):
mock_notif.return_value = 'mock_notif'

32
tacker/tests/unit/sol_refactored/samples/sample_cert/notification_client_cert.pem Normal file
View File

@@ -0,0 +1,32 @@
-----BEGIN CERTIFICATE-----
MIIFeTCCA2ECFAqJwwzS3qBO9ZXRARMCma8IAZmTMA0GCSqGSIb3DQEBDAUAMHkx
CzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMRkwFwYDVQQKDBB0YWNr
ZXItY2Etc2FtcGxlMRkwFwYDVQQLDBB0YWNrZXItY2Etc2FtcGxlMR8wHQYDVQQD
DBZ0YWNrZXItY2Etc2FtcGxlLmxvY2FsMB4XDTIzMDYxNTA1NTI1MVoXDTMzMDYx
MjA1NTI1MVoweTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxGTAX
BgNVBAoMEHRhY2tlci11dC1zYW1wbGUxGTAXBgNVBAsMEHRhY2tlci11dC1zYW1w
bGUxHzAdBgNVBAMMFnRhY2tlci11dC1zYW1wbGUubG9jYWwwggIiMA0GCSqGSIb3
DQEBAQUAA4ICDwAwggIKAoICAQDCPLgLsS4s9EK4ZiyZ5Jpr3x5ffxsji8FoEdav
oR/hoifKTbJUKUF9Nn+4i0dradxvIALnXUzLVyshn0GTqOWgouxr79kZGyzklZaN
2+cvXU+cGKgO2Wse2jtaUvdCRD1wS54CkfB1/f/6G1xStUe7xyftIdIYw58FLYEh
+KKEQOfl2rWsqqC0kJiMXdp7d5ClzkFS9fssZzGtvaH75DqbNjGOxV76ZxIgm26h
n8JQuRR+WSnA39TP30dTio7wG+VwW2w1FqzgrRdhReFfHUuWs3Bbg/BIMSsK+S/8
L3veTpDlXzV4TzCGzhQRYzLbNj+t5/qLiTJINO4Ah0OSki+LPa/AuiABicxOXFpu
S1y0cOtWvn7kAM3ioo+2caYIFcxcRrtKdUVrCjU3mHLus5UyiwT9DQLZF/2RGDed
Nbtn3OHMbpS1QW4miewMFa3vWo6wTn7ST5cJcKyEhDHgGG5UaYKxlhWXhnv0KbKP
FigPQWtnab46m6GYpSnAsccKFGBB8aqDXROzNXkDFTFjGfydUqr7iKmD7JqjjHV8
+6toZD7bc3hCUC43bL1/n5BZrISRXAXJRzg588/vbVjWR6zdfqkgkEvLZ1qIYg5V
X0KGHUGdUg/dMQq8WsnWU37GwZxBnAGSXs2BHfIk1sfEXbQjMLW5kpRTsOdemZr9
aYJYoQIDAQABMA0GCSqGSIb3DQEBDAUAA4ICAQCZwYTOafeEDS9HaYz2XNeTb60O
+LLN84zmA3S0ufithI7DQ4GA9ko6LvmHyDx6A4qIjrrRlZItwST9jjx2g2nriMRY
yTW9WNd86jdU74ujzNO5gvNKXZfkhhxknLGmv3wDTaBNZK7NgAmgI0ATNac1bbJ3
IRSrJEItoBmHxVFg3fbq5XS4fa5WawZlgiCmY0icYeFug5+RGZMh//KV6aXPNRUk
qx91E50YhTec83Hs5rZUQT24cd4lsiBgH5ibnRh0slwrKJirYfdp0FfjbbmPF0uj
Rv8hLWRjAfD4gk9LCO8UuU6ivRzbQnJ5RSQI46GtoqAEVMPGIMVyxqOvMa0hRo06
3cvRYfCDPSq5+IfHUaHqhpp5DTOA9NGuLjmD4qrz5IhbVbNr3RgG4IhcwbHkYaKz
XjfJa5ymo5eADfVYP5/VxhyzSYMFfWHwzqqmFF4gOB0DT3Kp00E9iHJSrSgI3LFi
W4lEE1FWP2ZpRZwmilPPjSSHanNOIBXpXD/H4SJpLZJGa1m+1jqVo5GhHpeEorXI
dZN4iJNNwe99Dq7dyCT39r1E1YvSekc+GWIJ4aUPRL2xIncKxOboT/0CnGMZ+/jF
H3PlIwrkYYJpHshcNHMS6GfkSANOB0okdtDU25c9+HNkEF3wJm/fLzJu36YholPt
wLSXLxUydHy6QdOg5w==
-----END CERTIFICATE-----