diff --git a/CHANGELOG.md b/CHANGELOG.md index 8876954bc..46c06baff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ All notable changes to Bambuddy will be documented in this file. - **Shortest Job First Queue Scheduling** ([#879](https://github.com/maziggy/bambuddy/issues/879)) — New SJF toggle badge on the queue page header. When enabled, the scheduler starts shorter print jobs before longer ones instead of FIFO order. A starvation guard ensures long jobs that get skipped once are protected from being skipped again — they move to the front of the queue on the next cycle. The queue display automatically reorders to show the scheduler's actual execution order. Print duration is cached on queue items at creation time from the 3MF metadata. - **Auto-Print G-code Injection** ([#422](https://github.com/maziggy/bambuddy/issues/422)) — Configure custom start and end G-code snippets per printer model in Settings (Workflow tab) for bed-clearing systems like Farmloop, SwapMod, AutoClear, and Printflow 3D. When adding a print to the queue, enable "Inject G-code" to have the scheduler inject the configured snippets into the 3MF before uploading to the printer. The original file is never modified — injection creates a temporary copy for upload only. - **External Folder Subfolder Preservation** ([#890](https://github.com/maziggy/bambuddy/issues/890)) — Scanning an external folder now mirrors the real directory structure into the file manager folder tree instead of flattening all files into the root. Subdirectories are created as child LibraryFolders with correct parent/child hierarchy, and files are assigned to their matching subfolder. Hidden directories are skipped when "Show hidden files" is disabled. Subfolders that are deleted from disk are automatically cleaned up on the next scan. Created subfolders inherit the parent's read-only and show-hidden settings. +- **LDAP Authentication** ([#794](https://github.com/maziggy/bambuddy/issues/794)) — Users can now authenticate against an LDAP/Active Directory server. Configure the LDAP server URL, bind DN, search base, and user filter in Settings > Authentication > LDAP. Supports StartTLS, LDAPS (SSL), and plaintext connections. LDAP groups can be mapped to BamBuddy groups (Administrators, Operators, Viewers) for automatic role assignment. Auto-provisioning creates BamBuddy accounts on first LDAP login when enabled. Local admin accounts remain as fallback when the LDAP server is unreachable. Password management features (change password, forgot password, admin reset) are automatically disabled for LDAP users. ### Improved - **Database Engine Info on System Page** — The System Information page now shows the active database engine (SQLite or PostgreSQL) and its version in the Database section, making it easy to verify which backend is in use. diff --git a/backend/app/api/routes/auth.py b/backend/app/api/routes/auth.py index d10c22e55..894341f81 100644 --- a/backend/app/api/routes/auth.py +++ b/backend/app/api/routes/auth.py @@ -62,6 +62,7 @@ def _user_to_response(user: User) -> UserResponse: role=user.role, is_active=user.is_active, is_admin=user.is_admin, + auth_source=getattr(user, "auth_source", "local"), groups=[GroupBrief(id=g.id, name=g.name) for g in user.groups], permissions=sorted(user.get_permissions()), created_at=user.created_at.isoformat(), @@ -289,11 +290,49 @@ async def login(request: LoginRequest, db: AsyncSession = Depends(get_db)): detail="Authentication is not enabled", ) - # Try username-based authentication first - user = await authenticate_user(db, request.username, request.password) + # Check if LDAP is enabled + ldap_user = None + ldap_settings = await _get_ldap_settings(db) + if ldap_settings: + try: + from backend.app.services.ldap_service import ( + authenticate_ldap_user, + parse_ldap_config, + ) + + ldap_config = parse_ldap_config(ldap_settings) + if ldap_config: + ldap_user = authenticate_ldap_user(ldap_config, request.username, request.password) + if ldap_user: + # LDAP auth succeeded — find or create local user + user = await get_user_by_username(db, ldap_user.username) + if user and user.auth_source != "ldap": + # Username exists as local user — don't override + user = None + ldap_user = None + elif not user: + if not ldap_config.auto_provision: + # User doesn't exist and auto-provision is off + ldap_user = None + else: + # Auto-provision LDAP user + user = await _provision_ldap_user(db, ldap_user, ldap_config) + + if user and ldap_user: + # Update email and group mappings on each login + await _sync_ldap_user(db, user, ldap_user, ldap_config) + except Exception as e: + import logging + + logging.getLogger(__name__).warning("LDAP authentication error, falling back to local: %s", e) + ldap_user = None + + # Try username-based authentication (skip if already authenticated via LDAP) + if not ldap_user: + user = await authenticate_user(db, request.username, request.password) # If username auth failed and advanced auth is enabled, try email-based authentication - if not user: + if not user and not ldap_user: advanced_auth = await is_advanced_auth_enabled(db) if advanced_auth: user = await authenticate_user_by_email(db, request.username, request.password) @@ -587,8 +626,8 @@ async def forgot_password(request: ForgotPasswordRequest, db: AsyncSession = Dep user = await get_user_by_email(db, request.email) # Always return success message to prevent email enumeration - # but only send email if user exists - if user and user.is_active: + # but only send email if user exists and is not an LDAP user + if user and user.is_active and user.auth_source != "ldap": try: # Generate new password new_password = generate_secure_password() @@ -659,6 +698,12 @@ async def reset_user_password( detail="User not found", ) + if user.auth_source == "ldap": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Cannot reset password for LDAP users — passwords are managed by the LDAP server", + ) + if not user.email: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, @@ -688,3 +733,146 @@ async def reset_user_password( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to reset password: {str(e)}", ) + + +# LDAP Authentication Helpers + + +async def _get_ldap_settings(db: AsyncSession) -> dict[str, str] | None: + """Get LDAP settings from the database. Returns None if LDAP is not enabled.""" + ldap_keys = [ + "ldap_enabled", + "ldap_server_url", + "ldap_bind_dn", + "ldap_bind_password", + "ldap_search_base", + "ldap_user_filter", + "ldap_security", + "ldap_group_mapping", + "ldap_auto_provision", + "ldap_ca_cert_path", + ] + result = await db.execute(select(Settings).where(Settings.key.in_(ldap_keys))) + settings = {s.key: s.value for s in result.scalars().all()} + if settings.get("ldap_enabled", "false").lower() != "true": + return None + return settings + + +async def _provision_ldap_user(db: AsyncSession, ldap_user, ldap_config) -> User: + """Create a new local user from LDAP authentication.""" + import logging + + from backend.app.services.ldap_service import resolve_group_mapping + + logger = logging.getLogger(__name__) + + new_user = User( + username=ldap_user.username, + email=ldap_user.email, + password_hash=None, + role="user", + auth_source="ldap", + is_active=True, + ) + + # Map LDAP groups to BamBuddy groups + mapped_group_names = resolve_group_mapping(ldap_user.groups, ldap_config.group_mapping) + if mapped_group_names: + groups_result = await db.execute(select(Group).where(Group.name.in_(mapped_group_names))) + new_user.groups = list(groups_result.scalars().all()) + + db.add(new_user) + await db.commit() + await db.refresh(new_user) + logger.info("Auto-provisioned LDAP user: %s (groups: %s)", new_user.username, mapped_group_names) + return new_user + + +async def _sync_ldap_user(db: AsyncSession, user: User, ldap_user, ldap_config) -> None: + """Sync LDAP user attributes (email, groups) on each login.""" + import logging + + from backend.app.services.ldap_service import resolve_group_mapping + + logger = logging.getLogger(__name__) + + changed = False + + # Update email if changed + if ldap_user.email and ldap_user.email != user.email: + user.email = ldap_user.email + changed = True + + # Sync group mappings — always update to match LDAP state (including revocation) + mapped_group_names = resolve_group_mapping(ldap_user.groups, ldap_config.group_mapping) + if mapped_group_names: + groups_result = await db.execute(select(Group).where(Group.name.in_(mapped_group_names))) + new_groups = list(groups_result.scalars().all()) + else: + new_groups = [] + current_group_ids = {g.id for g in user.groups} + new_group_ids = {g.id for g in new_groups} + if current_group_ids != new_group_ids: + user.groups = new_groups + changed = True + + if changed: + await db.commit() + logger.info("Synced LDAP user attributes: %s", user.username) + + +@router.post("/ldap/test") +async def test_ldap( + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE), + db: AsyncSession = Depends(get_db), +): + """Test LDAP connection using saved settings (admin only when auth enabled).""" + import logging + + from backend.app.services.ldap_service import parse_ldap_config, test_ldap_connection + + logger = logging.getLogger(__name__) + + ldap_settings = await _get_ldap_settings(db) + if not ldap_settings: + # LDAP might not be enabled yet but settings might still exist — read all keys + ldap_keys = [ + "ldap_enabled", + "ldap_server_url", + "ldap_bind_dn", + "ldap_bind_password", + "ldap_search_base", + "ldap_user_filter", + "ldap_security", + "ldap_group_mapping", + "ldap_auto_provision", + ] + result = await db.execute(select(Settings).where(Settings.key.in_(ldap_keys))) + ldap_settings = {s.key: s.value for s in result.scalars().all()} + # Force enabled for test + ldap_settings["ldap_enabled"] = "true" + + config = parse_ldap_config(ldap_settings) + if not config: + return {"success": False, "message": "LDAP server URL is not configured"} + + success, message = test_ldap_connection(config) + if success: + logger.info("LDAP connection test successful") + else: + logger.warning("LDAP connection test failed: %s", message) + return {"success": success, "message": message} + + +@router.get("/ldap/status") +async def get_ldap_status(db: AsyncSession = Depends(get_db)): + """Get LDAP authentication status.""" + # Only fetch the minimum keys needed — never load secrets + ldap_keys = ["ldap_enabled", "ldap_server_url"] + result = await db.execute(select(Settings).where(Settings.key.in_(ldap_keys))) + settings = {s.key: s.value for s in result.scalars().all()} + return { + "ldap_enabled": settings.get("ldap_enabled", "false").lower() == "true", + "ldap_configured": bool(settings.get("ldap_server_url")), + } diff --git a/backend/app/api/routes/settings.py b/backend/app/api/routes/settings.py index 13a4c4fec..2eaef62cf 100644 --- a/backend/app/api/routes/settings.py +++ b/backend/app/api/routes/settings.py @@ -106,6 +106,8 @@ async def get_settings( "default_vibration_cali", "default_layer_inspect", "default_timelapse", + "ldap_enabled", + "ldap_auto_provision", ]: settings_dict[setting.key] = setting.value.lower() == "true" elif setting.key in [ @@ -139,6 +141,9 @@ async def get_settings( ha_settings = await get_homeassistant_settings(db) settings_dict.update(ha_settings) + # Never return LDAP bind password in API responses + settings_dict["ldap_bind_password"] = "" + return AppSettings(**settings_dict) diff --git a/backend/app/api/routes/users.py b/backend/app/api/routes/users.py index fa3daa8fb..6ae765dd5 100644 --- a/backend/app/api/routes/users.py +++ b/backend/app/api/routes/users.py @@ -38,6 +38,7 @@ def _user_to_response(user: User) -> UserResponse: role=user.role, is_active=user.is_active, is_admin=user.is_admin, + auth_source=getattr(user, "auth_source", "local"), groups=[GroupBrief(id=g.id, name=g.name) for g in user.groups], permissions=sorted(user.get_permissions()), created_at=user.created_at.isoformat(), @@ -253,6 +254,11 @@ async def update_user( user.email = user_data.email if user_data.password is not None: + if getattr(user, "auth_source", "local") == "ldap": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Cannot set password for LDAP users", + ) user.password_hash = get_password_hash(user_data.password) if user_data.role is not None: @@ -402,7 +408,19 @@ async def change_own_password( detail="Authentication required to change password", ) + # Block password change for LDAP users + if getattr(current_user, "auth_source", "local") == "ldap": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Cannot change password for LDAP users — passwords are managed by the LDAP server", + ) + # Verify current password + if not current_user.password_hash: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Account has no local password set", + ) if not verify_password(password_data.current_password, current_user.password_hash): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, diff --git a/backend/app/core/auth.py b/backend/app/core/auth.py index ebe166be4..3548ca793 100644 --- a/backend/app/core/auth.py +++ b/backend/app/core/auth.py @@ -220,7 +220,9 @@ async def authenticate_user(db: AsyncSession, username: str, password: str) -> U user = await get_user_by_username(db, username) if not user: return None - if not verify_password(password, user.password_hash): + if getattr(user, "auth_source", "local") == "ldap": + return None # LDAP users authenticate via LDAP, not local password + if not user.password_hash or not verify_password(password, user.password_hash): return None if not user.is_active: return None @@ -235,7 +237,9 @@ async def authenticate_user_by_email(db: AsyncSession, email: str, password: str user = await get_user_by_email(db, email) if not user: return None - if not verify_password(password, user.password_hash): + if getattr(user, "auth_source", "local") == "ldap": + return None + if not user.password_hash or not verify_password(password, user.password_hash): return None if not user.is_active: return None diff --git a/backend/app/core/database.py b/backend/app/core/database.py index e9782a967..1aa91cc9b 100644 --- a/backend/app/core/database.py +++ b/backend/app/core/database.py @@ -1367,6 +1367,13 @@ async def run_migrations(conn): # Previously both None and [] meant "all printers"; now [] means "no printers" await _safe_execute(conn, "UPDATE api_keys SET printer_ids = NULL WHERE printer_ids = '[]'") + # Migration: Add auth_source column to users for LDAP support (#794) + await _safe_execute(conn, "ALTER TABLE users ADD COLUMN auth_source VARCHAR(20) DEFAULT 'local' NOT NULL") + + # Migration: Make password_hash nullable for LDAP users (#794) + if not is_sqlite(): + await _safe_execute(conn, "ALTER TABLE users ALTER COLUMN password_hash DROP NOT NULL") + # Seed default settings keys that must exist on fresh install default_settings = [ ("advanced_auth_enabled", "false"), diff --git a/backend/app/models/user.py b/backend/app/models/user.py index d720a2bf8..08fd069bd 100644 --- a/backend/app/models/user.py +++ b/backend/app/models/user.py @@ -26,10 +26,11 @@ class User(Base): id: Mapped[int] = mapped_column(primary_key=True) username: Mapped[str] = mapped_column(String(100), unique=True, index=True) email: Mapped[str | None] = mapped_column(String(255), unique=True, index=True, nullable=True) - password_hash: Mapped[str] = mapped_column(String(255)) + password_hash: Mapped[str | None] = mapped_column(String(255), nullable=True) role: Mapped[str] = mapped_column( String(20), default="user" ) # "admin" or "user" (legacy, kept for backward compat) + auth_source: Mapped[str] = mapped_column(String(20), default="local") # "local" or "ldap" is_active: Mapped[bool] = mapped_column(default=True) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now()) diff --git a/backend/app/schemas/auth.py b/backend/app/schemas/auth.py index 0e70a5df3..543577160 100644 --- a/backend/app/schemas/auth.py +++ b/backend/app/schemas/auth.py @@ -46,6 +46,7 @@ class UserResponse(BaseModel): role: str # Deprecated, kept for backward compatibility is_active: bool is_admin: bool # Computed from role and group membership + auth_source: str = "local" # "local" or "ldap" groups: list[GroupBrief] = [] permissions: list[str] = [] # All permissions from groups created_at: str diff --git a/backend/app/schemas/settings.py b/backend/app/schemas/settings.py index cd9848d51..da4c15cfc 100644 --- a/backend/app/schemas/settings.py +++ b/backend/app/schemas/settings.py @@ -225,6 +225,26 @@ class AppSettings(BaseModel): description="Shortest Job First — scheduler prioritizes shorter print jobs over longer ones", ) + # LDAP authentication (#794) + ldap_enabled: bool = Field(default=False, description="Enable LDAP authentication") + ldap_server_url: str = Field(default="", description="LDAP server URL (e.g., ldap://ldap.example.com:389)") + ldap_bind_dn: str = Field(default="", description="Bind DN for LDAP searches (e.g., cn=admin,dc=example,dc=com)") + ldap_bind_password: str = Field(default="", description="Bind password for LDAP searches") + ldap_search_base: str = Field(default="", description="Search base DN (e.g., ou=users,dc=example,dc=com)") + ldap_user_filter: str = Field( + default="(sAMAccountName={username})", + description="LDAP user search filter. {username} is replaced with the login username", + ) + ldap_security: str = Field(default="starttls", description="LDAP security: 'starttls' or 'ldaps'") + ldap_group_mapping: str = Field( + default="", + description="JSON: LDAP group to BamBuddy group mapping {ldap_group_dn: bambuddy_group_name}", + ) + ldap_auto_provision: bool = Field( + default=False, + description="Auto-create BamBuddy user on first successful LDAP login", + ) + # Default sidebar order (admin-set for all users) default_sidebar_order: str = Field( default="", @@ -310,6 +330,15 @@ class AppSettingsUpdate(BaseModel): require_plate_clear: bool | None = None queue_shortest_first: bool | None = None gcode_snippets: str | None = None + ldap_enabled: bool | None = None + ldap_server_url: str | None = None + ldap_bind_dn: str | None = None + ldap_bind_password: str | None = None + ldap_search_base: str | None = None + ldap_user_filter: str | None = None + ldap_security: str | None = None + ldap_group_mapping: str | None = None + ldap_auto_provision: bool | None = None default_sidebar_order: str | None = None @field_validator("gcode_snippets") @@ -325,6 +354,19 @@ class AppSettingsUpdate(BaseModel): raise ValueError("gcode_snippets must be a JSON object keyed by printer model") return v + @field_validator("ldap_group_mapping") + @classmethod + def validate_ldap_group_mapping(cls, v: str | None) -> str | None: + if v is None or v == "": + return v + try: + parsed = json.loads(v) + except json.JSONDecodeError: + raise ValueError("ldap_group_mapping must be valid JSON or empty") + if not isinstance(parsed, dict): + raise ValueError("ldap_group_mapping must be a JSON object mapping LDAP group DNs to BamBuddy group names") + return v + @field_validator("default_sidebar_order") @classmethod def validate_default_sidebar_order(cls, v: str | None) -> str | None: diff --git a/backend/app/services/ldap_service.py b/backend/app/services/ldap_service.py new file mode 100644 index 000000000..68267a037 --- /dev/null +++ b/backend/app/services/ldap_service.py @@ -0,0 +1,265 @@ +"""LDAP authentication service for BamBuddy (#794). + +Supports: +- LDAP bind authentication (simple bind with user's credentials) +- StartTLS, LDAPS, and plaintext connections +- User search with configurable filter +- Group membership resolution for role mapping +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass + +from ldap3 import ALL, SUBTREE, Connection, Server, Tls + +logger = logging.getLogger(__name__) + + +@dataclass +class LDAPUserInfo: + """User information retrieved from LDAP after successful authentication.""" + + username: str + email: str | None + display_name: str | None + groups: list[str] # List of group DNs the user belongs to + + +@dataclass +class LDAPConfig: + """LDAP configuration parsed from settings.""" + + server_url: str + bind_dn: str + bind_password: str + search_base: str + user_filter: str # e.g. "(sAMAccountName={username})" + security: str # "none", "starttls", "ldaps" + group_mapping: dict[str, str] # LDAP group DN -> BamBuddy group name + auto_provision: bool + ca_cert_path: str # Path to CA certificate file (empty = skip verification) + + +def parse_ldap_config(settings: dict[str, str]) -> LDAPConfig | None: + """Parse LDAP config from settings key-value pairs. Returns None if LDAP not enabled.""" + if settings.get("ldap_enabled", "false").lower() != "true": + return None + + server_url = settings.get("ldap_server_url", "").strip() + if not server_url: + return None + + group_mapping_raw = settings.get("ldap_group_mapping", "") + try: + group_mapping = json.loads(group_mapping_raw) if group_mapping_raw else {} + except json.JSONDecodeError: + group_mapping = {} + + return LDAPConfig( + server_url=server_url, + bind_dn=settings.get("ldap_bind_dn", "").strip(), + bind_password=settings.get("ldap_bind_password", ""), + search_base=settings.get("ldap_search_base", "").strip(), + user_filter=settings.get("ldap_user_filter", "(sAMAccountName={username})").strip(), + security=settings.get("ldap_security", "starttls").strip(), + group_mapping=group_mapping if isinstance(group_mapping, dict) else {}, + auto_provision=settings.get("ldap_auto_provision", "false").lower() == "true", + ca_cert_path=settings.get("ldap_ca_cert_path", "").strip(), + ) + + +def _create_server(config: LDAPConfig) -> Server: + """Create an ldap3 Server instance from config. + + Always uses TLS — either LDAPS (TLS from start) or StartTLS (upgrade after connect). + Plaintext LDAP is not supported. + """ + import ssl + + use_ssl = config.security == "ldaps" or config.server_url.startswith("ldaps://") + + if config.ca_cert_path: + tls = Tls(validate=ssl.CERT_REQUIRED, ca_certs_file=config.ca_cert_path) + else: + tls = Tls(validate=ssl.CERT_NONE) + + return Server(config.server_url, use_ssl=use_ssl, tls=tls, get_info=ALL, connect_timeout=10) + + +def authenticate_ldap_user(config: LDAPConfig, username: str, password: str) -> LDAPUserInfo | None: + """Authenticate a user via LDAP bind. + + 1. Bind with service account to search for the user DN + 2. Attempt bind with the user's DN and provided password + 3. On success, retrieve user attributes and group memberships + + Returns LDAPUserInfo on success, None on failure. + """ + if not password: + return None + + server = _create_server(config) + + # Step 1: Service account bind + user search + try: + service_conn = Connection( + server, + user=config.bind_dn, + password=config.bind_password, + auto_bind=False, + raise_exceptions=True, + read_only=True, + ) + service_conn.open() + if config.security == "starttls" and not config.server_url.startswith("ldaps://"): + service_conn.start_tls() + service_conn.bind() + except Exception as e: + logger.warning("LDAP service account bind failed: %s", e) + return None + + try: + # Search for the user + search_filter = config.user_filter.replace("{username}", _ldap_escape(username)) + service_conn.search( + search_base=config.search_base, + search_filter=search_filter, + search_scope=SUBTREE, + attributes=["*"], + ) + + if not service_conn.entries: + logger.info("LDAP user not found: %s", username) + return None + + user_entry = service_conn.entries[0] + user_dn = str(user_entry.entry_dn) + + # Step 2: Bind as the user to verify password + try: + user_conn = Connection( + server, + user=user_dn, + password=password, + auto_bind=False, + raise_exceptions=True, + read_only=True, + ) + user_conn.open() + if config.security == "starttls" and not config.server_url.startswith("ldaps://"): + user_conn.start_tls() + user_conn.bind() + user_conn.unbind() + except Exception as e: + logger.info("LDAP bind failed for user %s: %s", username, e) + return None + + # Step 3: Extract user info + email = str(user_entry.mail) if hasattr(user_entry, "mail") and user_entry.mail else None + display_name = ( + str(user_entry.displayName) if hasattr(user_entry, "displayName") and user_entry.displayName else None + ) + + # Collect groups from memberOf attribute (Active Directory / groupOfNames) + groups = ( + [str(g) for g in user_entry.memberOf] if hasattr(user_entry, "memberOf") and user_entry.memberOf else [] + ) + + # Also search for Posix groups (memberUid-based) using the service account + canonical_username = username + if hasattr(user_entry, "sAMAccountName") and user_entry.sAMAccountName: + canonical_username = str(user_entry.sAMAccountName) + elif hasattr(user_entry, "uid") and user_entry.uid: + canonical_username = str(user_entry.uid) + + posix_filter = f"(&(objectClass=posixGroup)(memberUid={_ldap_escape(canonical_username)}))" + service_conn.search( + search_base=config.search_base, + search_filter=posix_filter, + search_scope=SUBTREE, + attributes=["cn"], + ) + for entry in service_conn.entries: + groups.append(str(entry.entry_dn)) + + logger.info( + "LDAP authentication successful for user: %s (DN: %s, groups: %d)", canonical_username, user_dn, len(groups) + ) + + return LDAPUserInfo( + username=canonical_username, + email=email, + display_name=display_name, + groups=groups, + ) + finally: + service_conn.unbind() + + +def resolve_group_mapping(ldap_groups: list[str], group_mapping: dict[str, str]) -> list[str]: + """Map LDAP group DNs to BamBuddy group names. + + Returns list of BamBuddy group names that the user should be added to. + Comparison is case-insensitive on the LDAP group DN. + """ + if not group_mapping: + return [] + + # Build case-insensitive lookup + mapping_lower = {k.lower(): v for k, v in group_mapping.items()} + result = [] + for ldap_group in ldap_groups: + bambuddy_group = mapping_lower.get(ldap_group.lower()) + if bambuddy_group: + result.append(bambuddy_group) + return result + + +def test_ldap_connection(config: LDAPConfig) -> tuple[bool, str]: + """Test LDAP connection and service account bind. + + Returns (success, message). + """ + try: + server = _create_server(config) + conn = Connection( + server, + user=config.bind_dn, + password=config.bind_password, + auto_bind=False, + raise_exceptions=True, + read_only=True, + ) + conn.open() + if config.security == "starttls" and not config.server_url.startswith("ldaps://"): + conn.start_tls() + conn.bind() + + # Try a search to verify search base + conn.search( + search_base=config.search_base, + search_filter="(objectClass=*)", + search_scope=SUBTREE, + size_limit=1, + ) + conn.unbind() + return True, "LDAP connection successful" + except Exception as e: + return False, f"LDAP connection failed: {e}" + + +def _ldap_escape(value: str) -> str: + """Escape special characters in LDAP search filter values (RFC 4515).""" + replacements = { + "\\": "\\5c", + "*": "\\2a", + "(": "\\28", + ")": "\\29", + "\x00": "\\00", + } + for char, escaped in replacements.items(): + value = value.replace(char, escaped) + return value diff --git a/backend/tests/unit/services/test_ldap_service.py b/backend/tests/unit/services/test_ldap_service.py new file mode 100644 index 000000000..f418944bb --- /dev/null +++ b/backend/tests/unit/services/test_ldap_service.py @@ -0,0 +1,230 @@ +"""Tests for LDAP authentication service (#794). + +Tests the pure logic functions in ldap_service.py: +- Config parsing from settings dict +- LDAP filter escaping (RFC 4515) +- Group mapping resolution +- LDAPConfig/LDAPUserInfo dataclass construction + +Network-dependent functions (authenticate_ldap_user, test_ldap_connection) +are not tested here — they require a live LDAP server. +""" + +import pytest + +from backend.app.services.ldap_service import ( + LDAPConfig, + LDAPUserInfo, + _ldap_escape, + parse_ldap_config, + resolve_group_mapping, +) + + +class TestParseConfig: + """Verify parse_ldap_config builds LDAPConfig from settings dict.""" + + def test_returns_none_when_disabled(self): + settings = {"ldap_enabled": "false", "ldap_server_url": "ldaps://example.com"} + assert parse_ldap_config(settings) is None + + def test_returns_none_when_missing_enabled(self): + settings = {"ldap_server_url": "ldaps://example.com"} + assert parse_ldap_config(settings) is None + + def test_returns_none_when_no_server_url(self): + settings = {"ldap_enabled": "true", "ldap_server_url": ""} + assert parse_ldap_config(settings) is None + + def test_returns_none_when_server_url_whitespace(self): + settings = {"ldap_enabled": "true", "ldap_server_url": " "} + assert parse_ldap_config(settings) is None + + def test_parses_minimal_config(self): + settings = { + "ldap_enabled": "true", + "ldap_server_url": "ldaps://ldap.example.com:636", + } + config = parse_ldap_config(settings) + assert config is not None + assert config.server_url == "ldaps://ldap.example.com:636" + assert config.bind_dn == "" + assert config.search_base == "" + assert config.user_filter == "(sAMAccountName={username})" + assert config.security == "starttls" + assert config.group_mapping == {} + assert config.auto_provision is False + assert config.ca_cert_path == "" + + def test_parses_full_config(self): + settings = { + "ldap_enabled": "true", + "ldap_server_url": "ldaps://ldap.example.com:636", + "ldap_bind_dn": "cn=admin,dc=example,dc=com", + "ldap_bind_password": "secret", + "ldap_search_base": "ou=users,dc=example,dc=com", + "ldap_user_filter": "(uid={username})", + "ldap_security": "ldaps", + "ldap_group_mapping": '{"cn=admins,dc=example,dc=com": "Administrators"}', + "ldap_auto_provision": "true", + "ldap_ca_cert_path": "/path/to/ca.pem", + } + config = parse_ldap_config(settings) + assert config is not None + assert config.bind_dn == "cn=admin,dc=example,dc=com" + assert config.bind_password == "secret" + assert config.search_base == "ou=users,dc=example,dc=com" + assert config.user_filter == "(uid={username})" + assert config.security == "ldaps" + assert config.group_mapping == {"cn=admins,dc=example,dc=com": "Administrators"} + assert config.auto_provision is True + assert config.ca_cert_path == "/path/to/ca.pem" + + def test_handles_invalid_group_mapping_json(self): + settings = { + "ldap_enabled": "true", + "ldap_server_url": "ldaps://ldap.example.com", + "ldap_group_mapping": "not valid json", + } + config = parse_ldap_config(settings) + assert config is not None + assert config.group_mapping == {} + + def test_handles_non_dict_group_mapping(self): + settings = { + "ldap_enabled": "true", + "ldap_server_url": "ldaps://ldap.example.com", + "ldap_group_mapping": '["not", "a", "dict"]', + } + config = parse_ldap_config(settings) + assert config is not None + assert config.group_mapping == {} + + def test_enabled_case_insensitive(self): + settings = {"ldap_enabled": "True", "ldap_server_url": "ldaps://ldap.example.com"} + assert parse_ldap_config(settings) is not None + + settings = {"ldap_enabled": "TRUE", "ldap_server_url": "ldaps://ldap.example.com"} + assert parse_ldap_config(settings) is not None + + def test_strips_whitespace(self): + settings = { + "ldap_enabled": "true", + "ldap_server_url": " ldaps://ldap.example.com ", + "ldap_bind_dn": " cn=admin,dc=example,dc=com ", + "ldap_search_base": " dc=example,dc=com ", + } + config = parse_ldap_config(settings) + assert config.server_url == "ldaps://ldap.example.com" + assert config.bind_dn == "cn=admin,dc=example,dc=com" + assert config.search_base == "dc=example,dc=com" + + +class TestLDAPEscape: + """Verify RFC 4515 escaping for LDAP search filter values.""" + + def test_plain_string(self): + assert _ldap_escape("testuser") == "testuser" + + def test_escapes_backslash(self): + assert _ldap_escape("test\\user") == "test\\5cuser" + + def test_escapes_asterisk(self): + assert _ldap_escape("test*user") == "test\\2auser" + + def test_escapes_open_paren(self): + assert _ldap_escape("test(user") == "test\\28user" + + def test_escapes_close_paren(self): + assert _ldap_escape("test)user") == "test\\29user" + + def test_escapes_null(self): + assert _ldap_escape("test\x00user") == "test\\00user" + + def test_escapes_multiple_chars(self): + assert _ldap_escape("a*b(c)d\\e") == "a\\2ab\\28c\\29d\\5ce" + + def test_empty_string(self): + assert _ldap_escape("") == "" + + +class TestResolveGroupMapping: + """Verify LDAP group DN to BamBuddy group name resolution.""" + + def test_empty_mapping(self): + assert resolve_group_mapping(["cn=admins,dc=example"], {}) == [] + + def test_empty_groups(self): + mapping = {"cn=admins,dc=example": "Administrators"} + assert resolve_group_mapping([], mapping) == [] + + def test_single_match(self): + mapping = {"cn=admins,dc=example,dc=com": "Administrators"} + groups = ["cn=admins,dc=example,dc=com"] + assert resolve_group_mapping(groups, mapping) == ["Administrators"] + + def test_multiple_matches(self): + mapping = { + "cn=admins,dc=example,dc=com": "Administrators", + "cn=ops,dc=example,dc=com": "Operators", + } + groups = ["cn=admins,dc=example,dc=com", "cn=ops,dc=example,dc=com"] + result = resolve_group_mapping(groups, mapping) + assert set(result) == {"Administrators", "Operators"} + + def test_no_match(self): + mapping = {"cn=admins,dc=example,dc=com": "Administrators"} + groups = ["cn=users,dc=example,dc=com"] + assert resolve_group_mapping(groups, mapping) == [] + + def test_case_insensitive_dn(self): + mapping = {"CN=Admins,DC=Example,DC=Com": "Administrators"} + groups = ["cn=admins,dc=example,dc=com"] + assert resolve_group_mapping(groups, mapping) == ["Administrators"] + + def test_partial_match_not_matched(self): + mapping = {"cn=admins,dc=example,dc=com": "Administrators"} + groups = ["cn=admins,dc=other,dc=com"] + assert resolve_group_mapping(groups, mapping) == [] + + def test_extra_groups_ignored(self): + mapping = {"cn=admins,dc=example,dc=com": "Administrators"} + groups = ["cn=admins,dc=example,dc=com", "cn=users,dc=example,dc=com", "cn=devs,dc=example,dc=com"] + assert resolve_group_mapping(groups, mapping) == ["Administrators"] + + +class TestDataclasses: + """Verify dataclass construction.""" + + def test_ldap_user_info(self): + info = LDAPUserInfo( + username="testuser", + email="test@example.com", + display_name="Test User", + groups=["cn=admins,dc=example,dc=com"], + ) + assert info.username == "testuser" + assert info.email == "test@example.com" + assert info.display_name == "Test User" + assert info.groups == ["cn=admins,dc=example,dc=com"] + + def test_ldap_user_info_none_fields(self): + info = LDAPUserInfo(username="testuser", email=None, display_name=None, groups=[]) + assert info.email is None + assert info.display_name is None + assert info.groups == [] + + def test_ldap_config(self): + config = LDAPConfig( + server_url="ldaps://ldap.example.com:636", + bind_dn="cn=admin,dc=example,dc=com", + bind_password="secret", + search_base="dc=example,dc=com", + user_filter="(uid={username})", + security="ldaps", + group_mapping={"cn=admins": "Administrators"}, + auto_provision=True, + ca_cert_path="", + ) + assert config.server_url == "ldaps://ldap.example.com:636" + assert config.auto_provision is True diff --git a/frontend/src/__tests__/components/AssignSpoolModal.test.tsx b/frontend/src/__tests__/components/AssignSpoolModal.test.tsx index 4b8d9cf20..594145e89 100644 --- a/frontend/src/__tests__/components/AssignSpoolModal.test.tsx +++ b/frontend/src/__tests__/components/AssignSpoolModal.test.tsx @@ -34,6 +34,7 @@ const manualSpool = { weight_used: 0, tag_uid: null, tray_uuid: null, + slicer_filament_name: 'PLA', }; const blSpool = { @@ -47,11 +48,12 @@ const blSpool = { weight_used: 50, tag_uid: '05CC1E0F00000100', tray_uuid: 'A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4', + slicer_filament_name: 'PLA', }; const anotherManualSpool = { id: 3, - material: 'PETG', + material: 'PLA', subtype: 'HF', brand: 'Overture', color_name: 'Black', @@ -60,6 +62,7 @@ const anotherManualSpool = { weight_used: 200, tag_uid: null, tray_uuid: null, + slicer_filament_name: 'PLA', }; describe('AssignSpoolModal', () => { diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 1aa54ecc6..97c6ffb9a 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -903,6 +903,16 @@ export interface AppSettings { queue_shortest_first: boolean; // Default sidebar order (admin-set for all users) default_sidebar_order: string; + // LDAP authentication + ldap_enabled: boolean; + ldap_server_url: string; + ldap_bind_dn: string; + ldap_bind_password: string; + ldap_search_base: string; + ldap_user_filter: string; + ldap_security: string; + ldap_group_mapping: string; + ldap_auto_provision: boolean; } export type AppSettingsUpdate = Partial; @@ -2275,6 +2285,7 @@ export interface UserResponse { role: string; // Deprecated, kept for backward compatibility is_active: boolean; is_admin: boolean; // Computed from role and group membership + auth_source: string; // "local" or "ldap" groups: GroupBrief[]; permissions: Permission[]; // All permissions from groups created_at: string; @@ -2344,6 +2355,16 @@ export interface AdvancedAuthStatus { smtp_configured: boolean; } +export interface LDAPStatus { + ldap_enabled: boolean; + ldap_configured: boolean; +} + +export interface LDAPTestResponse { + success: boolean; + message: string; +} + export interface SetupResponse { auth_enabled: boolean; admin_created?: boolean; @@ -2399,6 +2420,12 @@ export const api = { method: 'POST', }), getAdvancedAuthStatus: () => request('/auth/advanced-auth/status'), + // LDAP Authentication + getLDAPStatus: () => request('/auth/ldap/status'), + testLDAP: () => + request('/auth/ldap/test', { + method: 'POST', + }), forgotPassword: (data: ForgotPasswordRequest) => request('/auth/forgot-password', { method: 'POST', diff --git a/frontend/src/components/LDAPSettings.tsx b/frontend/src/components/LDAPSettings.tsx new file mode 100644 index 000000000..7026944c7 --- /dev/null +++ b/frontend/src/components/LDAPSettings.tsx @@ -0,0 +1,418 @@ +import { useState, useEffect } from 'react'; +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; +import { useTranslation } from 'react-i18next'; +import { Shield, Lock, Unlock, AlertTriangle, CheckCircle, Loader2, Send } from 'lucide-react'; +import { api } from '../api/client'; +import type { AppSettings } from '../api/client'; +import { Card, CardContent, CardHeader } from './Card'; +import { Button } from './Button'; +import { useToast } from '../contexts/ToastContext'; +import { useAuth } from '../contexts/AuthContext'; + +const SECURITY_PORT_MAP: Record = { + starttls: '389', + ldaps: '636', +}; + +interface LDAPFormState { + ldap_server_url: string; + ldap_bind_dn: string; + ldap_bind_password: string; + ldap_search_base: string; + ldap_user_filter: string; + ldap_security: string; + ldap_group_mapping: string; + ldap_auto_provision: boolean; +} + +export function LDAPSettings() { + const { t } = useTranslation(); + const { showToast } = useToast(); + const queryClient = useQueryClient(); + const { authEnabled } = useAuth(); + + const [form, setForm] = useState({ + ldap_server_url: '', + ldap_bind_dn: '', + ldap_bind_password: '', + ldap_search_base: '', + ldap_user_filter: '(sAMAccountName={username})', + ldap_security: 'starttls', + ldap_group_mapping: '', + ldap_auto_provision: false, + }); + + // Fetch settings + const { data: settings, isLoading } = useQuery({ + queryKey: ['settings'], + queryFn: () => api.getSettings(), + }); + + // Fetch LDAP status + const { data: ldapStatus } = useQuery({ + queryKey: ['ldapStatus'], + queryFn: () => api.getLDAPStatus(), + }); + + // Fetch groups for mapping display + const { data: groups = [] } = useQuery({ + queryKey: ['groups'], + queryFn: () => api.getGroups(), + }); + + // Load settings into form + useEffect(() => { + if (settings) { + setForm({ + ldap_server_url: settings.ldap_server_url || '', + ldap_bind_dn: settings.ldap_bind_dn || '', + ldap_bind_password: '', // Never show password + ldap_search_base: settings.ldap_search_base || '', + ldap_user_filter: settings.ldap_user_filter || '(sAMAccountName={username})', + ldap_security: settings.ldap_security || 'starttls', + ldap_group_mapping: settings.ldap_group_mapping || '', + ldap_auto_provision: settings.ldap_auto_provision ?? false, + }); + } + }, [settings]); + + // Save settings + const saveMutation = useMutation({ + mutationFn: (data: Partial) => api.updateSettings(data), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: ['settings'] }); + queryClient.invalidateQueries({ queryKey: ['ldapStatus'] }); + showToast(t('settings.ldap.settingsSaved') || 'LDAP settings saved', 'success'); + }, + onError: (error: Error) => { + showToast(error.message, 'error'); + }, + }); + + // Toggle LDAP + const toggleMutation = useMutation({ + mutationFn: (enabled: boolean) => api.updateSettings({ ldap_enabled: enabled }), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: ['settings'] }); + queryClient.invalidateQueries({ queryKey: ['ldapStatus'] }); + showToast( + ldapStatus?.ldap_enabled + ? (t('settings.ldap.disabled') || 'LDAP authentication disabled') + : (t('settings.ldap.enabled') || 'LDAP authentication enabled'), + 'success' + ); + }, + onError: (error: Error) => { + showToast(error.message, 'error'); + }, + }); + + // Test connection + const testMutation = useMutation({ + mutationFn: () => api.testLDAP(), + onSuccess: (data: { success: boolean; message: string }) => { + showToast(data.message, data.success ? 'success' : 'error'); + }, + onError: (error: Error) => { + showToast(error.message, 'error'); + }, + }); + + const handleSave = () => { + if (!form.ldap_server_url) { + showToast(t('settings.ldap.errors.serverRequired') || 'LDAP server URL is required', 'error'); + return; + } + if (!form.ldap_search_base) { + showToast(t('settings.ldap.errors.searchBaseRequired') || 'Search base DN is required', 'error'); + return; + } + + // Build the update payload — only include password if user entered one + const update: Record = { + ldap_server_url: form.ldap_server_url, + ldap_bind_dn: form.ldap_bind_dn, + ldap_search_base: form.ldap_search_base, + ldap_user_filter: form.ldap_user_filter, + ldap_security: form.ldap_security, + ldap_group_mapping: form.ldap_group_mapping, + ldap_auto_provision: form.ldap_auto_provision, + }; + if (form.ldap_bind_password) { + update.ldap_bind_password = form.ldap_bind_password; + } + saveMutation.mutate(update as Partial); + }; + + const handleToggle = () => { + if (!authEnabled) { + showToast(t('settings.ldap.errors.enableAuthFirst') || 'Enable authentication first', 'error'); + return; + } + if (!ldapStatus?.ldap_enabled && !ldapStatus?.ldap_configured) { + showToast(t('settings.ldap.errors.configureLdapFirst') || 'Save LDAP settings first', 'error'); + return; + } + toggleMutation.mutate(!ldapStatus?.ldap_enabled); + }; + + if (isLoading) { + return ( +
+ +
+ ); + } + + const ldapEnabled = ldapStatus?.ldap_enabled ?? false; + const inputClasses = "w-full px-4 py-3 bg-bambu-dark-secondary border border-bambu-dark-tertiary rounded-lg text-white placeholder-bambu-gray focus:outline-none focus:ring-2 focus:ring-bambu-green/50 focus:border-bambu-green transition-colors"; + + return ( +
+ {/* LDAP Toggle */} + + +
+
+ +

+ {t('settings.ldap.title') || 'LDAP Authentication'} +

+
+ +
+ + + {ldapEnabled ? ( +
+
+ +
+

+ {t('settings.ldap.enabledDesc') || 'LDAP authentication is enabled'} +

+
    +
  • {t('settings.ldap.feature1') || 'Users can login with LDAP credentials'}
    • +
  • {t('settings.ldap.feature2') || 'Local admin account remains as fallback'}
    • +
  • {t('settings.ldap.feature3') || 'LDAP groups are mapped to BamBuddy groups on login'}
    • +
+
+
+
+ ) : ( +
+
+ +
+

+ {t('settings.ldap.disabledDesc') || 'LDAP authentication is disabled'} +

+

+ {t('settings.ldap.disabledHint') || 'Configure and save LDAP settings below, then enable.'} +

+
+
+
+ )} + + + + {/* LDAP Server Configuration */} + + +

+ {t('settings.ldap.serverConfig') || 'LDAP Server Configuration'} +

+ + +
+ {/* Server URL */} +
+ + setForm({ ...form, ldap_server_url: e.target.value })} + /> +

+ {t('settings.ldap.serverUrlHint') || 'Use ldaps:// for SSL or ldap:// with StartTLS'} +

+
+ + {/* Security */} +
+ +
+ {(['starttls', 'ldaps'] as const).map(sec => ( + + ))} +
+

+ {t('settings.ldap.securityHint') || `Default port: ${SECURITY_PORT_MAP[form.ldap_security]}`} +

+
+ + {/* Bind DN */} +
+ + setForm({ ...form, ldap_bind_dn: e.target.value })} + /> +
+ + {/* Bind Password */} +
+ + setForm({ ...form, ldap_bind_password: e.target.value })} + /> +
+ + {/* Search Base */} +
+ + setForm({ ...form, ldap_search_base: e.target.value })} + /> +
+ + {/* User Filter */} +
+ + setForm({ ...form, ldap_user_filter: e.target.value })} + /> +

+ {t('settings.ldap.userFilterHint') || '{username} is replaced with the login username. Use (uid={username}) for OpenLDAP.'} +

+
+ + {/* Auto Provision */} +
+
+ +

+ {t('settings.ldap.autoProvisionHint') || 'Automatically create a BamBuddy account on first LDAP login'} +

+
+ +
+ + {/* Group Mapping */} +
+ +