From c912b51e2aec383b4b0b3293b77d9305a1b8dd38 Mon Sep 17 00:00:00 2001
From: Julian Rother <julian@jrother.eu>
Date: Sun, 9 Jan 2022 19:44:57 +0100
Subject: [PATCH] Regex filter option for groups (--group-filter-regex)
---
debian/contrib/uffd-ldapd.conf | 1 +
uffd-ldapd | 20 ++++++++++++++++----
2 files changed, 17 insertions(+), 4 deletions(-)
diff --git a/debian/contrib/uffd-ldapd.conf b/debian/contrib/uffd-ldapd.conf
index 03a8fa0..1ee50d0 100644
--- a/debian/contrib/uffd-ldapd.conf
+++ b/debian/contrib/uffd-ldapd.conf
@@ -9,3 +9,4 @@
#SERVER_BIND_PASSWORD="SECRET-BIND-PASSWORD"
#SERVER_CACHE_TTL="60"
+#SERVER_GROUP_FILTER_REGEX=""
diff --git a/uffd-ldapd b/uffd-ldapd
index 0b64cf4..10fad3f 100755
--- a/uffd-ldapd
+++ b/uffd-ldapd
@@ -4,6 +4,7 @@ import sys
import socketserver
import logging
import socket
+import re
import click
import requests
@@ -89,6 +90,7 @@ class UffdLDAPRequestHandler(ldapserver.LDAPRequestHandler):
api = None
dn_base = None
bind_password = None # if None anonymous reads are allowed
+ group_filter_regex = None
def do_bind_simple_authenticated(self, dn, password):
dn = self.subschema.DN.from_str(dn)
@@ -186,6 +188,8 @@ class UffdLDAPRequestHandler(ldapserver.LDAPRequestHandler):
if value.is_direct_child_of(self.subschema.DN(self.dn_base, ou='groups')) and value.object_attribute == 'cn':
request_params = {'group': normalize_group_name(value.object_value)}
break
+ if 'group' in request_params and not self.group_filter_regex.match(request_params['group']):
+ return
for user in self.api.get_users(**request_params):
yield template.create_entry(user['loginname'],
cn=[user['displayname']],
@@ -195,7 +199,9 @@ class UffdLDAPRequestHandler(ldapserver.LDAPRequestHandler):
mail=[user['email']],
uid=[user['loginname']],
uidNumber=[user['id']],
- memberOf=[self.subschema.DN(self.subschema.DN(self.dn_base, ou='groups'), cn=group) for group in user['groups']],
+ memberOf=[self.subschema.DN(self.subschema.DN(self.dn_base, ou='groups'), cn=group)
+ for group in user['groups']
+ if self.group_filter_regex.match(group)],
)
def do_search_groups(self, baseobj, scope, filterobj):
@@ -220,20 +226,25 @@ class UffdLDAPRequestHandler(ldapserver.LDAPRequestHandler):
if value.is_direct_child_of(self.subschema.DN(self.dn_base, ou='users')) and value.object_attribute == 'uid':
request_params = {'member': normalize_user_loginname(value.object_value)}
break
+ if 'name' in request_params and not self.group_filter_regex.match(request_params['name']):
+ return
for group in self.api.get_groups(**request_params):
+ if not self.group_filter_regex.match(group['name']):
+ continue
yield template.create_entry(group['name'],
cn=[group['name']],
gidNumber=[group['id']],
uniqueMember=[self.subschema.DN(self.subschema.DN(self.dn_base, ou='users'), uid=user) for user in group['members']],
)
-def make_requesthandler(api, dn_base, bind_password=None):
+def make_requesthandler(api, dn_base, bind_password=None, group_filter_regex=None):
class RequestHandler(UffdLDAPRequestHandler):
pass
dn_base = RequestHandler.subschema.DN.from_str(dn_base)
RequestHandler.api = api
RequestHandler.dn_base = dn_base
RequestHandler.bind_password = bind_password.encode() if bind_password else None
+ RequestHandler.group_filter_regex = re.compile(group_filter_regex) if group_filter_regex else re.compile('')
return RequestHandler
class FilenoUnixStreamServer(socketserver.UnixStreamServer):
@@ -284,7 +295,8 @@ class StdoutFilter(logging.Filter):
@click.option('--cache-ttl', default=60, help='Time-to-live for API response caching in seconds')
@click.option('--base-dn', required=True, help='Base DN for user, group and system objects. E.g. "dc=example,dc=com"')
@click.option('--bind-password', help='Authentication password for the service connection to LDAP. Bind DN is always "cn=service,ou=system,BASEDN". If set, anonymous access is disabled.')
-def main(socket_address, socket_path, socket_fd, api_url, api_user, api_secret, cache_ttl, base_dn, bind_password):
+@click.option('--group-filter-regex', help='Python regular expression that group names must match for the group to be visible to LDAP clients')
+def main(socket_address, socket_path, socket_fd, api_url, api_user, api_secret, cache_ttl, base_dn, bind_password, group_filter_regex):
# pylint: disable=too-many-locals
if (socket_address is not None) \
+ (socket_path is not None) \
@@ -302,7 +314,7 @@ def main(socket_address, socket_path, socket_fd, api_url, api_user, api_secret,
root_logger.addHandler(stderr_handler)
api = UffdAPI(api_url, api_user, api_secret, cache_ttl)
- RequestHandler = make_requesthandler(api, base_dn, bind_password)
+ RequestHandler = make_requesthandler(api, base_dn, bind_password, group_filter_regex)
if socket_address is not None:
host, port = parse_network_address(socket_address)
server = socketserver.ThreadingTCPServer((host, int(port)), RequestHandler)
--
GitLab