feat(anhaenge): ATTACH-Zeile im erzeugten ICS

Jeder Anhang wird als ATTACH mit FMTTYPE, SIZE und FILENAME ausgegeben,
als URI statt inline base64 -- bei 10 MB je Datei wuerde sonst bei jeder
Synchronisation der ganze Anhang im ICS-Rumpf mitgeschleppt.

Die absolute URL kommt aus dav_util.public_base, das PUBLIC_BASE_URL und
die Forwarded-Header schon behandelt; ohne Basis-URL wird die Zeile
weggelassen, weil eine relative ATTACH-URL einem externen Client nichts
nuetzt. Abschaltbar ueber ATTACHMENT_PUBLIC_LINKS=0.

Der CalDAV-PUT ist geprueft: ein externer Client, der den Termin
zurueckschreibt, laesst die Anhaenge unberuehrt.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-09-14 10:19:14 +02:00
co-authored by Claude Opus 5
parent 4b9c09df74
commit a024fa54b9
6 changed files with 207 additions and 21 deletions
+23
View File
@@ -21,6 +21,7 @@ and forgets the helper.
import io
import logging
import os
import secrets
import time
import uuid
@@ -75,6 +76,11 @@ _TEXT_TYPES = {
# row is committed, and a sweep landing in between would delete a good file.
SWEEP_GRACE_SECONDS = 3600
# Deployments that would rather lose attachments in external calendar clients
# than hand out unauthenticated URLs set this to 0: no ATTACH line is emitted
# and the capability endpoint 404s.
PUBLIC_LINKS_ENABLED = os.environ.get("ATTACHMENT_PUBLIC_LINKS", "1") != "0"
def sniff(raw: bytes, filename: str) -> tuple[str, str]:
"""Determine ``(content_type, extension)`` from the BYTES.
@@ -243,6 +249,23 @@ def counts_for_events(db: Session, event_ids: Iterable[int]) -> dict[int, int]:
return {event_id: n for event_id, n in rows}
def by_event(db: Session, event_ids: Iterable[int]) -> dict[int, list]:
"""Attachments grouped by event id, in ONE query — for ICS generation."""
ids = [i for i in event_ids if i is not None]
if not ids:
return {}
out: dict[int, list] = {}
rows = (
db.query(models.EventAttachment)
.filter(models.EventAttachment.event_id.in_(ids))
.order_by(models.EventAttachment.id)
.all()
)
for row in rows:
out.setdefault(row.event_id, []).append(row)
return out
def sweep(db: Session) -> tuple[int, int]:
"""Drop attachments whose event is gone, and files with no row.
+28 -3
View File
@@ -9,9 +9,11 @@ from __future__ import annotations
import logging
import uuid
from urllib.parse import quote
from datetime import date, datetime, timedelta, timezone
from icalendar import Calendar, Event, vCalAddress, vRecur, vText
from icalendar import Calendar, Event, vCalAddress, vRecur, vText, vUri
from icalendar.parser import Parameters
logger = logging.getLogger(__name__)
@@ -150,8 +152,17 @@ def _parse_iso(s: str) -> datetime:
return dt
def build_ics(calendar, events, *, name_cache: dict | None = None) -> str:
"""Build a VCALENDAR string for a local calendar and its events."""
def build_ics(calendar, events, *, name_cache: dict | None = None,
attachments_by_event: dict | None = None,
base_url: str | None = None) -> str:
"""Build a VCALENDAR string for a local calendar and its events.
``attachments_by_event`` maps LocalEvent.id -> list of EventAttachment. With
a ``base_url`` each one becomes an ATTACH property pointing at its
capability URL, so external clients (Apple Calendar, Thunderbird) can fetch
it — they cannot send a bearer token. Both arguments are optional so every
existing caller keeps working and emits no ATTACH at all.
"""
cal = Calendar()
cal.add("prodid", "-//Calendarr//EN")
cal.add("version", "2.0")
@@ -200,6 +211,20 @@ def build_ics(calendar, events, *, name_cache: dict | None = None) -> str:
organizer.params["CN"] = vText(organizer_name.replace('"', ""))
item.add("organizer", organizer)
# ATTACH: one line per attachment, as a URI rather than inline base64 —
# the files are up to 10 MB each and would otherwise be copied into
# every ICS body on every sync.
for att in (attachments_by_event or {}).get(getattr(ev, "id", None), ()):
if not base_url:
continue # a relative ATTACH URL is useless to an external client
uri = vUri(f"{base_url}/api/attach/{att.token}/{quote(att.filename)}")
uri.params = Parameters({
"FMTTYPE": att.content_type,
"SIZE": str(att.size_bytes),
"FILENAME": att.filename,
})
item.add("attach", uri, encode=0)
cal.add_component(item)
return cal.to_ical().decode("utf-8")
+1 -8
View File
@@ -10,7 +10,6 @@ Two routers live here:
send a bearer token. See the README for the tradeoff.
"""
import os
from urllib.parse import quote
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
@@ -29,12 +28,6 @@ from database import get_db
router = APIRouter()
public_router = APIRouter()
# Deployments that would rather lose attachments in external calendar clients
# than hand out unauthenticated URLs can set this to 0: the capability endpoint
# then 404s and no ATTACH line is emitted into any ICS.
PUBLIC_LINKS_ENABLED = os.environ.get("ATTACHMENT_PUBLIC_LINKS", "1") != "0"
def _to_dict(att: models.EventAttachment) -> dict:
uploader = None
if att.uploader is not None:
@@ -261,7 +254,7 @@ def public_attachment(token: str, name: str = "", db: Session = Depends(get_db))
Never distinguish "wrong token" from "deleted": both are 404.
"""
if not PUBLIC_LINKS_ENABLED:
if not attachments_store.PUBLIC_LINKS_ENABLED:
raise HTTPException(404, "Not found")
att = (
db.query(models.EventAttachment)
+19 -9
View File
@@ -160,8 +160,15 @@ def _name_cache(cal: models.LocalCalendar, db: Session) -> dict:
return {}
def _build_ics(cal: models.LocalCalendar, evs: list[models.LocalEvent], db: Session) -> str:
return ical_io.build_ics(cal, evs, name_cache=_name_cache(cal, db))
def _build_ics(cal: models.LocalCalendar, evs: list[models.LocalEvent], db: Session,
base_url: str | None = None) -> str:
# Without a base URL (or with public links switched off) no ATTACH is
# emitted: a relative attachment URL is useless to an external client.
atts = None
if base_url and attachments_store.PUBLIC_LINKS_ENABLED:
atts = attachments_store.by_event(db, [e.id for e in evs])
return ical_io.build_ics(cal, evs, name_cache=_name_cache(cal, db),
attachments_by_event=atts, base_url=base_url)
# ── XML builders ──────────────────────────────────────────
@@ -277,7 +284,8 @@ def _handle_propfind(cal: models.LocalCalendar, base: str, resource: str, depth:
return _multistatus("\n".join(parts))
def _handle_report(cal: models.LocalCalendar, base: str, body: bytes, db: Session) -> Response:
def _handle_report(cal: models.LocalCalendar, base: str, body: bytes, db: Session,
base_url: str | None = None) -> Response:
report_type = None
hrefs: list[str] = []
if body:
@@ -297,17 +305,17 @@ def _handle_report(cal: models.LocalCalendar, base: str, body: bytes, db: Sessio
parts = []
for ev in evs:
ics = _build_ics(cal, [ev], db)
ics = _build_ics(cal, [ev], db, base_url)
parts.append(_event_propstat(base, ev, with_data=True, ics=ics))
return _multistatus("\n".join(parts) if parts else "")
def _handle_get(cal: models.LocalCalendar, resource: str, db: Session,
*, head: bool = False) -> Response:
*, head: bool = False, base_url: str | None = None) -> Response:
ev = _find_event(cal, resource, db)
if not ev:
return Response(status_code=404)
ics = _build_ics(cal, [ev], db)
ics = _build_ics(cal, [ev], db, base_url)
headers = {"ETag": f'"{_etag(ev)}"'}
return Response(
content=b"" if head else ics,
@@ -390,13 +398,15 @@ async def _dispatch_collection(request: Request, cal: models.LocalCalendar, base
depth = request.headers.get("Depth", "0")
return _handle_propfind(cal, base, resource, depth, db,
principal_href=principal_href, home_href=home_href)
# The public origin as clients actually reach us, for ATTACH URLs.
base_url = dav_util.public_base(request)
if method == "REPORT":
return _handle_report(cal, base, await request.body(), db)
return _handle_report(cal, base, await request.body(), db, base_url)
if method in ("GET", "HEAD"):
if not resource:
ics = _build_ics(cal, _events(cal, db), db)
ics = _build_ics(cal, _events(cal, db), db, base_url)
return Response(content=ics, media_type="text/calendar; charset=utf-8")
return _handle_get(cal, resource, db, head=(method == "HEAD"))
return _handle_get(cal, resource, db, head=(method == "HEAD"), base_url=base_url)
if method == "PUT":
return _handle_put(cal, resource, await request.body(), db)
if method == "DELETE":
+7 -1
View File
@@ -772,6 +772,7 @@ async def import_generic(
@router.get("/calendars/{calendar_id}/export")
def export_calendar(
calendar_id: int,
request: Request,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user),
):
@@ -783,7 +784,12 @@ def export_calendar(
)
# Resolve creator display names for ORGANIZER.
name_cache = {u.id: (u.display_name or u.username) for u in db.query(models.User).all()}
ics = ical_io.build_ics(cal, events, name_cache=name_cache)
base_url = dav_util.public_base(request)
atts = None
if base_url and attachments_store.PUBLIC_LINKS_ENABLED:
atts = attachments_store.by_event(db, [e.id for e in events])
ics = ical_io.build_ics(cal, events, name_cache=name_cache,
attachments_by_event=atts, base_url=base_url)
safe_name = "".join(c for c in cal.name if c.isalnum() or c in (" ", "-", "_")).strip() or "calendar"
return Response(
content=ics,
+129
View File
@@ -516,3 +516,132 @@ def test_busy_masked_event_reports_no_attachments(client):
masked = next(e for e in events if e["id"] == ev["id"])
assert masked["title"] == "Beschäftigt"
assert masked["attachment_count"] == 0
# ── ATTACH in the generated ICS ───────────────────────────
def _publish(client, token, cal_id):
"""Publish the calendar over CalDAV and return its secret token URL path."""
r = client.put(f"/api/local/calendars/{cal_id}", headers=auth(token),
json={"caldav_published": True})
assert r.status_code == 200, r.text
cals = client.get("/api/local/calendars", headers=auth(token)).json()
url = next(c["caldav_url"] for c in cals if c["id"] == cal_id)
return "/dav/" + url.rstrip("/").rsplit("/", 1)[-1] + "/"
def _attach_props(ics_text):
"""Re-parse instead of grepping: icalendar folds lines at 75 octets, so a
naive substring check would miss a long URL."""
from icalendar import Calendar
cal = Calendar.from_ical(ics_text)
out = []
for comp in cal.walk("VEVENT"):
prop = comp.get("attach")
if prop is None:
continue
for p in (prop if isinstance(prop, list) else [prop]):
out.append((str(p), dict(p.params)))
return out
def test_ics_carries_an_attach_line_per_attachment(client):
token = register_admin(client)
cal_id = _make_calendar(client, token)
ev = _make_event(client, token, cal_id)
_upload(client, token, ev["id"], name="Mietvertrag.pdf")
_upload(client, token, ev["id"], png_bytes(), "foto.png", "image/png")
dav_path = _publish(client, token, cal_id)
props = _attach_props(client.get(dav_path).text)
assert len(props) == 2
by_name = {params["FILENAME"]: (uri, params) for uri, params in props}
assert set(by_name) == {"Mietvertrag.pdf", "foto.png"}
uri, params = by_name["Mietvertrag.pdf"]
assert params["FMTTYPE"] == "application/pdf"
assert int(params["SIZE"]) == len(PDF_BYTES)
assert "/api/attach/" in uri and uri.startswith("http")
def test_attach_url_from_the_ics_actually_serves_the_file(client):
"""The whole point: an external client follows the URL with no credentials."""
token = register_admin(client)
cal_id = _make_calendar(client, token)
ev = _make_event(client, token, cal_id)
_upload(client, token, ev["id"])
dav_path = _publish(client, token, cal_id)
uri, _ = _attach_props(client.get(dav_path).text)[0]
path = uri.split("testserver", 1)[1]
r = client.get(path) # no Authorization header
assert r.status_code == 200
assert r.content == PDF_BYTES
def test_attach_url_honours_public_base_url(client, monkeypatch):
"""Behind a reverse proxy the app only sees its internal origin."""
monkeypatch.setenv("PUBLIC_BASE_URL", "https://cal.example.com")
token = register_admin(client)
cal_id = _make_calendar(client, token)
ev = _make_event(client, token, cal_id)
_upload(client, token, ev["id"])
dav_path = _publish(client, token, cal_id)
uri, _ = _attach_props(client.get(dav_path).text)[0]
assert uri.startswith("https://cal.example.com/api/attach/")
def test_no_attach_when_public_links_are_switched_off(client, monkeypatch):
monkeypatch.setattr(attachments_store, "PUBLIC_LINKS_ENABLED", False)
token = register_admin(client)
cal_id = _make_calendar(client, token)
ev = _make_event(client, token, cal_id)
_upload(client, token, ev["id"])
dav_path = _publish(client, token, cal_id)
assert _attach_props(client.get(dav_path).text) == []
def test_events_without_attachments_get_no_attach(client):
token = register_admin(client)
cal_id = _make_calendar(client, token)
_make_event(client, token, cal_id)
dav_path = _publish(client, token, cal_id)
assert _attach_props(client.get(dav_path).text) == []
def test_upload_changes_the_event_etag(client):
"""The ICS body changes, so CalDAV clients must be told to re-fetch."""
token = register_admin(client)
cal_id = _make_calendar(client, token)
ev = _make_event(client, token, cal_id)
db = SessionLocal()
before = db.query(models.LocalEvent).filter(
models.LocalEvent.uid == ev["id"]).first().etag
db.close()
_upload(client, token, ev["id"])
db = SessionLocal()
after = db.query(models.LocalEvent).filter(
models.LocalEvent.uid == ev["id"]).first().etag
db.close()
assert after != before
def test_caldav_put_leaves_attachments_alone(client):
"""An external client rewriting the event must not drop its attachments."""
token = register_admin(client)
cal_id = _make_calendar(client, token)
ev = _make_event(client, token, cal_id)
_upload(client, token, ev["id"])
dav_path = _publish(client, token, cal_id)
ics = client.get(dav_path).text
r = client.put(f"{dav_path}{ev['id']}.ics", content=ics.encode("utf-8"),
headers={"Content-Type": "text/calendar"})
assert r.status_code in (201, 204), r.text
assert _row_count() == 1
assert len(_attach_props(client.get(dav_path).text)) == 1