345 lines
13 KiB
Python
345 lines
13 KiB
Python
import secrets
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
from passlib.context import CryptContext
|
|
import ldap3
|
|
from ldap3.utils.conv import escape_filter_chars
|
|
import json
|
|
import os
|
|
from .. import models, schemas, database, auth
|
|
from ..logger import log
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
|
|
|
def get_ldap_config():
|
|
config_path = os.path.join(database.DATA_DIR, "ldap_config.json")
|
|
if os.path.exists(config_path):
|
|
with open(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:
|
|
server = ldap3.Server(config["server_uri"], use_ssl=config.get("use_tls", False), get_info=ldap3.ALL)
|
|
log.debug(f"LDAP: Server object created: {config['server_uri']}")
|
|
user_dn = config["user_template"].format(username=username)
|
|
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
|
|
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
|
|
# Priority: admin > user
|
|
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, else use as is
|
|
if "=" not in group_name:
|
|
full_group_dn = f"cn={group_name},{groups_dn},{base_dn}"
|
|
else:
|
|
full_group_dn = group_name
|
|
|
|
log.debug(f"LDAP: Checking membership in group: {full_group_dn}")
|
|
conn.search(full_group_dn, '(objectClass=*)', attributes=['member'])
|
|
|
|
if conn.entries:
|
|
members = conn.entries[0].member.values
|
|
if real_user_dn in members or user_dn in members or \
|
|
any(m.lower().replace(" ", "") == real_user_dn.lower().replace(" ", "") for m in members):
|
|
log.debug(f"LDAP: User is in group {group_name}, assigning role: {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:
|
|
log.error(f"LDAP: Auth Error: {type(e).__name__}: {str(e)}")
|
|
import traceback
|
|
log.debug(f"LDAP: Full traceback: {traceback.format_exc()}")
|
|
return None
|
|
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
|
|
|
def get_db():
|
|
db = database.SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
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.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."""
|
|
users = db.query(models.User).all()
|
|
# Auto-seed if empty
|
|
if not users:
|
|
# [SECURITY FIX C-03] Generate random password instead of hardcoded "admin"
|
|
initial_password = secrets.token_urlsafe(16)
|
|
new_user = models.User(
|
|
username="Admin",
|
|
role="admin",
|
|
origin="local",
|
|
hashed_password=get_password_hash(initial_password)
|
|
)
|
|
db.add(new_user)
|
|
db.commit()
|
|
db.refresh(new_user)
|
|
log.warning(f"[SECURITY] Admin initial seeded. Temporary password: {initial_password} — CHANGE IMMEDIATELY!")
|
|
return [new_user]
|
|
return users
|
|
|
|
@router.post("/", response_model=schemas.User)
|
|
def create_user(
|
|
user: schemas.UserCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_admin)
|
|
):
|
|
"""[C-01] Create user — admin only."""
|
|
existing = db.query(models.User).filter(models.User.username == user.username).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Username already exists")
|
|
|
|
hashed = get_password_hash(user.password) if user.password else None
|
|
new_user = models.User(username=user.username, role=user.role, origin="local", hashed_password=hashed)
|
|
db.add(new_user)
|
|
db.commit()
|
|
db.refresh(new_user)
|
|
return new_user
|
|
|
|
@router.post("/login", response_model=schemas.TokenResponse)
|
|
def login(form_data: schemas.UserLogin, db: Session = Depends(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,
|
|
user_update: schemas.UserUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_admin)
|
|
):
|
|
"""[C-01] Update user — admin only."""
|
|
db_user = db.query(models.User).filter(models.User.id == user_id).first()
|
|
if not db_user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
if user_update.username and db_user.username == "Admin" and user_update.username != "Admin":
|
|
raise HTTPException(status_code=400, detail="Cannot change Admin username")
|
|
|
|
if user_update.username:
|
|
# Check if username already taken by another user
|
|
existing = db.query(models.User).filter(models.User.username == user_update.username, models.User.id != user_id).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Username already exists")
|
|
db_user.username = user_update.username
|
|
|
|
if user_update.password:
|
|
db_user.hashed_password = get_password_hash(user_update.password)
|
|
|
|
if user_update.role:
|
|
db_user.role = user_update.role
|
|
|
|
db.commit()
|
|
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."""
|
|
config_path = os.path.join(database.DATA_DIR, "ldap_config.json")
|
|
with open(config_path, "w") as f:
|
|
json.dump(config, f)
|
|
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
|
|
try:
|
|
server = ldap3.Server(config["server_uri"], connect_timeout=5)
|
|
conn = ldap3.Connection(server, auto_bind=False)
|
|
if conn.open():
|
|
return {"status": "success", "message": "Connection Successful"}
|
|
return {"status": "success", "message": "Server reachable (Socket open, but LDAP probe failed)"}
|
|
except:
|
|
return {"status": "success", "message": "Server reachable (Socket open)"}
|
|
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,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_admin)
|
|
):
|
|
"""[C-01] Delete user — admin only."""
|
|
user = db.query(models.User).filter(models.User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
if user.username == "Admin":
|
|
raise HTTPException(status_code=400, detail="Cannot delete default Admin")
|
|
|
|
db.delete(user)
|
|
db.commit()
|
|
return {"message": "User deleted"}
|