#!/usr/bin/env python3 """Dependency-free mailbox HTTP example. Persists setup BEFORE the first request. Standalone example, not an SDK. Supply your mailbox API origin. """ import argparse import json import os from pathlib import Path import secrets import re import stat import sys from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None # Never forward credentials to another destination. def load_or_create_state(path, base_url): path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) try: fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) except FileExistsError: fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) with os.fdopen(fd) as handle: metadata = os.fstat(handle.fileno()) if not stat.S_ISREG(metadata.st_mode) or metadata.st_mode & 0o077: raise ValueError("State must be a private regular file (chmod 600).") state = json.load(handle) if state["base_url"] != base_url: raise ValueError("State belongs to another API origin. Use a separate state file.") return state state = {"base_url": base_url, "setup_token": "ridian_setup_" + secrets.token_hex(32)} with os.fdopen(fd, "w") as handle: json.dump(state, handle) handle.flush() os.fsync(handle.fileno()) return state def save_state(path, state): temporary = path.with_name(path.name + "." + secrets.token_hex(8) + ".tmp") fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(fd, "w") as handle: json.dump(state, handle) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) finally: temporary.unlink(missing_ok=True) def prepare_curl(path, state): """Prepare private curl files locally, reusing the durable setup identity.""" token = state["setup_token"] if not re.fullmatch(r"ridian_setup_[a-f0-9]{64}", token): raise ValueError("Invalid saved setup credential.") header = Path(str(path) + ".headers") response = Path(str(path) + ".response.json") # Do not overwrite an existing response: it may be the only saved proof. for target, content in [(header, "Authorization: Bearer " + token + "\n"), (response, "")]: try: fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) except FileExistsError: fd = os.open(target, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) with os.fdopen(fd) as handle: metadata = os.fstat(handle.fileno()) if not stat.S_ISREG(metadata.st_mode) or metadata.st_mode & 0o077: raise ValueError("curl files must be private regular files.") if target == header and handle.read() != content: raise ValueError("Header belongs to a different setup credential.") else: with os.fdopen(fd, "w") as handle: handle.write(content) handle.flush() os.fsync(handle.fileno()) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--base-url", required=True, help="Configured API origin, without /v1") parser.add_argument("--state", type=Path, required=True, help="Private credential file; keep it outside version control") parser.add_argument("operation", choices=["setup", "prepare-curl", "status", "list", "message", "events", "ack-events"]) parser.add_argument("--message-id") parser.add_argument("--cursor") args = parser.parse_args() base = args.base_url.rstrip("/") parsed = urlsplit(base) if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password or parsed.path or parsed.query or parsed.fragment: parser.error("Use a fixed HTTPS origin without path, credentials, query or fragment.") if args.operation not in ("setup", "prepare-curl") and not args.state.exists(): parser.error("Run setup first with this state file.") state = load_or_create_state(args.state, base) if args.operation == "prepare-curl": prepare_curl(args.state, state) print("Private curl header and response files prepared. No network request made.") return 0 if args.operation == "ack-events": if not args.cursor or len(args.cursor) > 4096: parser.error("Supply next_cursor from a fully processed events page.") state["event_cursor"] = args.cursor save_state(args.state, state) print("Event cursor saved. Replayed events must still be deduplicated by ID.") return 0 root = "/v1/mailboxes" if args.operation == "setup": method, path, token, body = "POST", root, state["setup_token"], b"{}" else: if "access_token" not in state: parser.error("Repeat setup with the same state file to recover its response.") method, token, body = "GET", state["access_token"], None path = root + "/current" if args.operation == "list": path += "/messages" + ("?" + urlencode({"cursor": args.cursor}) if args.cursor else "") if args.operation == "events": cursor = args.cursor or state.get("event_cursor") path += "/events" + ("?" + urlencode({"cursor": cursor}) if cursor else "") if args.operation == "message": if not args.message_id or not args.message_id.isalnum(): parser.error("Supply a message ID from list using --message-id.") path += "/messages/" + args.message_id request = Request(base + path, method=method, data=body, headers={ "Authorization": "Bearer " + token, "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "ridian-mailbox-cli"}) try: with build_opener(NoRedirect).open(request, timeout=30) as response: result = json.load(response) except HTTPError as error: # Do not print request headers or arbitrary upstream HTML/error bodies. print(f"HTTP {error.code}. Retry-After: {error.headers.get('Retry-After', 'not supplied')}. Keep the same state file.", file=sys.stderr) return 1 except URLError: print("Transport interrupted. Repeat with the same state file.", file=sys.stderr) return 1 if args.operation == "setup": state.update(result) save_state(args.state, state) print(json.dumps(result["mailbox"], indent=2)) # Never print secrets. else: print(json.dumps(result, indent=2, ensure_ascii=False)) return 0 if __name__ == "__main__": try: sys.exit(main()) except (ValueError, OSError, KeyError): print("Could not read or persist private state. Check the state path and permissions; do not replace an existing setup key.", file=sys.stderr) sys.exit(1)