"""book: a minimal request-and-approve booking page backed by CalDAV.
Guests pick a free slot and send a request. The request is held in SQLite,
mirrored into the owner's CalDAV calendar as a TENTATIVE event and mailed to
the owner with an approve/decline link. Approving turns the event CONFIRMED
and mails the guest an invitation; declining, cancelling or letting a request
run past its start time removes the event again.
Free/busy comes from every calendar in the CalDAV user's home collection plus
the held requests themselves. If the calendar cannot be read, no slots are
offered: an unreachable calendar must never look like an empty one.
The only mail a stranger can trigger goes to the owner. Guests receive mail
only after the owner acted on their request, so the form cannot be abused to
send mail to arbitrary addresses.
"""
import argparse
import base64
import json
import logging
import os
import re
import secrets
import smtplib
import sqlite3
import threading
import time
import urllib.error
import urllib.request
import xml.etree.ElementTree as ET
from collections import defaultdict, deque
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
from email.message import EmailMessage
from email.utils import formataddr, formatdate, make_msgid
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, quote, urlencode, urlsplit
from zoneinfo import ZoneInfo, available_timezones
import icalendar
import jinja2
import recurring_ical_events
log = logging.getLogger("book")
UTC = timezone.utc
WEEKDAYS = ("mon", "tue", "wed", "thu", "fri", "sat", "sun")
TIMEZONES = sorted(available_timezones())
SLOT_FORMAT = "%Y%m%dT%H%M%SZ"
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
TOKEN_RE = re.compile(r"^[A-Za-z0-9_-]{20,64}$")
MAX_BODY = 16 * 1024
LIMITS = {"name": 100, "email": 254, "note": 2000, "message": 2000}
DAV = "DAV:"
CALDAV = "urn:ietf:params:xml:ns:caldav"
# --- configuration -----------------------------------------------------------
@dataclass(frozen=True)
class EventType:
slug: str
title: str
minutes: int
description: str
@dataclass(frozen=True)
class Config:
listen_host: str
listen_port: int
base_url: str
owner: str
intro: str
tz: ZoneInfo
event_types: tuple[EventType, ...]
hours: dict[int, tuple[tuple[int, int], ...]] # weekday -> (start, end) minutes
slot_step: int
buffer: int
min_notice: timedelta
horizon: timedelta
caldav_url: str
caldav_user: str
caldav_password: str
caldav_calendar: str
smtp_host: str
smtp_port: int
mail_from: str
notify: str
db: str
templates: str
static: str
def event_type(self, slug: str) -> EventType | None:
return next((e for e in self.event_types if e.slug == slug), None)
def parse_hhmm(s: str) -> int:
h, m = s.split(":")
minutes = int(h) * 60 + int(m)
if not 0 <= minutes <= 24 * 60:
raise ValueError(f"time out of range: {s}")
return minutes
def load_config(path: str, password_file: str, templates: str, static: str) -> Config:
raw = json.loads(Path(path).read_text())
host, _, port = raw["listen"].rpartition(":")
hours: dict[int, tuple[tuple[int, int], ...]] = {}
for day, ranges in raw["hours"].items():
spans = []
for r in ranges:
a, b = (parse_hhmm(x) for x in r.split("-"))
if a >= b:
raise ValueError(f"empty working-hours range {r} on {day}")
spans.append((a, b))
hours[WEEKDAYS.index(day)] = tuple(sorted(spans))
event_types = tuple(
EventType(e["slug"], e["title"], int(e["minutes"]), e.get("description", ""))
for e in raw["eventTypes"]
)
for e in event_types:
if not re.fullmatch(r"[a-z0-9-]+", e.slug) or e.slug in ("a", "r"):
raise ValueError(f"invalid event type slug: {e.slug}")
if not event_types:
raise ValueError("no event types configured")
caldav = raw["caldav"]
smtp = raw["smtp"]
return Config(
listen_host=host,
listen_port=int(port),
base_url=raw["baseUrl"].rstrip("/"),
owner=raw["owner"],
intro=raw.get("intro", ""),
tz=ZoneInfo(raw["timezone"]),
event_types=event_types,
hours=hours,
slot_step=int(raw.get("slotStep", 30)),
buffer=int(raw.get("bufferMinutes", 0)),
min_notice=timedelta(hours=float(raw.get("minNoticeHours", 24))),
horizon=timedelta(days=int(raw.get("horizonDays", 28))),
caldav_url=caldav["url"].rstrip("/"),
caldav_user=caldav["user"],
caldav_password=Path(password_file).read_text().strip(),
caldav_calendar=caldav.get("calendar", "bookings"),
smtp_host=smtp.get("host", "127.0.0.1"),
smtp_port=int(smtp.get("port", 25)),
mail_from=smtp["from"],
notify=smtp["notify"],
db=raw["database"],
templates=templates,
static=static,
)
# --- time helpers ------------------------------------------------------------
def now_utc() -> datetime:
return datetime.now(UTC).replace(microsecond=0)
def to_utc(value: date | datetime, tz: ZoneInfo) -> datetime:
"""Normalise an iCalendar DATE/DATE-TIME to aware UTC.
All-day dates start at local midnight; floating times are the owner's.
"""
if not isinstance(value, datetime):
value = datetime(value.year, value.month, value.day)
if value.tzinfo is None:
value = value.replace(tzinfo=tz)
return value.astimezone(UTC)
def epoch(dt: datetime) -> int:
return int(dt.timestamp())
def from_epoch(ts: int) -> datetime:
return datetime.fromtimestamp(ts, UTC)
def overlaps(a0: datetime, a1: datetime, intervals: list[tuple[datetime, datetime]]) -> bool:
return any(a0 < b1 and b0 < a1 for b0, b1 in intervals)
# --- CalDAV ------------------------------------------------------------------
class CalDAVError(Exception):
pass
class CalDAV:
"""Just enough CalDAV for one user's home: list, query, put, delete."""
def __init__(self, cfg: Config):
self.base = cfg.caldav_url
self.home = f"/{quote(cfg.caldav_user)}/"
self.calendar = f"{self.home}{quote(cfg.caldav_calendar)}/"
token = base64.b64encode(f"{cfg.caldav_user}:{cfg.caldav_password}".encode()).decode()
self.auth = f"Basic {token}"
self.tz = cfg.tz
self._ensured = False
def _request(
self,
method: str,
path: str,
body: bytes | None = None,
headers: dict[str, str] | None = None,
accept: tuple[int, ...] = (),
) -> tuple[int, bytes]:
req = urllib.request.Request(self.base + path, data=body, method=method)
req.add_header("Authorization", self.auth)
for k, v in (headers or {}).items():
req.add_header(k, v)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return resp.status, resp.read()
except urllib.error.HTTPError as e:
if e.code in accept:
return e.code, e.read()
raise CalDAVError(f"{method} {path}: HTTP {e.code}") from e
except (urllib.error.URLError, TimeoutError, OSError) as e:
raise CalDAVError(f"{method} {path}: {e}") from e
def _xml(self, method: str, path: str, depth: str, body: str, accept=()) -> tuple[int, ET.Element | None]:
status, data = self._request(
method,
path,
body.encode(),
{"Depth": depth, "Content-Type": "application/xml; charset=utf-8"},
accept,
)
if status >= 300 or not data.strip():
return status, None
try:
return status, ET.fromstring(data)
except ET.ParseError as e:
raise CalDAVError(f"{method} {path}: unparseable response: {e}") from e
def ensure_calendar(self) -> None:
if self._ensured:
return
status, _ = self._xml(
"PROPFIND",
self.calendar,
"0",
f'',
accept=(404,),
)
if status == 404:
log.info("creating calendar %s", self.calendar)
self._xml(
"MKCALENDAR",
self.calendar,
"0",
f''
"Bookings"
''
"",
)
self._ensured = True
def calendars(self) -> list[str]:
_, root = self._xml(
"PROPFIND",
self.home,
"1",
f''
"",
)
found = []
for resp in root.iter(f"{{{DAV}}}response") if root is not None else ():
href = resp.findtext(f"{{{DAV}}}href")
if not href or resp.find(f".//{{{DAV}}}resourcetype/{{{CALDAV}}}calendar") is None:
continue
comps = [c.get("name") for c in resp.iter(f"{{{CALDAV}}}comp")]
if comps and "VEVENT" not in comps:
continue
found.append(urlsplit(href).path)
return found
def busy(self, start: datetime, end: datetime) -> list[tuple[datetime, datetime]]:
"""Opaque, non-cancelled event occurrences in [start, end), in UTC."""
span = (
f''
)
query = (
f''
''
f"{span}"
)
out: list[tuple[datetime, datetime]] = []
for path in self.calendars():
_, root = self._xml("REPORT", path, "1", query)
for data in root.iter(f"{{{CALDAV}}}calendar-data") if root is not None else ():
if not data.text:
continue
try:
cal = icalendar.Calendar.from_ical(data.text)
occurrences = recurring_ical_events.of(cal, skip_bad_series=True).between(start, end)
except Exception:
log.exception("unparseable event in %s", path)
continue
for ev in occurrences:
if str(ev.get("TRANSP", "OPAQUE")).upper() == "TRANSPARENT":
continue
if str(ev.get("STATUS", "")).upper() == "CANCELLED":
continue
try:
out.append((to_utc(ev.start, self.tz), to_utc(ev.end, self.tz)))
except Exception:
log.exception("event without usable start/end in %s", path)
return out
def put(self, uid: str, ics: bytes) -> None:
self.ensure_calendar()
self._request(
"PUT",
f"{self.calendar}{quote(uid)}.ics",
ics,
{"Content-Type": "text/calendar; charset=utf-8"},
)
def delete(self, uid: str) -> None:
self._request("DELETE", f"{self.calendar}{quote(uid)}.ics", accept=(404,))
# --- storage -----------------------------------------------------------------
SCHEMA = """
CREATE TABLE IF NOT EXISTS bookings (
id INTEGER PRIMARY KEY,
uid TEXT NOT NULL UNIQUE,
event TEXT NOT NULL,
title TEXT NOT NULL,
start INTEGER NOT NULL,
end INTEGER NOT NULL,
name TEXT NOT NULL,
email TEXT NOT NULL,
note TEXT NOT NULL,
guest_tz TEXT NOT NULL,
status TEXT NOT NULL CHECK (status IN ('pending', 'approved', 'declined', 'cancelled', 'expired')),
guest_token TEXT NOT NULL UNIQUE,
admin_token TEXT NOT NULL UNIQUE,
sequence INTEGER NOT NULL DEFAULT 0,
created INTEGER NOT NULL,
updated INTEGER NOT NULL,
ip TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS bookings_active ON bookings (status, start);
"""
ACTIVE = ("pending", "approved")
class Store:
"""One connection behind one lock; traffic here is a handful of requests a day."""
def __init__(self, path: str):
self.lock = threading.RLock()
self.db = sqlite3.connect(path, check_same_thread=False, isolation_level=None)
self.db.row_factory = sqlite3.Row
self.db.execute("PRAGMA journal_mode=WAL")
self.db.executescript(SCHEMA)
def holds(self, start: datetime, end: datetime) -> list[tuple[datetime, datetime]]:
with self.lock:
rows = self.db.execute(
"SELECT start, end FROM bookings WHERE status IN (?, ?) AND end > ? AND start < ?",
(*ACTIVE, epoch(start), epoch(end)),
).fetchall()
return [(from_epoch(r["start"]), from_epoch(r["end"])) for r in rows]
def by_token(self, column: str, token: str) -> sqlite3.Row | None:
assert column in ("guest_token", "admin_token")
with self.lock:
return self.db.execute(f"SELECT * FROM bookings WHERE {column} = ?", (token,)).fetchone()
def count(self, where: str, args: tuple) -> int:
with self.lock:
return self.db.execute(f"SELECT count(*) FROM bookings WHERE {where}", args).fetchone()[0]
def insert(self, **row) -> None:
cols = ", ".join(row)
marks = ", ".join("?" for _ in row)
with self.lock:
self.db.execute(f"INSERT INTO bookings ({cols}) VALUES ({marks})", tuple(row.values()))
def set_status(self, booking_id: int, status: str) -> None:
with self.lock:
self.db.execute(
"UPDATE bookings SET status = ?, sequence = sequence + 1, updated = ? WHERE id = ?",
(status, epoch(now_utc()), booking_id),
)
def get(self, booking_id: int) -> sqlite3.Row:
with self.lock:
return self.db.execute("SELECT * FROM bookings WHERE id = ?", (booking_id,)).fetchone()
def overdue(self, now: datetime) -> list[sqlite3.Row]:
with self.lock:
return self.db.execute(
"SELECT * FROM bookings WHERE status = 'pending' AND start <= ?", (epoch(now),)
).fetchall()
# --- the application ---------------------------------------------------------
class Unavailable(Exception):
"""The calendar could not be read; refuse to guess."""
class App:
def __init__(self, cfg: Config):
self.cfg = cfg
self.cal = CalDAV(cfg)
self.store = Store(cfg.db)
self.rate: dict[str, deque[float]] = defaultdict(deque)
self.jinja = jinja2.Environment(
loader=jinja2.FileSystemLoader(cfg.templates),
autoescape=jinja2.select_autoescape(["html.j2"]),
undefined=jinja2.StrictUndefined,
trim_blocks=True,
lstrip_blocks=True,
)
# availability
def candidates(self, et: EventType, start: datetime, end: datetime) -> list[datetime]:
"""Slot starts inside working hours, notice and horizon, ignoring busy times."""
cfg = self.cfg
now = now_utc()
earliest = max(start, now + cfg.min_notice)
latest = min(end, now + cfg.horizon)
if earliest >= latest:
return []
out = []
day = earliest.astimezone(cfg.tz).date() - timedelta(days=1)
last = latest.astimezone(cfg.tz).date()
while day <= last:
for a, b in cfg.hours.get(day.weekday(), ()):
m = a
while m + et.minutes <= b:
local = datetime(day.year, day.month, day.day) + timedelta(minutes=m)
slot = local.replace(tzinfo=cfg.tz).astimezone(UTC)
# Skip nonexistent local times (DST gap): they do not round-trip.
if slot.astimezone(cfg.tz).replace(tzinfo=None) == local and earliest <= slot < latest:
out.append(slot)
m += cfg.slot_step
day += timedelta(days=1)
return sorted(set(out))
def free_slots(self, et: EventType, start: datetime, end: datetime) -> list[datetime]:
slots = self.candidates(et, start, end)
if not slots:
return []
pad = timedelta(minutes=self.cfg.buffer)
length = timedelta(minutes=et.minutes)
window = (slots[0] - pad, slots[-1] + length + pad)
try:
busy = self.cal.busy(*window)
except CalDAVError as e:
log.error("cannot read calendar: %s", e)
raise Unavailable from e
busy += self.store.holds(*window)
return [s for s in slots if not overlaps(s - pad, s + length + pad, busy)]
# mail
def send(
self,
to: str,
subject: str,
body: str,
reply_to: str | None = None,
ics: bytes | None = None,
method: str | None = None,
) -> None:
msg = EmailMessage()
msg["From"] = formataddr((self.cfg.owner, self.cfg.mail_from))
msg["To"] = to
msg["Subject"] = subject
msg["Date"] = formatdate(localtime=False)
msg["Message-ID"] = make_msgid(domain=self.cfg.mail_from.rpartition("@")[2])
if reply_to:
msg["Reply-To"] = reply_to
msg.set_content(body)
if ics is not None:
msg.add_attachment(
ics,
maintype="text",
subtype="calendar",
filename="invite.ics",
params={"method": method or "PUBLISH"},
)
try:
with smtplib.SMTP(self.cfg.smtp_host, self.cfg.smtp_port, timeout=15) as s:
s.send_message(msg)
except (smtplib.SMTPException, OSError):
log.exception("sending mail to %s failed", to)
def render_text(self, template: str, **ctx) -> str:
return self.jinja.get_template(template).render(cfg=self.cfg, **ctx)
# calendar objects
def ics(self, b: sqlite3.Row, *, owner_copy: bool, method: str | None = None) -> bytes:
cfg = self.cfg
cal = icalendar.Calendar()
cal.add("prodid", "-//lassul.us//book//EN")
cal.add("version", "2.0")
if method:
cal.add("method", method)
ev = icalendar.Event()
ev.add("uid", b["uid"])
ev.add("sequence", b["sequence"])
ev.add("dtstamp", now_utc())
ev.add("dtstart", from_epoch(b["start"]))
ev.add("dtend", from_epoch(b["end"]))
if owner_copy:
pending = b["status"] == "pending"
ev.add("summary", f"{'? ' if pending else ''}{b['name']}: {b['title']}")
ev.add("status", "TENTATIVE" if pending else "CONFIRMED")
ev.add(
"description",
f"{b['name']} <{b['email']}>\n\n{b['note']}\n\n{cfg.base_url}/a/{b['admin_token']}",
)
else:
ev.add("summary", f"{b['title']} with {cfg.owner}")
ev.add("status", "CANCELLED" if method == "CANCEL" else "CONFIRMED")
ev.add("description", f"{b['note']}\n\n{cfg.base_url}/r/{b['guest_token']}")
ev.add("organizer", f"mailto:{cfg.notify}", parameters={"CN": cfg.owner})
ev.add(
"attendee",
f"mailto:{b['email']}",
parameters={"CN": b["name"], "ROLE": "REQ-PARTICIPANT", "PARTSTAT": "ACCEPTED"},
)
cal.add_component(ev)
return cal.to_ical()
def mirror(self, b: sqlite3.Row) -> None:
"""Keep the owner's calendar in line with the booking; best effort."""
try:
if b["status"] in ACTIVE:
self.cal.put(b["uid"], self.ics(b, owner_copy=True))
else:
self.cal.delete(b["uid"])
except CalDAVError as e:
log.error("calendar mirror for %s failed: %s", b["uid"], e)
# state transitions
def request(self, et: EventType, start: datetime, form: dict[str, str], guest_tz: str, ip: str) -> str:
cfg = self.cfg
with self.store.lock:
if start not in self.free_slots(et, start, start + timedelta(seconds=1)):
raise ValueError("That time is no longer available.")
token = secrets.token_urlsafe(24)
self.store.insert(
uid=f"{secrets.token_hex(16)}@{urlsplit(cfg.base_url).hostname}",
event=et.slug,
title=et.title,
start=epoch(start),
end=epoch(start + timedelta(minutes=et.minutes)),
name=form["name"],
email=form["email"],
note=form["note"],
guest_tz=guest_tz,
status="pending",
guest_token=token,
admin_token=secrets.token_urlsafe(24),
created=epoch(now_utc()),
updated=epoch(now_utc()),
ip=ip,
)
b = self.store.by_token("guest_token", token)
self.mirror(b)
self.send(
cfg.notify,
f"Booking request: {b['name']}, {self.when(b, cfg.tz)}",
self.render_text("mail/request.txt", b=b, when=self.when(b, cfg.tz)),
reply_to=formataddr((b["name"], b["email"])),
)
return token
def transition(self, b: sqlite3.Row, status: str, message: str = "") -> sqlite3.Row:
"""Move a booking to `status`, mirror it and mail whoever needs to know."""
cfg = self.cfg
with self.store.lock:
current = self.store.get(b["id"])
allowed = {
"approved": ("pending",),
"declined": ("pending",),
"cancelled": ACTIVE,
"expired": ("pending",),
}[status]
if current["status"] not in allowed:
return current
was = current["status"]
self.store.set_status(b["id"], status)
b = self.store.get(b["id"])
self.mirror(b)
guest_when = self.when(b, ZoneInfo(b["guest_tz"]))
ctx = {"b": b, "when": guest_when, "message": message}
if status == "approved":
self.send(
b["email"],
f"Confirmed: {b['title']} with {cfg.owner}, {guest_when}",
self.render_text("mail/approved.txt", **ctx),
reply_to=cfg.notify,
ics=self.ics(b, owner_copy=False, method="REQUEST"),
method="REQUEST",
)
elif status == "declined":
self.send(
b["email"],
f"Declined: {b['title']} with {cfg.owner}, {guest_when}",
self.render_text("mail/declined.txt", **ctx),
reply_to=cfg.notify,
)
elif status == "cancelled" and was == "approved":
# The guest holds an invitation; withdraw it either way.
self.send(
b["email"],
f"Cancelled: {b['title']} with {cfg.owner}, {guest_when}",
self.render_text("mail/cancelled.txt", **ctx),
reply_to=cfg.notify,
ics=self.ics(b, owner_copy=False, method="CANCEL"),
method="CANCEL",
)
return b
def guest_cancel(self, b: sqlite3.Row) -> sqlite3.Row:
b = self.transition(b, "cancelled")
if b["status"] == "cancelled":
self.send(
self.cfg.notify,
f"Cancelled by guest: {b['name']}, {self.when(b, self.cfg.tz)}",
self.render_text("mail/guest-cancelled.txt", b=b, when=self.when(b, self.cfg.tz)),
reply_to=formataddr((b["name"], b["email"])),
)
return b
def sweep(self) -> None:
for b in self.store.overdue(now_utc()):
log.info("expiring unanswered request %s", b["uid"])
self.transition(b, "expired")
def rate_limited(self, ip: str, limit: int = 5, window: float = 3600) -> bool:
with self.store.lock:
q = self.rate[ip]
t = time.monotonic()
while q and q[0] < t - window:
q.popleft()
if len(q) >= limit:
return True
q.append(t)
return False
# formatting
@staticmethod
def when(b: sqlite3.Row, tz: ZoneInfo) -> str:
s = from_epoch(b["start"]).astimezone(tz)
e = from_epoch(b["end"]).astimezone(tz)
return f"{s:%a %d %b %Y, %H:%M}–{e:%H:%M} ({tz.key})"
# --- HTTP --------------------------------------------------------------------
STATIC = {
"/style.css": ("style.css", "text/css; charset=utf-8"),
"/tz.js": ("tz.js", "text/javascript; charset=utf-8"),
"/font.woff": ("font.woff", "font/woff"),
"/robots.txt": ("robots.txt", "text/plain; charset=utf-8"),
}
CSP = (
"default-src 'none'; style-src 'self'; script-src 'self'; font-src 'self'; "
"form-action 'self'; base-uri 'none'; frame-ancestors 'none'"
)
class HTTPError(Exception):
def __init__(self, status: HTTPStatus, message: str):
super().__init__(message)
self.status = status
self.message = message
class Handler(BaseHTTPRequestHandler):
app: App
server_version = "book"
sys_version = ""
# plumbing
def log_message(self, fmt: str, *args) -> None:
log.info("%s %s", self.client_ip(), fmt % args)
def client_ip(self) -> str:
peer = self.client_address[0]
if peer in ("127.0.0.1", "::1"):
return self.headers.get("X-Real-IP", peer)
return peer
def respond(self, status: HTTPStatus, body: bytes, ctype: str, headers: dict[str, str] | None = None) -> None:
self.send_response(status)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.send_header("Content-Security-Policy", CSP)
self.send_header("Referrer-Policy", "no-referrer")
self.send_header("X-Content-Type-Options", "nosniff")
for k, v in (headers or {}).items():
self.send_header(k, v)
self.end_headers()
if self.command != "HEAD":
self.wfile.write(body)
def page(self, template: str, status: HTTPStatus = HTTPStatus.OK, **ctx) -> None:
html = self.app.jinja.get_template(template).render(cfg=self.app.cfg, **ctx)
self.respond(status, html.encode(), "text/html; charset=utf-8", {"Cache-Control": "no-store"})
def redirect(self, location: str) -> None:
self.respond(HTTPStatus.SEE_OTHER, b"", "text/plain", {"Location": location})
def query(self) -> dict[str, str]:
return {k: v[0] for k, v in parse_qs(urlsplit(self.path).query).items()}
def form(self) -> dict[str, str]:
length = int(self.headers.get("Content-Length") or 0)
if length > MAX_BODY:
raise HTTPError(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "Request too large.")
raw = self.rfile.read(length).decode("utf-8", "replace")
return {k: v[0] for k, v in parse_qs(raw, keep_blank_values=True).items()}
def guest_tz(self, q: dict[str, str]) -> ZoneInfo:
tz = q.get("tz", "")
return ZoneInfo(tz) if tz in TIMEZONES else self.app.cfg.tz
def dispatch(self, method: str) -> None:
path = urlsplit(self.path).path
parts = [p for p in path.split("/") if p]
try:
if method == "GET" and path in STATIC:
return self.static(*STATIC[path])
if not parts:
return self.index() if method == "GET" else self.not_found()
head, rest = parts[0], parts[1:]
if head in ("a", "r") and rest and TOKEN_RE.match(rest[0]):
column = "admin_token" if head == "a" else "guest_token"
b = self.app.store.by_token(column, rest[0])
if b is None:
return self.not_found()
action = rest[1] if len(rest) > 1 else ""
if head == "a":
return self.admin(method, b, action)
return self.guest(method, b, action)
et = self.app.cfg.event_type(head)
if et is None or len(rest) > 1:
return self.not_found()
if not rest:
return self.slots(et) if method == "GET" else self.not_found()
try:
start = datetime.strptime(rest[0], SLOT_FORMAT).replace(tzinfo=UTC)
except ValueError:
return self.not_found()
return self.book(method, et, start)
except HTTPError as e:
self.page("message.html.j2", e.status, title="Error", message=e.message)
except Unavailable:
self.page(
"message.html.j2",
HTTPStatus.SERVICE_UNAVAILABLE,
title="Unavailable",
message="The calendar cannot be read right now. Please try again later.",
)
def do_GET(self) -> None:
self.dispatch("GET")
def do_HEAD(self) -> None:
self.dispatch("GET")
def do_POST(self) -> None:
self.dispatch("POST")
def not_found(self) -> None:
self.page("message.html.j2", HTTPStatus.NOT_FOUND, title="Not found", message="Nothing here.")
def static(self, name: str, ctype: str) -> None:
try:
body = (Path(self.app.cfg.static) / name).read_bytes()
except FileNotFoundError:
return self.not_found()
self.respond(HTTPStatus.OK, body, ctype, {"Cache-Control": "public, max-age=86400"})
# public pages
def index(self) -> None:
cfg = self.app.cfg
q = self.query()
if len(cfg.event_types) == 1:
return self.redirect(f"/{cfg.event_types[0].slug}?{urlencode(q)}")
# Only pass on an explicit zone: pinning the owner's here would stop
# tz.js from detecting the visitor's zone on the next page.
tzq = urlencode({"tz": q["tz"]}) if q.get("tz") in TIMEZONES else ""
self.page("index.html.j2", tzq=tzq)
def slots(self, et: EventType) -> None:
q = self.query()
tz = self.guest_tz(q)
today = now_utc().astimezone(tz).date()
try:
first = date.fromisoformat(q.get("from", ""))
except ValueError:
first = today
first = max(first, today)
days = [first + timedelta(days=i) for i in range(7)]
start = datetime(first.year, first.month, first.day, tzinfo=tz).astimezone(UTC)
end = datetime(days[-1].year, days[-1].month, days[-1].day, tzinfo=tz).astimezone(UTC) + timedelta(days=1)
by_day: dict[date, list[datetime]] = {d: [] for d in days}
for s in self.app.free_slots(et, start, end):
local = s.astimezone(tz)
if local.date() in by_day:
by_day[local.date()].append(local)
horizon = (now_utc() + self.app.cfg.horizon).astimezone(tz).date()
prev_first = max(first - timedelta(days=7), today) if first > today else None
next_first = first + timedelta(days=7) if first + timedelta(days=7) <= horizon else None
self.page(
"slots.html.j2",
et=et,
tz=tz,
days=by_day,
fmt=SLOT_FORMAT,
utc=UTC,
timezones=TIMEZONES,
prev=prev_first and urlencode({"tz": tz.key, "from": prev_first.isoformat()}),
next=next_first and urlencode({"tz": tz.key, "from": next_first.isoformat()}),
tzq=urlencode({"tz": tz.key}),
tzaware=True,
)
def book(self, method: str, et: EventType, start: datetime) -> None:
q = self.query()
tz = self.guest_tz(q)
values = {"name": "", "email": "", "note": ""}
error = ""
if method == "POST":
form = self.form()
tz = self.guest_tz(form)
values = {k: form.get(k, "").strip() for k in values}
if form.get("website"): # honeypot: pretend it worked, do nothing
return self.page("message.html.j2", title="Request sent", message="Request sent.")
error = self.validate(values)
if not error:
ip = self.client_ip()
if self.app.rate_limited(ip):
error = "Too many requests from your address. Please try again later."
elif self.app.store.count("status = 'pending'", ()) >= 30:
error = "Too many open requests right now. Please try again later."
elif self.app.store.count("status = 'pending' AND email = ?", (values["email"],)) >= 3:
error = "You already have several open requests."
else:
try:
token = self.app.request(et, start, values, tz.key, ip)
except ValueError as e:
error = str(e)
else:
return self.redirect(f"/r/{token}")
elif start not in self.app.free_slots(et, start, start + timedelta(seconds=1)):
return self.page(
"message.html.j2",
HTTPStatus.GONE,
title="Taken",
message="That time is not available.",
back=f"/{et.slug}?{urlencode({'tz': tz.key})}",
)
self.page(
"book.html.j2",
HTTPStatus.BAD_REQUEST if error else HTTPStatus.OK,
et=et,
tz=tz,
start=start.astimezone(tz),
end=(start + timedelta(minutes=et.minutes)).astimezone(tz),
values=values,
error=error,
back=f"/{et.slug}?{urlencode({'tz': tz.key})}",
tzaware=True,
)
@staticmethod
def validate(values: dict[str, str]) -> str:
for k, limit in LIMITS.items():
if len(values.get(k, "")) > limit:
return f"The {k} is too long."
if not values["name"]:
return "Please tell me your name."
if any(c in values["name"] for c in "\r\n<>\"") or "@" in values["name"]:
return "Please use a plain name."
if not EMAIL_RE.match(values["email"]) or any(c in values["email"] for c in "\r\n<>,\"'"):
return "That email address does not look right."
return ""
# token pages: GET only renders, POST acts, so link scanners cannot click buttons
def guest(self, method: str, b: sqlite3.Row, action: str) -> None:
if action == "invite.ics" and method == "GET" and b["status"] == "approved":
return self.respond(
HTTPStatus.OK,
self.app.ics(b, owner_copy=False, method="PUBLISH"),
"text/calendar; charset=utf-8",
{"Content-Disposition": 'attachment; filename="invite.ics"', "Cache-Control": "no-store"},
)
if action == "cancel" and method == "POST":
self.app.guest_cancel(b)
return self.redirect(f"/r/{b['guest_token']}")
if action or method != "GET":
return self.not_found()
tz = ZoneInfo(b["guest_tz"])
self.page(
"request.html.j2",
b=b,
when=self.app.when(b, tz),
future=from_epoch(b["start"]) > now_utc(),
noindex=True,
)
def admin(self, method: str, b: sqlite3.Row, action: str) -> None:
if method == "POST" and action in ("approve", "decline", "cancel"):
message = self.form().get("message", "").strip()[: LIMITS["message"]]
status = {"approve": "approved", "decline": "declined", "cancel": "cancelled"}[action]
self.app.transition(b, status, message)
return self.redirect(f"/a/{b['admin_token']}")
if action or method != "GET":
return self.not_found()
self.page(
"admin.html.j2",
b=b,
when=self.app.when(b, self.app.cfg.tz),
guest_when=self.app.when(b, ZoneInfo(b["guest_tz"])),
future=from_epoch(b["start"]) > now_utc(),
noindex=True,
)
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("--config", required=True, help="JSON configuration file")
parser.add_argument("--caldav-password-file", required=True, help="file holding the CalDAV password")
parser.add_argument("--templates", default=os.environ.get("BOOK_TEMPLATES"), required="BOOK_TEMPLATES" not in os.environ)
parser.add_argument("--static", default=os.environ.get("BOOK_STATIC"), required="BOOK_STATIC" not in os.environ)
args = parser.parse_args()
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
cfg = load_config(args.config, args.caldav_password_file, args.templates, args.static)
app = App(cfg)
try:
app.cal.ensure_calendar()
except CalDAVError as e:
log.error("calendar not reachable yet: %s", e)
def sweeper() -> None:
while True:
try:
app.sweep()
except Exception:
log.exception("sweep failed")
time.sleep(300)
threading.Thread(target=sweeper, daemon=True).start()
Handler.app = app
server = ThreadingHTTPServer((cfg.listen_host, cfg.listen_port), Handler)
server.daemon_threads = True
log.info("listening on %s:%d", cfg.listen_host, cfg.listen_port)
server.serve_forever()
if __name__ == "__main__":
main()