feature: openldap support (#31)

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 - Refactoring - all sql functions to have session as first arg
 - Adjust oauth2 so that github or google providers can use used separately
+- Support for OpenLDAP (RFC 4510) authentication
 
 
 ## 0.6.4 - 2024-01-29

auth_server/auth.py

@@ -13,7 +13,11 @@
 from .database.models import User
 from . import schemas
 from .config import Settings
-from .backends import GoogleAuth, GithubAuth, OAuth2Provider
+from .backends import (
+    GoogleAuth,
+    GithubAuth,
+)
+from .backends import ldap
 from .utils import raise_on_empty
 
 
@@ -26,40 +30,51 @@ async def authenticate(
     *,
     username: str | None = None,
     password: str | None = None,
-    provider: OAuth2Provider | None = None,
+    provider: schemas.AuthProvider = schemas.AuthProvider.DB,
     client_id: str | None = None,
     code: str | None = None,
     redirect_uri: str | None = None
 ) -> schemas.User | None:
 
-    if username and password:
+    # provider = DB
+    if username and password and provider == schemas.AuthProvider.DB:
         # password based authentication against database
         return db_auth(db, username, password)
 
-    raise_on_empty(
-        code=code,
-        client_id=client_id,
-        provider=provider,
-        redirect_uri=redirect_uri
-    )
-
-    if provider == OAuth2Provider.GOOGLE:
+    if provider == schemas.AuthProvider.GOOGLE:
+        # provider = GOOGLE (oauth2/google)
+        raise_on_empty(
+            code=code,
+            client_id=client_id,
+            provider=provider,
+            redirect_uri=redirect_uri
+        )
         # oauth 2.0, google provider
         return await google_auth(
             db,
             client_id=client_id,
             code=code,
             redirect_uri=redirect_uri
         )
-    elif provider == OAuth2Provider.GITHUB:
+    elif provider == schemas.AuthProvider.GITHUB:
+        # provider = GitHub (oauth2/github)
+        raise_on_empty(
+            code=code,
+            client_id=client_id,
+            provider=provider,
+            redirect_uri=redirect_uri
+        )
         return await github_auth(
             db,
             client_id=client_id,
             code=code,
             redirect_uri=redirect_uri
         )
+    elif provider == schemas.AuthProvider.LDAP:
+        # provider = ldap
+        return await ldap_auth(db, username, password)
     else:
-        raise ValueError("Unknown or empty oauth2 provider")
+        raise ValueError("Unknown or empty auth provider")
 
 
 def verify_password(password: str, hashed_password: str) -> bool:
@@ -119,6 +134,35 @@ def db_auth(db: Session, username: str, password: str) -> schemas.User | None:
     return user
 
 
+async def ldap_auth(
+    db: Session,
+    username: str,
+    password: str
+) -> schemas.User | None:
+    client = ldap.get_client(username, password)
+
+    try:
+        await client.signin()
+    except Exception as ex:
+        logger.warning(f"Auth:LDAP: sign in failed with {ex}")
+
+        raise HTTPException(
+            status_code=401,
+            detail=f"401 Unauthorized. LDAP Auth error: {ex}."
+        )
+
+    email = ldap.get_default_email(username)
+    try:
+        email = await client.user_email()
+    except Exception as ex:
+        logger.warning(f"Auth:LDAP: cannot retrieve user email {ex}")
+        logger.warning(
+            f"Auth:LDAP: user email fallback to {email}"
+        )
+
+    return get_or_create_user_by_email(db, email)
+
+
 async def google_auth(
     db: Session,
     client_id: str,

auth_server/backends/__init__.py

@@ -1,9 +1,3 @@
-from enum import Enum
-
 from .google import GoogleAuth
 from .github import GithubAuth
-
-
-class OAuth2Provider(str, Enum):
-    GOOGLE = "google"
-    GITHUB = "github"
+from .ldap import LDAPAuth

auth_server/backends/ldap.py

@@ -0,0 +1,113 @@
+import logging
+from ldap3 import Server, Connection, ALL
+from auth_server.config import Settings
+
+
+logger = logging.getLogger(__name__)
+
+settings = Settings()
+
+
+class LDAPAuth:
+    name: str = 'ldap'
+
+    def __init__(self,
+        url: str,
+        username: str,
+        password: str,
+        user_dn_format: str,
+        email_attr: str = 'mail',
+        use_ssl: bool = True,
+    ):
+        self._username = username
+        self._password = password
+        self._url = url
+        self._user_dn_format = user_dn_format
+        self._use_ssl = use_ssl
+        self._email_attr = email_attr
+        self._conn = None
+
+    async def signin(self):
+        return self._signin()
+
+    def _signin(self):
+        server = Server(self._url, use_ssl=self._use_ssl, get_info=ALL)
+        user_dn = self.get_user_dn()
+        self._conn = Connection(server, user_dn, self._password)
+
+        if not self._conn.bind():
+            # this may happen from multiple reasons:
+            # 1. user simply provided wrong credentials
+            # 2. auth_server configuration issues
+            # 3. server side problem
+            logger.info(
+                f"LDAP conn.bind() returned falsy value: {self._conn}"
+            )
+            raise Exception("LDAP conn.bind() returned falsy value")
+
+        return self._conn
+
+    async def user_email(self) -> str | None:
+        if self._conn is None:
+            self._conn = await self.signin()
+
+        return self._user_email()
+
+    def _user_email(self) -> str | None:
+        user_dn = self.get_user_dn()
+        search_filter = f'(uid={self._username})'
+        attributes = [
+            'uid', self._email_attr
+        ]
+        msg = "User entry not found:" \
+              f"user_dn: {user_dn} " \
+              f"search filter: {search_filter}" \
+              f"attributes: {attributes}"
+        if self._conn.search(
+            user_dn,
+            search_filter,
+            attributes=attributes
+        ):
+            if len(self._conn.entries) == 0:
+                logger.info(msg)
+                return None
+
+            result = self._conn.entries[0]
+
+            if not result:
+                logger.info("conn.search returned an empty entry")
+                logger.info(msg)
+                return None
+
+            if result[self._email_attr] is None:
+                logger.info("con.search empty email attr")
+                logger.info(msg)
+                return None
+
+            return result[self._email_attr].value
+        else:
+            logger.info("conn.search returned falsy value")
+            logger.info(msg)
+
+        return None
+
+    def get_user_dn(self) -> str:
+        return self._user_dn_format.format(username=self._username)
+
+
+def get_client(username: str, password: str) -> LDAPAuth:
+    return LDAPAuth(
+        url=settings.papermerge__auth__ldap_url,
+        username=username,
+        password=password,
+        user_dn_format=settings.papermerge__auth__ldap_user_dn_format,
+        email_attr=settings.papermerge__auth__ldap_email_attr,
+        use_ssl=settings.papermerge__auth__ldap_use_ssl
+    )
+
+
+def get_default_email(username: str) -> str:
+    domain = settings.papermerge__auth__ldap_user_email_domain_fallback
+    return f"{username}@{domain}"
+
+

auth_server/cli/ldap.py

@@ -0,0 +1,43 @@
+import typer
+from rich.console import Console
+from typing_extensions import Annotated
+from auth_server.backends import ldap
+
+
+app = typer.Typer()
+console = Console()
+
+Password = Annotated[
+    str,
+    typer.Option(
+        prompt=True,
+        confirmation_prompt=False,
+        hide_input=True
+    )
+]
+
+
+@app.command()
+def auth(username: str, password: Password):
+    """Authenticates user with credentials"""
+    client = ldap.get_client(username, password)
+    try:
+        client._signin()
+        console.print("Authentication success", style="green")
+    except Exception:
+        console.print("Authentication failed", style="red")
+
+
+@app.command()
+def user_email(username: str, password: Password):
+    """Prints user email as retrieved from LDAP"""
+    client = ldap.get_client(username, password)
+    try:
+        client._signin()
+        console.print(f"User email: {client._user_email()}")
+    except Exception:
+        console.print("Authentication error", style="red")
+
+
+if __name__ == '__main__':
+    app()

auth_server/config.py

@@ -31,6 +31,16 @@ class Settings(BaseSettings):
     papermerge__auth__google_client_secret: str | None = None
     papermerge__auth__github_client_secret: str | None = None
 
+    papermerge__auth__ldap_url: str | None = None  # e.g. ldap.trusel.net
+    papermerge__auth__ldap_use_ssl: bool = True
+    # e.g. uid={username},ou=People,dc=ldap,dc=trusel,dc=net
+    papermerge__auth__ldap_user_dn_format: str | None = None
+    # LDAP Entry attribute name for the email
+    papermerge__auth__ldap_email_attr: str = 'mail'
+    # if there is an error retrieving ldap_email_attr, the
+    # fallback user email will be set to username@<email-domain-fallback>
+    papermerge__auth__ldap_user_email_domain_fallback: str = 'example-ldap.com'
+
 
 @lru_cache()
 def get_settings():

auth_server/main.py

@@ -6,7 +6,6 @@
 
 from .auth import authenticate, create_token
 from . import schemas
-from .backends import OAuth2Provider
 from .config import get_settings
 from auth_server.database import get_db
 
@@ -21,7 +20,7 @@
 @app.post("/token")
 async def retrieve_token(
     response: Response,
-    provider: OAuth2Provider | None = None,
+    provider: schemas.AuthProvider | None = None,
     client_id: str | None = None,
     code: str | None = None,
     redirect_uri: str | None = None,
@@ -40,7 +39,7 @@ async def retrieve_token(
     if creds:
         kwargs['username'] = creds.username
         kwargs['password'] = creds.password
-
+        kwargs['provider'] = creds.provider.value
     try:
         user = await authenticate(db, **kwargs)
     except ValueError as ex:

auth_server/schemas.py

@@ -1,4 +1,6 @@
 from uuid import UUID
+from enum import Enum
+from typing_extensions import Literal
 from pydantic import BaseModel, ConfigDict
 
 
@@ -20,8 +22,16 @@ class Token(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
 
+class AuthProvider(str, Enum):
+    GOOGLE = "google"
+    GITHUB = "github"
+    LDAP = "ldap"
+    DB = "db"
+
+
 class UserCredentials(BaseModel):
     username: str
     password: str
+    provider: AuthProvider = AuthProvider.DB
 
     model_config = ConfigDict(from_attributes=True)