bstein/titan-iac
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

vaultwarden: make cred sync idempotent

Browse Source
This commit is contained in:
Brad Stein 2026-01-03 18:18:31 -03:00
parent 12348258fa
commit 51a733096f
1 changed files with 13 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
scripts/vaultwarden_cred_sync.py
View File

@ -27,7 +27,9 @@ def _iter_keycloak_users(page_size: int = 200) -> Iterable[dict[str, Any]]:
first = 0 first = 0
while True: while True:
headers = client.headers() headers = client.headers()
params = {"first": str(first), "max": str(page_size)} # We need attributes for idempotency (vaultwarden_status/vaultwarden_email). Keycloak defaults to a
# brief representation which may omit these.
params = {"first": str(first), "max": str(page_size), "briefRepresentation": "false"}
with httpx.Client(timeout=settings.HTTP_CHECK_TIMEOUT_SEC) as http: with httpx.Client(timeout=settings.HTTP_CHECK_TIMEOUT_SEC) as http:
resp = http.get(url, params=params, headers=headers) resp = http.get(url, params=params, headers=headers)
resp.raise_for_status() resp.raise_for_status()
@ -79,7 +81,9 @@ def _vaultwarden_email_for_user(user: dict[str, Any]) -> str:
if email and email.lower().endswith(f"@{settings.MAILU_DOMAIN.lower()}"): if email and email.lower().endswith(f"@{settings.MAILU_DOMAIN.lower()}"):
return email return email
return f"{username}@{settings.MAILU_DOMAIN}" # Don't guess an internal mailbox address until Mailu sync has run and stored mailu_email.
# This avoids spamming Vaultwarden invites that can never be delivered (unknown recipient).
return ""
def _set_user_attribute_if_missing(username: str, user: dict[str, Any], key: str, value: str) -> None: def _set_user_attribute_if_missing(username: str, user: dict[str, Any], key: str, value: str) -> None:
@ -121,6 +125,7 @@ def main() -> int:
skipped += 1 skipped += 1
continue continue
current_status = _extract_attr(user.get("attributes"), VAULTWARDEN_STATUS_ATTR)
email = _vaultwarden_email_for_user(user) email = _vaultwarden_email_for_user(user)
if not email: if not email:
print(f"skip {username}: missing email", file=sys.stderr) print(f"skip {username}: missing email", file=sys.stderr)
@ -132,6 +137,12 @@ def main() -> int:
except Exception: except Exception:
pass pass
# If we've already successfully invited or confirmed presence, do not re-invite on every cron run.
# Vaultwarden returns 409 for "already exists", which is idempotent but noisy and can trigger rate limits.
if current_status in {"invited", "already_present"}:
skipped += 1
continue
processed += 1 processed += 1
result = invite_user(email) result = invite_user(email)
if result.ok: if result.ok: