feat(ical): abonnierbarer read-only Kalender-Feed pro Nutzer

Token-gescoper iCal-Feed (/absences/ical/<token>.ics), abonnierbar in
Outlook/Apple/Google. Anders als der CalDAV-Client (Push nach Nextcloud)
pollt der Kalender die URL selbst. Feed zeigt nur die eigenen bestätigten
Abwesenheiten des Token-Inhabers.

- users.ical_token_hash (SHA-256, rotierbar) + Migration 0041 (nur Spalte,
  keine RLS-Aenderung; users-Policy deckt neue nullable Spalte ab)
- Router ical.py: oeffentlicher Feed (kein JWT) + Token-Verwaltung
  POST/GET/DELETE /users/me/ical-token (authentifiziert)
- ProfilePage: Sektion "Kalender-Abo (iCal)" mit Erzeugen/Rotieren/
  Deaktivieren, URL-Anzeige einmalig + Kopieren
- test_ical.py: Token-Lifecycle + oeffentlicher Feed (3 Tests)

Deployed auf 137 (Migration 0041, 196/196 Tests gruen). 164 ausstehend.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-08 16:12:05 +02:00
co-authored by Claude Opus 4.8
parent d27be740a0
commit d13350b38b
6 changed files with 320 additions and 0 deletions
+2
View File
@@ -18,6 +18,7 @@ from app.routers import audit
from app.routers import special_assignments
from app.routers import hours_payouts
from app.routers import public_stamp
from app.routers import ical
from app.routers import reseller, tenants
@@ -97,6 +98,7 @@ app.include_router(users.router, prefix=API_PREFIX)
app.include_router(companies.router, prefix=API_PREFIX)
app.include_router(time_entries.router, prefix=API_PREFIX)
app.include_router(public_stamp.router, prefix=API_PREFIX)
app.include_router(ical.router, prefix=API_PREFIX)
app.include_router(absences.router, prefix=API_PREFIX)
app.include_router(reports.router, prefix=API_PREFIX)
app.include_router(ldap.router, prefix=API_PREFIX)
+4
View File
@@ -67,6 +67,10 @@ class User(Base):
# Kalender-Kürzel (vom Manager setzbar, für CalDAV-Template $kuerzel)
kuerzel: Mapped[str | None] = mapped_column(String(20))
# iCal-Abo: gehashtes rotierbares Token für den read-only Kalender-Feed
# (/absences/ical/<token>.ics). Nur eigene Abwesenheiten, kein JWT nötig.
ical_token_hash: Mapped[str | None] = mapped_column(Text, unique=True)
# Personalnummer (numerisch, eindeutig pro Firma; bleibt nach Deaktivierung reserviert)
personnel_number: Mapped[str | None] = mapped_column(String(50))
+132
View File
@@ -0,0 +1,132 @@
"""iCal-Abo-Feed pro Nutzer (read-only, abonnierbar in Outlook/Apple/Google).
Anders als der CalDAV-Client (der Events aktiv nach Nextcloud *pusht*), stellt
dieser Endpunkt einen abonnierbaren `text/calendar`-Feed bereit: der Kalender
des Nutzers pollt die URL selbst.
Sicherheit / DSGVO:
- Token-gescoped auf GENAU EINEN Nutzer (gehasht in `users.ical_token_hash`).
- Feed enthält ausschließlich die eigenen Abwesenheiten dieses Nutzers.
- Kein JWT (Kalender-Apps können kein Bearer) → Zugang allein über das
geheime, rotierbare Token in der URL. Opt-in: ohne Token kein Feed.
- Read-only, kein DB-Write → RLS-Bypass-Kontext (get_db default) bleibt stabil.
"""
from __future__ import annotations
import hashlib
import secrets
from datetime import date, datetime, timedelta, timezone
from fastapi import APIRouter, Depends, HTTPException, Response
from icalendar import Calendar, Event
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import settings
from app.core.database import get_db
from app.core.dependencies import get_current_user
from app.models.absence import Absence, AbsenceStatus
from app.models.absence_type import AbsenceType
from app.models.user import User
router = APIRouter(tags=["iCal"])
# Nur bestätigte / laufende Anträge im Kalender zeigen keine offenen/abgelehnten.
_FEED_STATUSES = (
AbsenceStatus.APPROVED,
AbsenceStatus.FIRST_APPROVED,
AbsenceStatus.CANCELLATION_REQUESTED,
)
def _hash_token(token: str) -> str:
return hashlib.sha256(token.encode("utf-8")).hexdigest()
def _feed_url(token: str) -> str:
base = settings.frontend_url.rstrip("/")
return f"{base}/api/v1/absences/ical/{token}.ics"
# ── Öffentlicher Feed (kein JWT) ──────────────────────────────────────────────
@router.get("/absences/ical/{token}.ics")
async def ical_feed(token: str, db: AsyncSession = Depends(get_db)) -> Response:
user = await db.scalar(select(User).where(User.ical_token_hash == _hash_token(token)))
if user is None or not user.is_active:
raise HTTPException(status_code=404, detail="Kalender-Feed nicht gefunden.")
rows = (
await db.execute(
select(Absence, AbsenceType.name, AbsenceType.color)
.join(AbsenceType, AbsenceType.id == Absence.type_id)
.where(
Absence.user_id == user.id,
Absence.status.in_(_FEED_STATUSES),
)
.order_by(Absence.start_date)
)
).all()
cal = Calendar()
cal.add("prodid", "-//TimeMaster//iCal Feed//DE")
cal.add("version", "2.0")
cal.add("x-wr-calname", f"TimeMaster {user.first_name} {user.last_name}")
cal.add("x-published-ttl", "PT12H")
now = datetime.now(timezone.utc)
for absence, type_name, _color in rows:
summary = type_name
if absence.status == AbsenceStatus.FIRST_APPROVED:
summary = f"{type_name} (vorläufig)"
elif absence.status == AbsenceStatus.CANCELLATION_REQUESTED:
summary = f"{type_name} (Storno beantragt)"
if absence.half_day_start or absence.half_day_end:
summary += " ½"
ev = Event()
ev.add("uid", f"absence-{absence.id}@timemaster")
ev.add("dtstart", absence.start_date)
ev.add("dtend", absence.end_date + timedelta(days=1)) # DTEND exklusiv
ev.add("summary", summary)
if absence.note:
ev.add("description", absence.note)
ev.add("status", "CONFIRMED")
ev.add("transp", "TRANSPARENT")
ev.add("dtstamp", now)
ev.add("last-modified", absence.created_at or now)
cal.add_component(ev)
return Response(
content=cal.to_ical(),
media_type="text/calendar; charset=utf-8",
headers={"Content-Disposition": 'inline; filename="timemaster.ics"'},
)
# ── Token-Verwaltung (authentifiziert) ────────────────────────────────────────
@router.get("/users/me/ical-token")
async def ical_token_status(current_user: User = Depends(get_current_user)) -> dict:
"""Ist der Feed aktiviert? Das Token selbst wird nie zurückgegeben (nur Hash gespeichert)."""
return {"enabled": current_user.ical_token_hash is not None}
@router.post("/users/me/ical-token")
async def create_or_rotate_ical_token(
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
) -> dict:
"""Erzeugt ein neues Token (rotiert ein bestehendes) und liefert die Feed-URL einmalig zurück."""
token = secrets.token_urlsafe(32)
current_user.ical_token_hash = _hash_token(token)
await db.commit()
return {"enabled": True, "url": _feed_url(token)}
@router.delete("/users/me/ical-token", status_code=204)
async def delete_ical_token(
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
) -> Response:
current_user.ical_token_hash = None
await db.commit()
return Response(status_code=204)
@@ -0,0 +1,29 @@
"""iCal-Abo: users.ical_token_hash
Revision ID: 0041
Revises: 0040
Create Date: 2026-07-08
Gehashtes, rotierbares Token für den read-only iCal-Kalender-Feed pro Nutzer
(/absences/ical/<token>.ics). NULL = Feed deaktiviert. Keine RLS-Änderung
(nur neue nullable Spalte auf bestehender users-Policy).
"""
from alembic import op
revision = "0041"
down_revision = "0040"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.execute("ALTER TABLE users ADD COLUMN IF NOT EXISTS ical_token_hash TEXT")
op.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS ix_users_ical_token_hash "
"ON users (ical_token_hash) WHERE ical_token_hash IS NOT NULL"
)
def downgrade() -> None:
op.execute("DROP INDEX IF EXISTS ix_users_ical_token_hash")
op.execute("ALTER TABLE users DROP COLUMN IF EXISTS ical_token_hash")
+66
View File
@@ -0,0 +1,66 @@
"""Tests für den iCal-Abo-Feed (router/ical.py)."""
import pytest
import pytest_asyncio
from httpx import AsyncClient
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def ical_headers(client: AsyncClient):
resp = await client.post("/api/v1/auth/register", json={
"company_name": "iCal GmbH",
"first_name": "Cal",
"last_name": "Endar",
"email": "admin@icalgmbh.de",
"password": "Secret123",
})
assert resp.status_code == 201, resp.text
return {"Authorization": f"Bearer {resp.json()['access_token']}"}
@pytest.mark.asyncio(loop_scope="session")
async def test_ical_token_disabled_by_default(client: AsyncClient, ical_headers):
r = await client.get("/api/v1/users/me/ical-token", headers=ical_headers)
assert r.status_code == 200
assert r.json()["enabled"] is False
@pytest.mark.asyncio(loop_scope="session")
async def test_ical_feed_lifecycle(client: AsyncClient, ical_headers):
# Aktivieren → URL zurück
r = await client.post("/api/v1/users/me/ical-token", json={}, headers=ical_headers)
assert r.status_code == 200, r.text
url = r.json()["url"]
assert url.endswith(".ics")
assert r.json()["enabled"] is True
# Öffentlicher Feed OHNE Auth erreichbar, liefert VCALENDAR
path = url.split("/api/v1", 1)[1]
feed = await client.get("/api/v1" + path) # kein Authorization-Header
assert feed.status_code == 200
assert "text/calendar" in feed.headers["content-type"]
assert feed.text.startswith("BEGIN:VCALENDAR")
# Status jetzt aktiv
st = await client.get("/api/v1/users/me/ical-token", headers=ical_headers)
assert st.json()["enabled"] is True
# Rotieren → altes Token wird ungültig
old_path = path
r2 = await client.post("/api/v1/users/me/ical-token", json={}, headers=ical_headers)
new_path = r2.json()["url"].split("/api/v1", 1)[1]
assert new_path != old_path
assert (await client.get("/api/v1" + old_path)).status_code == 404
assert (await client.get("/api/v1" + new_path)).status_code == 200
# Deaktivieren → Feed weg
d = await client.delete("/api/v1/users/me/ical-token", headers=ical_headers)
assert d.status_code == 204
assert (await client.get("/api/v1" + new_path)).status_code == 404
st2 = await client.get("/api/v1/users/me/ical-token", headers=ical_headers)
assert st2.json()["enabled"] is False
@pytest.mark.asyncio(loop_scope="session")
async def test_ical_feed_bad_token(client: AsyncClient):
r = await client.get("/api/v1/absences/ical/nonexistenttoken.ics")
assert r.status_code == 404