From 258d23175ce1dc30e80d9ab666f1e7644d68dc47 Mon Sep 17 00:00:00 2001 From: Arron Atchison Date: Wed, 1 Jul 2026 11:48:50 -0700 Subject: [PATCH] Add CalDAV intermittent-connection probe + log correlation Diagnostic tooling for the intermittent 'failed to connect to server mail.thundermail.com' calendar errors seen in Thunderbird desktop. A once-a-day integration test cannot catch a failure that occurs ~once every 2-3 hours, so this polls the CalDAV endpoint frequently and records which connection phase fails, then correlates failures against Stalwart CloudWatch logs to determine whether the root cause is Stalwart, the network/edge, or the client. - probe.py: stdlib-only poller timing DNS/TCP/TLS/PROPFIND phases with granular failure classification; appends JSONL results. - correlate.py: matches probe failures against /tb/prod/stalwart logs by public IP and timestamp window. - .env/.jsonl gitignored; README documents usage and result interpretation. Co-Authored-By: Claude Opus 4.8 (1M context) --- research/caldav-probe/.env.example | 9 + research/caldav-probe/.gitignore | 3 + research/caldav-probe/README.md | 47 +++++ research/caldav-probe/correlate.py | 129 +++++++++++++ research/caldav-probe/probe.py | 287 +++++++++++++++++++++++++++++ 5 files changed, 475 insertions(+) create mode 100644 research/caldav-probe/.env.example create mode 100644 research/caldav-probe/.gitignore create mode 100644 research/caldav-probe/README.md create mode 100644 research/caldav-probe/correlate.py create mode 100644 research/caldav-probe/probe.py diff --git a/research/caldav-probe/.env.example b/research/caldav-probe/.env.example new file mode 100644 index 0000000..b8a03bb --- /dev/null +++ b/research/caldav-probe/.env.example @@ -0,0 +1,9 @@ +# Copy to .env and fill in. .env is gitignored. +TEST_SERVER_HOST = "mail.thundermail.com" +TEST_ACCT_1_USERNAME = "username@thundermail.com" +TEST_ACCT_1_PASSWORD = "app password" + +# Optional overrides: +# CALDAV_PATH = "/dav/cal/" +# PROBE_INTERVAL_SECONDS = 60 +# PROBE_TIMEOUT_SECONDS = 30 diff --git a/research/caldav-probe/.gitignore b/research/caldav-probe/.gitignore new file mode 100644 index 0000000..0ed7f0c --- /dev/null +++ b/research/caldav-probe/.gitignore @@ -0,0 +1,3 @@ +# Local credentials and captured probe data -- never commit these. +.env +*.jsonl diff --git a/research/caldav-probe/README.md b/research/caldav-probe/README.md new file mode 100644 index 0000000..d603dbc --- /dev/null +++ b/research/caldav-probe/README.md @@ -0,0 +1,47 @@ +# CalDAV intermittent-connection probe + +Diagnostic tooling for the intermittent *"failed to connect to server +mail.thundermail.com"* calendar errors that Thunderbird desktop users +occasionally see. A once-a-day integration test can't catch a failure that +happens roughly once every 2–3 hours, so this polls the CalDAV endpoint +frequently and records exactly which connection phase fails, then correlates any +failures against Stalwart's CloudWatch logs. + +The point is to answer one question: **is the root cause Stalwart, the +network/edge, or the Thunderbird client?** The root cause may well be unrelated +to Stalwart — this is how we find out for sure. + +## Contents + +- `probe.py` — polls `https:///dav/cal/` on an interval, timing each phase + (DNS → TCP → TLS → CalDAV `PROPFIND`) and classifying failures + (`DNS_FAIL`, `TCP_FAIL`, `TLS_FAIL`, `AUTH_FAIL`, `HTTP_5XX`, `TIMEOUT`, + `DAV_ERROR`, `OK`). Standard library only — no install step. Appends one JSON + line per attempt to `probe-YYYYMMDD.jsonl`. +- `correlate.py` — reads the JSONL, finds failures, and queries the Stalwart + CloudWatch log group in a window around each failure, filtered to this + machine's public IP. + +## Usage + +```bash +cd research/caldav-probe +cp .env.example .env # then fill in TEST_ACCT_1 credentials + +# Run the probe (leave it going for several hours / overnight): +python3 probe.py + +# Later, correlate failures against Stalwart logs (needs AWS SSO to legacy): +AWS_PROFILE=mzla-legacy python3 correlate.py probe-*.jsonl +``` + +## Reading the results + +| Probe result | Stalwart logs | Verdict | +|---|---|---| +| Fails | error logged at our IP/time | **Server-side (Stalwart)** | +| `TLS_FAIL` (cert) | — | Cert / edge | +| `TCP_FAIL` / `TIMEOUT` | silent | **Network / edge / load balancer** | +| Always `OK` | — | **Client-side (Thunderbird)** or a layer not probed | + +`.env` and the `*.jsonl` capture files are gitignored. diff --git a/research/caldav-probe/correlate.py b/research/caldav-probe/correlate.py new file mode 100644 index 0000000..eff3ce2 --- /dev/null +++ b/research/caldav-probe/correlate.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 +"""Correlate CalDAV probe failures against Stalwart's CloudWatch logs. + +Reads the JSONL emitted by probe.py, finds every non-OK attempt, and for each one +queries the Stalwart CloudWatch log group in a window around the failure timestamp, +filtered to this machine's public IP. The report tells you, for each probe failure, +exactly what the server logged at that moment -- or that it logged nothing at all. + + probe fails + server logged an error -> server-side (Stalwart) + probe fails (TCP/TLS/timeout) + silent -> network / edge / load balancer + probe never fails but users still do -> client-side (Thunderbird desktop) + +Requires the AWS CLI, authenticated to the legacy profile: + + AWS_PROFILE=mzla-legacy python3 correlate.py probe-YYYYMMDD.jsonl + +Options: + --log-group CloudWatch log group (default: /tb/prod/stalwart) + --window seconds each side of the failure to search (default: 90) + --profile AWS profile to pass to the CLI (default: $AWS_PROFILE or mzla-legacy) +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path + + +def parse_ts(ts: str) -> int: + """ISO 'YYYY-MM-DDTHH:MM:SSZ' -> epoch milliseconds.""" + dt = datetime.strptime(ts, '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=timezone.utc) + return int(dt.timestamp() * 1000) + + +def query_logs(log_group: str, start_ms: int, end_ms: int, ip: str | None, + profile: str) -> list[dict]: + cmd = [ + 'aws', 'logs', 'filter-log-events', + '--log-group-name', log_group, + '--start-time', str(start_ms), + '--end-time', str(end_ms), + '--limit', '50', + '--output', 'json', + '--profile', profile, + ] + if ip: + # Filter to events mentioning our probe's public IP so we only see the + # server's view of *our* connections. + cmd += ['--filter-pattern', f'"{ip}"'] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode != 0: + print(f' ! aws query failed: {proc.stderr.strip()}', file=sys.stderr) + return [] + return json.loads(proc.stdout or '{}').get('events', []) + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument('jsonl', nargs='+', help='probe-*.jsonl file(s) to analyze') + ap.add_argument('--log-group', default='/tb/prod/stalwart') + ap.add_argument('--window', type=int, default=90, help='seconds each side of failure') + ap.add_argument('--profile', default=None) + args = ap.parse_args() + + import os + profile = args.profile or os.environ.get('AWS_PROFILE') or 'mzla-legacy' + + failures = [] + total = 0 + for path_str in args.jsonl: + path = Path(path_str) + if not path.exists(): + print(f'skipping missing file: {path}', file=sys.stderr) + continue + for line in path.read_text().splitlines(): + line = line.strip() + if not line: + continue + rec = json.loads(line) + total += 1 + if rec.get('outcome') != 'OK': + failures.append(rec) + + print(f'Analyzed {total} probe attempts, {len(failures)} failures.\n') + if not failures: + print('No probe failures recorded -- if users still report errors, the ' + 'issue is likely client-side (Thunderbird desktop) or in a layer this ' + 'probe does not exercise.') + return 0 + + window_ms = args.window * 1000 + server_confirmed = 0 + server_silent = 0 + + for rec in failures: + ts = rec['ts'] + ip = rec.get('public_ip') + print('=' * 78) + print(f'FAILURE {ts} outcome={rec["outcome"]} status={rec.get("http_status")}') + print(f' error : {rec.get("error")}') + print(f' phases: {rec.get("phases")} total={rec.get("total_ms")}ms probe_ip={ip}') + center = parse_ts(ts) + events = query_logs(args.log_group, center - window_ms, center + window_ms, ip, profile) + if not events: + server_silent += 1 + print(f' SERVER SILENT: no {args.log_group} events for {ip} in ' + f'+/-{args.window}s window --> points AWAY from Stalwart ' + f'(network / edge / client).') + else: + server_confirmed += 1 + print(f' SERVER LOGGED {len(events)} event(s) in window:') + for ev in events: + ev_ts = datetime.fromtimestamp(ev['timestamp'] / 1000, timezone.utc) + print(f' {ev_ts:%H:%M:%SZ} {ev["message"].strip()[:220]}') + print() + + print('=' * 78) + print(f'SUMMARY: {len(failures)} failures | server-logged: {server_confirmed} | ' + f'server-silent: {server_silent}') + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/research/caldav-probe/probe.py b/research/caldav-probe/probe.py new file mode 100644 index 0000000..8c5c22d --- /dev/null +++ b/research/caldav-probe/probe.py @@ -0,0 +1,287 @@ +#!/usr/bin/env python3 +"""Intermittent CalDAV connection probe. + +Polls the Thundermail CalDAV endpoint on a fixed interval and records, for every +attempt, a per-phase breakdown (DNS -> TCP -> TLS -> CalDAV request) with precise +failure classification. The goal is to catch the *intermittent* "failed to connect +to server" errors that Thunderbird desktop users see for the calendar, and to +pin down which layer fails so we can tell whether the root cause is Stalwart, the +network/edge, or the client. + +Deliberately uses only the Python standard library so it runs anywhere with no +install step, and so each connection phase can be timed and classified on its own +(the `caldav` library collapses every failure into a single opaque exception). + +Each attempt is appended as one JSON line to probe-YYYYMMDD.jsonl. Feed that file +to correlate.py to line failures up against Stalwart's CloudWatch logs. + +Config comes from the environment (optionally from a .env file next to this +script): + + TEST_SERVER_HOST host to probe (default: mail.thundermail.com) + TEST_ACCT_1_USERNAME CalDAV username (required) + TEST_ACCT_1_PASSWORD CalDAV app password (required) + CALDAV_PATH PROPFIND path (default: /dav/cal/) + PROBE_INTERVAL_SECONDS seconds between polls (default: 60) + PROBE_TIMEOUT_SECONDS per-phase timeout (default: 30) +""" + +from __future__ import annotations + +import base64 +import json +import os +import signal +import socket +import ssl +import sys +import time +import urllib.request +from datetime import datetime, timezone +from pathlib import Path + +HERE = Path(__file__).resolve().parent + +# Outcome classifications, ordered from earliest to latest failing phase. +OK = 'OK' +DNS_FAIL = 'DNS_FAIL' +TCP_FAIL = 'TCP_FAIL' +TLS_FAIL = 'TLS_FAIL' +AUTH_FAIL = 'AUTH_FAIL' +HTTP_5XX = 'HTTP_5XX' +DAV_ERROR = 'DAV_ERROR' +TIMEOUT = 'TIMEOUT' + +PROPFIND_BODY = ( + '' + '' +).encode('utf-8') + + +def load_dotenv(path: Path) -> None: + """Minimal .env loader (no dependency on python-dotenv).""" + if not path.exists(): + return + for raw in path.read_text().splitlines(): + line = raw.strip() + if not line or line.startswith('#') or '=' not in line: + continue + key, _, value = line.partition('=') + key = key.strip() + value = value.strip().strip('"').strip("'") + # Don't clobber a value already exported in the real environment. + os.environ.setdefault(key, value) + + +def now_iso() -> str: + return datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') + + +def get_public_ip() -> str | None: + """Best-effort discovery of this machine's public IP for log correlation.""" + for url in ('https://checkip.amazonaws.com', 'https://api.ipify.org'): + try: + with urllib.request.urlopen(url, timeout=10) as resp: + return resp.read().decode('utf-8').strip() + except Exception: + continue + return None + + +def probe_once(host: str, path: str, auth_header: str, timeout: float) -> dict: + """Run one full DNS -> TCP -> TLS -> PROPFIND attempt, timing each phase.""" + result = { + 'ts': now_iso(), + 'outcome': None, + 'http_status': None, + 'phases': {}, + 'total_ms': None, + 'error': None, + } + started = time.monotonic() + + def elapsed_ms(since: float) -> float: + return round((time.monotonic() - since) * 1000, 1) + + # Phase 1: DNS + t = time.monotonic() + try: + addrinfo = socket.getaddrinfo(host, 443, socket.AF_UNSPEC, socket.SOCK_STREAM) + except socket.gaierror as e: + result.update(outcome=DNS_FAIL, error=f'{type(e).__name__}: {e}') + result['total_ms'] = elapsed_ms(started) + return result + result['phases']['dns_ms'] = elapsed_ms(t) + family, socktype, proto, _canon, sockaddr = addrinfo[0] + result['resolved_ip'] = sockaddr[0] + + # Phase 2: TCP connect + t = time.monotonic() + raw_sock = socket.socket(family, socktype, proto) + raw_sock.settimeout(timeout) + try: + raw_sock.connect(sockaddr) + except socket.timeout as e: + raw_sock.close() + result.update(outcome=TIMEOUT, error=f'tcp connect timeout: {e}') + result['total_ms'] = elapsed_ms(started) + return result + except OSError as e: + raw_sock.close() + result.update(outcome=TCP_FAIL, error=f'{type(e).__name__}: {e}') + result['total_ms'] = elapsed_ms(started) + return result + result['phases']['tcp_ms'] = elapsed_ms(t) + + # Phase 3: TLS handshake (cert verification on) + t = time.monotonic() + ctx = ssl.create_default_context() + try: + tls_sock = ctx.wrap_socket(raw_sock, server_hostname=host) + except ssl.SSLCertVerificationError as e: + raw_sock.close() + result.update(outcome=TLS_FAIL, error=f'cert verification: {e}') + result['total_ms'] = elapsed_ms(started) + return result + except ssl.SSLError as e: + raw_sock.close() + result.update(outcome=TLS_FAIL, error=f'{type(e).__name__}: {e}') + result['total_ms'] = elapsed_ms(started) + return result + except (socket.timeout, OSError) as e: + raw_sock.close() + result.update(outcome=TLS_FAIL, error=f'{type(e).__name__}: {e}') + result['total_ms'] = elapsed_ms(started) + return result + result['phases']['tls_ms'] = elapsed_ms(t) + + # Phase 4: CalDAV PROPFIND + t = time.monotonic() + request = ( + f'PROPFIND {path} HTTP/1.1\r\n' + f'Host: {host}\r\n' + f'Authorization: {auth_header}\r\n' + f'Depth: 0\r\n' + f'Content-Type: application/xml; charset=utf-8\r\n' + f'Content-Length: {len(PROPFIND_BODY)}\r\n' + f'User-Agent: mailstrom-caldav-probe/1.0\r\n' + f'Connection: close\r\n' + f'\r\n' + ).encode('utf-8') + PROPFIND_BODY + try: + tls_sock.settimeout(timeout) + tls_sock.sendall(request) + # Read just enough to get the status line. + buf = b'' + while b'\r\n' not in buf and len(buf) < 4096: + chunk = tls_sock.recv(4096) + if not chunk: + break + buf += chunk + except socket.timeout as e: + tls_sock.close() + result.update(outcome=TIMEOUT, error=f'propfind read timeout: {e}') + result['total_ms'] = elapsed_ms(started) + return result + except OSError as e: + tls_sock.close() + result.update(outcome=DAV_ERROR, error=f'{type(e).__name__}: {e}') + result['total_ms'] = elapsed_ms(started) + return result + finally: + try: + tls_sock.close() + except OSError: + pass + result['phases']['req_ms'] = elapsed_ms(t) + + status_line = buf.split(b'\r\n', 1)[0].decode('latin-1', 'replace') + parts = status_line.split(' ', 2) + status = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else None + result['http_status'] = status + result['total_ms'] = elapsed_ms(started) + + if status in (200, 207): + result['outcome'] = OK + elif status in (401, 403): + result.update(outcome=AUTH_FAIL, error=status_line) + elif status is not None and 500 <= status < 600: + result.update(outcome=HTTP_5XX, error=status_line) + else: + result.update(outcome=DAV_ERROR, error=status_line) + return result + + +_stop = False + + +def _handle_signal(signum, _frame): + global _stop + _stop = True + print(f'\n[{now_iso()}] received signal {signum}, shutting down after current attempt...', + file=sys.stderr, flush=True) + + +def main() -> int: + load_dotenv(HERE / '.env') + + host = os.environ.get('TEST_SERVER_HOST', 'mail.thundermail.com').strip() + username = os.environ.get('TEST_ACCT_1_USERNAME', '').strip() + password = os.environ.get('TEST_ACCT_1_PASSWORD', '').strip() + path = os.environ.get('CALDAV_PATH', '/dav/cal/').strip() + interval = float(os.environ.get('PROBE_INTERVAL_SECONDS', '60')) + timeout = float(os.environ.get('PROBE_TIMEOUT_SECONDS', '30')) + + if not username or not password: + print('ERROR: TEST_ACCT_1_USERNAME and TEST_ACCT_1_PASSWORD must be set ' + '(export them or put them in research/caldav-probe/.env).', file=sys.stderr) + return 2 + + auth_header = 'Basic ' + base64.b64encode(f'{username}:{password}'.encode()).decode() + + signal.signal(signal.SIGINT, _handle_signal) + signal.signal(signal.SIGTERM, _handle_signal) + + public_ip = get_public_ip() + out_path = HERE / f'probe-{datetime.now(timezone.utc):%Y%m%d}.jsonl' + + print(f'[{now_iso()}] probing https://{host}{path} as {username} ' + f'every {interval:g}s (timeout {timeout:g}s)') + print(f'[{now_iso()}] this machine public IP: {public_ip or "unknown"} ' + f'(use this to filter Stalwart logs)') + print(f'[{now_iso()}] writing results to {out_path}') + + counts: dict[str, int] = {} + attempts = 0 + while not _stop: + cycle_start = time.monotonic() + record = probe_once(host, path, auth_header, timeout) + record['public_ip'] = public_ip + record['host'] = host + + with out_path.open('a') as f: + f.write(json.dumps(record) + '\n') + + attempts += 1 + counts[record['outcome']] = counts.get(record['outcome'], 0) + 1 + summary = ' '.join(f'{k}={v}' for k, v in sorted(counts.items())) + flag = '' if record['outcome'] == OK else ' <-- FAILURE' + print(f'[{record["ts"]}] {record["outcome"]:<9} ' + f'status={record["http_status"]} total={record["total_ms"]}ms ' + f'phases={record["phases"]}{flag}', flush=True) + if record['outcome'] != OK and record['error']: + print(f' error: {record["error"]}', flush=True) + if attempts % 10 == 0: + print(f'[{now_iso()}] --- {attempts} attempts: {summary} ---', flush=True) + + # Sleep the remainder of the interval, interruptibly. + while not _stop and (time.monotonic() - cycle_start) < interval: + time.sleep(min(1.0, interval)) + + print(f'[{now_iso()}] stopped after {attempts} attempts: ' + f'{" ".join(f"{k}={v}" for k, v in sorted(counts.items()))}') + return 0 + + +if __name__ == '__main__': + raise SystemExit(main())