"""Vikunja service account provisioner (T030). Creates a Vikunja account via the public registration API, then activates the user by setting status=0 through direct DB access (docker exec into the Vikunja DB container). This is needed because Vikunja marks new local users as 'email confirmation required' when the mailer is enabled. """ from __future__ import annotations import json import logging import re import secrets import subprocess from typing import TYPE_CHECKING import httpx from config import settings from provisioners.base import ProvisionResult, ServiceProvisioner if TYPE_CHECKING: from user_store import ServiceCredential, User, UserStore logger = logging.getLogger(__name__) _VIKUNJA_DB_CONTAINER = "vikunja-db" _MAX_USERNAME_RETRIES = 3 def _slugify(name: str) -> str: """Turn a display name into a safe username fragment.""" slug = re.sub(r"[^a-zA-Z0-9]", "", name).lower() return slug[:12] if slug else "user" def _generate_username(display_name: str) -> str: suffix = secrets.token_hex(3) return f"{_slugify(display_name)}_{suffix}" def _generate_password() -> str: return secrets.token_urlsafe(16) class VikunjaProvisioner(ServiceProvisioner): @property def service_name(self) -> str: return "vikunja" @property def capabilities(self) -> list[str]: return ["tasks"] @property def requires_consent(self) -> bool: return True async def provision(self, user: User, store: UserStore) -> ProvisionResult: base_url = settings.VIKUNJA_API_URL.rstrip("/") vikunja_user_id: str | None = None for attempt in range(_MAX_USERNAME_RETRIES): username = _generate_username(user.display_name) password = _generate_password() email = f"{username}@code.bytesizeprotip.com" try: # Step 1: Register async with httpx.AsyncClient(timeout=15) as client: reg_resp = await client.post( f"{base_url}/register", json={"username": username, "password": password, "email": email}, ) if reg_resp.status_code == 400 and "already exists" in reg_resp.text.lower(): logger.warning( "Username %s taken, retrying (%d/%d)", username, attempt + 1, _MAX_USERNAME_RETRIES ) continue if reg_resp.status_code not in (200, 201): return ProvisionResult( success=False, error=f"Registration failed ({reg_resp.status_code}): {reg_resp.text[:200]}", ) reg_data = reg_resp.json() vikunja_user_id = str(reg_data.get("id", "")) # Step 2: Admin-side activation — set user status=0 via DB if vikunja_user_id: try: _activate_vikunja_user(vikunja_user_id) except Exception: logger.exception("Failed to activate Vikunja user %s via DB", vikunja_user_id) # Continue anyway — user may still be able to log in # Step 3: Login to get JWT login_resp = await client.post( f"{base_url}/login", json={"username": username, "password": password}, ) if login_resp.status_code != 200: store.log_provisioning( user.id, "vikunja", "provision_failed", json.dumps({"step": "login", "status": login_resp.status_code}), ) return ProvisionResult( success=False, service_user_id=vikunja_user_id, error=f"Login failed after registration ({login_resp.status_code})", ) jwt_token = login_resp.json().get("token", "") # Step 4: Create long-lived API token from datetime import datetime, timedelta, timezone expires = datetime.now(timezone.utc) + timedelta(days=365) token_resp = await client.put( f"{base_url}/tokens", json={ "title": "CodeAnywhere", "expires_at": expires.strftime("%Y-%m-%dT%H:%M:%S+00:00"), "right": 2, # read+write }, headers={"Authorization": f"Bearer {jwt_token}"}, ) if token_resp.status_code not in (200, 201): store.log_provisioning( user.id, "vikunja", "provision_failed", json.dumps({"step": "create_token", "status": token_resp.status_code}), ) return ProvisionResult( success=False, service_user_id=vikunja_user_id, error=f"Token creation failed ({token_resp.status_code})", ) api_token = token_resp.json().get("token", "") # Step 5: Store credential store.store_credential( user.id, "vikunja", api_token, service_user_id=vikunja_user_id, service_username=username, expires_at=expires.isoformat(), ) store.log_provisioning( user.id, "vikunja", "provisioned", json.dumps({"username": username, "vikunja_user_id": vikunja_user_id}), ) logger.info("Provisioned Vikunja account %s for user %s", username, user.id) return ProvisionResult( success=True, service_username=username, service_user_id=vikunja_user_id, service_url=settings.VIKUNJA_API_URL, ) except httpx.HTTPError as exc: store.log_provisioning( user.id, "vikunja", "provision_failed", json.dumps({"error": str(exc)}), ) return ProvisionResult(success=False, error=f"HTTP error: {exc}") # All retries exhausted store.log_provisioning( user.id, "vikunja", "provision_failed", json.dumps({"error": "username collision after max retries"}), ) return ProvisionResult(success=False, error="Could not find available username after retries") async def health_check(self, credential: ServiceCredential) -> bool: """Verify the stored API token still works.""" from user_store import decrypt base_url = settings.VIKUNJA_API_URL.rstrip("/") try: token = decrypt(credential.encrypted_token) async with httpx.AsyncClient(timeout=10) as client: resp = await client.get( f"{base_url}/tasks", headers={"Authorization": f"Bearer {token}"}, params={"per_page": 1}, ) return resp.status_code == 200 except Exception: logger.exception("Vikunja health check failed") return False def _activate_vikunja_user(vikunja_user_id: str) -> None: """Set Vikunja user status to 0 (active) via docker exec into the DB.""" # Sanitise: vikunja_user_id must be numeric if not vikunja_user_id.isdigit(): raise ValueError(f"Invalid Vikunja user ID: {vikunja_user_id}") cmd = [ "docker", "exec", _VIKUNJA_DB_CONTAINER, "psql", "-U", "vikunja", "-d", "vikunja", "-c", f"UPDATE users SET status = 0 WHERE id = {vikunja_user_id};", ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if result.returncode != 0: logger.error("DB activation failed: %s", result.stderr) raise RuntimeError(f"DB activation failed: {result.stderr}") logger.info("Activated Vikunja user %s via DB", vikunja_user_id)