dataforge/osint_web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: descarga DAV concurrente + caché de contactos/calendario

Browse Source 
Las colecciones Xandikos son grandes (1064 contactos, 98 eventos). Descargar
los .vcf/.ics secuencialmente tardaba ~2 min para los contactos (timeout). Se
añade _fetch_resources con un ThreadPoolExecutor acotado (16 workers): primera
carga de /api/contacts baja a ~9s, segunda (cacheada) a ~10ms. La descarga
sigue delegada a dav_get_resource del registry (stdlib, thread-safe); solo se
paraleliza la orquestación.

Incluye caché en memoria de contactos y calendario (invalidada por
/api/refresh), DavUnavailable para degradación clara sin red, y campos
aliaseados en español (nombre/alias/telefonos/correos/osint) para el frontend.

Verificado contra el vault real (1199 nodos) y Xandikos real (1064 contactos,
98 eventos). 19 tests verdes.
This commit is contained in:
agent
2026-06-11 22:54:54 +02:00
parent 5b51e3d035
commit 59558d43cb
2 changed files with 369 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/main.py
+196 -105
View File
@@ -45,8 +45,14 @@ import os
import re
import sys
import threading
from concurrent.futures import ThreadPoolExecutor
from typing import Optional
# Nº de descargas DAV concurrentes al traer una colección completa (addressbook
# con ~1000 vCards). Secuencial son ~0.11s/recurso (~2 min para 1064); con un
# pool acotado baja a ~10s. Acotado para no saturar al servidor Xandikos.
_DAV_FETCH_WORKERS = 16
def _registry_functions_dir() -> str:
"""Localiza ``python/functions`` del fn_registry sin paths hardcodeados.
@@ -148,6 +154,15 @@ XANDIKOS_CALENDAR_COLLECTION = "/enmanuel/calendars/calendar/"
_IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".svg"}
class DavUnavailable(Exception):
"""Xandikos no responde (sin red, timeout, auth caída).
Los endpoints DAV la capturan y devuelven un 503 JSON claro, para que un
fallo de la agenda/calendario NUNCA tumbe el server ni afecte a los
endpoints del vault, que deben seguir funcionando offline.
"""
def _attachment_kind(name: str) -> str:
"""Clasifica un attachment por extensión: ``image`` | ``pdf`` | ``other``."""
ext = os.path.splitext(name)[1].lower()
@@ -396,6 +411,124 @@ class VaultState:
self._xandikos_password = _read_pass_secret(XANDIKOS_PASS_ENTRY)
return self._xandikos_password
def contacts(self) -> list:
"""Contactos del addressbook Xandikos, parseados y cacheados en memoria.
Llena la caché al primer acceso (descarga + parseo de todos los
``.vcf``); accesos posteriores la reutilizan hasta ``invalidate_dav()``.
Raises:
RuntimeError: si no se puede leer la password de ``pass``.
DavUnavailable: si Xandikos no responde (sin red, timeout, auth).
"""
with self._dav_lock:
if self._contacts_cache is not None:
return self._contacts_cache
password = self.xandikos_password()
listing = dav_list_resources(
XANDIKOS_BASE_URL,
XANDIKOS_USERNAME,
password,
XANDIKOS_CONTACTS_COLLECTION,
)
if listing.get("status") != "ok":
raise DavUnavailable(
"Xandikos no responde: %s" % listing.get("error")
)
contacts: list = []
# Descarga concurrente de los .vcf: secuencial son ~0.11s/recurso
# (~2 min para 1064 contactos); con el pool acotado baja a ~10s.
for res, got in self._fetch_resources(
listing.get("resources", []), ".vcf", password
):
href = res.get("href")
for card_text in split_vcards(got.get("text", "")):
card = _vcard_to_json(card_text)
card["etag"] = res.get("etag")
card["href"] = href
contacts.append(card)
contacts.sort(key=lambda c: (c.get("fn") or c.get("uid") or "").lower())
self._contacts_cache = contacts
return contacts
def calendar(self, dt_from: str = "", dt_to: str = "") -> list:
"""Eventos del calendario Xandikos, cacheados; filtrados por rango.
La descarga + parseo completos se cachean; el filtro por ``[from, to]``
se aplica sobre la caché en cada llamada (barato). Sin ``from``/``to``
devuelve todos.
Raises:
RuntimeError: si no se puede leer la password de ``pass``.
DavUnavailable: si Xandikos no responde (sin red, timeout, auth).
"""
with self._dav_lock:
if self._calendar_cache is None:
password = self.xandikos_password()
listing = dav_list_resources(
XANDIKOS_BASE_URL,
XANDIKOS_USERNAME,
password,
XANDIKOS_CALENDAR_COLLECTION,
)
if listing.get("status") != "ok":
raise DavUnavailable(
"Xandikos no responde: %s" % listing.get("error")
)
events: list = []
for res, got in self._fetch_resources(
listing.get("resources", []), ".ics", password
):
href = res.get("href")
for event in _vcalendar_to_events(got.get("text", "")):
event["etag"] = res.get("etag")
event["href"] = href
events.append(event)
events.sort(key=lambda e: e.get("dtstart") or "")
self._calendar_cache = events
all_events = self._calendar_cache
if not dt_from and not dt_to:
return list(all_events)
return [e for e in all_events if _event_in_range(e, dt_from, dt_to)]
def _fetch_resources(self, resources: list, suffix: str, password: str) -> list:
"""Descarga en paralelo los recursos DAV con la extensión ``suffix``.
Filtra los recursos por extensión (``.vcf`` / ``.ics``), los descarga con
``dav_get_resource`` (función del registry) usando un pool acotado de
hilos (``_DAV_FETCH_WORKERS``) y devuelve la lista de pares
``(res, got)`` de los que respondieron ``status == "ok"``, preservando el
orden del listing. La paralelización es solo de la orquestación: la
descarga sigue delegada a la función del registry, que es stdlib y
thread-safe (abre su propia conexión por request). Acotar el pool evita
saturar al servidor Xandikos.
"""
targets = [
res
for res in resources
if res.get("href") and res["href"].lower().endswith(suffix)
]
if not targets:
return []
def _get(res):
got = dav_get_resource(
XANDIKOS_BASE_URL, XANDIKOS_USERNAME, password, res["href"]
)
return res, got
workers = min(_DAV_FETCH_WORKERS, len(targets))
with ThreadPoolExecutor(max_workers=workers) as pool:
# pool.map preserva el orden de entrada.
results = list(pool.map(_get, targets))
return [(res, got) for res, got in results if got.get("status") == "ok"]
def invalidate_dav(self) -> None:
"""Vacía las cachés de contactos y calendario (no la password)."""
with self._dav_lock:
self._contacts_cache = None
self._calendar_cache = None
# ---------------------------------------------------------------------------
# Helpers DAV: parseo ligero de vCard / iCalendar a JSON
@@ -570,30 +703,22 @@ def _vcalendar_to_events(vcalendar_text: str) -> list:
def _event_in_range(event: dict, dt_from: str, dt_to: str) -> bool:
"""True si el evento cae (por DTSTART) dentro de ``[dt_from, dt_to]``.
La comparación es lexicográfica sobre el prefijo de fecha ``YYYYMMDD`` que
comparten todos los formatos iCal (date y date-time). ``dt_from``/``dt_to``
Comparación lexicográfica sobre el prefijo ``YYYYMMDD`` que comparten todos
los formatos iCal (date y date-time). Los límites se normalizan quitando los
guiones, así acepta tanto el formato documentado del endpoint
(``2026-06-11``) como el iCal crudo (``20260611``). ``dt_from``/``dt_to``
vacíos desactivan ese extremo del filtro.
"""
dtstart = (event.get("dtstart") or "")[:8]
dtstart = (event.get("dtstart") or "").replace("-", "")[:8]
if not dtstart:
return True
if dt_from and dtstart < dt_from[:8]:
if dt_from and dtstart < dt_from.replace("-", "")[:8]:
return False
if dt_to and dtstart > dt_to[:8]:
if dt_to and dtstart > dt_to.replace("-", "")[:8]:
return False
return True
def _uid_to_href(uid: str, resources: list) -> Optional[str]:
"""Localiza el href de un recurso DAV cuyo último segmento contiene el uid."""
for res in resources:
href = res.get("href", "")
tail = href.rstrip("/").rsplit("/", 1)[-1]
if uid in tail or tail.startswith(uid):
return href
return None
# ---------------------------------------------------------------------------
# Construcción de la app FastAPI
# ---------------------------------------------------------------------------
@@ -679,127 +804,93 @@ def create_app(vault_dir: str) -> FastAPI:
@app.get("/api/contacts")
def api_contacts() -> JSONResponse:
"""Contactos del addressbook Xandikos, parseados a JSON.
"""Contactos del addressbook Xandikos, parseados a JSON (cacheados).
Lista los recursos de la colección CardDAV, descarga cada ``.vcf`` y lo
parsea con ``split_vcards`` + parseo ligero. Si Xandikos no responde
(sin red) devuelve un error claro (502), no un crash.
Cada contacto: ``{uid, nombre, alias, nota, org, telefonos[], emails[],
osint{dni,pais,sexo,...}}`` (+ las formas tipadas ``phones``/``emails``).
La lista se cachea en memoria al primer acceso (``POST /api/refresh`` la
invalida). Si Xandikos no responde o falta la password → 503 con un JSON
de error claro, nunca un crash.
"""
try:
password = state.xandikos_password()
except RuntimeError as exc:
return JSONResponse(status_code=503, content={"status": "error", "error": str(exc)})
listing = dav_list_resources(
XANDIKOS_BASE_URL,
XANDIKOS_USERNAME,
password,
XANDIKOS_CONTACTS_COLLECTION,
)
if listing.get("status") != "ok":
contacts = state.contacts()
except (RuntimeError, DavUnavailable) as exc:
return JSONResponse(
status_code=502,
content={"status": "error", "error": "Xandikos no responde: %s" % listing.get("error")},
status_code=503, content={"status": "error", "error": str(exc)}
)
contacts: list = []
for res in listing.get("resources", []):
href = res.get("href")
if not href or not href.lower().endswith(".vcf"):
continue
got = dav_get_resource(XANDIKOS_BASE_URL, XANDIKOS_USERNAME, password, href)
if got.get("status") != "ok":
continue
for card_text in split_vcards(got.get("text", "")):
card = _vcard_to_json(card_text)
card["etag"] = res.get("etag")
contacts.append(card)
contacts.sort(key=lambda c: (c.get("fn") or c.get("uid") or "").lower())
return JSONResponse(content={"status": "ok", "count": len(contacts), "contacts": contacts})
return JSONResponse(
content={"status": "ok", "count": len(contacts), "contacts": contacts}
)
@app.get("/api/contact/{uid}")
def api_contact(uid: str) -> JSONResponse:
"""Un vCard concreto (por UID) a JSON."""
try:
password = state.xandikos_password()
except RuntimeError as exc:
return JSONResponse(status_code=503, content={"status": "error", "error": str(exc)})
"""Un contacto concreto (por UID) parseado a JSON, desde la caché.
listing = dav_list_resources(
XANDIKOS_BASE_URL, XANDIKOS_USERNAME, password, XANDIKOS_CONTACTS_COLLECTION
)
if listing.get("status") != "ok":
Resuelve sobre la lista cacheada de ``/api/contacts`` (mismo parseo
completo, todos los campos). 404 si el UID no existe; 503 si Xandikos no
responde o falta la password.
"""
try:
contacts = state.contacts()
except (RuntimeError, DavUnavailable) as exc:
return JSONResponse(
status_code=502,
content={"status": "error", "error": "Xandikos no responde: %s" % listing.get("error")},
status_code=503, content={"status": "error", "error": str(exc)}
)
href = _uid_to_href(uid, listing.get("resources", []))
if not href:
raise HTTPException(status_code=404, detail="contacto '%s' no encontrado" % uid)
got = dav_get_resource(XANDIKOS_BASE_URL, XANDIKOS_USERNAME, password, href)
if got.get("status") != "ok":
return JSONResponse(
status_code=502,
content={"status": "error", "error": "no se pudo descargar el vCard"},
match = next((c for c in contacts if c.get("uid") == uid), None)
if match is None:
# Tolerancia: aceptar también el segmento final del href (nombre del
# recurso .vcf) cuando el UID no coincide literalmente.
match = next(
(
c
for c in contacts
if uid in (c.get("href") or "").rsplit("/", 1)[-1]
),
None,
)
cards = split_vcards(got.get("text", ""))
if not cards:
raise HTTPException(status_code=404, detail="vCard vacío")
return JSONResponse(content={"status": "ok", "contact": _vcard_to_json(cards[0])})
if match is None:
raise HTTPException(
status_code=404, detail="contacto '%s' no encontrado" % uid
)
return JSONResponse(content={"status": "ok", "contact": match})
# -- Xandikos: calendario (CalDAV) --
@app.get("/api/calendar")
def api_calendar(
from_: str = Query("", alias="from", description="fecha inicio YYYYMMDD"),
to: str = Query("", description="fecha fin YYYYMMDD"),
from_: str = Query("", alias="from", description="fecha inicio YYYY-MM-DD"),
to: str = Query("", description="fecha fin YYYY-MM-DD"),
) -> JSONResponse:
"""Eventos del calendario Xandikos en ``[from, to]``, parseados a JSON.
"""Eventos del calendario Xandikos en ``[from, to]`` (cacheados).
Lista los recursos de la colección CalDAV, descarga cada ``.ics``,
extrae sus VEVENT y los filtra por DTSTART dentro del rango. Sin red ->
error claro (502), no crash.
Cada evento: ``{uid, summary, dtstart, dtend, location, description}``.
La descarga + parseo completos se cachean (``POST /api/refresh`` los
invalida); el filtro por rango se aplica sobre la caché. Sin ``from``/
``to`` devuelve todos. Si Xandikos no responde o falta la password →
503 con JSON de error claro, nunca un crash.
"""
try:
password = state.xandikos_password()
except RuntimeError as exc:
return JSONResponse(status_code=503, content={"status": "error", "error": str(exc)})
listing = dav_list_resources(
XANDIKOS_BASE_URL, XANDIKOS_USERNAME, password, XANDIKOS_CALENDAR_COLLECTION
)
if listing.get("status") != "ok":
events = state.calendar(from_, to)
except (RuntimeError, DavUnavailable) as exc:
return JSONResponse(
status_code=502,
content={"status": "error", "error": "Xandikos no responde: %s" % listing.get("error")},
status_code=503, content={"status": "error", "error": str(exc)}
)
events: list = []
for res in listing.get("resources", []):
href = res.get("href")
if not href or not href.lower().endswith(".ics"):
continue
got = dav_get_resource(XANDIKOS_BASE_URL, XANDIKOS_USERNAME, password, href)
if got.get("status") != "ok":
continue
for event in _vcalendar_to_events(got.get("text", "")):
if _event_in_range(event, from_, to):
events.append(event)
events.sort(key=lambda e: e.get("dtstart") or "")
return JSONResponse(content={"status": "ok", "count": len(events), "events": events})
return JSONResponse(
content={"status": "ok", "count": len(events), "events": events}
)
# -- Refresco de cachés --
@app.post("/api/refresh")
def api_refresh() -> dict:
"""Invalida y reconstruye la caché del grafo del vault.
"""Reconstruye la caché del grafo del vault e invalida las cachés DAV.
Los datos DAV no se cachean, así que esto solo afecta al grafo/tablas del
vault. Devuelve el conteo del grafo recién reconstruido.
Re-escanea el vault (grafo + tablas) y vacía las cachés de contactos y
calendario, que se recargarán perezosamente en el siguiente acceso.
Devuelve el conteo del grafo recién reconstruido.
"""
summary = state.refresh()
state.invalidate_dav()
return {"status": "refreshed", **summary}
return app
tests/test_server.py
+173 -3
View File
@@ -236,7 +236,10 @@ def test_vevent_to_json_and_range():
def test_dav_endpoints_degrade_without_network(client, monkeypatch):
"""Sin Xandikos accesible los endpoints DAV devuelven error claro, no crash."""
"""Sin Xandikos accesible los endpoints DAV devuelven 503 claro, no crash.
Y los endpoints del vault siguen funcionando offline (no se ven afectados).
"""
monkeypatch.setattr(
srv, "dav_list_resources", lambda *a, **k: {"status": "error", "error": "sin red"}
)
@@ -244,9 +247,176 @@ def test_dav_endpoints_degrade_without_network(client, monkeypatch):
client.app.state.vault._xandikos_password = "x"
r1 = client.get("/api/contacts")
assert r1.status_code == 502
assert r1.status_code == 503
assert r1.json()["status"] == "error"
r2 = client.get("/api/calendar")
assert r2.status_code == 502
assert r2.status_code == 503
assert r2.json()["status"] == "error"
# El fallo DAV NO contamina los endpoints del vault (offline-OK).
assert client.get("/api/graph").status_code == 200
assert client.get("/api/health").status_code == 200
# ---------------------------------------------------------------------------
# DAV: campos osint / alias / nota / itemN. + caché + invalidación
# ---------------------------------------------------------------------------
def test_vcard_to_json_alias_nota_osint_y_item_prefix():
"""Parsea alias (NICKNAME), nota (NOTE), osint (X-OSINT-*) y prefijo itemN."""
vcard = (
"BEGIN:VCARD\r\n"
"VERSION:3.0\r\n"
"UID:maria-001\r\n"
"FN:María del Mar Pérez\r\n"
"NICKNAME:Marimar\r\n"
"item1.TEL;TYPE=CELL:+34 600 111 222\r\n"
"item2.EMAIL;TYPE=INTERNET:maria@example.com\r\n"
"NOTE:Objetivo principal.\r\n"
"X-OSINT-DNI:12345678Z\r\n"
"X-OSINT-PAIS:España\r\n"
"X-OSINT-SEXO:F\r\n"
"END:VCARD\r\n"
)
out = srv._vcard_to_json(vcard)
assert out["uid"] == "maria-001"
assert out["nombre"] == "María del Mar Pérez"
assert out["alias"] == "Marimar"
assert out["nota"] == "Objetivo principal."
# El prefijo itemN. se elimina: TEL/EMAIL se reconocen.
assert out["telefonos"] == ["+34 600 111 222"]
assert out["correos"] == ["maria@example.com"]
# Bloque osint derivado de X-OSINT-*.
assert out["osint"] == {"dni": "12345678Z", "pais": "España", "sexo": "F"}
def test_vcard_to_json_nombre_desde_N_sin_fn():
vcard = (
"BEGIN:VCARD\r\nVERSION:3.0\r\nUID:juan-002\r\n"
"N:García;Juan;;;\r\nTEL:+34 611 222 333\r\nEND:VCARD\r\n"
)
out = srv._vcard_to_json(vcard)
assert out["nombre"] == "Juan García"
assert out["osint"] == {}
# Fixture DAV mockeado: dos contactos (uno con osint) y dos eventos.
_VCF_BODY = (
"BEGIN:VCARD\r\nVERSION:3.0\r\nUID:maria-001\r\nFN:María Pérez\r\n"
"NICKNAME:Mari\r\nX-OSINT-DNI:12345678Z\r\nX-OSINT-PAIS:España\r\n"
"item1.TEL;TYPE=CELL:+34600111222\r\nEND:VCARD\r\n"
)
_VCF_BODY_2 = (
"BEGIN:VCARD\r\nVERSION:3.0\r\nUID:juan-002\r\nFN:Juan García\r\n"
"EMAIL:juan@example.com\r\nEND:VCARD\r\n"
)
_ICS_BODY = (
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:evt-001\r\n"
"SUMMARY:Reunión\r\nDTSTART:20260611T090000Z\r\nDTEND:20260611T100000Z\r\n"
"LOCATION:Madrid\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n"
)
_ICS_BODY_2 = (
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:evt-002\r\n"
"SUMMARY:Vigilancia\r\nDTSTART:20260620T200000Z\r\nEND:VEVENT\r\n"
"END:VCALENDAR\r\n"
)
@pytest.fixture()
def fake_dav(monkeypatch):
"""Parchea las funciones del registry DAV con fixtures en memoria (sin red).
Devuelve un dict ``{"calls": int}`` que cuenta los PROPFIND para verificar
el cacheo (segunda lectura no re-llama a Xandikos).
"""
state = {"calls": 0}
contacts_res = [
{"href": "/enmanuel/contacts/addressbook/maria-001.vcf", "etag": '"a"'},
{"href": "/enmanuel/contacts/addressbook/juan-002.vcf", "etag": '"b"'},
]
calendar_res = [
{"href": "/enmanuel/calendars/calendar/evt-001.ics", "etag": '"c"'},
{"href": "/enmanuel/calendars/calendar/evt-002.ics", "etag": '"d"'},
]
bodies = {
"/enmanuel/contacts/addressbook/maria-001.vcf": _VCF_BODY,
"/enmanuel/contacts/addressbook/juan-002.vcf": _VCF_BODY_2,
"/enmanuel/calendars/calendar/evt-001.ics": _ICS_BODY,
"/enmanuel/calendars/calendar/evt-002.ics": _ICS_BODY_2,
}
def _list(base, user, pw, collection, **kw):
state["calls"] += 1
res = contacts_res if "contacts" in collection else calendar_res
return {"status": "ok", "http_status": 207, "resources": res}
def _get(base, user, pw, href, **kw):
return {"status": "ok", "http_status": 200, "text": bodies.get(href, "")}
monkeypatch.setattr(srv, "dav_list_resources", _list)
monkeypatch.setattr(srv, "dav_get_resource", _get)
monkeypatch.setattr(srv, "pass_get_secret", lambda *a, **k: {"status": "ok", "value": "x"})
return state
def test_contacts_endpoint_parsea_y_cachea(client, fake_dav):
r = client.get("/api/contacts")
assert r.status_code == 200
data = r.json()
assert data["status"] == "ok" and data["count"] == 2
by_uid = {c["uid"]: c for c in data["contacts"]}
maria = by_uid["maria-001"]
assert maria["nombre"] == "María Pérez"
assert maria["alias"] == "Mari"
assert maria["telefonos"] == ["+34600111222"]
assert maria["osint"] == {"dni": "12345678Z", "pais": "España"}
# Segunda llamada NO re-hace PROPFIND (sirve de la caché en memoria).
calls_after_first = fake_dav["calls"]
client.get("/api/contacts")
assert fake_dav["calls"] == calls_after_first
def test_contact_by_uid_desde_cache(client, fake_dav):
r = client.get("/api/contact/maria-001")
assert r.status_code == 200
assert r.json()["contact"]["nombre"] == "María Pérez"
assert client.get("/api/contact/no-existe").status_code == 404
def test_calendar_endpoint_rango_y_cache(client, fake_dav):
# Sin rango: ambos eventos.
r = client.get("/api/calendar")
assert r.status_code == 200 and r.json()["count"] == 2
# Con rango: solo evt-001 (11 junio).
r2 = client.get("/api/calendar", params={"from": "2026-06-01", "to": "2026-06-15"})
assert [e["uid"] for e in r2.json()["events"]] == ["evt-001"]
def test_refresh_invalida_cache_dav(client, fake_dav):
client.get("/api/contacts") # llena caché
calls_before = fake_dav["calls"]
client.post("/api/refresh") # invalida
client.get("/api/contacts") # vuelve a hacer PROPFIND
assert fake_dav["calls"] > calls_before
# ---------------------------------------------------------------------------
# Smoke real opcional contra Xandikos (gateado, no corre en CI)
# ---------------------------------------------------------------------------
@pytest.mark.skipif(
os.environ.get("OSINT_WEB_DAV_SMOKE") != "1",
reason="smoke DAV real desactivado (export OSINT_WEB_DAV_SMOKE=1 para correrlo)",
)
def test_smoke_dav_real(vault):
"""Smoke contra el Xandikos real: ≥1 contacto y ≥1 evento. Requiere red + pass."""
app = srv.create_app(vault)
real_client = TestClient(app)
rc = real_client.get("/api/contacts")
assert rc.status_code == 200
assert rc.json()["status"] == "ok" and rc.json()["count"] >= 1
re_ = real_client.get("/api/calendar")
assert re_.status_code == 200 and re_.json()["status"] == "ok"