ryan
38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
|
|
from .config import Config
|
|
|
|
|
|
@dataclass
|
|
class AuthentikSyncResult:
|
|
ok: bool
|
|
message: str
|
|
|
|
|
|
def trigger_ldap_sync(config: Config) -> AuthentikSyncResult:
|
|
if not config.authentik_sync_enabled:
|
|
return AuthentikSyncResult(True, "authentik sync disabled")
|
|
if not config.authentik_ldap_source_slugs:
|
|
return AuthentikSyncResult(False, "authentik sync enabled but AUTHENTIK_LDAP_SOURCE_SLUGS is empty")
|
|
|
|
cmd = [
|
|
"docker",
|
|
"exec",
|
|
"-d",
|
|
config.authentik_sync_container,
|
|
"ak",
|
|
"ldap_sync",
|
|
*config.authentik_ldap_source_slugs,
|
|
]
|
|
try:
|
|
subprocess.run(cmd, check=True, text=True, capture_output=True, timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
return AuthentikSyncResult(False, "authentik LDAP sync trigger timed out")
|
|
except subprocess.CalledProcessError as exc:
|
|
detail = exc.stderr.strip() or exc.stdout.strip() or str(exc)
|
|
return AuthentikSyncResult(False, f"authentik LDAP sync trigger failed: {detail}")
|
|
return AuthentikSyncResult(True, f"triggered authentik LDAP sync for {', '.join(config.authentik_ldap_source_slugs)}")
|