Protocol Types¶

def get_enabled_protocols(self) -> list[Protocoltypee]:
    """Gibt lis the enablethe protocole torück.

    Returns:
        lis the enablethe Protocoltypee Enums
    """
    enabled = []
    if self.rpc_enabled:
        enabled.append(Protocoltypee.RPC)
    if self.stream_enabled:
        enabled.append(Protocoltypee.STREAM)
    if self.bus_enabled:
        enabled.append(Protocoltypee.BUS)
    if self.mcp_enabled:
        enabled.append(Protocoltypee.MCP)
    return enabled

def is_protocol_enabled(self, protocol: Protocoltypee) -> bool:
    """Checks ob a specifics protocol enabled is.

    Args:
        protocol: protocol to check

    Returns:
        True if protocol enabled is
    """
    protocol_map = {
        Protocoltypee.RPC: self.rpc_enabled,
        Protocoltypee.STREAM: self.stream_enabled,
        Protocoltypee.BUS: self.bus_enabled,
        Protocoltypee.MCP: self.mcp_enabled,
    }
    return protocol_map.get(protocol, False)

def get_endpoint(self, protocol: Protocoltypee) -> str:
    """Gibt Endpunkt for specifics protocol torück.

    Args:
        protocol: protocol for the the Endpunkt benötigt is

    Returns:
        API-Endpunkt for the protocol

    Raises:
        ValueError: If protocol not supported is
    """
    endpoint_map = {
        Protocoltypee.RPC: self.rpc_endpoint,
        Protocoltypee.STREAM: self.stream_endpoint,
        Protocoltypee.BUS: self.bus_endpoint,
        Protocoltypee.MCP: self.mcp_endpoint,
    }

    if protocol not in endpoint_map:
        raise ValueError(f"Unknown protocol: {protocol}")

    return endpoint_map[protocol]

def validate(self) -> None:
    """Validates the security configuration using Pydantic models.

    Raises:
        ValidationError: On ungültiger configuration
    """
    try:
        # Convert to dict for validation
        config_dict = {
            "auth_type": self.auth_type.value
            if hasattr(self.auth_type, "value")
            else str(self.auth_type),
            "api_token": self.api_token,
            "oidc_issuer": self.oidc_issuer,
            "oidc_client_id": self.oidc_client_id,
            "oidc_client_secret": self.oidc_client_secret,
            "oidc_scope": self.oidc_scope,
            "mtls_cert_path": self.mtls_cert_path,
            "mtls_key_path": self.mtls_key_path,
            "mtls_ca_path": self.mtls_ca_path,
            "rbac_enabled": self.rbac_enabled,
            "audit_enabled": self.audit_enabled,
            "token_refresh_enabled": self.token_refresh_enabled,
            "token_cache_ttl": self.token_cache_ttl,
        }

        # Validate using Pydantic model
        validate_configuration(config_dict, "security")

    except ValidationError:
        raise
    except Exception as e:
        raise ValidationError(
            f"Security configuration validation failed: {e}"
        ) from e

def is_token_based(self) -> bool:
    """Checks ob Token-basierte authentication verwendet is.

    Returns:
        True if Bearer or OIDC verwendet is
    """
    return self.auth_type in [Authtypee.BEARER, Authtypee.OIDC]

def requires_refresh(self) -> bool:
    """Checks ob Token-Refresh erforthelich is.

    Returns:
        True if Token-Refresh enabled and erforthelich is
    """
    return self.token_refresh_enabled and self.is_token_based()