feat(invites): add cancel() for events you host

.vscode/cspell.json

@@ -127,6 +127,7 @@
     "fteu",
     "ftyp",
     "gare",
+    "gatewayws",
     "generatedcode",
     "geofence",
     "guids",
@@ -222,6 +223,7 @@
     "sharedstreams",
     "sharingtype",
     "slomo",
+    "softwarecapabilityflags",
     "sonarcube",
     "sonarlint",
     "sonarqube",

README.md

@@ -1759,12 +1759,38 @@ enums used elsewhere in this service.
 `resolve()` only looks; `accept()` joins the event, after which it appears in
 `events()` with the shared scope.
 
+### Cancelling an event you host
+
+```python
+event = api.invites.event("EVENT-UUID")
+
+cancelled = api.invites.cancel(event)
+print(cancelled.is_cancelled)  # True
+
+# Changed your mind:
+api.invites.cancel(cancelled, cancelled=False)
+```
+
+Guests see the cancellation, so pass an event you really do host. The call
+works in both directions, so a cancellation is recoverable.
+
+Pass an `event` that carries a current record change tag -- one from `event()`
+or from a previous `cancel()`, as above. Apple rejects a stale one with an
+opaque `502`.
+
+Cancelling is the only event flag this library can write. Apple stores
+`isPublished`, `isPrivate` and `title` as PCS-encrypted fields, and writing
+those needs encryption support pyicloud does not have yet, so publishing an
+event or renaming it still has to happen in the Invites app.
+
 ### Errors
 
 Every call raises a subclass of `InvitesError`: `InvitesAuthError` when the
 session needs renewing, `InvitesRateLimited` when Apple asks you to slow down,
 and `InvitesApiError` for everything else. `event()` raises `EventNotFound` for
-an unknown event id.
+an unknown event id, and `cancel()` raises `InvitesEntitlementError` when the
+account is not entitled to edit events -- Apple gates them behind an iCloud
+subscription feature.
 
 ## Examples
 

pyicloud/services/invites/__init__.py

@@ -4,6 +4,7 @@
     CloudKitInvitesClient,
     InvitesApiError,
     InvitesAuthError,
+    InvitesEntitlementError,
     InvitesError,
     InvitesRateLimited,
 )
@@ -34,6 +35,7 @@
     "EventTime",
     "InvitesApiError",
     "InvitesAuthError",
+    "InvitesEntitlementError",
     "InvitesError",
     "InvitesRateLimited",
     "InvitesService",

pyicloud/services/invites/client.py

@@ -47,6 +47,11 @@
     CloudKitRateLimited,
 )
 from pyicloud.exceptions import PyiCloudAPIResponseException
+from pyicloud.services.invites.entitlement import (
+    GATEWAY_BASE_URL,
+    FeatureAccess,
+    parse_feature_access,
+)
 
 LOGGER = logging.getLogger(__name__)
 DEFAULT_TIMEOUT = (10.0, 60.0)
@@ -90,6 +95,15 @@ def __init__(self, message: str, retry_after: float | None = None) -> None:
         self.retry_after = retry_after
 
 
+class InvitesEntitlementError(InvitesError):
+    """The account may not perform this write.
+
+    Apple gates event edits behind a subscription feature and answers an
+    unproven write with ``502 INTERNAL_ERROR``, which says nothing useful. This
+    is raised before the request instead.
+    """
+
+
 class InvitesApiError(InvitesError):
     """Catch-all API error."""
 
@@ -174,6 +188,31 @@ def _raise_invites_error(exc: Exception) -> NoReturn:
 
     # ----- Records-in-zones scoped wrappers ----------------------------------
 
+    def feature_access(self, feature: str) -> FeatureAccess:
+        """Ask Apple whether this account may use ``feature``, and for a token.
+
+        Not a CloudKit call: the gateway lives outside the container and is not
+        in the advertised webservices map, so it is reached directly.
+        """
+
+        dsid = self._base_params.get("dsid")
+        url = f"{GATEWAY_BASE_URL}/accounts/{dsid}/subscriptions/features"
+        try:
+            response = self._session.get(
+                url,
+                params=self._base_params,
+                headers={"x-apple-softwarecapabilityflags": feature},
+                timeout=self._timeout,
+            )
+            payload = response.json()
+        except PyiCloudAPIResponseException as exc:
+            self._raise_invites_error(exc)
+        except ValueError as exc:
+            raise InvitesApiError(
+                "Entitlement gateway returned a non-JSON response"
+            ) from exc
+        return parse_feature_access(payload, feature)
+
     def zones_list(self, scope: ScopeLiteral) -> CKZoneListResponse:
         """List the zones in the given scope.
 

pyicloud/services/invites/entitlement.py

@@ -0,0 +1,83 @@
+"""Subscription entitlements for Invites writes.
+
+Editing an event is gated behind an iCloud subscription feature. Apple proves
+the entitlement with a short-lived token that the write must carry, and without
+it every modify comes back ``502 INTERNAL_ERROR`` with nothing to act on -- so
+the failure looks like an unsupported operation rather than a missing
+credential.
+
+The token comes from a host that is not in the ``webservices`` map Apple
+advertises at login, so unlike every other endpoint in this library it cannot
+be resolved and is named here.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import logging
+from typing import Any
+
+LOGGER = logging.getLogger(__name__)
+
+#: Not advertised in the webservices map, so unlike every other endpoint
+#: here it cannot be resolved from the login response.
+GATEWAY_BASE_URL = "https://gatewayws.icloud.com/acsegateway/v4"
+
+#: The capability an event write requires.
+CREATE_EVENT_FEATURE = "apps.rsvp.create-event"
+
+#: Re-request a token this long before it expires, so a write near the boundary
+#: does not race the clock.
+EXPIRY_MARGIN_SECONDS = 60.0
+
+
+@dataclass(frozen=True, slots=True)
+class FeatureAccess:
+    """Whether an account may use a feature, and the token that proves it."""
+
+    feature_key: str
+    can_use: bool
+    access_token: str | None = None
+    cache_till: datetime | None = None
+
+    def usable_at(self, moment: datetime) -> bool:
+        """Return whether this grant is still good at ``moment``."""
+
+        if not self.can_use or not self.access_token:
+            return False
+        if self.cache_till is None:
+            return False
+        return (self.cache_till - moment).total_seconds() > EXPIRY_MARGIN_SECONDS
+
+
+def _parse_cache_till(value: Any) -> datetime | None:
+    """Parse Apple's ``cacheTill`` timestamp, tolerating a missing one."""
+
+    if not isinstance(value, str):
+        return None
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        LOGGER.debug("invites.entitlement.unparsed_cache_till value=%r", value)
+        return None
+    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+
+
+def parse_feature_access(payload: Any, feature: str) -> FeatureAccess:
+    """Read one feature's grant out of the gateway's response.
+
+    The response is a list -- one entry per requested feature -- so the wanted
+    grant is matched on ``featureKey`` rather than taken positionally.
+    """
+
+    entries = payload if isinstance(payload, list) else []
+    for entry in entries:
+        if isinstance(entry, dict) and entry.get("featureKey") == feature:
+            return FeatureAccess(
+                feature_key=feature,
+                can_use=bool(entry.get("canUse")),
+                access_token=entry.get("accessToken"),
+                cache_till=_parse_cache_till(entry.get("cacheTill")),
+            )
+    return FeatureAccess(feature_key=feature, can_use=False)