LDAP Connections¶

LDAP Connection.

class freeiam.ldap.connection.Connection(uri: str | None = '', *, start_tls: bool = False, timeout: int = -1, automatic_reconnect: bool = True, max_connection_attempts: int = 10, retry_delay: float = 0.0, _hide_parent_exception: bool = True, _conn: SimpleLDAPObject | None = None)[source]¶

Bases: object

A LDAP Connection.

Variables:

uri (str) – The LDAP URI.
timelimit (int) – The global timelimit.
automatic_reconnect (bool) – Whether automatic reconnection is enabled.
max_connection_attempts (int) – number of connection attempt on connection loss.
retry_delay (float) – The retry delay (in seconds) between the reconnection attempts.

async abandon(msgid: int, *, controls: Controls | None = None) → Result[source]¶: Abandon a LDAP operation.

async add(dn: DN | str, attrs: dict[str, list[bytes]] | Attributes, *, controls: Controls | None = None) → Result[source]¶: Create a LDAP object.

async add_al(dn: DN | str, al: list[tuple[str, list[bytes]]], *, controls: Controls | None = None) → Result[source]¶: Create a LDAP object from addlist.

automatic_reconnect¶

async bind(authzid: str | None, password: str | None, *, controls: Controls | None = None) → Result[source]¶: Authenticate via plaintext credentials.

async bind_external() → None[source]¶: Authenticate via EXTERNAL method e.g. UNIX socket or TLS client certificate.

async bind_oauthbearer(authzid: str | None, token: str) → None[source]¶: Authenticate via OAuth 2.0 Access Token.

async bind_sasl_gssapi() → None[source]¶: Authenticate via GSSAPI e.g. via Kerberos ticket.

async cancel(msgid: int, *, controls: Controls | None = None) → bool[source]¶: Cancel a LDAP operation.

async change_password(dn: DN | str, old_password: str, new_password: str, *, controls: Controls | None = None) → Result[source]¶: Change password.

async compare(dn: DN | str, attr: str, value: bytes, *, controls: Controls | None = None) → bool[source]¶: Compare the value of an LDAP object.

async compare_dn(entry: DN | str, dn: DN | str) → bool[source]¶: Compare LDAP DN with existing entry.

property conn: SimpleLDAPObject¶: The underlying connection.

connect(fileno: bool | None = None) → None[source]¶: Connect to the LDAP server.

property connected: bool¶: Whether the connection is established.

async delete(dn: DN | str, *, controls: Controls | None = None) → Result[source]¶: Delete a LDAP object.

async delete_recursive(dn: DN | str, *, controls: Controls | None = None) → Result[source]¶: Delete a LDAP object recursively.

property dereference: int¶: Get the de-reference setting.

disconnect() → None[source]¶: Disconnect from LDAP server.

async exists(dn: DN | str, unique: bool = False, *, controls: Controls | None = None) → bool[source]¶: Check if LDAP object exists.

async extended(request: ExtendedRequest, response_class: type[ExtendedResponse] | None = None, *, controls: Controls | None = None) → Result[source]¶: Perform extended operation.

property fileno: int¶: Get the file descriptor number of the active socket connection.

property follow_referrals: bool | None¶: Follow referrals enabled.

async get(dn: DN | str, attrs: list[str] | None = None, filter_expr: str = '(objectClass=*)', *, unique: bool = False, controls: Controls | None = None) → Result[source]¶: Get a LDAP object.

async get_attr(dn: DN | str, attr: str, filter_expr: str = '(objectClass=*)', *, unique: bool = False, controls: Controls | None = None) → list[bytes][source]¶: Get attribute of an LDAP object.

classmethod get_global_option(option: Option | SASLOption | TLSOption | int) → OptionValue | TLSCRLCheck | TLSRequireCert | TLSProtocol | KeepAlive | Dereference | int | str[source]¶: Get a LDAP connection option.

async get_naming_contexts() → list[str][source]¶: Return namingContexts of Root DSE.

get_option(option: Literal[20503]) → int[source]¶
get_option(option: Option | SASLOption | TLSOption | int) → OptionValue | TLSCRLCheck | TLSRequireCert | TLSProtocol | KeepAlive | Dereference | int | str: Get a LDAP connection option.

get_result(conn: SimpleLDAPObject, msgid: int = ResponseType.Any, _all: int = 0, timeout: int = 0) → _Response[source]¶: Get the LDAP result for the given msgid.

async get_root_dse(attrs: list[str] | None = None, filter_expr: str = '(objectClass=*)') → Result[source]¶: Get Root DSE (Directory Server Entry).

async get_schema(subschema_dn: DN | str | None = None) → Schema[source]¶: Get LDAP Schema.

max_connection_attempts¶

async modify(dn: DN | str, oldattr: dict[str, list[bytes]] | Attributes, newattr: dict[str, list[bytes]] | Attributes, *, controls: Controls | None = None) → Result[source]¶: Modify a LDAP object.

async modify_ml(dn: DN | str, ml: list[tuple[int, str, list[bytes]]], *, controls: Controls | None = None) → Result[source]¶: Modify a LDAP object from modlist.

async modrdn(dn: DN | str, newrdn: DN | str, delete_old: bool = True, *, controls: Controls | None = None) → Result[source]¶: Rename a LDAP object.

async move(dn: DN | str, newposition: DN | str, *, controls: Controls | None = None) → Result[source]¶: Move a LDAP object.

property network_timeout: int¶: Get the LDAP network timeout.

property protocol_version: Version¶: Get the LDAP protocol version.

reconnect(*, force: bool = True) → None[source]¶: Reconnect to the LDAP server.

async refresh_ttl(dn: DN | str, ttl: int) → Result[source]¶: Perform Refresh extended operation.

async rename(dn: DN | str, newdn: DN | str, delete_old: bool = True, *, controls: Controls | None = None) → Result[source]¶: Rename a LDAP object.

request(operation: Callable[[...], int], *args: Any, **kwargs: Any) → int | None[source]¶: Make the LDAP request for the given operation.

retry_delay¶

async search(base: DN | str = '', scope: Scope = Scope.Subtree, filter_expr: str = '(objectClass=*)', attrs: list[str] | None = None, *, unique: bool = False, sizelimit: bool | None = None, sorting: list[str | tuple[str, str | None, bool]] | None = None, controls: Controls | None = None, _attrsonly: bool = False) → list[Result][source]¶: Search for DN and Attributes of LDAP objects.

async search_dn(base: DN | str = '', scope: Scope = Scope.Subtree, filter_expr: str = '(objectClass=*)', *, unique: bool = False, sizelimit: bool | None = None, sorting: list[str | tuple[str, str | None, bool]] | None = None, controls: Controls | None = None) → AsyncGenerator[DN, None][source]¶: Search for DNs of LDAP objects.

async search_iter(base: DN | str = '', scope: Scope = Scope.Subtree, filter_expr: str = '(objectClass=*)', attrs: list[str] | None = None, *, unique: bool = False, sizelimit: bool | None = None, sorting: list[str | tuple[str, str | None, bool]] | None = None, controls: Controls | None = None, _attrsonly: bool = False) → AsyncGenerator[Result, None][source]¶: Search iterative for DN and Attributes of LDAP objects.

async search_paged(base: DN | str = '', scope: Scope = Scope.Subtree, filter_expr: str = '(objectClass=*)', attrs: list[str] | None = None, page_size: int = 100, *, unique: bool = False, sizelimit: bool | None = None, sorting: list[str | tuple[str, str | None, bool]] | None = None, controls: Controls | None = None) → AsyncGenerator[Result, None][source]¶: Search paginated using SimplePagedResults control.

async search_paginated(base: DN | str = '', scope: Scope = Scope.Subtree, filter_expr: str = '(objectClass=*)', attrs: list[str] | None = None, *, page_size: int = 100, sorting: list[str | tuple[str, str | None, bool]], unique: bool = False, sizelimit: bool | None = None, controls: Controls | None = None) → AsyncGenerator[Result, None][source]¶: Search paginated using Virtual List View control.

set_controls(controls: Controls) → None[source]¶: Set LDAP controls for all operations on this connection.

classmethod set_global_option(option: Option | SASLOption | TLSOption | int, value: OptionValue | TLSCRLCheck | TLSRequireCert | TLSProtocol | KeepAlive | Dereference | int | str) → None[source]¶: Set a global LDAP option.

set_option(option: Literal[Option.ServerControls], value: Sequence[RequestControl]) → None[source]¶
set_option(option: Literal[Option.ClientControls], value: Sequence[RequestControl]) → None
set_option(option: Literal[Option.ProtocolVersion], value: int) → None
set_option(option: Option | SASLOption | TLSOption | int, value: OptionValue | TLSCRLCheck | TLSRequireCert | TLSProtocol | KeepAlive | Dereference | int | str | Sequence[RequestControl], *, append: bool = True) → None: Set a LDAP connection option.

classmethod set_tls(*, ca_certfile: str | None = None, ca_certdir: str | None = None, certfile: str | None = None, keyfile: str | None = None, require_cert: TLSRequireCert = TLSRequireCert.Demand, require_san: TLSRequireCert | None = None, minimum_protocol: TLSProtocol | None = None, cipher_suite: str | None = None, crlfile: None = None, crl_check: TLSCRLCheck | None = None) → None[source]¶: Set the TLS certificate settings globally.

property sizelimit: int¶: Get the sizelimit setting.

start_tls() → None[source]¶: Start TLS.

property timelimit: int¶: Get the LDAP time limit.

timeout¶

transaction(set_controls: bool = True) → AsyncGenerator[bytes, None][source]¶: Context manager to make a transaction, which is aborted on errors.

async unbind(*, controls: Controls | None = None) → Result | None[source]¶: Unbind.

uri¶

async whoami(*, controls: Controls | None = None) → DN | str | None[source]¶: Get authenticated user DN (authzid). “Who am I?” Operation.