betterbot/provisioners/vikunja.py

"""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)