sientia-dataops-opc-ingestor/ingestor/connection/certs.py

import asyncio
import os
from os import getenv
from pathlib import Path

from sientia_do.repository.minio_repository import MinioRepository

from ingestor.config import _env_bool

COLLECTION_NAME = 'platform_certificates'


class CertificateNotConfiguredError(RuntimeError):
    """No usable platform OPC client certificate is registered (Mongo + MinIO)."""


class CertProvider:
    """Downloads the platform's client cert + private key (Mongo metadata,
    MinIO bytes) into a writable cache dir — pod root FS is read-only (§12),
    so `cache_dir` must be an emptyDir/tmpfs mount.

    The certificate is single and shared by every connection (reads, writes,
    gRPC), so it is downloaded once and cached; `invalidate()` forces a fresh
    download (the pool calls it when a secured connection fails, e.g. after the
    certificate was replaced in MinIO). The lock serializes downloads so two
    connections never rewrite the cache files at the same time."""

    def __init__(self, mongo, minio, cache_dir: str, bucket: str, metadata: dict | None = None):
        self.mongo = mongo
        self.minio = minio
        self.cache_dir = cache_dir
        self.bucket = bucket
        self.metadata = metadata or {}
        self._lock = asyncio.Lock()
        self._paths: tuple[str, str] | None = None

    async def fetch(self) -> tuple[str, str]:
        async with self._lock:
            if self._paths is None:
                self._paths = await self._download_all()
            return self._paths

    def invalidate(self) -> None:
        self._paths = None

    async def _download_all(self) -> tuple[str, str]:
        doc = await self.mongo.find_one(COLLECTION_NAME, {}, self.metadata)
        try:
            cert_object = doc['client_cert']['minio_object_name']
            key_object = doc['private_key']['minio_object_name']
        except (KeyError, TypeError) as e:
            raise CertificateNotConfiguredError(
                f'No OPC client certificate configured: the Mongo collection '
                f"'{COLLECTION_NAME}' needs a document with client_cert.minio_object_name "
                f'and private_key.minio_object_name (objects in the MinIO bucket '
                f"'{self.bucket}'). Register the platform certificate before using "
                f'secured OPC servers.'
            ) from e
        Path(self.cache_dir).mkdir(parents=True, exist_ok=True)
        cert_path = await self._download(cert_object)
        key_path = await self._download(key_object)
        return cert_path, key_path

    async def _download(self, object_name: str) -> str:
        data = await self.minio.download_file(object_name, self.bucket, self.metadata)
        local_path = Path(self.cache_dir) / Path(object_name).name
        # Atomic replace: a connection loading the previous file never sees it half-written.
        tmp_path = local_path.with_name(local_path.name + '.tmp')
        tmp_path.write_bytes(data)
        os.replace(tmp_path, local_path)
        return str(local_path)


class LazyMinio:
    """Builds the MinIO client on first use. MinioRepository's constructor hits the
    network (bucket check), so building it eagerly would let a MinIO outage at pod
    start take the whole read path down. Built lazily, only the servers that need a
    certificate fail — and they are retried on the next loop."""

    def __init__(self, factory):
        self._factory = factory
        self._client = None

    async def download_file(self, object_name: str, bucket: str, metadata: dict):
        if self._client is None:
            self._client = self._factory()
        return await self._client.download_file(object_name, bucket, metadata)


def build_cert_provider(
    mongo, config, logger, notification_handler, metrics_controller, metadata=None
) -> CertProvider:
    """CertProvider shared by the read path (IngestorManager) and the write path
    (GatewayRuntime): one certificate source — MinIO + Mongo — for every connection."""

    def minio_factory() -> MinioRepository:
        return MinioRepository(
            endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'),
            access_key=getenv('MINIO_ACCESS_KEY', 'minioadmin'),
            secret_key=getenv('MINIO_SECRET_KEY', 'minioadmin'),
            secure=_env_bool('MINIO_SECURE', True),
            bucket=config.cert_bucket,
            logger=logger,
            notification_handler=notification_handler,
            metrics_controller=metrics_controller,
        )

    return CertProvider(
        mongo=mongo,
        minio=LazyMinio(minio_factory),
        cache_dir=config.cert_cache_dir,
        bucket=config.cert_bucket,
        metadata=metadata,
    )