Daniel Bedeleanu
dc999991a5
- Added /users/auth-mode public endpoint to detect LDAP status - Updated frontend to call getAuthMode() instead of hardcoded false - Login page now automatically switches to enterprise mode when LDAP is enabled - Fixes missing LDAP user login UI that was previously disabled with TODO comment This restores the functionality that was commented out earlier where the login page would auto-detect whether to show local user list or enterprise username input.
375 lines
16 KiB
Python
375 lines
16 KiB
Python
import secrets
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from sqlalchemy.orm import Session
|
|
from slowapi import Limiter
|
|
from slowapi.util import get_remote_address
|
|
from passlib.context import CryptContext
|
|
import ldap3
|
|
from ldap3 import Tls
|
|
from ldap3.utils.conv import escape_filter_chars
|
|
from ldap3.utils.dn import escape_rdn
|
|
import ssl
|
|
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
from .. import models, schemas, database, auth
|
|
from ..config_loader import get_config
|
|
from ..config_manager import ConfigManager
|
|
from ..logger import log
|
|
|
|
router = APIRouter(prefix="/users", tags=["auth"])
|
|
limiter = Limiter(key_func=get_remote_address)
|
|
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
|
|
|
|
|
def get_ldap_config():
|
|
config = get_config()
|
|
auth_config = config.get("auth", {})
|
|
return {
|
|
"ldap_enabled": auth_config.get("ldap_enabled", False),
|
|
"server_uri": auth_config.get("ldap_server"),
|
|
"base_dn": auth_config.get("ldap_base_dn"),
|
|
"user_template": auth_config.get("ldap_user_template"),
|
|
"groups_dn": auth_config.get("ldap_groups_dn"),
|
|
"use_tls": auth_config.get("ldap_use_tls", True),
|
|
"ignore_cert": auth_config.get("ldap_ignore_cert", False),
|
|
"role_mappings": auth_config.get("ldap_role_mappings", [])
|
|
}
|
|
|
|
|
|
def authenticate_ldap(username, password):
|
|
config = get_ldap_config()
|
|
if not config.get("ldap_enabled"):
|
|
log.debug("LDAP: LDAP is disabled in config")
|
|
return None
|
|
|
|
log.debug(f"LDAP: Config loaded: server_uri={config.get('server_uri')}, base_dn={config.get('base_dn')}")
|
|
try:
|
|
tls_config = None
|
|
if config.get("use_tls", False):
|
|
if config.get("ignore_cert", False):
|
|
# [SECURITY] CERT_NONE is only for internal test environments with self-signed certs
|
|
tls_config = Tls(validate=ssl.CERT_NONE, version=ssl.PROTOCOL_TLSv1_2)
|
|
log.warning("LDAP: TLS Certificate Validation DISABLED (ignore_cert=true)")
|
|
else:
|
|
tls_config = Tls(validate=ssl.CERT_REQUIRED, version=ssl.PROTOCOL_TLSv1_2)
|
|
log.debug("LDAP: TLS Certificate Validation ENABLED (CERT_REQUIRED)")
|
|
|
|
server = ldap3.Server(
|
|
config["server_uri"],
|
|
use_ssl=config.get("use_tls", False),
|
|
tls=tls_config,
|
|
get_info=ldap3.ALL
|
|
)
|
|
log.debug(f"LDAP: Server object created: {config['server_uri']}")
|
|
safe_username_rdn = escape_rdn(username)
|
|
user_dn = config["user_template"].format(username=safe_username_rdn)
|
|
log.debug(f"LDAP: Attempting bind for DN: {user_dn}")
|
|
|
|
# [FIX] Ensure password is UTF-8 encoded for LDAP protocol
|
|
password_bytes = password.encode('utf-8') if isinstance(password, str) else password
|
|
conn = ldap3.Connection(server, user=user_dn, password=password_bytes, auto_bind=True)
|
|
log.debug(f"LDAP: Bind successful for {user_dn}")
|
|
|
|
# Search for the user to get their CANONICAL DN
|
|
# [SECURITY FIX H-01] Escape username before interpolating into LDAP filter
|
|
base_dn = config.get("base_dn", "dc=example,dc=org")
|
|
safe_username = escape_filter_chars(username)
|
|
search_filter = f"(|(cn={safe_username})(uid={safe_username}))"
|
|
conn.search(base_dn, search_filter, attributes=['cn', 'uid'])
|
|
|
|
if not conn.entries:
|
|
log.debug(f"LDAP: User not found in search after bind.")
|
|
return None
|
|
|
|
real_user_dn = conn.entries[0].entry_dn
|
|
user_groups = []
|
|
if hasattr(conn.entries[0], 'memberOf'):
|
|
user_groups = [str(g).lower() for g in conn.entries[0].memberOf.values]
|
|
log.debug(f"LDAP: Found memberOf groups on user: {user_groups}")
|
|
|
|
log.debug(f"LDAP: Canonical DN found: {real_user_dn}")
|
|
|
|
# Check roles based on group membership
|
|
assigned_role = None
|
|
|
|
# New multi-group mapping support
|
|
role_mappings = config.get("role_mappings", [])
|
|
if not role_mappings and config.get("required_group"):
|
|
# Fallback to legacy single-group config
|
|
role_mappings = [{"group": config["required_group"], "role": "user"}]
|
|
|
|
groups_dn = config.get("groups_dn", "ou=groups")
|
|
|
|
# Iterate through mappings to find the highest role
|
|
potential_roles = []
|
|
|
|
for mapping in role_mappings:
|
|
group_name = mapping["group"]
|
|
target_role = mapping["role"]
|
|
|
|
# Construct group DN if it's just a common name
|
|
if "=" not in group_name:
|
|
full_group_dn = f"cn={group_name},{groups_dn},{base_dn}"
|
|
else:
|
|
full_group_dn = group_name
|
|
|
|
full_group_dn_lower = full_group_dn.lower()
|
|
|
|
log.debug(f"LDAP: Checking membership in group: {full_group_dn}")
|
|
|
|
# Method 1: Check memberOf if available (AD/LLDAP)
|
|
if full_group_dn_lower in user_groups:
|
|
log.debug(f"LDAP: Match found via memberOf for {target_role}")
|
|
potential_roles.append(target_role)
|
|
continue
|
|
|
|
# Method 2: Search group's member attribute (Standard LDAP)
|
|
conn.search(full_group_dn, '(objectClass=*)', attributes=['member', 'uniqueMember'])
|
|
if conn.entries:
|
|
members = []
|
|
if hasattr(conn.entries[0], 'member'):
|
|
members = [str(m).lower() for m in conn.entries[0].member.values]
|
|
elif hasattr(conn.entries[0], 'uniqueMember'):
|
|
members = [str(m).lower() for m in conn.entries[0].uniqueMember.values]
|
|
|
|
if real_user_dn.lower() in members or user_dn.lower() in members:
|
|
log.debug(f"LDAP: Match found via group search for {target_role}")
|
|
potential_roles.append(target_role)
|
|
|
|
if "admin" in potential_roles:
|
|
assigned_role = "admin"
|
|
elif "user" in potential_roles:
|
|
assigned_role = "user"
|
|
elif potential_roles:
|
|
assigned_role = potential_roles[0]
|
|
|
|
return assigned_role
|
|
except Exception as e:
|
|
err_msg = str(e)
|
|
err_type = type(e).__name__
|
|
log.error(f"LDAP: Auth Error: {err_type}: {err_msg}")
|
|
|
|
# Broad detection for SSL/TLS certificate/handshake or connectivity errors
|
|
# handles both ldapsearch style "Can't contact" and ldap3 style "socket ssl wrapping error"
|
|
ssl_indicators = ["certificate", "ssl", "tls", "handshake", "verify failed", "contact", "socket"]
|
|
|
|
if any(ind in err_msg.lower() for ind in ssl_indicators):
|
|
log.warning(f"LDAP: SSL/TLS or Connectivity issue detected: {err_msg}")
|
|
|
|
# User-friendly error message, hiding raw socket traces
|
|
friendly_msg = "Secure Connection Failed: The enterprise server's security certificate is not trusted or the connection dropped."
|
|
if config.get("use_tls"):
|
|
friendly_msg += " If this is an internal test environment, please ask an Admin to enable 'Ignore Certificate Validation'."
|
|
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail=friendly_msg
|
|
)
|
|
|
|
import traceback
|
|
log.debug(f"LDAP: Full traceback: {traceback.format_exc()}")
|
|
return None
|
|
|
|
|
|
def get_password_hash(password):
|
|
return pwd_context.hash(password)
|
|
|
|
|
|
def verify_password(plain_password, hashed_password):
|
|
if not hashed_password: return False
|
|
return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
|
|
@router.post("/login", response_model=schemas.TokenResponse)
|
|
@limiter.limit("5/minute")
|
|
def login(request: Request, form_data: schemas.UserLogin, db: Session = Depends(database.get_db)):
|
|
"""
|
|
[C-01] Login endpoint: validates credentials and returns JWT Bearer token.
|
|
"""
|
|
user = db.query(models.User).filter(models.User.username == form_data.username).first()
|
|
|
|
# Try local authentication
|
|
authenticated = False
|
|
if user and user.hashed_password:
|
|
if verify_password(form_data.password, user.hashed_password):
|
|
log.debug(f"Local auth successful for {form_data.username}")
|
|
authenticated = True
|
|
else:
|
|
log.debug(f"Local auth failed: password mismatch for {form_data.username}")
|
|
elif user and not user.hashed_password:
|
|
log.debug(f"User {form_data.username} exists but has no hashed password (LDAP user), skipping local auth")
|
|
# [SECURITY FIX C-02] Bypass for passwordless users has been removed.
|
|
# LDAP users must authenticate via the LDAP flow below.
|
|
pass
|
|
elif not user:
|
|
log.debug(f"User {form_data.username} not found in database, will try LDAP")
|
|
|
|
# If local failed, try LDAP
|
|
if not authenticated:
|
|
log.debug(f"Local auth failed for {form_data.username}, attempting LDAP")
|
|
ldap_role = authenticate_ldap(form_data.username, form_data.password)
|
|
if ldap_role:
|
|
log.debug(f"LDAP auth successful for {form_data.username}, role={ldap_role}")
|
|
authenticated = True
|
|
# Cache hash for offline support
|
|
new_hash = get_password_hash(form_data.password)
|
|
|
|
# If user doesn't exist locally, create a stub for role management
|
|
if not user:
|
|
user = models.User(
|
|
username=form_data.username,
|
|
role=ldap_role,
|
|
origin="ldap",
|
|
hashed_password=new_hash
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
else:
|
|
# Update role if it changed in LDAP and refresh cached hash
|
|
user.role = ldap_role
|
|
user.hashed_password = new_hash
|
|
db.commit()
|
|
db.refresh(user)
|
|
else:
|
|
log.warning(f"Login failed: LDAP auth also failed for {form_data.username}")
|
|
raise HTTPException(status_code=401, detail="Invalid username or password, or insufficient permissions")
|
|
|
|
if not authenticated or not user:
|
|
raise HTTPException(status_code=401, detail="Invalid username or password")
|
|
|
|
# [C-01] Generate JWT token
|
|
token = auth.create_access_token(
|
|
user_id=user.id,
|
|
username=user.username,
|
|
role=user.role
|
|
)
|
|
|
|
return schemas.TokenResponse(
|
|
access_token=token,
|
|
token_type="bearer",
|
|
user_id=user.id,
|
|
username=user.username,
|
|
role=user.role,
|
|
user=schemas.User(id=user.id, username=user.username, role=user.role, origin=user.origin)
|
|
)
|
|
|
|
|
|
@router.get("/auth-mode")
|
|
def get_auth_mode():
|
|
"""[C-01] Get authentication mode (public endpoint for login page)."""
|
|
config = get_ldap_config()
|
|
return {
|
|
"mode": "enterprise" if config.get("ldap_enabled") else "local",
|
|
"ldap_enabled": config.get("ldap_enabled", False)
|
|
}
|
|
|
|
|
|
@router.get("/ldap-config")
|
|
def get_ldap_settings(current_user: auth.TokenData = Depends(auth.get_current_admin)):
|
|
"""[C-01] Get LDAP config — admin only."""
|
|
return get_ldap_config()
|
|
|
|
|
|
@router.post("/ldap-config")
|
|
def update_ldap_settings(
|
|
config: dict,
|
|
current_user: auth.TokenData = Depends(auth.get_current_admin)
|
|
):
|
|
"""[C-01] Update LDAP config — admin only."""
|
|
# Map frontend keys back to backend YAML structure
|
|
updates = {
|
|
"auth": {
|
|
"ldap_enabled": config.get("ldap_enabled", False),
|
|
"ldap_server": config.get("server_uri", ""),
|
|
"ldap_base_dn": config.get("base_dn", ""),
|
|
"ldap_user_template": config.get("user_template", ""),
|
|
"ldap_groups_dn": config.get("groups_dn", ""),
|
|
"ldap_use_tls": config.get("use_tls", True),
|
|
"ldap_ignore_cert": config.get("ignore_cert", False),
|
|
"ldap_role_mappings": config.get("role_mappings", [])
|
|
}
|
|
}
|
|
|
|
try:
|
|
ConfigManager.update_config(updates)
|
|
log.info(f"LDAP config updated in backend.yaml by {current_user.username}")
|
|
return {"message": "Config saved successfully"}
|
|
except Exception as e:
|
|
log.error(f"Failed to update LDAP config: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to save configuration: {str(e)}")
|
|
|
|
|
|
@router.post("/test-ldap")
|
|
def test_ldap_connection(
|
|
config: dict,
|
|
current_user: auth.TokenData = Depends(auth.get_current_admin)
|
|
):
|
|
try:
|
|
# Extract host and port
|
|
uri = config["server_uri"]
|
|
host = uri.replace("ldap://", "").replace("ldaps://", "")
|
|
port = 389
|
|
if ":" in host:
|
|
host, port_str = host.split(":")
|
|
port = int(port_str)
|
|
elif "ldaps://" in uri:
|
|
port = 636
|
|
elif uri.endswith(":3890"): # Special case for LLDAP
|
|
port = 3890
|
|
|
|
# Try raw socket first
|
|
log.debug(f"LDAP test: Probing raw socket {host}:{port}")
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(5)
|
|
result = s.connect_ex((host, port))
|
|
s.close()
|
|
|
|
if result == 0:
|
|
# Socket is open! Now try LDAP library probe
|
|
try:
|
|
tls_config = None
|
|
if config.get("use_tls", False):
|
|
if config.get("ignore_cert", False):
|
|
tls_config = Tls(validate=ssl.CERT_NONE, version=ssl.PROTOCOL_TLSv1_2)
|
|
else:
|
|
tls_config = Tls(validate=ssl.CERT_REQUIRED, version=ssl.PROTOCOL_TLSv1_2)
|
|
|
|
server = ldap3.Server(
|
|
config["server_uri"],
|
|
connect_timeout=5,
|
|
get_info=ldap3.BASIC,
|
|
use_ssl=config.get("use_tls", False),
|
|
tls=tls_config
|
|
)
|
|
# Try a connection without auto-bind first to see if it's an LDAP server
|
|
conn = ldap3.Connection(server, auto_bind=False)
|
|
if conn.open():
|
|
return {"status": "success", "message": "LDAP Connection Successful (Server Reachable)"}
|
|
|
|
# If open fails, it might just be the server policy.
|
|
# Since the port is open, we report success at the network level.
|
|
return {"status": "success", "message": "Connection Successful (Network reachable, protocol handshake restricted by server security)"}
|
|
except Exception as e:
|
|
# Any LDAP level error while socket is open is still a partial success
|
|
err_msg = str(e)
|
|
if "certificate verify failed" in err_msg.lower() or "self signed certificate" in err_msg.lower():
|
|
return {"status": "error", "message": f"SSL/TLS Certificate Rejected: The server certificate is self-signed or invalid. Enable 'Ignore Certificate Validation' to bypass."}
|
|
return {"status": "success", "message": f"Partial Success: TCP Port {port} is open, but LDAP handshake was rejected: {err_msg}"}
|
|
else:
|
|
# Socket failed, let's try calling system 'ldapsearch' as a last resort diagnostic
|
|
try:
|
|
# We just try to reach the server with a 2s timeout
|
|
cmd = ["ldapsearch", "-h", host, "-p", str(port), "-x", "-s", "base", "-b", "", "namingContexts"]
|
|
proc = subprocess.run(cmd, capture_output=True, timeout=2)
|
|
if proc.returncode == 0 or b"namingContexts" in proc.stdout:
|
|
return {"status": "error", "message": f"SYSTEM CAN CONNECT, BUT PYTHON IS BLOCKED. Check Mac Firewall settings for Python."}
|
|
except:
|
|
pass
|
|
return {"status": "error", "message": f"TCP Port {port} is closed or unreachable (Error code: {result}). Check firewall on {host}."}
|
|
|
|
except Exception as e:
|
|
return {"status": "error", "message": f"Network Error: {str(e)}"}
|