diff --git a/backend/main.py b/backend/main.py index e4bbcf3f..28f0a5af 100644 --- a/backend/main.py +++ b/backend/main.py @@ -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) diff --git a/backend/routers/auth.py b/backend/routers/auth.py new file mode 100644 index 00000000..3ee7393f --- /dev/null +++ b/backend/routers/auth.py @@ -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)}"} diff --git a/backend/routers/users.py b/backend/routers/users.py index 8a8e6de6..bd3aa538 100644 --- a/backend/routers/users.py +++ b/backend/routers/users.py @@ -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, diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 451c92b9..5410159f 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -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 diff --git a/backend/tests/test_users.py b/backend/tests/test_users.py index 6bf652ca..763969f6 100644 --- a/backend/tests/test_users.py +++ b/backend/tests/test_users.py @@ -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