ldap auth module

PKGBUILD

@@ -66,6 +66,7 @@ depends=(
 	python-hidapi
 	python-six
 	python-pyrad
+	python-ldap
 	python-zstandard
 	libgpiod
 	freetype2

kvmd/plugins/auth/ldap.py

@@ -0,0 +1,98 @@
+# ========================================================================== #
+#                                                                            #
+#    KVMD - The main PiKVM daemon.                                           #
+#                                                                            #
+#    Copyright (C) 2018-2022  Maxim Devaev <mdevaev@gmail.com>               #
+#                                                                            #
+#    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 <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+# ========================================================================== #
+
+
+import ldap
+
+from ...yamlconf import Option
+
+from ...validators.basic import valid_stripped_string_not_empty
+from ...validators.basic import valid_int_f1
+
+from ...logging import get_logger
+
+from ... import tools
+from ... import aiotools
+
+from . import BaseAuthService
+
+
+# =====
+class Plugin(BaseAuthService):
+    def __init__(  # pylint: disable=super-init-not-called
+        self,
+        url: str,
+        base: str,
+        group: str,
+        user_domain: str,
+        timeout: float,
+    ) -> None:
+
+        self.__url = url
+        self.__base = base
+        self.__group = group
+        self.__user_domain = user_domain
+        self.__timeout = timeout
+
+    @classmethod
+    def get_plugin_options(cls) -> dict:
+        return {
+            "url":         Option("", type=valid_stripped_string_not_empty),
+            "base":        Option("", type=valid_stripped_string_not_empty),
+            "group":       Option("", type=valid_stripped_string_not_empty),
+            "user_domain": Option(""),
+            "timeout":     Option(5, type=valid_int_f1),
+        }
+
+    async def authorize(self, user: str, passwd: str) -> bool:
+        return (await aiotools.run_async(self.__inner_authorize, user, passwd))
+
+    def __inner_authorize(self, user: str, passwd: str) -> bool:
+        if self.__user_domain:
+            user = f"{user}@{self.__user_domain}"
+        conn: (ldap.ldapobject.LDAPObject | None) = None
+        try:
+            conn = ldap.initialize(self.__url)
+            conn.set_option(ldap.OPT_REFERRALS, 0)
+            conn.set_option(ldap.OPT_TIMEOUT, self.__timeout)
+            conn.simple_bind_s(user, passwd)
+            for (dn, attrs) in (conn.search_st(
+                base=self.__base,
+                scope=ldap.SCOPE_SUBTREE,
+                filterstr=f"(&(objectClass=user)(userPrincipalName={user})(memberOf={self.__group}))",
+                attrlist=["userPrincipalName", "memberOf"],
+                timeout=self.__timeout,
+            ) or []):
+                if dn is not None and isinstance(attrs, dict) and attrs.get("memberOf"):
+                    return True
+        except ldap.INVALID_CREDENTIALS:
+            pass
+        except ldap.SERVER_DOWN:
+            get_logger().error("LDAP server is down")
+        except Exception as err:
+            get_logger().error("Unexpected LDAP error: %s", tools.efmt(err))
+        finally:
+            if conn is not None:
+                try:
+                    conn.unbind()
+                except Exception:
+                    pass
+        return False

testenv/Dockerfile

@@ -58,6 +58,7 @@ RUN pacman --noconfirm --ask=4 -Syy \
 		python-xlib \
 		libxkbcommon \
 		python-hidapi \
+		python-ldap \
 		python-zstandard \
 		freetype2 \
 		nginx-mainline \

testenv/linters/pylint.ini

@@ -5,6 +5,7 @@ extension-pkg-whitelist =
 	gpiod,
 	spidev,
 	netifaces,
+	_ldap,
 	ustreamer,
 	hid,