merge: push masivo CardDAV por disco + 1 commit

app.md

@@ -131,7 +131,36 @@ Health check: `curl http://127.0.0.1:8771/api/health`.
 | POST/PUT/DELETE | `/api/event[/{uid}]` | CRUD de eventos CalDAV. Push `caldav_put_event`/`dav_delete_resource` |
 | POST | `/api/addressbook` | `{slug, display_name?, description?, color?}` → `dav_make_addressbook` + INSERT en `addressbooks` |
 | POST | `/api/calendar` | `{slug, display_name?, color?}` → `dav_make_calendar` (paridad) |
-| POST | `/api/push/dav` | reconcilia en bloque: recorre `contacts` y `events` de la DB y los empuja a Xandikos (PUT, sin borrar). Útil tras la migración |
+| POST | `/api/push/dav` | reconcilia en bloque por HTTP: recorre `contacts` y `events` de la DB y los empuja a Xandikos (1 PUT + 1 PROPFIND + 1 commit git por recurso, sin borrar). Fallback cuando no hay SSH al host de Xandikos |
+| POST | `/api/push/dav-bulk` | vía RÁPIDA del push de **contactos** por DISCO: genera todos los `.vcf` en un tmpdir local y hace **1 rsync + 1 commit + 1 PROPFIND** contra el working tree git que Xandikos sirve. Reconcilia ~1000 contactos en <1s en vez de ~6 min. Requiere SSH por clave |
+
+### `/api/push/dav` (HTTP) vs `/api/push/dav-bulk` (disco)
+
+Ambos vuelcan los contactos de la DB a Xandikos, pero por mecanismos distintos:
+
+| | `/api/push/dav` (HTTP) | `/api/push/dav-bulk` (disco) |
+|---|---|---|
+| Mecanismo | N PUT WebDAV (uno por contacto) | 1 rsync de todos los `.vcf` a la vez |
+| PROPFIND | 1 por contacto (lee el etag tras cada PUT) | 1 al final, lee todos los etags de golpe |
+| Commits git en el remoto | N (Xandikos commitea cada PUT) | 1 (`git add -A && commit` único) |
+| Coste para ~1000 contactos | ~6 min (≥1 PROPFIND completo/contacto) | <1s (dominado por 2-3 round-trips SSH) |
+| Eventos CalDAV | sí (también empuja `events`) | no (solo `contacts`) |
+| Borra huérfanos remotos | no | sí — `rsync --delete` deja la colección `.vcf` == DB |
+| Requisitos | red HTTPS a Xandikos + `pass` | SSH por clave al host + `rsync` ambos lados |
+
+**Usa `/api/push/dav-bulk`** para reconciliar en bloque (tras una migración o un
+ingest masivo) cuando hay SSH al host de Xandikos: es el camino normal por
+rapidez. **Usa `/api/push/dav`** como fallback cuando no hay SSH, o cuando
+también necesitas empujar eventos CalDAV.
+
+Cómo lo ve Xandikos: sirve cada colección desde el working tree de un repo git y
+calcula el ctag desde el HEAD. El push por disco escribe los `.vcf` directamente
+en ese working tree y hace un único commit; Xandikos ve el nuevo HEAD en el
+siguiente request (nuevo ctag → DAVx5 detecta el cambio). El rsync sincroniza
+SOLO los `*.vcf` (`--include='*.vcf' --exclude='*'`), preservando `.git/`,
+`.xandikos` (tipo de colección) y `push-subscriptions.json` (suscripciones
+WebDAV-Push). Config del host/working tree: `DAV_BULK_SSH_HOST` /
+`DAV_BULK_REMOTE_DIR` en `server/config.py`.
 
 Queries con nombre incluidas: `personas_por_contexto`, `personas_recientes`,
 `eventos_proximos`, `contactos_sin_nota`, `stats_personas`,

server/config.py

@@ -25,6 +25,19 @@ DAV_CONTACTS_COLLECTION = "/enmanuel/contacts/addressbook/"
 DAV_CALENDAR_HOME = "/enmanuel/calendars/"
 PASS_SECRET = "dav/xandikos-enmanuel"
 
+# Vía rápida de push masivo por disco (POST /api/push/dav-bulk): Xandikos sirve
+# cada colección desde el working tree de un repo git en el host que la aloja, y
+# calcula el ctag desde el HEAD. Escribiendo los .vcf directos al working tree y
+# haciendo UN solo commit, Xandikos ve el cambio en el siguiente request (nuevo
+# ctag -> DAVx5 lo detecta), evitando los N PUT + N PROPFIND + N commit del push
+# HTTP. Solo aplica a la colección CardDAV por defecto (la del addressbook).
+#
+# - DAV_BULK_SSH_HOST: alias SSH (~/.ssh/config) del host de Xandikos (magnus).
+# - DAV_BULK_REMOTE_DIR: working tree de la colección CardDAV por defecto en el
+#   host. Debe terminar en '/'. Es un repo git que Xandikos sirve.
+DAV_BULK_SSH_HOST = "magnus"
+DAV_BULK_REMOTE_DIR = "/srv/xandikos/data/enmanuel/contacts/addressbook/"
+
 # Carpetas del vault que mapean a tablas maestras de entidades.
 # (carpeta del vault, tipo de frontmatter, tabla destino)
 ENTITY_FOLDERS = (
@@ -53,3 +66,6 @@ class Config:
     dav_calendar_home: str = DAV_CALENDAR_HOME
     pass_secret: str = PASS_SECRET
     entity_folders: tuple = field(default=ENTITY_FOLDERS)
+    # Push masivo por disco (POST /api/push/dav-bulk).
+    dav_bulk_ssh_host: str = DAV_BULK_SSH_HOST
+    dav_bulk_remote_dir: str = DAV_BULK_REMOTE_DIR

server/main.py

@@ -28,6 +28,8 @@ endpoints de datos responden SIEMPRE 200 con status ok|error en el body):
     POST /api/query/named   ejecuta una query del catálogo {name, max_rows}
     POST /api/ingest/vault  re-escanea el vault y reconstruye maestras+derivadas
     POST /api/ingest/dav    baja Xandikos y reconstruye contacts/events+derivadas
+    POST /api/push/dav      push masivo HTTP de contacts+events (N PUT, fallback sin SSH)
+    POST /api/push/dav-bulk push masivo de contacts por DISCO (1 rsync + 1 commit, requiere SSH)
     POST /api/sync/dav-pull sync inverso incremental por etag (ediciones del móvil)
     POST /api/render/note   ejecuta query y la upserta como bloque sentinel en una nota
 """
@@ -166,6 +168,24 @@ def create_app(cfg: Config) -> FastAPI:
         allowed_hosts=["127.0.0.1", "localhost", "testserver"],
     )
 
+    # Anti-CSRF de navegador: rechaza las peticiones mutantes que el navegador
+    # marca como cross-site (header Sec-Fetch-Site). Cierra el hueco de las
+    # peticiones "simples" (POST sin preflight CORS) que el TrustedHost no filtra
+    # porque su Host sigue siendo 127.0.0.1. Los clientes server-to-server (urllib,
+    # curl) y el frontend mismo-origen no envían 'cross-site', así que no se ven
+    # afectados.
+    from starlette.responses import JSONResponse
+
+    @app.middleware("http")
+    async def _reject_cross_site(request, call_next):
+        if request.method in ("POST", "PUT", "PATCH", "DELETE"):
+            if request.headers.get("sec-fetch-site") == "cross-site":
+                return JSONResponse(
+                    status_code=403,
+                    content={"status": "error", "error": "petición cross-site rechazada"},
+                )
+        return await call_next(request)
+
     def run_readonly(sql: str, params: list, max_rows: int) -> dict:
         """Ejecuta un SELECT con la conexión read_only del registry, acotado."""
         max_rows = max(1, min(int(max_rows), 10000))
@@ -360,6 +380,13 @@ def create_app(cfg: Config) -> FastAPI:
     def push_dav() -> dict:
         return _guard(lambda: writes.push_all_dav(cfg))
 
+    @app.post("/api/push/dav-bulk")
+    def push_dav_bulk() -> dict:
+        # Vía RÁPIDA del push de contactos: 1 rsync + 1 commit + 1 PROPFIND en
+        # el host de Xandikos (requiere SSH por clave). Si no hay SSH, usar
+        # /api/push/dav (HTTP) como fallback.
+        return _guard(lambda: writes.push_all_dav_bulk(cfg))
+
     @app.post("/api/sync/dav-pull")
     def sync_dav_pull() -> dict:
         # Sync inverso: trae a la DB las ediciones del móvil/DAVx5,

server/writes.py

@@ -19,6 +19,11 @@ from __future__ import annotations
 
 import json
 import os
+import re
+import shutil
+import subprocess
+import tempfile
+import time
 from datetime import datetime, timezone
 
 from .config import Config
@@ -60,6 +65,19 @@ def _now():
     return datetime.now(tz=timezone.utc)
 
 
+# Clave (slug de persona, uid de contacto/evento) admisible: empieza por
+# alfanumérico y solo contiene alfanuméricos y `._-`. Rechaza explícitamente
+# separadores de ruta (`/`, `\`), saltos de línea y secuencias `..`, que de otro
+# modo entrarían como filas con clave indeleble por la API REST (los `/` rompen
+# el routing de DELETE /api/<x>/{key}) y se acercan a paths fuera del vault.
+_VALID_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+
+
+def _valid_key(value: str) -> bool:
+    """True si ``value`` es una clave admisible (sin `/`, `\\`, `..`, controles)."""
+    return bool(_VALID_KEY_RE.match(value or "")) and ".." not in value
+
+
 def _as_list(value) -> list:
     """Normaliza a lista de strings no vacíos (string suelto -> [string])."""
     if value is None:
@@ -134,6 +152,8 @@ def upsert_person(cfg: Config, slug: str, fields: dict, *, render: bool = True)
     slug = (slug or "").strip()
     if not slug:
         return {"status": "error", "error": "slug vacío"}
+    if not _valid_key(slug):
+        return {"status": "error", "error": f"slug inválido: {slug!r}"}
 
     telefonos = _as_list(fields.get("telefonos"))
     emails = _as_list(fields.get("emails"))
@@ -402,6 +422,8 @@ def upsert_contact(cfg: Config, uid: str, fields: dict) -> dict:
     uid = (uid or "").strip()
     if not uid:
         return {"status": "error", "error": "uid vacío"}
+    if not _valid_key(uid):
+        return {"status": "error", "error": f"uid inválido: {uid!r}"}
 
     tels = _as_list(fields.get("tels") or fields.get("telefonos"))
     emails = _as_list(fields.get("emails") or fields.get("correos"))
@@ -570,6 +592,8 @@ def upsert_event(cfg: Config, uid: str, fields: dict) -> dict:
     uid = (uid or "").strip()
     if not uid:
         return {"status": "error", "error": "uid vacío"}
+    if not _valid_key(uid):
+        return {"status": "error", "error": f"uid inválido: {uid!r}"}
 
     calendar = fields.get("calendar") or "default"
     raw = _build_vcalendar(uid, fields)
@@ -669,6 +693,8 @@ def make_addressbook(cfg: Config, fields: dict) -> dict:
     slug = (fields.get("slug") or "").strip()
     if not slug:
         return {"status": "error", "error": "slug vacío"}
+    if not _valid_key(slug):
+        return {"status": "error", "error": f"slug inválido: {slug!r}"}
     display_name = fields.get("display_name") or ""
     description = fields.get("description") or ""
     color = fields.get("color") or ""
@@ -728,6 +754,8 @@ def make_calendar(cfg: Config, fields: dict) -> dict:
     slug = (fields.get("slug") or "").strip()
     if not slug:
         return {"status": "error", "error": "slug vacío"}
+    if not _valid_key(slug):
+        return {"status": "error", "error": f"slug inválido: {slug!r}"}
     display_name = fields.get("display_name") or ""
     color = fields.get("color") or ""
 
@@ -835,6 +863,297 @@ def push_all_dav(cfg: Config) -> dict:
     }
 
 
+# ---------------------------------------------------------------------------
+# push masivo POR DISCO (vía rápida: 1 rsync + 1 commit + 1 PROPFIND)
+# ---------------------------------------------------------------------------
+
+
+def _write_agenda_vcards_to_dir(cfg: Config, rows: list, out_dir: str) -> dict:
+    """Genera el .vcf de agenda (SIN OSINT) de cada contacto en ``out_dir``.
+
+    Para cada fila de ``contacts`` (uid, fn, tels, emails, note_path) compone el
+    vCard con ``_compose_agenda_vcard`` y lo escribe a
+    ``out_dir/<_safe_resource(uid)>.vcf`` — EXACTAMENTE el mismo nombre de
+    recurso que usa el push HTTP (``carddav_put_vcard``), para que rsync no cree
+    duplicados ni huérfanos respecto a la colección remota.
+
+    Es la parte testeable sin red/SSH del push masivo por disco: genera los
+    ficheros locales y devuelve el mapa nombre_recurso -> uid (necesario luego
+    para casar los etags del PROPFIND con sus uids).
+
+    Args:
+        cfg: configuración del service (db_path para resolver la persona enlazada).
+        rows: filas de contacts como dicts con uid, fn, tels, emails, note_path.
+        out_dir: directorio temporal local donde escribir los .vcf.
+
+    Returns:
+        {"written": N, "by_resource": {"<safe>.vcf": uid, ...}}.
+    """
+    by_resource: dict = {}
+    written = 0
+    for row in rows:
+        uid = row["uid"]
+        vcard = _compose_agenda_vcard(
+            cfg,
+            uid,
+            {
+                "fn": row.get("fn"),
+                "tels": _decode_json_field(row.get("tels")),
+                "emails": _decode_json_field(row.get("emails")),
+                "note_path": row.get("note_path"),
+            },
+        )
+        resource = _resource_href_tail(uid)  # _safe_resource(uid) + ".vcf"
+        # En el caso (raro) de que dos uids saneen al mismo recurso, gana el
+        # último, igual que haría el push HTTP secuencial.
+        by_resource[resource] = uid
+        with open(os.path.join(out_dir, resource), "w", encoding="utf-8") as fh:
+            fh.write(vcard)
+        written += 1
+    return {"written": written, "by_resource": by_resource}
+
+
+def _rsync_vcards(local_dir: str, ssh_host: str, remote_dir: str) -> dict:
+    """rsync de los .vcf del directorio local al working tree remoto de Xandikos.
+
+    Sincroniza SOLO los ``*.vcf`` (``--include='*.vcf' --exclude='*'``), de modo
+    que ningún otro fichero del working tree remoto se toca: ni ``.git/`` (el
+    historial), ni ``.xandikos`` (metadata del tipo de colección), ni
+    ``push-subscriptions.json`` (suscripciones WebDAV-Push), ni un ``.gitignore``.
+    ``--delete`` borra del remoto los .vcf que ya no están en local — como local
+    contiene TODOS los contactos de la DB, esto deja la colección de .vcf
+    EXACTAMENTE igual a la DB (limpia recursos .vcf huérfanos) sin afectar a los
+    ficheros no-.vcf, que quedan protegidos por estar excluidos.
+
+    Returns:
+        {"status":"ok", "stdout":..., "stderr":...} o {"status":"error", "error":...}.
+    """
+    src = local_dir.rstrip("/") + "/"
+    dst = f"{ssh_host}:{remote_dir}"
+    cmd = [
+        "rsync",
+        "-az",
+        "--delete",
+        "--include=*.vcf",
+        "--exclude=*",
+        src,
+        dst,
+    ]
+    try:
+        proc = subprocess.run(
+            cmd, capture_output=True, text=True, timeout=300, check=False
+        )
+    except (OSError, subprocess.TimeoutExpired) as e:
+        return {"status": "error", "error": f"rsync falló: {e}"}
+    if proc.returncode != 0:
+        return {
+            "status": "error",
+            "error": f"rsync rc={proc.returncode}: {proc.stderr.strip()}",
+        }
+    return {"status": "ok", "stdout": proc.stdout, "stderr": proc.stderr}
+
+
+def _git_commit_remote(ssh_host: str, remote_dir: str) -> dict:
+    """UN solo commit en el working tree remoto (lo que Xandikos sirve).
+
+    Hace ``git add -A`` (recoge altas, bajas y modificaciones de .vcf que dejó el
+    rsync) y un único commit con identidad fija ``osint_db``. El ``|| true``
+    evita fallar cuando no hay cambios (commit vacío). Captura el HEAD antes y
+    después para confirmar que SOLO se añadió un commit (o ninguno).
+
+    Returns:
+        {"status":"ok", "head_before":..., "head_after":..., "committed":bool} o
+        {"status":"error", "error":...}.
+    """
+    # rev-parse HEAD antes.
+    head_before = _ssh_capture(ssh_host, f"cd {remote_dir} && git rev-parse HEAD")
+    if head_before.get("status") != "ok":
+        return {"status": "error", "error": f"rev-parse(before): {head_before.get('error')}"}
+
+    script = (
+        f"cd {remote_dir} && git add -A && "
+        "git -c user.email=osint_db -c user.name=osint_db "
+        "commit -m 'bulk push from DuckDB' || true"
+    )
+    commit = _ssh_capture(ssh_host, script)
+    if commit.get("status") != "ok":
+        return {"status": "error", "error": f"git commit: {commit.get('error')}"}
+
+    head_after = _ssh_capture(ssh_host, f"cd {remote_dir} && git rev-parse HEAD")
+    if head_after.get("status") != "ok":
+        return {"status": "error", "error": f"rev-parse(after): {head_after.get('error')}"}
+
+    before = (head_before.get("stdout") or "").strip()
+    after = (head_after.get("stdout") or "").strip()
+    return {
+        "status": "ok",
+        "head_before": before,
+        "head_after": after,
+        "committed": before != after,
+    }
+
+
+def _ssh_capture(ssh_host: str, remote_cmd: str) -> dict:
+    """Ejecuta un comando en el host remoto vía ssh y captura stdout/stderr.
+
+    Usa ``BatchMode=yes`` para fallar rápido si no hay auth por clave (nunca pide
+    contraseña interactiva, que colgaría el service). NO interpola secretos en el
+    comando — solo paths y verbos git fijos.
+    """
+    cmd = [
+        "ssh",
+        "-o",
+        "BatchMode=yes",
+        "-o",
+        "ConnectTimeout=10",
+        ssh_host,
+        remote_cmd,
+    ]
+    try:
+        proc = subprocess.run(
+            cmd, capture_output=True, text=True, timeout=60, check=False
+        )
+    except (OSError, subprocess.TimeoutExpired) as e:
+        return {"status": "error", "error": f"ssh falló: {e}"}
+    if proc.returncode != 0:
+        return {
+            "status": "error",
+            "error": f"ssh rc={proc.returncode}: {proc.stderr.strip()}",
+        }
+    return {"status": "ok", "stdout": proc.stdout, "stderr": proc.stderr}
+
+
+def _capture_etags_after_bulk(
+    cfg: Config, pwd: str, collection: str, by_resource: dict
+) -> dict:
+    """Lee en UN PROPFIND los etags de la colección y los persiste por uid.
+
+    Tras el commit remoto, lista la colección entera (PROPFIND Depth:1 -> [{href,
+    etag}]) y casa cada recurso con su uid usando ``by_resource`` (nombre .vcf ->
+    uid construido al generar los ficheros). Hace un UPDATE de ``contacts.etag``
+    por uid encontrado, dejando los etags guardados == los del servidor, para que
+    el próximo ``/api/sync/dav-pull`` no lo confunda con una edición del móvil.
+
+    Returns:
+        {"status":"ok", "updated":N} o {"status":"error", "error":...}.
+    """
+    listing = dav_list_resources(cfg.dav_base, cfg.dav_user, pwd, collection)
+    if listing.get("status") != "ok":
+        return {
+            "status": "error",
+            "error": f"PROPFIND {collection}: {listing.get('error')} "
+            f"(http {listing.get('http_status')})",
+        }
+    updated = 0
+    for res in listing.get("resources", []):
+        href = res.get("href") or ""
+        etag = res.get("etag")
+        tail = href.rstrip("/").rsplit("/", 1)[-1]
+        uid = by_resource.get(tail)
+        if uid is None or not etag:
+            continue
+        up = duckdb_execute(
+            cfg.db_path, "UPDATE contacts SET etag = ? WHERE uid = ?", [etag, uid]
+        )
+        if up.get("status") == "ok":
+            updated += up.get("rowcount", 0) or 0
+    return {"status": "ok", "updated": updated}
+
+
+def push_all_dav_bulk(cfg: Config) -> dict:
+    """Push masivo DB -> Xandikos por DISCO: 1 rsync + 1 commit + 1 PROPFIND.
+
+    Vía RÁPIDA equivalente a ``push_all_dav`` para los CONTACTOS, pensada para
+    reconciliar las ~1000 fichas de la DB en segundos en vez de minutos. Evita
+    los tres cuellos del push HTTP secuencial:
+
+      - 1 PUT HTTP por contacto  -> 1 rsync de todos los .vcf de golpe.
+      - 1 commit git por PUT     -> 1 solo commit en el working tree remoto.
+      - 1 PROPFIND por contacto  -> 1 PROPFIND de toda la colección al final.
+
+    Flujo:
+      1. Lee TODOS los contacts de la DB (uid, collection, fn, tels, emails,
+         note_path) de la colección CardDAV por defecto.
+      2. Genera el vCard de AGENDA (SIN OSINT) de cada uno en un tmpdir local,
+         con el nombre de recurso EXACTO del push HTTP (``_safe_resource``).
+      3. rsync ``--delete`` ese tmpdir al working tree remoto, sincronizando SOLO
+         los .vcf (protege .git/.xandikos/push-subscriptions.json).
+      4. UN solo commit en el remoto -> Xandikos recalcula el ctag (DAVx5 detecta).
+      5. UN PROPFIND de la colección -> {uid: etag} -> UPDATE de ``contacts.etag``.
+
+    Solo cubre la colección CardDAV por defecto (``cfg.dav_contacts_collection``),
+    donde viven todos los contactos del ecosistema OSINT hoy. Los eventos CalDAV
+    siguen yendo por el push HTTP (``push_all_dav``).
+
+    Requisitos: SSH por clave al host de Xandikos (``cfg.dav_bulk_ssh_host``) con
+    acceso de escritura al working tree (``cfg.dav_bulk_remote_dir``), y ``rsync``
+    instalado en ambos lados. Si no hay SSH, usar ``push_all_dav`` (HTTP) como
+    fallback.
+
+    Returns:
+        {"status":"ok", "written":N, "rsynced":bool, "committed":bool,
+         "etags_updated":N, "head_before":..., "head_after":..., "elapsed_s":F} o
+        {"status":"error", "error":...}.
+    """
+    started = time.monotonic()
+    collection = _default_collection(cfg)
+
+    contacts = duckdb_query_readonly(
+        cfg.db_path,
+        "SELECT uid, collection, fn, tels, emails, note_path FROM contacts "
+        "WHERE collection = ?",
+        [collection],
+        1000000,
+    )
+    if contacts.get("status") != "ok":
+        return {"status": "error", "error": f"lectura contacts: {contacts.get('error')}"}
+    rows = contacts.get("rows", [])
+
+    # 1+2. Genera los .vcf de agenda en un tmpdir local (parte sin red).
+    tmp_dir = tempfile.mkdtemp(prefix="osint_db_bulk_")
+    try:
+        gen = _write_agenda_vcards_to_dir(cfg, rows, tmp_dir)
+        by_resource = gen["by_resource"]
+        written = gen["written"]
+
+        # 3. rsync de todos los .vcf al working tree remoto (solo *.vcf).
+        rs = _rsync_vcards(tmp_dir, cfg.dav_bulk_ssh_host, cfg.dav_bulk_remote_dir)
+        if rs.get("status") != "ok":
+            return {"status": "error", "error": rs.get("error"), "written": written}
+
+        # 4. UN solo commit en el remoto (Xandikos recalcula el ctag).
+        commit = _git_commit_remote(cfg.dav_bulk_ssh_host, cfg.dav_bulk_remote_dir)
+        if commit.get("status") != "ok":
+            return {"status": "error", "error": commit.get("error"), "written": written}
+    finally:
+        shutil.rmtree(tmp_dir, ignore_errors=True)
+
+    # 5. UN PROPFIND -> {uid: etag} -> UPDATE contacts.etag (sync inverso fiable).
+    pwd, err = _resolve_password(cfg)
+    etags_updated = 0
+    etag_error = None
+    if err is None:
+        cap = _capture_etags_after_bulk(cfg, pwd, collection, by_resource)
+        if cap.get("status") == "ok":
+            etags_updated = cap.get("updated", 0)
+        else:
+            etag_error = cap.get("error")
+    else:
+        etag_error = err
+
+    return {
+        "status": "ok",
+        "written": written,
+        "rsynced": True,
+        "committed": commit.get("committed", False),
+        "etags_updated": etags_updated,
+        "etag_error": etag_error,
+        "head_before": commit.get("head_before"),
+        "head_after": commit.get("head_after"),
+        "elapsed_s": round(time.monotonic() - started, 3),
+    }
+
+
 # ---------------------------------------------------------------------------
 # sync inverso Xandikos -> DB (pull incremental por etag)
 # ---------------------------------------------------------------------------

tests/test_osint_db.py

@@ -700,3 +700,146 @@ def test_pull_dav_incremental_por_etag(client, cfg, monkeypatch):
     assert by_uid["c-same"]["etag"] == '"e-same"'
     assert by_uid["c-changed"]["fn"] == "Despues"   # FN del vCard nuevo
     assert by_uid["c-changed"]["etag"] == '"e-new"'  # etag remoto persistido
+
+
+# --- F5: push masivo por disco (1 rsync + 1 commit + 1 PROPFIND) ------------
+
+
+def test_write_agenda_vcards_to_dir_nombres_y_sin_osint(client, cfg, tmp_path):
+    """_write_agenda_vcards_to_dir escribe un .vcf por contacto, sin OSINT.
+
+    Parte LOCAL del push masivo por disco, testeable sin SSH: genera los .vcf en
+    un tmpdir con el nombre de recurso EXACTO del push HTTP (_safe_resource(uid)
+    + '.vcf') y compone el vCard de agenda (con ADR de la persona enlazada, SIN
+    ninguna línea X-OSINT-*).
+    """
+    from server import writes
+
+    client.post("/api/ingest/vault")
+    # Persona con campos OSINT + dirección, contacto enlazado por teléfono.
+    client.post(
+        "/api/person",
+        json={
+            "slug": "disco-osint",
+            "nombre": "Disco Osint",
+            "dni": "11111111H",
+            "sexo": "hombre",
+            "telefonos": ["+34 655 900 900"],
+            "direcciones": ["Av. Disco 7, Málaga"],
+        },
+    )
+    now = datetime.now(tz=timezone.utc)
+    with write_conn(cfg.db_path) as conn:
+        conn.execute(
+            "INSERT INTO contacts VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            [
+                "uid con espacios/raro",  # fuerza saneo del nombre del recurso
+                "/enmanuel/contacts/addressbook/",
+                None,
+                "Disco Osint",
+                '["+34 655 900 900"]',
+                "[]",
+                "BEGIN:VCARD...",
+                None,
+                now,
+            ],
+        )
+    client.post("/api/ingest/vault")  # enlaza el contacto a la ficha
+
+    rows = client.post(
+        "/api/query",
+        json={
+            "sql": "SELECT uid, collection, fn, tels, emails, note_path "
+            "FROM contacts WHERE uid = 'uid con espacios/raro'"
+        },
+    ).json()["rows"]
+
+    out_dir = str(tmp_path / "vcards_out")
+    os.makedirs(out_dir)
+    res = writes._write_agenda_vcards_to_dir(cfg, rows, out_dir)
+
+    assert res["written"] == 1
+    # Nombre de recurso = saneo del uid + .vcf (idéntico al push HTTP).
+    expected_name = writes._safe_resource("uid con espacios/raro") + ".vcf"
+    assert expected_name == "uid_con_espacios_raro.vcf"
+    assert list(res["by_resource"]) == [expected_name]
+    assert res["by_resource"][expected_name] == "uid con espacios/raro"
+
+    written_path = os.path.join(out_dir, expected_name)
+    assert os.path.exists(written_path)
+    content = open(written_path, encoding="utf-8").read()
+    # (a) Privacidad: sin OSINT.
+    assert "X-OSINT-" not in content
+    assert "11111111H" not in content
+    # (b) Agenda: la dirección de la persona enlazada SÍ viaja como ADR.
+    assert "ADR;TYPE=HOME:;;Av. Disco 7\\, Málaga;;;;" in content
+
+
+def test_push_all_dav_bulk_flujo_mockeado(client, cfg, monkeypatch):
+    """push_all_dav_bulk: genera .vcf, rsync+commit (mock) y persiste etags por uid.
+
+    Mockea rsync (subprocess), el commit remoto (HEAD before/after) y el PROPFIND
+    final, verificando: written = nº de contactos, committed True (HEAD cambió) y
+    que contacts.etag queda poblado con el etag del PROPFIND casado por uid.
+    """
+    from server import writes
+
+    client.post("/api/ingest/vault")
+    coll = "/enmanuel/contacts/addressbook/"
+    now = datetime.now(tz=timezone.utc)
+    with write_conn(cfg.db_path) as conn:
+        for uid, fn in [("c-a", "Contacto A"), ("c-b", "Contacto B")]:
+            conn.execute(
+                "INSERT INTO contacts VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                [uid, coll, None, fn, "[]", "[]", "BEGIN:VCARD...", None, now],
+            )
+
+    # rsync + ssh (commit / rev-parse) mockeados a nivel de helper.
+    monkeypatch.setattr(
+        writes, "_rsync_vcards", lambda *a, **k: {"status": "ok", "stdout": "", "stderr": ""}
+    )
+    monkeypatch.setattr(
+        writes,
+        "_git_commit_remote",
+        lambda *a, **k: {
+            "status": "ok",
+            "head_before": "aaaa111",
+            "head_after": "bbbb222",
+            "committed": True,
+        },
+    )
+    monkeypatch.setattr(
+        writes, "pass_get_secret", lambda *_: {"status": "ok", "value": "pw"}
+    )
+
+    # PROPFIND: devuelve el etag de cada recurso por su nombre saneado.
+    def fake_list(base, user, pwd, collection, **kw):
+        return {
+            "status": "ok",
+            "http_status": 207,
+            "resources": [
+                {"href": coll + "c-a.vcf", "etag": '"etag-a"'},
+                {"href": coll + "c-b.vcf", "etag": '"etag-b"'},
+            ],
+        }
+
+    monkeypatch.setattr(writes, "dav_list_resources", fake_list)
+
+    r = writes.push_all_dav_bulk(cfg)
+    assert r["status"] == "ok"
+    assert r["written"] == 2
+    assert r["rsynced"] is True
+    assert r["committed"] is True
+    assert r["head_before"] == "aaaa111"
+    assert r["head_after"] == "bbbb222"
+    assert r["etags_updated"] == 2
+    assert isinstance(r["elapsed_s"], float)
+
+    # Los etags del PROPFIND quedaron persistidos por uid (sync inverso fiable).
+    rows = client.post(
+        "/api/query",
+        json={"sql": "SELECT uid, etag FROM contacts ORDER BY uid"},
+    ).json()["rows"]
+    by_uid = {row["uid"]: row["etag"] for row in rows}
+    assert by_uid["c-a"] == '"etag-a"'
+    assert by_uid["c-b"] == '"etag-b"'