From d7483f5d9ca8c44ce356283fed423c06f850bb4d Mon Sep 17 00:00:00 2001
From: Benjamin Graillot <graillot@crans.org>
Date: Sun, 9 Aug 2020 15:00:13 +0200
Subject: [PATCH] [ldap.py] LDAP lookup plugin
---
lookup_plugins/ldap.py | 101 +++++++++++++++++++++++++++++++++++++++++
1 file changed, 101 insertions(+)
create mode 100644 lookup_plugins/ldap.py
diff --git a/lookup_plugins/ldap.py b/lookup_plugins/ldap.py
new file mode 100644
index 00000000..87cee458
--- /dev/null
+++ b/lookup_plugins/ldap.py
@@ -0,0 +1,101 @@
+import ipaddress
+
+from ansible.errors import AnsibleError, AnsibleParserError
+from ansible.plugins.lookup import LookupBase
+from ansible.utils.display import Display
+
+import ldap
+
+display = Display()
+
+def decode_object(object):
+ return {attribute: [value.decode('utf-8') for value in object[attribute]] for attribute in object}
+
+class LookupModule(LookupBase):
+
+ def __init__(self, **kwargs):
+ self.base = ldap.initialize('ldaps://localhost:1636/')
+ self.base.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_ALLOW)
+ self.base.set_option(ldap.OPT_X_TLS_NEWCTX, 0)
+ self.base_dn = 'dc=crans,dc=org'
+
+ def query(self, base, scope, filter='(objectClass=*)', attr=None):
+ """
+ Make a LDAP query
+ query('ldap', 'query', BASE, SCOPE[, FILTER[, ATTR]])
+ BASE: base dn
+ SCOPE: 'base', 'one' or 'sub'
+ FILTER: ldap filter (optional)
+ ATTR: list of attributes (optional)
+ """
+ scope = { 'base': ldap.SCOPE_BASE, 'one': ldap.SCOPE_ONELEVEL, 'sub': ldap.SCOPE_SUBTREE }[scope]
+ query_id = self.base.search(f"{base}", scope, filter, attr)
+ result = self.base.result(query_id)[1]
+ result = { dn: decode_object(entry) for dn, entry in result }
+ return result
+
+ def ip(self, host, vlan):
+ """
+ Retrieve IP addresses of an interface of a device
+ query('ldap', 'ip', HOST, VLAN)
+ """
+ if isinstance(vlan, int):
+ network_query_id = self.base.search(f"ou=networks,{self.base_dn}", ldap.SCOPE_ONELEVEL, f"description={vlan}")
+ network_result = self.base.result(network_query_id)
+ vlan = network_result[1][0][1]['cn'][0].decode('utf-8')
+ if vlan == 'srv':
+ query_id = self.base.search(f"cn={host}.crans.org,cn={host},ou=hosts,{self.base_dn}", ldap.SCOPE_BASE)
+ else:
+ query_id = self.base.search(f"cn={host}.{vlan}.crans.org,cn={host},ou=hosts,{self.base_dn}", ldap.SCOPE_BASE)
+ result = self.base.result(query_id)
+ result = result[1][0][1]
+ result = [res.decode('utf-8') for res in result['ipHostNumber']]
+ return result
+
+ def run(self, terms, variables=None, **kwargs):
+ if terms[0] == 'query':
+ result = self.query(*terms[1:])
+ elif terms[0] == 'ip':
+ result = self.ip(*terms[1:])
+ elif terms[0] == 'group':
+ query_id = self.base.search(f"ou=group,{self.base_dn}", ldap.SCOPE_SUBTREE, "objectClass=posixGroup")
+ result = self.base.result(query_id)
+ result = result[1]
+ # query interface attribute
+ # query('ldap', 'hosts', HOST, VLAN, ATTR)
+ # HOST: device name
+ # VLAN: vlan name
+ # ATTR: attribute
+ elif terms[0] == 'hosts':
+ host = terms[1]
+ vlan = terms[2]
+ attr = terms[3]
+ if isinstance(vlan, int):
+ network_query_id = self.base.search(f"ou=networks,{self.base_dn}", ldap.SCOPE_ONELEVEL, f"description={vlan}")
+ network_result = self.base.result(network_query_id)
+ vlan = network_result[1][0][1]['cn'][0].decode('utf-8')
+ if vlan == 'srv':
+ query_id = self.base.search(f"cn={host}.crans.org,cn={host},ou=hosts,{self.base_dn}", ldap.SCOPE_BASE)
+ else:
+ query_id = self.base.search(f"cn={host}.{vlan}.crans.org,cn={host},ou=hosts,{self.base_dn}", ldap.SCOPE_BASE)
+ result = self.base.result(query_id)
+ result = result[1][0][1]
+ result = [res.decode('utf-8') for res in result[attr]]
+ elif terms[0] == 'networks':
+ network = terms[1]
+ query_id = self.base.search(f"cn={network},ou=networks,{self.base_dn}", ldap.SCOPE_BASE, "objectClass=ipNetwork")
+ result = self.base.result(query_id)
+ result = result[1][0][1]
+ return [str(ipaddress.ip_network('{}/{}'.format(result['ipNetworkNumber'][0].decode('utf-8'), result['ipNetmaskNumber'][0].decode('utf-8'))))]
+ elif terms[0] == 'vlanid':
+ network = terms[1]
+ query_id = self.base.search(f"cn={network},ou=networks,{self.base_dn}", ldap.SCOPE_BASE, "objectClass=ipNetwork")
+ result = self.base.result(query_id)
+ result = result[1][0][1]
+ return int(result['description'][0])
+ elif terms[0] == 'role':
+ role = terms[1]
+ query_id = self.base.search(f"ou=hosts,{self.base_dn}", ldap.SCOPE_ONELEVEL, f"description={role}")
+ result = self.base.result(query_id)
+ result = [cn.decode('utf-8') for res in result[1] for cn in res[1]['cn']]
+ return result
--
GitLab