diff --git a/README.md b/README.md index 52c2f5c..f39c4ab 100644 --- a/README.md +++ b/README.md @@ -191,8 +191,9 @@ ocl repo export status OWNER REPO VERSION --type source|collection # Trigger export creation (if not already cached) ocl repo export create OWNER REPO VERSION --type source|collection -# Download to a local file -ocl repo export download OWNER REPO VERSION --type source|collection -o FILENAME +# Download, saved under the API's name; never replaces an existing file +# (-o FILE saves it as FILE, overwriting; -o DIR saves it in that existing directory) +ocl repo export download OWNER REPO VERSION --type source|collection [-o FILE|DIR] # Delete a cached export ocl repo export delete OWNER REPO VERSION --type source|collection @@ -202,9 +203,12 @@ Example — download the latest CIEL release export: ```bash ocl repo export status CIEL CIEL v2026-03-23 --type source -ocl repo export download CIEL CIEL v2026-03-23 --type source -o CIEL_v2026-03-23.zip +ocl repo export download CIEL CIEL v2026-03-23 --type source +# Saved to orgs_CIEL_sources_CIEL_v2026-03-23_.zip ``` +The API names exports `[owner-type]_[owner]_[repo-type]_[repo]_[version]_[expansion]_[lastUpdated].zip`, with the expansion when the export has one. An older export the API can't name keeps its storage name; if no usable name comes back at all, the command asks for `-o FILE`. + ### Concepts ```bash diff --git a/src/ocl_cli/commands/export.py b/src/ocl_cli/commands/export.py index 925b274..c7ddcf7 100644 --- a/src/ocl_cli/commands/export.py +++ b/src/ocl_cli/commands/export.py @@ -1,6 +1,9 @@ """Export commands: status, create, delete, download.""" -import sys +import errno +import os +import re +from urllib.parse import parse_qs, unquote, urlsplit import click @@ -80,26 +83,111 @@ def delete(ctx, owner, repo, version, repo_type, owner_type): handle_api_error(e) +# OCL names (and so export and storage names) use only these characters; a leading '.' is never a name. +SAFE_FILENAME = re.compile(r"[A-Za-z0-9_@-][A-Za-z0-9._@-]*") +WINDOWS_DEVICE_NAMES = {"CON", "PRN", "AUX", "NUL", *(f"COM{i}" for i in range(1, 10)), + *(f"LPT{i}" for i in range(1, 10))} +DISPOSITION_PARAM = re.compile(r'\s*;\s*([^\s=;]+)\s*=\s*("(?:[^"\\]|\\.)*"|[^;]*)') + + +def _disposition_params(value): + """Parameters of a Content-Disposition value, lower-cased names, quoted values unescaped.""" + params = {} + pos = value.find(";") + while 0 <= pos < len(value): + match = DISPOSITION_PARAM.match(value, pos) + if not match: + break + raw = match.group(2).strip() + if raw.startswith('"') and raw.endswith('"') and len(raw) > 1: + raw = re.sub(r"\\(.)", r"\1", raw[1:-1]) + params.setdefault(match.group(1).lower(), raw) + pos = match.end() + return params + + +def _filename_from_content_disposition(value): + """The filename in a Content-Disposition value, preferring filename* (RFC 6266), or None.""" + params = _disposition_params(value or "") + extended = params.get("filename*", "") + if extended.count("'") >= 2: + charset, _, encoded = extended.split("'", 2) + try: + return unquote(encoded, encoding=charset or "utf-8", errors="strict") + except (LookupError, UnicodeDecodeError): + pass + return params.get("filename") + + +def _usable_filename(name): + if not name or not SAFE_FILENAME.fullmatch(name): + return False + return name.split(".")[0].upper() not in WINDOWS_DEVICE_NAMES + + +def export_filename(response): + """The name the API gives the export, or None. + + Tried in order: the Content-Disposition header, the signed URL's + response-content-disposition, then the URL's last path segment. A name + must be a plain file name made of the characters the API uses. + """ + url = urlsplit(str(response.url)) + candidates = ( + lambda: _filename_from_content_disposition(response.headers.get("content-disposition")), + lambda: _filename_from_content_disposition( + parse_qs(url.query).get("response-content-disposition", [None])[0]), + lambda: unquote(url.path.rsplit("/", 1)[-1]), + ) + for candidate in candidates: + try: + name = candidate() + except ValueError: + continue + if _usable_filename(name): + return name + return None + + @export.command() @export_args @click.option( "-o", "--output", "output_path", type=click.Path(), - help="Output file path (required).", - required=True, + help="Output file, or an existing directory (default: the API's name for the export, in the current directory).", ) @click.pass_context def download(ctx, owner, repo, version, repo_type, owner_type, output_path): - """Download an export file to a local path.""" + """Download an export file, named as the API names it unless -o gives a file.""" client = ctx.obj["client"] try: - click.echo(f"Downloading export...", err=True) + click.echo("Downloading export...", err=True) response = client.export_download( owner, repo, version, owner_type=owner_type, repo_type=repo_type, ) - with open(output_path, "wb") as f: - f.write(response.content) + # An explicit -o FILE is overwritten, as before. A name the API chose never replaces an existing file. + mode = "wb" + if not output_path or os.path.isdir(output_path): + filename = export_filename(response) + if not filename: + raise click.ClickException("The API didn't name the export. Pass -o FILE to choose a name.") + output_path = os.path.join(output_path or "", filename) + mode = "xb" + elif output_path.endswith(("/", os.sep)): + raise click.ClickException(f"Directory {output_path} doesn't exist.") + + try: + with open(output_path, mode) as f: + f.write(response.content) + except FileExistsError: + raise click.ClickException(f"{output_path} already exists. Remove it, or pass -o FILE to overwrite.") from None + except OSError as e: + if mode != "xb" or e.errno != errno.ENAMETOOLONG: + raise + raise click.ClickException( + "The export's name is too long for this filesystem. Pass -o FILE to choose a name." + ) from None size = len(response.content) click.echo(f"Saved to {output_path} ({size:,} bytes)", err=True) diff --git a/tests/test_export_commands.py b/tests/test_export_commands.py new file mode 100644 index 0000000..38e972d --- /dev/null +++ b/tests/test_export_commands.py @@ -0,0 +1,192 @@ +import os +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +import httpx +from click.testing import CliRunner + +from ocl_cli.main import cli + +API_NAME = "orgs_PIH_sources_PIH_1.8.24_2026-09-30_123456.zip" +SIGNED_URL = "https://exports.example.test/orgs/PIH/PIH_PIH_v1.8.24.2026-09-30_123456.zip?X-Amz-Signature=abc" + + +def export_response(url=SIGNED_URL, headers=None): + return httpx.Response(200, headers=headers, content=b"zip-bytes", request=httpx.Request("GET", url)) + + +def make_runner(): + # Click 8.1 mixes stderr into stdout unless told not to; 8.2 dropped mix_stderr. + try: + return CliRunner(mix_stderr=False) + except TypeError: + return CliRunner() + + +class FakeConfig: + def get_server(self, server_id): + return SimpleNamespace(base_url="https://api.example.test") + + def resolve_token(self, server, token_override=None): + return token_override + + +class FakeExportClient: + def __init__(self, response): + self.response = response + + def export_download(self, *args, **kwargs): + return self.response + + def close(self): + pass + + +class ExportDownloadCommandTest(unittest.TestCase): + def download(self, response, *args, exit_code=0): + runner = make_runner() + with ( + runner.isolated_filesystem(), + patch("ocl_cli.main.CLIConfig.load", return_value=FakeConfig()), + patch("ocl_cli.main.OCLAPIClient", return_value=FakeExportClient(response)), + ): + os.mkdir("downloads") + result = runner.invoke( + cli, ["repo", "export", "download", "PIH", "PIH", "1.8.24", "--type", "source", *args] + ) + self.assertEqual(result.exit_code, exit_code, result.output) + files = sorted( + os.path.join(root, name) for root, _, names in os.walk(".") for name in names + ) + return files, result + + def test_saves_under_the_content_disposition_name_by_default(self): + response = export_response(headers={"content-disposition": f'attachment; filename="{API_NAME}"'}) + + files, _ = self.download(response) + + self.assertEqual(files, [f"./{API_NAME}"]) + + def test_reads_the_name_from_the_signed_url_without_a_header(self): + url = f"{SIGNED_URL}&response-content-disposition=attachment%3B%20filename%3D%22{API_NAME}%22" + + files, _ = self.download(export_response(url=url)) + + self.assertEqual(files, [f"./{API_NAME}"]) + + def test_falls_back_to_the_storage_name(self): + files, _ = self.download(export_response()) + + self.assertEqual(files, ["./PIH_PIH_v1.8.24.2026-09-30_123456.zip"]) + + def test_output_directory_keeps_the_api_name(self): + response = export_response(headers={"content-disposition": f'attachment; filename="{API_NAME}"'}) + + files, _ = self.download(response, "-o", "downloads") + + self.assertEqual(files, [f"./downloads/{API_NAME}"]) + + def test_output_file_overrides_the_api_name(self): + response = export_response(headers={"content-disposition": f'attachment; filename="{API_NAME}"'}) + + files, _ = self.download(response, "-o", "mine.zip") + + self.assertEqual(files, ["./mine.zip"]) + + def test_a_path_in_the_name_is_rejected(self): + response = export_response(headers={"content-disposition": 'attachment; filename="../../evil.zip"'}) + + files, _ = self.download(response, "-o", "downloads") + + self.assertEqual(files, ["./downloads/PIH_PIH_v1.8.24.2026-09-30_123456.zip"]) + + def test_filename_star_wins_over_filename(self): + disposition = f"attachment; filename=\"fallback.zip\"; filename*=UTF-8''{API_NAME}" + + files, _ = self.download(export_response(headers={"content-disposition": disposition})) + + self.assertEqual(files, [f"./{API_NAME}"]) + + def test_unusable_header_name_falls_back_to_the_signed_url(self): + url = f"{SIGNED_URL}&response-content-disposition=attachment%3B%20filename%3D%22{API_NAME}%22" + for disposition in ("attachment; filename*=UTF-8''evil%00.zip", 'attachment; filename="NUL.zip"', + 'attachment; filename="victim.txt:stream.zip"', "attachment; filename*=UTF-8''%FF.zip"): + with self.subTest(disposition=disposition): + files, _ = self.download(export_response(url=url, headers={"content-disposition": disposition})) + + self.assertEqual(files, [f"./{API_NAME}"]) + + def test_quoted_parameters_can_contain_semicolons(self): + disposition = 'attachment; note="x; filename=wrong.zip; y"; filename="right.zip"' + + files, _ = self.download(export_response(headers={"content-disposition": disposition})) + + self.assertEqual(files, ["./right.zip"]) + + def test_storage_name_may_start_with_an_underscore(self): + url = "https://exports.example.test/orgs/_PIH/_PIH_PIH_v1.8.24.zip?X-Amz-Signature=abc" + + files, _ = self.download(export_response(url=url)) + + self.assertEqual(files, ["./_PIH_PIH_v1.8.24.zip"]) + + def test_an_existing_file_is_not_overwritten(self): + response = export_response(headers={"content-disposition": f'attachment; filename="{API_NAME}"'}) + runner = make_runner() + with ( + runner.isolated_filesystem(), + patch("ocl_cli.main.CLIConfig.load", return_value=FakeConfig()), + patch("ocl_cli.main.OCLAPIClient", return_value=FakeExportClient(response)), + ): + with open("important.txt", "w") as f: + f.write("keep") + os.symlink("important.txt", API_NAME) + result = runner.invoke(cli, ["repo", "export", "download", "PIH", "PIH", "1.8.24", "--type", "source"]) + + self.assertEqual(result.exit_code, 1, result.output) + self.assertIn("already exists", result.stderr) + with open("important.txt") as f: + self.assertEqual(f.read(), "keep") + + def test_a_name_too_long_for_the_filesystem_asks_for_output(self): + long_name = "orgs_PIH_sources_" + "x" * 300 + ".zip" + response = export_response(headers={"content-disposition": f'attachment; filename="{long_name}"'}) + + files, result = self.download(response, exit_code=1) + + self.assertEqual(files, []) + self.assertIn("too long", result.stderr) + + def test_a_missing_output_directory_is_an_error(self): + files, result = self.download(export_response(), "-o", "missing/", exit_code=1) + + self.assertEqual(files, []) + self.assertIn("doesn't exist", result.stderr) + + def test_unnamed_export_asks_for_output(self): + response = export_response(url="https://api.example.test/orgs/PIH/sources/PIH/1.8.24/export/") + + files, result = self.download(response, exit_code=1) + + self.assertEqual(files, []) + self.assertIn("Pass -o FILE", result.stderr) + + +def test_export_download_follows_the_redirect_to_the_named_file(httpx_mock): + from ocl_cli.api_client import OCLAPIClient + from ocl_cli.commands.export import export_filename + + httpx_mock.add_response( + method="GET", url="https://api.example.test/orgs/PIH/sources/PIH/1.8.24/export/", + status_code=302, headers={"location": SIGNED_URL}, + ) + httpx_mock.add_response( + method="GET", url=SIGNED_URL, content=b"zip-bytes", + headers={"content-disposition": f'attachment; filename="{API_NAME}"'}, + ) + + response = OCLAPIClient(base_url="https://api.example.test", token="t").export_download("PIH", "PIH", "1.8.24") + + assert response.content == b"zip-bytes" + assert export_filename(response) == API_NAME