scripts/approval_gated_calendar.py — Approval-Gated Calendar

#!/usr/bin/env python3
# By Zero Width Lab (zerowidthlab.com)
"""Approval-Gated Calendar command line tool."""
from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import os
import sys
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any

__author__ = "Zero Width Lab"
SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
GRAPH = "https://graph.microsoft.com/v1.0"
LOGIN = "https://login.microsoftonline.com"
ACK = "zero-width-lab-test-suite-only"
MAX_INPUT_BYTES = 1_048_576
_responses: list[dict[str, Any]] | None = None


class CalendarError(Exception):
    pass


def display_text(value: Any) -> str:
    parts = []
    for char in str(value):
        point = ord(char)
        if char == "\\":
            parts.append("\\\\")
        elif unicodedata.category(char).startswith("C") or point in (0x2028, 0x2029):
            parts.append(f"\\u{point:04x}" if point <= 0xffff else f"\\U{point:08x}")
        else:
            parts.append(char)
    return "".join(parts)


def reject_traversal(value: str) -> None:
    decoded = value
    for _ in range(8):
        if "\\" in decoded or any(part in (".", "..") for part in decoded.split("/")):
            raise CalendarError("path traversal is not allowed")
        next_value = urllib.parse.unquote(decoded)
        if next_value.count("/") != decoded.count("/"):
            raise CalendarError("encoded path separator is not allowed")
        if next_value == decoded:
            return
        decoded = next_value
    raise CalendarError("excessively encoded path is not allowed")


def bounded_json(stream: Any, label: str) -> Any:
    raw = stream.read(MAX_INPUT_BYTES + 1)
    if len(raw) > MAX_INPUT_BYTES:
        raise CalendarError(f"{label} exceeds 1048576-byte limit")
    try:
        def invalid_constant(value: str) -> None:
            raise ValueError("non-finite JSON constant")
        result = json.loads(raw, parse_constant=invalid_constant) if raw else {}
    except (ValueError, UnicodeError, RecursionError) as error:
        raise CalendarError(f"{label} is not valid JSON") from error
    pending = [(result, 0)]
    while pending:
        value, depth = pending.pop()
        if depth > 64:
            raise CalendarError(f"{label} is not valid JSON: nesting exceeds 64 levels")
        if isinstance(value, dict):
            pending.extend((child, depth + 1) for child in value.values())
        elif isinstance(value, list):
            pending.extend((child, depth + 1) for child in value)
    return result


def guard(url: str) -> None:
    if not isinstance(url, str):
        raise CalendarError("destination URL must be text")
    try:
        parsed = urllib.parse.urlsplit(url)
        port = parsed.port
    except ValueError as error:
        raise CalendarError("invalid destination URL or port") from error
    reject_traversal(parsed.path)
    if parsed.scheme != "https" or parsed.hostname not in ("graph.microsoft.com", "login.microsoftonline.com") or port not in (None, 443) or "@" in parsed.netloc:
        raise CalendarError("destination must use HTTPS at a declared host")


class GuardedRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> Any:
        if req.get_method() not in ("GET", "HEAD"):
            raise CalendarError("outcome unknown: write redirected; inspect the event before retrying")
        guard(newurl)
        if urllib.parse.urlsplit(newurl).hostname != urllib.parse.urlsplit(req.full_url).hostname:
            raise CalendarError("redirect must stay at the original host")
        if urllib.parse.urlsplit(newurl).path != urllib.parse.urlsplit(req.full_url).path:
            raise CalendarError("redirect must stay at the selected endpoint")
        preserve_interval(req.full_url, newurl)
        return super().redirect_request(req, fp, code, msg, headers, newurl)


OPENER = urllib.request.build_opener(GuardedRedirect())


def preserve_interval(original: str, target: str) -> None:
    initial = urllib.parse.parse_qs(urllib.parse.urlsplit(original).query)
    following = urllib.parse.parse_qs(urllib.parse.urlsplit(target).query)
    for key in ("startDateTime", "endDateTime"):
        if key not in initial:
            continue
        try:
            expected = initial[key]
            actual = following.get(key, [])
            if len(expected) != 1 or len(actual) != 1:
                raise ValueError("interval must occur exactly once")
            first = dt.datetime.fromisoformat(expected[0].replace("Z", "+00:00"))
            last = dt.datetime.fromisoformat(actual[0].replace("Z", "+00:00"))
            if first.tzinfo is None or last.tzinfo is None or first != last:
                raise ValueError("interval changed")
        except ValueError as error:
            raise CalendarError("pagination or redirect changed the selected interval") from error


def calendar_path(path: str) -> tuple[str, ...]:
    parts = urllib.parse.unquote(path).split("/")
    normalized: list[str] = []
    for part in parts:
        if part.startswith("users('") and part.endswith("')"):
            normalized.extend(("users", part[7:-2].replace("''", "'")))
        else:
            normalized.append(part)
    return tuple(normalized)


def fake() -> list[dict[str, Any]] | None:
    global _responses
    path = os.environ.get("M365_CALENDAR_FAKE_TRANSPORT")
    ack = os.environ.get("M365_CALENDAR_ALLOW_FAKE")
    if not path and not ack:
        return None
    if not path or ack != ACK:
        raise CalendarError("fake transport requires both test environment variables")
    if _responses is None:
        try:
            with Path(path).open("rb") as fixture:
                rows = bounded_json(fixture, "fake fixture")["responses"]
            if not isinstance(rows, list) or not all(isinstance(row, dict) for row in rows):
                raise ValueError("responses must be objects")
            for row in rows:
                if not isinstance(row.get("method"), str) or not isinstance(row.get("url_prefix"), str) or type(row.get("status")) is not int or not 100 <= row["status"] <= 599:
                    raise ValueError("response needs method, url_prefix and HTTP status")
                if "location" in row and not isinstance(row["location"], str):
                    raise ValueError("redirect location must be text")
        except (OSError, ValueError, KeyError, TypeError) as error:
            raise CalendarError(f"fake fixture unavailable: {display_text(error)}") from error
        _responses = rows
        print("WARNING: M365 calendar fake transport engaged; no real network request is made.", file=sys.stderr)
    return _responses


def request(method: str, url: str, access: str = "", body: Any = None, write: bool = False, form: bool = False, extra_headers: dict[str, str] | None = None) -> Any:
    guard(url)
    rows = fake()  # Fixture preflight precedes every network request.
    for attempt in range(3 if not write else 1):
        if rows is not None:
            redirects = 0
            while True:
                matches = [row for row in rows if row.get("method") == method and url.startswith(row.get("url_prefix", "")) and ("match_body" not in row or row["match_body"] == body)]
                match = max(matches, key=lambda row: len(row.get("url_prefix", "")), default=None)
                if match is None:
                    raise CalendarError(f"no fake response for {method} {display_text(url)}")
                if match.get("once"):
                    rows.remove(match)
                status, result = int(match["status"]), match.get("json", {})
                if status not in (301, 302, 303, 307, 308) or "location" not in match:
                    break
                if redirects >= 10:
                    raise CalendarError("redirect limit reached")
                destination = urllib.parse.urljoin(url, match["location"])
                redirected = GuardedRedirect().redirect_request(
                    urllib.request.Request(url, method=method), None, status, "", {}, destination
                )
                if redirected is None:
                    raise CalendarError(f"redirect refused (HTTP {status})")
                method, url = redirected.get_method(), redirected.full_url
                redirects += 1
        else:
            data = urllib.parse.urlencode(body).encode() if form else (json.dumps(body).encode() if body is not None else None)
            headers = {"Accept": "application/json"}
            headers.update(extra_headers or {})
            if access:
                headers["Authorization"] = f"Bearer {access}"
            if data is not None:
                headers["Content-Type"] = "application/x-www-form-urlencoded" if form else "application/json"
            req = urllib.request.Request(url, data=data, headers=headers, method=method)
            try:
                with OPENER.open(req, timeout=30) as response:
                    status = response.status
                    result = bounded_json(response, "provider response")
            except urllib.error.HTTPError as error:
                status = error.code
                try:
                    result = bounded_json(error, "provider error response")
                except CalendarError:
                    result = {}
                finally:
                    error.close()
            except (urllib.error.URLError, TimeoutError, OSError) as error:
                if write:
                    raise CalendarError("outcome unknown: write transport failed; inspect the event before retrying") from error
                raise CalendarError("network request failed") from error
            except CalendarError as error:
                if write and "outcome unknown" not in str(error) and "redirect" not in str(error):
                    raise CalendarError("outcome unknown: malformed write response; inspect the event before retrying") from error
                raise
        if 200 <= status < 300:
            return result
        if write and status >= 500:
            raise CalendarError(f"outcome unknown: write returned HTTP {status}; inspect the event before retrying")
        if write and status in (409, 412) and url.endswith("/cancel"):
            raise CalendarError(f"cancel conflicts with a concurrent change (HTTP {status}); rerun --dry-run for a fresh approval")
        if not write and status in (429, 503) and attempt < 2:
            continue
        raise CalendarError(f"{method} failed with HTTP {status}")
    raise CalendarError("request retry limit reached")


def token() -> str:
    tenant, client, secret = (os.environ.get(name, "") for name in ("M365_TENANT_ID", "M365_CLIENT_ID", "M365_CLIENT_SECRET"))
    if not all((tenant, client, secret)):
        raise CalendarError("M365_TENANT_ID, M365_CLIENT_ID and M365_CLIENT_SECRET are required")
    segment(tenant)
    if "/" in tenant or any(ord(char) < 32 for char in tenant):
        raise CalendarError("invalid tenant identifier")
    body = {"client_id": client, "client_secret": secret, "scope": "https://graph.microsoft.com/.default", "grant_type": "client_credentials"}
    result = request("POST", f"{LOGIN}/{urllib.parse.quote(tenant, safe='')}/oauth2/v2.0/token", body=body, form=True)
    access = result.get("access_token") if isinstance(result, dict) else None
    if not isinstance(access, str) or not access:
        raise CalendarError("token response omitted access token")
    return access


def segment(value: str) -> str:
    reject_traversal(value)
    if not value or "/" in value or "%" in value or any(ord(char) < 32 for char in value):
        raise CalendarError("identifier must be nonempty and single-line")
    return urllib.parse.quote(value, safe="")


def interval(start: str, end: str) -> tuple[str, str]:
    try:
        first = dt.datetime.fromisoformat(start.replace("Z", "+00:00"))
        last = dt.datetime.fromisoformat(end.replace("Z", "+00:00"))
    except ValueError as error:
        raise CalendarError("time must be ISO 8601") from error
    if first.tzinfo is None or last.tzinfo is None or first >= last:
        raise CalendarError("times need UTC offsets and end after start")
    return first.isoformat(), last.isoformat()


def digest(plan: dict[str, Any]) -> str:
    return hashlib.sha256(json.dumps(plan, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest()


def output(value: Any, as_json: bool) -> None:
    if as_json:
        print(json.dumps(value, indent=2, ensure_ascii=False))
        return
    def visit(item: Any, label: str = "") -> None:
        if isinstance(item, dict):
            for key, child in item.items():
                visit(child, f"{label}.{key}" if label else key)
        elif isinstance(item, list):
            for index, child in enumerate(item):
                visit(child, f"{label}[{index}]")
        else:
            print(f"{display_text(label)}: {display_text(item)}")
    visit(value)


def event_plan(args: argparse.Namespace, user: str) -> dict[str, Any]:
    start, end = interval(args.start, args.end)
    if not args.time_zone.strip():
        raise CalendarError("time zone is required")
    event = {"organizer": user, "attendees": [{"emailAddress": {"address": address}, "type": "required"} for address in args.attendee], "subject": args.subject, "start": {"dateTime": start, "timeZone": args.time_zone}, "end": {"dateTime": end, "timeZone": args.time_zone}, "body": {"contentType": "text", "content": args.body}, "location": {"displayName": args.location}, "isOnlineMeeting": args.online_meeting}
    return {"action": "create", "mailbox": user, "event": event}


def collect(url: str, access: str) -> dict[str, Any]:
    events: list[Any] = []
    skipped: list[str] = []
    seen = set()
    original = urllib.parse.urlsplit(url)
    initial_url = url
    while url:
        if len(seen) >= 100:
            skipped.append("pagination limit reached")
            break
        if url in seen:
            skipped.append("pagination loop")
            break
        seen.add(url)
        try:
            guard(url)
            target = urllib.parse.urlsplit(url)
            if (target.hostname, calendar_path(target.path)) != (original.hostname, calendar_path(original.path)):
                raise CalendarError("pagination must stay beneath the selected calendar endpoint")
            preserve_interval(initial_url, url)
            page = request("GET", url, access)
            if not isinstance(page, dict) or not isinstance(page.get("value"), list):
                raise CalendarError("invalid list response")
            if not all(isinstance(event, dict) and isinstance(event.get("id"), str) and event["id"] and ("subject" not in event or isinstance(event["subject"], str)) for event in page["value"]):
                raise CalendarError("invalid event in list response")
            next_url = page.get("@odata.nextLink", "")
            if not isinstance(next_url, str):
                raise CalendarError("pagination link must be text")
            events.extend(page["value"])
            url = next_url
        except CalendarError as error:
            skipped.append(str(error))
            break
    return {"events": events, "incomplete": bool(skipped), "skipped": skipped}


def require_event(value: Any) -> None:
    if not isinstance(value, dict) or not isinstance(value.get("id"), str) or not value["id"]:
        raise CalendarError("event response requires a nonempty id")
    if "subject" in value and not isinstance(value["subject"], str):
        raise CalendarError("event subject must be text")


def run(args: argparse.Namespace) -> int:
    user = args.user or os.environ.get("M365_DEFAULT_USER", "")
    base = f"{GRAPH}/users/{segment(user)}"
    if args.command in ("show", "cancel"):
        segment(args.event_id)
    if args.command in ("list", "search", "availability", "create"):
        interval(args.start, args.end)
    if args.command == "create":
        plan = event_plan(args, user)
        approval = digest(plan)
        if args.dry_run:
            output({"plan": plan, "approval_digest": approval, "dry_run": True}, args.json)
            return 0
        if args.approve != approval:
            raise CalendarError("missing or mismatched approval digest")
        graph_event = {key: value for key, value in plan["event"].items() if key != "organizer"}
        if args.online_meeting:
            graph_event["onlineMeetingProvider"] = "teamsForBusiness"
        result = request("POST", f"{base}/events", token(), graph_event, write=True)
        try:
            require_event(result)
        except CalendarError as error:
            raise CalendarError("outcome unknown: invalid create event response; inspect the calendar before retrying") from error
        output({"event": result, "approval_digest": approval}, args.json)
        return 0
    if args.command == "cancel":
        url = f"{base}/events/{segment(args.event_id)}"
        if args.dry_run:
            current = request("GET", url, token())
            require_event(current)
            change_key = current.get("changeKey") if isinstance(current, dict) else None
            if not isinstance(change_key, str) or not change_key:
                raise CalendarError("event response omitted changeKey")
            plan = {"action": "cancel", "mailbox": user, "event_id": args.event_id, "changeKey": change_key, "comment": args.comment}
            output({"plan": plan, "current_event": current, "approval_digest": digest(plan), "dry_run": True}, args.json)
            return 0
        if not args.change_key:
            raise CalendarError("--change-key from dry run is required")
        plan = {"action": "cancel", "mailbox": user, "event_id": args.event_id, "changeKey": args.change_key, "comment": args.comment}
        if args.approve != digest(plan):
            raise CalendarError("missing or mismatched approval digest")
        access = token()
        current = request("GET", url, access)
        require_event(current)
        if current.get("changeKey") != args.change_key:
            raise CalendarError("event changed since dry run; approval expired")
        result = request("POST", f"{url}/cancel", access, {"comment": args.comment}, write=True)
        if result != {}:
            raise CalendarError("outcome unknown: cancel response must be empty; inspect the event before retrying")
        output({"cancelled_event_id": args.event_id, "changeKey": args.change_key, "result": result}, args.json)
        return 0
    access = token()
    if args.command in ("list", "search"):
        start, end = interval(args.start, args.end)
        query = urllib.parse.urlencode({"startDateTime": start, "endDateTime": end, "$top": "100"})
        data = collect(f"{base}/calendarView?{query}", access)
        if args.command == "search":
            data["events"] = [event for event in data["events"] if args.query.casefold() in str(event.get("subject", "")).casefold()]
        output(data, args.json)
        return 2 if data["incomplete"] and not args.allow_partial else 0
    if args.command == "show":
        event = request("GET", f"{base}/events/{segment(args.event_id)}", access)
        require_event(event)
        output(event, args.json)
        return 0
    start, end = interval(args.start, args.end)
    body = {"schedules": args.attendee, "startTime": {"dateTime": start, "timeZone": args.time_zone}, "endTime": {"dateTime": end, "timeZone": args.time_zone}, "availabilityViewInterval": 30}
    result = request("POST", f"{base}/calendar/getSchedule", access, body)
    if not isinstance(result, dict) or not isinstance(result.get("value"), list) or not all(isinstance(row, dict) and isinstance(row.get("scheduleId"), str) for row in result["value"]):
        raise CalendarError("invalid availability response")
    output(result, args.json)
    return 0


def parser() -> argparse.ArgumentParser:
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--user")
    common.add_argument("--json", action="store_true")
    common.add_argument("--allow-partial", action="store_true")
    root = argparse.ArgumentParser(description="Approval-Gated Calendar", epilog=SIGNATURE)
    commands = root.add_subparsers(dest="command", required=True)
    for name in ("list", "search", "show", "availability", "create", "cancel"):
        command = commands.add_parser(name, parents=[common], epilog=SIGNATURE)
        if name in ("list", "search", "availability", "create"):
            command.add_argument("--start", required=True)
            command.add_argument("--end", required=True)
        if name == "search":
            command.add_argument("--query", required=True)
        if name in ("show", "cancel"):
            command.add_argument("event_id")
        if name in ("availability", "create"):
            command.add_argument("--attendee", action="append", default=[], required=name == "availability")
            command.add_argument("--time-zone", default="UTC")
        if name == "create":
            for field in ("subject", "body", "location"):
                command.add_argument(f"--{field}", required=field == "subject", default="")
            command.add_argument("--online-meeting", action="store_true")
        if name == "cancel":
            command.add_argument("--comment", required=True)
            command.add_argument("--change-key")
        if name in ("create", "cancel"):
            mode = command.add_mutually_exclusive_group()
            mode.add_argument("--dry-run", action="store_true")
            mode.add_argument("--approve")
    return root


def main() -> int:
    args = parser().parse_args()
    try:
        return run(args)
    except CalendarError as error:
        print(f"Error: {display_text(error)}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

License: MIT (read it).

Back to the listing