#!/usr/bin/env python3 """Download one explicitly selected BOX event page; no pagination loop or retries.""" import argparse import json import math import os import ssl import sys from http.client import HTTPException from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlsplit from urllib.request import HTTPSHandler, HTTPRedirectHandler, Request, build_opener ENDPOINT = "/api/statistics/event-storage/events/" MAX_BYTES = 10 * 1024 * 1024 class FetchError(Exception): """A safe diagnostic that contains neither response data nor credentials.""" class NoRedirect(HTTPRedirectHandler): def redirect_request(self, request, fp, code, message, headers, newurl): # An auth header must never follow a redirect to another origin. return None def origin_url(value): try: parsed = urlsplit(value) valid = ( parsed.scheme == "https" and bool(parsed.hostname) and parsed.username is None and parsed.password is None and parsed.path in ("", "/") and not parsed.query and not parsed.fragment and not any(char.isspace() or ord(char) < 32 or ord(char) > 126 for char in value) ) parsed.port # Validate the optional port without displaying the supplied URL. except ValueError: valid = False if not valid: raise FetchError("BOX_ORIGIN must be an HTTPS origin without a path, query or credentials; use an ASCII/Punycode hostname") return value.rstrip("/") def positive_integer(value): number = int(value) if number < 1: raise argparse.ArgumentTypeError("a positive integer is required") return number def fetch_page(origin, token, page, page_size, sorting="-timestamp", timeout=15): origin = origin_url(origin) if not token or any(ord(char) <= 32 or ord(char) > 126 for char in token): raise FetchError("BOX_ACCESS_TOKEN must contain the access_token without the Bearer prefix") if type(page) is not int or page < 1 or type(page_size) is not int or page_size < 1: raise FetchError("page and page_size must be positive integers") if not isinstance(sorting, str) or not sorting or any(ord(char) < 32 for char in sorting): raise FetchError("sorting must be a nonempty string supported by the server contract") if not math.isfinite(timeout) or timeout <= 0: raise FetchError("timeout must be a positive finite number") query = urlencode({"page": page, "page_size": page_size, "sorting": sorting}) request = Request(origin + ENDPOINT + "?" + query, headers={ "Authorization": "Bearer " + token, "Accept": "application/json", }, method="GET") # The standard trust store and hostname verification remain enabled. opener = build_opener(HTTPSHandler(context=ssl.create_default_context()), NoRedirect()) try: with opener.open(request, timeout=timeout) as response: if response.status != 200: raise FetchError("HTTP " + str(response.status) + ": page was not saved") body = response.read(MAX_BYTES + 1) except HTTPError as error: code = error.code error.close() if 300 <= code < 400: raise FetchError("HTTP " + str(code) + ": redirects are blocked; check BOX_ORIGIN") from None raise FetchError("HTTP " + str(code) + ": check permissions, token and parameters in Swagger") from None except (URLError, TimeoutError, OSError, HTTPException, ValueError): # Exception messages may contain URL, proxy details or response data. raise FetchError("Network error, TLS error or timeout; page was not saved") from None if len(body) > MAX_BYTES: raise FetchError("Response exceeds 10 MiB; reduce page_size") try: data = json.loads(body) except (ValueError, UnicodeError, RecursionError): raise FetchError("Response does not contain valid JSON; page was not saved") from None if not isinstance(data, dict) or not isinstance(data.get("events"), list): raise FetchError("Expected a JSON object with an events array; check the server contract") return data def download_page(output, **options): output = Path(output) if os.path.lexists(output): raise FetchError("Output file already exists; choose a new name") data = fetch_page(**options) try: # O_EXCL also protects against a file appearing after the check above. descriptor = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) except FileExistsError: raise FetchError("Output file already exists; choose a new name") from None except OSError: raise FetchError("Cannot create output file; check the directory and permissions") from None try: with os.fdopen(descriptor, "w", encoding="utf-8") as stream: json.dump(data, stream, ensure_ascii=False, indent=2, allow_nan=False) stream.write("\n") except (OSError, ValueError, RecursionError): try: output.unlink(missing_ok=True) except OSError: raise FetchError("JSON write was interrupted; inspect and remove the incomplete output file") from None raise FetchError("Cannot save JSON; check the data, free space and permissions") from None return {"status": "saved", "events": len(data["events"]), "page_requested": options["page"], "page_size_requested": options["page_size"]} def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--page", required=True, type=positive_integer) parser.add_argument("--page-size", required=True, type=positive_integer) parser.add_argument("--sorting", default="-timestamp", help="for a leading minus, use --sorting=-timestamp") parser.add_argument("--timeout", type=float, default=15, help="timeout for network operations, in seconds") parser.add_argument("--output", required=True, help="new JSON file; overwriting is not allowed") args = parser.parse_args(argv) try: result = download_page(args.output, origin=os.environ.get("BOX_ORIGIN", ""), token=os.environ.get("BOX_ACCESS_TOKEN", "").strip(), page=args.page, page_size=args.page_size, sorting=args.sorting, timeout=args.timeout) except FetchError as error: print("Error: " + str(error), file=sys.stderr) return 1 print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 0 if __name__ == "__main__": sys.exit(main())