# limitations under the License.
# ============LICENSE_END=========================================================
#
+from distutils import util
import os
import re
import uuid
from binascii import hexlify
from kubernetes import config, client, stream
+from .sans_parser import SansParser
# Default values for readiness probe
PROBE_DEFAULT_PERIOD = 15
return service
-def create_secret_with_password(namespace, secret_prefix, password_length):
+def create_secret_with_password(namespace, secret_prefix, password_key, password_length):
"""
Creates K8s secret object with a generated password.
Returns: secret name and data key.
password_base64 = _encode_base64(password)
metadata = {'generateName': secret_prefix, 'namespace': namespace}
- key = 'data'
+ key = password_key
data = {key: password_base64}
response = _create_k8s_secret(namespace, metadata, data, 'Opaque')
def _add_cert_post_processor_init_container(ctx, init_containers, tls_info, tls_config, external_cert,
- cert_post_processor_config):
+ cert_post_processor_config, isCertManagerIntegration):
# Adds an InitContainer to the pod to merge TLS and external TLS truststore into single file.
docker_image = cert_post_processor_config["image_tag"]
ctx.logger.info("Creating init container: cert post processor \n * [" + docker_image + "]")
# Create the volumes and volume mounts
init_volume_mounts = [client.V1VolumeMount(name="tls-info", mount_path=tls_cert_dir)]
-
+ if isCertManagerIntegration:
+ init_volume_mounts.append(client.V1VolumeMount(
+ name="certmanager-certs-volume", mount_path=ext_cert_dir))
# Create the init container
init_containers.append(
_create_container_object("cert-post-processor", docker_image, False, volume_mounts=init_volume_mounts, env=env))
return {"pod": pod_name, "output": output}
+def _create_certificate_subject(external_tls_config):
+ """
+ Map parameters to custom resource subject
+ """
+ organization = external_tls_config.get("organization")
+ organization_unit = external_tls_config.get("organizational_unit")
+ country = external_tls_config.get("country")
+ location = external_tls_config.get("location")
+ state = external_tls_config.get("state")
+ subject = {
+ "organizations": [organization],
+ "countries": [country],
+ "localities": [location],
+ "provinces": [state],
+ "organizationalUnits": [organization_unit]
+ }
+ return subject
+
+
+def _create_keystores_object(type, password_secret):
+ """
+ Create keystore property (JKS and PKC12 certificate) for custom resource
+ """
+ return {type: {
+ "create": True,
+ "passwordSecretRef": {
+ "name": password_secret,
+ "key": "password"
+ }}}
+
+
+def _get_keystores_object_type(output_type):
+ """
+ Map config type to custom resource cert type
+ """
+ return {
+ 'p12': 'pkcs12',
+ 'jks': 'jks',
+ }[output_type]
+
+
+def _create_projected_volume_with_password(cert_type, cert_secret_name, password_secret_name, password_secret_key):
+ """
+ Create volume for password protected certificates.
+ Secret contains passwords must be provided
+ """
+ extension = _get_file_extension(cert_type)
+ keystore_file_name = "keystore." + extension
+ truststore_file_name = "truststore." + extension
+ items = [client.V1KeyToPath(key=keystore_file_name, path=keystore_file_name),
+ client.V1KeyToPath(key=truststore_file_name, path=truststore_file_name)]
+ passwords = [client.V1KeyToPath(key=password_secret_key, path="keystore.pass"), client.V1KeyToPath(key=password_secret_key, path="truststore.pass")]
+
+ sec_projection = client.V1SecretProjection(name=cert_secret_name, items=items)
+ sec_passwords_projection = client.V1SecretProjection(name=password_secret_name, items=passwords)
+ sec_volume_projection = client.V1VolumeProjection(secret=sec_projection)
+ sec_passwords_volume_projection = client.V1VolumeProjection(secret=sec_passwords_projection)
+
+ return [sec_volume_projection, sec_passwords_volume_projection]
+
+
+def _create_pem_projected_volume(cert_secret_name):
+ """
+ Create volume for pem certificate
+ """
+ items = [client.V1KeyToPath(key="tls.crt", path="keystore.pem"),
+ client.V1KeyToPath(key="ca.crt", path="truststore.pem"),
+ client.V1KeyToPath(key="tls.key", path="key.pem")]
+ sec_projection = client.V1SecretProjection(name=cert_secret_name, items=items)
+ return [client.V1VolumeProjection(secret=sec_projection)]
+
+
+def create_certificate_object(ctx, cert_secret_name, external_cert_data, external_tls_config, cert_name, issuer):
+ """
+ Create cert-manager certificate custom resource object
+ """
+ common_name = external_cert_data.get("external_certificate_parameters").get("common_name")
+ subject = _create_certificate_subject(external_tls_config)
+
+ custom_resource = {
+ "apiVersion": "cert-manager.io/v1",
+ "kind": "Certificate",
+ "metadata": {"name": cert_name },
+ "spec": {
+ "secretName": cert_secret_name,
+ "commonName": common_name,
+ "issuerRef": {
+ "group": "certmanager.onap.org",
+ "kind": "CMPv2Issuer",
+ "name": issuer
+ }
+ }
+ }
+ custom_resource.get("spec")["subject"] = subject
+
+ raw_sans = external_cert_data.get("external_certificate_parameters").get("sans")
+ ctx.logger.info("Read SANS property: " + str(raw_sans))
+ sans = SansParser().parse_sans(raw_sans)
+ ctx.logger.info("Parsed SANS: " + str(sans))
+
+ if len(sans["ips"]) > 0:
+ custom_resource.get("spec")["ipAddresses"] = sans["ips"]
+ if len(sans["dnss"]) > 0:
+ custom_resource.get("spec")["dnsNames"] = sans["dnss"]
+ if len(sans["emails"]) > 0:
+ custom_resource.get("spec")["emailAddresses"] = sans["emails"]
+ if len(sans["uris"]) > 0:
+ custom_resource.get("spec")["uris"] = sans["uris"]
+
+ return custom_resource
+
+
+def _create_certificate_custom_resource(ctx, external_cert_data, external_tls_config, issuer, namespace, component_name, volumes, volume_mounts, deployment_description):
+ """
+ Create certificate custom resource for provided configuration
+ :param ctx: context
+ :param external_cert_data: object contains certificate common name and
+ SANs list
+ :param external_tls_config: object contains information about certificate subject
+ :param issuer: issuer-name
+ :param namespace: namespace
+ :param component_name: component name
+ :param volumes: list of deployment volume
+ :param volume_mounts: list of deployment volume mounts
+ :param deployment_description: list contains deployment information,
+ method appends created cert and secrets
+ """
+ ctx.logger.info("Creating certificate custom resource")
+ ctx.logger.info("External cert data: " + str(external_cert_data))
+
+ cert_type = (external_cert_data.get("cert_type") or DEFAULT_CERT_TYPE).lower()
+
+ api = client.CustomObjectsApi()
+ cert_secret_name = component_name + "-secret"
+ cert_name = component_name + "-cert"
+ cert_dir = external_cert_data.get("external_cert_directory") + "external/"
+ custom_resource = create_certificate_object(ctx, cert_secret_name,
+ external_cert_data,
+ external_tls_config,
+ cert_name, issuer)
+ # Create the volumes
+ if cert_type != 'pem':
+ ctx.logger.info("Creating volume with passwords")
+ password_secret_name, password_secret_key = create_secret_with_password(namespace, component_name + "-cert-password", "password", 30)
+ deployment_description["secrets"].append(password_secret_name)
+ custom_resource.get("spec")["keystores"] = _create_keystores_object(_get_keystores_object_type(cert_type), password_secret_name)
+ projected_volume_sources = _create_projected_volume_with_password(
+ cert_type, cert_secret_name, password_secret_name, password_secret_key)
+ else:
+ ctx.logger.info("Creating PEM volume")
+ projected_volume_sources = _create_pem_projected_volume(cert_secret_name)
+
+ # Create the volume mounts
+ projected_volume = client.V1ProjectedVolumeSource(sources=projected_volume_sources)
+ volumes.append(client.V1Volume(name="certmanager-certs-volume", projected=projected_volume))
+ volume_mounts.append(client.V1VolumeMount(name="certmanager-certs-volume", mount_path=cert_dir))
+
+ #Create certificate custom resource
+ ctx.logger.info("Certificate CRD: " + str(custom_resource))
+ api.create_namespaced_custom_object(
+ group="cert-manager.io",
+ version="v1",
+ namespace=namespace,
+ plural="certificates",
+ body=custom_resource
+ )
+ deployment_description["certificates"].append(cert_name)
+ deployment_description["secrets"].append(cert_secret_name)
+ ctx.logger.info("CRD certificate created")
+
+
def deploy(ctx, namespace, component_name, image, replicas, always_pull, k8sconfig, **kwargs):
"""
This will create a k8s Deployment and, if needed, one or two k8s Services.
"namespace": namespace,
"location": kwargs.get("k8s_location"),
"deployment": '',
- "services": []
+ "services": [],
+ "certificates": [],
+ "secrets": []
}
try:
# Set up external TLS information
external_cert = kwargs.get("external_cert")
+ cmpv2_issuer_config = k8sconfig.get("cmpv2_issuer")
+ ctx.logger.info("CMPv2 Issuer properties: " + str(cmpv2_issuer_config))
+
+ cmpv2_integration_enabled = bool(util.strtobool(cmpv2_issuer_config.get("enabled")))
+ ctx.logger.info("CMPv2 integration enabled: " + str(cmpv2_integration_enabled))
+
+
if external_cert and external_cert.get("use_external_tls"):
- _add_external_tls_init_container(ctx, init_containers, volumes, external_cert,
- k8sconfig.get("external_cert"))
+ if cmpv2_integration_enabled:
+ _create_certificate_custom_resource(ctx, external_cert,
+ k8sconfig.get("external_cert"),
+ cmpv2_issuer_config.get("name"),
+ namespace,
+ component_name, volumes,
+ volume_mounts, deployment_description)
+ else:
+ _add_external_tls_init_container(ctx, init_containers, volumes, external_cert,
+ k8sconfig.get("external_cert"))
_add_cert_post_processor_init_container(ctx, init_containers, kwargs.get("tls_info") or {},
- k8sconfig.get("tls"), external_cert,
- k8sconfig.get("cert_post_processor"))
+ k8sconfig.get("tls"), external_cert,
+ k8sconfig.get(
+ "cert_post_processor"),cmpv2_integration_enabled)
# Create the container for the component
# Make it the first container in the pod
for service in deployment_description["services"]:
client.CoreV1Api().delete_namespaced_service(service, namespace)
+ for secret in deployment_description["secrets"]:
+ client.CoreV1Api().delete_namespaced_secret(secret, namespace)
+
+ for cert in deployment_description["certificates"]:
+ # client.CoreV1Api().delete_namespaced_service(service, namespace)
+ client.CustomObjectsApi().delete_namespaced_custom_object(
+ group="cert-manager.io",
+ version="v1",
+ name=cert,
+ namespace=namespace,
+ plural="certificates"
+ )
# Have k8s delete the underlying pods and replicaset when deleting the deployment.
options = client.V1DeleteOptions(propagation_policy="Foreground")
client.AppsV1Api().delete_namespaced_deployment(deployment_description["deployment"], namespace, body=options)
--- /dev/null
+# ============LICENSE_START=======================================================
+# org.onap.dcae
+# ================================================================================
+# Copyright (c) 2021 Nokia. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+
+from uritools import urisplit
+from fqdn import FQDN
+import validators
+from validators.utils import ValidationFailure
+
+
+class SansParser:
+ def parse_sans(self, sans):
+ """
+ Method for parsing sans. As input require SANs separated by comma (,)
+ Return Map with sorted SANs by type:
+ ips -> IPv4 or IPv6
+ dnss -> dns name
+ emails -> email
+ uris -> uri
+
+ Example usage:
+ SansParser().parse_sans("example.org,onap@onap.org,127.0.0.1,onap://cluster.local/")
+ Output: { "ips": [127.0.0.1],
+ "uris": [onap://cluster.local/],
+ "dnss": [example.org],
+ "emails": [onap@onap.org]}
+ """
+ sans_map = {"ips": [],
+ "uris": [],
+ "dnss": [],
+ "emails": []}
+ sans_arr = sans.split(",")
+ for san in sans_arr:
+ if self._is_email(san):
+ sans_map["emails"].append(san)
+ elif self._is_ip_v4(san) or self._is_ip_v6(san):
+ sans_map["ips"].append(san)
+ elif self._is_dns(san):
+ sans_map["dnss"].append(san)
+ elif self._is_uri(san):
+ sans_map["uris"].append(san)
+
+ return sans_map
+
+ def _is_email(self, san):
+ try:
+ return validators.email(san)
+ except ValidationFailure:
+ return False
+
+ def _is_ip_v4(self, san):
+ try:
+ return validators.ipv4(san)
+ except ValidationFailure:
+ return False
+
+ def _is_ip_v6(self, san):
+ try:
+ return validators.ipv6(san)
+ except ValidationFailure:
+ return False
+
+ def _is_uri(self, san):
+ parts = urisplit(san)
+ return parts.isuri()
+
+ def _is_dns(self, san):
+ fqdn = FQDN(san, min_labels=1)
+ return fqdn.is_valid
--- /dev/null
+# ============LICENSE_START=======================================================
+# org.onap.dcae
+# ================================================================================
+# Copyright (c) 2021 Nokia. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+
+# import pytest
+
+SAMPLE_SANS_INPUT = "example.org,test.onap.org,onap@onap.org,127.0.0.1,2001:0db8:85a3:0000:0000:8a2e:0370:7334,onap://cluster.local/"
+
+
+def test_parse_dns_name():
+ from k8sclient.sans_parser import SansParser
+ result = SansParser().parse_sans(SAMPLE_SANS_INPUT)
+ dnss_array = result["dnss"]
+ assert len(dnss_array) == 2
+ assert assert_item_in_list("example.org", dnss_array)
+
+
+def test_parse_ips():
+ from k8sclient.sans_parser import SansParser
+ result = SansParser().parse_sans(SAMPLE_SANS_INPUT)
+ ips_array = result["ips"]
+ assert len(ips_array) == 2
+ assert assert_item_in_list("127.0.0.1", ips_array)
+ assert assert_item_in_list("2001:0db8:85a3:0000:0000:8a2e:0370:7334", ips_array)
+
+
+def test_parse_emails():
+ from k8sclient.sans_parser import SansParser
+ result = SansParser().parse_sans(SAMPLE_SANS_INPUT)
+ emails_array = result["emails"]
+ assert len(emails_array) == 1
+ assert assert_item_in_list("onap@onap.org", emails_array)
+
+
+def test_parse_uri():
+ from k8sclient.sans_parser import SansParser
+ result = SansParser().parse_sans(SAMPLE_SANS_INPUT)
+ uris_array = result["uris"]
+ assert len(uris_array) == 1
+ assert assert_item_in_list("onap://cluster.local/", uris_array)
+
+
+def assert_item_in_list(item, list):
+ if item in list:
+ return True
+ else:
+ return False
Code Review / dcaegen2 / platform / plugins.git / commitdiff