DataOps-platform/app/core/connectors/secrets.py

"""Deployable, fail-closed connector secret reference resolution."""

from __future__ import annotations

import os
import re
from collections.abc import Mapping

from app.core.connectors.errors import ConnectorConfigurationError

ENV_REFERENCE = re.compile(r"^env:(DATAOPS_CONNECTOR_[A-Z0-9_]{1,200})$")


class EnvironmentSecretResolver:
    """Resolve only the dedicated DATAOPS_CONNECTOR_* environment namespace."""

    def __init__(self, environ: Mapping[str, str] | None = None):
        self.environ = os.environ if environ is None else environ

    def __call__(self, reference: str) -> str:
        value = str(reference or "")
        match = ENV_REFERENCE.fullmatch(value)
        if match:
            secret = self.environ.get(match.group(1))
            if not isinstance(secret, str) or not secret.strip():
                raise ConnectorConfigurationError(
                    "connector environment secret is unavailable"
                )
            return secret
        if value.startswith(("vault:", "secret:")):
            raise ConnectorConfigurationError(
                "connector secret backend is not configured"
            )
        raise ConnectorConfigurationError(
            "connector environment secret reference is not allowed"
        )


__all__ = ["EnvironmentSecretResolver", "ENV_REFERENCE"]