ldap implementation

README.md

@@ -16,6 +16,7 @@ Current supported features are:
 * [Last.fm][lastfm] scrobbling
 * [ListenBrainz][listenbrainz] scrobbling
 * Jukebox mode
+* LDAP authentication
 
 Supysonic currently targets the version 1.12.0 of the Subsonic API. For more
 details, go check the [API implementation status][docs-api].

ci-requirements.txt

@@ -2,3 +2,4 @@
 
 lxml
 coverage
+ldap3

config.sample

@@ -99,3 +99,14 @@ default_transcode_target = mp3
 ;mp3 = audio/mpeg
 ;ogg = audio/vorbis
 
+[ldap]
+;server_url = ldapi://%2Frun%2Fslapd%2Fldapi ;server_url = ldap://127.0.0.1:389
+;bind_dn = cn=username,dc=example,dc=org
+;bind_pw = password
+;base_dn = ou=Users,dc=example,dc=org
+
+; Optional parameters with default values
+;user_filter = (&(objectClass=inetOrgperson)(uid={username}))
+;admin_filter =
+;username_attr = uid
+;email_attr = mail

docs/setup/configuration.rst

@@ -363,3 +363,35 @@ See the following links for a list of examples:
    ; Default: none
    ;mp3 = audio/mpeg
    ;ogg = audio/vorbis
+
+``[ldap]`` section
+-----------------------
+
+This section defines the LDAP connection parameters.
+when an LDAP user is found on a server and doesn't exist in the Supysonic database,
+a new user is created.
+
+``server_url``
+    URL of the LDAP server
+
+``bind_dn``
+``bind_pw``
+    Bind credentials used for the search query
+
+``base_dn``
+    Base DN where the search is performed
+
+``user_filter``
+    Filter for finding users
+    A special variable ``{username}`` can be used for filtering
+
+``admin_filter``
+    Same as ``user_filter`` but for finding admins
+
+``username_attr``
+    Attribute containing the username
+    Default is ``uid``
+
+``email_attr``
+    Attribute containing the e-mail address
+    Default is ``mail``

setup.cfg

@@ -54,6 +54,7 @@ python_requires = >=3.7
 install_requires =
     click
     flask >=0.11
+    ldap3
     peewee
     Pillow >=9.1.0
     requests >=1.0.0
@@ -73,6 +74,8 @@ console_scripts =
     supysonic-cli = supysonic.cli:main
     supysonic-daemon = supysonic.daemon:main
     supysonic-server = supysonic.server:main
-
+[options.extras_require]
+  ldap= 
+    ldap3
 [options.data_files]
 share/man/man1 = man/*.1

supysonic/config.py

@@ -55,6 +55,18 @@ class DefaultConfig:
     }
     LASTFM = {"api_key": None, "secret": None}
     LISTENBRAINZ = {"api_url": "https://api.listenbrainz.org"}
+    LDAP = {
+        "server_url": False,
+        "bind_dn": None,
+        "bind_pw": None,
+        "base_dn": None,
+        "user_filter": "(&(objectClass=inetOrgperson)(uid={username}))",
+        "admin_filter": False,
+        "username_attr": "uid",
+        "email_attr": "mail",
+    }
+
+
     TRANSCODING = {}
     MIMETYPES = {}
 

supysonic/db.py

@@ -430,8 +430,8 @@ class User(_Model):
     id = PrimaryKeyField()
     name = CharField(64, unique=True)
     mail = CharField(null=True)
-    password = FixedCharField(40)
-    salt = FixedCharField(6)
+    password = FixedCharField(40,null=True)
+    salt = FixedCharField(6,null=True)
 
     admin = BooleanField(default=False)
     jukebox = BooleanField(default=False)

supysonic/managers/ldap.py

@@ -0,0 +1,65 @@
+import logging
+try:
+    import ldap3
+except ModuleNotFoundError:
+    ldap3 = None
+
+logger = logging.getLogger(__name__)
+
+class LdapManager:
+    def __init__(self, **config):
+        if not config["server_url"]:
+            logger.debug("LDAP 'server_url' is not configured.")
+            raise ValueError
+        elif not ldap3:
+            logger.error("Module 'ldap3' is not installed.")
+            raise ValueError
+        elif None in config.values():
+            logger.error("Some required LDAP parameters are missing.")
+            raise ValueError
+
+        self.server = ldap3.Server(config["server_url"], get_info=None)
+        self.config = config
+
+    def try_auth(self, username, password):
+        admin = False
+
+        if self.config["admin_filter"]:
+            entry = self.search_user(username, self.config["admin_filter"])
+            if entry:
+                logger.info(f"User '{username}' is admin.")
+                admin = True
+
+        if not admin:
+            entry = self.search_user(username, self.config["user_filter"])
+
+        if entry:
+            try:
+                with ldap3.Connection(self.server, entry.entry_dn, password, read_only=True):
+                    return {
+                        "mail": entry[self.config["email_attr"]],
+                        "admin": admin
+                    }
+            except ldap3.core.exceptions.LDAPBindError:
+                logger.error(f"Bind failed for '{entry.entry_dn}'.")
+            except Exception as e:
+                logger.error(f"LDAP error: {e}")
+
+    def search_user(self, username, _filter):
+        try:
+            with ldap3.Connection(self.server, self.config["bind_dn"], self.config["bind_pw"], read_only=True) as conn:
+                conn.search(
+                    self.config["base_dn"],
+                    _filter.format(username=username),
+                    attributes=[self.config["username_attr"], self.config["email_attr"]],
+                    size_limit=1
+                )
+                entries = conn.entries
+                if entries and entries[0][self.config["username_attr"]] == username:
+                    return entries[0]
+                else:
+                    logger.info(f"User '{username}' not found in LDAP database.")
+        except ldap3.core.exceptions.LDAPBindError:
+            logger.error(f"Bind failed for '{self.config['bind_dn']}'.")
+        except Exception as e:
+            logger.error(f"LDAP error: {e}")

supysonic/managers/user.py

@@ -12,7 +12,8 @@
 import uuid
 
 from ..db import User
-
+from ..config import get_current_config
+from .ldap import LdapManager
 
 class UserManager:
     @staticmethod
@@ -46,13 +47,28 @@ def delete_by_name(name):
 
     @staticmethod
     def try_auth(name, password):
+        try:
+            ldap = LdapManager(**get_current_config().LDAP)
+        except ValueError:
+            ldap = None
+        ldap_user = ldap.try_auth(name, password) if ldap else None
         user = User.get_or_none(name=name)
-        if user is None:
-            return None
-        elif UserManager.__encrypt_password(password, user.salt)[0] != user.password:
-            return None
-        else:
+        if ldap_user:
+            if user is None:
+                user = User.create(name=name, mail=ldap_user["mail"], admin=ldap_user["admin"])
+            else:
+                if user.admin != ldap_user["admin"]:
+                    user.admin = ldap_user["admin"]
+                if user.mail != ldap_user["mail"]:
+                    user.mail = ldap_user["mail"]
             return user
+        else:
+            if user is None:
+                return None
+            elif UserManager.__encrypt_password(password, user.salt)[0] != user.password:
+                return None
+            else:
+                return user
 
     @staticmethod
     def change_password(uid, old_pass, new_pass):

supysonic/schema/migration/mysql/20230314.sql

@@ -0,0 +1,2 @@
+ALTER TABLE user MODIFY password CHAR(40);
+ALTER TABLE user MODIFY salt CHAR(6);

supysonic/schema/migration/postgres/20230314.sql

@@ -0,0 +1,3 @@
+ALTER TABLE "user" 
+    ALTER COLUMN password DROP NOT NULL,
+    ALTER COLUMN salt DROP NOT NULL;

supysonic/schema/migration/sqlite/20230314.sql

@@ -0,0 +1,28 @@
+COMMIT;
+PRAGMA foreign_keys = OFF;
+BEGIN TRANSACTION;
+DROP INDEX index_user_last_play_id_fk;
+create TABLE user_new (
+    id CHAR(36) PRIMARY KEY,
+    name VARCHAR(64) NOT NULL,
+    mail VARCHAR(256),
+    password CHAR(40),
+    salt CHAR(6),
+    admin BOOLEAN NOT NULL,
+    jukebox BOOLEAN NOT NULL,
+    lastfm_session CHAR(32),
+    lastfm_status BOOLEAN NOT NULL,
+    last_play_id CHAR(36) REFERENCES track,
+    last_play_date DATETIME
+);
+CREATE INDEX IF NOT EXISTS index_user_last_play_id_fk ON user_new(last_play_id);
+INSERT INTO user_new(id, name, mail, password, salt, admin, jukebox, lastfm_session, lastfm_status, last_play_id, last_play_date)
+SELECT id, name, mail, password, salt, admin, jukebox, lastfm_session, lastfm_status, last_play_id, last_play_date
+FROM user;
+
+DROP TABLE user;
+ALTER TABLE user_new RENAME TO user;
+COMMIT;
+VACUUM;
+PRAGMA foreign_keys = ON;
+BEGIN TRANSACTION;

supysonic/schema/mysql.sql

@@ -53,8 +53,8 @@ CREATE TABLE IF NOT EXISTS user (
     id CHAR(32) PRIMARY KEY,
     name VARCHAR(64) NOT NULL,
     mail VARCHAR(256),
-    password CHAR(40) NOT NULL,
-    salt CHAR(6) NOT NULL,
+    password CHAR(40),
+    salt CHAR(6),
     admin BOOLEAN NOT NULL,
     jukebox BOOLEAN NOT NULL,
     listenbrainz_session CHAR(36),

supysonic/schema/postgres.sql

@@ -53,8 +53,8 @@ CREATE TABLE IF NOT EXISTS "user" (
     id UUID PRIMARY KEY,
     name VARCHAR(64) NOT NULL,
     mail VARCHAR(256),
-    password CHAR(40) NOT NULL,
-    salt CHAR(6) NOT NULL,
+    password CHAR(40),
+    salt CHAR(6),
     admin BOOLEAN NOT NULL,
     jukebox BOOLEAN NOT NULL,
     listenbrainz_session CHAR(36),

supysonic/schema/sqlite.sql

@@ -55,8 +55,8 @@ CREATE TABLE IF NOT EXISTS user (
     id CHAR(36) PRIMARY KEY,
     name VARCHAR(64) NOT NULL,
     mail VARCHAR(256),
-    password CHAR(40) NOT NULL,
-    salt CHAR(6) NOT NULL,
+    password CHAR(40),
+    salt CHAR(6),
     admin BOOLEAN NOT NULL,
     jukebox BOOLEAN NOT NULL,
     listenbrainz_session CHAR(36),

tests/managers/test_manager_ldap.py

@@ -0,0 +1,54 @@
+from supysonic import db
+from supysonic.managers.ldap import LdapManager
+import unittest
+from unittest.mock import patch
+
+LDAP = {
+    "server_url": "fakeServer",
+    "bind_dn": "cn=my_user,ou=test,o=lab",
+    "bind_pw": "my_password",
+    "base_dn": "ou=test,o=lab",
+    "user_filter": "(&(objectClass=inetOrgPerson))",
+    "admin_filter": False,
+    "username_attr": "uid",
+    "email_attr": "mail",
+}
+
+class MockEntrie ():
+    def __init__(self,dn,attr):
+        self.entry_dn=dn
+        self.attribute=attr
+    def __getitem__(self, item):
+        return self.attribute[item]
+
+class LdapManagerTestCase(unittest.TestCase):
+
+    def setUp(self):
+        # Create an empty sqlite database in memory
+        pass
+
+    def tearDown(self):
+        pass
+
+    @patch("supysonic.managers.ldap.ldap3.Connection")
+    def test_ldapManager_searchUser(self, mock_object):
+        mock_object.return_value.__enter__.return_value.entries = [
+            {LDAP["email_attr"]:"[email protected]",
+             LDAP["username_attr"]:"toto"
+            }]
+        ldap = LdapManager(**LDAP)
+        ldap_user = ldap.search_user("toto", LDAP["user_filter"])
+        self.assertEqual(ldap_user[LDAP["email_attr"]], "[email protected]")
+        ldap_user = ldap.search_user("tata", LDAP["user_filter"])
+        self.assertIsNone(ldap_user)
+
+    @patch("supysonic.managers.ldap.ldap3.Connection")
+    def test_ldapManager_try_auth(self, mock_object):
+        mock_object.return_value.__enter__.return_value.entries = [
+            MockEntrie ("cn=toto",{LDAP["email_attr"]:"[email protected]", LDAP["username_attr"]:"toto"})]
+        ldap = LdapManager(**LDAP)
+        ldap_user = ldap.try_auth("toto", "toto")
+        self.assertFalse(ldap_user["admin"])
+        self.assertEqual(ldap_user[LDAP["email_attr"]], "[email protected]")
+        ldap_user = ldap.try_auth("tata", "tata")
+        self.assertIsNone(ldap_user)

tests/managers/test_manager_user.py

@@ -8,10 +8,12 @@
 
 from supysonic import db
 from supysonic.managers.user import UserManager
-
+from supysonic.config import get_current_config
 import unittest
+from unittest.mock import patch
 import uuid
 
+from .test_manager_ldap import MockEntrie
 
 class UserManagerTestCase(unittest.TestCase):
     def setUp(self):
@@ -142,6 +144,34 @@ def test_try_auth(self):
         # Non-existent user
         self.assertIsNone(UserManager.try_auth("null", "null"))
 
+
+    @patch("supysonic.managers.ldap.ldap3.Connection")
+    def test_try_auth_ldap(self,mock_object):
+        config = get_current_config()
+        config.LDAP["server_url"] = "fakeserver"
+        config.LDAP["bind_dn"]="cn=my_user,ou=test,o=lab"
+        config.LDAP["bind_pw"]= "my_password",
+        config.LDAP["base_dn"]= "ou=test,o=lab",
+        mock_object.return_value.__enter__.return_value.entries = [
+            MockEntrie ("cn=toto",{config.LDAP["email_attr"]:"[email protected]",config.LDAP["username_attr"]:"toto"})]
+        authed = UserManager.try_auth("toto","toto")
+        user = db.User.get(name="toto")
+        self.assertEqual(authed, user)
+
+        # test admin and mail change
+        config.LDAP["admin_filter"] = "fake_admin_filter"
+        mock_object.return_value.__enter__.return_value.entries = [
+        MockEntrie ("cn=toto",{config.LDAP["email_attr"]:"[email protected]",config.LDAP["username_attr"]:"toto"})]
+        authed= UserManager.try_auth("toto","toto")
+        self.assertEqual(authed.mail,"[email protected]")
+        self.assertEqual(authed.admin,True)
+
+        # Non-existent user
+
+        self.assertIsNone(UserManager.try_auth("tata","toto"))
+
+
+
     def test_change_password(self):
         self.create_data()