refactor: split LDAP auth into backend/routers/auth.py
This commit is contained in:
@@ -6,7 +6,7 @@ from slowapi import Limiter
|
||||
from slowapi.util import get_remote_address
|
||||
from . import models
|
||||
from .database import engine
|
||||
from .routers import items, operations, users, categories
|
||||
from .routers import items, operations, users, auth, categories
|
||||
from .routers.admin import backups, config
|
||||
from .logger import log
|
||||
from .scheduler import scheduler, sync_scheduler_config
|
||||
@@ -87,6 +87,7 @@ app.state.limiter = limiter
|
||||
app.include_router(items.router)
|
||||
app.include_router(operations.router)
|
||||
app.include_router(users.router)
|
||||
app.include_router(auth.router)
|
||||
app.include_router(categories.router)
|
||||
app.include_router(backups.router)
|
||||
app.include_router(config.router)
|
||||
|
||||
348
backend/routers/auth.py
Normal file
348
backend/routers/auth.py
Normal file
@@ -0,0 +1,348 @@
|
||||
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 ..logger import log
|
||||
|
||||
router = APIRouter(prefix="/auth", tags=["auth"])
|
||||
limiter = Limiter(key_func=get_remote_address)
|
||||
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
||||
|
||||
|
||||
def get_ldap_config():
|
||||
# Priority 1: Check in DATA_DIR (for Docker production)
|
||||
config_path = os.path.join(database.DATA_DIR, "config", "ldap_config.json")
|
||||
if os.path.exists(config_path):
|
||||
with open(config_path, "r") as f:
|
||||
return json.load(f)
|
||||
|
||||
# Priority 2: Fallback to source-relative config (for local dev)
|
||||
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
source_config_path = os.path.join(root_dir, "config", "ldap_config.json")
|
||||
if os.path.exists(source_config_path):
|
||||
with open(source_config_path, "r") as f:
|
||||
return json.load(f)
|
||||
|
||||
return {"ldap_enabled": False}
|
||||
|
||||
|
||||
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}")
|
||||
|
||||
conn = ldap3.Connection(server, user=user_dn, password=password, 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
|
||||
)
|
||||
|
||||
|
||||
@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."""
|
||||
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
config_dir = os.path.join(root_dir, "config")
|
||||
os.makedirs(config_dir, exist_ok=True)
|
||||
config_path = os.path.join(config_dir, "ldap_config.json")
|
||||
with open(config_path, "w") as f:
|
||||
json.dump(config, f, indent=2)
|
||||
log.info(f"LDAP config updated by {current_user.username}")
|
||||
return {"message": "Config saved"}
|
||||
|
||||
|
||||
@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)}"}
|
||||
@@ -1,172 +1,11 @@
|
||||
import secrets
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
from typing import List
|
||||
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
|
||||
from .. import models, schemas, database, auth
|
||||
from ..logger import log
|
||||
|
||||
router = APIRouter(prefix="/users", tags=["users"])
|
||||
limiter = Limiter(key_func=get_remote_address)
|
||||
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
||||
|
||||
def get_ldap_config():
|
||||
# Priority 1: Check in DATA_DIR (for Docker production)
|
||||
config_path = os.path.join(database.DATA_DIR, "config", "ldap_config.json")
|
||||
if os.path.exists(config_path):
|
||||
with open(config_path, "r") as f:
|
||||
return json.load(f)
|
||||
|
||||
# Priority 2: Fallback to source-relative config (for local dev)
|
||||
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
source_config_path = os.path.join(root_dir, "config", "ldap_config.json")
|
||||
if os.path.exists(source_config_path):
|
||||
with open(source_config_path, "r") as f:
|
||||
return json.load(f)
|
||||
|
||||
return {"ldap_enabled": False}
|
||||
|
||||
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}")
|
||||
|
||||
conn = ldap3.Connection(server, user=user_dn, password=password, 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
|
||||
|
||||
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
||||
|
||||
def get_db():
|
||||
@@ -183,6 +22,7 @@ def verify_password(plain_password, hashed_password):
|
||||
if not hashed_password: return False
|
||||
return pwd_context.verify(plain_password, hashed_password)
|
||||
|
||||
|
||||
@router.get("/", response_model=List[schemas.User])
|
||||
def get_users(db: Session = Depends(get_db)):
|
||||
"""[C-01] User list — public endpoint for login page to enumerate local users."""
|
||||
@@ -223,79 +63,6 @@ def create_user(
|
||||
db.refresh(new_user)
|
||||
return new_user
|
||||
|
||||
@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
|
||||
)
|
||||
|
||||
@router.put("/{user_id}", response_model=schemas.User)
|
||||
def update_user(
|
||||
user_id: int,
|
||||
@@ -328,99 +95,6 @@ def update_user(
|
||||
db.refresh(db_user)
|
||||
return db_user
|
||||
|
||||
@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."""
|
||||
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
config_dir = os.path.join(root_dir, "config")
|
||||
os.makedirs(config_dir, exist_ok=True)
|
||||
config_path = os.path.join(config_dir, "ldap_config.json")
|
||||
with open(config_path, "w") as f:
|
||||
json.dump(config, f, indent=2)
|
||||
log.info(f"LDAP config updated by {current_user.username}")
|
||||
return {"message": "Config saved"}
|
||||
|
||||
@router.post("/test-ldap")
|
||||
def test_ldap_connection(
|
||||
config: dict,
|
||||
current_user: auth.TokenData = Depends(auth.get_current_admin)
|
||||
):
|
||||
import socket
|
||||
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
|
||||
import subprocess
|
||||
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)}"}
|
||||
|
||||
@router.delete("/{user_id}")
|
||||
def delete_user(
|
||||
user_id: int,
|
||||
|
||||
@@ -14,6 +14,7 @@ from backend.models import User
|
||||
from backend.config_manager import ConfigManager
|
||||
from backend.auth import get_current_admin, TokenData
|
||||
import backend.routers.users as users_router
|
||||
import backend.routers.auth as auth_router
|
||||
import backend.routers.categories as categories_router
|
||||
|
||||
|
||||
@@ -84,8 +85,8 @@ def test_client(test_db: Session) -> Generator[TestClient, None, None]:
|
||||
@pytest.fixture(scope="function")
|
||||
def mock_ldap() -> Generator[MagicMock, None, None]:
|
||||
"""Mock LDAP authentication."""
|
||||
with patch("backend.routers.users.ldap3.Server") as mock_server, \
|
||||
patch("backend.routers.users.ldap3.Connection") as mock_conn_class:
|
||||
with patch("backend.routers.auth.ldap3.Server") as mock_server, \
|
||||
patch("backend.routers.auth.ldap3.Connection") as mock_conn_class:
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.bind.return_value = True
|
||||
|
||||
@@ -20,9 +20,9 @@ class TestUserAuthentication:
|
||||
"groups_dn": "ou=groups,dc=ainventory,dc=local",
|
||||
"role_mappings": [{"group": "cn=inventory_users,ou=groups,dc=ainventory,dc=local", "role": "user"}],
|
||||
}
|
||||
with patch("backend.routers.users.get_ldap_config", return_value=ldap_config):
|
||||
with patch("backend.routers.auth.get_ldap_config", return_value=ldap_config):
|
||||
response = test_client.post(
|
||||
"/users/login",
|
||||
"/auth/login",
|
||||
json={"username": "testuser", "password": "password123"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
@@ -42,11 +42,11 @@ class TestUserAuthentication:
|
||||
"groups_dn": "ou=groups,dc=ainventory,dc=local",
|
||||
"role_mappings": [{"group": "cn=inventory_users,ou=groups,dc=ainventory,dc=local", "role": "user"}],
|
||||
}
|
||||
with patch("backend.routers.users.get_ldap_config", return_value=ldap_config), \
|
||||
patch("backend.routers.users.ldap3.Server"), \
|
||||
patch("backend.routers.users.ldap3.Connection", side_effect=Exception("Invalid credentials")):
|
||||
with patch("backend.routers.auth.get_ldap_config", return_value=ldap_config), \
|
||||
patch("backend.routers.auth.ldap3.Server"), \
|
||||
patch("backend.routers.auth.ldap3.Connection", side_effect=Exception("Invalid credentials")):
|
||||
response = test_client.post(
|
||||
"/users/login",
|
||||
"/auth/login",
|
||||
json={"username": "testuser", "password": "wrongpassword"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
@@ -54,7 +54,7 @@ class TestUserAuthentication:
|
||||
def test_login_local_password(self, test_client, test_db):
|
||||
"""Test local password authentication (fallback)."""
|
||||
from backend.models import User
|
||||
from backend.routers.users import get_password_hash
|
||||
from backend.routers.auth import get_password_hash
|
||||
|
||||
# Create local user
|
||||
user = User(
|
||||
@@ -67,7 +67,7 @@ class TestUserAuthentication:
|
||||
test_db.commit()
|
||||
|
||||
response = test_client.post(
|
||||
"/users/login",
|
||||
"/auth/login",
|
||||
json={"username": "localuser", "password": "password123"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
Reference in New Issue
Block a user