feat(auth): add LDAP directory authentication

Introduce AUTH_METHOD values ldap and all, with LDAP_* environment settings, ldap3-based LDAPService (search, optional groupOfNames checks, user bind, DB sync), and users.auth_provider (local|oidc|ldap) via migration 153_add_user_auth_provider.

Login supports LDAP-only and combined all (local then LDAP where appropriate); OIDC callback sets auth_provider. Forgot/reset/change password flows skip LDAP-managed accounts. Admin System Settings gains a read-only LDAP summary and POST /admin/ldap/test. Production env validation requires core LDAP variables when LDAP is enabled; OIDC registration and docs recognize all.

Documentation: new docs/admin/configuration/LDAP_SETUP.md; updates to OIDC_SETUP, GETTING_STARTED, Docker guides, Render deploy notes, docs README, and CHANGELOG. Tests: tests/test_ldap_auth.py; test_oidc_logout allows auth_method all.
This commit is contained in:
Dries Peeters
2026-04-27 19:08:08 +02:00
parent 8fc823c252
commit e34a668ddc
25 changed files with 1199 additions and 150 deletions
+3
View File
@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- **LDAP authentication** — Optional directory login via `AUTH_METHOD=ldap` or combined `AUTH_METHOD=all` (with local + OIDC). New `LDAP_*` settings in `app/config.py`, `LDAPService` (`app/services/ldap_service.py`), login and password-reset behaviour keyed off `users.auth_provider` (`local` | `oidc` | `ldap`), admin **System Settings** LDAP panel and `POST /admin/ldap/test`, production env validation for required LDAP variables, Alembic `153_add_user_auth_provider`, and tests in `tests/test_ldap_auth.py`. Dependency: `ldap3`. Documentation: [docs/admin/configuration/LDAP_SETUP.md](docs/admin/configuration/LDAP_SETUP.md); OIDC and getting-started guides updated for `ldap` / `all`.
### Fixed
- **API integration test for project tasks** — `tests/test_api_comprehensive.py` now matches `GET /api/projects/<id>/tasks`, which returns **all** tasks (including done and cancelled) for the time-entry UI.
- **Quote create returned HTTP 500 after save (#583)** — The quote was saved, but the redirect to the quote detail page crashed when **Valid until** was set: the template compared `valid_until` to `now()`, and `now` was never defined in the Jinja context. The expired badge now uses `Quote.is_expired` (same rule, app timezone). Regression coverage in `tests/test_routes/test_quotes_web.py` posts `valid_until` so the view path is exercised.
+3 -1
View File
@@ -1009,12 +1009,14 @@ def create_app(config=None):
initialize_ip_cache(ip_cache_ttl)
# Register OAuth OIDC client if enabled
from app.utils.auth_method import auth_includes_oidc
try:
auth_method = (app.config.get("AUTH_METHOD") or "local").strip().lower()
except Exception:
auth_method = "local"
if auth_method in ("oidc", "both"):
if auth_includes_oidc(auth_method):
issuer = app.config.get("OIDC_ISSUER")
client_id = app.config.get("OIDC_CLIENT_ID")
client_secret = app.config.get("OIDC_CLIENT_SECRET")
+30 -4
View File
@@ -57,14 +57,40 @@ class Config:
API_TOKEN_RATE_LIMIT_PER_MINUTE = int(os.getenv("API_TOKEN_RATE_LIMIT_PER_MINUTE", "100"))
API_TOKEN_RATE_LIMIT_PER_HOUR = int(os.getenv("API_TOKEN_RATE_LIMIT_PER_HOUR", "1000"))
# Authentication method: 'none' | 'local' | 'oidc' | 'both'
# Authentication method: 'none' | 'local' | 'oidc' | 'ldap' | 'both' | 'all'
# 'none' = no password authentication (username only)
# 'local' = password authentication required
# 'oidc' = OIDC/Single Sign-On only
# 'both' = OIDC + local password authentication
AUTH_METHOD = os.getenv("AUTH_METHOD", "local").strip().lower()
# 'ldap' = LDAP bind only
# 'both' = OIDC + local password (backwards compatible)
# 'all' = local + OIDC + LDAP
_auth_method_raw = os.getenv("AUTH_METHOD", "local").strip().lower()
_auth_method_valid = frozenset({"none", "local", "oidc", "ldap", "both", "all"})
AUTH_METHOD = _auth_method_raw if _auth_method_raw in _auth_method_valid else "local"
# OIDC settings (used when AUTH_METHOD is 'oidc' or 'both')
# LDAP settings (used when AUTH_METHOD is 'ldap' or 'all')
LDAP_ENABLED = AUTH_METHOD in ("ldap", "all")
LDAP_HOST = os.environ.get("LDAP_HOST", "localhost")
LDAP_PORT = int(os.environ.get("LDAP_PORT", "389"))
LDAP_USE_SSL = os.environ.get("LDAP_USE_SSL", "false").lower() == "true"
LDAP_USE_TLS = os.environ.get("LDAP_USE_TLS", "false").lower() == "true"
LDAP_BIND_DN = os.environ.get("LDAP_BIND_DN", "")
LDAP_BIND_PASSWORD = os.environ.get("LDAP_BIND_PASSWORD", "")
LDAP_BASE_DN = os.environ.get("LDAP_BASE_DN", "dc=example,dc=com")
LDAP_USER_DN = os.environ.get("LDAP_USER_DN", "ou=users")
LDAP_USER_OBJECT_CLASS = os.environ.get("LDAP_USER_OBJECT_CLASS", "inetOrgPerson")
LDAP_USER_LOGIN_ATTR = os.environ.get("LDAP_USER_LOGIN_ATTR", "uid")
LDAP_USER_EMAIL_ATTR = os.environ.get("LDAP_USER_EMAIL_ATTR", "mail")
LDAP_USER_FNAME_ATTR = os.environ.get("LDAP_USER_FNAME_ATTR", "givenName")
LDAP_USER_LNAME_ATTR = os.environ.get("LDAP_USER_LNAME_ATTR", "sn")
LDAP_GROUP_DN = os.environ.get("LDAP_GROUP_DN", "ou=groups")
LDAP_GROUP_OBJECT_CLASS = os.environ.get("LDAP_GROUP_OBJECT_CLASS", "groupOfNames")
LDAP_ADMIN_GROUP = os.environ.get("LDAP_ADMIN_GROUP", "")
LDAP_REQUIRED_GROUP = os.environ.get("LDAP_REQUIRED_GROUP", "")
LDAP_TLS_CA_CERT_FILE = os.environ.get("LDAP_TLS_CA_CERT_FILE", "")
LDAP_TIMEOUT = int(os.environ.get("LDAP_TIMEOUT", "10"))
# OIDC settings (used when AUTH_METHOD is 'oidc', 'both', or 'all')
OIDC_ISSUER = os.getenv("OIDC_ISSUER") # e.g., https://login.microsoftonline.com/<tenant>/v2.0
OIDC_CLIENT_ID = os.getenv("OIDC_CLIENT_ID")
OIDC_CLIENT_SECRET = os.getenv("OIDC_CLIENT_SECRET")
+2
View File
@@ -28,6 +28,8 @@ class User(UserMixin, db.Model):
dismissed_release_version = db.Column(db.String(64), nullable=True)
oidc_sub = db.Column(db.String(255), nullable=True)
oidc_issuer = db.Column(db.String(255), nullable=True)
# Authentication source: local password, OIDC, or LDAP (synced on login where applicable)
auth_provider = db.Column(db.String(20), nullable=False, default="local", server_default="local")
avatar_filename = db.Column(db.String(255), nullable=True)
password_hash = db.Column(db.String(255), nullable=True)
password_change_required = db.Column(
+61 -10
View File
@@ -44,12 +44,52 @@ from app.utils.db import safe_commit
from app.utils.error_handling import safe_file_remove, safe_log
from app.utils.installation import get_installation_config
from app.utils.invoice_numbering import sanitize_invoice_pattern, sanitize_invoice_prefix, validate_invoice_pattern
from app.utils.auth_method import auth_includes_oidc, normalize_auth_method
from app.utils.permissions import admin_or_permission_required
from app.utils.telemetry import get_telemetry_fingerprint, is_telemetry_enabled
from app.utils.timezone import get_available_timezones
admin_bp = Blueprint("admin", __name__)
def _ldap_admin_display():
"""Read-only LDAP config summary for admin settings (from env / app config)."""
try:
cfg = current_app.config
ag = (cfg.get("LDAP_ADMIN_GROUP") or "").strip()
rg = (cfg.get("LDAP_REQUIRED_GROUP") or "").strip()
return {
"enabled": bool(cfg.get("LDAP_ENABLED")),
"host": cfg.get("LDAP_HOST") or "",
"port": int(cfg.get("LDAP_PORT") or 389),
"use_ssl": bool(cfg.get("LDAP_USE_SSL")),
"use_tls": bool(cfg.get("LDAP_USE_TLS")),
"base_dn": cfg.get("LDAP_BASE_DN") or "",
"user_dn": cfg.get("LDAP_USER_DN") or "",
"login_attr": cfg.get("LDAP_USER_LOGIN_ATTR") or "",
"admin_group": ag or "",
"required_group": rg or "",
}
except Exception:
return {
"enabled": False,
"host": "",
"port": 389,
"use_ssl": False,
"use_tls": False,
"base_dn": "",
"user_dn": "",
"login_attr": "",
"admin_group": "",
"required_group": "",
}
@admin_bp.context_processor
def _inject_ldap_admin_display():
return {"ldap_settings": _ldap_admin_display()}
# In-memory restore progress tracking (simple, per-process)
RESTORE_PROGRESS = {}
@@ -619,8 +659,8 @@ def admin_dashboard():
)
# Get OIDC status
auth_method = (getattr(Config, "AUTH_METHOD", "local") or "local").strip().lower()
oidc_enabled = auth_method in ("oidc", "both")
auth_method = normalize_auth_method(getattr(Config, "AUTH_METHOD", "local"))
oidc_enabled = auth_includes_oidc(auth_method)
oidc_issuer = getattr(Config, "OIDC_ISSUER", None)
oidc_configured = (
oidc_enabled
@@ -1560,6 +1600,17 @@ def settings():
)
@admin_bp.route("/admin/ldap/test", methods=["POST"])
@limiter.limit("10 per minute")
@login_required
@admin_or_permission_required("manage_settings")
def admin_ldap_test():
"""Test LDAP connectivity (service bind + user subtree count). Returns JSON only."""
from app.services.ldap_service import LDAPService
return jsonify(LDAPService.test_connection())
@admin_bp.route("/admin/settings/verify-donate-hide-code", methods=["POST"])
@login_required
@admin_or_permission_required("manage_settings")
@@ -4246,8 +4297,8 @@ def oidc_debug():
}
# Check if OIDC is enabled
auth_method = (oidc_config["auth_method"] or "local").strip().lower()
oidc_config["enabled"] = auth_method in ("oidc", "both")
auth_method = normalize_auth_method(oidc_config["auth_method"] or "local")
oidc_config["enabled"] = auth_includes_oidc(auth_method)
# Try to get OIDC client metadata
metadata = None
@@ -4306,9 +4357,9 @@ def oidc_test():
test_dns_resolution,
)
auth_method = (getattr(Config, "AUTH_METHOD", "local") or "local").strip().lower()
if auth_method not in ("oidc", "both"):
flash(_('OIDC is not enabled. Set AUTH_METHOD to "oidc" or "both".'), "warning")
auth_method = normalize_auth_method(getattr(Config, "AUTH_METHOD", "local"))
if not auth_includes_oidc(auth_method):
flash(_('OIDC is not enabled. Set AUTH_METHOD to "oidc", "both", or "all".'), "warning")
return redirect(url_for("admin.oidc_debug"))
issuer = getattr(Config, "OIDC_ISSUER", None)
@@ -4633,9 +4684,9 @@ def oidc_wizard_validate_config():
errors.append({"field": "client_secret", "message": "Client Secret is required"})
# Validate auth method
auth_method = data.get("auth_method", "").strip().lower()
if auth_method not in ("oidc", "both"):
errors.append({"field": "auth_method", "message": "Auth method must be 'oidc' or 'both'"})
auth_method = normalize_auth_method(data.get("auth_method", ""))
if not auth_includes_oidc(auth_method):
errors.append({"field": "auth_method", "message": "Auth method must be 'oidc', 'both', or 'all'"})
# Validate redirect URI if provided
redirect_uri = data.get("redirect_uri", "").strip()
+153 -91
View File
@@ -16,6 +16,14 @@ from app import db, limiter, log_event, oauth, track_event
from app.config import Config
from app.models import User
from app.utils.cache import get_cache
from app.utils.auth_method import (
auth_includes_ldap,
auth_includes_local,
auth_includes_oidc,
forgot_password_available,
normalize_auth_method,
requires_password_form,
)
from app.utils.config_manager import ConfigManager
from app.utils.db import safe_commit
from app.utils.posthog_funnels import track_onboarding_started
@@ -43,12 +51,15 @@ def get_avatar_upload_folder() -> str:
def _login_template_vars():
"""Common template variables for auth/login.html, including demo mode when enabled."""
allow_self_register = ConfigManager.get_setting("allow_self_register", Config.ALLOW_SELF_REGISTER)
auth_method = (current_app.config.get("AUTH_METHOD", "local") or "local").strip().lower()
requires_password = auth_method in ("local", "both")
auth_method = normalize_auth_method(current_app.config.get("AUTH_METHOD", "local"))
requires_password = requires_password_form(auth_method)
vars = {
"allow_self_register": allow_self_register,
"auth_method": auth_method,
"requires_password": requires_password,
"auth_includes_ldap": auth_includes_ldap(auth_method),
"auth_includes_oidc": auth_includes_oidc(auth_method),
"show_forgot_password": forgot_password_available(auth_method) and auth_method != "ldap",
}
if current_app.config.get("DEMO_MODE"):
vars["demo_mode"] = True
@@ -92,18 +103,78 @@ def _verify_password_reset_token(token: str, *, max_age_seconds: int) -> User |
return user
def _finalize_login_after_verification(user: User, *, log_auth_method: str):
"""After successful local or LDAP verification: 2FA gate or establish session and redirect."""
if getattr(user, "two_factor_enabled", False):
session["pre_2fa_user_id"] = user.id
next_page = request.args.get("next")
if next_page and next_page.startswith("/"):
session["pre_2fa_next"] = next_page
else:
session.pop("pre_2fa_next", None)
return redirect(url_for("auth.two_factor"))
from app.telemetry.otel_setup import business_span
with business_span("auth.login", user_id=user.id, auth_method=log_auth_method):
login_user(user, remember=True)
if not user.roles and user.role:
from app.utils.role_migration import migrate_single_user
if migrate_single_user(user.id):
current_app.logger.info(
"Auto-migrated user '%s' from legacy role '%s' to new role system",
user.username,
user.role,
)
user.update_last_login()
current_app.logger.info("User '%s' logged in successfully", user.username)
log_event("auth.login", user_id=user.id, auth_method=log_auth_method)
import threading
def track_login_async():
try:
track_event(user.id, "auth.login", {"auth_method": log_auth_method})
except Exception:
pass
threading.Thread(target=track_login_async, daemon=True).start()
if user.password_change_required:
flash(_("You must change your password before continuing."), "warning")
return redirect(url_for("auth.change_password"))
try:
require_admin_2fa = bool(getattr(Config, "REQUIRE_2FA_FOR_ADMINS", False))
except Exception:
require_admin_2fa = False
if require_admin_2fa and user.role == "admin" and not getattr(user, "two_factor_enabled", False):
flash(_("Administrator accounts must enable two-factor authentication."), "warning")
return redirect(url_for("auth.two_factor_setup"))
next_page = request.args.get("next")
if not next_page or not next_page.startswith("/"):
next_page = url_for("main.dashboard")
current_app.logger.info("Redirecting '%s' to %s", user.username, next_page)
flash(_("Welcome back, %(username)s!", username=user.username), "success")
return redirect(next_page)
@auth_bp.route("/forgot-password", methods=["GET", "POST"])
@limiter.limit("5 per minute", methods=["POST"])
def forgot_password():
if current_user.is_authenticated:
return redirect(url_for("main.dashboard"))
# Password reset only makes sense for password-based modes.
# Password reset only when local accounts may exist (not pure LDAP / OIDC-only).
try:
auth_method = (current_app.config.get("AUTH_METHOD", "local") or "local").strip().lower()
auth_method = normalize_auth_method(current_app.config.get("AUTH_METHOD", "local"))
except Exception:
auth_method = "local"
if auth_method not in ("local", "both"):
if not forgot_password_available(auth_method):
flash(_("Password reset is not available for this authentication method."), "warning")
return redirect(url_for("auth.login"))
@@ -132,7 +203,7 @@ def forgot_password():
if identifier:
user = User.query.filter((User.username == identifier) | (User.email == identifier)).first()
if user and user.is_active and user.email:
if user and user.is_active and user.email and getattr(user, "auth_provider", "local") == "local":
try:
token = _make_password_reset_token(user)
reset_url = url_for("auth.reset_password", token=token, _external=True)
@@ -175,6 +246,10 @@ def reset_password(token: str):
flash(_("This reset link is invalid or has expired. Please request a new one."), "error")
return redirect(url_for("auth.forgot_password"))
if getattr(user, "auth_provider", "local") == "ldap":
flash(_("Password reset is not available for this account."), "warning")
return redirect(url_for("auth.login"))
if request.method == "POST":
new_password = (request.form.get("new_password") or "").strip()
confirm_password = (request.form.get("confirm_password") or "").strip()
@@ -217,13 +292,13 @@ def login():
# Get authentication method from Flask app config (reads from environment)
try:
auth_method = (current_app.config.get("AUTH_METHOD", "local") or "local").strip().lower()
auth_method = normalize_auth_method(current_app.config.get("AUTH_METHOD", "local"))
except Exception:
auth_method = "local"
# Determine if password authentication is required
# 'none' = no password, 'local' = password required, 'oidc' = OIDC only, 'both' = OIDC + password
requires_password = auth_method in ("local", "both")
requires_password = requires_password_form(auth_method)
include_ldap = auth_includes_ldap(auth_method)
include_local = auth_includes_local(auth_method)
# If OIDC-only mode, redirect to OIDC login start
if auth_method == "oidc":
@@ -274,6 +349,20 @@ def login():
flash(_("Only the demo account can be used. Please use the credentials shown below."), "error")
return render_template("auth/login.html", **_login_template_vars())
if auth_method == "ldap":
from app.services.ldap_service import LDAPService
if requires_password and not password:
log_event("auth.login_failed", reason="password_required", auth_method=auth_method)
flash(_("Password is required"), "error")
return render_template("auth/login.html", **_login_template_vars())
ldap_user = LDAPService.authenticate(username, password)
if ldap_user:
return _finalize_login_after_verification(ldap_user, log_auth_method="ldap")
log_event("auth.login_failed", username=username, reason="ldap_auth_failed", auth_method=auth_method)
flash(_("Invalid username or password"), "error")
return render_template("auth/login.html", **_login_template_vars())
# Normalize admin usernames from config
try:
admin_usernames = [u.strip().lower() for u in (Config.ADMIN_USERNAMES or [])]
@@ -284,10 +373,18 @@ def login():
user = User.query.filter_by(username=username).first()
current_app.logger.info("User lookup for '%s': %s", username, "found" if user else "not found")
if auth_method == "all" and not user and include_ldap:
from app.services.ldap_service import LDAPService
if password:
ldap_user = LDAPService.authenticate(username, password)
if ldap_user:
return _finalize_login_after_verification(ldap_user, log_auth_method="ldap")
if not user:
# Check if self-registration is allowed (use ConfigManager to respect database settings)
allow_self_register = ConfigManager.get_setting("allow_self_register", Config.ALLOW_SELF_REGISTER)
if allow_self_register:
if allow_self_register and (include_local or auth_method == "none"):
# If password auth is required, validate password during self-registration
if requires_password:
if not password:
@@ -338,7 +435,10 @@ def login():
flash(_("Welcome! Your account has been created."), "success")
else:
log_event("auth.login_failed", username=username, reason="user_not_found", auth_method=auth_method)
flash(_("User not found. Please contact an administrator."), "error")
if auth_method == "all" and include_ldap:
flash(_("Invalid username or password"), "error")
else:
flash(_("User not found. Please contact an administrator."), "error")
return render_template("auth/login.html", **_login_template_vars())
else:
# If existing user matches admin usernames, ensure admin role
@@ -355,6 +455,22 @@ def login():
flash(_("Account is disabled. Please contact an administrator."), "error")
return render_template("auth/login.html", **_login_template_vars())
if auth_method == "all" and include_ldap and getattr(user, "auth_provider", "local") == "ldap":
from app.services.ldap_service import LDAPService
if requires_password and not password:
log_event(
"auth.login_failed", user_id=user.id, reason="password_required", auth_method=auth_method
)
flash(_("Password is required"), "error")
return render_template("auth/login.html", **_login_template_vars())
lu = LDAPService.authenticate(username, password)
if lu:
return _finalize_login_after_verification(lu, log_auth_method="ldap")
log_event("auth.login_failed", user_id=user.id, reason="ldap_auth_failed", auth_method=auth_method)
flash(_("Invalid username or password"), "error")
return render_template("auth/login.html", **_login_template_vars())
# Handle password authentication based on mode
if requires_password:
# Password authentication is required
@@ -371,6 +487,12 @@ def login():
log_event(
"auth.login_failed", user_id=user.id, reason="invalid_password", auth_method=auth_method
)
if auth_method == "all" and include_ldap:
from app.services.ldap_service import LDAPService
lu = LDAPService.authenticate(username, password)
if lu:
return _finalize_login_after_verification(lu, log_auth_method="ldap")
flash(_("Invalid username or password"), "error")
return render_template("auth/login.html", **_login_template_vars())
else:
@@ -409,76 +531,7 @@ def login():
# This mode is for trusted environments only
pass
# If 2FA is enabled for this user, require TOTP verification before creating a session.
if getattr(user, "two_factor_enabled", False):
session["pre_2fa_user_id"] = user.id
# Preserve intended redirect
next_page = request.args.get("next")
if next_page and next_page.startswith("/"):
session["pre_2fa_next"] = next_page
else:
session.pop("pre_2fa_next", None)
return redirect(url_for("auth.two_factor"))
from app.telemetry.otel_setup import business_span
with business_span("auth.login", user_id=user.id, auth_method=auth_method):
# Log in the user (password validation passed or password not required)
login_user(user, remember=True)
# Auto-migrate user from legacy role to new role system if needed
if not user.roles and user.role:
from app.utils.role_migration import migrate_single_user
if migrate_single_user(user.id):
current_app.logger.info(
"Auto-migrated user '%s' from legacy role '%s' to new role system",
user.username,
user.role,
)
user.update_last_login()
current_app.logger.info("User '%s' logged in successfully", user.username)
# Track successful login (log_event is fast, track_event is deferred to avoid blocking)
log_event("auth.login", user_id=user.id, auth_method=auth_method)
# Defer track_event to avoid blocking redirect - PostHog calls can be slow/timeout
import threading
def track_login_async():
try:
track_event(user.id, "auth.login", {"auth_method": auth_method})
except Exception:
pass # Don't let analytics errors affect login
threading.Thread(target=track_login_async, daemon=True).start()
# Note: identify_user_with_segments and set_super_properties are deferred to dashboard
# to avoid blocking the login redirect. The dashboard calls update_user_segments_if_needed
# which has caching logic and will handle this efficiently.
# Check if password change is required
if user.password_change_required:
flash(_("You must change your password before continuing."), "warning")
return redirect(url_for("auth.change_password"))
# Optionally enforce 2FA for admins (after login; they will be prompted to enroll).
try:
require_admin_2fa = bool(getattr(Config, "REQUIRE_2FA_FOR_ADMINS", False))
except Exception:
require_admin_2fa = False
if require_admin_2fa and user.role == "admin" and not getattr(user, "two_factor_enabled", False):
flash(_("Administrator accounts must enable two-factor authentication."), "warning")
return redirect(url_for("auth.two_factor_setup"))
# Redirect to intended page or dashboard
next_page = request.args.get("next")
if not next_page or not next_page.startswith("/"):
next_page = url_for("main.dashboard")
current_app.logger.info("Redirecting '%s' to %s", user.username, next_page)
flash(_("Welcome back, %(username)s!", username=user.username), "success")
return redirect(next_page)
return _finalize_login_after_verification(user, log_auth_method=auth_method)
except Exception as e:
current_app.logger.exception("Login error: %s", e)
flash(_("Unexpected error during login. Please try again or check server logs."), "error")
@@ -641,7 +694,7 @@ def logout():
# Try OIDC end-session if enabled and configured
try:
auth_method = (current_app.config.get("AUTH_METHOD", "local") or "local").strip().lower()
auth_method = normalize_auth_method(current_app.config.get("AUTH_METHOD", "local"))
except Exception:
auth_method = "local"
@@ -673,7 +726,7 @@ def logout():
pass
flash(_("Goodbye, %(username)s!", username=username), "info")
if auth_method in ("oidc", "both"):
if auth_includes_oidc(auth_method):
# Only perform RP-Initiated Logout if OIDC_POST_LOGOUT_REDIRECT_URI is explicitly configured
post_logout = getattr(Config, "OIDC_POST_LOGOUT_REDIRECT_URI", None)
if post_logout:
@@ -710,11 +763,12 @@ def edit_profile():
"""Edit user profile"""
# Get authentication method from Flask app config (reads from environment)
try:
auth_method = (current_app.config.get("AUTH_METHOD", "local") or "local").strip().lower()
auth_method = normalize_auth_method(current_app.config.get("AUTH_METHOD", "local"))
except Exception:
auth_method = "local"
requires_password = auth_method in ("local", "both")
requires_password = requires_password_form(auth_method)
allow_password_change = requires_password and getattr(current_user, "auth_provider", "local") != "ldap"
if request.method == "POST":
from app.utils.validation import sanitize_input
@@ -730,8 +784,8 @@ def edit_profile():
# Also set session so it applies immediately
session["preferred_language"] = preferred_language
# Handle password update if password auth is required
if requires_password:
# Handle password update if password auth is required (not LDAP-managed accounts)
if allow_password_change:
password = request.form.get("password", "").strip()
password_confirm = request.form.get("password_confirm", "").strip()
@@ -808,13 +862,17 @@ def edit_profile():
flash(_("Could not update your profile due to a database error."), "error")
return redirect(url_for("auth.profile"))
return render_template("auth/edit_profile.html", requires_password=requires_password)
return render_template("auth/edit_profile.html", requires_password=allow_password_change)
@auth_bp.route("/change-password", methods=["GET", "POST"])
@login_required
def change_password():
"""Change password page - required when password_change_required is True"""
if getattr(current_user, "auth_provider", "local") == "ldap":
flash(_("Password cannot be changed here for your account."), "warning")
return redirect(url_for("main.dashboard"))
if request.method == "POST":
current_password = request.form.get("current_password", "").strip()
new_password = request.form.get("new_password", "").strip()
@@ -925,11 +983,11 @@ def login_oidc():
return redirect(url_for("auth.login"))
try:
auth_method = (current_app.config.get("AUTH_METHOD", "local") or "local").strip().lower()
auth_method = normalize_auth_method(current_app.config.get("AUTH_METHOD", "local"))
except Exception:
auth_method = "local"
if auth_method not in ("oidc", "both"):
if not auth_includes_oidc(auth_method):
return redirect(url_for("auth.login"))
client = oauth.create_client("oidc")
@@ -1294,6 +1352,7 @@ def oidc_callback():
user.is_active = True
user.oidc_issuer = issuer
user.oidc_sub = sub
user.auth_provider = "oidc"
# Apply company default for daily working hours (overtime)
try:
from app.models import Settings
@@ -1334,6 +1393,9 @@ def oidc_callback():
else:
# Update linkage and profile fields
changed = False
if getattr(user, "auth_provider", "local") != "oidc":
user.auth_provider = "oidc"
changed = True
if not user.oidc_issuer or not user.oidc_sub:
user.oidc_issuer = issuer
user.oidc_sub = sub
+6 -5
View File
@@ -98,15 +98,16 @@ def kiosk_login():
return redirect(url_for("kiosk.kiosk_dashboard"))
# Get authentication method
try:
from app.config import Config
from app.config import Config
from app.utils.auth_method import normalize_auth_method, requires_password_form
auth_method = (getattr(Config, "AUTH_METHOD", "local") or "local").strip().lower()
try:
auth_method = normalize_auth_method(getattr(Config, "AUTH_METHOD", "local"))
except Exception:
auth_method = "local"
# Determine if password authentication is required (kiosk doesn't support OIDC)
requires_password = auth_method in ("local", "both")
# Determine if password authentication is required (kiosk doesn't support OIDC/LDAP flows)
requires_password = requires_password_form(auth_method)
if request.method == "POST":
username = request.form.get("username", "").strip()
+390
View File
@@ -0,0 +1,390 @@
"""
LDAP authentication: service-account search, optional group checks, user bind, DB sync.
"""
from __future__ import annotations
import logging
import secrets
from typing import Any, Mapping, MutableMapping, Optional
from flask import current_app
from app import db
from app.models import User
from app.utils.db import safe_commit
logger = logging.getLogger(__name__)
try:
from ldap3 import Connection, SIMPLE, SUBTREE, Tls
from ldap3 import Server
from ldap3.core.exceptions import LDAPException
from ldap3.utils.conv import escape_filter_chars
except ImportError: # pragma: no cover - exercised when ldap3 missing
Connection = None # type: ignore[misc, assignment]
Server = None # type: ignore[misc, assignment]
LDAPException = Exception # type: ignore[misc, assignment]
def _config() -> MutableMapping[str, Any]:
return current_app.config
def _group_search_base(cfg: Mapping[str, Any]) -> str:
gdn = (cfg.get("LDAP_GROUP_DN") or "").strip().strip(",")
bdn = (cfg.get("LDAP_BASE_DN") or "").strip().strip(",")
if not gdn:
return bdn
if not bdn:
return gdn
return f"{gdn},{bdn}"
def _user_search_base(cfg: Mapping[str, Any]) -> str:
udn = (cfg.get("LDAP_USER_DN") or "").strip().strip(",")
bdn = (cfg.get("LDAP_BASE_DN") or "").strip().strip(",")
if not udn:
return bdn
if not bdn:
return udn
return f"{udn},{bdn}"
def _make_server(cfg: Mapping[str, Any]) -> Any:
host = (cfg.get("LDAP_HOST") or "localhost").strip()
port = int(cfg.get("LDAP_PORT") or 389)
use_ssl = bool(cfg.get("LDAP_USE_SSL"))
timeout = int(cfg.get("LDAP_TIMEOUT") or 10)
ca_file = (cfg.get("LDAP_TLS_CA_CERT_FILE") or "").strip()
tls = None
if ca_file:
tls = Tls(ca_certs_file=ca_file)
return Server(
host,
port=port,
use_ssl=use_ssl,
get_info=None,
connect_timeout=timeout,
tls=tls,
)
def _service_connection(cfg: Mapping[str, Any]) -> Optional[Any]:
if Connection is None or Server is None:
return None
server = _make_server(cfg)
bind_dn = (cfg.get("LDAP_BIND_DN") or "").strip()
bind_pw = cfg.get("LDAP_BIND_PASSWORD") or ""
timeout = int(cfg.get("LDAP_TIMEOUT") or 10)
conn = Connection(
server,
user=bind_dn,
password=bind_pw,
authentication=SIMPLE,
receive_timeout=timeout,
auto_bind=False,
)
conn.open()
if cfg.get("LDAP_USE_TLS") and not cfg.get("LDAP_USE_SSL"):
conn.start_tls(read_server_info=False)
conn.bind()
return conn
def _user_dn_member_of_group(
conn: Any,
cfg: Mapping[str, Any],
group_cn: str,
user_dn: str,
) -> bool:
group_cn = (group_cn or "").strip()
if not group_cn or not user_dn:
return False
base = _group_search_base(cfg)
oc = escape_filter_chars((cfg.get("LDAP_GROUP_OBJECT_CLASS") or "groupOfNames").strip())
cn_esc = escape_filter_chars(group_cn)
ud_esc = escape_filter_chars(user_dn)
filt = f"(&(objectClass={oc})(cn={cn_esc})(member={ud_esc}))"
conn.search(search_base=base, search_filter=filt, search_scope=SUBTREE, size_limit=1, attributes=["1.1"])
return bool(conn.entries)
class LDAPService:
"""LDAP bind-authenticate and sync users to the local User model."""
@staticmethod
def authenticate(username: str, password: str) -> Optional[User]:
"""
Validate credentials against LDAP and return the linked User, or None.
Never raises LDAP errors to callers; failures are logged at WARNING without passwords.
"""
if Connection is None:
logger.warning("LDAP authenticate skipped: ldap3 is not installed")
return None
username = (username or "").strip()
password = password or ""
if not username or not password:
return None
cfg = _config()
svc_conn = None
try:
try:
svc_conn = _service_connection(cfg)
except LDAPException:
logger.warning("LDAP service bind failed")
return None
except Exception:
logger.warning("LDAP service connection error")
return None
if not svc_conn:
return None
user_base = _user_search_base(cfg)
login_attr = (cfg.get("LDAP_USER_LOGIN_ATTR") or "uid").strip()
user_oc = (cfg.get("LDAP_USER_OBJECT_CLASS") or "inetOrgPerson").strip()
u_esc = escape_filter_chars(username.lower())
la_esc = escape_filter_chars(login_attr)
oc_esc = escape_filter_chars(user_oc)
filt = f"(&(objectClass={oc_esc})({la_esc}={u_esc}))"
fetch_attrs = {
(cfg.get("LDAP_USER_LOGIN_ATTR") or "uid").strip(),
(cfg.get("LDAP_USER_EMAIL_ATTR") or "mail").strip(),
(cfg.get("LDAP_USER_FNAME_ATTR") or "givenName").strip(),
(cfg.get("LDAP_USER_LNAME_ATTR") or "sn").strip(),
}
svc_conn.search(
search_base=user_base,
search_filter=filt,
search_scope=SUBTREE,
size_limit=2,
attributes=list(fetch_attrs),
)
if not svc_conn.entries:
logger.warning("LDAP user not found for login attribute match")
return None
if len(svc_conn.entries) > 1:
logger.warning("LDAP search returned multiple entries; refusing login")
return None
entry = svc_conn.entries[0]
user_dn = entry.entry_dn
attrs = entry.entry_attributes_as_dict
req_group = (cfg.get("LDAP_REQUIRED_GROUP") or "").strip()
if req_group and not _user_dn_member_of_group(svc_conn, cfg, req_group, user_dn):
logger.warning("LDAP user not in required group")
return None
try:
if svc_conn.bound:
svc_conn.unbind()
except Exception:
pass
svc_conn = None
try:
user_conn = Connection(
_make_server(cfg),
user=user_dn,
password=password,
authentication=SIMPLE,
receive_timeout=int(cfg.get("LDAP_TIMEOUT") or 10),
auto_bind=True,
)
user_conn.unbind()
except LDAPException:
logger.warning("LDAP user bind failed")
return None
except Exception:
logger.warning("LDAP user bind error")
return None
ldap_attrs = LDAPService._entry_to_attrs(cfg, attrs, username.lower())
if not ldap_attrs.get("email"):
logger.warning("LDAP user has no email; cannot provision local user")
return None
admin_group = (cfg.get("LDAP_ADMIN_GROUP") or "").strip()
is_admin = False
if admin_group:
try:
c2 = _service_connection(cfg)
if c2:
is_admin = _user_dn_member_of_group(c2, cfg, admin_group, user_dn)
c2.unbind()
except LDAPException:
pass
synced = LDAPService._get_or_create_user(cfg, ldap_attrs, is_admin_member=is_admin)
if not synced:
logger.warning("LDAP user could not be persisted to the database")
return None
return synced
except LDAPException:
logger.warning("LDAP authenticate failed")
return None
except Exception:
logger.warning("LDAP authenticate unexpected error")
return None
finally:
if svc_conn is not None:
try:
if svc_conn.bound:
svc_conn.unbind()
except Exception:
pass
@staticmethod
def _entry_to_attrs(
cfg: Mapping[str, Any],
raw_attrs: Mapping[str, Any],
username_lower: str,
) -> dict[str, Optional[str]]:
def first(attr: str) -> Optional[str]:
if not attr:
return None
vals = raw_attrs.get(attr) or []
if not vals:
return None
v = vals[0]
if hasattr(v, "value"):
v = v.value
s = str(v).strip()
return s or None
login_attr = (cfg.get("LDAP_USER_LOGIN_ATTR") or "uid").strip()
email_attr = (cfg.get("LDAP_USER_EMAIL_ATTR") or "mail").strip()
fn_attr = (cfg.get("LDAP_USER_FNAME_ATTR") or "givenName").strip()
ln_attr = (cfg.get("LDAP_USER_LNAME_ATTR") or "sn").strip()
email = first(email_attr)
if email:
email = email.lower()
fn = first(fn_attr) or ""
ln = first(ln_attr) or ""
parts = [p for p in (fn.strip(), ln.strip()) if p]
full_name = " ".join(parts).strip() or None
un = first(login_attr) or username_lower
if un:
un = un.lower().strip()
return {
"username": un or username_lower,
"email": email,
"full_name": full_name,
}
@staticmethod
def _get_or_create_user(
cfg: Mapping[str, Any],
ldap_attrs: Mapping[str, Any],
*,
is_admin_member: bool,
) -> Optional[User]:
"""Create or update a User from LDAP attributes; commit and return user, or None on DB failure."""
email = ldap_attrs.get("email")
username = ldap_attrs.get("username") or ""
full_name = ldap_attrs.get("full_name")
user = User.query.filter_by(email=email).first() if email else None
if not user:
role_name = "admin" if is_admin_member else "user"
user = User(username=username, role=role_name, email=email, full_name=full_name)
user.auth_provider = "ldap"
user.set_password(secrets.token_urlsafe(48))
user.is_active = True
try:
from app.models import Role
role_obj = Role.query.filter_by(name=role_name).first()
if role_obj:
user.roles.append(role_obj)
except Exception:
pass
try:
from app.models import Settings
settings = Settings.get_settings()
user.standard_hours_per_day = float(getattr(settings, "default_daily_working_hours", 8.0) or 8.0)
except Exception:
pass
db.session.add(user)
else:
user.auth_provider = "ldap"
if username and user.username != username:
user.username = username
if full_name is not None:
user.full_name = full_name
if email and user.email != email:
user.email = email
if is_admin_member:
if user.role != "admin":
user.role = "admin"
else:
if user.role == "admin" and getattr(user, "auth_provider", None) == "ldap":
user.role = "user"
if not safe_commit("ldap_sync_user", {"user_id": getattr(user, "id", None), "email": email}):
db.session.rollback()
logger.warning("LDAP user DB commit failed")
return None
return User.query.filter_by(email=email).first() or user
@staticmethod
def test_connection() -> dict[str, Any]:
"""
Verify service bind and count users under the user subtree.
Returns dict: success (bool), message (str), user_count (int|None).
Never raises.
"""
if Connection is None:
return {"success": False, "message": "ldap3 is not installed", "user_count": None}
conn = None
cfg = _config()
try:
conn = _service_connection(cfg)
if not conn:
return {"success": False, "message": "Could not create LDAP connection", "user_count": None}
except LDAPException as e:
return {"success": False, "message": f"LDAP error: {type(e).__name__}", "user_count": None}
except Exception as e:
return {"success": False, "message": f"Error: {type(e).__name__}", "user_count": None}
try:
user_base = _user_search_base(cfg)
user_oc = escape_filter_chars((cfg.get("LDAP_USER_OBJECT_CLASS") or "inetOrgPerson").strip())
filt = f"(objectClass={user_oc})"
conn.search(
search_base=user_base,
search_filter=filt,
search_scope=SUBTREE,
attributes=["1.1"],
size_limit=2001,
)
n = len(conn.entries)
if n > 2000:
return {
"success": True,
"message": "Connected; user count exceeds 2000 (showing as 2000+)",
"user_count": n,
}
return {"success": True, "message": "Connected successfully", "user_count": n}
except LDAPException as e:
return {"success": False, "message": f"LDAP search failed: {type(e).__name__}", "user_count": None}
except Exception as e:
return {"success": False, "message": f"Search error: {type(e).__name__}", "user_count": None}
finally:
try:
conn.unbind()
except Exception:
pass
+83
View File
@@ -36,6 +36,7 @@
<a href="#section-invoices" class="px-3 py-1.5 text-xs font-medium rounded-full bg-card-light dark:bg-card-dark border border-border-light dark:border-border-dark whitespace-nowrap">Invoices</a>
<a href="#section-peppol" class="px-3 py-1.5 text-xs font-medium rounded-full bg-card-light dark:bg-card-dark border border-border-light dark:border-border-dark whitespace-nowrap">Peppol</a>
<a href="#section-backup" class="px-3 py-1.5 text-xs font-medium rounded-full bg-card-light dark:bg-card-dark border border-border-light dark:border-border-dark whitespace-nowrap">Backup</a>
<a href="#section-ldap" class="px-3 py-1.5 text-xs font-medium rounded-full bg-card-light dark:bg-card-dark border border-border-light dark:border-border-dark whitespace-nowrap">LDAP</a>
<a href="#section-kiosk" class="px-3 py-1.5 text-xs font-medium rounded-full bg-card-light dark:bg-card-dark border border-border-light dark:border-border-dark whitespace-nowrap">Kiosk</a>
<a href="#section-ai" class="px-3 py-1.5 text-xs font-medium rounded-full bg-card-light dark:bg-card-dark border border-border-light dark:border-border-dark whitespace-nowrap">AI Helper</a>
<a href="#section-analytics" class="px-3 py-1.5 text-xs font-medium rounded-full bg-card-light dark:bg-card-dark border border-border-light dark:border-border-dark whitespace-nowrap">Analytics</a>
@@ -400,6 +401,55 @@
</div>
</div>
<!-- LDAP (read-only; configured via environment) -->
<div id="section-ldap" class="border-b border-border-light dark:border-border-dark pb-6">
<h2 class="text-lg font-semibold mb-4">{{ _('LDAP') }}</h2>
<p class="text-sm text-text-muted-light dark:text-text-muted-dark mb-4">
{{ _('LDAP authentication is configured with environment variables. Values below are read-only.') }}
</p>
<div class="grid grid-cols-1 md:grid-cols-2 gap-6 text-sm">
<div>
<span class="form-label text-text-muted-light dark:text-text-muted-dark">{{ _('Enabled for auth') }}</span>
<p class="mt-1 font-medium">{{ _('Yes') if ldap_settings.enabled else _('No') }}</p>
</div>
<div>
<span class="form-label text-text-muted-light dark:text-text-muted-dark">{{ _('Host') }} / {{ _('Port') }}</span>
<p class="mt-1 font-mono text-xs break-all">{{ ldap_settings.host }}:{{ ldap_settings.port }}</p>
</div>
<div>
<span class="form-label text-text-muted-light dark:text-text-muted-dark">{{ _('SSL') }} / {{ _('TLS') }}</span>
<p class="mt-1">{{ _('SSL') }}: {% if ldap_settings.use_ssl %}{{ _('on') }}{% else %}{{ _('off') }}{% endif %},
{{ _('TLS') }}: {% if ldap_settings.use_tls %}{{ _('on') }}{% else %}{{ _('off') }}{% endif %}</p>
</div>
<div>
<span class="form-label text-text-muted-light dark:text-text-muted-dark">{{ _('Base DN') }}</span>
<p class="mt-1 font-mono text-xs break-all">{{ ldap_settings.base_dn }}</p>
</div>
<div>
<span class="form-label text-text-muted-light dark:text-text-muted-dark">{{ _('User DN') }}</span>
<p class="mt-1 font-mono text-xs break-all">{{ ldap_settings.user_dn }}</p>
</div>
<div>
<span class="form-label text-text-muted-light dark:text-text-muted-dark">{{ _('User login attribute') }}</span>
<p class="mt-1 font-mono text-xs">{{ ldap_settings.login_attr }}</p>
</div>
<div>
<span class="form-label text-text-muted-light dark:text-text-muted-dark">{{ _('Admin group (CN)') }}</span>
<p class="mt-1 font-mono text-xs break-all">{{ ldap_settings.admin_group }}</p>
</div>
<div>
<span class="form-label text-text-muted-light dark:text-text-muted-dark">{{ _('Required group (CN)') }}</span>
<p class="mt-1 font-mono text-xs break-all">{{ ldap_settings.required_group }}</p>
</div>
</div>
<div class="mt-4 flex flex-wrap items-center gap-3">
<button type="button" id="ldapTestConnectionBtn" class="btn btn-secondary">
{{ _('Test LDAP Connection') }}
</button>
<span id="ldapTestConnectionResult" class="text-sm text-text-muted-light dark:text-text-muted-dark"></span>
</div>
</div>
<!-- Export Settings -->
<div class="border-b border-border-light dark:border-border-dark pb-6">
<h2 class="text-lg font-semibold mb-4">{{ _('Export Settings') }}</h2>
@@ -757,6 +807,39 @@ document.addEventListener('DOMContentLoaded', function() {
togglePeppolTransport();
}
const ldapTestBtn = document.getElementById('ldapTestConnectionBtn');
const ldapTestResult = document.getElementById('ldapTestConnectionResult');
if (ldapTestBtn && ldapTestResult) {
ldapTestBtn.addEventListener('click', async function() {
ldapTestBtn.disabled = true;
ldapTestResult.textContent = '{{ _("Testing connection...") }}';
ldapTestResult.className = 'text-sm text-text-muted-light dark:text-text-muted-dark';
const csrfToken = document.querySelector('meta[name="csrf-token"]')?.content || '';
try {
const response = await fetch('{{ url_for("admin.admin_ldap_test") }}', {
method: 'POST',
credentials: 'same-origin',
headers: { 'Content-Type': 'application/json', 'X-CSRFToken': csrfToken },
body: JSON.stringify({})
});
const data = await response.json().catch(function() { return {}; });
if (data.success) {
const cnt = data.user_count != null ? ' (' + data.user_count + ' {{ _("users") }})' : '';
ldapTestResult.textContent = (data.message || '{{ _("OK") }}') + cnt;
ldapTestResult.className = 'text-sm text-green-600 dark:text-green-400';
} else {
ldapTestResult.textContent = data.message || '{{ _("Connection failed.") }}';
ldapTestResult.className = 'text-sm text-red-600 dark:text-red-400';
}
} catch (err) {
ldapTestResult.textContent = '{{ _("Request failed.") }}';
ldapTestResult.className = 'text-sm text-red-600 dark:text-red-400';
} finally {
ldapTestBtn.disabled = false;
}
});
}
const aiTestBtn = document.getElementById('aiTestConnectionBtn');
const aiTestResult = document.getElementById('aiTestConnectionResult');
if (aiTestBtn && aiTestResult) {
+6 -2
View File
@@ -66,7 +66,7 @@
<i class="fa-solid fa-lock absolute left-3 top-1/2 -translate-y-1/2 text-text-muted-light dark:text-text-muted-dark"></i>
<input type="password" name="password" id="password" autocomplete="current-password" class="w-full pl-10 bg-background-light dark:bg-gray-700 border border-border-light dark:border-border-dark rounded-lg px-3 py-2.5 text-base text-text-light dark:text-text-dark placeholder-text-muted-light dark:placeholder-text-muted-dark focus:outline-none focus:ring-2 focus:ring-primary" placeholder="{{ _('Password') }}">
</div>
{% if not demo_mode %}
{% if not demo_mode and show_forgot_password %}
<div class="mt-2 text-right">
<a href="{{ url_for('auth.forgot_password') }}" class="text-xs text-primary hover:underline">
{{ _('Forgot your password?') }}
@@ -75,6 +75,10 @@
{% endif %}
{% endif %}
{% if auth_includes_ldap %}
<p class="mt-2 text-xs text-text-muted-light dark:text-text-muted-dark">{{ _('LDAP authentication is enabled') }}</p>
{% endif %}
<button type="submit" class="btn btn-primary w-full mt-6">{{ _('Sign in') }}</button>
{% if demo_mode %}
@@ -83,7 +87,7 @@
<p class="mt-3 text-xs text-text-muted-light dark:text-text-muted-dark text-center">{{ _('Tip: Enter a new username to create your account.') }}</p>
{% endif %}
{% if not demo_mode and auth_method and auth_method|string|lower == 'both' %}
{% if not demo_mode and auth_includes_oidc %}
<div class="relative my-6">
<div class="absolute inset-0 flex items-center" aria-hidden="true">
<div class="w-full border-t border-border-light dark:border-border-dark"></div>
+1 -1
View File
@@ -41,7 +41,7 @@
<div class="flex items-center gap-2 mb-4">
{% set auth_method = config.get('AUTH_METHOD', 'local') or 'local' %}
{% set oidc_enabled = auth_method.lower() in ['oidc', 'both'] %}
{% set oidc_enabled = auth_method.lower() in ['oidc', 'both', 'all'] %}
{% if oidc_enabled %}
<span class="px-2 py-1 text-xs rounded-full bg-green-100 text-green-800 dark:bg-green-900 dark:text-green-200 font-medium">
<i class="fas fa-check-circle mr-1"></i>{{ _('Enabled') }}
+42
View File
@@ -0,0 +1,42 @@
"""Helpers for AUTH_METHOD parsing (none | local | oidc | ldap | both | all)."""
from __future__ import annotations
_VALID = frozenset({"none", "local", "oidc", "ldap", "both", "all"})
def normalize_auth_method(raw: str | None) -> str:
"""Return a valid auth method string; unknown values become 'local' in non-production via caller."""
s = (raw or "local").strip().lower()
return s if s in _VALID else "local"
def auth_includes_local(auth_method: str | None) -> bool:
m = normalize_auth_method(auth_method)
return m in ("local", "both", "all")
def auth_includes_oidc(auth_method: str | None) -> bool:
m = normalize_auth_method(auth_method)
return m in ("oidc", "both", "all")
def auth_includes_ldap(auth_method: str | None) -> bool:
m = normalize_auth_method(auth_method)
return m in ("ldap", "all")
def requires_password_form(auth_method: str | None) -> bool:
"""True when login form should collect a password (local, ldap, or combined modes)."""
m = normalize_auth_method(auth_method)
return m in ("local", "both", "ldap", "all")
def forgot_password_available(auth_method: str | None) -> bool:
"""Forgot-password link when any local-password account may exist."""
return auth_includes_local(auth_method)
def ldap_enabled_from_auth_method(auth_method: str | None) -> bool:
"""LDAP auth is active for this AUTH_METHOD (same as Config LDAP_ENABLED)."""
return auth_includes_ldap(auth_method)
+14 -3
View File
@@ -107,6 +107,14 @@ def validate_production_config() -> Tuple[bool, List[str]]:
if not session_secure:
issues.append("SESSION_COOKIE_SECURE should be true in production")
# LDAP required vars when LDAP authentication is enabled
auth_method = (os.getenv("AUTH_METHOD", "local") or "local").strip().lower()
if auth_method in ("ldap", "all"):
for var in ("LDAP_HOST", "LDAP_BASE_DN", "LDAP_BIND_DN", "LDAP_BIND_PASSWORD"):
val = (os.getenv(var) or "").strip()
if not val:
issues.append(f"{var} must be set when AUTH_METHOD enables LDAP ({auth_method})")
return len(issues) == 0, issues
@@ -117,12 +125,15 @@ def validate_optional_env_vars() -> Dict[str, bool]:
Returns:
Dict mapping env var names to their validation status
"""
auth_m = (os.getenv("AUTH_METHOD", "") or "").strip().lower()
oidc_required = auth_m in ("oidc", "both", "all")
optional_vars = {
"TZ": lambda v: bool(v),
"CURRENCY": lambda v: bool(v),
"OIDC_ISSUER": lambda v: bool(v) if os.getenv("AUTH_METHOD", "").lower() in ("oidc", "both") else True,
"OIDC_CLIENT_ID": lambda v: bool(v) if os.getenv("AUTH_METHOD", "").lower() in ("oidc", "both") else True,
"OIDC_CLIENT_SECRET": lambda v: bool(v) if os.getenv("AUTH_METHOD", "").lower() in ("oidc", "both") else True,
"OIDC_ISSUER": lambda v: bool(v) if oidc_required else True,
"OIDC_CLIENT_ID": lambda v: bool(v) if oidc_required else True,
"OIDC_CLIENT_SECRET": lambda v: bool(v) if oidc_required else True,
}
results = {}
+6 -2
View File
@@ -101,6 +101,8 @@ python app.py
- **No authentication (`AUTH_METHOD=none`)**: Enter username only (no password)
- **OIDC (`AUTH_METHOD=oidc`)**: Click "Sign in with SSO" button
- **Both (`AUTH_METHOD=both`)**: Choose either SSO or local username/password
- **LDAP (`AUTH_METHOD=ldap`)**: Sign in with directory username and password (no SSO button)
- **All methods (`AUTH_METHOD=all`)**: SSO button plus username/password (local and/or LDAP, depending on account)
3. **Admin users are configured in the environment**
- Set via `ADMIN_USERNAMES` environment variable (default: `admin`)
@@ -114,9 +116,11 @@ python app.py
> - `none`: Username only (for trusted internal networks)
> - `local`: Username + password (default, recommended)
> - `oidc`: Single Sign-On only
> - `both`: Both OIDC and local password authentication
> - `ldap`: LDAP directory only
> - `both`: OIDC and local password (no LDAP)
> - `all`: Local + OIDC + LDAP
>
> See [OIDC Setup Guide](OIDC_SETUP.md#5-authentication-methods) for detailed explanations of all authentication modes.
> See [OIDC Setup Guide](admin/configuration/OIDC_SETUP.md#5-authentication-methods) and [LDAP Setup](admin/configuration/LDAP_SETUP.md) for details.
---
+1
View File
@@ -90,6 +90,7 @@ See [features/](features/) for additional feature documentation.
- **[Docker Startup Troubleshooting](admin/configuration/DOCKER_STARTUP_TROUBLESHOOTING.md)** — Fix startup issues
- **[Email Configuration](admin/configuration/EMAIL_CONFIGURATION.md)** — Email setup
- **[OIDC Setup](admin/configuration/OIDC_SETUP.md)** — OIDC/SSO authentication setup
- **[LDAP Setup](admin/configuration/LDAP_SETUP.md)** — LDAP directory authentication (`AUTH_METHOD=ldap` or `all`)
- **[Support visibility](admin/configuration/SUPPORT_VISIBILITY.md)** — Hide donate/support UI with a purchased key; [purchase key](https://timetracker.drytrix.com/support.html)
### Deployment
@@ -94,9 +94,11 @@ All environment variables can be provided via `.env` and are consumed by the `ap
- `none`: No password authentication (username only). Use only in trusted environments.
- `local`: Password authentication required (default). Users must set and use passwords.
- `oidc`: OIDC/Single Sign-On only. Local login form is hidden.
- `both`: OIDC + local password authentication. Users can choose either method.
- `ldap`: LDAP directory authentication only (username/password against LDAP).
- `both`: OIDC + local password (no LDAP). Users can choose SSO or local login.
- `all`: Local + OIDC + LDAP combined (see [OIDC Setup](OIDC_SETUP.md) and [LDAP Setup](LDAP_SETUP.md)).
Default: `local`. See [OIDC Setup Guide](OIDC_SETUP.md) for detailed explanations.
Default: `local`. See [OIDC Setup Guide](OIDC_SETUP.md) and [LDAP Setup](LDAP_SETUP.md) for details.
- OIDC_ISSUER: OIDC provider issuer URL.
- OIDC_CLIENT_ID: OIDC client id.
- OIDC_CLIENT_SECRET: OIDC client secret.
@@ -185,6 +185,7 @@ If the issue persists, check:
- Use the first username from `ADMIN_USERNAMES` (default: "admin")
- If using `AUTH_METHOD=local`, the default admin has no password initially. On first login, enter the username and choose any password (minimum 8 characters)—it will be set and you will be logged in. There is no default password; you define it yourself on first use.
- If using `AUTH_METHOD=none`, you can login immediately (no password required)
- If using `AUTH_METHOD=ldap` or `all`, configure all required `LDAP_*` variables (see `env.example` and [LDAP Setup](LDAP_SETUP.md)); the first admin may still be created locally depending on your process
2. **Create additional admin users**:
+29
View File
@@ -0,0 +1,29 @@
# LDAP authentication
TimeTracker can authenticate users against an LDAP directory (OpenLDAP-style `groupOfNames` / `member` checks). LDAP is optional and is controlled with **`AUTH_METHOD`** and environment variables (see root **`env.example`** for a commented template).
## When to use which `AUTH_METHOD`
| Value | Meaning |
|---------|---------|
| `ldap` | Directory login only (same username/password form; users are provisioned or synced in the local DB on success). |
| `all` | Local passwords, OIDC SSO, and LDAP are all available (see [OIDC Setup](OIDC_SETUP.md) for SSO). Login tries local first for users whose `auth_provider` is not `ldap`, then LDAP. |
For LDAP only or combined mode, set the variables documented in `env.example` under **LDAP Authentication**. In production, if LDAP is enabled, **`LDAP_HOST`**, **`LDAP_BASE_DN`**, **`LDAP_BIND_DN`**, and **`LDAP_BIND_PASSWORD`** are required (startup validation).
## Behaviour summary
- **Service account**: Binds with `LDAP_BIND_DN` / `LDAP_BIND_PASSWORD`, searches for the user under `LDAP_USER_DN` + `LDAP_BASE_DN`, optionally verifies membership in `LDAP_REQUIRED_GROUP` (by `cn` under `LDAP_GROUP_DN`), then verifies the password with a second bind as the user.
- **Provisioning**: Users are matched primarily by **email** from `LDAP_USER_EMAIL_ATTR`. Without an email, login cannot create or link an account.
- **Profile sync**: On each successful LDAP login, `full_name` (from `givenName` + `sn`) and admin flag (via `LDAP_ADMIN_GROUP` and legacy `role` field) are updated from the directory.
- **Local passwords**: LDAP-managed accounts have `auth_provider=ldap` and cannot use forgot-password, reset-password, or in-app password change flows.
- **Admin UI**: **Admin → System Settings** includes a read-only LDAP summary and **Test LDAP Connection** (`POST /admin/ldap/test`) for a non-destructive bind and user count under the configured user subtree.
## Kiosk mode
Kiosk login continues to use **local passwords only** (same `requires_password` rules as `local` / `both` / `all` for the form). LDAP-only users must have a usable local password for kiosk, or use standard web login.
## Further reading
- [OIDC Setup](OIDC_SETUP.md) — `AUTH_METHOD` overview including `all`.
- [Docker Compose environment](DOCKER_COMPOSE_SETUP.md#authentication) — variable list entry point.
+35 -24
View File
@@ -1,10 +1,10 @@
## OpenID Connect (OIDC) Setup Guide
This guide explains how to enable Single Sign-On (SSO) with OpenID Connect for TimeTracker. OIDC is optional; you can run with local login only, OIDC only, or both.
This guide explains how to enable Single Sign-On (SSO) with OpenID Connect for TimeTracker. OIDC is optional; you can run with local login only, OIDC only, both, or combined with LDAP using `AUTH_METHOD=all` (see [LDAP Setup](LDAP_SETUP.md)).
### Quick Summary
- Set `AUTH_METHOD=oidc` (SSO only) or `AUTH_METHOD=both` (SSO + local password authentication).
- Set `AUTH_METHOD=oidc` (SSO only), `AUTH_METHOD=both` (SSO + local password), or `AUTH_METHOD=all` (local + SSO + LDAP; LDAP is documented separately).
- Configure `OIDC_ISSUER`, `OIDC_CLIENT_ID`, `OIDC_CLIENT_SECRET`, and `OIDC_REDIRECT_URI`.
- Optional: Configure admin mapping via `OIDC_ADMIN_GROUP` or `OIDC_ADMIN_EMAILS`.
- Restart the app. The login page will show an “Sign in with SSO” button when enabled.
@@ -31,7 +31,7 @@ Make sure your external URL and protocol (HTTP/HTTPS) match how users access the
Add these to your environment (e.g., `.env`, Docker Compose, or Kubernetes Secrets):
```
AUTH_METHOD=oidc # Options: none | local | oidc | both (see section 5 for details)
AUTH_METHOD=oidc # Options: none | local | oidc | ldap | both | all (see section 5; LDAP: LDAP_SETUP.md)
# Core OIDC settings
OIDC_ISSUER=https://idp.example.com/realms/your-realm
@@ -109,7 +109,7 @@ Also ensure the standard app settings are configured (database, secret key, etc.
### 5) Authentication Methods
The `AUTH_METHOD` environment variable controls how users authenticate with TimeTracker. It supports four options:
The `AUTH_METHOD` environment variable controls how users authenticate with TimeTracker. It supports these options:
#### Available Options
@@ -136,21 +136,33 @@ The `AUTH_METHOD` environment variable controls how users authenticate with Time
- Requires OIDC configuration (see Required Environment Variables above)
- Self-registration still works if `ALLOW_SELF_REGISTER=true` (users created on first OIDC login)
4. **`both`** - OIDC + Local password authentication
4. **`both`** - OIDC + Local password authentication (no LDAP)
- Shows both SSO button and local login form
- Users can choose to log in with OIDC or use username/password
- Local authentication requires passwords (same as `local` mode)
- Best for organizations transitioning to SSO or supporting mixed authentication
- Requires OIDC configuration to be set up
5. **`ldap`** - LDAP directory authentication only
- Same username/password form; no OIDC button
- Users are created or updated from the directory after a successful bind
- Configure all `LDAP_*` variables; see [LDAP Setup](LDAP_SETUP.md) and `env.example`
6. **`all`** - Local + OIDC + LDAP
- SSO button plus username/password form; LDAP is tried when appropriate (after local password failure for non-LDAP accounts, or as primary for LDAP-only accounts)
- Requires OIDC env vars when using SSO, and LDAP env vars for directory login
- See [LDAP Setup](LDAP_SETUP.md) for LDAP-specific behaviour
#### Summary Table
| Mode | Password Field | Password Required | OIDC Available | Self-Register | Use Case |
|------|---------------|-------------------|----------------|---------------|----------|
| `none` | ❌ No | ❌ No | ❌ No | ✅ Yes | Trusted internal networks, development |
| `local` | Yes | ✅ Yes | ❌ No | ✅ Yes | Standard password authentication |
| `oidc` | ❌ No | ❌ No | ✅ Yes | ✅ Yes | Enterprise SSO only |
| `both` | Yes | ✅ Yes | ✅ Yes | ✅ Yes | Mixed authentication (SSO + local) |
| Mode | Password field | OIDC | LDAP | Self-register (local) | Typical use |
|------|----------------|------|------|------------------------|-------------|
| `none` | Optional | ❌ | ❌ | ✅ | Trusted / dev |
| `local` | Yes | | ❌ | ✅ | Default self-hosted |
| `oidc` | N/A (redirect) | ✅ | ❌ | ✅ (first SSO) | SSO only |
| `both` | Yes | ✅ | ❌ | ✅ | SSO + local |
| `ldap` | Yes | ❌ | ✅ | ❌ | Directory only |
| `all` | Yes | ✅ | ✅ | ✅ | Enterprise: all methods |
### 6) Docker Compose Example
@@ -183,7 +195,7 @@ services:
### 8) Troubleshooting
- “SSO button doesnt appear”
- Check `AUTH_METHOD`. Must be `oidc` or `both`.
- Check `AUTH_METHOD`. Must be `oidc`, `both`, or `all` (SSO is not shown for `local`, `ldap`, or `none`).
- “Redirect URI mismatch”
- The `OIDC_REDIRECT_URI` must exactly match the value registered at your IdP.
@@ -218,25 +230,24 @@ services:
### 10) Database Changes
The app includes a migration that adds the following to `users`:
Relevant columns on `users` for SSO and account linking include:
- `email` (nullable)
- `oidc_issuer` (nullable)
- `oidc_issuer`, `oidc_sub` (nullable), with a unique constraint on `(oidc_issuer, oidc_sub)` where applicable
- `auth_provider` (`local`, `oidc`, or `ldap`) — set automatically from the login path; used to avoid local password login for LDAP-managed users when `AUTH_METHOD=all`
### Advanced Configuration: DNS Resolution
If your DB was not migrated automatically, run your usual migration flow (e.g. `flask db upgrade` / Alembic).
If you're experiencing DNS resolution issues (especially in Docker environments), TimeTracker includes enhanced DNS resolution features:
#### Advanced: DNS resolution (OIDC metadata)
- **Multiple DNS Strategies**: Automatically tries different DNS resolution methods
- **IP Address Caching**: Caches resolved IPs to reduce lookup overhead
- **Docker Network Detection**: Automatically tries Docker internal service names
- **Background Refresh**: Periodically refreshes metadata to keep it current
If you're experiencing DNS resolution issues (especially in Docker environments), TimeTracker includes enhanced DNS resolution for OIDC metadata:
See [TROUBLESHOOTING_OIDC_DNS.md](../../TROUBLESHOOTING_OIDC_DNS.md) for detailed information and troubleshooting steps.
- `oidc_sub` (nullable)
- Unique constraint on `(oidc_issuer, oidc_sub)`
- **Multiple DNS strategies**: Automatically tries different resolution methods
- **IP address caching**: Reduces lookup overhead
- **Docker network detection**: Tries internal service names when external DNS fails
- **Background refresh**: Keeps metadata current
If your DB wasnt migrated automatically, run your usual migration flow.
See [TROUBLESHOOTING_OIDC_DNS.md](../../TROUBLESHOOTING_OIDC_DNS.md) for detailed steps.
### 11) Support
+1 -1
View File
@@ -38,7 +38,7 @@ The blueprint sets:
- **FLASK_APP**: `app:create_app()` (used for `flask db upgrade` in the pre-deploy step)
- **SECRET_KEY**: Auto-generated by Render (recommended; you can override in the Dashboard)
- **DATABASE_URL**: Filled automatically from the linked PostgreSQL database
- **AUTH_METHOD**: `local` (username/password login)
- **AUTH_METHOD**: `local` (default), or `none` | `oidc` | `ldap` | `both` | `all` — see `env.example` and [LDAP Setup](../admin/configuration/LDAP_SETUP.md) / [OIDC Setup](../admin/configuration/OIDC_SETUP.md)
- **REDIS_ENABLED**: `false` (rate limiting uses in-memory storage; no Redis required for demo)
## Demo mode (single-user demo)
+26 -3
View File
@@ -60,11 +60,13 @@ ALLOW_SELF_REGISTER=true
ADMIN_USERNAMES=admin
# Authentication
# Options: none | local | oidc | both
# Options: none | local | oidc | ldap | both | all
# none = No password authentication (username only)
# local = Password authentication required
# oidc = OIDC/Single Sign-On only
# both = OIDC + local password authentication
# ldap = LDAP bind only
# both = OIDC + local password (backwards compatible)
# all = local + OIDC + LDAP
AUTH_METHOD=local
# Security hardening (recommended)
@@ -80,7 +82,28 @@ AUTH_METHOD=local
# Require TOTP 2FA for admin accounts (admins will be prompted to enroll)
# REQUIRE_2FA_FOR_ADMINS=false
# OIDC (used when AUTH_METHOD=oidc or both)
# LDAP Authentication (used when AUTH_METHOD=ldap or all)
# LDAP_HOST=ldap.example.com
# LDAP_PORT=389
# LDAP_USE_SSL=false
# LDAP_USE_TLS=false
# LDAP_BIND_DN=cn=serviceaccount,dc=example,dc=com
# LDAP_BIND_PASSWORD=secret
# LDAP_BASE_DN=dc=example,dc=com
# LDAP_USER_DN=ou=users
# LDAP_USER_OBJECT_CLASS=inetOrgPerson
# LDAP_USER_LOGIN_ATTR=uid
# LDAP_USER_EMAIL_ATTR=mail
# LDAP_USER_FNAME_ATTR=givenName
# LDAP_USER_LNAME_ATTR=sn
# LDAP_GROUP_DN=ou=groups
# LDAP_GROUP_OBJECT_CLASS=groupOfNames
# LDAP_ADMIN_GROUP=timetracker-admins
# LDAP_REQUIRED_GROUP=timetracker-users
# LDAP_TLS_CA_CERT_FILE=/certs/ldap-ca.crt
# LDAP_TIMEOUT=10
# OIDC (used when AUTH_METHOD=oidc, both, or all)
# OIDC_ISSUER=https://login.microsoftonline.com/<tenant>/v2.0
# OIDC_CLIENT_ID=
# OIDC_CLIENT_SECRET=
@@ -0,0 +1,53 @@
"""Add users.auth_provider for local / oidc / ldap.
Revision ID: 153_add_user_auth_provider
Revises: 152_add_user_totp_2fa
Create Date: 2026-04-27
"""
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "153_add_user_auth_provider"
down_revision = "152_add_user_totp_2fa"
branch_labels = None
depends_on = None
def _has_column(inspector, table_name: str, column_name: str) -> bool:
try:
return column_name in {c["name"] for c in inspector.get_columns(table_name)}
except Exception:
return False
def upgrade():
bind = op.get_bind()
inspector = inspect(bind)
if "users" not in inspector.get_table_names():
return
if not _has_column(inspector, "users", "auth_provider"):
op.add_column(
"users",
sa.Column(
"auth_provider",
sa.String(length=20),
nullable=False,
server_default="local",
),
)
try:
op.alter_column("users", "auth_provider", server_default=None)
except Exception:
pass
def downgrade():
bind = op.get_bind()
inspector = inspect(bind)
if "users" not in inspector.get_table_names():
return
if _has_column(inspector, "users", "auth_provider"):
op.drop_column("users", "auth_provider")
+3
View File
@@ -9,6 +9,9 @@ Flask-SocketIO==5.3.6
Authlib==1.3.1
PyJWT==2.8.0
# LDAP (directory authentication)
ldap3==2.9.1
# Database
SQLAlchemy==2.0.23
alembic==1.13.1
+245
View File
@@ -0,0 +1,245 @@
"""Tests for LDAP authentication service and login integration."""
from __future__ import annotations
import re
import tempfile
import uuid
from unittest.mock import MagicMock, patch
import pytest
from app import db
from app.models import User
@pytest.fixture
def ldap_app_config(app_config):
cfg = dict(app_config)
cfg["AUTH_METHOD"] = "ldap"
cfg["LDAP_ENABLED"] = True
cfg["LDAP_HOST"] = "ldap.test"
cfg["LDAP_PORT"] = 389
cfg["LDAP_USE_SSL"] = False
cfg["LDAP_USE_TLS"] = False
cfg["LDAP_BIND_DN"] = "cn=svc,dc=example,dc=com"
cfg["LDAP_BIND_PASSWORD"] = "svc-secret"
cfg["LDAP_BASE_DN"] = "dc=example,dc=com"
cfg["LDAP_USER_DN"] = "ou=users"
cfg["LDAP_USER_OBJECT_CLASS"] = "inetOrgPerson"
cfg["LDAP_USER_LOGIN_ATTR"] = "uid"
cfg["LDAP_USER_EMAIL_ATTR"] = "mail"
cfg["LDAP_USER_FNAME_ATTR"] = "givenName"
cfg["LDAP_USER_LNAME_ATTR"] = "sn"
cfg["LDAP_GROUP_DN"] = "ou=groups"
cfg["LDAP_GROUP_OBJECT_CLASS"] = "groupOfNames"
cfg["LDAP_ADMIN_GROUP"] = ""
cfg["LDAP_REQUIRED_GROUP"] = ""
cfg["LDAP_TLS_CA_CERT_FILE"] = ""
cfg["LDAP_TIMEOUT"] = 10
return cfg
@pytest.fixture
def ldap_app(ldap_app_config):
from app import create_app
unique_db_path = tempfile.gettempdir() + f"/pytest_ldap_{uuid.uuid4().hex}.sqlite"
cfg = dict(ldap_app_config)
cfg["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{unique_db_path}"
application = create_app(cfg)
with application.app_context():
db.create_all()
from app.models import Role
for name in ("user", "admin"):
if not Role.query.filter_by(name=name).first():
db.session.add(Role(name=name))
db.session.commit()
yield application
with application.app_context():
db.drop_all()
def _mock_ldap_entry(dn: str, uid: str, mail: str, given: str = "A", sn: str = "B"):
entry = MagicMock()
entry.entry_dn = dn
entry.entry_attributes_as_dict = {
"uid": [uid],
"mail": [mail],
"givenName": [given],
"sn": [sn],
}
return entry
@patch("app.services.ldap_service._service_connection")
@patch("app.services.ldap_service.Connection")
def test_ldap_authenticate_success(mock_conn_cls, mock_svc, ldap_app):
svc = MagicMock()
svc.entries = [_mock_ldap_entry("uid=test,ou=users,dc=example,dc=com", "test", "test@example.com")]
svc.bound = True
mock_svc.return_value = svc
user_conn = MagicMock()
mock_conn_cls.return_value = user_conn
with ldap_app.app_context():
from app.services.ldap_service import LDAPService
u = LDAPService.authenticate("test", "secret")
assert u is not None
assert u.email == "test@example.com"
assert u.auth_provider == "ldap"
assert User.query.filter_by(email="test@example.com").first() is not None
@patch("app.services.ldap_service._service_connection")
@patch("app.services.ldap_service.Connection")
def test_ldap_authenticate_wrong_password(mock_conn_cls, mock_svc, ldap_app):
from ldap3.core.exceptions import LDAPBindError
svc = MagicMock()
svc.entries = [_mock_ldap_entry("uid=test,ou=users,dc=example,dc=com", "test", "test@example.com")]
svc.bound = True
mock_svc.return_value = svc
mock_conn_cls.side_effect = LDAPBindError("bad")
with ldap_app.app_context():
from app.services.ldap_service import LDAPService
assert LDAPService.authenticate("test", "wrong") is None
@patch("app.services.ldap_service._user_dn_member_of_group")
@patch("app.services.ldap_service._service_connection")
@patch("app.services.ldap_service.Connection")
def test_ldap_required_group_blocks_non_member(mock_conn_cls, mock_svc, mock_member, ldap_app):
ldap_app.config["LDAP_REQUIRED_GROUP"] = "users"
mock_member.return_value = False
svc = MagicMock()
svc.entries = [_mock_ldap_entry("uid=test,ou=users,dc=example,dc=com", "test", "test@example.com")]
svc.bound = True
mock_svc.return_value = svc
with ldap_app.app_context():
from app.services.ldap_service import LDAPService
assert LDAPService.authenticate("test", "secret") is None
@patch("app.services.ldap_service._service_connection")
@patch("app.services.ldap_service.Connection")
def test_ldap_admin_group_grants_admin(mock_conn_cls, mock_svc, ldap_app):
ldap_app.config["LDAP_ADMIN_GROUP"] = "admins"
svc1 = MagicMock()
svc1.entries = [_mock_ldap_entry("uid=adm,ou=users,dc=example,dc=com", "adm", "adm@example.com")]
svc1.bound = True
svc2 = MagicMock()
svc2.bound = True
mock_svc.side_effect = [svc1, svc2]
with patch("app.services.ldap_service._user_dn_member_of_group", return_value=True):
mock_conn_cls.return_value = MagicMock()
with ldap_app.app_context():
from app.services.ldap_service import LDAPService
u = LDAPService.authenticate("adm", "pw")
assert u is not None
assert u.role == "admin"
@patch("app.services.ldap_service._service_connection")
@patch("app.services.ldap_service.Connection")
def test_ldap_syncs_attributes_on_relogin(mock_conn_cls, mock_svc, ldap_app):
svc1 = MagicMock()
svc1.entries = [_mock_ldap_entry("uid=u1,ou=users,dc=example,dc=com", "u1", "sync@example.com", "Old", "Name")]
svc1.bound = True
mock_svc.return_value = svc1
mock_conn_cls.return_value = MagicMock()
with ldap_app.app_context():
from app.services.ldap_service import LDAPService
u1 = LDAPService.authenticate("u1", "pw")
assert u1.full_name == "Old Name"
svc2 = MagicMock()
svc2.entries = [_mock_ldap_entry("uid=u1,ou=users,dc=example,dc=com", "u1", "sync@example.com", "New", "Name")]
svc2.bound = True
mock_svc.return_value = svc2
with ldap_app.app_context():
from app.services.ldap_service import LDAPService
u2 = LDAPService.authenticate("u1", "pw")
assert u2.full_name == "New Name"
@patch("app.services.ldap_service._service_connection")
def test_ldap_exception_returns_none(mock_svc, ldap_app):
from ldap3.core.exceptions import LDAPException
mock_svc.side_effect = LDAPException("network")
with ldap_app.app_context():
from app.services.ldap_service import LDAPService
assert LDAPService.authenticate("x", "y") is None
def test_login_route_ldap_success(client, app):
app.config["AUTH_METHOD"] = "ldap"
app.config["LDAP_ENABLED"] = True
with app.app_context():
from app.models import Role
for name in ("user", "admin"):
if not Role.query.filter_by(name=name).first():
db.session.add(Role(name=name))
db.session.commit()
u = User(username="ldapuser", role="user", email="ldapuser@example.com")
u.auth_provider = "ldap"
u.set_password("unused")
u.is_active = True
ro = Role.query.filter_by(name="user").first()
if ro:
u.roles.append(ro)
db.session.add(u)
db.session.commit()
def fake_authenticate(username, password):
if username == "ldapuser" and password == "ok":
with app.app_context():
return User.query.filter_by(username="ldapuser").first()
return None
with patch("app.services.ldap_service.LDAPService.authenticate", staticmethod(fake_authenticate)):
resp = client.post(
"/login",
data={"username": "ldapuser", "password": "ok"},
follow_redirects=False,
)
assert resp.status_code in (302, 303)
def test_login_route_ldap_failure_generic_message(client, app):
app.config["AUTH_METHOD"] = "ldap"
app.config["LDAP_ENABLED"] = True
with patch("app.services.ldap_service.LDAPService.authenticate", staticmethod(lambda u, p: None)):
resp = client.post(
"/login",
data={"username": "nouser", "password": "bad"},
follow_redirects=False,
)
assert resp.status_code == 200
body = resp.get_data(as_text=True)
m = re.search(r'data-toast-message="([^"]*)"', body)
assert m, "expected flash toast in response"
assert m.group(1) == "Invalid username or password"
assert "ldap" not in m.group(1).lower()
+1 -1
View File
@@ -268,7 +268,7 @@ def test_logout_configuration_keys_valid(app):
# These should be accessible without errors
auth_method = getattr(Config, "AUTH_METHOD", None)
assert auth_method in ["local", "oidc", "both", None]
assert auth_method in ["local", "oidc", "both", "all", None]
# OIDC_POST_LOGOUT_REDIRECT_URI should be optional
post_logout = getattr(Config, "OIDC_POST_LOGOUT_REDIRECT_URI", None)