LDAPConnector

class LDAPConnector:
    """Intended for providers that connect to LDAP directories. This class
    provides methods to establish and manage a connection to an LDAP server,
    including connecting, disconnecting, and checking the connection status. It
    also allows retrieval of the current connection and server information.
    """

    def __init__(
        self,
        server_url: str,
        bind_dn: str,
        bind_password: str,
        server_port: int = 636,
        use_tls: bool = True,
        client_strategy: ClientStrategyType = SYNC,
        connect_timeout: float | None = 5.0,
        additional_connector_params: dict[str, Any] | None = None,
        additional_server_params: dict[str, Any] | None = None,
    ) -> None:
        """
        Parameters
        ----------
        server_url
            URL of the LDAP server.
        bind_dn
            Distinguished Name (DN) for binding to the LDAP server.
        bind_password
            Password for the bind DN.
        use_tls
            Whether to use TLS for the connection.
        server_port
            Port of the LDAP server.
        client_strategy
            The client strategy to use for the connection. This can be one of the
            following values:

            - 'SYNC': Synchronous strategy (default).
            - 'SAFE_RESTARTABLE': Safe restartable strategy.
            - 'SAFE_SYNC': Safe synchronous strategy.
            - 'ASYNC': Asynchronous strategy.
            - 'LDIF': LDIF strategy.
            - 'RESTARTABLE': Restartable strategy.
            - 'REUSABLE': Reusable strategy.
            - 'MOCK_SYNC': Mock synchronous strategy.
            - 'MOCK_ASYNC': Mock asynchronous strategy.
            - 'ASYNC_STREAM': Asynchronous stream strategy.
        connect_timeout
            Maximum number of seconds to wait while opening the socket.
        additional_connector_params
            Additional parameters to pass to the LDAP connection, by default
            {"receive_timeout": 5}
        additional_server_params
            Additional parameters to pass to the LDAP server, by default None
        """
        self._server_url = server_url
        self._bind_dn = bind_dn
        self._bind_password = bind_password
        self._client_strategy = client_strategy
        self._connect_timeout = connect_timeout
        self._additional_connector_params: dict[str, Any] = (
            {"receive_timeout": 5}
            if additional_connector_params is None
            else dict(additional_connector_params)
        )
        self._additional_server_params: dict[str, Any] = (
            {}
            if additional_server_params is None
            else dict(additional_server_params)
        )
        tls = None
        if use_tls:
            tls = Tls(
                validate=ssl.CERT_REQUIRED,
                version=ssl.PROTOCOL_TLS_CLIENT,
            )

        self._server = Server(
            host=self._server_url,
            port=server_port,
            use_ssl=use_tls,
            tls=tls,
            get_info=ALL,
            connect_timeout=self._connect_timeout,
            **self._additional_server_params,
        )
        self._connection: Connection | None = None

    @property
    def connection_cls(self) -> type[Connection]:
        """Get the connection class."""
        return Connection

    @property
    def bind_dn(self) -> str:
        """Get the bind DN."""
        return self._bind_dn

    def connect(self) -> None:
        """Method to establish a connection to the LDAP server."""
        self._connection = self.connection_cls(
            self._server,
            user=self._bind_dn,
            password=self._bind_password,
            client_strategy=self._client_strategy,  # type: ignore
            auto_bind=True,
            **self._additional_connector_params,
        )

    def disconnect(self) -> None:
        """Method to close the connection to the LDAP server."""
        if self.is_connected():
            cast(Any, self._connection).unbind()
        self._connection = None

    def is_connected(self) -> bool:
        """Check if the connection to the LDAP server is active."""
        return self._connection is not None and self._connection.bound

    def get_connection(self) -> Connection:
        """Get the current LDAP connection.

        Returns
        -------
        Connection
            The active LDAP connection.

        Raises
        ------
        RuntimeError
            If the connection is not established.
        """
        if not self._connection or not self.is_connected():
            raise RuntimeError("LDAP connection is not established.")
        return self._connection

    def get_server(self) -> Server:
        """Get the LDAP server."""
        return self._server