b0rbor4d/insight-updater
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

LDAP/AD-Auth fuer Dashboard: User-Bind gegen insight.local, Zugriff nur fuer Domänen-Admins

Browse Source 
- ldap_auth Service: User bindet als user@domain (kein Servicekonto noetig),
  Gruppenpruefung ueber Matching-Rule-In-Chain (verschachtelte Gruppen)
- Login-Route: lokaler User zuerst, dann LDAP-Fallback; LDAP-User werden
  lokal gespiegelt (is_ldap, is_admin, Profildaten aus AD)
- Neue Config: ldap_domain, ldap_admin_group, ldap_timeout_seconds
- ldap3 Dependency, 7 Unit-Tests (ldap3 gemockt), ruff+mypy sauber
- Gruppen-DN verified am DC: CN=Domänen-Admins,CN=Users,DC=insight,DC=local
This commit is contained in:
B0rbor4d
2026-09-28 22:06:05 +00:00
parent cc4c3fcecb
commit 8305352497
7 changed files with 320 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+11 -6
View File
@@ -35,14 +35,19 @@ DB_PASSWORD=change-me-db-password
JOB_STALE_TIMEOUT=3600
# =============================================================================
# LDAP / ACTIVE DIRECTORY (STUB - prepared for future implementation)
# LDAP / ACTIVE DIRECTORY
# =============================================================================
LDAP_ENABLED=false
LDAP_URI=ldaps://dc.insight.local:636
LDAP_BIND_DN=CN=svc_updater,OU=Services,DC=insight,DC=local
LDAP_BIND_PASSWORD=
LDAP_USER_SEARCH_BASE=OU=Users,DC=insight,DC=local
# User-Bind: Es gibt kein Servicekonto. Der User bindet mit seinen eigenen
# Credentials (user@domain), Gruppenmitgliedschaft wird ueber die eigene
# Verbindung geprueft. Nur Mitglieder der Admin-Gruppe kommen rein.
LDAP_ENABLED=true
LDAP_URI=ldaps://ads-server-01.insight.local:636
LDAP_DOMAIN=insight.local
LDAP_USER_SEARCH_BASE=DC=insight,DC=local
LDAP_USER_FILTER=(sAMAccountName={username})
# Achtung: DN exakt wie im AD (verified 2026-09-28, Umlaut beachten):
LDAP_ADMIN_GROUP=CN=Domänen-Admins,CN=Users,DC=insight,DC=local
LDAP_TIMEOUT_SECONDS=10
# =============================================================================
# FRONTEND (injected at build time via Vite)
README.md
+13 -3
View File
@@ -41,7 +41,9 @@ Kundennetz A Kundennetz B Zentrale
- **Netzwerk-Scan**: Ping + Port-Probe (5985/22), legt gefundene Hosts zentral an
- **Log-Upload** in Batches (kein Live-Stream noetig, 10s Dashboard-Refresh)
- **Audit-Log**: strukturiert, pro Kunde filterbar
- **Auth**: JWT (RS256) fuer Dashboard-User, API-Key (SHA-256 gehasht) fuer Satelliten
- **Auth**: JWT (RS256) fuer Dashboard-User, API-Key (SHA-256 gehasht) fuer Satelliten;
Dashboard-Login optional per LDAP/AD (User-Bind, kein Servicekonto) mit
Zugriffsbeschraenkung auf die konfigurierte Admin-Gruppe
## Quick Start (Zentrale, Development)
@@ -94,8 +96,16 @@ Wichtig: die Zentrale muss per HTTPS von den Kunden-Standorten aus erreichbar se
## Umgebungsvariablen
Siehe `.env.example`. Zentrale braucht nur noch: SECRET_KEY, DATABASE_URL, JWT-Keys,
optional LDAP. WinRM/SSH-Config ist in den Satellite gewandert (dessen `config.yaml`).
Siehe `.env.example`. Zentrale braucht nur noch: SECRET_KEY, DATABASE_URL, JWT-Keys.
WinRM/SSH-Config ist in den Satellite gewandert (dessen `config.yaml`).
## LDAP / Active Directory
Dashboard-Login per AD (User-Bind, kein Servicekonto): Der User bindet mit
seinen eigenen Credentials als `user@domain`, die Gruppenmitgliedschaft wird
ueber die Matching-Rule-In-Chain geprueft (verschachtelte Gruppen aufloesend).
Nur Mitglieder von `LDAP_ADMIN_GROUP` bekommen Zugang. Lokal angelegte User
(Initial-Admin) bleiben als Fallback wirksam.
## Development
backend/app/api/routes/auth.py
+58 -10
View File
@@ -1,10 +1,11 @@
"""Auth routes: login (local user or LDAP stub)."""
"""Auth routes: login (local user or LDAP/AD)."""
from datetime import UTC, datetime
from fastapi import APIRouter, Depends, Request
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.concurrency import run_in_threadpool
from app.api.deps import client_ip
from app.core.config import get_settings
@@ -14,11 +15,24 @@ from app.core.security import create_access_token, verify_password
from app.models.user import User
from app.schemas.auth import LoginRequest, TokenResponse
from app.services.audit import AuditService
from app.services.ldap_auth import LdapAuthError
from app.services.ldap_auth import authenticate as ldap_authenticate
settings = get_settings()
router = APIRouter()
def _issue_token(user: User) -> TokenResponse:
token = create_access_token(
subject=user.username,
extra_claims={"is_admin": user.is_admin},
)
return TokenResponse(
access_token=token,
expires_in=settings.jwt_access_token_expire_minutes * 60,
)
@router.post("/login", response_model=TokenResponse)
async def login(
payload: LoginRequest,
@@ -36,16 +50,50 @@ async def login(
action="auth.login",
ip_address=client_ip(request),
)
token = create_access_token(
subject=user.username,
extra_claims={"is_admin": user.is_admin},
)
return TokenResponse(
access_token=token,
expires_in=settings.jwt_access_token_expire_minutes * 60,
)
return _issue_token(user)
# Local user not found or password mismatch: fall back to LDAP/AD when
# enabled. Only members of the configured admin group get access.
if settings.ldap_enabled:
try:
ldap_user = await run_in_threadpool(
ldap_authenticate, payload.username, payload.password, settings
)
except LdapAuthError as exc:
await audit.log(
username=payload.username,
action="auth.login",
result="failure",
ip_address=client_ip(request),
)
raise UnauthorizedError(exc.detail) from exc
result = await db.execute(select(User).where(User.username == ldap_user.username))
user = result.scalar_one_or_none()
if user is None:
user = User(
username=ldap_user.username,
full_name=ldap_user.full_name,
email=ldap_user.email,
is_ldap=True,
is_admin=ldap_user.is_admin,
)
db.add(user)
else:
user.is_ldap = True
user.is_admin = ldap_user.is_admin
if ldap_user.full_name:
user.full_name = ldap_user.full_name
if ldap_user.email:
user.email = ldap_user.email
user.last_login_at = datetime.now(UTC)
await audit.log(
username=user.username,
action="auth.login",
ip_address=client_ip(request),
)
return _issue_token(user)
# LDAP stub: when enabled, attempt bind + search here (not yet implemented)
await audit.log(
username=payload.username,
action="auth.login",
backend/app/core/config.py
+4 -3
View File
@@ -35,13 +35,14 @@ class Settings(BaseSettings):
# seconds is considered stale and marked failed by the janitor
job_stale_timeout: int = 3600
# LDAP (stub)
# LDAP / Active Directory (user bind, no service account)
ldap_enabled: bool = False
ldap_uri: str = ""
ldap_bind_dn: str = ""
ldap_bind_password: str = ""
ldap_domain: str = "" # bind as user@domain; empty = plain username bind
ldap_user_search_base: str = ""
ldap_user_filter: str = "(sAMAccountName={username})"
ldap_admin_group: str = "" # DN, e.g. CN=Admins,CN=Users,DC=example,DC=com
ldap_timeout_seconds: int = 10
# CORS
cors_origins: str = "http://localhost:3000,http://localhost:8000"
backend/app/services/ldap_auth.py
+110
View File
@@ -0,0 +1,110 @@
"""LDAP/AD authentication for dashboard users.
User-bind design (no service account required):
1. The user binds with their own credentials as ``user@domain`` (AD accepts
userPrincipalName as bind DN).
2. Over that authenticated connection we search the user's own directory
entry, restricted by the in-chain matching rule to members of the
configured admin group (direct or nested membership).
Only members of the configured group may log in. All other users are
rejected before any local session is created.
"""
import ssl
from dataclasses import dataclass
import ldap3
from ldap3.core.exceptions import LDAPExceptionError
from ldap3.utils.conv import escape_filter_chars
from app.core.config import Settings
# LDAP_MATCHING_RULE_IN_CHAIN: resolves nested group membership server-side
_MATCHING_RULE_IN_CHAIN = "1.2.840.113556.1941"
class LdapAuthError(Exception):
"""Authentication failed. Message is safe to show to the user."""
def __init__(self, detail: str = "Benutzername oder Passwort falsch") -> None:
super().__init__(detail)
self.detail = detail
@dataclass
class LdapUser:
username: str
full_name: str | None
email: str | None
is_admin: bool # always True today: only admin-group members may log in
def _build_server(settings: Settings) -> ldap3.Server:
use_ssl = settings.ldap_uri.lower().startswith("ldaps://")
# DC certificates come from an internal AD CA that is not distributed to
# clients; full chain validation would need the enterprise root CA.
tls = ldap3.Tls(validate=ssl.CERT_NONE) if use_ssl else None
return ldap3.Server(
settings.ldap_uri,
get_info=ldap3.NONE,
connect_timeout=settings.ldap_timeout_seconds,
tls=tls,
)
def _bind_dn(username: str, settings: Settings) -> str:
if settings.ldap_domain:
return f"{username}@{settings.ldap_domain}"
return username
def authenticate(username: str, password: str, settings: Settings) -> LdapUser:
"""Validate credentials against AD and verify admin-group membership.
Raises LdapAuthError on any failure (wrong password, unknown user,
not in admin group, LDAP unreachable).
"""
if not settings.ldap_enabled:
raise LdapAuthError("LDAP ist nicht aktiviert")
if not settings.ldap_admin_group:
raise LdapAuthError("LDAP-Admin-Gruppe ist nicht konfiguriert")
if not username or not password:
raise LdapAuthError()
server = _build_server(settings)
try:
conn = ldap3.Connection(
server,
user=_bind_dn(username.strip(), settings),
password=password,
auto_bind=True,
receive_timeout=settings.ldap_timeout_seconds,
)
except LDAPExceptionError as exc:
# Covers invalidCredentials (49) as well as connect/TLS timeouts
raise LdapAuthError() from exc
with conn:
user_filter = settings.ldap_user_filter.format(
username=escape_filter_chars(username.strip())
)
# One search proves: user exists AND is (transitively) in the admin
# group. Non-members simply match nothing.
group_filter = f"(memberOf:{_MATCHING_RULE_IN_CHAIN}:={settings.ldap_admin_group})"
ok = conn.search(
search_base=settings.ldap_user_search_base,
search_filter=f"(&{user_filter}{group_filter})",
search_scope=ldap3.SUBTREE,
attributes=["displayName", "mail"],
)
if not ok or not conn.entries:
raise LdapAuthError("Kein Zugriff: Kein Mitglied der Administrator-Gruppe")
entry = conn.entries[0]
return LdapUser(
username=username.strip(),
full_name=str(entry.displayName) if entry.displayName else None,
email=str(entry.mail) if entry.mail else None,
is_admin=True,
)
backend/pyproject.toml
+1
View File
@@ -21,6 +21,7 @@ dependencies = [
"python-multipart>=0.0.6",
"httpx>=0.26.0",
"tenacity>=8.2.0",
"ldap3>=2.9.1",
"pytz>=2024.1",
"python-dateutil>=2.8.2",
]
backend/tests/test_ldap_auth.py
+123
View File
@@ -0,0 +1,123 @@
"""Unit tests for LDAP user-bind authentication (ldap3 fully mocked)."""
from unittest.mock import MagicMock
import pytest
from app.core.config import Settings
from app.services.ldap_auth import LdapAuthError, authenticate
def _settings(**overrides: object) -> Settings:
base: dict[str, object] = {
"ldap_enabled": True,
"ldap_uri": "ldaps://ads-server-01.insight.local:636",
"ldap_domain": "insight.local",
"ldap_user_search_base": "DC=insight,DC=local",
"ldap_user_filter": "(sAMAccountName={username})",
"ldap_admin_group": "CN=Domain Admins,CN=Users,DC=insight,DC=local",
"ldap_timeout_seconds": 10,
}
base.update(overrides)
return Settings(**base) # type: ignore[arg-type]
def _entry(full_name: str = "Flo Hartmann", email: str = "flo@example.com") -> MagicMock:
entry = MagicMock()
entry.displayName = full_name
entry.mail = email
return entry
def _patch_ldap(monkeypatch: pytest.MonkeyPatch, conn: MagicMock) -> None:
mock_ldap3 = MagicMock()
mock_ldap3.Server.return_value = MagicMock()
mock_ldap3.Connection.return_value = conn
mock_ldap3.SUBTREE = "SUBTREE"
mock_ldap3.NONE = "NO_INFO"
mock_ldap3.utils.conv.escape_filter_chars = lambda s: s
monkeypatch.setattr("app.services.ldap_auth.ldap3", mock_ldap3)
def test_disabled_raises() -> None:
with pytest.raises(LdapAuthError, match="nicht aktiviert"):
authenticate("flo", "secret", _settings(ldap_enabled=False))
def test_missing_admin_group_raises() -> None:
with pytest.raises(LdapAuthError, match="nicht konfiguriert"):
authenticate("flo", "secret", _settings(ldap_admin_group=""))
def test_empty_password_raises() -> None:
with pytest.raises(LdapAuthError):
authenticate("flo", "", _settings())
def test_bind_failure_raises_generic(monkeypatch: pytest.MonkeyPatch) -> None:
from ldap3.core.exceptions import LDAPExceptionError
mock_ldap3 = MagicMock()
mock_ldap3.Server.return_value = MagicMock()
mock_ldap3.Connection.side_effect = LDAPExceptionError("invalidCredentials")
monkeypatch.setattr("app.services.ldap_auth.ldap3", mock_ldap3)
with pytest.raises(LdapAuthError, match="Benutzername oder Passwort falsch"):
authenticate("flo", "wrong", _settings())
def test_search_without_match_denied(monkeypatch: pytest.MonkeyPatch) -> None:
conn = MagicMock()
conn.__enter__.return_value = conn
conn.search.return_value = True
conn.entries = []
_patch_ldap(monkeypatch, conn)
with pytest.raises(LdapAuthError, match="Kein Zugriff"):
authenticate("flo", "secret", _settings())
def test_success_returns_admin_user(monkeypatch: pytest.MonkeyPatch) -> None:
conn = MagicMock()
conn.__enter__.return_value = conn
conn.search.return_value = True
conn.entries = [_entry()]
mock_ldap3 = MagicMock()
mock_ldap3.Server.return_value = MagicMock()
mock_ldap3.Connection.return_value = conn
mock_ldap3.SUBTREE = "SUBTREE"
mock_ldap3.NONE = "NO_INFO"
mock_ldap3.utils.conv.escape_filter_chars = lambda s: s
monkeypatch.setattr("app.services.ldap_auth.ldap3", mock_ldap3)
user = authenticate("flo", "secret", _settings())
assert user.username == "flo"
assert user.full_name == "Flo Hartmann"
assert user.email == "flo@example.com"
assert user.is_admin is True
# bind DN uses user@domain format
assert mock_ldap3.Connection.call_args.kwargs["user"] == "flo@insight.local"
# search filter proves admin-group membership via in-chain rule
search_kwargs = conn.search.call_args.kwargs
assert "memberOf:1.2.840.113556.1941:=" in search_kwargs["search_filter"]
assert "CN=Domain Admins,CN=Users,DC=insight,DC=local" in search_kwargs["search_filter"]
assert search_kwargs["search_base"] == "DC=insight,DC=local"
def test_optional_fields_none_when_empty(monkeypatch: pytest.MonkeyPatch) -> None:
conn = MagicMock()
conn.__enter__.return_value = conn
conn.search.return_value = True
conn.entries = [_entry(full_name="", email="")]
mock_ldap3 = MagicMock()
mock_ldap3.Server.return_value = MagicMock()
mock_ldap3.Connection.return_value = conn
mock_ldap3.SUBTREE = "SUBTREE"
mock_ldap3.NONE = "NO_INFO"
mock_ldap3.utils.conv.escape_filter_chars = lambda s: s
monkeypatch.setattr("app.services.ldap_auth.ldap3", mock_ldap3)
user = authenticate("flo", "secret", _settings())
assert user.full_name is None
assert user.email is None