Python - MTLS Changes

charts/eric-oss-hello-world-python-app/templates/_helpers.tpl

@@ -341,4 +341,21 @@ Define the annotations for security policy
 */}}
 {{- define "eric-oss-hello-world-python-app.securityPolicy.annotations" -}}
 # Automatically generated annotations for documentation purposes.
-{{- end -}}
+{{- end -}}
+
+{{/*
+Define the function to get the secret name
+ */}}
+{{- define "eric-oss-hello-world-python-app.clientSecret" -}}
+{{- $clientSecret := "" -}}
+{{- if .Values.global }}
+  {{- if .Values.global.clientCredentials }}
+    {{- if .Values.global.clientCredentials.secret }}
+      {{- if .Values.global.clientCredentials.secret.name }}
+        {{- $clientSecret = .Values.global.clientCredentials.secret.name }}
+      {{- end }}
+    {{- end }}
+  {{- end }}
+{{- end }}
+{{- print $clientSecret }}
+{{- end }}

charts/eric-oss-hello-world-python-app/templates/deployment/deployment.yaml

@@ -63,6 +63,10 @@ spec:
           secret:
             secretName: {{ index .Values "appSecretName" | quote }}
             defaultMode: 420
+        - name: client-creds
+          secret:
+            secretName: {{ include "eric-oss-hello-world-python-app.clientSecret" . | quote }}
+            defaultMode: 420
       containers:
         - name: eric-oss-hello-world-python-app
           image: {{ template "eric-oss-hello-world-python-app.imagePath" (dict "imageId" "eric-oss-hello-world-python-app" "values" .Values "files" .Files) }}
@@ -89,6 +93,9 @@ spec:
             - name: app-certs
               mountPath: {{ index .Values "appCertMountPath" | default .Values.instantiationDefaults.appCertMountPath | quote }}
               readOnly: true
+            - name: client-creds
+              mountPath: {{ index .Values "clientCredsMountPath" | default .Values.instantiationDefaults.clientCredsMountPath | quote }}
+              readOnly: true
           env:
             - name: IAM_CLIENT_ID
               value: {{ index .Values "clientId" | quote }}
@@ -108,6 +115,10 @@ spec:
               value: {{ index .Values "appCertFileName" | quote }}
             - name: APP_CERT_FILE_PATH
               value: {{ index .Values "appCertMountPath" | default .Values.instantiationDefaults.appCertMountPath | quote }}
+            - name: CLIENT_CREDS_FILE_PATH
+              value: {{ index .Values "clientCredsMountPath" | default .Values.instantiationDefaults.clientCredsMountPath | quote }}
+            - name: CLIENT_ID_FILE_NAME
+              value: {{ .Values.global.clientCredentials.secret.clientIdKey | quote }}
             - name: SERVICE_NAME
               value: {{ .Chart.Name }}
             - name: CONTAINER_NAME

charts/eric-oss-hello-world-python-app/values.yaml

@@ -126,3 +126,10 @@ podPriority:
 instantiationDefaults:
   platformCaCertMountPath: "/etc/tls-ca/platform/"
   appCertMountPath: "/etc/tls/log/"
+  clientCredsMountPath: "/etc/client-creds/"
+
+global:
+  clientCredentials:
+    secret:
+      clientIdKey: "clientId"
+      name: "<instance id>-cc"

csar/Definitions/AppDescriptor.yaml

@@ -3,8 +3,12 @@ Description of an APP:
     APPName: eric-oss-hello-world-python-app
     APPVersion: VERSION
     APPType: rApp
-APPComponent:
-    NameofComponent: eric-oss-hello-world-python-app
-    Version: VERSION
-    Path: OtherDefinitions/ASD/eric-oss-hello-world-python-appASD.yaml
-    ArtefactType: ASD
+AppComponentList:
+    - NameofComponent: eric-oss-hello-world-python-app
+      Version: VERSION
+      Path: OtherDefinitions/ASD/eric-oss-hello-world-python-appASD.yaml
+      ArtefactType: ASD
+    - NameofComponent: security-mgmt
+      Version: 1.0.0
+      Path: OtherDefinitions/SecurityManagement
+      ArtefactType: SecurityManagement

csar/OtherDefinitions/SecurityManagement/security-metadata.json

@@ -0,0 +1,3 @@
+{
+	"authenticatorType": "client-x509"
+}

eric-oss-hello-world-python-app/config.py

@@ -1,38 +1,40 @@
-'''This module handles environment variables'''
+"""This module handles environment variables"""
+
 import os
 
+
 def get_config():
-    '''
-    get env and return config with all env vals required
-    '''
-    iam_client_id           = get_os_env_string("IAM_CLIENT_ID", "")
-    iam_client_secret       = get_os_env_string("IAM_CLIENT_SECRET", "")
-    iam_base_url            = get_os_env_string("IAM_BASE_URL", "")
-    ca_cert_file_name       = get_os_env_string("CA_CERT_FILE_NAME", "")
-    ca_cert_file_path       = get_os_env_string("CA_CERT_FILE_PATH", "")
-    log_ctrl_file           = get_os_env_string("LOG_CTRL_FILE", "")
-    log_endpoint            = get_os_env_string("LOG_ENDPOINT", "")
-    app_key                 = get_os_env_string("APP_KEY", "")
-    app_cert                = get_os_env_string("APP_CERT", "")
-    app_cert_file_path      = get_os_env_string("APP_CERT_FILE_PATH", "")
+    """get env and return config with all env vals required"""
+    iam_client_id = get_os_env_string("IAM_CLIENT_ID", "")
+    iam_client_secret = get_os_env_string("IAM_CLIENT_SECRET", "")
+    iam_base_url = get_os_env_string("IAM_BASE_URL", "")
+    ca_cert_file_name = get_os_env_string("CA_CERT_FILE_NAME", "")
+    ca_cert_file_path = get_os_env_string("CA_CERT_FILE_PATH", "")
+    log_ctrl_file = get_os_env_string("LOG_CTRL_FILE", "")
+    log_endpoint = get_os_env_string("LOG_ENDPOINT", "")
+    app_key = get_os_env_string("APP_KEY", "")
+    app_cert = get_os_env_string("APP_CERT", "")
+    app_cert_file_path = get_os_env_string("APP_CERT_FILE_PATH", "")
+    client_creds_file_path = get_os_env_string("CLIENT_CREDS_FILE_PATH", "")
+    client_id_file_name = get_os_env_string("CLIENT_ID_FILE_NAME", "")
 
     config = {
-            "iam_client_id": iam_client_id,
-            "iam_client_secret": iam_client_secret,
-            "iam_base_url": iam_base_url,
-            "ca_cert_file_name": ca_cert_file_name,
-            "ca_cert_file_path": ca_cert_file_path,
-            "log_ctrl_file": log_ctrl_file,
-            "log_endpoint": log_endpoint,
-            "app_key": app_key,
-            "app_cert": app_cert,
-            "app_cert_file_path": app_cert_file_path
+        "iam_client_id": iam_client_id,
+        "iam_client_secret": iam_client_secret,
+        "iam_base_url": iam_base_url,
+        "ca_cert_file_name": ca_cert_file_name,
+        "ca_cert_file_path": ca_cert_file_path,
+        "log_ctrl_file": log_ctrl_file,
+        "log_endpoint": log_endpoint,
+        "app_key": app_key,
+        "app_cert": app_cert,
+        "app_cert_file_path": app_cert_file_path,
+        "client_creds_file_path": client_creds_file_path,
+        "client_id_file_name": client_id_file_name,
     }
     return config
 
 
 def get_os_env_string(env_name, default_value):
-    '''
-    get env
-    '''
+    """get env"""
     return os.getenv(env_name, default_value).strip()

eric-oss-hello-world-python-app/login.py

@@ -1,52 +1,65 @@
-'''
+"""
 This module performs client credentials grant authentication
 by sending HTTP requests with TLS and with required environment
 variables.
- '''
+ """
+
 import os
 from urllib.parse import urljoin
 import json
+import time
 import requests
 from config import get_config
 
+
 class LoginError(Exception):
     """Raised when EIC login fails"""
 
 def login():
-    '''
+    """
     Get bearer token for accessing platform REST APIs:
     https://developer.intelligentautomationplatform.ericsson.net/#tutorials/app-authentication
-    '''
+    """
     config = get_config()
     login_path = "/auth/realms/master/protocol/openid-connect/token"
     login_url = urljoin(config.get("iam_base_url"), login_path)
-    headers = {
-        "Content-Type": "application/x-www-form-urlencoded"
-    }
-    form_data = {
-        "grant_type": "client_credentials",
-        "client_id": config.get("iam_client_id"),
-        "client_secret": config.get("iam_client_secret"),
-        "tenant_id": "master"
-    }
-    try:
-        resp = tls_login(login_url, form_data, headers)
-    except LoginError:
-        return None, 0
-
-    resp = json.loads(resp.decode('utf-8'))
+    headers = {"Content-Type": "application/x-www-form-urlencoded"}
+    resp = tls_login(login_url, headers)
+    resp = json.loads(resp.decode("utf-8"))
     token, time_until_expiry = resp["access_token"], resp["expires_in"]
-    time_until_expiry -= 10 # add a buffer to ensure our session doesn't expire mid-request
-    return token, time_until_expiry
+    time_until_expiry -= (
+        10  # add a buffer to ensure our session doesn't expire mid-request
+    )
+    return token, time.time() + time_until_expiry
 
-def tls_login(url, form_data, headers):
-    '''
-    This function sends an HTTP POST request with TLS for the login operation
-    '''
+
+def tls_login(url, headers):
+    """This function sends an HTTP POST request with TLS for the login operation"""
     config = get_config()
-    cert = os.path.join("/", config.get("ca_cert_file_path"), config.get("ca_cert_file_name"))
+    ca_cert = os.path.join(
+        "/", config.get("ca_cert_file_path"), config.get("ca_cert_file_name")
+    )
+    app_cert = os.path.join(
+        "/", config.get("app_cert_file_path"), config.get("app_cert")
+    )
+    app_key = os.path.join(
+        "/", config.get("app_cert_file_path"), config.get("app_key"))
+    client_id_path = os.path.join(
+        "/", config.get("client_creds_file_path"), config.get("client_id_file_name")
+    )
+    form_data = {"grant_type": "client_credentials", "tenant_id": "master"}
+    cert = (app_cert, app_key)
+
+    try:
+        with open(client_id_path, "r") as f:
+            form_data["client_id"] = f.read().strip()
+    except OSError as e:
+        raise LoginError(f"Error while reading client id: {e}")
+
     try:
-        response = requests.post(url, data=form_data, headers = headers, timeout=5, verify=cert)
+        response = requests.post(
+            url, data=form_data, headers=headers, timeout=5, verify=ca_cert, cert=cert
+        )
         if response.status_code != 200:
             raise LoginError(f"Login failed ({response.status_code})")
     except Exception as exception:

eric-oss-hello-world-python-app/main.py

@@ -1,82 +1,96 @@
 #!/usr/bin/env python3
-'''
+"""
 Flask Application for Hello World Service
 
 This Python script defines a Flask application that implements a simple "Hello World" service
 along with a health check and metrics endpoints.
-'''
+"""
 import time
 from flask import abort
 from flask import Flask
 from login import login
 from mtls_logging import MtlsLogging, Severity
 from werkzeug.middleware.dispatcher import DispatcherMiddleware
-from prometheus_client import disable_created_metrics, make_wsgi_app, CollectorRegistry, Counter
+from prometheus_client import (
+    disable_created_metrics,
+    make_wsgi_app,
+    CollectorRegistry,
+    Counter,
+)
 
 SERVICE_PREFIX = "python_hello_world"
 
 class Application(Flask):
-    '''The Flask application itself. Subclassed for testing.'''
+    """The Flask application itself. Subclassed for testing."""
     def __init__(self):
         super().__init__(__name__)
         disable_created_metrics()
         self.counters = {"total_failed": 0, "total_requests": 0}
         self.session = {"token": None, "expiry_time": 0}
         self.create_metrics()
-        self.wsgi_app = DispatcherMiddleware(self.wsgi_app, {
-            '/sample-app/python/metrics': make_wsgi_app(registry=self.registry)
-        })
+        self.wsgi_app = DispatcherMiddleware(
+            self.wsgi_app,
+            {"/sample-app/python/metrics": make_wsgi_app(registry=self.registry)},
+        )
         self.logger = MtlsLogging()
 
         @self.route("/sample-app/python/")
         def root():
-            '''This route returns a 400 Bad Request HTTP response.'''
-            self.logger.log("400 Bad request: User tried accessing '/sample-app/python/'", Severity.INFO)
+            """This route returns a 400 Bad Request HTTP response."""
+            self.logger.log(
+                "400 Bad request: User tried accessing '/sample-app/python/'",
+                Severity.INFO,
+            )
             abort(400)
 
         @self.route("/sample-app/python/hello")
         def hello():
-            '''
+            """
             This route performs a login operation and returns
             a simple "Hello World!" greeting and increments the
             total request counter.
-            '''
+            """
             self.update_session()
             self.requests_total.inc()
             self.logger.log("200 OK: Hello World!", Severity.INFO)
             return "Hello World!\n"
 
         @self.route("/sample-app/python/health")
         def health():
-            '''
+            """
             This route provides a simple health check endpoint, returning "Ok" to
             indicate that the application is healthy.
-            '''
+            """
             self.update_session()
             self.logger.log("200 OK: Health check", Severity.INFO)
             return "Ok\n"
 
-
     def update_session(self):
-        '''Refresh session if it expires.'''
+        """Refresh session if it expires."""
         if int(time.time()) >= self.session["expiry_time"]:
-            self.session["token"], self.session["expiry_time"] = login()
-        if not self.session["token"]:
-            # since the token isn't used for anything,
-            # this is just a WARNING level log instead of ERROR
-            self.logger.log("Login failed", Severity.WARNING)
+            try:
+                self.session["token"], self.session["expiry_time"] = login()
+            except Exception as e:
+                # since the token isn't used for anything,
+                # this is just a WARNING level log instead of ERROR
+                self.logger.log(f"Login failed: {e}", Severity.WARNING)
+
     def create_metrics(self):
         self.registry = CollectorRegistry()
-        self.requests_total = Counter(namespace=SERVICE_PREFIX,
-                                      name="requests_total",
-                                      documentation="Total number of API requests")
-        self.requests_failed = Counter(namespace=SERVICE_PREFIX,
-                                       name="requests_failed_total",
-                                       documentation="Total number of API request failures")
+        self.requests_total = Counter(
+            namespace=SERVICE_PREFIX,
+            name="requests_total",
+            documentation="Total number of API requests",
+        )
+        self.requests_failed = Counter(
+            namespace=SERVICE_PREFIX,
+            name="requests_failed_total",
+            documentation="Total number of API request failures",
+        )
         self.registry.register(self.requests_total)
         self.registry.register(self.requests_failed)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     instance = Application()
-    instance.run(host = '0.0.0.0', port = '8050')
+    instance.run(host="0.0.0.0", port="8050")