Certificate implementation

Signed-off-by: Anton Kremenetsky <anton.kremenetsky@gmail.com>

Author: akremenetsky

etc/genesis_universal_agent/genesis_universal_agent.conf

@@ -11,11 +11,11 @@ connection_pool_size = 2
 [universal_agent]
 orch_endpoint = http://localhost:11011
 status_endpoint = http://localhost:11012
-caps_drivers = CoreCapabilityDriver,PasswordCapabilityDriver
+caps_drivers = CoreCapabilityDriver,PasswordCapabilityDriver,CoreDNSCertificateCapabilityDriver
 
 
 [universal_agent_scheduler]
-capabilities = em_core_*,password
+capabilities = em_core_*,password,certificate
 
 
 [CoreCapabilityDriver]
@@ -26,3 +26,4 @@ project_id = 12345678-c625-4fee-81d5-f691897b8142
 em_core_compute_nodes = /v1/nodes/
 em_core_config_configs = /v1/config/configs/
 em_core_secret_passwords = /v1/secret/passwords/
+em_core_secret_certificates = /v1/secret/certificates/

genesis_core/agent/universal/drivers/secret/backend/cert.py

@@ -0,0 +1,84 @@
+#    Copyright 2025 Genesis Corporation.
+#
+#    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.
+from __future__ import annotations
+
+import logging
+import typing as tp
+
+from gcl_sdk.agents.universal.dm import models
+from gcl_sdk.agents.universal.clients.backend import base
+from gcl_sdk.agents.universal.clients.backend import exceptions
+
+
+LOG = logging.getLogger(__name__)
+
+
+class CertBotBackendClient(base.AbstractBackendClient):
+    """Cert bot backend client."""
+
+    def __init__(self):
+        self._certs = []
+
+    def get(self, resource: models.Resource) -> dict[str, tp.Any]:
+        """Get the resource value in dictionary format."""
+        for cert in self._certs:
+            if cert["uuid"] == str(resource.uuid):
+                return cert
+
+        raise exceptions.ResourceNotFound(resource=resource)
+
+    def create(self, resource: models.Resource) -> dict[str, tp.Any]:
+        """Creates the resource. Returns the created resource."""
+        try:
+            self.get(resource)
+        except exceptions.ResourceNotFound:
+            pass
+        else:
+            raise exceptions.ResourceAlreadyExists(resource=resource)
+
+        value = resource.value.copy()
+        value["status"] = "ACTIVE"
+        value["key"] = "PRIVATE KEY"
+        value["cert"] = "CERTIFICATE"
+
+        self._certs.append(value)
+        return value
+
+    def update(self, resource: models.Resource) -> dict[str, tp.Any]:
+        """Update the resource. Returns the updated resource."""
+
+        # The simplest implementation. Update via recreation.
+        self.delete(resource)
+        return self.create(resource)
+
+    def list(self, kind: str, **kwargs) -> list[dict[str, tp.Any]]:
+        """Lists all resources by kind."""
+        return self._certs
+
+    def delete(self, resource: models.Resource) -> None:
+        """Delete the resource."""
+        try:
+            self.get(resource)
+        except exceptions.ResourceNotFound:
+            raise exceptions.ResourceNotFound(resource=resource)
+
+        for i, cert in enumerate(self._certs):
+            if cert["uuid"] == str(resource.uuid):
+                break
+        else:
+            raise exceptions.ResourceNotFound(resource=resource)
+
+        self._certs.pop(i)

genesis_core/agent/universal/drivers/secret/cert.py

@@ -0,0 +1,44 @@
+#    Copyright 2025 Genesis Corporation.
+#
+#    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.
+from __future__ import annotations
+
+import logging
+
+from gcl_sdk.agents.universal.drivers import direct
+from gcl_sdk.agents.universal.storage import fs
+
+from genesis_core.agent.universal.drivers.secret.backend import (
+    cert as cert_back,
+)
+
+
+LOG = logging.getLogger(__name__)
+
+AGENT_WORK_DIR = "/var/lib/genesis/universal_agent/"
+
+
+class CoreDNSCertificateCapabilityDriver(direct.DirectAgentDriver):
+    """Certificate capability driver."""
+
+    def __init__(self):
+        storage = fs.FileAgentStorage(AGENT_WORK_DIR, "cert_cap_storage.json")
+        client = cert_back.CertBotBackendClient()
+
+        super().__init__(storage=storage, client=client)
+
+    def get_capabilities(self) -> list[str]:
+        """Returns a list of capabilities supported by the driver."""
+        return ["certificate"]

genesis_core/secret/constants.py

@@ -18,6 +18,7 @@
 
 DEFAULT_SQL_LIMIT = 100
 PASSWORD_KIND = "password"
+CERTIFICATE_KIND = "certificate"
 
 
 class SecretStatus(str, enum.Enum):

genesis_core/secret/dm/models.py

@@ -27,43 +27,48 @@
 from genesis_core.secret import constants as sc
 
 
-class AbstractPasswordConstructor(
+class AbstractSecretConstructor(
     types_dynamic.AbstractKindModel, ra_models.SimpleViewMixin
 ):
 
-    def build(self, plain_password: str) -> str:
+    def build(self, plain_secret: str) -> str:
         raise NotImplementedError()
 
 
-class PlainPasswordConstructor(AbstractPasswordConstructor):
+class PlainSecretConstructor(AbstractSecretConstructor):
     KIND = "plain"
 
-    def build(self, plain_password: str) -> str:
-        return plain_password
+    def build(self, plain_secret: str) -> str:
+        return plain_secret
 
 
-class Password(
+class Secret(
     cm.ModelWithFullAsset,
-    orm.SQLStorableMixin,
     ua_models.TargetResourceMixin,
-    ua_models.TargetResourceSQLStorableMixin,
 ):
-    __tablename__ = "secret_passwords"
-
-    method = properties.property(
-        types.Enum([s.value for s in sc.SecretMethod]),
-        default=sc.SecretMethod.AUTO_HEX.value,
-    )
     status = properties.property(
         types.Enum([s.value for s in sc.SecretStatus]),
         default=sc.SecretStatus.NEW.value,
     )
     constructor = properties.property(
         types_dynamic.KindModelSelectorType(
-            types_dynamic.KindModelType(PlainPasswordConstructor),
+            types_dynamic.KindModelType(PlainSecretConstructor),
         ),
         required=True,
-        default=PlainPasswordConstructor,
+        default=PlainSecretConstructor,
+    )
+
+
+class Password(
+    Secret,
+    orm.SQLStorableMixin,
+    ua_models.TargetResourceSQLStorableMixin,
+):
+    __tablename__ = "secret_passwords"
+
+    method = properties.property(
+        types.Enum([s.value for s in sc.SecretMethod]),
+        default=sc.SecretMethod.AUTO_HEX.value,
     )
     value = properties.property(
         types.AllowNone(types.String(min_length=1, max_length=512)),
@@ -105,3 +110,87 @@ def get_deleted_passwords(
         return cls.get_deleted_target_resources(
             cls.__tablename__, sc.PASSWORD_KIND, limit
         )
+
+
+class AbstractCertificateMethod(
+    types_dynamic.AbstractKindModel, ra_models.SimpleViewMixin
+):
+    pass
+
+
+class DNSCoreCertificateMethod(AbstractCertificateMethod):
+    KIND = "dns_core"
+
+
+class Certificate(
+    Secret,
+    orm.SQLStorableWithJSONFieldsMixin,
+    ua_models.TargetResourceSQLStorableMixin,
+):
+    __tablename__ = "secret_certificates"
+    __jsonfields__ = ["domains"]
+
+    method = properties.property(
+        types_dynamic.KindModelSelectorType(
+            types_dynamic.KindModelType(DNSCoreCertificateMethod),
+        ),
+        required=True,
+        default=DNSCoreCertificateMethod,
+    )
+    valid_until = properties.property(
+        types.AllowNone(types.UTCDateTimeZ()),
+        default=None,
+    )
+    email = properties.property(types.Email())
+    domains = properties.property(
+        types.TypedList(types.String()),
+        required=True,
+    )
+    key = properties.property(
+        types.AllowNone(types.String(min_length=1, max_length=10240)),
+        default=None,
+    )
+    cert = properties.property(
+        types.AllowNone(types.String(min_length=1, max_length=10240)),
+        default=None,
+    )
+
+    def get_resource_target_fields(self) -> set[str]:
+        """Return the collection of target fields.
+
+        Refer to the Resource model for more details about target fields.
+        """
+        return {
+            "method",
+            "email",
+            "domains",
+            "constructor",
+            "name",
+            "description",
+            "project_id",
+            "uuid",
+        }
+
+    @classmethod
+    def get_new_certificates(
+        cls, limit: int = sc.DEFAULT_SQL_LIMIT
+    ) -> list["Certificate"]:
+        return cls.get_new_entities(
+            cls.__tablename__, sc.CERTIFICATE_KIND, limit
+        )
+
+    @classmethod
+    def get_updated_certificates(
+        cls, limit: int = sc.DEFAULT_SQL_LIMIT
+    ) -> list["Certificate"]:
+        return cls.get_updated_entities(
+            cls.__tablename__, sc.CERTIFICATE_KIND, limit
+        )
+
+    @classmethod
+    def get_deleted_certificates(
+        cls, limit: int = sc.DEFAULT_SQL_LIMIT
+    ) -> list[ua_models.TargetResource]:
+        return cls.get_deleted_target_resources(
+            cls.__tablename__, sc.CERTIFICATE_KIND, limit
+        )