Add signify-certificate-rotator image

.github/workflows/pull-build-rotate-signify-certificate.yml

@@ -0,0 +1,15 @@
+name: pull-build-rotate-signify-certificate
+#  description: "Build rotate-signify-certificate image for rotating signify certificates.
+on:
+  pull_request_target:
+    types: [ opened, edited, synchronize, reopened, ready_for_review ]
+    paths:
+      - "cmd/cloud-run/rotate-signify-certificate/**"
+
+jobs:
+  build-image:
+    uses: ./.github/workflows/image-builder.yml
+    with:
+      name: test-infra/rotatesignifycertificate
+      dockerfile: cmd/cloud-run/rotate-signify-certificate/Dockerfile
+      context: .

.github/workflows/push-build-rotate-signify-certificate.yml

@@ -0,0 +1,16 @@
+name: push-build-rotate-signify-certificate
+#  description: "Build rotate-signify-certificate image for rotating signify certificates.
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - "cmd/cloud-run/rotate-signify-certificate/**"
+
+jobs:
+  build-image:
+    uses: ./.github/workflows/image-builder.yml
+    with:
+      name: test-infra/rotatesignifycertificate
+      dockerfile: cmd/cloud-run/rotate-signify-certificate/Dockerfile
+      context: .

.gitignore

@@ -12,4 +12,4 @@ __debug_bin
 .terraform
 
 # Ignore generated credentials from google-github-actions/auth
-gha-creds-*.json
+gha-creds-*.json

cmd/cloud-run/signifysecretrotator/Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.12.6-slim
+
+# Allow statements and log messages to immediately appear in the Knative logs
+ENV PYTHONUNBUFFERED True
+
+WORKDIR /app
+
+COPY ./cmd/cloud-run/signifysecretrotator .
+COPY ./cmd/cloud-run/signifysecretrotator/requirements.txt .
+
+RUN pip3 install --upgrade -r requirements.txt && \
+	apt-get install ca-certificates
+
+CMD ["gunicorn", "--bind", "0.0.0.0:8080", "--workers", "4", "--timeout", "0", "signifysecretrotator:app"]

cmd/cloud-run/signifysecretrotator/pylogger/logger.py

@@ -0,0 +1,69 @@
+"""Encapsulates logging for cloud runs"""
+
+import json
+import logging
+from typing import Any, Dict
+
+from flask import Request
+
+
+class GoogleCloudFormatter(logging.Formatter):
+    """Wraps the formatting the logs using google cloud format"""
+
+    def __init__(
+        self, component_name: str, application_name: str, log_fields: Dict[str, Any]
+    ) -> None:
+        self.component_name: str = component_name
+        self.application_name: str = application_name
+        self.log_fields: Dict[str, Any] = log_fields
+
+        super().__init__()
+
+    def format(self, record: logging.LogRecord) -> str:
+        """Formats record into cloud event log"""
+
+        return json.dumps(
+            {
+                "timestamp": record.created,
+                "severity": record.levelname,
+                "message": record.getMessage(),
+            }
+        )
+
+
+def create_logger(
+    component_name: str,
+    application_name: str,
+    project_id: str = None,
+    request: Request = None,
+    log_level=logging.INFO,
+) -> logging.Logger:
+    """Creates instance of stdout logger for aplication's component"""
+    logger: logging.Logger = logging.getLogger(f"{application_name}/{component_name}")
+    logger.setLevel(log_level)
+
+    log_fields = {
+        "component": component_name,
+        "labels": {"io.kyma.component": application_name},
+    }
+
+    if request:
+        trace_header: str | None = request.headers.get("X-Cloud-Trace-Context")
+
+        if trace_header and project_id:
+            trace: list[str] = trace_header.split("/")
+            log_fields["logging.googleapi.com/trace"] = (
+                f"projects/{project_id}/traces/{trace[0]}"
+            )
+
+    formatter = GoogleCloudFormatter(
+        component_name=component_name,
+        application_name=application_name,
+        log_fields=log_fields,
+    )
+    handler = logging.StreamHandler()
+    handler.setFormatter(formatter)
+
+    logger.addHandler(handler)
+
+    return logger

cmd/cloud-run/signifysecretrotator/pylogger/test_logger.py

@@ -0,0 +1,106 @@
+"""Tests for logger module"""
+
+import time
+from typing import Any
+import unittest
+import logging
+import json
+from unittest.mock import Mock
+
+from flask import Request
+
+# pylint: disable=import-error
+# False positive see: https://github.com/pylint-dev/pylint/issues/3984
+from logger import GoogleCloudFormatter, create_logger
+
+
+class TestGoogleCloudFormatter(unittest.TestCase):
+    """Tests for custom formatter"""
+
+    def setUp(self):
+        self.component_name = "test_component"
+        self.application_name = "test_application"
+        self.log_fields = {"key1": "value1"}
+        self.formatter = GoogleCloudFormatter(
+            self.component_name, self.application_name, self.log_fields
+        )
+
+    def test_format(self) -> None:
+        """Test format function"""
+        log_record = logging.LogRecord(
+            name="test",
+            level=logging.INFO,
+            pathname="test_path",
+            lineno=10,
+            msg="This is a test message",
+            args=(),
+            exc_info=None,
+        )
+        formatted_log = self.formatter.format(log_record)
+        expected_log = {
+            "timestamp": log_record.created,
+            "severity": "INFO",
+            "message": "This is a test message",
+        }
+        self.assertEqual(json.loads(formatted_log), expected_log)
+
+
+class TestCreateLogger(unittest.TestCase):
+    """Tests for logger factory"""
+
+    def setUp(self):
+        # Ensure that each test has logger with unique name
+        self.component_name = f"test_component_{time.time()}"
+        self.application_name = "test_application"
+        self.project_id = "test_project_id"
+        self.request = Mock(spec=Request)
+        self.request.headers = {"X-Cloud-Trace-Context": "1234567890/other-info"}
+
+    def test_create_logger(self):
+        """Tests create logger with trace"""
+
+        logger: logging.Logger = create_logger(
+            component_name=self.component_name,
+            application_name=self.application_name,
+            project_id=self.project_id,
+            request=self.request,
+            log_level=logging.INFO,
+        )
+
+        self.assertFalse(logger.isEnabledFor(logging.DEBUG))
+
+        self.assertTrue(logger.hasHandlers())
+
+        handler = logger.handlers[0]
+        self.assertIsInstance(handler.formatter, GoogleCloudFormatter)
+
+        expected_log_fields: dict[str, Any] = {
+            "component": self.component_name,
+            "labels": {"io.kyma.component": self.application_name},
+            "logging.googleapi.com/trace": f"projects/{self.project_id}/traces/1234567890",
+        }
+        self.assertDictEqual(expected_log_fields, handler.formatter.log_fields)
+
+    def test_create_logger_without_trace(self):
+        """Tests create logger without request"""
+        logger: logging.Logger = create_logger(
+            component_name=self.component_name,
+            application_name=self.application_name,
+            log_level=logging.INFO,
+        )
+
+        self.assertFalse(logger.isEnabledFor(logging.DEBUG))
+        self.assertTrue(logger.hasHandlers())
+
+        handler = logger.handlers[0]
+        self.assertIsInstance(handler.formatter, GoogleCloudFormatter)
+
+        expected_log_fields: dict[str, Any] = {
+            "component": self.component_name,
+            "labels": {"io.kyma.component": self.application_name},
+        }
+        self.assertDictEqual(expected_log_fields, handler.formatter.log_fields)
+
+
+if __name__ == "__main__":
+    unittest.main()

cmd/cloud-run/signifysecretrotator/requirements.txt

@@ -0,0 +1,5 @@
+flask==2.3.2
+cloudevents==1.9.0
+gunicorn==22.0.0
+google-cloud-secret-manager==2.20.2
+cryptography==43.0.1

cmd/cloud-run/signifysecretrotator/secretmanager/client.py

@@ -0,0 +1,66 @@
+"""Custom wrapper for Google's Secret Manager Service client"""
+
+import json
+from typing import Any, Dict
+from google.cloud import secretmanager
+from google.api_core.exceptions import GoogleAPIError
+
+
+class SecretManagerClient:
+    """
+    Wraps the google's secret manager client implementation.
+    Provides more efficient way to retrieve and set secrets in kyma-project secret manager
+    """
+
+    def __init__(self, client=secretmanager.SecretManagerServiceClient()) -> None:
+        self.client: secretmanager.SecretManagerServiceClient = client
+
+    def get_secret(
+        self, secret_id: str, secret_version: str = "latest", is_json: bool = True
+    ) -> Dict[str, Any]:
+        """Fetches value of secret with given version
+
+        Args:
+            secret_id (str): Secret id in format "projects/<project_id>/secrets/<secret name>"
+            secret_version (str, optional): Version of the secret. Defaults to "latest".
+            is_json (bool): Secret is json struct. Defaults to True
+
+        Returns:
+            Dict[str, Any]: JSON decoded or str depending on is_json value
+        """
+
+        secret_name = f"{secret_id}/versions/{secret_version}"
+
+        try:
+            response: secretmanager.AccessSecretVersionResponse = (
+                self.client.access_secret_version(name=secret_name)
+            )
+            secret_value = response.payload.data.decode()
+
+            if is_json:
+                return json.loads(secret_value)
+
+            return secret_value
+        except GoogleAPIError as e:
+            raise SecretManagerError(secret_id, e) from e
+
+    def add_secret_version(self, secret_id: str, data: str) -> None:
+        """Adds new secret version with given data
+
+        Args:
+            secret_id (str): Secret id in format "projects/<project_id>/secrets/<secret name>"
+            data (str): Value that should be set as new secret version
+        """
+        payload = {"data": data.encode()}
+
+        try:
+            self.client.add_secret_version(parent=secret_id, payload=payload)
+        except GoogleAPIError as e:
+            raise SecretManagerError(secret_id, e) from e
+
+
+class SecretManagerError(Exception):
+    """Common class for Secret Manager client exceptions"""
+
+    def __init__(self, secret_id: str, e: Exception) -> None:
+        self.add_note(f"Failed to access secret {secret_id}, error: {e}")

cmd/cloud-run/signifysecretrotator/secretmanager/test_client.py

@@ -0,0 +1,87 @@
+"""Contains tests for secret manager client module"""
+
+import json
+import unittest
+from unittest.mock import AsyncMock, MagicMock, patch
+from google.cloud import secretmanager
+
+# pylint: disable=import-error
+# False positive see: https://github.com/pylint-dev/pylint/issues/3984
+from client import SecretManagerClient
+
+
+class TestSecretManagerClient(unittest.TestCase):
+    """Tests for secret manager client"""
+
+    def setUp(self) -> None:
+        access_secret_patcher = patch.object(
+            secretmanager.SecretManagerServiceClient, "access_secret_version"
+        )
+        add_secret_version_patcher = patch.object(
+            secretmanager.SecretManagerServiceClient, "add_secret_version"
+        )
+
+        self.mock_access_secret_version: MagicMock | AsyncMock = (
+            access_secret_patcher.start()
+        )
+        self.mock_add_secret_version: MagicMock | AsyncMock = (
+            add_secret_version_patcher.start()
+        )
+
+        self.addCleanup(access_secret_patcher.stop)
+        self.addCleanup(add_secret_version_patcher.stop)
+
+        self.client = SecretManagerClient()
+
+    def test_get_secret_json(self) -> None:
+        """Tests fetching json secret data"""
+        # Arrange
+        mock_response = MagicMock()
+        mock_response.payload.data.decode.return_value = json.dumps({"key": "value"})
+        self.mock_access_secret_version.return_value = mock_response
+
+        # Act
+        secret = self.client.get_secret("projects/test-project/secrets/test-secret")
+
+        # Assert
+        self.assertEqual(secret, {"key": "value"})
+        self.mock_access_secret_version.assert_called_once_with(
+            secret_name="projects/test-project/secrets/test-secret/versions/latest"
+        )
+
+    def test_get_secret_plain_string(self) -> None:
+        """Tests fetching string secret data"""
+        # Arrange
+        mock_response = MagicMock()
+        mock_response.payload.data.decode.return_value = "some-secret-value"
+        self.mock_access_secret_version.return_value = mock_response
+
+        # Act
+        secret = self.client.get_secret(
+            "projects/test-project/secrets/test-secret", is_json=False
+        )
+
+        # Assert
+        self.assertEqual(secret, "some-secret-value")
+        self.mock_access_secret_version.assert_called_once_with(
+            secret_name="projects/test-project/secrets/test-secret/versions/latest"
+        )
+
+    def test_add_secret_version(self) -> None:
+        """Tests setting a new secret version"""
+        # Arrange
+        secret_id = "projects/test-project/secrets/test-secret"
+        secret_data = "new-secret-value"
+
+        # Act
+        self.client.add_secret_version(secret_id, secret_data)
+
+        # Assert
+        payload: dict[str, bytes] = {"data": secret_data.encode()}
+        self.mock_add_secret_version.assert_called_once_with(
+            parent=secret_id, payload=payload
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()