IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2025-02-25 18:55:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

Use new LDAPClient constructors

Browse Source 
Replace get_ldap_uri() + LDAPClient() with new LDAPClient constructors
like LDAPClient.from_realm().

Some places now use LDAPI with external bind instead of LDAP with simple
bind. Although the FQDN *should* resolve to 127.0.0.1 / [::1], there is
no hard guarantee. The draft
https://tools.ietf.org/html/draft-west-let-localhost-be-localhost-04#section-5.1
specifies that applications must verify that the resulting IP is a
loopback API. LDAPI is always local and a bit more efficient, too.

The simple_bind() method also prevents the caller from sending a
password over an insecure line.

Signed-off-by: Christian Heimes <cheimes@redhat.com>
Reviewed-By: Rob Crittenden <rcritten@redhat.com>
This commit is contained in:
Christian Heimes 2018-11-30 10:28:32 +01:00 committed by Rob Crittenden
parent 5be9341fba
commit a3934a211d
12 changed files with 55 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
install/migration/migration.py
View File

@ -54,8 +54,11 @@ def bind(ldap_uri, base_dn, username, password):
logger.error('migration unable to get base dn')
raise IOError(errno.EIO, 'Cannot get Base DN')
bind_dn = DN(('uid', username), ('cn', 'users'), ('cn', 'accounts'), base_dn)
# ldap_uri should be ldapi:// in all common cases. Enforce start_tls just
# in case it's a plain LDAP connection.
start_tls = ldap_uri.startswith('ldap://')
try:
conn = ipaldap.LDAPClient(ldap_uri)
conn = ipaldap.LDAPClient(ldap_uri, start_tls=start_tls)
conn.simple_bind(bind_dn, password)
except (errors.ACIError, errors.DatabaseError, errors.NotFound) as e:
logger.error(

3
ipaclient/install/client.py
View File

@ -1637,8 +1637,7 @@ def cert_summary(msg, certs, indent=' '):
def get_certs_from_ldap(server, base_dn, realm, ca_enabled):
ldap_uri = ipaldap.get_ldap_uri(server)
conn = ipaldap.LDAPClient(ldap_uri)
conn = ipaldap.LDAPClient.from_hostname_plain(server)
try:
conn.gssapi_bind()
certs = certstore.get_ca_certs(conn, base_dn, realm, ca_enabled)

3
ipaclient/install/ipa_certupdate.py
View File

@ -70,8 +70,7 @@ def run_with_args(api):
"""
server = urlsplit(api.env.jsonrpc_uri).hostname
ldap_uri = ipaldap.get_ldap_uri(server)
ldap = ipaldap.LDAPClient(ldap_uri)
ldap = ipaldap.LDAPClient.from_hostname_secure(server)
tmpdir = tempfile.mkdtemp(prefix="tmp-")
ccache_name = os.path.join(tmpdir, 'ccache')

7
ipaserver/dcerpc.py
View File

@ -724,9 +724,10 @@ class DomainValidator:
entries = None
try:
ldap_uri = ipaldap.get_ldap_uri(host)
conn = ipaldap.LDAPClient(
ldap_uri,
# AD does not support SASL + TLS at the same time
# https://msdn.microsoft.com/en-us/library/cc223500.aspx
conn = ipaldap.LDAPClient.from_hostname_plain(
host,
no_schema=True,
decode_attrs=False
)

8
ipaserver/install/dogtaginstance.py
View File

@ -386,8 +386,7 @@ class DogtagInstance(service.Service):
conn = None
try:
ldap_uri = ipaldap.get_ldap_uri(protocol='ldapi', realm=self.realm)
conn = ipaldap.LDAPClient(ldap_uri)
conn = ipaldap.LDAPClient.from_realm(self.realm)
conn.external_bind()
entry_attrs = conn.get_entry(self.admin_dn, ['usercertificate'])
@ -465,8 +464,9 @@ class DogtagInstance(service.Service):
wait_groups.append(group_dn)
# Now wait until the other server gets replicated this data
ldap_uri = ipaldap.get_ldap_uri(self.master_host)
master_conn = ipaldap.LDAPClient(ldap_uri, start_tls=True)
master_conn = ipaldap.LDAPClient.from_hostname_secure(
self.master_host
)
logger.debug(
"Waiting for %s to appear on %s", self.admin_dn, master_conn
)

57
ipaserver/install/dsinstance.py
View File

@ -162,18 +162,17 @@ def is_ds_running(server_id=''):
def get_domain_level(api=api):
ldap_uri = ipaldap.get_ldap_uri(protocol='ldapi', realm=api.env.realm)
conn = ipaldap.LDAPClient(ldap_uri)
conn.external_bind()
dn = DN(('cn', 'Domain Level'),
('cn', 'ipa'), ('cn', 'etc'), api.env.basedn)
try:
entry = conn.get_entry(dn, ['ipaDomainLevel'])
except errors.NotFound:
return constants.DOMAIN_LEVEL_0
return int(entry.single_value['ipaDomainLevel'])
with ipaldap.LDAPClient.from_realm(api.env.realm) as conn:
conn.external_bind()
try:
entry = conn.get_entry(dn, ['ipaDomainLevel'])
except errors.NotFound:
return constants.DOMAIN_LEVEL_0
else:
return int(entry.single_value['ipaDomainLevel'])
def get_all_external_schema_files(root):
@ -392,8 +391,7 @@ class DsInstance(service.Service):
def _get_replication_manager(self):
# Always connect to self over ldapi
ldap_uri = ipaldap.get_ldap_uri(protocol='ldapi', realm=self.realm)
conn = ipaldap.LDAPClient(ldap_uri)
conn = ipaldap.LDAPClient.from_realm(self.realm)
conn.external_bind()
repl = replication.ReplicationManager(
self.realm, self.fqdn, self.dm_password, conn=conn
@ -680,7 +678,6 @@ class DsInstance(service.Service):
self._ldap_mod("memberof-conf.ldif")
def init_memberof(self):
if not self.run_init_memberof:
return
@ -689,15 +686,9 @@ class DsInstance(service.Service):
dn = DN(('cn', 'IPA install %s' % self.sub_dict["TIME"]), ('cn', 'memberof task'),
('cn', 'tasks'), ('cn', 'config'))
logger.debug("Waiting for memberof task to complete.")
ldap_uri = ipaldap.get_ldap_uri(self.fqdn)
conn = ipaldap.LDAPClient(ldap_uri)
if self.dm_password:
conn.simple_bind(bind_dn=ipaldap.DIRMAN_DN,
bind_password=self.dm_password)
else:
conn.gssapi_bind()
replication.wait_for_task(conn, dn)
conn.unbind()
with ipaldap.LDAPClient.from_realm(self.realm) as conn:
conn.external_bind()
replication.wait_for_task(conn, dn)
def apply_updates(self):
schema_files = get_all_external_schema_files(paths.EXTERNAL_SCHEMA_DIR)
@ -861,10 +852,9 @@ class DsInstance(service.Service):
self.cacert_name = dsdb.cacert_name
ldap_uri = ipaldap.get_ldap_uri(self.fqdn)
conn = ipaldap.LDAPClient(ldap_uri)
conn.simple_bind(bind_dn=ipaldap.DIRMAN_DN,
bind_password=self.dm_password)
# use LDAPI?
conn = ipaldap.LDAPClient.from_realm(self.realm)
conn.external_bind()
encrypt_entry = conn.make_entry(
DN(('cn', 'encryption'), ('cn', 'config')),
@ -917,10 +907,8 @@ class DsInstance(service.Service):
subject_base=self.subject_base)
trust_flags = dict(reversed(dsdb.list_certs()))
ldap_uri = ipaldap.get_ldap_uri(self.fqdn)
conn = ipaldap.LDAPClient(ldap_uri)
conn.simple_bind(bind_dn=ipaldap.DIRMAN_DN,
bind_password=self.dm_password)
conn = ipaldap.LDAPClient.from_realm(self.realm)
conn.external_bind()
nicknames = dsdb.find_root_cert(self.cacert_name)[:-1]
for nickname in nicknames:
@ -951,14 +939,9 @@ class DsInstance(service.Service):
dsdb = certs.CertDB(self.realm, nssdir=dirname,
subject_base=self.subject_base)
ldap_uri = ipaldap.get_ldap_uri(self.fqdn)
conn = ipaldap.LDAPClient(ldap_uri)
conn.simple_bind(bind_dn=ipaldap.DIRMAN_DN,
bind_password=self.dm_password)
self.export_ca_certs_nssdb(dsdb, self.ca_is_configured, conn)
conn.unbind()
with ipaldap.LDAPClient.from_realm(self.realm) as conn:
conn.external_bind()
self.export_ca_certs_nssdb(dsdb, self.ca_is_configured, conn)
def __add_default_layout(self):
self._ldap_mod("bootstrap-template.ldif", self.sub_dict)

3
ipaserver/install/ipa_backup.py
View File

@ -393,8 +393,7 @@ class Backup(admintool.AdminTool):
if self._conn is not None:
return self._conn
ldap_uri = ipaldap.get_ldap_uri(protocol='ldapi', realm=api.env.realm)
self._conn = ipaldap.LDAPClient(ldap_uri)
self._conn = ipaldap.LDAPClient.from_realm(api.env.realm)
try:
self._conn.external_bind()

3
ipaserver/install/ipa_restore.py
View File

@ -475,8 +475,7 @@ class Restore(admintool.AdminTool):
if self._conn is not None:
return self._conn
ldap_uri = ipaldap.get_ldap_uri(protocol='ldapi', realm=api.env.realm)
self._conn = ipaldap.LDAPClient(ldap_uri)
self._conn = ipaldap.LDAPClient.from_realm(api.env.realm)
try:
self._conn.external_bind()

8
ipaserver/install/ldapupdate.py
View File

@ -54,8 +54,12 @@ UPDATE_SEARCH_TIME_LIMIT = 30 # seconds
def connect(ldapi=False, realm=None, fqdn=None, dm_password=None):
"""Create a connection for updates"""
ldap_uri = ipaldap.get_ldap_uri(fqdn, ldapi=ldapi, realm=realm)
conn = ipaldap.LDAPClient(ldap_uri, decode_attrs=False)
if ldapi:
conn = ipaldap.LDAPClient.from_realm(realm, decode_attrs=False)
else:
conn = ipaldap.LDAPClient.from_hostname_secure(
fqdn, decode_attrs=False
)
try:
if dm_password:
conn.simple_bind(bind_dn=ipaldap.DIRMAN_DN,

37
ipaserver/install/replication.py
View File

@ -137,8 +137,7 @@ def enable_replication_version_checking(realm, dirman_passwd):
enabled then enable it and restart 389-ds. If it is enabled
the do nothing.
"""
ldap_uri = ipaldap.get_ldap_uri(protocol='ldapi', realm=realm)
conn = ipaldap.LDAPClient(ldap_uri)
conn = ipaldap.LDAPClient.from_realm(realm)
if dirman_passwd:
conn.simple_bind(bind_dn=ipaldap.DIRMAN_DN,
bind_password=dirman_passwd)
@ -619,8 +618,9 @@ class ReplicationManager:
"""
self._finalize_replica_settings(self.conn)
ldap_uri = ipaldap.get_ldap_uri(r_hostname)
r_conn = ipaldap.LDAPClient(ldap_uri, cacert=cacert)
r_conn = ipaldap.LDAPClient.from_hostname_secure(
r_hostname, cacert=cacert
)
if r_bindpw:
r_conn.simple_bind(r_binddn, r_bindpw)
else:
@ -1148,12 +1148,7 @@ class ReplicationManager:
local_port = r_port
# note - there appears to be a bug in python-ldap - it does not
# allow connections using two different CA certs
ldap_uri = ipaldap.get_ldap_uri(r_hostname, r_port,
cacert=paths.IPA_CA_CRT,
protocol='ldap')
r_conn = ipaldap.LDAPClient(ldap_uri,
cacert=paths.IPA_CA_CRT,
start_tls=True)
r_conn = ipaldap.LDAPClient.from_hostname_secure(r_hostname)
if r_bindpw:
r_conn.simple_bind(r_binddn, r_bindpw)
@ -1259,9 +1254,7 @@ class ReplicationManager:
raise RuntimeError("Failed to start replication")
def convert_to_gssapi_replication(self, r_hostname, r_binddn, r_bindpw):
ldap_uri = ipaldap.get_ldap_uri(r_hostname, PORT,
cacert=paths.IPA_CA_CRT)
r_conn = ipaldap.LDAPClient(ldap_uri, cacert=paths.IPA_CA_CRT)
r_conn = ipaldap.LDAPClient.from_hostname_secure(r_hostname)
if r_bindpw:
r_conn.simple_bind(r_binddn, r_bindpw)
else:
@ -1289,11 +1282,7 @@ class ReplicationManager:
Only usable to connect 2 existing replicas (needs existing kerberos
principals)
"""
# note - there appears to be a bug in python-ldap - it does not
# allow connections using two different CA certs
ldap_uri = ipaldap.get_ldap_uri(r_hostname, PORT,
cacert=paths.IPA_CA_CRT)
r_conn = ipaldap.LDAPClient(ldap_uri, cacert=paths.IPA_CA_CRT)
r_conn = ipaldap.LDAPClient.from_hostname_secure(r_hostname)
if r_bindpw:
r_conn.simple_bind(r_binddn, r_bindpw)
else:
@ -1789,10 +1778,8 @@ class ReplicationManager:
def setup_promote_replication(self, r_hostname, r_binddn=None,
r_bindpw=None, cacert=paths.IPA_CA_CRT):
# note - there appears to be a bug in python-ldap - it does not
# allow connections using two different CA certs
ldap_uri = ipaldap.get_ldap_uri(r_hostname)
r_conn = ipaldap.LDAPClient(ldap_uri, cacert=cacert)
r_conn = ipaldap.LDAPClient.from_hostname_secure(
r_hostname, cacert=cacert)
if r_bindpw:
r_conn.simple_bind(r_binddn, r_bindpw)
else:
@ -1931,8 +1918,7 @@ class CAReplicationManager(ReplicationManager):
def __init__(self, realm, hostname):
# Always connect to self over ldapi
ldap_uri = ipaldap.get_ldap_uri(protocol='ldapi', realm=realm)
conn = ipaldap.LDAPClient(ldap_uri)
conn = ipaldap.LDAPClient.from_realm(realm)
conn.external_bind()
super(CAReplicationManager, self).__init__(
realm, hostname, None, port=DEFAULT_PORT, conn=conn)
@ -1944,8 +1930,7 @@ class CAReplicationManager(ReplicationManager):
Assumes a promote replica with working GSSAPI for replication
and unified DS instance.
"""
ldap_uri = ipaldap.get_ldap_uri(r_hostname)
r_conn = ipaldap.LDAPClient(ldap_uri)
r_conn = ipaldap.LDAPClient.from_hostname_secure(r_hostname)
r_conn.gssapi_bind()
# Setup the first half

3
ipatests/pytest_ipa/integration/host.py
View File

@ -45,8 +45,7 @@ class Host(pytest_multihost.host.Host):
"""Return an LDAPClient authenticated to this host as directory manager
"""
self.log.info('Connecting to LDAP at %s', self.external_hostname)
ldap_uri = ipaldap.get_ldap_uri(self.external_hostname)
ldap = ipaldap.LDAPClient(ldap_uri)
ldap = ipaldap.LDAPClient.from_hostname_secure(self.external_hostname)
binddn = self.config.dirman_dn
self.log.info('LDAP bind as %s' % binddn)
ldap.simple_bind(binddn, self.config.dirman_password)

3
ipatests/test_install/test_updates.py
View File

@ -65,8 +65,7 @@ class test_update(unittest.TestCase):
else:
raise unittest.SkipTest("No directory manager password")
self.updater = LDAPUpdate(dm_password=self.dm_password, sub_dict={})
ldap_uri = ipaldap.get_ldap_uri(fqdn)
self.ld = ipaldap.LDAPClient(ldap_uri)
self.ld = ipaldap.LDAPClient.from_hostname_secure(fqdn)
self.ld.simple_bind(bind_dn=ipaldap.DIRMAN_DN,
bind_password=self.dm_password)
self.testdir = os.path.abspath(os.path.dirname(__file__))