--- /dev/null
+###############################################################################
+# #
+# dnsbl - A DNS Blocklist Compositor For IPFire #
+# Copyright (C) 2026 IPFire Development Team #
+# #
+# This program is free software: you can redistribute it and/or modify #
+# it under the terms of the GNU General Public License as published by #
+# the Free Software Foundation, either version 3 of the License, or #
+# (at your option) any later version. #
+# #
+# This program is distributed in the hope that it will be useful, #
+# but WITHOUT ANY WARRANTY; without even the implied warranty of #
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
+# GNU General Public License for more details. #
+# #
+# You should have received a copy of the GNU General Public License #
+# along with this program. If not, see <http://www.gnu.org/licenses/>. #
+# #
+###############################################################################
+
+import email.message
+import email.utils
+import ldap
+import logging
+import subprocess
+
+# Setup logging
+log = logging.getLogger(__name__)
+
+class Users(object):
+ def __init__(self, backend):
+ self.backend = backend
+
+ def connect(self):
+ """
+ Creates a new LDAP connection
+ """
+ uri = self.backend.config.get("ldap", "uri", fallback=None)
+
+ # Fail if we have no LDAP URI configured
+ if not uri:
+ raise RuntimeError("LDAP URI is not configured")
+
+ # Log action
+ log.debug("Connecting to LDAP %s" % uri)
+
+ # Connect to LDAP
+ return ldap.initialize(uri)
+
+ def _query(self, filter, base=None, attrlist=None, **kwargs):
+ """
+ Performs a query against the LDAP database
+ """
+ # Connect to LDAP
+ conn = self.connect()
+
+ # Fetch the base
+ if base is None:
+ base = self.backend.config.get("ldap", "base", fallback=None)
+ if not base:
+ raise RuntimeError("LDAP base is not configured")
+
+ log.debug("Performing LDAP query: %s" % filter)
+
+ # Send the query
+ id = conn.search_ext(base, ldap.SCOPE_SUBTREE, filter,
+ attrlist=attrlist, **kwargs)
+
+ # Fetch the result
+ type, result = conn.result(id)
+
+ return result
+
+ def get_by_uid(self, uid):
+ """
+ Fetches a user by its UID
+ """
+ result = self._query(
+ "(&(objectClass=person)(uid=%s))" % uid, sizelimit=1,
+ )
+
+ # Return the user
+ for dn, attrs in result:
+ return User(self.backend, dn, attrs)
+
+ def get_by_dn(self, dn):
+ """
+ Fetches a user by its DN
+ """
+ result = self._query("(objectClass=*)", base=dn, sizelimit=1)
+
+ # Return the user
+ for dn, attrs in result:
+ return User(self.backend, dn, attrs)
+
+ def get_group(self, name):
+ """
+ Fetches all users in a specific group
+ """
+ result = self._query(
+ "(|"
+ "(&(objectClass=groupOfNames)(cn=%s))"
+ "(&(objectClass=posixGroup)(cn=%s))"
+ ")" % (name, name), sizelimit=1,
+ )
+
+ # Return the group
+ for dn, attrs in result:
+ return Group(self.backend, dn, attrs)
+
+ # Admins
+
+ @property
+ def admins(self):
+ return self.get_group("dnsbl-admins")
+
+ # Moderators
+
+ @property
+ def moderators(self):
+ return self.get_group("dnsbl-moderators")
+
+
+class LDAPObject(object):
+ def __init__(self, backend, dn, attrs):
+ self.backend, self.dn, self.attrs = backend, dn, attrs
+
+ def _get(self, key):
+ for value in self.attrs.get(key, []):
+ yield value
+
+ def _get_strings(self, key):
+ for value in self._get(key):
+ yield value.decode()
+
+ def _get_string(self, key, default=None):
+ for value in self._get_strings(key):
+ return value
+
+ return default
+
+
+class User(LDAPObject):
+ def __str__(self):
+ return self.cn
+
+ @property
+ def uid(self):
+ """
+ UID
+ """
+ return self._get_string("uid")
+
+ @property
+ def cn(self):
+ """
+ Common Name
+ """
+ return self._get_string("cn")
+
+ @property
+ def mail(self):
+ """
+ Mail
+ """
+ return self._get_string("mail")
+
+ def sendmail(self, message, headers=None):
+ """
+ Sends the given message to this user
+ """
+ if not isinstance(message, email.message.EmailMessage):
+ content = message
+
+ # Create a new message object
+ message = email.message.EmailMessage()
+ message.set_content(content)
+
+ # Set the recipient
+ message["To"] = email.utils.formataddr((self.cn, self.mail))
+
+ # Set headers
+ if headers:
+ for header in headers:
+ message[header] = headers[header]
+
+ # Set a sender if none set
+ if not "From" in message:
+ message["From"] = email.utils.formataddr(("IPFire DNSBL", "no-reply@ipfire.org"))
+
+ # Log the email
+ log.debug("Sending email:\n%s" % message.as_string())
+
+ # Launch sendmail
+ sendmail = subprocess.Popen(
+ ["/usr/sbin/sendmail", "-t", "-oi"], stdin=subprocess.PIPE, text=True,
+ )
+
+ # Pipe the email into sendmail
+ sendmail.communicate(message.as_string())
+
+
+class Group(LDAPObject):
+ def __str__(self):
+ return self.cn
+
+ def __iter__(self):
+ """
+ Returns all users in this group
+ """
+ # Fetch all users by their DN
+ for dn in self._get_strings("member"):
+ yield self.backend.users.get_by_dn(dn)
+
+ # Fetch all users by their UID
+ for uid in self._get_strings("memberUid"):
+ yield self.backend.users.get_by_uid(uid)
+
+ @property
+ def cn(self):
+ """
+ Common Name
+ """
+ return self._get_string("cn")
projects / dbl.git / commitdiff
