from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timezone import re import time from typing import Any import httpx import psycopg from ..k8s.exec import ExecError, PodExecutor from ..k8s.pods import PodSelectionError from ..settings import settings from ..utils.logging import get_logger from ..utils.passwords import random_password from .keycloak_admin import keycloak_admin logger = get_logger(__name__) def _extract_attr(attrs: Any, key: str) -> str: if not isinstance(attrs, dict): return "" raw = attrs.get(key) if isinstance(raw, list): for item in raw: if isinstance(item, str) and item.strip(): return item.strip() return "" if isinstance(raw, str) and raw.strip(): return raw.strip() return "" def _resolve_mailu_email(username: str, user: dict[str, Any]) -> str: attrs = user.get("attributes") mailu_email = _extract_attr(attrs, "mailu_email") if mailu_email: return mailu_email email = user.get("email") if isinstance(email, str) and email.strip(): email = email.strip() if email.lower().endswith(f"@{settings.mailu_domain.lower()}"): return email return f"{username}@{settings.mailu_domain}" def _parse_mail_export(output: str) -> list[tuple[str, str]]: accounts: list[tuple[str, str]] = [] account_id = "" for line in output.splitlines(): line = line.strip() if not line: continue match = re.match(r"^Account\s+(\d+):", line, flags=re.IGNORECASE) if match: account_id = match.group(1) continue match = re.match(r"^-\s*E-?mail:\s*(\S+)", line, flags=re.IGNORECASE) if match and account_id: accounts.append((account_id, match.group(1))) return accounts @dataclass(frozen=True) class NextcloudMailSyncSummary: processed: int created: int updated: int deleted: int skipped: int failures: int detail: str = "" @dataclass class MailSyncCounters: processed: int = 0 created: int = 0 updated: int = 0 deleted: int = 0 skipped: int = 0 failures: int = 0 last_error: str = "" def summary(self) -> NextcloudMailSyncSummary: return NextcloudMailSyncSummary( processed=self.processed, created=self.created, updated=self.updated, deleted=self.deleted, skipped=self.skipped, failures=self.failures, detail=self.last_error, ) def status(self) -> str: return "ok" if self.failures == 0 else "error" def record_failure(self, detail: str) -> None: self.failures += 1 if detail and not self.last_error: self.last_error = detail class NextcloudService: def __init__(self) -> None: self._executor = PodExecutor( settings.nextcloud_namespace, settings.nextcloud_pod_label, settings.nextcloud_container, ) def _exec_with_fallback( self, primary: list[str], fallback: list[str], env: dict[str, str] | None = None, check: bool = True, ) -> ExecResult: try: result = self._executor.exec( primary, env=env, timeout_sec=settings.nextcloud_exec_timeout_sec, check=check, ) except ExecError as exc: if "runuser: may not be used by non-root users" not in str(exc): raise return self._executor.exec( fallback, env=env, timeout_sec=settings.nextcloud_exec_timeout_sec, check=check, ) if not result.ok and "runuser: may not be used by non-root users" in result.stderr: return self._executor.exec( fallback, env=env, timeout_sec=settings.nextcloud_exec_timeout_sec, check=check, ) return result def _occ_exec( self, args: list[str], env: dict[str, str] | None = None, check: bool = True, ) -> ExecResult: command = ["runuser", "-u", "www-data", "--", "php", "/var/www/html/occ", *args] fallback = ["php", "/var/www/html/occ", *args] return self._exec_with_fallback(command, fallback, env=env, check=check) def _occ(self, args: list[str]) -> str: result = self._occ_exec(args, check=True) return result.stdout def _display_name(self, user: dict[str, Any]) -> str: first = user.get("firstName") if isinstance(user.get("firstName"), str) else "" last = user.get("lastName") if isinstance(user.get("lastName"), str) else "" first = first.strip() last = last.strip() if first and last: return f"{first} {last}" return last or first def _ensure_nextcloud_user( self, username: str, mailu_email: str, display_name: str, ) -> None: result = self._occ_exec(["user:info", username], check=False) if result.ok: return detail = f"{result.stdout}\n{result.stderr}".strip().lower() missing_markers = ("not found", "not exist", "does not exist", "unknown user") if detail and not any(marker in detail for marker in missing_markers): raise RuntimeError(f"nextcloud user lookup failed: {detail[:200]}") password = random_password(24) env = {"OC_PASS": password} args = ["user:add", "--password-from-env"] name = display_name or username if name: args += ["--display-name", name] if mailu_email: args += ["--email", mailu_email] args.append(username) self._occ_exec(args, env=env, check=True) def run_cron(self) -> dict[str, Any]: if not settings.nextcloud_namespace: raise RuntimeError("nextcloud cron not configured") try: self._exec_with_fallback( ["runuser", "-u", "www-data", "--", "php", "-f", "/var/www/html/cron.php"], ["php", "-f", "/var/www/html/cron.php"], ) except (ExecError, PodSelectionError, TimeoutError) as exc: return {"status": "error", "detail": str(exc)} return {"status": "ok"} def _list_mail_accounts(self, username: str) -> list[tuple[str, str]]: output = self._occ(["mail:account:export", username]) return _parse_mail_export(output) def _set_editor_mode_richtext(self, account_ids: list[str]) -> None: safe_ids = [item for item in account_ids if item.isdigit()] if not safe_ids: return if not settings.nextcloud_db_host or not settings.nextcloud_db_password: logger.info( "nextcloud editor_mode skipped", extra={"event": "nextcloud_mail_editor_mode", "status": "skip", "reason": "missing db config"}, ) return ids_csv = ",".join(safe_ids) query = ( "UPDATE oc_mail_accounts SET editor_mode='richtext' " f"WHERE id IN ({ids_csv}) AND editor_mode <> 'richtext';" ) try: with psycopg.connect( host=settings.nextcloud_db_host, port=settings.nextcloud_db_port, dbname=settings.nextcloud_db_name, user=settings.nextcloud_db_user, password=settings.nextcloud_db_password, ) as conn: with conn.cursor() as cur: cur.execute(query) except Exception as exc: logger.info( "nextcloud editor_mode update failed", extra={"event": "nextcloud_mail_editor_mode", "status": "error", "detail": str(exc)}, ) def _set_user_mail_meta(self, user_id: str, primary_email: str, account_count: int) -> None: synced_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) attrs = { "nextcloud_mail_primary_email": [primary_email], "nextcloud_mail_account_count": [str(account_count)], "nextcloud_mail_synced_at": [synced_at], } try: keycloak_admin.update_user_safe(user_id, {"attributes": attrs}) except Exception: return def _collect_users(self, username: str | None) -> list[dict[str, Any]]: if username is not None: user = keycloak_admin.find_user(username) return [user] if user else [] return list(keycloak_admin.iter_users(page_size=200, brief=False)) def _normalize_user(self, user: dict[str, Any]) -> tuple[str, str, dict[str, Any]] | None: username_val = user.get("username") if isinstance(user.get("username"), str) else "" username_val = username_val.strip() if not username_val: return None if user.get("enabled") is False: return None if user.get("serviceAccountClientId") or username_val.startswith("service-account-"): return None user_id = user.get("id") if isinstance(user.get("id"), str) else "" full_user = user if user_id: try: full_user = keycloak_admin.get_user(user_id) except Exception: full_user = user return username_val, user_id, full_user def _list_mail_accounts_safe( self, username: str, counters: MailSyncCounters, ) -> list[tuple[str, str]] | None: try: return self._list_mail_accounts(username) except Exception as exc: detail = f"mail export failed: {exc}" counters.record_failure(detail) logger.info( "nextcloud mail export failed", extra={"event": "nextcloud_mail_export", "status": "error", "detail": detail}, ) return None def _select_primary_account( self, mailu_accounts: list[tuple[str, str]], mailu_email: str, ) -> tuple[str, str]: primary_id = "" primary_email = "" for account_id, account_email in mailu_accounts: if not primary_id: primary_id = account_id primary_email = account_email if account_email.lower() == mailu_email.lower(): primary_id = account_id primary_email = account_email break return primary_id, primary_email def _update_mail_account( self, username: str, primary_id: str, mailu_email: str, app_pw: str, ) -> str | None: try: self._occ( [ "mail:account:update", "-q", primary_id, "--name", username, "--email", mailu_email, "--imap-host", settings.mailu_host, "--imap-port", "993", "--imap-ssl-mode", "ssl", "--imap-user", mailu_email, "--imap-password", app_pw, "--smtp-host", settings.mailu_host, "--smtp-port", "587", "--smtp-ssl-mode", "tls", "--smtp-user", mailu_email, "--smtp-password", app_pw, "--auth-method", "password", ] ) return None except Exception as exc: return str(exc) def _create_mail_account(self, username: str, mailu_email: str, app_pw: str) -> str | None: try: self._occ( [ "mail:account:create", "-q", username, username, mailu_email, settings.mailu_host, "993", "ssl", mailu_email, app_pw, settings.mailu_host, "587", "tls", mailu_email, app_pw, "password", ] ) return None except Exception as exc: return str(exc) def _delete_extra_accounts( self, mailu_accounts: list[tuple[str, str]], primary_id: str, counters: MailSyncCounters, ) -> int: deleted = 0 for account_id, _account_email in mailu_accounts: if account_id == primary_id: continue try: self._occ(["mail:account:delete", "-q", account_id]) deleted += 1 except Exception as exc: counters.record_failure(f"mail account delete failed: {exc}") return deleted def _mailu_accounts(self, accounts: list[tuple[str, str]]) -> list[tuple[str, str]]: return [ (account_id, email) for account_id, email in accounts if email.lower().endswith(f"@{settings.mailu_domain.lower()}") ] def _summarize_mail_accounts( self, accounts: list[tuple[str, str]], mailu_email: str, ) -> tuple[int, str, list[str]]: mailu_accounts = self._mailu_accounts(accounts) account_count = len(mailu_accounts) primary_email = "" editor_mode_ids: list[str] = [] for account_id, account_email in mailu_accounts: editor_mode_ids.append(account_id) if account_email.lower() == mailu_email.lower(): primary_email = account_email break if not primary_email: primary_email = account_email return account_count, primary_email, editor_mode_ids def _mail_sync_context( self, user: dict[str, Any], counters: MailSyncCounters, ) -> tuple[str, str, str, str, dict[str, Any]] | None: normalized = self._normalize_user(user) if not normalized: counters.skipped += 1 return None username, user_id, full_user = normalized attrs = full_user.get("attributes") if isinstance(full_user.get("attributes"), dict) else {} mailu_email = _resolve_mailu_email(username, full_user) app_pw = _extract_attr(attrs, "mailu_app_password") if not mailu_email or not app_pw: counters.skipped += 1 return None if mailu_email and not _extract_attr(attrs, "mailu_email"): try: keycloak_admin.set_user_attribute(username, "mailu_email", mailu_email) except Exception: pass return username, user_id, mailu_email, app_pw, full_user def _sync_mail_accounts( self, username: str, mailu_email: str, app_pw: str, accounts: list[tuple[str, str]], counters: MailSyncCounters, ) -> bool: mailu_accounts = self._mailu_accounts(accounts) if mailu_accounts: primary_id, _primary_email = self._select_primary_account(mailu_accounts, mailu_email) error = self._update_mail_account(username, primary_id, mailu_email, app_pw) if error: counters.record_failure(f"mail account update failed: {error}") return False counters.updated += 1 counters.deleted += self._delete_extra_accounts(mailu_accounts, primary_id, counters) else: error = self._create_mail_account(username, mailu_email, app_pw) if error: counters.record_failure(f"mail account create failed: {error}") return False counters.created += 1 return True def _apply_mail_metadata( self, user_id: str, mailu_email: str, accounts: list[tuple[str, str]], ) -> None: account_count, primary_email, editor_mode_ids = self._summarize_mail_accounts(accounts, mailu_email) self._set_editor_mode_richtext(editor_mode_ids) if user_id: self._set_user_mail_meta(user_id, primary_email, account_count) def _sync_user_mail(self, user: dict[str, Any], counters: MailSyncCounters) -> None: context = self._mail_sync_context(user, counters) if not context: return username, user_id, mailu_email, app_pw, full_user = context try: display_name = self._display_name(full_user) self._ensure_nextcloud_user(username, mailu_email, display_name) except Exception as exc: counters.record_failure(f"nextcloud user ensure failed: {exc}") return accounts = self._list_mail_accounts_safe(username, counters) if accounts is None: return counters.processed += 1 if not self._sync_mail_accounts(username, mailu_email, app_pw, accounts, counters): return accounts_after = self._list_mail_accounts_safe(username, counters) if accounts_after is None: return self._apply_mail_metadata(user_id, mailu_email, accounts_after) def sync_mail(self, username: str | None = None, wait: bool = True) -> dict[str, Any]: if not settings.nextcloud_namespace: raise RuntimeError("nextcloud mail sync not configured") cleaned_username = None if username is not None: cleaned_username = username.strip() if not cleaned_username: raise RuntimeError("missing username") if not keycloak_admin.ready(): return {"status": "error", "detail": "keycloak admin not configured"} users = self._collect_users(cleaned_username) if cleaned_username is not None and not users: return {"status": "ok", "detail": "no matching user"} counters = MailSyncCounters() for user in users: self._sync_user_mail(user, counters) summary = counters.summary() summary_payload = { "processed": summary.processed, "created": summary.created, "updated": summary.updated, "deleted": summary.deleted, "skipped": summary.skipped, "failures": summary.failures, "detail": summary.detail, } logger.info( "nextcloud mail sync finished", extra={ "event": "nextcloud_mail_sync", "status": counters.status(), "processed_count": counters.processed, "created_count": counters.created, "updated_count": counters.updated, "deleted_count": counters.deleted, "skipped_count": counters.skipped, "failures_count": counters.failures, "detail": summary.detail, }, ) return {"status": counters.status(), "summary": summary_payload, "detail": summary.detail} def _run_shell(self, script: str, check: bool = True) -> None: self._executor.exec( script, timeout_sec=settings.nextcloud_exec_timeout_sec, check=check, ) def _external_api(self, method: str, path: str, data: dict[str, Any] | None = None) -> dict[str, Any]: if not settings.nextcloud_url: raise RuntimeError("nextcloud url not configured") if not settings.nextcloud_admin_user or not settings.nextcloud_admin_password: raise RuntimeError("nextcloud admin credentials missing") url = f"{settings.nextcloud_url}/ocs/v2.php/apps/external/api/v1{path}" headers = {"OCS-APIRequest": "true"} with httpx.Client(timeout=settings.nextcloud_exec_timeout_sec) as client: resp = client.request( method, url, headers=headers, auth=(settings.nextcloud_admin_user, settings.nextcloud_admin_password), data=data, ) resp.raise_for_status() try: return resp.json() except Exception: return {} def run_maintenance(self) -> dict[str, Any]: if not settings.nextcloud_namespace: raise RuntimeError("nextcloud maintenance not configured") try: self._run_shell( """ set -euo pipefail if [ ! -d /var/www/html/lib ] && [ -d /usr/src/nextcloud/lib ]; then if command -v rsync >/dev/null 2>&1; then rsync -a --delete --exclude config --exclude data /usr/src/nextcloud/ /var/www/html/ else cp -a /usr/src/nextcloud/. /var/www/html/ fi fi mkdir -p /var/www/html/data chown 33:33 /var/www/html || true chmod 775 /var/www/html || true chown -R 33:33 /var/www/html/apps /var/www/html/custom_apps /var/www/html/data /var/www/html/config 2>/dev/null || true """, check=False, ) self._occ(["config:app:set", "theming", "name", "--value", "Atlas Cloud"]) self._occ(["config:app:set", "theming", "slogan", "--value", "Unified access to Atlas services"]) theming_url = settings.nextcloud_url or "https://cloud.bstein.dev" self._occ(["config:app:set", "theming", "url", "--value", theming_url]) self._occ(["config:app:set", "theming", "color", "--value", "#0f172a"]) self._occ(["config:app:set", "theming", "disable-user-theming", "--value", "yes"]) self._executor.exec( ["runuser", "-u", "www-data", "--", "php", "/var/www/html/occ", "app:install", "customcss"], timeout_sec=settings.nextcloud_exec_timeout_sec, check=False, ) self._executor.exec( ["runuser", "-u", "www-data", "--", "php", "/var/www/html/occ", "app:enable", "customcss"], timeout_sec=settings.nextcloud_exec_timeout_sec, check=False, ) mail_css = ( ".mail-message-body, .mail-message-body pre, .mail-message-body code, .mail-message-body table {\n" " font-family: \"Inter\", \"Source Sans 3\", \"Helvetica Neue\", Arial, sans-serif;\n" " font-size: 14px;\n" " line-height: 1.6;\n" " color: var(--color-main-text);\n" "}\n" ".mail-message-body pre {\n" " background: rgba(15, 23, 42, 0.06);\n" " padding: 12px;\n" " border-radius: 8px;\n" "}\n" ".mail-message-body blockquote {\n" " border-left: 3px solid var(--color-border);\n" " padding-left: 12px;\n" " margin: 8px 0;\n" " color: var(--color-text-lighter);\n" "}\n" ".mail-message-body img {\n" " max-width: 100%;\n" " border-radius: 6px;\n" "}\n" ) self._occ(["config:app:set", "customcss", "css", "--value", mail_css]) self._occ(["config:app:set", "files", "default_quota", "--value", "250 GB"]) payload = self._external_api("GET", "?format=json") links = payload.get("ocs", {}).get("data", []) if isinstance(payload, dict) else [] for link in links: link_id = link.get("id") if isinstance(link, dict) else None if link_id is not None: self._external_api("DELETE", f"/sites/{link_id}?format=json") sites = [ ("Vaultwarden", "https://vault.bstein.dev"), ("Jellyfin", "https://stream.bstein.dev"), ("Gitea", "https://scm.bstein.dev"), ("Jenkins", "https://ci.bstein.dev"), ("Harbor", "https://registry.bstein.dev"), ("Vault", "https://secret.bstein.dev"), ("Jitsi", "https://meet.bstein.dev"), ("Grafana", "https://metrics.bstein.dev"), ("Chat LLM", "https://chat.ai.bstein.dev"), ("Vision", "https://draw.ai.bstein.dev"), ("STT/TTS", "https://talk.ai.bstein.dev"), ] for name, url in sites: self._external_api( "POST", "/sites?format=json", data={ "name": name, "url": url, "lang": "", "type": "link", "device": "", "icon": "", "groups[]": "", "redirect": "1", }, ) except (ExecError, PodSelectionError, TimeoutError) as exc: return {"status": "error", "detail": str(exc)} except Exception as exc: # noqa: BLE001 return {"status": "error", "detail": str(exc)} return {"status": "ok", "detail": "maintenance complete"} nextcloud = NextcloudService()