diff --git a/python/functions/infra/dav_collection_ctag.md b/python/functions/infra/dav_collection_ctag.md new file mode 100644 index 00000000..4bdb1a7c --- /dev/null +++ b/python/functions/infra/dav_collection_ctag.md @@ -0,0 +1,88 @@ +--- +name: dav_collection_ctag +kind: function +lang: py +domain: infra +version: "1.0.0" +purity: impure +signature: "def dav_collection_ctag(base_url: str, username: str, password: str, collection_path: str, *, timeout_s: float = 10.0, verify_tls: bool = True) -> dict" +description: "Lee el ctag (token de version) de una coleccion DAV en UNA peticion PROPFIND Depth:0 barata. Pide el getctag de CalendarServer (http://calendarserver.org/ns/) y, como respaldo, el getetag DAV de la propia coleccion. El ctag cambia solo cuando cambia algun recurso de la coleccion: comparandolo con un ctag cacheado se decide si recargar el contenido (REPORT) o servir de cache sin tocar la red. Construye el header Authorization: Basic base64(user:pass) a mano con stdlib y parsea el multistatus con regex simple. verify_tls=True por defecto. Maneja errores sin lanzar. Solo stdlib (urllib, base64, re, ssl). Probado contra Xandikos." +tags: [dav, carddav, caldav, ctag, getctag, propfind, cache, sync, http, infra] +uses_functions: [] +uses_types: [] +returns: [] +returns_optional: false +error_type: "error_go_core" +imports: [base64, re, ssl, urllib.error, urllib.request] +params: + - name: base_url + desc: "URL base del servidor DAV (p.ej. 'https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com')." + - name: username + desc: "usuario para HTTP Basic auth (p.ej. 'enmanuel')." + - name: password + desc: "contrasena para HTTP Basic auth. Resolver desde pass con pass_get_secret, nunca hardcodear." + - name: collection_path + desc: "ruta de la coleccion (CardDAV '/enmanuel/contacts/addressbook/' o CalDAV '/enmanuel/calendars/calendar/')." + - name: timeout_s + desc: "timeout de la peticion HTTP en segundos. Default 10.0 (la respuesta es minuscula)." + - name: verify_tls + desc: "si True (default) verifica el certificado TLS. No desactivar salvo entorno de prueba." +output: "dict. En exito: {status:'ok', http_status:int, ctag:str} donde ctag es el getctag de CalendarServer si el servidor lo expone, o el getetag DAV de la coleccion como respaldo. En error (sin lanzar): {status:'error', error:str, http_status:int|None}." +tested: true +tests: + - "test_construye_propfind_depth_0" + - "test_basic_auth_header_correcto" + - "test_devuelve_getctag" + - "test_fallback_a_getetag" + - "test_sin_ctag_ni_etag_devuelve_error" + - "test_httperror_devuelve_status_error" +test_file_path: "python/functions/infra/dav_collection_ctag_test.py" +file_path: "python/functions/infra/dav_collection_ctag.py" +--- + +## Ejemplo + +```python +import sys +sys.path.insert(0, "python/functions") +from infra.pass_get_secret import pass_get_secret +from infra.dav_collection_ctag import dav_collection_ctag + +pw = pass_get_secret("dav/xandikos-enmanuel")["value"] # NO logear + +res = dav_collection_ctag( + base_url="https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com", + username="enmanuel", + password=pw, + collection_path="/enmanuel/calendars/calendar/", +) +print(res["status"], res["ctag"]) # ok e8b39a8b180d25a674b35f0fee3013992b59e51e + +# Patron de cache: si el ctag no cambio, sirve del disco sin descargar. +if res["ctag"] == cached_ctag: + return cached_payload +# ...si cambio, recargar con dav_get_collection y guardar el nuevo ctag. +``` + +## Cuando usarla + +Antes de descargar una coleccion DAV completa: pides el ctag (peticion +minuscula, ~10ms) y lo comparas con el ctag de tu cache. Si coincide, sirves de +cache sin tocar la red (arranque instantaneo); si difiere, recargas con +`dav_get_collection` y guardas el nuevo ctag. Es el primitivo de validacion de +cache para CardDAV/CalDAV: una sola comprobacion barata decide si la copia local +sigue vigente. + +## Gotchas + +- El `getctag` es la extension de CalendarServer (`http://calendarserver.org/ns/`), + ampliamente soportada (Xandikos la expone). Si el servidor no la implementa, la + funcion cae al `getetag` DAV de la coleccion, que en Xandikos tambien cambia al + cambiar cualquier recurso — sirve igual como token de version. +- El ctag es OPACO: no lo interpretes, solo comparalo por igualdad con el que + guardaste. No asumas orden ni formato (Xandikos usa un hash hex; otros + servidores usan timestamps u otros formatos). +- No garantiza deteccion de cambios sub-recurso (etag por recurso): solo dice si + ALGO cambio en la coleccion. Para sync incremental fino combina con + `dav_list_resources` (mapa href->etag). +- Lectura remota real sobre TLS; password de `pass`, no se logea. diff --git a/python/functions/infra/dav_collection_ctag.py b/python/functions/infra/dav_collection_ctag.py new file mode 100644 index 00000000..9965284e --- /dev/null +++ b/python/functions/infra/dav_collection_ctag.py @@ -0,0 +1,104 @@ +"""Lee el ctag de una coleccion DAV en UNA peticion barata (PROPFIND Depth:0). + +Funcion impura: hace un unico PROPFIND Depth:0 sobre la coleccion pidiendo el +`getctag` de CalendarServer (`http://calendarserver.org/ns/`) y, como respaldo, +el `getetag` DAV de la propia coleccion. El ctag es un token opaco que cambia +SOLO cuando cambia algun recurso de la coleccion: comparandolo con el ctag +cacheado se decide si hay que recargar el contenido (REPORT) o servir de cache +sin tocar la red. + +Construye el header `Authorization: Basic base64(user:pass)` a mano con stdlib y +parsea el multistatus con regex simple. Maneja errores sin lanzar. Solo usa +stdlib (urllib, base64, re, ssl). Probado contra Xandikos. +""" + +import base64 +import re +import ssl +import urllib.error +import urllib.request + +_CTAG_RE = re.compile( + r"<(?:[A-Za-z0-9]+:)?getctag>\s*(.*?)\s*", + re.DOTALL | re.IGNORECASE, +) +_ETAG_RE = re.compile( + r"<(?:[A-Za-z0-9]+:)?getetag>\s*(.*?)\s*", + re.DOTALL | re.IGNORECASE, +) + +_PROPFIND_BODY = ( + '' + '' + "" + "" +) + + +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 _join_url(base_url: str, collection_path: str) -> str: + return base_url.rstrip("/") + "/" + collection_path.strip("/") + "/" + + +def dav_collection_ctag( + base_url: str, + username: str, + password: str, + collection_path: str, + *, + timeout_s: float = 10.0, + verify_tls: bool = True, +) -> dict: + """Lee el ctag (token de version) de una coleccion DAV. + + Args: + base_url: URL base del servidor DAV. + username: usuario para HTTP Basic auth. + password: contrasena para HTTP Basic auth. Resolver desde pass. + collection_path: ruta de la coleccion (CardDAV o CalDAV). + timeout_s: timeout de la peticion en segundos. Default 10.0. + verify_tls: si True (default) verifica el certificado TLS. + + Returns: + dict. En exito: {status:'ok', http_status:int, ctag:str} donde ctag es + el getctag de CalendarServer si el servidor lo expone, o el getetag DAV + de la coleccion como respaldo. En error (sin lanzar): + {status:'error', error:str, http_status:int|None}. + """ + url = _join_url(base_url, collection_path) + headers = { + "Authorization": _basic_auth_header(username, password), + "Content-Type": "application/xml; charset=utf-8", + "Depth": "0", + } + req = urllib.request.Request( + url, data=_PROPFIND_BODY.encode("utf-8"), method="PROPFIND", headers=headers + ) + + context = None if verify_tls else ssl._create_unverified_context() + try: + with urllib.request.urlopen(req, timeout=timeout_s, context=context) as resp: + status = resp.status + xml = resp.read().decode("utf-8", "replace") + except urllib.error.HTTPError as e: + return {"status": "error", "error": "http %s" % e.code, "http_status": e.code} + except urllib.error.URLError as e: + return {"status": "error", "error": str(e.reason), "http_status": None} + except Exception as e: # noqa: BLE001 + return {"status": "error", "error": str(e), "http_status": None} + + ctag_m = _CTAG_RE.search(xml) + if ctag_m: + return {"status": "ok", "http_status": status, "ctag": ctag_m.group(1).strip()} + etag_m = _ETAG_RE.search(xml) + if etag_m: + return {"status": "ok", "http_status": status, "ctag": etag_m.group(1).strip()} + return { + "status": "error", + "error": "ni getctag ni getetag en la respuesta", + "http_status": status, + } diff --git a/python/functions/infra/dav_collection_ctag_test.py b/python/functions/infra/dav_collection_ctag_test.py new file mode 100644 index 00000000..c6cd4df9 --- /dev/null +++ b/python/functions/infra/dav_collection_ctag_test.py @@ -0,0 +1,119 @@ +"""Tests para dav_collection_ctag. + +Smoke deterministas: monkeypatchean urllib.request.urlopen para capturar el +Request (method PROPFIND, Depth:0, auth) y devolver un multistatus simulado. +Cubren: getctag presente, fallback a getetag cuando no hay getctag, ninguno de +los dos, y el path de error HTTP. +""" + +import base64 +import sys + +import infra.dav_collection_ctag # noqa: F401 + +mod = sys.modules["infra.dav_collection_ctag"] +dav_collection_ctag = mod.dav_collection_ctag + +_XML_CTAG = ( + '' + '' + "/enmanuel/calendars/calendar/" + "HTTP/1.1 200 OK" + "ctag-abc123" + '"etag-abc123"' + "" +) + +_XML_SOLO_ETAG = ( + '' + '' + "/enmanuel/contacts/addressbook/" + "HTTP/1.1 200 OK" + '"etag-only-999"' + "" +) + +_XML_VACIO = ( + '' + '' + "/enmanuel/contacts/addressbook/" + "HTTP/1.1 404 Not Found" + "" +) + + +class _FakeResp: + def __init__(self, payload: str): + self._payload = payload + self.status = 207 + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return self._payload.encode("utf-8") + + +def _capture(monkeypatch, payload: str): + captured = {} + + def fake_urlopen(req, timeout=None, context=None): + captured["method"] = req.get_method() + captured["headers"] = {k.lower(): v for k, v in req.header_items()} + return _FakeResp(payload) + + monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen) + return captured + + +def _call(path="/enmanuel/calendars/calendar/"): + return dav_collection_ctag( + "https://dav.example.com", "enmanuel", "secret-pw", path + ) + + +def test_construye_propfind_depth_0(monkeypatch): + cap = _capture(monkeypatch, _XML_CTAG) + _call() + assert cap["method"] == "PROPFIND" + assert cap["headers"]["depth"] == "0" + + +def test_basic_auth_header_correcto(monkeypatch): + cap = _capture(monkeypatch, _XML_CTAG) + _call() + expected = "Basic " + base64.b64encode(b"enmanuel:secret-pw").decode("ascii") + assert cap["headers"]["authorization"] == expected + + +def test_devuelve_getctag(monkeypatch): + _capture(monkeypatch, _XML_CTAG) + res = _call() + assert res["status"] == "ok" + assert res["ctag"] == "ctag-abc123" + + +def test_fallback_a_getetag(monkeypatch): + _capture(monkeypatch, _XML_SOLO_ETAG) + res = _call("/enmanuel/contacts/addressbook/") + assert res["status"] == "ok" + assert res["ctag"] == '"etag-only-999"' + + +def test_sin_ctag_ni_etag_devuelve_error(monkeypatch): + _capture(monkeypatch, _XML_VACIO) + res = _call("/enmanuel/contacts/addressbook/") + assert res["status"] == "error" + + +def test_httperror_devuelve_status_error(monkeypatch): + def fake_urlopen(req, timeout=None, context=None): + raise mod.urllib.error.HTTPError(req.full_url, 401, "Unauthorized", {}, None) + + monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen) + res = _call() + assert res["status"] == "error" + assert res["http_status"] == 401 diff --git a/python/functions/infra/dav_get_collection.md b/python/functions/infra/dav_get_collection.md new file mode 100644 index 00000000..1b94cd6a --- /dev/null +++ b/python/functions/infra/dav_get_collection.md @@ -0,0 +1,107 @@ +--- +name: dav_get_collection +kind: function +lang: py +domain: infra +version: "1.0.0" +purity: impure +signature: "def dav_get_collection(base_url: str, username: str, password: str, collection_path: str, content_type: str = 'vcard', *, timeout_s: float = 30.0, verify_tls: bool = True) -> dict" +description: "Descarga TODOS los recursos de una coleccion DAV en UNA peticion HTTP REPORT con el contenido inline, evitando el patron N+1 (PROPFIND + un GET por recurso). Usa addressbook-query (CardDAV, content_type='vcard') o calendar-query (CalDAV, content_type='ical'); el servidor responde un multistatus con el vCard/VCALENDAR de cada recurso embebido. Construye el header Authorization: Basic base64(user:pass) a mano con stdlib, parsea el XML con regex simple y des-escapa las entidades XML del contenido. Para 1064 contactos baja de ~9s (N GETs) a ~1s (1 REPORT). verify_tls=True por defecto. Maneja errores sin lanzar. Solo stdlib (urllib, base64, re, ssl, html). Probado contra Xandikos." +tags: [dav, carddav, caldav, report, multiget, addressbook-query, calendar-query, bulk, http, infra] +uses_functions: [] +uses_types: [] +returns: [] +returns_optional: false +error_type: "error_go_core" +imports: [base64, html, re, ssl, urllib.error, urllib.request] +params: + - name: base_url + desc: "URL base del servidor DAV (p.ej. 'https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com')." + - name: username + desc: "usuario para HTTP Basic auth (p.ej. 'enmanuel')." + - name: password + desc: "contrasena para HTTP Basic auth. Resolver desde pass con pass_get_secret, nunca hardcodear." + - name: collection_path + desc: "ruta de la coleccion (CardDAV '/enmanuel/contacts/addressbook/' o CalDAV '/enmanuel/calendars/calendar/')." + - name: content_type + desc: "tipo de la coleccion: 'vcard' (CardDAV, default) o 'ical' (CalDAV). Acepta sinonimos: 'carddav'/'contacts'/'addressbook' -> vcard; 'caldav'/'calendar'/'icalendar' -> ical." + - name: timeout_s + desc: "timeout de la peticion HTTP en segundos. Default 30.0 (la respuesta puede ser grande: ~600KB para 1000 contactos)." + - name: verify_tls + desc: "si True (default) verifica el certificado TLS. No desactivar salvo entorno de prueba." +output: "dict. En exito: {status:'ok', http_status:int, resources:[{href:str, etag:str|None, data:str}, ...]} con un elemento por recurso de la coleccion; data es el vCard / VCALENDAR completo ya des-escapado. En error (sin lanzar): {status:'error', error:str, http_status:int|None}." +tested: true +tests: + - "test_vcard_construye_report_addressbook_query" + - "test_ical_construye_report_calendar_query_con_filtro" + - "test_basic_auth_header_correcto" + - "test_parsea_resources_con_data_inline" + - "test_desescapa_entidades_xml_del_data" + - "test_ical_parsea_calendar_data" + - "test_acepta_sinonimos_de_content_type" + - "test_content_type_invalido_devuelve_error" + - "test_httperror_devuelve_status_error" +test_file_path: "python/functions/infra/dav_get_collection_test.py" +file_path: "python/functions/infra/dav_get_collection.py" +--- + +## Ejemplo + +```python +import sys +sys.path.insert(0, "python/functions") +from infra.pass_get_secret import pass_get_secret +from infra.dav_get_collection import dav_get_collection + +pw = pass_get_secret("dav/xandikos-enmanuel")["value"] # NO logear + +# Todos los contactos en UNA peticion (~1s para 1064 vCards): +res = dav_get_collection( + base_url="https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com", + username="enmanuel", + password=pw, + collection_path="/enmanuel/contacts/addressbook/", + content_type="vcard", +) +print(res["status"], len(res["resources"])) # ok 1064 +print(res["resources"][0]["data"][:40]) # BEGIN:VCARD\nVERSION:3.0\nFN:... + +# Todos los eventos del calendario en UNA peticion: +cal = dav_get_collection( + base_url="https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com", + username="enmanuel", + password=pw, + collection_path="/enmanuel/calendars/calendar/", + content_type="ical", +) +print(cal["status"], len(cal["resources"])) # ok 98 +``` + +## Cuando usarla + +Cuando necesitas el contenido de TODOS los recursos de una coleccion CardDAV o +CalDAV (renderizar la agenda completa, listar todos los eventos, sincronizar en +bloque) y no solo sus hrefs. Sustituye a `dav_list_resources` + un +`dav_get_resource` por recurso: una sola ida y vuelta en lugar de N+1, lo que +para colecciones de cientos/miles de recursos es la diferencia entre ~9s y ~1s. +Si solo necesitas los hrefs/etags (sin contenido), usa `dav_list_resources`; si +necesitas un unico recurso, usa `dav_get_resource`. + +## Gotchas + +- Usa los REPORT `addressbook-query` / `calendar-query` (RFC 6352 / 4791) con + Depth:1, NO `addressbook-multiget` (que en Xandikos exige Depth:0 + una lista + explicita de hrefs en el cuerpo). El query no necesita conocer los hrefs de + antemano: una sola peticion trae todo. +- El namespace CardDAV/CalDAV es el "legacy" `urn:ietf:params:xml:ns:carddav` + (con `:ns:`), que es el que Xandikos anuncia en su `supported-report-set`. El + namespace sin `:ns:` (`urn:ietf:params:xml:carddav`) provoca un 403 + "Unknown report" en Xandikos. +- El contenido inline viene XML-escapado en el multistatus (`<`, `>`, + `&`); la funcion lo des-escapa con `html.unescape` antes de devolverlo. + El `data` resultante es el vCard / VCALENDAR tal cual lo guardo el servidor. +- El parseo es regex simple sobre el multistatus (KISS, sin parser XML): robusto + para la salida estandar de Xandikos, podria fallar con XML muy exotico. +- La respuesta puede ser grande (~600KB para 1000 contactos): el timeout default + es 30s, mayor que el de `dav_list_resources` por eso. +- Lectura remota real sobre TLS; password de `pass`, no se logea. diff --git a/python/functions/infra/dav_get_collection.py b/python/functions/infra/dav_get_collection.py new file mode 100644 index 00000000..af40e607 --- /dev/null +++ b/python/functions/infra/dav_get_collection.py @@ -0,0 +1,200 @@ +"""Descarga TODOS los recursos de una coleccion DAV en UNA peticion (REPORT). + +Funcion impura: hace una unica peticion HTTP REPORT (`addressbook-query` para +CardDAV, `calendar-query` para CalDAV) que el servidor responde con un XML +multistatus que lleva el contenido de cada recurso INLINE (vCard / VCALENDAR). +Esto reemplaza el patron N+1 (un PROPFIND + un GET por recurso) por una sola +ida y vuelta: para 1064 contactos baja de ~9s a ~1s. + +Construye el header `Authorization: Basic base64(user:pass)` a mano con stdlib y +parsea el multistatus con regex simple (sin parser XML externo), des-escapando +las entidades XML del contenido inline. Maneja errores sin lanzar. Solo usa +stdlib (urllib, base64, re, ssl, html). Probado contra Xandikos. +""" + +import base64 +import html +import re +import ssl +import urllib.error +import urllib.request + +# Namespaces y elementos por tipo de coleccion. Xandikos (y la mayoria de +# servidores) usan el namespace "legacy" con `:ns:` que es el que aparece en el +# supported-report-set, no `urn:ietf:params:xml:carddav`. +_PROFILES = { + "vcard": { + "ns": "urn:ietf:params:xml:ns:carddav", + "report": "addressbook-query", + "data_prop": "address-data", + }, + "ical": { + "ns": "urn:ietf:params:xml:ns:caldav", + "report": "calendar-query", + "data_prop": "calendar-data", + }, +} + +# Aceptamos sinonimos comunes para no atar al caller a un literal exacto. +_ALIASES = { + "vcard": "vcard", + "carddav": "vcard", + "contacts": "vcard", + "addressbook": "vcard", + "ical": "ical", + "icalendar": "ical", + "caldav": "ical", + "calendar": "ical", +} + +_RESPONSE_RE = re.compile( + r"<(?:[A-Za-z0-9]+:)?response>(.*?)", + re.DOTALL | re.IGNORECASE, +) +_HREF_RE = re.compile( + r"<(?:[A-Za-z0-9]+:)?href>\s*(.*?)\s*", + re.DOTALL | re.IGNORECASE, +) +_ETAG_RE = re.compile( + r"<(?:[A-Za-z0-9]+:)?getetag>\s*(.*?)\s*", + re.DOTALL | re.IGNORECASE, +) + + +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 _join_url(base_url: str, collection_path: str) -> str: + return base_url.rstrip("/") + "/" + collection_path.strip("/") + "/" + + +def _report_body(profile: dict) -> str: + """Construye el cuerpo XML del REPORT query para el perfil dado. + + `addressbook-query` (CardDAV) no lleva filtro de tiempo: trae todos los + vCards. `calendar-query` (CalDAV) exige un `` con un comp-filter de + VCALENDAR; sin un comp-filter interno trae todos los componentes (todos los + eventos), que es lo que queremos. + """ + ns = profile["ns"] + report = profile["report"] + data_prop = profile["data_prop"] + prop = "" % data_prop + if report == "calendar-query": + filt = '' + else: + filt = "" + return ( + '' + '%s%s' + % (report, ns, prop, filt, report) + ) + + +def _data_re(data_prop: str) -> "re.Pattern": + """Regex para extraer el contenido inline del elemento de datos. + + El servidor namespacea el elemento (``); el contenido va + XML-escapado. Capturamos el cuerpo y lo des-escapamos con html.unescape. + """ + return re.compile( + r"<(?:[A-Za-z0-9]+:)?%s[^>]*>(.*?)" + % (re.escape(data_prop), re.escape(data_prop)), + re.DOTALL | re.IGNORECASE, + ) + + +def dav_get_collection( + base_url: str, + username: str, + password: str, + collection_path: str, + content_type: str = "vcard", + *, + timeout_s: float = 30.0, + verify_tls: bool = True, +) -> dict: + """Descarga el contenido de TODOS los recursos de una coleccion en 1 request. + + Hace un REPORT `addressbook-query` (vcard) o `calendar-query` (ical) que + devuelve el multistatus con el contenido inline de cada recurso, evitando + el patron N+1 (PROPFIND + un GET por recurso). + + Args: + base_url: URL base del servidor DAV. + username: usuario para HTTP Basic auth. + password: contrasena para HTTP Basic auth. Resolver desde pass. + collection_path: ruta de la coleccion (CardDAV o CalDAV). + content_type: 'vcard' (CardDAV) o 'ical' (CalDAV). Acepta sinonimos + ('carddav', 'contacts', 'caldav', 'calendar', ...). + timeout_s: timeout de la peticion en segundos. Default 30.0 (la + respuesta puede ser grande: ~600KB para 1000 contactos). + verify_tls: si True (default) verifica el certificado TLS. + + Returns: + dict. En exito: {status:'ok', http_status:int, + resources:[{href:str, etag:str|None, data:str}, ...]} con un elemento + por recurso de la coleccion (el `data` es el vCard / VCALENDAR ya + des-escapado). En error (sin lanzar): {status:'error', error:str, + http_status:int|None}. + """ + key = _ALIASES.get((content_type or "").strip().lower()) + if key is None: + return { + "status": "error", + "error": "content_type invalido: %r (usa 'vcard' o 'ical')" + % content_type, + "http_status": None, + } + profile = _PROFILES[key] + + url = _join_url(base_url, collection_path) + headers = { + "Authorization": _basic_auth_header(username, password), + "Content-Type": "application/xml; charset=utf-8", + # RFC 6352 / 4791: el query REPORT se aplica con Depth:1 sobre la + # coleccion (multiget exigiria Depth:0 + lista de hrefs; query no). + "Depth": "1", + } + req = urllib.request.Request( + url, + data=_report_body(profile).encode("utf-8"), + method="REPORT", + headers=headers, + ) + + context = None if verify_tls else ssl._create_unverified_context() + try: + with urllib.request.urlopen(req, timeout=timeout_s, context=context) as resp: + status = resp.status + xml = resp.read().decode("utf-8", "replace") + except urllib.error.HTTPError as e: + return {"status": "error", "error": "http %s" % e.code, "http_status": e.code} + except urllib.error.URLError as e: + return {"status": "error", "error": str(e.reason), "http_status": None} + except Exception as e: # noqa: BLE001 + return {"status": "error", "error": str(e), "http_status": None} + + coll_tail = collection_path.strip("/").rsplit("/", 1)[-1] + data_re = _data_re(profile["data_prop"]) + resources = [] + for block in _RESPONSE_RE.findall(xml): + href_m = _HREF_RE.search(block) + if not href_m: + continue + href = href_m.group(1).strip() + # Skip la propia coleccion si el servidor la incluyera. + tail = href.rstrip("/").rsplit("/", 1)[-1] + if tail == coll_tail: + continue + data_m = data_re.search(block) + if not data_m: + # Recurso sin contenido inline (404 en ese propstat): se omite. + continue + data = html.unescape(data_m.group(1)).strip() + etag_m = _ETAG_RE.search(block) + etag = etag_m.group(1).strip() if etag_m else None + resources.append({"href": href, "etag": etag, "data": data}) + return {"status": "ok", "http_status": status, "resources": resources} diff --git a/python/functions/infra/dav_get_collection_test.py b/python/functions/infra/dav_get_collection_test.py new file mode 100644 index 00000000..42ab62f9 --- /dev/null +++ b/python/functions/infra/dav_get_collection_test.py @@ -0,0 +1,203 @@ +"""Tests para dav_get_collection. + +Smoke deterministas: monkeypatchean urllib.request.urlopen para capturar el +Request (method REPORT, Depth, auth, cuerpo del query) y devolver un XML +multistatus simulado con contenido inline. Cubren ambos perfiles (vcard / ical), +el des-escapado de entidades XML, los sinonimos de content_type, el content_type +invalido y el path de error HTTP. +""" + +import base64 +import sys + +import infra.dav_get_collection # noqa: F401 + +mod = sys.modules["infra.dav_get_collection"] +dav_get_collection = mod.dav_get_collection + +# Multistatus de un addressbook-query: 2 vCards inline. El segundo contiene una +# entidad XML (<) que debe des-escaparse a '<' en el campo data. +_VCARD_XML = ( + '' + '' + "/enmanuel/contacts/addressbook/ada.vcf" + "HTTP/1.1 200 OK" + '"etag-ada"' + "BEGIN:VCARD\nFN:Ada\nUID:ada\nEND:VCARD\n" + "" + "/enmanuel/contacts/addressbook/alan.vcf" + "HTTP/1.1 200 OK" + '"etag-alan"' + "BEGIN:VCARD\nFN:Alan <turing>\nUID:alan\nEND:VCARD\n" + "" + "" +) + +# Multistatus de un calendar-query: 1 VEVENT inline. +_ICAL_XML = ( + '' + '' + "/enmanuel/calendars/calendar/e1.ics" + "HTTP/1.1 200 OK" + '"etag-e1"' + "BEGIN:VCALENDAR\nBEGIN:VEVENT\nSUMMARY:Cita\nUID:e1\nEND:VEVENT\nEND:VCALENDAR\n" + "" + "" +) + + +class _FakeResp: + def __init__(self, payload: str): + self._payload = payload + self.status = 207 + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return self._payload.encode("utf-8") + + +def _capture(monkeypatch, payload: str): + captured = {} + + def fake_urlopen(req, timeout=None, context=None): + captured["url"] = req.full_url + captured["method"] = req.get_method() + captured["headers"] = {k.lower(): v for k, v in req.header_items()} + captured["body"] = req.data.decode("utf-8") if req.data else "" + return _FakeResp(payload) + + monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen) + return captured + + +def test_vcard_construye_report_addressbook_query(monkeypatch): + cap = _capture(monkeypatch, _VCARD_XML) + dav_get_collection( + "https://dav.example.com", + "enmanuel", + "secret-pw", + "/enmanuel/contacts/addressbook/", + content_type="vcard", + ) + assert cap["method"] == "REPORT" + assert cap["headers"]["depth"] == "1" + assert "addressbook-query" in cap["body"] + assert "address-data" in cap["body"] + assert "urn:ietf:params:xml:ns:carddav" in cap["body"] + + +def test_ical_construye_report_calendar_query_con_filtro(monkeypatch): + cap = _capture(monkeypatch, _ICAL_XML) + dav_get_collection( + "https://dav.example.com", + "enmanuel", + "secret-pw", + "/enmanuel/calendars/calendar/", + content_type="ical", + ) + assert cap["method"] == "REPORT" + assert "calendar-query" in cap["body"] + assert 'comp-filter name="VCALENDAR"' in cap["body"] + assert "urn:ietf:params:xml:ns:caldav" in cap["body"] + + +def test_basic_auth_header_correcto(monkeypatch): + cap = _capture(monkeypatch, _VCARD_XML) + dav_get_collection( + "https://dav.example.com", + "enmanuel", + "secret-pw", + "/enmanuel/contacts/addressbook/", + ) + expected = "Basic " + base64.b64encode(b"enmanuel:secret-pw").decode("ascii") + assert cap["headers"]["authorization"] == expected + + +def test_parsea_resources_con_data_inline(monkeypatch): + _capture(monkeypatch, _VCARD_XML) + res = dav_get_collection( + "https://dav.example.com", + "enmanuel", + "secret-pw", + "/enmanuel/contacts/addressbook/", + ) + assert res["status"] == "ok" + assert len(res["resources"]) == 2 + by_href = {r["href"]: r for r in res["resources"]} + ada = by_href["/enmanuel/contacts/addressbook/ada.vcf"] + assert ada["etag"] == '"etag-ada"' + assert "BEGIN:VCARD" in ada["data"] + assert "FN:Ada" in ada["data"] + + +def test_desescapa_entidades_xml_del_data(monkeypatch): + _capture(monkeypatch, _VCARD_XML) + res = dav_get_collection( + "https://dav.example.com", + "enmanuel", + "secret-pw", + "/enmanuel/contacts/addressbook/", + ) + alan = next(r for r in res["resources"] if r["href"].endswith("alan.vcf")) + # <turing> debe quedar des-escapado a . + assert "FN:Alan " in alan["data"] + + +def test_ical_parsea_calendar_data(monkeypatch): + _capture(monkeypatch, _ICAL_XML) + res = dav_get_collection( + "https://dav.example.com", + "enmanuel", + "secret-pw", + "/enmanuel/calendars/calendar/", + content_type="ical", + ) + assert res["status"] == "ok" + assert len(res["resources"]) == 1 + assert "BEGIN:VEVENT" in res["resources"][0]["data"] + + +def test_acepta_sinonimos_de_content_type(monkeypatch): + _capture(monkeypatch, _VCARD_XML) + for alias in ("contacts", "carddav", "addressbook"): + res = dav_get_collection( + "https://dav.example.com", + "enmanuel", + "secret-pw", + "/enmanuel/contacts/addressbook/", + content_type=alias, + ) + assert res["status"] == "ok" + + +def test_content_type_invalido_devuelve_error(monkeypatch): + _capture(monkeypatch, _VCARD_XML) + res = dav_get_collection( + "https://dav.example.com", + "enmanuel", + "secret-pw", + "/enmanuel/contacts/addressbook/", + content_type="json", + ) + assert res["status"] == "error" + assert res["http_status"] is None + + +def test_httperror_devuelve_status_error(monkeypatch): + def fake_urlopen(req, timeout=None, context=None): + raise mod.urllib.error.HTTPError(req.full_url, 401, "Unauthorized", {}, None) + + monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen) + res = dav_get_collection( + "https://dav.example.com", + "enmanuel", + "secret-pw", + "/enmanuel/contacts/addressbook/", + ) + assert res["status"] == "error" + assert res["http_status"] == 401