fn_registry/python/functions/infra/dav_make_addressbook.py

"""Crea una nueva coleccion de contactos CardDAV bajo un contacts-home.

Funcion impura: hace una peticion HTTP MKCOL extendido (RFC 5689) para crear una
"libreta/agenda de contactos" nueva bajo el contacts-home de un principal. El
cuerpo XML del MKCOL declara el resourcetype como addressbook
(`{urn:ietf:params:xml:ns:carddav}addressbook`) y fija de paso el nombre visible
(DAV:displayname) y la descripcion (CardDAV addressbook-description).

El slug (segmento de path de la coleccion) se sanea a `[a-z0-9_-]` (minusculas,
espacios -> '-'); si queda vacio se devuelve un error de validacion. La coleccion
se crea en `<contacts_home><slug>/`.

Idempotente: un 201 (Created) es exito; un 405 (Method Not Allowed) o un 301 (la
coleccion ya existe en ese path) se devuelven como {status:'ok', existed:True}.
El display_name y la description se escapan para XML. Construye
`Authorization: Basic base64(user:pass)` a mano con stdlib. Maneja errores sin
lanzar (salvo validacion de args). Solo usa stdlib (urllib, base64, re, ssl,
xml.sax.saxutils). Probado contra Xandikos.
"""

import base64
import re
import ssl
import urllib.error
import urllib.request
from xml.sax.saxutils import escape as _xml_escape

_UNSAFE_SLUG_RE = re.compile(r"[^a-z0-9_-]")


def _basic_auth_header(username: str, password: str) -> str:
    raw = ("%s:%s" % (username, password)).encode("utf-8")
    return "Basic " + base64.b64encode(raw).decode("ascii")


def _sanitize_slug(slug: str) -> str:
    """Sanea un slug a `[a-z0-9_-]`.

    Pasa a minusculas, convierte espacios (y runs de espacios) en un guion, y
    elimina cualquier otro caracter no permitido. Colapsa guiones repetidos y
    recorta guiones de los extremos. Puede devolver "" si no queda nada usable;
    el caller trata "" como error de validacion.
    """
    s = slug.strip().lower()
    s = re.sub(r"\s+", "-", s)
    s = _UNSAFE_SLUG_RE.sub("", s)
    s = re.sub(r"-{2,}", "-", s).strip("-")
    return s


def _build_mkcol_xml(display_name: str, description: str = "") -> str:
    """Cuerpo XML del MKCOL extendido (RFC 5689) para crear un addressbook.

    Declara el resourcetype como `D:collection` + `C:addressbook` (CardDAV) y
    setea el displayname; si hay descripcion, anade `C:addressbook-description`.
    Ambos valores se escapan para XML.
    """
    name = _xml_escape(display_name)
    props = [
        "<D:resourcetype><D:collection/><C:addressbook/></D:resourcetype>",
        "<D:displayname>%s</D:displayname>" % name,
    ]
    if description:
        props.append(
            "<C:addressbook-description>%s</C:addressbook-description>"
            % _xml_escape(description)
        )
    return (
        '<?xml version="1.0" encoding="utf-8" ?>'
        '<D:mkcol xmlns:D="DAV:" '
        'xmlns:C="urn:ietf:params:xml:ns:carddav">'
        "<D:set><D:prop>%s</D:prop></D:set>"
        "</D:mkcol>"
    ) % "".join(props)


def _join_url(base_url: str, contacts_home: str, slug: str) -> str:
    return base_url.rstrip("/") + "/" + contacts_home.strip("/") + "/" + slug + "/"


def dav_make_addressbook(
    base_url: str,
    username: str,
    password: str,
    contacts_home: str,
    slug: str,
    display_name: str = "",
    description: str = "",
    *,
    timeout_s: float = 20.0,
    verify_tls: bool = True,
) -> dict:
    """Crea una nueva coleccion de contactos CardDAV (MKCOL extendido RFC 5689).

    Crea la coleccion en `<contacts_home><slug>/` via MKCOL extendido, declarando
    el resourcetype como addressbook y fijando el displayname (y la descripcion si
    se pasa) en el propio cuerpo. Idempotente: si la coleccion ya existe (405/301)
    devuelve {status:'ok', existed:True}.

    Args:
        base_url: URL base del servidor DAV (sin barra final), p.ej.
            'https://dav-x.organic-machine.com'.
        username: usuario para HTTP Basic auth.
        password: contrasena para HTTP Basic auth. Resolver desde pass.
        contacts_home: ruta del contacts-home del principal (con barra final),
            p.ej. '/enmanuel/contacts/'. La coleccion cuelga de el.
        slug: segmento de path de la coleccion (p.ej. 'trabajo'); se sanea a
            [a-z0-9_-]. Si queda vacio tras sanear, error de validacion.
        display_name: nombre visible (DAV:displayname). Si vacio, usa el slug.
        description: descripcion (CardDAV addressbook-description). Opcional.
        timeout_s: timeout de cada peticion en segundos. Default 20.0.
        verify_tls: si True (default) verifica el certificado TLS.

    Returns:
        dict. En exito: {status:'ok', http_status:int, href:str} (y existed:True
        si ya existia). En error (sin lanzar): {status:'error', http_status:
        int|None, href:str, error:str}.
    """
    clean = _sanitize_slug(slug)
    href = (contacts_home.rstrip("/") + "/" + clean + "/") if clean else ""
    if not clean:
        return {
            "status": "error",
            "http_status": None,
            "href": href,
            "error": "slug invalido: queda vacio tras sanear a [a-z0-9_-]",
        }

    name = display_name if display_name else clean
    url = _join_url(base_url, contacts_home, clean)
    context = None if verify_tls else ssl._create_unverified_context()
    headers = {
        "Authorization": _basic_auth_header(username, password),
        "Content-Type": "application/xml; charset=utf-8",
    }

    # MKCOL extendido (RFC 5689) — crea la coleccion + resourcetype addressbook +
    # displayname + (opcional) descripcion, todo en un solo request.
    mk_body = _build_mkcol_xml(name, description).encode("utf-8")
    req = urllib.request.Request(url, data=mk_body, method="MKCOL", headers=headers)
    existed = False
    http_status = None
    try:
        with urllib.request.urlopen(req, timeout=timeout_s, context=context) as resp:
            http_status = resp.status
    except urllib.error.HTTPError as e:
        # 405/301: la coleccion ya existe en ese path -> idempotente.
        if e.code in (301, 405):
            existed = True
            http_status = e.code
        else:
            return {
                "status": "error",
                "http_status": e.code,
                "href": href,
                "error": "http %s" % e.code,
            }
    except urllib.error.URLError as e:
        return {
            "status": "error",
            "http_status": None,
            "href": href,
            "error": str(e.reason),
        }
    except Exception as e:  # noqa: BLE001
        return {"status": "error", "http_status": None, "href": href, "error": str(e)}

    result = {"status": "ok", "http_status": http_status, "href": href}
    if existed:
        result["existed"] = True
    return result