def __connect_to_activedirectory()

in ad-joining/register-computer/main.py [0:0]

• 24 lines of code
• 9 McCabe index (conditional complexity)

def __connect_to_activedirectory(ad_site=None):
    domain = __read_required_setting("AD_DOMAIN")

    if "AD_DOMAINCONTROLLER" in os.environ:
        # Use explicitly defined DC.
        domain_controllers = [os.environ["AD_DOMAINCONTROLLER"]]
    else:
        # Look up DC in DNS.
        domain_controllers = ad.domain.ActiveDirectoryConnection.locate_domain_controllers(
            domain, ad_site)

    # Determine if LDAPS should be used for Active Directory connection
    # Environmental variable stores strings convert to bool
    use_ldaps = __read_setting("USE_LDAPS")
    use_ldaps = False if use_ldaps is None else use_ldaps.lower() == "true"

    certificate_data = None
    if use_ldaps:
        # Retrieve certificate data from Secret Manager
        certificate_data = __read_certificate_data()

    # If we used SRV records to look up domain controllers, then it is possible that
    # the highest-priority one is offline. So loop over the records to fine one
    # that works.
    for dc in domain_controllers:
        try:
            return ad.domain.ActiveDirectoryConnection.connect(
                    dc,
                    ",".join(["DC=" + dc for dc in domain.split(".")]),
                    __read_required_setting("AD_USERNAME"),
                    __read_ad_password(),
                    use_ldaps,
                    certificate_data)
        except Exception as e:
            logging.exception("Failed to connect to DC '%s': %s" % (dc, e))

    raise ad.domain.LdapException("No more DCs left to try")