LDAP: restriction groupe AD + auto-provisioning users (sans permissions)
- Settings ldap_required_group (DN groupe autorise) + ldap_default_role - ldap_authenticate verifie memberOf vs required_group avant bind - auth.py: si user inconnu + LDAP + groupe OK -> auto-create user, role default, zero permission (admin doit assigner via /users)
This commit is contained in:
parent
d72d4a711f
commit
53d4f71607
@ -40,16 +40,34 @@ async def login(request: Request, username: str = Form(...), password: str = For
|
||||
"error": msg, "ldap_enabled": ldap_is_on
|
||||
})
|
||||
|
||||
if not row:
|
||||
# Auto-provision LDAP : user inconnu en base mais peut etre dans AD groupe autorise
|
||||
if not row and auth_method == "ldap" and ldap_is_on:
|
||||
result = ldap_authenticate(db, username, password)
|
||||
if not result.get("ok"):
|
||||
log_login_failed(db, request, username)
|
||||
db.commit()
|
||||
return err_template(result.get("msg") or "Authentification LDAP echouee")
|
||||
# Cree l'user en local avec role par defaut
|
||||
default_role = result.get("default_role", "operator")
|
||||
db.execute(text("""
|
||||
INSERT INTO users (username, email, full_name, role, is_active, auth_type, password_hash)
|
||||
VALUES (:u, :e, :n, :r, true, 'ldap', '')
|
||||
"""), {"u": username, "e": result.get("email", ""),
|
||||
"n": result.get("name", username), "r": default_role})
|
||||
db.commit()
|
||||
row = db.execute(text("SELECT id, username, password_hash, role, is_active, auth_type FROM users WHERE LOWER(username)=LOWER(:u)"),
|
||||
{"u": username}).fetchone()
|
||||
ok = True
|
||||
elif not row:
|
||||
log_login_failed(db, request, username)
|
||||
db.commit()
|
||||
return err_template("Utilisateur inconnu")
|
||||
if not row.is_active:
|
||||
elif not row.is_active:
|
||||
log_login_failed(db, request, username)
|
||||
db.commit()
|
||||
return err_template("Compte desactive")
|
||||
|
||||
# Choix de la methode d'auth
|
||||
else:
|
||||
# Choix de la methode d'auth pour user existant
|
||||
use_ldap = (auth_method == "ldap") or (row.auth_type == "ldap")
|
||||
if use_ldap and not ldap_is_on:
|
||||
return err_template("LDAP non active")
|
||||
|
||||
@ -85,6 +85,8 @@ SECTIONS = {
|
||||
("ldap_email_attr", "Attribut email (ex: mail)", False),
|
||||
("ldap_name_attr", "Attribut nom affiché (ex: displayName)", False),
|
||||
("ldap_tls", "TLS (true/false)", False),
|
||||
("ldap_required_group", "Groupe AD autorise (DN complet, vide = tous)", False),
|
||||
("ldap_default_role", "Role par defaut auto-provision (admin/operator/viewer)", False),
|
||||
],
|
||||
"itop_contacts": [
|
||||
("itop_contact_teams", "Teams iTop à synchroniser (séparées par ,)", False),
|
||||
|
||||
@ -39,6 +39,8 @@ def _get_config(db):
|
||||
"tls": s("ldap_tls", "true").lower() == "true",
|
||||
"email_attr": s("ldap_email_attr", "mail"),
|
||||
"name_attr": s("ldap_name_attr", "displayName"),
|
||||
"required_group": s("ldap_required_group", ""),
|
||||
"default_role": s("ldap_default_role", "operator"),
|
||||
}
|
||||
|
||||
|
||||
@ -72,7 +74,7 @@ def authenticate(db, username, password):
|
||||
user_filter = cfg["user_filter"].replace("{username}", username)
|
||||
try:
|
||||
conn.search(cfg["base_dn"], user_filter,
|
||||
attributes=[cfg["email_attr"], cfg["name_attr"], "distinguishedName"])
|
||||
attributes=[cfg["email_attr"], cfg["name_attr"], "distinguishedName", "memberOf"])
|
||||
except Exception as e:
|
||||
conn.unbind()
|
||||
return {"ok": False, "msg": f"Recherche LDAP échouée: {e}"}
|
||||
@ -85,16 +87,27 @@ def authenticate(db, username, password):
|
||||
user_dn = str(entry.distinguishedName) if hasattr(entry, "distinguishedName") else entry.entry_dn
|
||||
email = str(getattr(entry, cfg["email_attr"], "")) or ""
|
||||
name = str(getattr(entry, cfg["name_attr"], "")) or username
|
||||
groups = list(entry.memberOf.values) if hasattr(entry, "memberOf") and entry.memberOf else []
|
||||
conn.unbind()
|
||||
|
||||
# 3. Bind avec les credentials fournis
|
||||
# 3. Verification appartenance groupe (si configure)
|
||||
required = (cfg.get("required_group") or "").strip()
|
||||
if required:
|
||||
# Match insensible a la casse + espaces normalises
|
||||
norm_required = required.lower().replace(" ", "")
|
||||
is_member = any(norm_required == g.lower().replace(" ", "") for g in groups)
|
||||
if not is_member:
|
||||
return {"ok": False, "msg": f"Acces refuse: utilisateur non membre du groupe requis"}
|
||||
|
||||
# 4. Bind avec les credentials fournis
|
||||
try:
|
||||
user_conn = Connection(server, user=user_dn, password=password, auto_bind=True)
|
||||
user_conn.unbind()
|
||||
except Exception as e:
|
||||
return {"ok": False, "msg": "Mot de passe incorrect"}
|
||||
|
||||
return {"ok": True, "dn": user_dn, "email": email, "name": name}
|
||||
return {"ok": True, "dn": user_dn, "email": email, "name": name,
|
||||
"groups": groups, "default_role": cfg.get("default_role", "operator")}
|
||||
|
||||
|
||||
def test_connection(db):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user