From 78fc275fe7ac345118b24cd36d32237f23d97583 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Schotte?= Date: Wed, 7 Oct 2026 05:28:22 +0200 Subject: [PATCH 1/6] Talk: join link, conversation password and lobby MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit create_conversation and get_conversation return url, the link to join (/index.php/call/, works without pretty URLs). set_conversation_password sets the password guests need for a public conversation; removing it needs the destructive level; the password is never echoed, also not in errors. set_conversation_lobby turns the lobby on (optionally until a future time) or off. Signed-off-by: Björn Schotte Co-Authored-By: Claude Opus 5.5 --- README.md | 10 ++-- src/nc_mcp_server/tools/talk.py | 86 +++++++++++++++++++++++++++++++-- tests/test_talk_admin.py | 83 +++++++++++++++++++++++++++++++ 3 files changed, 171 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 4c7d152..964eb6b 100644 --- a/README.md +++ b/README.md @@ -30,9 +30,9 @@ export NEXTCLOUD_PASSWORD=your-app-password nc-mcp-server ``` -## 222 Tools Across 24 Nextcloud Apps +## 224 Tools Across 24 Nextcloud Apps -A 223rd tool, `upload_file_from_path`, is registered only when the operator sets +A 225th tool, `upload_file_from_path`, is registered only when the operator sets `NEXTCLOUD_MCP_UPLOAD_ROOT`. See [Files](#files) for details. | Category | Tools | Protocol | @@ -314,7 +314,7 @@ level, as does disabling an account with `set_user_enabled`. | Tool | Permission | Description | |------|-----------|-------------| | `list_conversations` | read | List all Talk conversations | -| `get_conversation` | read | Get conversation details | +| `get_conversation` | read | Get conversation details, with the link to join it | | `get_messages` | read | Get messages from a conversation, or from one thread | | `get_participants` | read | List participants in a conversation | | `list_threads` | read | List the most recently active threads in a conversation | @@ -328,7 +328,9 @@ level, as does disabling an account with `set_user_enabled`. | `list_conversation_tags` | read | List your personal conversation tags | | `list_conversation_presets` | read | List the presets create_conversation can start from | | `send_message` | write | Send a message; can start a thread or post into one | -| `create_conversation` | write | Create a one-to-one, group or public conversation, optionally from a preset | +| `create_conversation` | write | Create a one-to-one, group or public conversation, optionally from a preset; returns the link to join it | +| `set_conversation_password` | write | Set the password guests need for a public conversation (moderators); removing it needs `destructive` | +| `set_conversation_lobby` | write | Turn the lobby on, optionally until a time, or off (moderators) | | `update_conversation` | write | Rename, describe, lock (read-only) or open (public) a conversation (moderators); making it private needs `destructive`; owners can preserve it (Talk 25+) | | `add_participant` | write | Add a user, group, team, email guest or federated user (moderators) | | `set_participant_role` | write | Make a participant owner (Talk 25+), moderator or user | diff --git a/src/nc_mcp_server/tools/talk.py b/src/nc_mcp_server/tools/talk.py index 358a48e..83e882d 100644 --- a/src/nc_mcp_server/tools/talk.py +++ b/src/nc_mcp_server/tools/talk.py @@ -125,6 +125,17 @@ def _format_conversation(room: dict[str, Any]) -> dict[str, Any]: } +def _with_link(room: dict[str, Any]) -> dict[str, Any]: + """The formatted conversation plus its join link (url), as Talk shows it under "Copy link". + + index.php keeps the link working on servers without pretty URLs. Guests can only open it + for public conversations; for others it works for participants who are logged in. + """ + out = _format_conversation(room) + out["url"] = f"{get_client().base_url.rstrip('/')}/index.php/call/{quote(room['token'], safe='')}" + return out + + _PLACEHOLDER = re.compile(r"\{([A-Za-z0-9_-]+)\}") @@ -368,11 +379,11 @@ async def get_conversation(token: str) -> str: Use list_conversations to find tokens. Returns: - JSON object with conversation details. + JSON object with conversation details, including url, the link to join it. """ client = get_client() data = await client.ocs_get(f"apps/spreed/api/v4/room/{token}") - return json.dumps(_format_conversation(data), default=str) + return json.dumps(_with_link(data), default=str) @mcp.tool(annotations=READONLY) @require_permission(PermissionLevel.READ) @@ -648,7 +659,8 @@ async def create_conversation( conversation. An explicit room_type still wins over the preset's. Returns: - JSON object with the created conversation details, including its token. + JSON object with the created conversation details, including its token + and url, the link to join it. """ if room_type not in _VALID_ROOM_TYPES: valid = ", ".join(f"{k} ({v})" for k, v in _VALID_ROOM_TYPES.items()) @@ -665,7 +677,7 @@ async def create_conversation( # Talk only records which preset was used; its settings have to be sent along post_data = {**await _preset_settings(preset), **post_data, "preset": preset} data = await client.ocs_post("apps/spreed/api/v4/room", data=post_data) - return json.dumps(_format_conversation(data), default=str) + return json.dumps(_with_link(data), default=str) @mcp.tool(annotations=DESTRUCTIVE) @require_permission(PermissionLevel.DESTRUCTIVE) @@ -1521,6 +1533,71 @@ async def update_conversation( return json.dumps(_format_conversation(room), ensure_ascii=False) +def _register_access_tools(mcp: FastMCP) -> None: + @mcp.tool(annotations=ADDITIVE_IDEMPOTENT) + @require_permission(PermissionLevel.WRITE) + async def set_conversation_password(token: str, password: str) -> str: + """Set or remove the password of a public conversation. Needs moderator rights. + + Guests who join through the link have to enter it; users of this Nextcloud + who are participants do not. Only public conversations have a password. + + Args: + token: The conversation token. + password: The new password; Talk checks it against the password policy + of the instance. An empty string removes the password, which needs + the destructive permission level. + + Returns: + Confirmation message (never the password). + """ + current = get_permission_level() + if not password and not current.includes(PermissionLevel.DESTRUCTIVE): + raise PermissionDeniedError( + "set_conversation_password without a password", PermissionLevel.DESTRUCTIVE, current + ) + try: + await get_client().ocs_put(f"{_ROOM_API}/{token}/password", data={"password": password}) + except NextcloudError as e: + if e.status_code != 400: + raise + # Talk names the policy rule in the message when it has one; never echo the password + raise NextcloudError( + f"{e}: only public conversations have a password, and it must match the password policy", 400 + ) from e + return f"Password {'set' if password else 'removed'} for {token}." + + @mcp.tool(annotations=ADDITIVE_IDEMPOTENT) + @require_permission(PermissionLevel.WRITE) + async def set_conversation_lobby(token: str, enabled: bool, open_at: str = "") -> str: + """Turn the lobby of a group or public conversation on or off. Needs moderator rights. + + With the lobby on, only moderators can join the call and read the chat; + everyone else waits until a moderator turns it off or open_at is reached. + + Args: + token: The conversation token. + enabled: True to turn the lobby on, False to open the conversation. + open_at: Only with enabled: when Talk opens the lobby by itself, as an + ISO 8601 time with time zone in the future, e.g. + "2026-10-12T19:00:00+02:00" (default: stays closed until a + moderator opens it). + + Returns: + JSON with the conversation afterwards, including lobby and lobby_opens_at. + """ + if open_at and not enabled: + raise ValueError("open_at only goes with enabled=true.") + body: dict[str, Any] = {"state": 1 if enabled else 0} + if open_at: + body["timer"] = _timestamp(open_at, "open_at") + data = await get_client().ocs_put(f"{_ROOM_API}/{token}/webinar/lobby", data=body) + room = _with_link(data) + room["lobby"] = data.get("lobbyState", 0) == 1 + room["lobby_opens_at"] = _iso(data.get("lobbyTimer")) + return json.dumps(room, ensure_ascii=False) + + def _register_participant_admin_tools(mcp: FastMCP) -> None: @mcp.tool(annotations=ADDITIVE_IDEMPOTENT) @require_permission(PermissionLevel.WRITE) @@ -1771,6 +1848,7 @@ def register(mcp: FastMCP) -> None: _register_pin_and_reminder_tools(mcp) _register_pin_and_reminder_destructive_tools(mcp) _register_conversation_admin_tools(mcp) + _register_access_tools(mcp) _register_participant_admin_tools(mcp) _register_conversation_admin_destructive_tools(mcp) _register_tag_and_preset_tools(mcp) diff --git a/tests/test_talk_admin.py b/tests/test_talk_admin.py index 6b6f23c..f34b275 100644 --- a/tests/test_talk_admin.py +++ b/tests/test_talk_admin.py @@ -243,3 +243,86 @@ async def test_delete_one_to_one_explained(self, mcp: FastMCP, client: MagicMock client.ocs_delete.side_effect = NextcloudError("OCS DELETE x: HTTP 400", 400) with pytest.raises(ToolError, match="cannot be deleted; one-to-one ones can only be left"): await _call(mcp, "delete_conversation", token="tok") + + +class TestJoinLink: + async def test_create_returns_the_link(self, mcp: FastMCP, client: MagicMock) -> None: + client.base_url = "https://cloud.example/" + client.ocs_post.return_value = {"token": "ab12cd34", "type": 3} + data = json.loads(await _call(mcp, "create_conversation", room_type=3, name="Meeting")) + assert data["url"] == "https://cloud.example/index.php/call/ab12cd34" + assert data["type"] == "public" + + async def test_get_returns_the_link(self, mcp: FastMCP, client: MagicMock) -> None: + client.base_url = "https://cloud.example" + client.ocs_get.return_value = {"token": "ab12cd34", "type": 2} + data = json.loads(await _call(mcp, "get_conversation", token="ab12cd34")) + assert data["url"] == "https://cloud.example/index.php/call/ab12cd34" + + async def test_token_is_quoted(self, mcp: FastMCP, client: MagicMock) -> None: + client.base_url = "https://cloud.example" + client.ocs_get.return_value = {"token": "a/b?c", "type": 2} + data = json.loads(await _call(mcp, "get_conversation", token="x")) + assert data["url"] == "https://cloud.example/index.php/call/a%2Fb%3Fc" + + +class TestPassword: + async def test_set(self, mcp: FastMCP, client: MagicMock) -> None: + set_permission_level(PermissionLevel.WRITE) + result = await _call(mcp, "set_conversation_password", token="tok", password="geheim-123") + assert result == "Password set for tok." + assert "geheim" not in result + client.ocs_put.assert_awaited_once_with(f"{ROOM}/password", data={"password": "geheim-123"}) + + async def test_removing_needs_the_destructive_level(self, mcp: FastMCP, client: MagicMock) -> None: + set_permission_level(PermissionLevel.WRITE) + with pytest.raises(ToolError, match="destructive"): + await _call(mcp, "set_conversation_password", token="tok", password="") + client.ocs_put.assert_not_awaited() + + async def test_remove(self, mcp: FastMCP, client: MagicMock) -> None: + assert await _call(mcp, "set_conversation_password", token="tok", password="") == "Password removed for tok." + client.ocs_put.assert_awaited_once_with(f"{ROOM}/password", data={"password": ""}) + + async def test_refusal_is_explained_without_the_password(self, mcp: FastMCP, client: MagicMock) -> None: + client.ocs_put.side_effect = NextcloudError("OCS PUT x: HTTP 400", 400) + with pytest.raises(ToolError, match="only public conversations have a password") as e: + await _call(mcp, "set_conversation_password", token="tok", password="geheim-123") + assert "geheim" not in str(e.value) + + async def test_read_level_cannot_set(self, mcp: FastMCP, client: MagicMock) -> None: + set_permission_level(PermissionLevel.READ) + with pytest.raises(ToolError, match="write"): + await _call(mcp, "set_conversation_password", token="tok", password="x") + client.ocs_put.assert_not_awaited() + + +class TestLobby: + async def test_on_with_time(self, mcp: FastMCP, client: MagicMock) -> None: + client.base_url = "https://cloud.example" + client.ocs_put.return_value = {"token": "tok", "type": 3, "lobbyState": 1, "lobbyTimer": 4102444800} + data = json.loads( + await _call(mcp, "set_conversation_lobby", token="tok", enabled=True, open_at="2100-01-01T00:00:00Z") + ) + client.ocs_put.assert_awaited_once_with(f"{ROOM}/webinar/lobby", data={"state": 1, "timer": 4102444800}) + assert data["lobby"] is True + assert data["lobby_opens_at"] == "2100-01-01T00:00:00+00:00" + assert data["url"] == "https://cloud.example/index.php/call/tok" + + async def test_off(self, mcp: FastMCP, client: MagicMock) -> None: + client.base_url = "https://cloud.example" + client.ocs_put.return_value = {"token": "tok", "type": 3, "lobbyState": 0, "lobbyTimer": 0} + data = json.loads(await _call(mcp, "set_conversation_lobby", token="tok", enabled=False)) + client.ocs_put.assert_awaited_once_with(f"{ROOM}/webinar/lobby", data={"state": 0}) + assert data["lobby"] is False + assert data["lobby_opens_at"] == "" + + async def test_time_only_with_lobby_on(self, mcp: FastMCP, client: MagicMock) -> None: + with pytest.raises(ToolError, match="only goes with enabled"): + await _call(mcp, "set_conversation_lobby", token="tok", enabled=False, open_at="2100-01-01T00:00:00Z") + client.ocs_put.assert_not_awaited() + + async def test_past_time_refused(self, mcp: FastMCP, client: MagicMock) -> None: + with pytest.raises(ToolError, match="in the future"): + await _call(mcp, "set_conversation_lobby", token="tok", enabled=True, open_at="2020-01-01T00:00:00Z") + client.ocs_put.assert_not_awaited() From 6a94de9a712e9e32b64d54342b0eb1f5e3f4fd8b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Schotte?= Date: Wed, 7 Oct 2026 08:08:50 +0200 Subject: [PATCH 2/6] OCS 998 on an app path: say the app is not available for this account MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nextcloud answers 998 ("Invalid query, please check the syntax") when no route matches. For apps// that means the app does not serve this account: not installed, disabled, or limited to groups the user is not in (seen with Talk restricted to some groups). The old message misled the caller; app-level 404s keep their own message. Signed-off-by: Björn Schotte Co-Authored-By: Claude Opus 5.5 --- src/nc_mcp_server/client.py | 21 +++++++++++++++++++++ tests/test_client_errors.py | 27 +++++++++++++++++++++++++++ 2 files changed, 48 insertions(+) diff --git a/src/nc_mcp_server/client.py b/src/nc_mcp_server/client.py index 25e4280..513deb2 100644 --- a/src/nc_mcp_server/client.py +++ b/src/nc_mcp_server/client.py @@ -3,6 +3,7 @@ import asyncio import contextlib import logging +import re import xml.etree.ElementTree as ET from collections.abc import AsyncIterable, Callable from typing import Any, cast @@ -60,6 +61,8 @@ def _raise_for_ocs_status(response: niquests.Response, context: str = "") -> Non prefix = f"{context}: " if context else "" try: ocs = response.json()["ocs"] + if ocs["meta"].get("statuscode") == _OCS_NO_ROUTE: + raise NextcloudError(f"{prefix}{_no_route_message(context)}", code) ocs_message: str = ( ocs["meta"]["message"] or _ocs_field_errors(ocs.get("data")) @@ -76,6 +79,24 @@ def _raise_for_ocs_status(response: niquests.Response, context: str = "") -> Non raise NextcloudError(f"{prefix}{detail}", code) +# OCS answers 998 when no route matches the path. For an app path that means the app does not +# serve this account: not installed, disabled, or limited to groups the user is not in. Its own +# message ("Invalid query, please check the syntax...") misleads; a missing object inside an app +# comes back with the app's own status instead. +_OCS_NO_ROUTE = 998 +_APP_PATH = re.compile(r"\bapps/([A-Za-z0-9_-]+)/") + + +def _no_route_message(context: str) -> str: + match = _APP_PATH.search(context) + if match is None: + return "Not found. This Nextcloud has no such API endpoint." + return ( + f"The '{match.group(1)}' app is not available for this Nextcloud account" + " (not installed, disabled, or limited to groups the account is not in)" + ) + + _CONFIRMATION_REQUIRED = "Password confirmation is required" _CONFIRMATION_HINT = ( ". Nextcloud does not count an app password as a confirmed password for this action; use the" diff --git a/tests/test_client_errors.py b/tests/test_client_errors.py index eb86d06..affe984 100644 --- a/tests/test_client_errors.py +++ b/tests/test_client_errors.py @@ -196,3 +196,30 @@ def test_no_usable_message(self, data: Any) -> None: body["ocs"]["data"] = data with pytest.raises(NextcloudError, match=r"^HTTP 400$"): _raise_for_ocs_status(_fake_response(400, body)) + + +class TestOcsNoRoute: + """OCS 998 on an app path: the app does not serve this account (e.g. Talk limited to groups).""" + + _BODY = _ocs_error_body( + "Invalid query, please check the syntax. API specifications are here: " + "http://www.freedesktop.org/wiki/Specifications/open-collaboration-services.\n", + statuscode=998, + ) + + def test_app_path_names_the_unavailable_app(self) -> None: + with pytest.raises(NextcloudError) as exc_info: + _raise_for_ocs_status(_fake_response(404, self._BODY), "OCS POST apps/spreed/api/v4/room") + message = str(exc_info.value) + assert message.startswith("OCS POST apps/spreed/api/v4/room: The 'spreed' app is not available") + assert "Invalid query" not in message + assert exc_info.value.status_code == 404 + + def test_path_outside_apps_says_not_found(self) -> None: + with pytest.raises(NextcloudError, match="no such API endpoint"): + _raise_for_ocs_status(_fake_response(404, self._BODY), "OCS GET cloud/nothing") + + def test_app_level_404_keeps_its_message(self) -> None: + body = _ocs_error_body("Conversation not found", statuscode=404) + with pytest.raises(NextcloudError, match="Conversation not found"): + _raise_for_ocs_status(_fake_response(404, body), "OCS GET apps/spreed/api/v4/room/abc") From 94047c4db052a1e1f0f9b8015da605b615294ed1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Schotte?= Date: Wed, 7 Oct 2026 08:41:00 +0200 Subject: [PATCH 3/6] Calendar: time zones, checked RRULE, video link, creator guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - NEXTCLOUD_MCP_TIMEZONE and a timezone argument: times without offset are wall time in that zone; events get TZID + VTIMEZONE (recurring events keep their wall time across DST). get_events accepts offsets in the range. - RRULE is checked (DAILY..YEARLY, known parts, COUNT/INTERVAL <= 1000, not COUNT and UNTIL together); UNTIL converted to UTC or a date. - conference_url: CONFERENCE (RFC 7986) plus LOCATION when empty, else a line in the description; update can replace or remove it. - create_event marks X-NC-MCP-CREATED-BY; update_event refuses events of others or with attendees unless allow_foreign=true; expected_summary on update and delete. Output has has_attendees/created_by_me, never attendee addresses. - Moving only the start keeps the duration; end before start is refused. Tests: 43 unit, 6 integration against Nextcloud 34 (shared calendar _shared_by_, read-only sharee cannot write). Signed-off-by: Björn Schotte Co-Authored-By: Claude Opus 5.5 --- README.md | 16 +- src/nc_mcp_server/config.py | 15 + src/nc_mcp_server/tools/calendar.py | 439 ++++++++++++++++++++--- tests/integration/test_calendar_guard.py | 258 +++++++++++++ tests/test_calendar_helpers.py | 216 +++++++++++ 5 files changed, 891 insertions(+), 53 deletions(-) create mode 100644 tests/integration/test_calendar_guard.py create mode 100644 tests/test_calendar_helpers.py diff --git a/README.md b/README.md index 964eb6b..e5c6ebb 100644 --- a/README.md +++ b/README.md @@ -386,10 +386,24 @@ instead of the placeholders Talk stores (`{mention-user1}`, `{file}`). | `list_calendars` | read | List user's calendars | | `get_events` | read | Get events from a calendar (with date filtering) | | `get_event` | read | Get a single event by UID | -| `create_event` | write | Create a calendar event | +| `create_event` | write | Create a calendar event (time zone, checked RRULE, video link) | | `update_event` | write | Update an event (partial updates supported) | | `delete_event` | destructive | Delete a calendar event | +- Time zone: times without an offset are wall time in `timezone` (tool argument), else + `NEXTCLOUD_MCP_TIMEZONE` (IANA name, e.g. `Europe/Berlin`), else UTC. Zoned events get a `TZID` and a + `VTIMEZONE`, so weekly events stay at 19:00 across daylight saving changes. Moving only the start keeps the + duration. +- RRULE: only `DAILY`/`WEEKLY`/`MONTHLY`/`YEARLY`, known parts, `COUNT`/`INTERVAL` at most 1000, never `COUNT` + and `UNTIL` together; `UNTIL` is converted to UTC (or a date for all-day events). +- Video link: `conference_url` is stored as `CONFERENCE` (RFC 7986) and as `LOCATION` when empty (as Nextcloud + Calendar does for Talk rooms), else as a `Video call: ` line in the description. +- Creator guard: `create_event` marks events with `X-NC-MCP-CREATED-BY:`. `update_event` refuses events + without the caller's mark, or with attendees (Nextcloud would notify them), unless `allow_foreign=true`, so a + client can ask its user first. `update_event`/`delete_event` take `expected_summary` and refuse to act if the + title differs. Event output never lists attendees or organizers, only `has_attendees` and `created_by_me`. +- Shared calendars appear as `_shared_by_` for the people they are shared with. + ### Contacts | Tool | Permission | Description | diff --git a/src/nc_mcp_server/config.py b/src/nc_mcp_server/config.py index fa2dbd7..a1de6f0 100644 --- a/src/nc_mcp_server/config.py +++ b/src/nc_mcp_server/config.py @@ -3,6 +3,7 @@ import os from dataclasses import dataclass, field from pathlib import Path +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from .permissions import PermissionLevel @@ -25,6 +26,8 @@ class Config: NEXTCLOUD_MCP_UPLOAD_ROOT: Absolute path to a local directory. When set, enables the upload_file_from_path tool, restricted to files under this directory (symlinks are resolved before the containment check). Unset by default — tool disabled. + NEXTCLOUD_MCP_TIMEZONE: IANA time zone (e.g. Europe/Berlin) for calendar times given without + an offset. Events then carry a TZID and a VTIMEZONE. Unset: such times are UTC. """ nextcloud_url: str = field(default="") @@ -36,6 +39,8 @@ class Config: retry_max: int = field(default=3) is_app_password: bool = field(default=False) upload_root: str = field(default="") + # IANA zone for calendar times without an offset; "" = UTC (see NEXTCLOUD_MCP_TIMEZONE). + timezone: str = field(default="") @classmethod def from_env(cls) -> "Config": @@ -67,6 +72,15 @@ def from_env(cls) -> "Config": else: raise ValueError(f"Invalid NEXTCLOUD_MCP_APP_PASSWORD='{app_pw_raw}'. Expected: true/false, 1/0, yes/no.") + timezone = os.environ.get("NEXTCLOUD_MCP_TIMEZONE", "").strip() + if timezone: + try: + ZoneInfo(timezone) + except (ZoneInfoNotFoundError, ValueError): + raise ValueError( + f"Invalid NEXTCLOUD_MCP_TIMEZONE='{timezone}'. Expected an IANA name like Europe/Berlin." + ) from None + upload_root_raw = os.environ.get("NEXTCLOUD_MCP_UPLOAD_ROOT", "").strip() if upload_root_raw: root = Path(upload_root_raw).expanduser() @@ -88,6 +102,7 @@ def from_env(cls) -> "Config": retry_max=max(0, retry_max), is_app_password=is_app_password, upload_root=upload_root, + timezone=timezone, ) def validate(self) -> None: diff --git a/src/nc_mcp_server/tools/calendar.py b/src/nc_mcp_server/tools/calendar.py index 13ead44..9a46949 100644 --- a/src/nc_mcp_server/tools/calendar.py +++ b/src/nc_mcp_server/tools/calendar.py @@ -1,10 +1,12 @@ """Calendar tools — list calendars, query/create/update/delete events via CalDAV.""" import json +import re import uuid import xml.etree.ElementTree as ET -from datetime import UTC, date, datetime, timedelta +from datetime import UTC, date, datetime, time, timedelta, tzinfo from typing import Any +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from xml.sax.saxutils import escape as xml_escape from icalendar import Calendar as ICal @@ -40,6 +42,13 @@ SKIP_CALENDARS = {"inbox", "outbox", "trashbin"} +# Marks events created through this server with the creating user's ID. update_event refuses +# events without the caller's mark (or with attendees) unless allow_foreign=true, so a client +# can ask its user before changing someone else's event. +CREATED_BY_PROP = "X-NC-MCP-CREATED-BY" +CONFERENCE_LABEL = "Video call" +VIDEO_LINE_PREFIX = "Video call: " + def _caldav_path(user: str, calendar_id: str = "", resource: str = "") -> str: path = f"calendars/{user}/" @@ -179,8 +188,12 @@ def _is_all_day(dt_prop: Any) -> bool: return dt_prop is not None and isinstance(dt_prop.dt, date) and not isinstance(dt_prop.dt, datetime) -def _format_event(ical_text: str) -> dict[str, Any]: - """Parse iCalendar text and extract VEVENT fields into a dict.""" +def _format_event(ical_text: str, me: str | None = None) -> dict[str, Any]: + """Parse iCalendar text and extract VEVENT fields into a dict. + + Attendee and organizer addresses are never returned, only whether there are attendees. + With ``me`` (the caller's user ID) the result says whether the caller created the event here. + """ cal = ICal.from_ical(ical_text) for component in cal.walk(): if component.name != "VEVENT": @@ -200,19 +213,96 @@ def _format_event(ical_text: str) -> dict[str, Any]: result["rrule"] = rrule.to_ical().decode() if isinstance(component, IEvent) and component.categories: result["categories"] = [str(c) for c in component.categories] + conference = _conference_url(component) + if conference: + result["conference"] = conference + result["has_attendees"] = _has_attendees(component) + if me is not None: + result["created_by_me"] = _created_by(component) == me return result msg = "No VEVENT found in calendar data" raise ValueError(msg) -def _parse_dt(value: str, all_day: bool = False) -> date | datetime: - """Parse an ISO date or datetime string.""" +def _zone(name: str) -> ZoneInfo | None: + """IANA zone for ``name``; None for "".""" + if not name: + return None + try: + return ZoneInfo(name) + except (ZoneInfoNotFoundError, ValueError): + raise ValueError(f"Unknown timezone '{name}'. Use an IANA name like 'Europe/Berlin'.") from None + + +def _effective_zone(timezone: str, fallback: tzinfo | None = None) -> tzinfo | None: + """Zone for times without an offset: the timezone argument, else the fallback + (the event's own zone on updates), else NEXTCLOUD_MCP_TIMEZONE, else None (UTC).""" + if timezone: + return _zone(timezone) + if fallback is not None: + return fallback + return _zone(get_config().timezone) + + +def _parse_dt(value: str, all_day: bool = False, tz: tzinfo | None = None) -> date | datetime: + """Parse an ISO date or datetime string. + + A datetime without an offset is wall time in ``tz`` (UTC without ``tz``). A datetime with an + offset keeps its instant and is shown in ``tz`` when given, so the event carries that TZID. + """ if all_day or len(value) == 10: return date.fromisoformat(value) dt = datetime.fromisoformat(value) if dt.tzinfo is None: - dt = dt.replace(tzinfo=UTC) - return dt + return dt.replace(tzinfo=tz or UTC) + return dt.astimezone(tz) if tz is not None else dt + + +def _zone_of(dt_prop: Any) -> tzinfo | None: + """The IANA zone of an existing DTSTART (None for UTC, floating or all-day values).""" + val = getattr(dt_prop, "dt", None) + if isinstance(val, datetime) and isinstance(val.tzinfo, ZoneInfo) and val.tzinfo.key != "UTC": + return val.tzinfo + return None + + +def _to_caldav_utc(value: str, tz: tzinfo | None) -> str: + """ISO date/datetime -> CalDAV UTC time (YYYYMMDDTHHMMSSZ) for time-range filters.""" + value = value.strip() + try: + if len(value) == 10: + dt = datetime.combine(date.fromisoformat(value), time(), tzinfo=tz or UTC) + else: + dt = datetime.fromisoformat(value) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=tz or UTC) + except ValueError: + raise ValueError(f"Invalid date/time '{value}'. Use ISO 8601, e.g. 2026-04-01T00:00:00Z.") from None + return dt.astimezone(UTC).strftime("%Y%m%dT%H%M%SZ") + + +def _created_by(component: Any) -> str: + return str(component.get(CREATED_BY_PROP, "")).strip() + + +def _has_attendees(component: Any) -> bool: + return component.get("ATTENDEE") is not None + + +def _conference_url(component: Any) -> str: + conf = component.get("CONFERENCE") + if conf is None: + return "" + if isinstance(conf, list): + conf = conf[0] if conf else "" + return str(conf) + + +def _validate_url(url: str) -> str: + url = url.strip() + if not re.fullmatch(r"https?://[^\s<>\"]+", url): + raise ValueError(f"Invalid conference_url '{url}'. Expected an http(s) URL.") + return url def _build_ical( @@ -225,8 +315,10 @@ def _build_ical( status: str = "CONFIRMED", categories: list[str] | None = None, rrule: str = "", + conference_url: str = "", + created_by: str = "", ) -> str: - """Build a minimal iCalendar VEVENT string.""" + """Build a minimal iCalendar VEVENT string (with VTIMEZONE for zoned times).""" cal = ICal() cal.add("prodid", "-//nc-mcp-server//EN") cal.add("version", "2.0") @@ -245,25 +337,147 @@ def _build_ical( if categories: event.add("categories", categories) if rrule: - event.add("rrule", _parse_rrule(rrule)) + event.add("rrule", _parse_rrule(rrule, dtstart)) + if conference_url: + _set_conference(event, conference_url) + if created_by: + event.add(CREATED_BY_PROP, created_by) cal.add_component(event) + cal.add_missing_timezones() return cal.to_ical().decode() -def _parse_rrule(rrule_str: str) -> dict[str, list[Any]]: - """Parse an RRULE string like 'FREQ=WEEKLY;COUNT=4;BYDAY=MO,WE' into a dict.""" +def _set_conference(component: Any, url: str) -> None: + """Video link per RFC 7986 (CONFERENCE) plus where common clients show it: LOCATION when it + is empty (as Nextcloud Calendar does for Talk rooms), else a line in DESCRIPTION.""" + old = _conference_url(component) + _clear_conference(component, old) + if not url: + return + component.add( + "conference", url, parameters={"VALUE": "URI", "FEATURE": ["AUDIO", "VIDEO"], "LABEL": CONFERENCE_LABEL} + ) + if not str(component.get("LOCATION", "")).strip(): + _set_prop(component, "LOCATION", url) + else: + desc = str(component.get("DESCRIPTION", "")).rstrip() + line = f"{VIDEO_LINE_PREFIX}{url}" + _set_prop(component, "DESCRIPTION", f"{desc}\n\n{line}" if desc else line) + + +def _clear_conference(component: Any, old: str) -> None: + """Remove the video link this server set (CONFERENCE, LOCATION equal to it, its DESCRIPTION line).""" + component.pop("CONFERENCE", None) + if not old: + return + if str(component.get("LOCATION", "")).strip() == old: + component.pop("LOCATION", None) + desc = str(component.get("DESCRIPTION", "")) + line = f"{VIDEO_LINE_PREFIX}{old}" + if line in desc: + _set_prop(component, "DESCRIPTION", desc.replace(line, "").strip(), clear_if_empty=True) + + +RRULE_FREQ = {"DAILY", "WEEKLY", "MONTHLY", "YEARLY"} +RRULE_BYDAY = re.compile(r"^([+-]?[1-9][0-9]?)?(MO|TU|WE|TH|FR|SA|SU)$") +RRULE_INT_RANGES = { + "BYMONTHDAY": (-31, 31), + "BYMONTH": (1, 12), + "BYSETPOS": (-366, 366), + "BYWEEKNO": (-53, 53), + "BYYEARDAY": (-366, 366), +} +RRULE_MAX = 1000 + + +def _rrule_int(key: str, raw: str, low: int, high: int, zero_ok: bool = False) -> int: + try: + n = int(raw) + except ValueError: + raise ValueError(f"RRULE {key}={raw}: expected a number.") from None + if n < low or n > high or (n == 0 and not zero_ok): + raise ValueError(f"RRULE {key}={raw}: out of range {low}..{high}.") + return n + + +def _rrule_until(raw: str, dtstart: date | datetime | None) -> date | datetime: + """UNTIL as RFC 5545 wants it: a date for all-day events, else a UTC datetime. A value + without offset is wall time in the event's zone; a date means the end of that day there.""" + compact = raw.strip() + try: + if re.fullmatch(r"\d{8}", compact): + day = date(int(compact[:4]), int(compact[4:6]), int(compact[6:])) + parsed: date | datetime = day + elif re.fullmatch(r"\d{8}T\d{6}Z?", compact): + parsed = datetime.strptime(compact[:15] + "+0000", "%Y%m%dT%H%M%S%z") + if not compact.endswith("Z"): + parsed = parsed.replace(tzinfo=None) + elif len(compact) == 10: + parsed = date.fromisoformat(compact) + else: + parsed = datetime.fromisoformat(compact) + except ValueError: + raise ValueError(f"RRULE UNTIL={raw}: expected a date like 20261231 or 20261231T235959Z.") from None + if isinstance(dtstart, datetime): + zone = dtstart.tzinfo or UTC + if not isinstance(parsed, datetime): + parsed = datetime.combine(parsed, time(23, 59, 59), tzinfo=zone) + elif parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=zone) + return parsed.astimezone(UTC) + if isinstance(dtstart, date): + return parsed.date() if isinstance(parsed, datetime) else parsed + return parsed + + +WEEKDAYS = {"MO", "TU", "WE", "TH", "FR", "SA", "SU"} + + +def _rrule_value(key: str, val: str, dtstart: date | datetime | None) -> list[Any]: + """One checked RRULE part as icalendar wants it (a list of values).""" + if key == "FREQ": + if val.upper() not in RRULE_FREQ: + raise ValueError(f"RRULE FREQ={val}: use one of {', '.join(sorted(RRULE_FREQ))}.") + return [val.upper()] + if key == "UNTIL": + return [_rrule_until(val, dtstart)] + if key in {"COUNT", "INTERVAL"}: + return [_rrule_int(key, val, 1, RRULE_MAX)] + if key == "BYDAY": + days = [d.strip().upper() for d in val.split(",")] + if not all(RRULE_BYDAY.match(d) for d in days): + raise ValueError(f"RRULE BYDAY={val}: expected days like MO,WE or 1MO,-1FR.") + return days + if key in RRULE_INT_RANGES: + low, high = RRULE_INT_RANGES[key] + return [_rrule_int(key, v.strip(), low, high) for v in val.split(",")] + if key == "WKST" and val.upper() in WEEKDAYS: + return [val.upper()] + raise ValueError(f"RRULE part '{key}={val}' is not supported.") + + +def _parse_rrule(rrule_str: str, dtstart: date | datetime | None = None) -> dict[str, list[Any]]: + """Parse and check an RRULE like 'FREQ=WEEKLY;COUNT=4;BYDAY=MO,WE'. + + Only DAILY/WEEKLY/MONTHLY/YEARLY rules with known parts are accepted (no sub-daily + frequencies, COUNT and INTERVAL at most 1000, never COUNT and UNTIL together). + """ + text = rrule_str.strip() + if text.upper().startswith("RRULE:"): + text = text[6:] result: dict[str, list[Any]] = {} - for part in rrule_str.split(";"): + for part in filter(None, (p.strip() for p in text.split(";"))): if "=" not in part: - continue - key, val = part.split("=", 1) - key = key.strip() - if key == "UNTIL": - result[key] = [datetime.fromisoformat(val.strip())] - elif key in {"COUNT", "INTERVAL"}: - result[key] = [int(val.strip())] - else: - result[key] = [v.strip() for v in val.split(",")] + raise ValueError(f"RRULE part '{part}' is not KEY=VALUE.") + key, val = (x.strip() for x in part.split("=", 1)) + key = key.upper() + if key in result: + raise ValueError(f"RRULE {key} given twice.") + result[key] = _rrule_value(key, val, dtstart) + if "FREQ" not in result: + raise ValueError("RRULE needs FREQ (DAILY, WEEKLY, MONTHLY or YEARLY).") + if "COUNT" in result and "UNTIL" in result: + raise ValueError("RRULE takes COUNT or UNTIL, not both.") return result @@ -291,14 +505,14 @@ def _apply_event_updates( location: str | None, status: str | None, categories: list[str] | None = None, + tz: tzinfo | None = None, + rrule: str | None = None, + conference_url: str | None = None, ) -> None: if summary is not None: _set_prop(component, "SUMMARY", summary) - if start is not None: - _set_prop(component, "DTSTART", _parse_dt(start, _is_all_day(component.get("DTSTART")))) - if end is not None: - ref = component.get("DTEND") or component.get("DTSTART") - _set_prop(component, "DTEND", _parse_dt(end, _is_all_day(ref))) + if start is not None or end is not None: + _apply_times(component, start, end, tz) if description is not None: _set_prop(component, "DESCRIPTION", description, clear_if_empty=True) if location is not None: @@ -307,9 +521,48 @@ def _apply_event_updates( _set_prop(component, "STATUS", status) if categories is not None: _set_prop(component, "CATEGORIES", categories, clear_if_empty=True) + if rrule is not None: + start_prop = component.get("DTSTART") + _set_prop( + component, + "RRULE", + _parse_rrule(rrule, start_prop.dt if start_prop else None) if rrule else "", + clear_if_empty=True, + ) + if conference_url is not None: + _set_conference(component, _validate_url(conference_url) if conference_url else "") _set_prop(component, "DTSTAMP", datetime.now(UTC)) +def _apply_times(component: Any, start: str | None, end: str | None, tz: tzinfo | None) -> None: + if start is not None: + old_start, old_end = component.get("DTSTART"), component.get("DTEND") + new_start = _parse_dt(start, _is_all_day(old_start), tz) + _set_prop(component, "DTSTART", new_start) + # Moving only the start keeps the duration (DTEND follows), as calendar apps do. + if end is None and old_start is not None and old_end is not None and type(old_start.dt) is type(new_start): + _set_prop(component, "DTEND", new_start + (old_end.dt - old_start.dt)) + if end is not None: + ref = component.get("DTEND") or component.get("DTSTART") + _set_prop(component, "DTEND", _parse_dt(end, _is_all_day(ref), tz)) + + +def _check_summary(component: Any, expected_summary: str, event_uid: str) -> None: + """Guard against acting on the wrong event: the title must match when one is expected.""" + if not expected_summary: + return + actual = str(component.get("SUMMARY", "")) + if actual.strip().casefold() != expected_summary.strip().casefold(): + raise ValueError(f"Event '{event_uid}' is titled '{actual}', not '{expected_summary}'. Nothing was changed.") + + +def _vevent(cal: Any) -> Any: + for component in cal.walk(): + if component.name == "VEVENT": + return component + raise ValueError("No VEVENT found in calendar data") + + async def _find_event(calendar_id: str, event_uid: str) -> tuple[str, str, str]: """Find an event by UID. Returns (href, etag, ical_data) or raises.""" client = get_client() @@ -372,10 +625,10 @@ async def get_events( Args: calendar_id: Calendar identifier (default "personal"). Use list_calendars to find IDs. - start: Optional range start in ISO 8601 UTC format: "2026-04-01T00:00:00Z". - Required if end is provided. - end: Optional range end in ISO 8601 UTC format: "2026-04-30T23:59:59Z". - Required if start is provided. + start: Optional range start in ISO 8601, e.g. "2026-04-01T00:00:00Z" or + "2026-04-01T00:00:00+02:00". Without an offset the server time zone + (NEXTCLOUD_MCP_TIMEZONE, else UTC) applies. Required if end is provided. + end: Optional range end, same format. Required if start is provided. limit: Maximum number of events to return (1-500, default 50). offset: Number of events to skip for pagination (default 0). @@ -387,12 +640,9 @@ async def get_events( raise ValueError("Both start and end are required for time-range filtering, or omit both.") limit = max(1, min(500, limit)) offset = max(0, offset) - caldav_start = start.replace("-", "").replace(":", "").replace(".", "") if start else None - caldav_end = end.replace("-", "").replace(":", "").replace(".", "") if end else None - if caldav_start: - caldav_start = caldav_start[:15] + "Z" if not caldav_start.endswith("Z") else caldav_start - if caldav_end: - caldav_end = caldav_end[:15] + "Z" if not caldav_end.endswith("Z") else caldav_end + zone = _effective_zone("") + caldav_start = _to_caldav_utc(start, zone) if start else None + caldav_end = _to_caldav_utc(end, zone) if end else None client = get_client() user = get_config().user @@ -408,7 +658,7 @@ async def get_events( results = _parse_report_xml(response.text or "") all_events = [] for _href, etag, ical_data in results: - event = _format_event(ical_data) + event = _format_event(ical_data, user) event["etag"] = etag all_events.append(event) page = all_events[offset : offset + limit] @@ -433,10 +683,11 @@ async def get_event(calendar_id: str, event_uid: str) -> str: Returns: JSON object with full event details: uid, summary, dtstart, dtend, - description, location, status, all_day, etag, and optionally rrule, categories. + description, location, status, all_day, has_attendees, created_by_me, etag, + and optionally rrule, categories, conference (video link). """ _href, etag, ical_data = await _find_event(calendar_id, event_uid) - event = _format_event(ical_data) + event = _format_event(ical_data, get_config().user) event["etag"] = etag return json.dumps(event) @@ -455,6 +706,8 @@ async def create_event( status: str = "CONFIRMED", categories: str = "", rrule: str = "", + timezone: str = "", + conference_url: str = "", ) -> str: """Create a new calendar event. @@ -462,7 +715,8 @@ async def create_event( calendar_id: Calendar identifier (e.g. "personal"). summary: Event title/summary. start: Start date or datetime in ISO 8601 format. - For timed events: "2026-04-01T10:00:00Z" or "2026-04-01T10:00:00". + For timed events: "2026-04-01T10:00:00Z" or "2026-04-01T10:00:00" + (without offset: wall time in `timezone`, else the server time zone, else UTC). For all-day events: "2026-04-01". end: End date or datetime. Optional — defaults to 1 hour after start for timed events, or next day for all-day events. @@ -474,24 +728,49 @@ async def create_event( rrule: Optional recurrence rule in iCalendar RRULE format. Examples: "FREQ=DAILY;COUNT=5", "FREQ=WEEKLY;BYDAY=MO,WE,FR", "FREQ=MONTHLY;BYMONTHDAY=15;UNTIL=20261231T235959Z". + Only DAILY/WEEKLY/MONTHLY/YEARLY; COUNT or UNTIL, not both. + timezone: Optional IANA time zone (e.g. "Europe/Berlin") for the event. Times are + stored with this TZID, so recurring events keep their wall time across + daylight saving changes. Default: NEXTCLOUD_MCP_TIMEZONE, else UTC. + conference_url: Optional video call link (e.g. a Talk room URL). Stored as + CONFERENCE (RFC 7986) and as LOCATION when no location is given, + otherwise as a line in the description. Returns: - JSON object with the created event's uid and summary. + JSON object with the created event's uid, summary, dtstart, dtend and, + when set, conference. """ status_upper = _validate_status(status) cat_list = [c.strip() for c in categories.split(",") if c.strip()] if categories else None - dtstart = _parse_dt(start, all_day) + zone = _effective_zone(timezone) + conference = _validate_url(conference_url) if conference_url else "" + dtstart = _parse_dt(start, all_day, zone) if end: - dtend = _parse_dt(end, all_day) + dtend = _parse_dt(end, all_day, zone) elif all_day or not isinstance(dtstart, datetime): dtend = dtstart + timedelta(days=1) else: dtend = dtstart + timedelta(hours=1) + if type(dtend) is type(dtstart) and dtend < dtstart: + raise ValueError("The event ends before it starts.") + uid = str(uuid.uuid4()) - ical_data = _build_ical(uid, summary, dtstart, dtend, description, location, status_upper, cat_list, rrule) client = get_client() user = get_config().user + ical_data = _build_ical( + uid, + summary, + dtstart, + dtend, + description, + location, + status_upper, + cat_list, + rrule, + conference, + created_by=user, + ) path = _caldav_path(user, calendar_id, f"{uid}.ics") await client.dav_request( "PUT", @@ -500,7 +779,15 @@ async def create_event( headers={"Content-Type": "text/calendar; charset=utf-8"}, context=f"Create event in '{calendar_id}'", ) - return json.dumps({"uid": uid, "summary": summary}) + result: dict[str, Any] = { + "uid": uid, + "summary": summary, + "dtstart": _dt_to_str(dtstart), + "dtend": _dt_to_str(dtend), + } + if conference: + result["conference"] = conference + return json.dumps(result) def _register_update_event(mcp: FastMCP) -> None: @@ -516,22 +803,39 @@ async def update_event( location: str | None = None, status: str | None = None, categories: str | None = None, + rrule: str | None = None, + timezone: str = "", + conference_url: str | None = None, + allow_foreign: bool = False, + expected_summary: str = "", ) -> str: """Update an existing calendar event. Only provided fields are changed. Uses the event's ETag for safe concurrent updates — if the event was modified since it was last read, the update will fail with a conflict error. + Events that were not created by the calling user through this server, or that + have attendees (who Nextcloud would notify), are only changed with + allow_foreign=true. Ask the user before passing it. + Args: calendar_id: Calendar identifier (e.g. "personal"). event_uid: The event's UID to update. Use get_events to find UIDs. summary: New event title. - start: New start date/datetime in ISO 8601 format. + start: New start date/datetime in ISO 8601 format. Without `end` the + event keeps its duration. end: New end date/datetime in ISO 8601 format. description: New description. Pass "" to clear. location: New location. Pass "" to clear. status: New status: "CONFIRMED", "TENTATIVE", or "CANCELLED". categories: New categories as comma-separated string. Pass "" to clear. + rrule: New recurrence rule (see create_event). Pass "" to end the recurrence. + timezone: IANA zone for new start/end without offset. Default: the event's + own zone, else NEXTCLOUD_MCP_TIMEZONE, else UTC. + conference_url: New video call link. Pass "" to remove the link set before. + allow_foreign: Also change events created by someone else (or with attendees). + expected_summary: Optional current title; the update is refused if it differs + (protects against changing the wrong event). Returns: Confirmation message with the updated event UID. @@ -542,10 +846,38 @@ async def update_event( cat_list = [c.strip() for c in categories.split(",") if c.strip()] if categories else [] href, etag, ical_data = await _find_event(calendar_id, event_uid) cal = ICal.from_ical(ical_data) - for component in cal.walk(): - if component.name == "VEVENT": - _apply_event_updates(component, summary, start, end, description, location, validated_status, cat_list) - break + component = _vevent(cal) + _check_summary(component, expected_summary, event_uid) + me = get_config().user + if not allow_foreign and (_created_by(component) != me or _has_attendees(component)): + raise ValueError( + f"Event '{event_uid}' was not created by you through this server, or it has attendees " + "who would be notified. Nothing was changed. Ask the user, then call update_event " + "again with allow_foreign=true." + ) + zone = _effective_zone(timezone, _zone_of(component.get("DTSTART"))) + _apply_event_updates( + component, + summary, + start, + end, + description, + location, + validated_status, + cat_list, + zone, + rrule, + conference_url, + ) + new_start, new_end = component.get("DTSTART"), component.get("DTEND") + if ( + new_start is not None + and new_end is not None + and type(new_start.dt) is type(new_end.dt) + and new_end.dt < new_start.dt + ): + raise ValueError("The event would end before it starts. Nothing was changed.") + cal.add_missing_timezones() client = get_client() await client.dav_request( @@ -561,7 +893,7 @@ async def update_event( def _register_destructive_tools(mcp: FastMCP) -> None: @mcp.tool(annotations=DESTRUCTIVE) @require_permission(PermissionLevel.DESTRUCTIVE) - async def delete_event(calendar_id: str, event_uid: str) -> str: + async def delete_event(calendar_id: str, event_uid: str, expected_summary: str = "") -> str: """Delete a calendar event by its UID. The event is moved to the calendar trashbin and can be restored @@ -570,11 +902,14 @@ async def delete_event(calendar_id: str, event_uid: str) -> str: Args: calendar_id: Calendar identifier (e.g. "personal"). event_uid: The event's UID to delete. Use get_events to find UIDs. + expected_summary: Optional current title; nothing is deleted if it differs + (protects against deleting the wrong event). Returns: Confirmation message. """ - href, _etag, _ical = await _find_event(calendar_id, event_uid) + href, _etag, ical_data = await _find_event(calendar_id, event_uid) + _check_summary(_vevent(ICal.from_ical(ical_data)), expected_summary, event_uid) client = get_client() await client.dav_request("DELETE", _href_to_dav_path(href), context=f"Delete event '{event_uid}'") return f"Event '{event_uid}' deleted." diff --git a/tests/integration/test_calendar_guard.py b/tests/integration/test_calendar_guard.py new file mode 100644 index 0000000..37784ad --- /dev/null +++ b/tests/integration/test_calendar_guard.py @@ -0,0 +1,258 @@ +"""Calendar against a real Nextcloud: time zones, video links, the creator guard and shared calendars. + +A calendar shared with another account shows up there as ``_shared_by_``; events +written there carry the writer's mark, and update_event refuses other people's events unless +allow_foreign=true. +""" + +import contextlib +import json +import secrets +import uuid +from collections.abc import AsyncGenerator + +import pytest +from mcp.server.fastmcp.exceptions import ToolError + +from nc_mcp_server.client import NextcloudClient +from nc_mcp_server.config import Config +from nc_mcp_server.permissions import PermissionLevel +from nc_mcp_server.server import create_server +from nc_mcp_server.state import get_client + +from .conftest import McpTestHelper, _get_integration_config + +pytestmark = pytest.mark.integration + +CAL_ID = "personal" +VIDEO = "https://cloud.example.org/index.php/call/mcptest1" + + +def _server(config: Config) -> McpTestHelper: + mcp = create_server(config) + return McpTestHelper(mcp, get_client()) + + +async def _delete_quietly(helper: McpTestHelper, calendar_id: str, uid: str) -> None: + with contextlib.suppress(ToolError): + await helper.call("delete_event", calendar_id=calendar_id, event_uid=uid) + + +class TestTimezone: + @pytest.mark.asyncio + async def test_wall_time_in_zone(self, nc_mcp: McpTestHelper) -> None: + created = json.loads( + await nc_mcp.call( + "create_event", + calendar_id=CAL_ID, + summary="mcp-test-tz", + start="2027-03-20T19:00:00", + end="2027-03-20T21:00:00", + timezone="Europe/Berlin", + rrule="FREQ=WEEKLY;UNTIL=20270410", + ) + ) + try: + assert created["dtstart"] == "2027-03-20T19:00:00+01:00" + event = json.loads(await nc_mcp.call("get_event", calendar_id=CAL_ID, event_uid=created["uid"])) + assert event["dtstart"] == "2027-03-20T19:00:00+01:00" + assert event["created_by_me"] is True + assert "UNTIL=20270410T215959Z" in event["rrule"] + # Range given with an offset finds it (18:00 UTC is inside 19:00-21:00 CET). + found = json.loads( + await nc_mcp.call( + "get_events", calendar_id=CAL_ID, start="2027-03-20T19:30:00+01:00", end="2027-03-20T19:45:00+01:00" + ) + )["data"] + assert created["uid"] in [e["uid"] for e in found] + # A later move keeps the event's zone; summer time gives +02:00. + await nc_mcp.call("update_event", calendar_id=CAL_ID, event_uid=created["uid"], start="2027-04-03T19:00:00") + moved = json.loads(await nc_mcp.call("get_event", calendar_id=CAL_ID, event_uid=created["uid"])) + assert moved["dtstart"] == "2027-04-03T19:00:00+02:00" + finally: + await _delete_quietly(nc_mcp, CAL_ID, created["uid"]) + + @pytest.mark.asyncio + async def test_bad_rrule_creates_nothing(self, nc_mcp: McpTestHelper) -> None: + before = json.loads(await nc_mcp.call("get_events", calendar_id=CAL_ID, limit=500))["pagination"]["count"] + with pytest.raises((ToolError, ValueError)): + await nc_mcp.call( + "create_event", + calendar_id=CAL_ID, + summary="mcp-test-bad", + start="2027-01-01T10:00:00Z", + rrule="FREQ=MINUTELY", + ) + after = json.loads(await nc_mcp.call("get_events", calendar_id=CAL_ID, limit=500))["pagination"]["count"] + assert after == before + + +class TestVideoLink: + @pytest.mark.asyncio + async def test_link_set_replaced_removed(self, nc_mcp: McpTestHelper) -> None: + created = json.loads( + await nc_mcp.call( + "create_event", + calendar_id=CAL_ID, + summary="mcp-test-video", + start="2027-02-01T18:00:00Z", + conference_url=VIDEO, + ) + ) + try: + event = json.loads(await nc_mcp.call("get_event", calendar_id=CAL_ID, event_uid=created["uid"])) + assert event["conference"] == VIDEO + assert event["location"] == VIDEO + await nc_mcp.call("update_event", calendar_id=CAL_ID, event_uid=created["uid"], conference_url="") + event = json.loads(await nc_mcp.call("get_event", calendar_id=CAL_ID, event_uid=created["uid"])) + assert "conference" not in event + assert event["location"] == "" + finally: + await _delete_quietly(nc_mcp, CAL_ID, created["uid"]) + + +class TestGuard: + @pytest.mark.asyncio + async def test_event_without_mark_needs_allow_foreign(self, nc_mcp: McpTestHelper) -> None: + uid = f"mcp-test-foreign-{uuid.uuid4()}" + ical = ( + "BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//test//EN\r\nBEGIN:VEVENT\r\n" + f"UID:{uid}\r\nDTSTAMP:20260101T000000Z\r\nDTSTART:20270301T090000Z\r\nDTEND:20270301T100000Z\r\n" + "SUMMARY:mcp-test-foreign\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n" + ) + user = nc_mcp.client._config.user + await nc_mcp.client.dav_request( + "PUT", + f"calendars/{user}/{CAL_ID}/{uid}.ics", + body=ical, + headers={"Content-Type": "text/calendar; charset=utf-8"}, + ) + try: + with pytest.raises(ToolError, match="allow_foreign"): + await nc_mcp.call("update_event", calendar_id=CAL_ID, event_uid=uid, summary="mcp-test-changed") + event = json.loads(await nc_mcp.call("get_event", calendar_id=CAL_ID, event_uid=uid)) + assert event["summary"] == "mcp-test-foreign" + assert event["created_by_me"] is False + await nc_mcp.call( + "update_event", calendar_id=CAL_ID, event_uid=uid, summary="mcp-test-changed", allow_foreign=True + ) + event = json.loads(await nc_mcp.call("get_event", calendar_id=CAL_ID, event_uid=uid)) + assert event["summary"] == "mcp-test-changed" + finally: + await _delete_quietly(nc_mcp, CAL_ID, uid) + + @pytest.mark.asyncio + async def test_expected_summary_protects_delete(self, nc_mcp: McpTestHelper) -> None: + created = json.loads( + await nc_mcp.call("create_event", calendar_id=CAL_ID, summary="mcp-test-keep", start="2027-02-02T18:00:00Z") + ) + try: + with pytest.raises(ToolError, match="Nothing was changed"): + await nc_mcp.call( + "delete_event", calendar_id=CAL_ID, event_uid=created["uid"], expected_summary="other" + ) + json.loads(await nc_mcp.call("get_event", calendar_id=CAL_ID, event_uid=created["uid"])) + await nc_mcp.call( + "delete_event", calendar_id=CAL_ID, event_uid=created["uid"], expected_summary="MCP-test-keep" + ) + with pytest.raises(ToolError): + await nc_mcp.call("get_event", calendar_id=CAL_ID, event_uid=created["uid"]) + finally: + await _delete_quietly(nc_mcp, CAL_ID, created["uid"]) + + +@pytest.fixture +async def shared_calendar() -> AsyncGenerator[tuple[Config, Config, str, str]]: + """admin owns a calendar shared read-write with a new member account and read-only with another.""" + admin_config = _get_integration_config() + admin = NextcloudClient(admin_config) + suffix = secrets.token_hex(4) + uri = f"mcp-test-verein-{suffix}" + members: list[Config] = [] + try: + for name in ("kal-anna", "kal-leser"): + user_id = f"{name}-{suffix}" + password = f"Mcp-{secrets.token_hex(10)}!" + await admin.ocs_post("cloud/users", data={"userid": user_id, "password": password}) + members.append( + Config( + nextcloud_url=admin_config.nextcloud_url, + user=user_id, + password=password, + permission_level=PermissionLevel.DESTRUCTIVE, + ) + ) + await admin.dav_request( + "MKCALENDAR", + f"calendars/{admin_config.user}/{uri}", + body=( + '' + "Vereinstermine Test" + ), + headers={"Content-Type": "application/xml; charset=utf-8"}, + ) + for member, right in ((members[0], ""), (members[1], "")): + await admin.dav_request( + "POST", + f"calendars/{admin_config.user}/{uri}", + body=( + '' + f"principal:principals/users/{member.user}{right}" + ), + headers={"Content-Type": "application/xml; charset=utf-8"}, + ) + yield admin_config, members[0], members[1], uri + finally: + with contextlib.suppress(Exception): + await admin.dav_request("DELETE", f"calendars/{admin_config.user}/{uri}") + for member in members: + with contextlib.suppress(Exception): + await admin.ocs_delete(f"cloud/users/{member.user}") + await admin.close() + + +class TestSharedCalendar: + @pytest.mark.asyncio + async def test_member_writes_owner_needs_allow_foreign( + self, shared_calendar: tuple[Config, Config, str, str] + ) -> None: + admin_config, anna, leser, uri = shared_calendar + shared_id = f"{uri}_shared_by_{admin_config.user}" + + member = _server(anna) + try: + calendars = {c["id"]: c for c in json.loads(await member.call("list_calendars"))} + assert shared_id in calendars + assert calendars[shared_id]["writable"] is True + created = json.loads( + await member.call( + "create_event", + calendar_id=shared_id, + summary="mcp-test-stammtisch", + start="2027-05-04T19:00:00", + timezone="Europe/Berlin", + ) + ) + await member.call("update_event", calendar_id=shared_id, event_uid=created["uid"], description="Neu") + finally: + await member.client.close() + + owner = _server(admin_config) + try: + event = json.loads(await owner.call("get_event", calendar_id=uri, event_uid=created["uid"])) + assert event["created_by_me"] is False + with pytest.raises(ToolError, match="allow_foreign"): + await owner.call("update_event", calendar_id=uri, event_uid=created["uid"], summary="x") + finally: + await owner.client.close() + + reader = _server(leser) + try: + calendars = {c["id"]: c for c in json.loads(await reader.call("list_calendars"))} + assert calendars[shared_id]["writable"] is False + with pytest.raises(ToolError): + await reader.call( + "create_event", calendar_id=shared_id, summary="mcp-test-nope", start="2027-05-05T19:00:00Z" + ) + finally: + await reader.client.close() diff --git a/tests/test_calendar_helpers.py b/tests/test_calendar_helpers.py new file mode 100644 index 0000000..982de81 --- /dev/null +++ b/tests/test_calendar_helpers.py @@ -0,0 +1,216 @@ +"""Unit tests for calendar helpers: time zones, RRULE checks, video links, creator guard.""" + +from datetime import UTC, date, datetime +from zoneinfo import ZoneInfo + +import pytest +from icalendar import Calendar as ICal + +from nc_mcp_server.tools.calendar import ( + CREATED_BY_PROP, + _apply_event_updates, + _build_ical, + _check_summary, + _format_event, + _parse_dt, + _parse_rrule, + _to_caldav_utc, + _validate_url, + _vevent, + _zone, + _zone_of, +) + +BERLIN = ZoneInfo("Europe/Berlin") + + +class TestParseDt: + def test_naive_without_zone_is_utc(self) -> None: + assert _parse_dt("2026-10-20T19:00:00") == datetime(2026, 10, 20, 19, 0, tzinfo=UTC) + + def test_naive_is_wall_time_in_zone(self) -> None: + dt = _parse_dt("2026-10-20T19:00:00", tz=BERLIN) + assert isinstance(dt, datetime) + assert dt.tzinfo == BERLIN + assert dt.astimezone(UTC) == datetime(2026, 10, 20, 17, 0, tzinfo=UTC) + + def test_offset_keeps_instant_and_takes_zone(self) -> None: + dt = _parse_dt("2026-12-01T18:00:00Z", tz=BERLIN) + assert isinstance(dt, datetime) + assert dt.tzinfo == BERLIN + assert dt.hour == 19 + + def test_all_day(self) -> None: + assert _parse_dt("2026-10-20", tz=BERLIN) == date(2026, 10, 20) + + def test_unknown_zone(self) -> None: + with pytest.raises(ValueError, match="Unknown timezone"): + _zone("Mars/Olympus") + + +class TestCaldavUtc: + def test_offset(self) -> None: + assert _to_caldav_utc("2026-10-20T00:00:00+02:00", None) == "20261019T220000Z" + + def test_naive_in_zone(self) -> None: + assert _to_caldav_utc("2026-01-10T00:00:00", BERLIN) == "20260109T230000Z" + + def test_date_in_zone(self) -> None: + assert _to_caldav_utc("2026-07-01", BERLIN) == "20260630T220000Z" + + def test_utc_z(self) -> None: + assert _to_caldav_utc("2026-04-01T00:00:00Z", BERLIN) == "20260401T000000Z" + + def test_garbage(self) -> None: + with pytest.raises(ValueError, match="Invalid date/time"): + _to_caldav_utc("next tuesday", None) + + +class TestRrule: + def test_weekly(self) -> None: + assert _parse_rrule("FREQ=WEEKLY;COUNT=4;BYDAY=MO,WE") == { + "FREQ": ["WEEKLY"], + "COUNT": [4], + "BYDAY": ["MO", "WE"], + } + + def test_prefix_and_case(self) -> None: + assert _parse_rrule("RRULE:freq=monthly;byday=1tu") == {"FREQ": ["MONTHLY"], "BYDAY": ["1TU"]} + + @pytest.mark.parametrize( + "rule", + [ + "COUNT=3", + "FREQ=HOURLY", + "FREQ=SECONDLY", + "FREQ=WEEKLY;COUNT=0", + "FREQ=WEEKLY;COUNT=100000", + "FREQ=WEEKLY;COUNT=2;UNTIL=20261231", + "FREQ=WEEKLY;BYDAY=XX", + "FREQ=MONTHLY;BYMONTHDAY=32", + "FREQ=WEEKLY;X-FOO=1", + "FREQ=WEEKLY;FREQ=DAILY", + "FREQ=WEEKLY;BYHOUR=3", + "FREQ", + ], + ) + def test_rejects(self, rule: str) -> None: + with pytest.raises(ValueError, match="RRULE"): + _parse_rrule(rule) + + def test_until_date_for_zoned_start_is_end_of_day_utc(self) -> None: + start = datetime(2026, 10, 20, 19, 0, tzinfo=BERLIN) + until = _parse_rrule("FREQ=WEEKLY;UNTIL=20261215", start)["UNTIL"][0] + assert until == datetime(2026, 12, 15, 22, 59, 59, tzinfo=UTC) + + def test_until_naive_datetime_in_start_zone(self) -> None: + start = datetime(2026, 7, 1, 19, 0, tzinfo=BERLIN) + until = _parse_rrule("FREQ=DAILY;UNTIL=20260705T190000", start)["UNTIL"][0] + assert until == datetime(2026, 7, 5, 17, 0, tzinfo=UTC) + + def test_until_for_all_day_is_date(self) -> None: + until = _parse_rrule("FREQ=DAILY;UNTIL=20260705T235959Z", date(2026, 7, 1))["UNTIL"][0] + assert until == date(2026, 7, 5) + + def test_until_utc_kept(self) -> None: + until = _parse_rrule("FREQ=DAILY;UNTIL=20270705T235959Z")["UNTIL"][0] + assert until == datetime(2027, 7, 5, 23, 59, 59, tzinfo=UTC) + + +class TestBuildIcal: + def test_zoned_event_has_tzid_vtimezone_and_creator(self) -> None: + start = datetime(2026, 10, 20, 19, 0, tzinfo=BERLIN) + end = datetime(2026, 10, 20, 21, 0, tzinfo=BERLIN) + text = _build_ical("u1", "Stammtisch", start, end, rrule="FREQ=MONTHLY;BYDAY=3TU", created_by="anna") + assert "DTSTART;TZID=Europe/Berlin:20261020T190000" in text + assert "BEGIN:VTIMEZONE" in text + assert f"{CREATED_BY_PROP}:anna" in text + event = _format_event(text, "anna") + assert event["dtstart"] == "2026-10-20T19:00:00+02:00" + assert event["created_by_me"] is True + assert event["has_attendees"] is False + assert _format_event(text, "bert")["created_by_me"] is False + + def test_conference_goes_to_location_when_empty(self) -> None: + start = datetime(2026, 10, 20, 17, 0, tzinfo=UTC) + url = "https://cloud.example.org/index.php/call/abc123" + text = _build_ical("u2", "Call", start, start, conference_url=url) + event = _format_event(text) + assert event["conference"] == url + assert event["location"] == url + assert "FEATURE=AUDIO,VIDEO" in text.replace("\r\n ", "") + + def test_conference_goes_to_description_with_location(self) -> None: + start = datetime(2026, 10, 20, 17, 0, tzinfo=UTC) + url = "https://cloud.example.org/index.php/call/abc123" + text = _build_ical("u3", "Meeting", start, start, description="Agenda", location="Room 1", conference_url=url) + event = _format_event(text) + assert event["location"] == "Room 1" + assert event["description"] == f"Agenda\n\nVideo call: {url}" + + def test_attendees_never_listed(self) -> None: + text = ( + "BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:x\r\nBEGIN:VEVENT\r\nUID:a\r\n" + "DTSTART:20261020T170000Z\r\nDTEND:20261020T180000Z\r\nSUMMARY:x\r\n" + "ORGANIZER:mailto:o@example.org\r\nATTENDEE:mailto:p@example.org\r\n" + "END:VEVENT\r\nEND:VCALENDAR\r\n" + ) + event = _format_event(text, "o") + assert event["has_attendees"] is True + assert "example.org" not in str(event) + + +class TestUpdates: + def _event(self, **kw: object) -> ICal: + start = datetime(2026, 10, 20, 19, 0, tzinfo=BERLIN) + return ICal.from_ical(_build_ical("u", "Stammtisch", start, start.replace(hour=21), **kw)) # type: ignore[arg-type] + + def test_new_start_keeps_event_zone(self) -> None: + component = _vevent(self._event()) + zone = _zone_of(component.get("DTSTART")) + assert zone == BERLIN + _apply_event_updates(component, None, "2026-11-03T18:30:00", None, None, None, None, tz=zone) + assert component.get("DTSTART").dt == datetime(2026, 11, 3, 18, 30, tzinfo=BERLIN) + + def test_moving_start_keeps_duration(self) -> None: + component = _vevent(self._event()) + _apply_event_updates(component, None, "2026-11-03T18:00:00", None, None, None, None, tz=BERLIN) + assert component.get("DTEND").dt == datetime(2026, 11, 3, 20, 0, tzinfo=BERLIN) + + def test_replace_and_remove_conference(self) -> None: + component = _vevent(self._event(conference_url="https://x.example/call/old")) + _apply_event_updates(component, None, None, None, None, None, None, conference_url="https://x.example/call/new") + assert str(component.get("LOCATION")) == "https://x.example/call/new" + _apply_event_updates(component, None, None, None, None, None, None, conference_url="") + assert component.get("LOCATION") is None + assert component.get("CONFERENCE") is None + + def test_remove_conference_line_from_description(self) -> None: + component = _vevent(self._event(location="Room 1", description="Agenda", conference_url="https://x.example/c")) + _apply_event_updates(component, None, None, None, None, None, None, conference_url="") + assert str(component.get("DESCRIPTION")) == "Agenda" + assert str(component.get("LOCATION")) == "Room 1" + + def test_rrule_set_and_cleared(self) -> None: + component = _vevent(self._event()) + _apply_event_updates(component, None, None, None, None, None, None, rrule="FREQ=WEEKLY;COUNT=3") + assert component.get("RRULE") is not None + _apply_event_updates(component, None, None, None, None, None, None, rrule="") + assert component.get("RRULE") is None + + def test_check_summary(self) -> None: + component = _vevent(self._event()) + _check_summary(component, " stammtisch ", "u") + _check_summary(component, "", "u") + with pytest.raises(ValueError, match="Nothing was changed"): + _check_summary(component, "Vorstandssitzung", "u") + + +class TestUrl: + def test_ok(self) -> None: + assert _validate_url(" https://a.example/call/x ") == "https://a.example/call/x" + + @pytest.mark.parametrize("url", ["javascript:alert(1)", "ftp://a", "https://a b", ""]) + def test_rejects(self, url: str) -> None: + with pytest.raises(ValueError, match="conference_url"): + _validate_url(url) From 9e198e01351886778b62a9cfffbfdbbbc906bf70 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Schotte?= Date: Wed, 7 Oct 2026 11:03:37 +0200 Subject: [PATCH 4/6] Calendar administration: create (optionally shared), share/unshare, public link, delete, shares MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit create_calendar (unique URI, also past trashed calendars; optional group share), share_calendar/unshare_calendar (users and groups), publish_calendar/unpublish_calendar, delete_calendar (expected_name guard), get_calendar_shares (owner only; Depth 1 on the home because Nextcloud omits owner-principal with Depth 0). Tests: unit + integration against Nextcloud 34 (group share visible to the member, sharees see no shares and cannot share). Signed-off-by: Björn Schotte Co-Authored-By: Claude Opus 5.5 --- README.md | 7 + src/nc_mcp_server/tools/calendar.py | 318 ++++++++++++++++++++++- tests/integration/test_calendar_admin.py | 121 +++++++++ tests/test_calendar_helpers.py | 27 ++ 4 files changed, 472 insertions(+), 1 deletion(-) create mode 100644 tests/integration/test_calendar_admin.py diff --git a/README.md b/README.md index e5c6ebb..77983e2 100644 --- a/README.md +++ b/README.md @@ -389,6 +389,13 @@ instead of the placeholders Talk stores (`{mention-user1}`, `{file}`). | `create_event` | write | Create a calendar event (time zone, checked RRULE, video link) | | `update_event` | write | Update an event (partial updates supported) | | `delete_event` | destructive | Delete a calendar event | +| `get_calendar_shares` | read | Who a calendar is shared with, public link (owner only) | +| `create_calendar` | write | Create a calendar, optionally shared with a group right away | +| `share_calendar` | write | Share a calendar with a user/group or change their rights | +| `unshare_calendar` | write | Stop sharing a calendar with a user/group | +| `publish_calendar` | write | Create a public read-only link | +| `unpublish_calendar` | write | Remove the public link | +| `delete_calendar` | destructive | Delete a calendar (`expected_name` guard); for a sharee: remove it from their account | - Time zone: times without an offset are wall time in `timezone` (tool argument), else `NEXTCLOUD_MCP_TIMEZONE` (IANA name, e.g. `Europe/Berlin`), else UTC. Zoned events get a `TZID` and a diff --git a/src/nc_mcp_server/tools/calendar.py b/src/nc_mcp_server/tools/calendar.py index 9a46949..3d58d8d 100644 --- a/src/nc_mcp_server/tools/calendar.py +++ b/src/nc_mcp_server/tools/calendar.py @@ -6,8 +6,9 @@ import xml.etree.ElementTree as ET from datetime import UTC, date, datetime, time, timedelta, tzinfo from typing import Any -from zoneinfo import ZoneInfo, ZoneInfoNotFoundError +from urllib.parse import unquote from xml.sax.saxutils import escape as xml_escape +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from icalendar import Calendar as ICal from icalendar import Event as IEvent @@ -915,9 +916,324 @@ async def delete_event(calendar_id: str, event_uid: str, expected_summary: str = return f"Event '{event_uid}' deleted." +CAL_ID_RE = re.compile(r"^[A-Za-z0-9_@-][A-Za-z0-9_.@' -]{0,199}$") +SHARE_TYPES = {"user": "users", "group": "groups"} +SHARE_WITH_RE = re.compile(r"^[A-Za-z0-9_.@' -]{1,64}$") + + +def _check_calendar_id(calendar_id: str) -> str: + if not CAL_ID_RE.match(calendar_id) or ".." in calendar_id or calendar_id in SKIP_CALENDARS: + raise ValueError(f"Invalid calendar_id '{calendar_id}'.") + return calendar_id + + +def _slug(name: str) -> str: + """URI for a new calendar, like Nextcloud Calendar derives it from the name.""" + table = str.maketrans({"ä": "ae", "ö": "oe", "ü": "ue", "ß": "ss", "Ä": "ae", "Ö": "oe", "Ü": "ue"}) + slug = re.sub(r"[^a-z0-9]+", "-", name.translate(table).lower()).strip("-") + return (slug or "calendar")[:60] + + +def _share_body(principal: str, write: bool | None) -> str: + if write is None: + inner = f"principal:{xml_escape(principal)}" + else: + access = "" if write else "" + inner = f"principal:{xml_escape(principal)}{access}" + return f'{inner}' + + +def _principal(share_with: str, share_type: str) -> str: + kind = SHARE_TYPES.get(share_type.strip().lower()) + if kind is None: + raise ValueError("share_type must be 'user' or 'group'.") + if not SHARE_WITH_RE.match(share_with.strip()): + raise ValueError(f"Invalid share_with '{share_with}'.") + return f"principals/{kind}/{share_with.strip()}" + + +SHARES_PROPFIND = ( + '' + "" +) +OC_NS = "http://owncloud.org/ns" + + +async def _calendar_info(calendar_id: str) -> dict[str, Any]: + """Name, owner, shares and public URL of one calendar (shares only visible to the owner).""" + client = get_client() + user = get_config().user + # Depth 1 on the calendar home: Nextcloud returns owner-principal only there, not with + # Depth 0 on the calendar itself (seen on Nextcloud 34). + response = await client.dav_request( + "PROPFIND", + _caldav_path(user), + body=SHARES_PROPFIND, + headers={"Depth": "1", "Content-Type": "application/xml; charset=utf-8"}, + context=f"Calendar '{calendar_id}'", + ) + root = ET.fromstring(response.text or "") # noqa: S314 + prop = None + for resp in root.findall(f"{{{DAV_NS}}}response"): + href = resp.find(f"{{{DAV_NS}}}href") + if href is None or not href.text or unquote(href.text.rstrip("/").rsplit("/", 1)[-1]) != calendar_id: + continue + prop = find_ok_prop(resp) + break + if prop is None: + raise NextcloudError(f"Calendar '{calendar_id}' not found", 404) + owner = (_el_text(prop, OC_NS, "owner-principal") or "").rsplit("/", 1)[-1] + shares = [] + invite = prop.find(f"{{{OC_NS}}}invite") + if invite is not None: + for entry in invite.findall(f"{{{OC_NS}}}user"): + href = entry.find(f"{{{DAV_NS}}}href") + principal = (href.text or "").removeprefix("principal:") if href is not None else "" + kind, _, name = principal.removeprefix("principals/").partition("/") + write = entry.find(f"{{{OC_NS}}}access/{{{OC_NS}}}read-write") is not None + shares.append({"type": "group" if kind == "groups" else "user", "share_with": name, "write": write}) + publish = prop.find(f"{{{CS_NS}}}publish-url") + public_url = None + if publish is not None: + href = publish.find(f"{{{DAV_NS}}}href") + public_url = href.text if href is not None and href.text else None + return { + "id": calendar_id, + "name": _el_text(prop, DAV_NS, "displayname") or calendar_id, + "owner": owner, + "owned_by_me": owner == user, + "shares": shares if owner == user else None, + "public_url": public_url, + } + + +def _register_calendar_admin(mcp: FastMCP) -> None: + @mcp.tool(annotations=READONLY) + @require_permission(PermissionLevel.READ) + async def get_calendar_shares(calendar_id: str) -> str: + """Who a calendar is shared with and whether it has a public link. + + Shares are only visible to the calendar's owner (null for everyone else). + + Args: + calendar_id: Calendar identifier from list_calendars. + + Returns: + JSON with id, name, owner, owned_by_me, shares ([{type, share_with, write}] or null), public_url. + """ + return json.dumps(await _calendar_info(_check_calendar_id(calendar_id))) + + @mcp.tool(annotations=ADDITIVE) + @require_permission(PermissionLevel.WRITE) + async def create_calendar(name: str, color: str = "", share_with_group: str = "", group_write: bool = False) -> str: + """Create a new calendar (events only) for the current user, optionally shared with a group. + + Args: + name: Display name, e.g. "Sommerfest 2027". + color: Optional color as #RRGGBB. + share_with_group: Optional Nextcloud group ID to share the new calendar with right away. + group_write: Whether that group may also add and change events (default: read only). + + Returns: + JSON with id (use it as calendar_id), name and, when shared, the share. + """ + name = name.strip() + if not name or len(name) > 100 or any(ord(c) < 32 for c in name): + raise ValueError("Calendar name must be 1-100 characters without control characters.") + if color and not re.fullmatch(r"#[0-9A-Fa-f]{6}", color): + raise ValueError("color must look like #1E78C1.") + principal = _principal(share_with_group, "group") if share_with_group else "" + client = get_client() + user = get_config().user + existing = { + c["id"] + for c in _parse_calendars_xml( + ( + await client.dav_request( + "PROPFIND", + _caldav_path(user), + body=CALENDAR_PROPFIND, + headers={"Depth": "1", "Content-Type": "application/xml; charset=utf-8"}, + context="List calendars", + ) + ).text + or "", + user, + ) + } + base = _slug(name) + uri = base + n = 2 + while uri in existing or uri in SKIP_CALENDARS: + uri = f"{base}-{n}" + n += 1 + color_prop = f"{color}" if color else "" + body = ( + '' + f"{xml_escape(name)}{color_prop}" + '' + "" + ) + # 405: the URI is taken, e.g. by a deleted calendar still in the trash bin -> next suffix. + for _attempt in range(10): + try: + await client.dav_request( + "MKCALENDAR", + _caldav_path(user, uri), + body=body, + headers={"Content-Type": "application/xml; charset=utf-8"}, + context=f"Create calendar '{name}'", + ) + break + except NextcloudError as err: + if err.status_code != 405: + raise + uri = f"{base}-{n}" + n += 1 + else: + raise NextcloudError(f"No free URI for calendar '{name}'", 409) + result: dict[str, Any] = {"id": uri, "name": name} + if principal: + await client.dav_request( + "POST", + _caldav_path(user, uri), + body=_share_body(principal, group_write), + headers={"Content-Type": "application/xml; charset=utf-8"}, + context=f"Share calendar '{uri}'", + ) + result["share"] = {"type": "group", "share_with": share_with_group.strip(), "write": group_write} + return json.dumps(result) + + +def _register_calendar_sharing(mcp: FastMCP) -> None: + @mcp.tool(annotations=ADDITIVE_IDEMPOTENT) + @require_permission(PermissionLevel.WRITE) + async def share_calendar(calendar_id: str, share_with: str, share_type: str = "user", write: bool = False) -> str: + """Share one of your calendars with a user or group of this Nextcloud, or change their rights. + + Args: + calendar_id: Your calendar's identifier (only the owner can share). + share_with: User ID or group ID. + share_type: "user" or "group". + write: Whether they may add and change events (default: read only). + + Returns: + Confirmation message. + """ + principal = _principal(share_with, share_type) + await get_client().dav_request( + "POST", + _caldav_path(get_config().user, _check_calendar_id(calendar_id)), + body=_share_body(principal, write), + headers={"Content-Type": "application/xml; charset=utf-8"}, + context=f"Share calendar '{calendar_id}'", + ) + rights = "read-write" if write else "read" + return f"Calendar '{calendar_id}' shared with {share_type} '{share_with.strip()}' ({rights})." + + @mcp.tool(annotations=DESTRUCTIVE) + @require_permission(PermissionLevel.WRITE) + async def unshare_calendar(calendar_id: str, share_with: str, share_type: str = "user") -> str: + """Stop sharing one of your calendars with a user or group. + + Args: + calendar_id: Your calendar's identifier. + share_with: User ID or group ID. + share_type: "user" or "group". + + Returns: + Confirmation message. + """ + principal = _principal(share_with, share_type) + await get_client().dav_request( + "POST", + _caldav_path(get_config().user, _check_calendar_id(calendar_id)), + body=_share_body(principal, None), + headers={"Content-Type": "application/xml; charset=utf-8"}, + context=f"Unshare calendar '{calendar_id}'", + ) + return f"Calendar '{calendar_id}' no longer shared with {share_type} '{share_with.strip()}'." + + @mcp.tool(annotations=ADDITIVE_IDEMPOTENT) + @require_permission(PermissionLevel.WRITE) + async def publish_calendar(calendar_id: str) -> str: + """Create a public read-only link for one of your calendars (anyone with the link can read it). + + Args: + calendar_id: Your calendar's identifier. + + Returns: + JSON with id and public_url. + """ + calendar_id = _check_calendar_id(calendar_id) + await get_client().dav_request( + "POST", + _caldav_path(get_config().user, calendar_id), + body='', + headers={"Content-Type": "application/xml; charset=utf-8"}, + context=f"Publish calendar '{calendar_id}'", + ) + info = await _calendar_info(calendar_id) + return json.dumps({"id": calendar_id, "public_url": info["public_url"]}) + + @mcp.tool(annotations=DESTRUCTIVE) + @require_permission(PermissionLevel.WRITE) + async def unpublish_calendar(calendar_id: str) -> str: + """Remove the public link of one of your calendars. + + Args: + calendar_id: Your calendar's identifier. + + Returns: + Confirmation message. + """ + calendar_id = _check_calendar_id(calendar_id) + await get_client().dav_request( + "POST", + _caldav_path(get_config().user, calendar_id), + body='', + headers={"Content-Type": "application/xml; charset=utf-8"}, + context=f"Unpublish calendar '{calendar_id}'", + ) + return f"Public link of calendar '{calendar_id}' removed." + + @mcp.tool(annotations=DESTRUCTIVE) + @require_permission(PermissionLevel.DESTRUCTIVE) + async def delete_calendar(calendar_id: str, expected_name: str = "") -> str: + """Delete one of your calendars with all its events (Nextcloud keeps it in the trash bin + for a while). For a calendar shared with you, this removes it from your account only. + + Args: + calendar_id: Calendar identifier. + expected_name: Optional current display name; nothing is deleted if it differs. + + Returns: + Confirmation message. + """ + calendar_id = _check_calendar_id(calendar_id) + if expected_name: + info = await _calendar_info(calendar_id) + names = { + info["name"].strip().casefold(), + info["name"].removesuffix(f" ({info['owner']})").strip().casefold(), + } + if expected_name.strip().casefold() not in names: + raise ValueError( + f"Calendar '{calendar_id}' is named '{info['name']}', not '{expected_name}'. Nothing was deleted." + ) + await get_client().dav_request( + "DELETE", _caldav_path(get_config().user, calendar_id), context=f"Delete calendar '{calendar_id}'" + ) + return f"Calendar '{calendar_id}' deleted." + + def register(mcp: FastMCP) -> None: """Register calendar tools with the MCP server.""" _register_read_tools(mcp) _register_create_event(mcp) _register_update_event(mcp) _register_destructive_tools(mcp) + _register_calendar_admin(mcp) + _register_calendar_sharing(mcp) diff --git a/tests/integration/test_calendar_admin.py b/tests/integration/test_calendar_admin.py new file mode 100644 index 0000000..e495f88 --- /dev/null +++ b/tests/integration/test_calendar_admin.py @@ -0,0 +1,121 @@ +"""Calendar administration against a real Nextcloud: create (optionally shared with a group), +share/unshare, public link, delete; shares are visible to the owner only.""" + +import contextlib +import json +import secrets +from collections.abc import AsyncGenerator + +import pytest +from mcp.server.fastmcp.exceptions import ToolError + +from nc_mcp_server.client import NextcloudClient +from nc_mcp_server.config import Config +from nc_mcp_server.permissions import PermissionLevel +from nc_mcp_server.server import create_server +from nc_mcp_server.state import get_client + +from .conftest import McpTestHelper, _get_integration_config + +pytestmark = pytest.mark.integration + + +def _server(config: Config) -> McpTestHelper: + return McpTestHelper(create_server(config), get_client()) + + +@pytest.fixture +async def member_and_group() -> AsyncGenerator[tuple[Config, Config, str]]: + """admin (owner) and a member in a fresh group.""" + admin_config = _get_integration_config() + admin = NextcloudClient(admin_config) + suffix = secrets.token_hex(4) + group = f"mcp-grp-{suffix}" + user_id = f"kal-mitglied-{suffix}" + password = f"Mcp-{secrets.token_hex(10)}-x" + try: + await admin.ocs_post("cloud/groups", data={"groupid": group}) + await admin.ocs_post("cloud/users", data={"userid": user_id, "password": password, "groups[]": group}) + member = Config( + nextcloud_url=admin_config.nextcloud_url, + user=user_id, + password=password, + permission_level=PermissionLevel.DESTRUCTIVE, + ) + yield admin_config, member, group + finally: + with contextlib.suppress(Exception): + await admin.ocs_delete(f"cloud/users/{user_id}") + with contextlib.suppress(Exception): + await admin.ocs_delete(f"cloud/groups/{group}") + await admin.close() + + +class TestCalendarAdmin: + @pytest.mark.asyncio + async def test_lifecycle(self, member_and_group: tuple[Config, Config, str]) -> None: + admin_config, member, group = member_and_group + owner = _server(admin_config) + created = json.loads( + await owner.call("create_calendar", name="Sommerfest Ü 2027", share_with_group=group, group_write=True) + ) + cal_id = created["id"] + try: + assert cal_id.startswith("sommerfest-ue-2027") + assert created["share"] == {"type": "group", "share_with": group, "write": True} + again = json.loads(await owner.call("create_calendar", name="Sommerfest Ü 2027")) + assert again["id"] != cal_id, "unique URI" + await owner.call("delete_calendar", calendar_id=again["id"]) + + info = json.loads(await owner.call("get_calendar_shares", calendar_id=cal_id)) + assert info["owned_by_me"] is True + assert info["shares"] == [{"type": "group", "share_with": group, "write": True}] + assert info["public_url"] is None + + published = json.loads(await owner.call("publish_calendar", calendar_id=cal_id)) + assert published["public_url"] + await owner.call("unpublish_calendar", calendar_id=cal_id) + assert json.loads(await owner.call("get_calendar_shares", calendar_id=cal_id))["public_url"] is None + + await owner.call("share_calendar", calendar_id=cal_id, share_with=member.user, share_type="user") + shares = json.loads(await owner.call("get_calendar_shares", calendar_id=cal_id))["shares"] + assert {"type": "user", "share_with": member.user, "write": False} in shares + await owner.call("unshare_calendar", calendar_id=cal_id, share_with=member.user, share_type="user") + shares = json.loads(await owner.call("get_calendar_shares", calendar_id=cal_id))["shares"] + assert all(s["share_with"] != member.user for s in shares) + finally: + await owner.client.close() + + sharee = _server(member) + shared_id = f"{cal_id}_shared_by_{admin_config.user}" + try: + ids = [c["id"] for c in json.loads(await sharee.call("list_calendars"))] + assert shared_id in ids, "group share visible to the member" + info = json.loads(await sharee.call("get_calendar_shares", calendar_id=shared_id)) + assert info["owned_by_me"] is False + assert info["shares"] is None, "shares hidden from sharees" + with pytest.raises(ToolError): + await sharee.call("share_calendar", calendar_id=shared_id, share_with="admin", share_type="user") + finally: + await sharee.client.close() + + owner = _server(admin_config) + try: + with pytest.raises(ToolError, match="Nothing was deleted"): + await owner.call("delete_calendar", calendar_id=cal_id, expected_name="Anderer") + await owner.call("delete_calendar", calendar_id=cal_id, expected_name="sommerfest ü 2027") + ids = [c["id"] for c in json.loads(await owner.call("list_calendars"))] + assert cal_id not in ids + finally: + await owner.client.close() + + @pytest.mark.asyncio + async def test_rejects_bad_input(self, nc_mcp: McpTestHelper) -> None: + for kwargs in ({"name": ""}, {"name": "x", "color": "red"}, {"name": "x", "share_with_group": "a/b"}): + with pytest.raises(ToolError): + await nc_mcp.call("create_calendar", **kwargs) + for cal in ("../x", "inbox", "a/b"): + with pytest.raises(ToolError, match="Invalid calendar_id"): + await nc_mcp.call("delete_calendar", calendar_id=cal) + with pytest.raises(ToolError, match="share_type"): + await nc_mcp.call("share_calendar", calendar_id="personal", share_with="x", share_type="email") diff --git a/tests/test_calendar_helpers.py b/tests/test_calendar_helpers.py index 982de81..0604dae 100644 --- a/tests/test_calendar_helpers.py +++ b/tests/test_calendar_helpers.py @@ -10,10 +10,14 @@ CREATED_BY_PROP, _apply_event_updates, _build_ical, + _check_calendar_id, _check_summary, _format_event, _parse_dt, _parse_rrule, + _principal, + _share_body, + _slug, _to_caldav_utc, _validate_url, _vevent, @@ -214,3 +218,26 @@ def test_ok(self) -> None: def test_rejects(self, url: str) -> None: with pytest.raises(ValueError, match="conference_url"): _validate_url(url) + + +class TestCalendarAdminHelpers: + def test_slug(self) -> None: + assert _slug("Sommerfest Ü 2027") == "sommerfest-ue-2027" + assert _slug("!!!") == "calendar" + + def test_principal_and_share_body(self) -> None: + assert _principal(" Vorstand ", "group") == "principals/groups/Vorstand" + assert _principal("anna", "USER") == "principals/users/anna" + with pytest.raises(ValueError, match="share_type"): + _principal("anna", "email") + with pytest.raises(ValueError, match="share_with"): + _principal("a/b", "user") + assert "" in _share_body("principals/users/a", True) + assert "" not in _share_body("principals/users/a", False) + assert "" in _share_body("principals/users/a", None) + + def test_calendar_id(self) -> None: + assert _check_calendar_id("team_shared_by_alice") == "team_shared_by_alice" + for bad in ("../x", "a/b", "inbox", "", "x" * 201): + with pytest.raises(ValueError, match="calendar_id"): + _check_calendar_id(bad) From 2507653cb0e039d88922fe146316d0892d8aab94 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Schotte?= Date: Wed, 7 Oct 2026 06:30:25 +0200 Subject: [PATCH 5/6] Refuse ".." in search_files paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit search_files built its SEARCH scope from the path as given, so "a/../.." went to Nextcloud and failed there with an unclear error. The scope is now built from the path's segments: empty and "." segments are dropped, and ".." is refused with a clear error before any request is sent. The scope stays unencoded on purpose. Nextcloud's search plugin takes a relative scope href as a path without URL-decoding it, which the existing integration test for a folder named "search #1 %41 Ü" relies on. Signed-off-by: Björn Schotte Co-Authored-By: Claude Opus 5.5 --- src/nc_mcp_server/tools/files.py | 21 +++++++++++++++++---- tests/integration/test_files.py | 5 +++++ tests/test_files_helpers.py | 32 ++++++++++++++++++++++++++++++++ 3 files changed, 54 insertions(+), 4 deletions(-) diff --git a/src/nc_mcp_server/tools/files.py b/src/nc_mcp_server/tools/files.py index 3f382e4..35fe620 100644 --- a/src/nc_mcp_server/tools/files.py +++ b/src/nc_mcp_server/tools/files.py @@ -112,6 +112,21 @@ def _like_literal(text: str) -> str: return text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") +def _search_scope(user: str, path: str) -> str: + """The SEARCH scope href of a directory in the user's files. + + Unlike a request URL, the scope is not percent-encoded: Nextcloud's search plugin takes a + relative scope href as a path as it stands, without URL-decoding it, so "%41" or "#" in a + folder name must be sent raw. Empty and "." segments are dropped. ".." is refused, so the + scope cannot point outside the given folder; Nextcloud would refuse it later with a less + clear error. + """ + segments = [segment for segment in path.split("/") if segment not in ("", ".")] + if ".." in segments: + raise ValueError(f"Invalid path {path!r}: '..' is not allowed.") + return "/".join(["/files", user, *segments]) + + def _build_search_xml(user: str, query: str, path: str, limit: int, offset: int, mimetype: str) -> str: """Build a WebDAV SEARCH request body.""" where_parts: list[str] = [] @@ -127,9 +142,7 @@ def _build_search_xml(user: str, query: str, path: str, limit: int, offset: int, where_clause = where_parts[0] else: where_clause = "" + "".join(where_parts) + "" - safe_user = xml_escape(user) - safe_path = xml_escape(path.strip("/")) - scope = f"/files/{safe_user}/{safe_path}" if safe_path else f"/files/{safe_user}" + scope = xml_escape(_search_scope(user, path)) return ( '' '' @@ -230,7 +243,7 @@ async def search_files( (% and _ are not wildcards). Example: "report" matches "quarterly-report.pdf", "report-2026.docx". path: Directory to search in (default: "/" for entire user folder). - Example: "Documents" to only search in Documents. + Example: "Documents" to only search in Documents. ".." is not allowed. mimetype: Filter by MIME type prefix. Example: "image" for all images, "application/pdf" for PDFs, "text" for all text files. limit: Maximum number of results (1-100, default: 20). diff --git a/tests/integration/test_files.py b/tests/integration/test_files.py index fb93444..68ee5f5 100644 --- a/tests/integration/test_files.py +++ b/tests/integration/test_files.py @@ -532,6 +532,11 @@ async def test_search_scoped_to_path(self, nc_mcp: McpTestHelper) -> None: assert len(data) >= 1 assert all(TEST_BASE_DIR in e["path"] for e in data) + @pytest.mark.asyncio + async def test_search_refuses_parent_segments(self, nc_mcp: McpTestHelper) -> None: + with pytest.raises((ToolError, ValueError), match="not allowed"): + await nc_mcp.call("search_files", query="x", path=f"{TEST_BASE_DIR}/../..") + @pytest.mark.asyncio async def test_search_empty_query_and_mimetype_raises(self, nc_mcp: McpTestHelper) -> None: with pytest.raises((ToolError, ValueError)): diff --git a/tests/test_files_helpers.py b/tests/test_files_helpers.py index 348227d..ff2f2ea 100644 --- a/tests/test_files_helpers.py +++ b/tests/test_files_helpers.py @@ -11,6 +11,7 @@ _open_no_follow, _resolve_content_type, _resolve_local_upload_path, + _search_scope, ) @@ -202,3 +203,34 @@ def test_query_is_escaped_and_mimetype_keeps_its_wildcard(self) -> None: body = _build_search_xml("admin", "a_b<", "/", 20, 0, "image") assert "%a\\_b<%" in body assert "image/%" in body + + +class TestSearchScope: + @pytest.mark.parametrize( + ("path", "scope"), + [ + ("/", "/files/admin"), + ("", "/files/admin"), + ("Documents", "/files/admin/Documents"), + ("/Documents/Sub/", "/files/admin/Documents/Sub"), + ("a//./b", "/files/admin/a/b"), + ("..foo/bar..", "/files/admin/..foo/bar.."), + ("%2e%2e", "/files/admin/%2e%2e"), + ("search #1 %41 Ü", "/files/admin/search #1 %41 Ü"), + ], + ) + def test_path_is_normalized_and_kept_literal(self, path: str, scope: str) -> None: + assert _search_scope("admin", path) == scope + + @pytest.mark.parametrize("path", ["..", "../bob", "/Documents/../../bob", "a/./..", "a/../b"]) + def test_parent_segments_are_refused(self, path: str) -> None: + with pytest.raises(ValueError, match=r"'\.\.' is not allowed"): + _search_scope("admin", path) + + def test_scope_in_body_is_xml_escaped(self) -> None: + body = _build_search_xml("admin", "x", "R&D/", 20, 0, "") + assert "/files/admin/R&D/<x>" in body + + def test_body_refuses_parent_segments(self) -> None: + with pytest.raises(ValueError, match="not allowed"): + _build_search_xml("admin", "x", "../bob", 20, 0, "") From f345cf8310a2daa167bda6a4d7e0a10c3683a731 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Schotte?= Date: Wed, 7 Oct 2026 18:04:00 +0200 Subject: [PATCH 6/6] Count the new tools and tests, use neutral test data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit README and PROGRESS.md list the nine new tools (231 plus upload_file_from_path), sync_progress.py maps the new calendar test modules to the Calendar row, and the counts are recomputed with scripts/sync_progress.py --write. Fixes the pyright error in the shared calendar fixture's type. Signed-off-by: Björn Schotte Co-Authored-By: Claude Opus 5.5 --- PROGRESS.md | 12 ++++++------ README.md | 6 +++--- scripts/sync_progress.py | 7 ++++++- src/nc_mcp_server/tools/calendar.py | 2 +- tests/integration/test_calendar_admin.py | 2 +- tests/integration/test_calendar_guard.py | 20 ++++++++++---------- tests/test_calendar_helpers.py | 10 +++++----- 7 files changed, 32 insertions(+), 27 deletions(-) diff --git a/PROGRESS.md b/PROGRESS.md index 6041dfc..d786ad9 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -83,7 +83,7 @@ | Module | Tools | Tests | |--------|-------|-------| -| Files | 10 | 166 | +| Files | 10 | 167 | | Users | 7 | 72 | | Groups | 4 | 15 | | Notifications | 3 | 11 | @@ -92,7 +92,7 @@ | Talk Threads | 5 | 78 | | Talk Chat | 9 | 58 | | Talk Settings | 6 | 41 | -| Talk Admin | 5 | 61 | +| Talk Admin | 7 | 73 | | Talk Tags | 6 | 20 | | Activity | 3 | 65 | | Comments | 4 | 29 | @@ -105,25 +105,25 @@ | Mail | 10 | 102 | | Collectives | 25 | 106 | | App Management | 4 | 14 | -| Calendar | 6 | 44 | +| Calendar | 13 | 98 | | Contacts | 6 | 78 | | Tasks | 7 | 51 | | Search | 2 | 17 | | User Permissions | — | 25 | | Server | — | 8 | | Permissions | — | 34 | -| Errors | — | 55 | +| Errors | — | 58 | | Client | — | 75 | | Config | — | 26 | | State | — | 2 | -| File Helpers | — | 36 | +| File Helpers | — | 51 | | File Reminders | 3 | 22 | | Forms | 25 | 34 | | Circles | 14 | 64 | | Cospend | 16 | 41 | | Flow | 5 | 58 | | Pagination | — | 28 | -| **Total** | **223** | **1827** | +| **Total** | **232** | **1912** | The test counts are what pytest collects, recomputed with `python scripts/sync_progress.py --write`, which also fails when a test file is not assigned to a row. diff --git a/README.md b/README.md index 77983e2..cb334e4 100644 --- a/README.md +++ b/README.md @@ -30,9 +30,9 @@ export NEXTCLOUD_PASSWORD=your-app-password nc-mcp-server ``` -## 224 Tools Across 24 Nextcloud Apps +## 231 Tools Across 24 Nextcloud Apps -A 225th tool, `upload_file_from_path`, is registered only when the operator sets +A 232nd tool, `upload_file_from_path`, is registered only when the operator sets `NEXTCLOUD_MCP_UPLOAD_ROOT`. See [Files](#files) for details. | Category | Tools | Protocol | @@ -52,7 +52,7 @@ A 225th tool, `upload_file_from_path`, is registered only when the operator sets | [Talk](#talk) | conversations, messages, threads, participants, edits, reactions, read state, shared items, pins, reminders, personal settings and tags, participant and conversation management | OCS | | [Talk Polls](#talk-polls) | get, create, vote, close polls | OCS | | [Announcements](#announcements) | list, create, delete announcements | OCS | -| [Calendar](#calendar) | list calendars, CRUD events | CalDAV | +| [Calendar](#calendar) | list, create, share, publish and delete calendars; CRUD events with time zones and recurrence checks | CalDAV | | [Contacts](#contacts) | list address books, CRUD contacts | CardDAV | | [Tasks](#tasks) | list lists, CRUD tasks, complete | CalDAV | | [Mail](#mail) | accounts, mailboxes, messages, send, move, flags, tags | OCS + REST | diff --git a/scripts/sync_progress.py b/scripts/sync_progress.py index c8d0203..7a404bf 100644 --- a/scripts/sync_progress.py +++ b/scripts/sync_progress.py @@ -31,7 +31,12 @@ "Mail": ["integration/test_mail.py", "test_mail_tools.py"], "Collectives": ["integration/test_collectives.py", "test_collectives_pages.py", "test_collectives_sharing.py"], "App Management": ["integration/test_app_management.py"], - "Calendar": ["integration/test_calendar.py"], + "Calendar": [ + "integration/test_calendar.py", + "integration/test_calendar_admin.py", + "integration/test_calendar_guard.py", + "test_calendar_helpers.py", + ], "Contacts": ["integration/test_contacts.py"], "Tasks": ["integration/test_tasks.py"], "Search": ["integration/test_search.py"], diff --git a/src/nc_mcp_server/tools/calendar.py b/src/nc_mcp_server/tools/calendar.py index 3d58d8d..0fc6e95 100644 --- a/src/nc_mcp_server/tools/calendar.py +++ b/src/nc_mcp_server/tools/calendar.py @@ -1030,7 +1030,7 @@ async def create_calendar(name: str, color: str = "", share_with_group: str = "" """Create a new calendar (events only) for the current user, optionally shared with a group. Args: - name: Display name, e.g. "Sommerfest 2027". + name: Display name, e.g. "Summer party 2027". color: Optional color as #RRGGBB. share_with_group: Optional Nextcloud group ID to share the new calendar with right away. group_write: Whether that group may also add and change events (default: read only). diff --git a/tests/integration/test_calendar_admin.py b/tests/integration/test_calendar_admin.py index e495f88..ebfd82a 100644 --- a/tests/integration/test_calendar_admin.py +++ b/tests/integration/test_calendar_admin.py @@ -31,7 +31,7 @@ async def member_and_group() -> AsyncGenerator[tuple[Config, Config, str]]: admin = NextcloudClient(admin_config) suffix = secrets.token_hex(4) group = f"mcp-grp-{suffix}" - user_id = f"kal-mitglied-{suffix}" + user_id = f"mcp-member-{suffix}" password = f"Mcp-{secrets.token_hex(10)}-x" try: await admin.ocs_post("cloud/groups", data={"groupid": group}) diff --git a/tests/integration/test_calendar_guard.py b/tests/integration/test_calendar_guard.py index 37784ad..7243dc9 100644 --- a/tests/integration/test_calendar_guard.py +++ b/tests/integration/test_calendar_guard.py @@ -162,15 +162,15 @@ async def test_expected_summary_protects_delete(self, nc_mcp: McpTestHelper) -> @pytest.fixture -async def shared_calendar() -> AsyncGenerator[tuple[Config, Config, str, str]]: +async def shared_calendar() -> AsyncGenerator[tuple[Config, Config, Config, str]]: """admin owns a calendar shared read-write with a new member account and read-only with another.""" admin_config = _get_integration_config() admin = NextcloudClient(admin_config) suffix = secrets.token_hex(4) - uri = f"mcp-test-verein-{suffix}" + uri = f"mcp-test-shared-{suffix}" members: list[Config] = [] try: - for name in ("kal-anna", "kal-leser"): + for name in ("mcp-writer", "mcp-reader"): user_id = f"{name}-{suffix}" password = f"Mcp-{secrets.token_hex(10)}!" await admin.ocs_post("cloud/users", data={"userid": user_id, "password": password}) @@ -187,7 +187,7 @@ async def shared_calendar() -> AsyncGenerator[tuple[Config, Config, str, str]]: f"calendars/{admin_config.user}/{uri}", body=( '' - "Vereinstermine Test" + "Shared events test" ), headers={"Content-Type": "application/xml; charset=utf-8"}, ) @@ -214,12 +214,12 @@ async def shared_calendar() -> AsyncGenerator[tuple[Config, Config, str, str]]: class TestSharedCalendar: @pytest.mark.asyncio async def test_member_writes_owner_needs_allow_foreign( - self, shared_calendar: tuple[Config, Config, str, str] + self, shared_calendar: tuple[Config, Config, Config, str] ) -> None: - admin_config, anna, leser, uri = shared_calendar + admin_config, writer_config, reader_config, uri = shared_calendar shared_id = f"{uri}_shared_by_{admin_config.user}" - member = _server(anna) + member = _server(writer_config) try: calendars = {c["id"]: c for c in json.loads(await member.call("list_calendars"))} assert shared_id in calendars @@ -228,12 +228,12 @@ async def test_member_writes_owner_needs_allow_foreign( await member.call( "create_event", calendar_id=shared_id, - summary="mcp-test-stammtisch", + summary="mcp-test-meetup", start="2027-05-04T19:00:00", timezone="Europe/Berlin", ) ) - await member.call("update_event", calendar_id=shared_id, event_uid=created["uid"], description="Neu") + await member.call("update_event", calendar_id=shared_id, event_uid=created["uid"], description="New") finally: await member.client.close() @@ -246,7 +246,7 @@ async def test_member_writes_owner_needs_allow_foreign( finally: await owner.client.close() - reader = _server(leser) + reader = _server(reader_config) try: calendars = {c["id"]: c for c in json.loads(await reader.call("list_calendars"))} assert calendars[shared_id]["writable"] is False diff --git a/tests/test_calendar_helpers.py b/tests/test_calendar_helpers.py index 0604dae..814d162 100644 --- a/tests/test_calendar_helpers.py +++ b/tests/test_calendar_helpers.py @@ -125,7 +125,7 @@ class TestBuildIcal: def test_zoned_event_has_tzid_vtimezone_and_creator(self) -> None: start = datetime(2026, 10, 20, 19, 0, tzinfo=BERLIN) end = datetime(2026, 10, 20, 21, 0, tzinfo=BERLIN) - text = _build_ical("u1", "Stammtisch", start, end, rrule="FREQ=MONTHLY;BYDAY=3TU", created_by="anna") + text = _build_ical("u1", "Meetup", start, end, rrule="FREQ=MONTHLY;BYDAY=3TU", created_by="anna") assert "DTSTART;TZID=Europe/Berlin:20261020T190000" in text assert "BEGIN:VTIMEZONE" in text assert f"{CREATED_BY_PROP}:anna" in text @@ -167,7 +167,7 @@ def test_attendees_never_listed(self) -> None: class TestUpdates: def _event(self, **kw: object) -> ICal: start = datetime(2026, 10, 20, 19, 0, tzinfo=BERLIN) - return ICal.from_ical(_build_ical("u", "Stammtisch", start, start.replace(hour=21), **kw)) # type: ignore[arg-type] + return ICal.from_ical(_build_ical("u", "Meetup", start, start.replace(hour=21), **kw)) # type: ignore[arg-type] def test_new_start_keeps_event_zone(self) -> None: component = _vevent(self._event()) @@ -204,10 +204,10 @@ def test_rrule_set_and_cleared(self) -> None: def test_check_summary(self) -> None: component = _vevent(self._event()) - _check_summary(component, " stammtisch ", "u") + _check_summary(component, " meetup ", "u") _check_summary(component, "", "u") with pytest.raises(ValueError, match="Nothing was changed"): - _check_summary(component, "Vorstandssitzung", "u") + _check_summary(component, "Board meeting", "u") class TestUrl: @@ -226,7 +226,7 @@ def test_slug(self) -> None: assert _slug("!!!") == "calendar" def test_principal_and_share_body(self) -> None: - assert _principal(" Vorstand ", "group") == "principals/groups/Vorstand" + assert _principal(" Board ", "group") == "principals/groups/Board" assert _principal("anna", "USER") == "principals/users/anna" with pytest.raises(ValueError, match="share_type"): _principal("anna", "email")