diff --git a/CHANGELOG.md b/CHANGELOG.md index c19a90ce..4d613f53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ## [Unreleased] +### Added + +- `cloudsmith credential-helper domains` emits a versioned JSON document — `{"version": 1, "domains": [{"host": ..., "format": ..., "enabled": ..., "validated": ..., "type": ..., "domain_type": ..., "org": ..., "repository": ...}]}` — listing the hosts Cloudsmith can authenticate: the built-in service domains plus any custom domains configured for an organisation, along with the package format each one serves (`format` is `null` for hosts that serve no single format, like the download CDN). Each entry's `domain_type` says what the host is for — `download` for the CDN, `upload` for the generic upload endpoint, or `native_api` for a host that speaks a single package format's own protocol — so a consumer can pick the right host without pattern-matching on hostnames. `org` and `repository` say which organisation a custom domain belongs to and, when it is scoped to one, which repository — built-in entries carry `null` for both. Disabled and unvalidated custom domains are included too, with their state intact, so a misconfigured domain is visible rather than silently missing. A failed custom-domain lookup (typo'd organisation, missing permission, unreachable API) exits non-zero with a message on stderr instead of silently listing built-in hosts only. The command takes no arguments: the organisation is resolved from `CLOUDSMITH_ORG` or `oidc_org` in `config.ini`. With an organisation configured, its custom domains are looked up (and a failed lookup still exits non-zero); with none configured, the command instead lists whatever custom domains an earlier run already cached, at no API cost, rather than querying every organisation the caller belongs to. Custom-domain lookups are cached on disk for seven days, since custom domains change rarely; pass `--refresh` to bypass the cache and fetch fresh data, for example right after adding or validating a domain. Like `credential-helper generic`, this command is for programmatic consumers only and has no human-readable output mode. +- The built-in domain list can be replaced via a `[domains]` section in `config.ini`, mapping hostname to package format, for dedicated and on-premise deployments. For safety this is honoured only from trusted configuration — your user-level config directory, `~/.cloudsmith`, or an explicit `--config-file`. A `[domains]` section in a directory-relative `config.ini` is ignored with a warning, because that file can be committed to a repository and this list decides which hosts may receive a credential. + ## [1.20.2] - 2026-07-31 ### Fixed diff --git a/cloudsmith_cli/cli/commands/credential_helper/__init__.py b/cloudsmith_cli/cli/commands/credential_helper/__init__.py index 93e5feae..3c9fd746 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/__init__.py +++ b/cloudsmith_cli/cli/commands/credential_helper/__init__.py @@ -10,6 +10,7 @@ from ..main import main from .docker import docker as docker_cmd +from .domains import domains_cmd from .manage import install_cmd, list_cmd, uninstall_cmd @@ -32,6 +33,7 @@ def credential_helper(): credential_helper.add_command(docker_cmd, name="docker") +credential_helper.add_command(domains_cmd, name="domains") credential_helper.add_command(install_cmd, name="install") credential_helper.add_command(uninstall_cmd, name="uninstall") credential_helper.add_command(list_cmd, name="list") diff --git a/cloudsmith_cli/cli/commands/credential_helper/domains.py b/cloudsmith_cli/cli/commands/credential_helper/domains.py new file mode 100644 index 00000000..dbde55a1 --- /dev/null +++ b/cloudsmith_cli/cli/commands/credential_helper/domains.py @@ -0,0 +1,158 @@ +# Copyright 2026 Cloudsmith Ltd +"""List the hosts Cloudsmith can authenticate: built-in and custom domains. + +The built-in ``*.cloudsmith.io`` service hosts ship with the CLI so that every +consumer shares one authoritative table; custom domains cannot be inferred, so +discovering them requires an API lookup. +""" + +from __future__ import annotations + +import json +import os + +import click + +from ....credential_helpers.custom_domains import ( + get_custom_domains, + read_all_cached_domains, +) +from ....credential_helpers.default_domains import ( + domain_type_for_backend_kind, + format_for_backend_kind, + load_default_domains, + untrusted_config_declares_domains, +) +from ...config import get_default_config_path +from ...decorators import ( + common_api_auth_options, + common_cli_config_options, + resolve_credentials, +) + +PROTOCOL_VERSION = 1 + + +def _custom_entries(records) -> list[dict]: + """Render custom-domain records as document entries.""" + return [ + { + "host": record.host, + "format": format_for_backend_kind(record.backend_kind), + "enabled": record.enabled, + "validated": record.validated, + "type": "custom", + "domain_type": domain_type_for_backend_kind(record.backend_kind).value, + "org": record.org, + "repository": record.repository, + } + for record in sorted(records, key=lambda record: (record.org, record.host)) + ] + + +@click.command("domains") +@click.option( + "--refresh", + is_flag=True, + default=False, + help="Bypass the custom-domain cache and fetch fresh data from the API.", +) +@common_cli_config_options +@common_api_auth_options +@resolve_credentials +@click.pass_context +def domains_cmd(ctx, opts, refresh: bool) -> None: + """Emit the Cloudsmith hosts and their package formats as JSON. + + Emits machine-readable JSON for programmatic consumers listing the + built-in Cloudsmith service hosts (``*.cloudsmith.io``) alongside custom + domains. Each entry's ``type`` field distinguishes ``default`` from + ``custom``, and ``domain_type`` says what the host is for: ``download``, + ``upload``, or ``native_api`` for a host speaking one package format's own + protocol. Every domain is included, even custom ones that are disabled or + not yet validated, so a domain that is present but not working is visible + rather than silently absent. Consumers that need a usable host should + filter on both ``enabled`` and ``validated``. + + The built-in hosts are always listed and need no organisation or + authentication. When an organisation is configured - CLOUDSMITH_ORG, or + ``oidc_org`` in ``config.ini`` - its custom domains are looked up, and a + failed lookup exits non-zero rather than being rendered as "no domains". + With none configured, the command instead lists the custom domains any + earlier run already cached and makes no API call at all: there is + deliberately no lookup of which organisations the caller belongs to, since + that would cost one API call per organisation on every run. A custom + entry's ``org`` and ``repository`` fields say which organisation it + belongs to and, when scoped to one, which repository - built-in entries + carry ``null`` for both. Pass ``--refresh`` to bypass the cache and fetch + fresh data from the API for a configured organisation. + + Output (stdout): + JSON: {"version": 1, "domains": [{"host": ..., "format": ..., + "enabled": ..., "validated": ..., "type": ..., "domain_type": ..., + "org": ..., "repository": ...}]} + + Examples: + + \b + # List built-in hosts plus any custom domains already cached + $ cloudsmith credential-helper domains + + \b + # List built-in hosts plus an organisation's custom domains + $ CLOUDSMITH_ORG=my-org cloudsmith credential-helper domains + """ + explicit_config = ctx.meta.get("config_file") + if explicit_config and os.path.isdir(explicit_config): + explicit_config = os.path.join(explicit_config, "config.ini") + + # An explicit --config-file is trusted and read directly, so the + # untrusted-cwd scan does not apply — warning about the very file the + # user passed would claim the opposite of what happens. + if explicit_config is None and untrusted_config_declares_domains(): + click.secho( + "Warning: ignoring [domains] section in ./config.ini -- config.ini " + "in the current directory is untrusted. Put it in " + f"{get_default_config_path()} or ~/.cloudsmith instead.", + fg="yellow", + err=True, + ) + + data = [ + { + "host": domain.host, + "format": domain.format_label, + "enabled": True, + "validated": True, + "type": "default", + "domain_type": domain.domain_type.value, + "org": None, + "repository": None, + } + for domain in load_default_domains(config_path=explicit_config) + ] + + org = opts.oidc_org + if org: + try: + records = get_custom_domains( + org, + credential=opts.credential, + api_host=opts.api_host, + refresh=refresh, + strict=True, + ) + except Exception as exc: # pylint: disable=broad-except + # Presentation boundary: a named organisation that cannot be read + # is the user's error, and must not be rendered as "no domains". + raise click.ClickException( + f"Failed to fetch custom domains for {org!r}: {exc}" + ) from exc + else: + # No organisation named: list what earlier runs cached rather than + # spending an API call per organisation to rediscover it. + records = read_all_cached_domains() + + data.extend(_custom_entries(records)) + + click.echo(json.dumps({"version": PROTOCOL_VERSION, "domains": data})) diff --git a/cloudsmith_cli/cli/tests/commands/test_credential_helper.py b/cloudsmith_cli/cli/tests/commands/test_credential_helper.py index 113f9366..a1e6f7a4 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper.py @@ -2,6 +2,8 @@ import io import json +import os +import time from unittest.mock import patch import httpretty @@ -9,17 +11,21 @@ import pytest from ....cli.commands.credential_helper.docker import docker +from ....core.api.exceptions import ApiException from ....core.api.init import unset_api_key from ....core.credentials.models import CredentialResult from ....credential_helpers.backends import BackendKind from ....credential_helpers.custom_domains import ( + CACHE_FORMAT_VERSION, CustomDomain, get_cache_path, get_custom_domains, get_format_domains, + is_cache_valid, read_cache, write_cache, ) +from ....credential_helpers.default_domains import DomainScope, domain_scope from ....credential_helpers.docker.runtime import ( _REFUSAL_MESSAGE, execute, @@ -237,10 +243,11 @@ def _cache_dir(tmp_path, monkeypatch): [ # 200 → records built and cached (200, True, True), - # 402 → [] and cached (feature not enabled) + # 402 → [] and cached (feature not enabled - a stable answer) (402, False, True), - # 404 → [] and cached (org not found) - (404, False, True), + # 404 → [] but NOT cached (org not found - probably a typo, must + # fail loudly every time rather than replay a cached failure) + (404, False, False), # 403 → [] but NOT cached (may succeed after auth) (403, False, False), # 401 → [] but NOT cached (same branch as 403) @@ -303,6 +310,151 @@ def test_get_custom_domains_status_matrix( assert httpretty.last_request().headers.get("X-Api-Key") == "k_abc" +@pytest.mark.parametrize( + "status,expect_raise", + [ + (200, False), + # 402 genuinely means "no custom domains" (feature not enabled). + (402, False), + (401, True), + (403, True), + (404, True), + (500, True), + ], +) +@httpretty.activate(allow_net_connect=False) +def test_get_custom_domains_strict_raises_on_failure( + tmp_path, monkeypatch, status, expect_raise +): + """strict=True re-raises failures instead of degrading to []. + + Consumers presenting results to a user (`credential-helper domains`) must + not render a typo'd org or an unreachable API as "no custom domains". + """ + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + lambda: str(tmp_path), + ) + body = json.dumps([]) if status == 200 else json.dumps({"detail": "error"}) + httpretty.register_uri( + httpretty.GET, + f"{API_HOST}/orgs/acme/custom-domains/", + body=body, + status=status, + content_type="application/json", + ) + + credential = CredentialResult(api_key="k_abc", source_name="test") + if expect_raise: + with pytest.raises(ApiException): + get_custom_domains( + "acme", credential=credential, api_host=API_HOST, strict=True + ) + else: + result = get_custom_domains( + "acme", credential=credential, api_host=API_HOST, strict=True + ) + assert result == [] + + +@httpretty.activate(allow_net_connect=False) +def test_get_custom_domains_does_not_cache_a_failure(tmp_path, monkeypatch): + """A failed lookup writes no cache entry. + + Caching [] for a 404 is what forced strict mode to bypass the cache: a + typo'd organisation would otherwise be served back as a successful + "no custom domains" for the whole TTL. + """ + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + lambda: str(tmp_path), + ) + httpretty.register_uri( + httpretty.GET, + f"{API_HOST}/orgs/my-ogr/custom-domains/", + body=json.dumps({"detail": "not found"}), + status=404, + content_type="application/json", + ) + credential = CredentialResult(api_key="k_abc", source_name="test") + + assert get_custom_domains("my-ogr", credential=credential, api_host=API_HOST) == [] + + assert not get_cache_path("my-ogr").exists() + with pytest.raises(ApiException): + get_custom_domains( + "my-ogr", credential=credential, api_host=API_HOST, strict=True + ) + + +@httpretty.activate(allow_net_connect=False) +def test_get_custom_domains_caches_a_402(tmp_path, monkeypatch): + """A 402 (feature not enabled) is a stable answer, not a failure. + + Unlike a 404, it is cached: the org's product entitlement isn't going to + change between one credential-helper invocation and the next, so caching + it avoids hitting the API forever. A second lookup within the TTL must + be served from the cache, not a second request. + """ + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + lambda: str(tmp_path), + ) + httpretty.register_uri( + httpretty.GET, + f"{API_HOST}/orgs/acme/custom-domains/", + body=json.dumps({"detail": "payment required"}), + status=402, + content_type="application/json", + ) + credential = CredentialResult(api_key="k_abc", source_name="test") + + assert get_custom_domains("acme", credential=credential, api_host=API_HOST) == [] + assert read_cache(get_cache_path("acme")) == [] + assert len(httpretty.latest_requests()) == 1 + + assert get_custom_domains("acme", credential=credential, api_host=API_HOST) == [] + assert len(httpretty.latest_requests()) == 1 + + +def test_get_custom_domains_strict_is_served_from_cache(tmp_path, monkeypatch): + """Failures are never cached, so a cache hit is always a real result. + + httpretty is not active here: a request would raise, proving the cache + answered. + """ + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + lambda: str(tmp_path), + ) + domains = [ + CustomDomain( + host="dl.acme.com", + backend_kind=None, + enabled=True, + validated=True, + org="acme", + ) + ] + write_cache(get_cache_path("acme"), domains) + + assert get_custom_domains("acme", api_host=API_HOST, strict=True) == domains + + +def test_cache_is_valid_for_seven_days(tmp_path, monkeypatch): + """Custom domains change rarely; a week-long cache keeps CI off the API.""" + cache_path = tmp_path / "acme.json" + cache_path.write_text("{}", encoding="utf-8") + + six_days = time.time() - 6 * 24 * 3600 + os.utime(cache_path, (six_days, six_days)) + assert is_cache_valid(cache_path) is True + + eight_days = time.time() - 8 * 24 * 3600 + os.utime(cache_path, (eight_days, eight_days)) + assert is_cache_valid(cache_path) is False + + @httpretty.activate(allow_net_connect=False) def test_get_custom_domains_bearer_credential_sends_authorization_header( tmp_path, monkeypatch @@ -353,8 +505,6 @@ def test_get_custom_domains_bearer_credential_sends_authorization_header( ) def test_get_custom_domains_cache_edge(tmp_path, monkeypatch, scenario, expected): """Cache format edge cases: legacy string-list is a miss; empty list is a hit.""" - import time - monkeypatch.setattr( "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", lambda: str(tmp_path), @@ -363,9 +513,17 @@ def test_get_custom_domains_cache_edge(tmp_path, monkeypatch, scenario, expected cache_path = get_cache_path("acme") if scenario == "legacy_string_list": - data = {"domains": ["docker.acme.com"], "cached_at": time.time()} + data = { + "format_version": CACHE_FORMAT_VERSION, + "domains": ["docker.acme.com"], + "cached_at": time.time(), + } else: - data = {"domains": [], "cached_at": time.time()} + data = { + "format_version": CACHE_FORMAT_VERSION, + "domains": [], + "cached_at": time.time(), + } cache_path.write_text(json.dumps(data), encoding="utf-8") assert read_cache(cache_path) == expected @@ -468,7 +626,11 @@ def test_get_format_domains_filters_correctly(tmp_path, monkeypatch): "acme", [ CustomDomain( - host="docker.acme.com", backend_kind=6, enabled=True, validated=True + host="docker.acme.com", + backend_kind=6, + enabled=True, + validated=True, + org="acme", ) ], None, @@ -484,6 +646,7 @@ def test_get_format_domains_filters_correctly(tmp_path, monkeypatch): backend_kind=6, enabled=False, validated=True, + org="acme", ) ], None, @@ -499,6 +662,7 @@ def test_get_format_domains_filters_correctly(tmp_path, monkeypatch): backend_kind=6, enabled=True, validated=False, + org="acme", ) ], None, @@ -510,7 +674,11 @@ def test_get_format_domains_filters_correctly(tmp_path, monkeypatch): "acme", [ CustomDomain( - host="docker.acme.com", backend_kind=6, enabled=True, validated=True + host="docker.acme.com", + backend_kind=6, + enabled=True, + validated=True, + org="acme", ) ], BackendKind.DOCKER, @@ -522,7 +690,11 @@ def test_get_format_domains_filters_correctly(tmp_path, monkeypatch): "acme", [ CustomDomain( - host="npm.acme.com", backend_kind=9, enabled=True, validated=True + host="npm.acme.com", + backend_kind=9, + enabled=True, + validated=True, + org="acme", ) ], BackendKind.DOCKER, @@ -535,7 +707,11 @@ def test_get_format_domains_filters_correctly(tmp_path, monkeypatch): "acme", [ CustomDomain( - host="docker.acme.com", backend_kind=6, enabled=True, validated=True + host="docker.acme.com", + backend_kind=6, + enabled=True, + validated=True, + org="acme", ) ], BackendKind.DOCKER, @@ -603,7 +779,11 @@ def test_get_credentials_refuses_npm_custom_domain(tmp_path, monkeypatch): cache_path, [ CustomDomain( - host="npm.acme.com", backend_kind=9, enabled=True, validated=True + host="npm.acme.com", + backend_kind=9, + enabled=True, + validated=True, + org="acme", ) ], ) @@ -614,3 +794,165 @@ def test_get_credentials_refuses_npm_custom_domain(tmp_path, monkeypatch): ) assert result is None + + +# --------------------------------------------------------------------------- +# 11. Custom domain repository scope +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "repository,repository_only,expected", + [ + ("prod", True, DomainScope.REPOSITORY), + ("prod", False, DomainScope.ORGANIZATION), + # repository_only with no repository is a server-side contradiction: + # honouring it would build a URL with no identifying segment at all. + (None, True, DomainScope.ORGANIZATION), + (None, False, DomainScope.ORGANIZATION), + ], +) +def test_custom_domain_scope(repository, repository_only, expected): + """A domain is repository-scoped only when it names a repo and is bound to it.""" + domain = CustomDomain( + host="dl-prod.acme.com", + backend_kind=None, + enabled=True, + validated=True, + org="acme", + repository=repository, + repository_only=repository_only, + ) + assert domain.scope is expected + + +@pytest.mark.parametrize( + "value,expected", + [ + ("repository", DomainScope.REPOSITORY), + ("organization", DomainScope.ORGANIZATION), + # A hand-edited config must not break every wrapped run. + ("nonsense", DomainScope.ORGANIZATION), + (None, DomainScope.ORGANIZATION), + ("", DomainScope.ORGANIZATION), + ], +) +def test_domain_scope_resolves_persisted_values(value, expected): + """A persisted scope string resolves, degrading to organisation.""" + assert domain_scope(value) is expected + + +@httpretty.activate(allow_net_connect=False) +def test_get_custom_domains_parses_repository_scope(tmp_path, monkeypatch): + """The nested repository payload and repository_only reach the record.""" + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + lambda: str(tmp_path), + ) + httpretty.register_uri( + httpretty.GET, + f"{API_HOST}/orgs/acme/custom-domains/", + body=json.dumps( + [ + { + "host": "dl-prod.acme.com", + "backend_kind": None, + "enabled": True, + "validated": True, + "repository": {"name": "Prod", "slug": "prod"}, + "repository_only": True, + }, + { + "host": "dl.acme.com", + "backend_kind": None, + "enabled": True, + "validated": True, + "repository": None, + "repository_only": False, + }, + ] + ), + status=200, + content_type="application/json", + ) + + records = get_custom_domains( + "acme", + credential=CredentialResult(api_key="k_abc", source_name="test"), + api_host=API_HOST, + ) + + by_host = {record.host: record for record in records} + assert by_host["dl-prod.acme.com"].repository == "prod" + assert by_host["dl-prod.acme.com"].scope is DomainScope.REPOSITORY + assert by_host["dl-prod.acme.com"].org == "acme" + assert by_host["dl.acme.com"].repository is None + assert by_host["dl.acme.com"].scope is DomainScope.ORGANIZATION + + +def test_custom_domain_cache_round_trips_scope(tmp_path): + """Scope survives the cache: a miss here would build wrong URLs for 7 days.""" + cache_path = tmp_path / "acme.json" + domains = [ + CustomDomain( + host="dl-prod.acme.com", + backend_kind=None, + enabled=True, + validated=True, + org="acme", + repository="prod", + repository_only=True, + ) + ] + + write_cache(cache_path, domains) + + assert read_cache(cache_path) == domains + + +def test_custom_domain_cache_rejects_an_older_format_version(tmp_path): + """A pre-scope cache document is a miss, not an organisation-scoped record. + + Serving it would report a repository-scoped domain as organisation-wide + for the whole TTL. + """ + cache_path = tmp_path / "acme.json" + cache_path.write_text( + json.dumps( + { + "format_version": 1, + "cached_at": time.time(), + "domains": [ + { + "host": "dl-prod.acme.com", + "backend_kind": None, + "enabled": True, + "validated": True, + } + ], + } + ), + encoding="utf-8", + ) + + assert read_cache(cache_path) is None + + +@pytest.mark.parametrize( + "top_level", + [ + [], + ["dl.acme.com"], + ], +) +def test_custom_domain_cache_rejects_a_non_dict_document(tmp_path, top_level): + """A cache file whose top-level JSON is a list is a miss, not a crash. + + A hand-edited or corrupted cache file isn't bound by what this codebase + writes; ``.get`` on a list would raise before the version gate got a + chance to degrade it to a miss. + """ + cache_path = tmp_path / "acme.json" + cache_path.write_text(json.dumps(top_level), encoding="utf-8") + + assert read_cache(cache_path) is None diff --git a/cloudsmith_cli/cli/tests/commands/test_credential_helper_domains.py b/cloudsmith_cli/cli/tests/commands/test_credential_helper_domains.py new file mode 100644 index 00000000..e5cd3b64 --- /dev/null +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper_domains.py @@ -0,0 +1,431 @@ +# Copyright 2026 Cloudsmith Ltd +"""Tests for the `cloudsmith credential-helper domains` command.""" + +import json +import os +import time +from unittest.mock import patch + +import pytest + +from ....cli import config as cli_config +from ....cli.commands.credential_helper.domains import domains_cmd +from ....credential_helpers import default_domains +from ....credential_helpers.custom_domains import ( + CustomDomain, + get_cache_path, + write_cache, +) + +_PATCH_TARGET = ( + "cloudsmith_cli.cli.commands.credential_helper.domains.get_custom_domains" +) + +HERMETIC_ARGS = ["--api-key", "fake-api-key"] + + +@pytest.fixture(autouse=True) +def hermetic_environment(monkeypatch): + """Keep the developer's real environment out of these tests. + + An inherited CLOUDSMITH_ORG would trigger a live custom-domain lookup, an + inherited CLOUDSMITH_CONFIG_FILE would supply a foreign domain table, and a + [domains] section in the developer's real trusted config.ini would replace + the built-in hosts these tests assert on. The CLI's Options object is a + process-wide thread-local and ConfigReader.load_config permanently + prepends any explicit --config-file to class-level state, so both are + pinned per test to stop one test's organisation leaking into the next. + """ + monkeypatch.delenv("CLOUDSMITH_ORG", raising=False) + monkeypatch.delenv("CLOUDSMITH_CONFIG_FILE", raising=False) + monkeypatch.setattr(default_domains, "_trusted_domains", lambda: None) + monkeypatch.delattr(cli_config.OPTIONS, "value", raising=False) + monkeypatch.setattr(cli_config.ConfigReader, "config_files", ["config.ini"]) + monkeypatch.setattr(cli_config.ConfigReader, "config_searchpath", ["."]) + + +def _domain( + host, + backend_kind, + enabled=True, + validated=True, + org="acme", + repository=None, + repository_only=False, +): + """Build a CustomDomain record for a test.""" + return CustomDomain( + host=host, + backend_kind=backend_kind, + enabled=enabled, + validated=validated, + org=org, + repository=repository, + repository_only=repository_only, + ) + + +def _invoke(runner, *args, custom_domains=()): + """Run the command with the custom-domain lookup stubbed; assert a clean exit. + + Returns (result, mock_get) so a caller can read stderr or assert on how the + lookup was called. + """ + with patch(_PATCH_TARGET, return_value=list(custom_domains)) as mock_get: + result = runner.invoke( + domains_cmd, [*args, *HERMETIC_ARGS], catch_exceptions=False + ) + + assert result.exit_code == 0, result.output + return result, mock_get + + +def _by_host(result): + """Index the emitted domains document by host.""" + return {entry["host"]: entry for entry in json.loads(result.stdout)["domains"]} + + +def test_document_has_exactly_version_and_domains_keys(runner): + """The top-level document has exactly the keys `version` and `domains`.""" + result, _ = _invoke(runner) + + document = json.loads(result.stdout) + assert set(document.keys()) == {"version", "domains"} + assert document["version"] == 1 + + +def test_custom_entry_has_exact_key_set(runner, monkeypatch): + """A custom entry's full key set is exactly the eight expected fields.""" + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + + result, _ = _invoke(runner, custom_domains=[_domain("pypi.acme.example.com", 3)]) + + assert _by_host(result)["pypi.acme.example.com"] == { + "host": "pypi.acme.example.com", + "format": "python", + "enabled": True, + "validated": True, + "type": "custom", + "domain_type": "native_api", + "org": "acme", + "repository": None, + } + + +def test_custom_domains_resolve_format_and_domain_type(runner, monkeypatch): + """A custom domain's backend kind resolves to its format and domain type.""" + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + records = [ + _domain("pypi.acme.example.com", 3), + _domain("dl.acme.example.com", None), + ] + + by_host = _by_host(_invoke(runner, custom_domains=records)[0]) + + assert by_host["pypi.acme.example.com"]["format"] == "python" + assert by_host["pypi.acme.example.com"]["domain_type"] == "native_api" + assert by_host["dl.acme.example.com"]["format"] is None + assert by_host["dl.acme.example.com"]["domain_type"] == "download" + + +def test_formatless_hosts_have_null_format(runner): + """Hosts serving no single format say so as JSON null, not a sentinel.""" + by_host = _by_host(_invoke(runner)[0]) + + assert by_host["dl.cloudsmith.io"]["format"] is None + assert by_host["upload.cloudsmith.io"]["format"] is None + assert by_host["maven.cloudsmith.io"]["format"] == "maven" + + +def test_disabled_and_unvalidated_domains_are_listed(runner, monkeypatch): + """A domain that is not usable is still present, with its state intact.""" + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + records = [ + _domain("stale.acme.example.com", 3, enabled=False, validated=False), + _domain("healthy.acme.example.com", 3, enabled=True, validated=True), + ] + + by_host = _by_host(_invoke(runner, custom_domains=records)[0]) + + assert by_host["stale.acme.example.com"]["enabled"] is False + assert by_host["stale.acme.example.com"]["validated"] is False + assert by_host["healthy.acme.example.com"]["enabled"] is True + assert by_host["healthy.acme.example.com"]["validated"] is True + + +@pytest.mark.parametrize( + "host,domain_type", + [ + ("dl.cloudsmith.io", "download"), + ("upload.cloudsmith.io", "upload"), + ("python.cloudsmith.io", "native_api"), + ], +) +def test_default_entries_expose_domain_type(runner, host, domain_type): + """Built-in entries say whether they are for download, upload or the API.""" + result, _ = _invoke(runner) + + assert _by_host(result)[host]["domain_type"] == domain_type + + +@pytest.mark.parametrize("org", [None, "acme"]) +def test_builtin_hosts_are_always_listed(runner, monkeypatch, org): + """Built-in hosts need no organisation, and survive an empty custom lookup.""" + if org: + monkeypatch.setenv("CLOUDSMITH_ORG", org) + + result, _ = _invoke(runner) + + assert "python.cloudsmith.io" in _by_host(result) + + +def test_defaults_are_listed_alongside_custom_domains(runner, monkeypatch): + """Built-in hosts and custom domains appear in one document, typed.""" + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + + by_host = _by_host( + _invoke(runner, custom_domains=[_domain("pypi.acme.example.com", 3)])[0] + ) + + assert by_host["python.cloudsmith.io"]["type"] == "default" + assert by_host["pypi.acme.example.com"]["type"] == "custom" + + +def test_org_resolves_from_environment(runner, monkeypatch): + """CLOUDSMITH_ORG selects the organisation whose custom domains are listed.""" + monkeypatch.setenv("CLOUDSMITH_ORG", "acme-from-env") + + _, mock_get = _invoke(runner) + + assert mock_get.call_args.args[0] == "acme-from-env" + + +def test_org_resolves_from_config_file(runner, tmp_path): + """oidc_org in a trusted config.ini selects the organisation.""" + config = tmp_path / "config.ini" + config.write_text("[default]\noidc_org = acme-from-config\n", encoding="utf-8") + + _, mock_get = _invoke(runner, "--config-file", str(config)) + + assert mock_get.call_args.args[0] == "acme-from-config" + + +def test_without_org_no_custom_domain_lookup(runner): + """With no organisation configured, no custom-domain lookup is attempted.""" + _, mock_get = _invoke(runner) + + mock_get.assert_not_called() + + +def test_lookup_uses_resolved_credential(runner, monkeypatch): + """The custom-domain lookup authenticates with the resolved credential.""" + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + + _, mock_get = _invoke(runner) + + credential = mock_get.call_args.kwargs["credential"] + assert credential.api_key == "fake-api-key" + + +def test_lookup_is_strict_so_failures_are_loud(runner, monkeypatch): + """The lookup runs in strict mode: failures raise instead of degrading. + + A best-effort lookup would silently pretend a typo'd org, missing + permission or unreachable API means "no custom domains" (exit 0). + """ + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + + _, mock_get = _invoke(runner) + + assert mock_get.call_args.kwargs["strict"] is True + + +def test_lookup_failure_is_a_clean_error(runner, monkeypatch): + """A failed lookup exits non-zero with a message and no partial document.""" + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + + with patch(_PATCH_TARGET, side_effect=RuntimeError("boom")): + result = runner.invoke(domains_cmd, HERMETIC_ARGS, catch_exceptions=False) + + assert result.exit_code != 0 + assert "Failed to fetch custom domains" in result.stderr + assert result.stdout.strip() == "" + + +def test_explicit_config_file_supplies_default_domains(runner, tmp_path): + """An explicit --config-file is a trusted source for [domains].""" + config = tmp_path / "config.ini" + config.write_text( + "[domains]\nindex.internal.example.com = python\n", encoding="utf-8" + ) + + result, _ = _invoke(runner, "--config-file", str(config)) + + # The built-in table was replaced wholesale, not merged. + assert set(_by_host(result)) == {"index.internal.example.com"} + + +def test_untrusted_config_warning_does_not_contaminate_stdout( + runner, monkeypatch, tmp_path +): + """A cwd config.ini's [domains] section triggers a stderr-only warning.""" + monkeypatch.chdir(tmp_path) + (tmp_path / "config.ini").write_text( + "[domains]\nindex.internal.example.com = python\n", encoding="utf-8" + ) + + result, _ = _invoke(runner) + + assert "Warning" in result.stderr + # The untrusted [domains] section was ignored, so built-in hosts remain. + assert "python.cloudsmith.io" in _by_host(result) + + +def test_untrusted_warning_suppressed_for_explicit_config_file( + runner, monkeypatch, tmp_path +): + """--config-file silences the cwd warning: the explicit file is honoured. + + Warning "ignoring [domains] in ./config.ini" while honouring that very + file (the user explicitly passed it) would claim the opposite of what + happens. + """ + monkeypatch.chdir(tmp_path) + (tmp_path / "config.ini").write_text( + "[domains]\nindex.internal.example.com = python\n", encoding="utf-8" + ) + + result, _ = _invoke(runner, "--config-file", str(tmp_path / "config.ini")) + + assert "Warning" not in result.stderr + assert set(_by_host(result)) == {"index.internal.example.com"} + + +def test_lists_cached_domains_when_no_organisation_is_configured( + runner, monkeypatch, tmp_path +): + """A run with no org lists what earlier runs cached, without any API call. + + Attribution comes from the record, not the filename: the cache filename is + a sanitised slug and cannot be relied on. + """ + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + lambda: str(tmp_path), + ) + called = [] + monkeypatch.setattr( + "cloudsmith_cli.cli.commands.credential_helper.domains.get_custom_domains", + lambda *a, **k: called.append("api") or [], + ) + write_cache(get_cache_path("acme"), [_domain("dl.acme.com", None, org="acme")]) + write_cache( + get_cache_path("widgets"), [_domain("dl.widgets.com", None, org="widgets")] + ) + + result = runner.invoke(domains_cmd, ["-k", "fake-key"]) + + assert result.exit_code == 0, result.output + assert not called + custom = [d for d in json.loads(result.output)["domains"] if d["type"] == "custom"] + assert {(d["host"], d["org"]) for d in custom} == { + ("dl.acme.com", "acme"), + ("dl.widgets.com", "widgets"), + } + + +def test_stale_cache_entries_are_not_listed(runner, monkeypatch, tmp_path): + """An expired entry is skipped, not served past its TTL.""" + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + lambda: str(tmp_path), + ) + cache_path = get_cache_path("acme") + write_cache(cache_path, [_domain("dl.acme.com", None, org="acme")]) + stale = time.time() - 8 * 24 * 3600 + os.utime(cache_path, (stale, stale)) + + result = runner.invoke(domains_cmd, ["-k", "fake-key"]) + + assert result.exit_code == 0, result.output + assert all(d["type"] == "default" for d in json.loads(result.output)["domains"]) + + +def test_configured_organisation_is_looked_up_not_read_from_cache_dir( + runner, monkeypatch, tmp_path +): + """A named organisation still goes through the normal lookup.""" + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + lambda: str(tmp_path), + ) + write_cache( + get_cache_path("widgets"), [_domain("dl.widgets.com", None, org="widgets")] + ) + monkeypatch.setattr( + "cloudsmith_cli.cli.commands.credential_helper.domains.get_custom_domains", + lambda org, **_kwargs: [_domain("dl.acme.com", None, org=org)], + ) + + result = runner.invoke(domains_cmd, ["-k", "fake-key"]) + + hosts = {d["host"] for d in json.loads(result.output)["domains"]} + assert "dl.acme.com" in hosts + assert "dl.widgets.com" not in hosts + + +def test_repository_scoped_domain_is_reported(runner, monkeypatch): + """A consumer must see which repository a domain is bound to.""" + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + monkeypatch.setattr( + "cloudsmith_cli.cli.commands.credential_helper.domains.get_custom_domains", + lambda org, **_kwargs: [ + _domain( + "dl-prod.acme.com", + None, + org="acme", + repository="prod", + repository_only=True, + ) + ], + ) + + result = runner.invoke(domains_cmd, ["-k", "fake-key"]) + + entry = [d for d in json.loads(result.output)["domains"] if d["type"] == "custom"][ + 0 + ] + assert entry["org"] == "acme" + assert entry["repository"] == "prod" + + +def test_builtin_entries_carry_null_org_and_repository(runner): + """Built-in hosts belong to no organisation.""" + result = runner.invoke(domains_cmd, []) + + defaults = [ + d for d in json.loads(result.output)["domains"] if d["type"] == "default" + ] + assert defaults + assert all(d["org"] is None and d["repository"] is None for d in defaults) + + +def test_refresh_flag_is_forwarded(runner, monkeypatch): + """--refresh must reach the lookup so a new domain shows up immediately.""" + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + captured = {} + + def _lookup(org, **kwargs): + captured.update(kwargs) + return [] + + monkeypatch.setattr( + "cloudsmith_cli.cli.commands.credential_helper.domains.get_custom_domains", + _lookup, + ) + + result = runner.invoke(domains_cmd, ["-k", "fake-key", "--refresh"]) + + assert result.exit_code == 0, result.output + assert captured["refresh"] is True diff --git a/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py b/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py index d44289ca..8009dfaa 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py @@ -446,13 +446,13 @@ def test_refresh_flag(tmp_path, monkeypatch, refresh): cache_path = get_cache_path("acme") cached_domain = CustomDomain( - host="docker.acme.com", backend_kind=6, enabled=True, validated=True + host="docker.acme.com", backend_kind=6, enabled=True, validated=True, org="acme" ) write_cache(cache_path, [cached_domain]) os.utime(cache_path, (time.time(), time.time())) fresh_domain = CustomDomain( - host="new.acme.com", backend_kind=6, enabled=True, validated=True + host="new.acme.com", backend_kind=6, enabled=True, validated=True, org="acme" ) api_calls = [] diff --git a/cloudsmith_cli/cli/tests/commands/test_default_domains.py b/cloudsmith_cli/cli/tests/commands/test_default_domains.py new file mode 100644 index 00000000..fd92fa12 --- /dev/null +++ b/cloudsmith_cli/cli/tests/commands/test_default_domains.py @@ -0,0 +1,354 @@ +# Copyright 2026 Cloudsmith Ltd +"""Tests for the built-in default domain table and its config override.""" + +import pytest + +from ....cli import config as cli_config +from ....credential_helpers.backends import BackendKind +from ....credential_helpers.default_domains import ( + BUILTIN_DOMAINS, + DefaultDomain, + DomainType, + builtin_host, + builtin_host_for_type, + default_host, + default_host_for_type, + domain_type_for_backend_kind, + format_for_backend_kind, + load_default_domains, + untrusted_config_declares_domains, +) + + +def _by_host(domains): + """Index a list of DefaultDomain records by host.""" + return {domain.host: domain for domain in domains} + + +def test_builtin_table_has_expected_size(): + """The built-in table is a fixed, reviewed list.""" + assert len(BUILTIN_DOMAINS) == 21 + + +@pytest.mark.parametrize( + "host,label,kind", + [ + ("python.cloudsmith.io", "python", BackendKind.PYTHON), + ("docker.cloudsmith.io", "docker", BackendKind.DOCKER), + ("maven.cloudsmith.io", "maven", BackendKind.MAVEN), + ("golang.cloudsmith.io", "go", BackendKind.GO), + ("helm.oci.cloudsmith.io", "helm", BackendKind.HELM), + ("dl.cloudsmith.io", None, None), + ("upload.cloudsmith.io", None, None), + ("nix.cloudsmith.io", "nix", BackendKind.NIX), + ], +) +def test_builtin_entries_map_to_expected_kinds(host, label, kind): + """Each built-in host carries the right label and backend kind.""" + domain = _by_host(BUILTIN_DOMAINS)[host] + + assert domain.format_label == label + assert domain.backend_kind == kind + + +def test_load_returns_builtins_when_no_config(tmp_path): + """With no config file present, the built-in table is used.""" + assert load_default_domains(config_path=tmp_path / "absent.ini") == list( + BUILTIN_DOMAINS + ) + + +def test_config_section_replaces_builtins_wholesale(tmp_path): + """A [domains] section replaces the built-in table entirely.""" + config = tmp_path / "config.ini" + config.write_text( + "[domains]\n" + "packages.internal.example.com = python\n" + "cdn.internal.example.com =\n", + encoding="utf-8", + ) + + domains = load_default_domains(config_path=config) + + assert len(domains) == 2 + indexed = _by_host(domains) + assert indexed["packages.internal.example.com"].backend_kind == (BackendKind.PYTHON) + assert indexed["packages.internal.example.com"].format_label == "python" + # An empty value means "no single format". + assert indexed["cdn.internal.example.com"].backend_kind is None + assert indexed["cdn.internal.example.com"].format_label is None + assert indexed["cdn.internal.example.com"].domain_type is DomainType.DOWNLOAD + assert indexed["packages.internal.example.com"].domain_type is DomainType.NATIVE_API + + +def test_config_label_without_backend_kind_is_formatless(tmp_path): + """A label with no matching BackendKind resolves to no format at all. + + A host without a backend kind serves no single format, so its label + cannot name one - the entry degrades to a formatless download host. + """ + config = tmp_path / "config.ini" + config.write_text( + "[domains]\nwidget.internal.example.com = widget\n", encoding="utf-8" + ) + + domain = _by_host(load_default_domains(config_path=config))[ + "widget.internal.example.com" + ] + + assert domain.format_label is None + assert domain.backend_kind is None + assert domain.domain_type is DomainType.DOWNLOAD + + +def test_config_without_domains_section_falls_back_to_builtins(tmp_path): + """A config file that has no [domains] section changes nothing.""" + config = tmp_path / "config.ini" + config.write_text( + "[default]\napi_host = https://api.cloudsmith.io\n", encoding="utf-8" + ) + + assert load_default_domains(config_path=config) == list(BUILTIN_DOMAINS) + + +def test_unreadable_config_falls_back_to_builtins(tmp_path): + """A malformed config must not break the command.""" + config = tmp_path / "config.ini" + config.write_text("this is not valid ini [[[", encoding="utf-8") + + assert load_default_domains(config_path=config) == list(BUILTIN_DOMAINS) + + +def test_trusted_lookup_skips_a_config_without_a_domains_section(tmp_path, monkeypatch): + """The table comes from the first trusted config that declares one. + + Existence is not enough: an app-dir config.ini holding only [default] api + settings would otherwise mask the deployment's [domains] override and + silently restore the public *.cloudsmith.io table. + """ + without_domains = tmp_path / "app-dir" + with_domains = tmp_path / "home-dir" + without_domains.mkdir() + with_domains.mkdir() + (without_domains / "config.ini").write_text( + "[default]\napi_host = https://api.internal.example.com\n", encoding="utf-8" + ) + (with_domains / "config.ini").write_text( + "[domains]\ncdn.internal.example.com =\n", encoding="utf-8" + ) + monkeypatch.setattr(cli_config.ConfigReader, "config_files", ["config.ini"]) + monkeypatch.setattr( + cli_config.ConfigReader, + "config_searchpath", + [str(without_domains), str(with_domains)], + ) + + assert load_default_domains() == [ + DefaultDomain("cdn.internal.example.com", None, DomainType.DOWNLOAD) + ] + + +def test_default_hosts_honour_a_config_override(tmp_path, monkeypatch): + """default_host/_for_type resolve against the override, not the builtins.""" + (tmp_path / "config.ini").write_text( + "[domains]\ncdn.internal.example.com =\nmvn.internal.example.com = maven\n", + encoding="utf-8", + ) + monkeypatch.setattr(cli_config.ConfigReader, "config_files", ["config.ini"]) + monkeypatch.setattr(cli_config.ConfigReader, "config_searchpath", [str(tmp_path)]) + + assert default_host_for_type(DomainType.DOWNLOAD) == "cdn.internal.example.com" + assert default_host(BackendKind.MAVEN) == "mvn.internal.example.com" + + +def test_default_host_falls_back_when_the_override_omits_it(tmp_path, monkeypatch): + """An override that declares no host for a kind keeps the built-in one.""" + (tmp_path / "config.ini").write_text( + "[domains]\ncdn.internal.example.com =\n", encoding="utf-8" + ) + monkeypatch.setattr(cli_config.ConfigReader, "config_files", ["config.ini"]) + monkeypatch.setattr(cli_config.ConfigReader, "config_searchpath", [str(tmp_path)]) + + assert default_host(BackendKind.MAVEN) == "maven.cloudsmith.io" + + +def test_untrusted_cwd_config_is_not_honoured(tmp_path, monkeypatch): + """A [domains] section in a cwd config.ini is ignored, and detectable. + + config.ini is searched in the working directory first, so a repository can + ship one. Honouring it here would let a malicious repo declare its own host + a Cloudsmith host and harvest a token. + """ + monkeypatch.chdir(tmp_path) + # Pin the trusted lookup to "nothing found" so the developer's real + # config.ini cannot influence what this test observes. + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.default_domains._trusted_domains", + lambda: None, + ) + (tmp_path / "config.ini").write_text( + "[domains]\nevil.example.com = python\n", encoding="utf-8" + ) + + domains = load_default_domains() + + assert all(domain.host != "evil.example.com" for domain in domains) + assert untrusted_config_declares_domains() is True + + +def test_untrusted_config_predicate_false_without_section(tmp_path, monkeypatch): + """No [domains] section in the cwd config means nothing to warn about.""" + monkeypatch.chdir(tmp_path) + (tmp_path / "config.ini").write_text("[default]\n", encoding="utf-8") + + assert untrusted_config_declares_domains() is False + + +def test_absolute_config_file_is_not_scanned_as_untrusted(tmp_path, monkeypatch): + """An explicit --config-file is trusted, so it must not trip the warning. + + load_config prepends an explicit path to ConfigReader.config_files, and + joining an absolute path onto a relative search directory yields the + absolute path - so an unfiltered scan would read a trusted file and warn + that it was ignored. + """ + monkeypatch.chdir(tmp_path) + trusted = tmp_path / "explicit.ini" + trusted.write_text("[domains]\nindex.example.com = python\n", encoding="utf-8") + monkeypatch.setattr( + cli_config.ConfigReader, "config_files", [str(trusted), "config.ini"] + ) + monkeypatch.setattr(cli_config.ConfigReader, "config_searchpath", ["."]) + + assert untrusted_config_declares_domains() is False + + +def test_explicit_config_file_is_honoured_as_trusted(tmp_path, monkeypatch): + """An explicit --config-file's [domains] table is honoured, like -C . + + load_config prepends an explicit --config-file naming a file to + ConfigReader.config_files as an absolute path. _config_candidates skips + absolute filenames (to protect the untrusted scan), which must not also + make the no-argument load_default_domains() path drop it - CHANGELOG + promises an explicit --config-file is honoured, and -C already + works. + """ + monkeypatch.chdir(tmp_path) + explicit = tmp_path / "on-prem.ini" + explicit.write_text( + "[domains]\ncdn.internal.example.com =\nmvn.internal.example.com = maven\n", + encoding="utf-8", + ) + monkeypatch.setattr( + cli_config.ConfigReader, "config_files", [str(explicit), "config.ini"] + ) + monkeypatch.setattr(cli_config.ConfigReader, "config_searchpath", ["."]) + + assert default_host_for_type(DomainType.DOWNLOAD) == "cdn.internal.example.com" + assert default_host(BackendKind.MAVEN) == "mvn.internal.example.com" + assert untrusted_config_declares_domains() is False + + +@pytest.mark.parametrize( + "host,domain_type", + [ + ("dl.cloudsmith.io", DomainType.DOWNLOAD), + ("upload.cloudsmith.io", DomainType.UPLOAD), + ("maven.cloudsmith.io", DomainType.NATIVE_API), + ], +) +def test_builtin_entries_carry_their_domain_type(host, domain_type): + """The two format-less hosts are download/upload; the rest are native API.""" + assert _by_host(BUILTIN_DOMAINS)[host].domain_type is domain_type + + +@pytest.mark.parametrize("domain_type", [DomainType.DOWNLOAD, DomainType.UPLOAD]) +def test_backend_kind_requires_native_api_domain_type(domain_type): + """A backend kind is only possible on a NATIVE_API host. + + A host with a backend kind speaks that format's native protocol, so a + download or upload row carrying one is a contradiction and must not be + constructible. + """ + with pytest.raises(ValueError): + DefaultDomain( + host="maven.example.com", + backend_kind=BackendKind.MAVEN, + domain_type=domain_type, + ) + + +@pytest.mark.parametrize( + "backend_kind,expected", + [ + (BackendKind.PYTHON, "python"), + (BackendKind.MAVEN, "maven"), + (BackendKind.GO, "go"), + (None, None), + (9999, "unknown"), + ], +) +def test_format_for_backend_kind(backend_kind, expected): + """A kind resolves to its lowercased format name, with safe fallbacks.""" + assert format_for_backend_kind(backend_kind) == expected + + +def test_builtin_domains_with_a_backend_kind_are_native_api(): + """A host that serves one format is a native-API host, and vice versa. + + Guards the table against a new row whose declared type contradicts its + backend kind - the default is NATIVE_API, so a format-less host added + without an explicit type would otherwise be mistyped silently. + """ + for domain in BUILTIN_DOMAINS: + if domain.backend_kind is None: + assert domain.domain_type is not DomainType.NATIVE_API, domain.host + else: + assert domain.domain_type is DomainType.NATIVE_API, domain.host + + +def test_builtin_host_for_type_resolves_download_and_upload(): + """The download/upload hosts are looked up by type, not by constant.""" + assert builtin_host_for_type(DomainType.DOWNLOAD) == "dl.cloudsmith.io" + assert builtin_host_for_type(DomainType.UPLOAD) == "upload.cloudsmith.io" + + +def test_builtin_host_for_type_rejects_ambiguous_native_api(): + """NATIVE_API covers many hosts, so there is no single one to return.""" + with pytest.raises(ValueError): + builtin_host_for_type(DomainType.NATIVE_API) + + +@pytest.mark.parametrize( + "backend_kind,expected", + [ + (BackendKind.MAVEN, DomainType.NATIVE_API), + # DEB is 0 and therefore falsy: a truthiness check here would + # misclassify it as a download host. + (BackendKind.DEB, DomainType.NATIVE_API), + (None, DomainType.DOWNLOAD), + ], +) +def test_domain_type_for_backend_kind(backend_kind, expected): + """A kind implies native API; its absence implies a download host.""" + assert domain_type_for_backend_kind(backend_kind) is expected + + +def test_builtin_host_resolves_backend_kind(): + """builtin_host maps a backend kind to its built-in service host.""" + assert builtin_host(BackendKind.MAVEN) == "maven.cloudsmith.io" + assert builtin_host(BackendKind.PYTHON) == "python.cloudsmith.io" + + +def test_builtin_host_rejects_kind_without_dedicated_host(): + """Formats served only via the CDN have no dedicated built-in host.""" + with pytest.raises(ValueError): + builtin_host(BackendKind.DEB) + + +def test_default_domain_is_frozen(): + """Records are immutable so callers cannot mutate the shared table.""" + domain = DefaultDomain(host="a.cloudsmith.io", backend_kind=BackendKind.PYTHON) + + with pytest.raises(Exception): + domain.host = "b.cloudsmith.io" diff --git a/cloudsmith_cli/credential_helpers/backends.py b/cloudsmith_cli/credential_helpers/backends.py index d6c32b25..174f25e9 100644 --- a/cloudsmith_cli/credential_helpers/backends.py +++ b/cloudsmith_cli/credential_helpers/backends.py @@ -38,4 +38,5 @@ class BackendKind(IntEnum): GENERIC = 28 VSX = 29 MCP = 30 + NIX = 31 DEFAULT = 99 diff --git a/cloudsmith_cli/credential_helpers/custom_domains.py b/cloudsmith_cli/credential_helpers/custom_domains.py index 5c92cf03..92d6a2ee 100644 --- a/cloudsmith_cli/credential_helpers/custom_domains.py +++ b/cloudsmith_cli/credential_helpers/custom_domains.py @@ -18,11 +18,17 @@ from ..core.api.orgs import list_custom_domains from ..core.cache_utils import atomic_write_json from ..core.credentials.models import CredentialResult +from .default_domains import DomainScope logger = logging.getLogger(__name__) -# Cache custom domains for 1 hour -CACHE_TTL_SECONDS = 3600 +# Custom domains change rarely, so the cache is long-lived; `--refresh` +# bypasses it when a domain has just been added or validated. +CACHE_TTL_SECONDS = 7 * 24 * 3600 + +# Bump when the cached record shape changes: an older document is a miss, not +# a record with the new fields defaulted. +CACHE_FORMAT_VERSION = 2 @dataclass(frozen=True) @@ -33,6 +39,21 @@ class CustomDomain: backend_kind: int | None enabled: bool validated: bool + org: str + repository: str | None = None + repository_only: bool = False + + @property + def scope(self) -> DomainScope: + """What this domain is bound to. + + ``repository_only`` without a repository is a server-side + contradiction; treating it as repository-scoped would build a URL with + no identifying path segment at all, so it degrades to organisation. + """ + if self.repository_only and self.repository: + return DomainScope.REPOSITORY + return DomainScope.ORGANIZATION def get_cache_dir() -> Path: @@ -80,6 +101,32 @@ def is_cache_valid(cache_path: Path) -> bool: return False +def _repository_slug(raw) -> str | None: + """Return the repository slug from an API or cache payload. + + The API nests the repository as ``{"name": ..., "slug": ...}``; the cache + stores the slug flat. One reader serves both. + """ + if isinstance(raw, dict): + return raw.get("slug") or None + if isinstance(raw, str): + return raw or None + return None + + +def _record_from_payload(raw: dict, org: str) -> CustomDomain: + """Build a CustomDomain from an API or cache payload for *org*.""" + return CustomDomain( + host=raw["host"], + backend_kind=raw.get("backend_kind"), + enabled=bool(raw.get("enabled", False)), + validated=bool(raw.get("validated", False)), + org=org, + repository=_repository_slug(raw.get("repository")), + repository_only=bool(raw.get("repository_only", False)), + ) + + def read_cache(cache_path: Path) -> list[CustomDomain] | None: """ Read custom domains from cache file. @@ -96,6 +143,17 @@ def read_cache(cache_path: Path) -> list[CustomDomain] | None: try: with open(cache_path, encoding="utf-8") as f: data = json.load(f) + if ( + isinstance(data, dict) + and data.get("format_version") != CACHE_FORMAT_VERSION + ): + logger.debug( + "Cache %s has format version %s, expected %s - treating as miss", + cache_path, + data.get("format_version"), + CACHE_FORMAT_VERSION, + ) + return None if isinstance(data, dict) and "domains" in data: domains = data["domains"] if isinstance(domains, list): @@ -110,12 +168,7 @@ def read_cache(cache_path: Path) -> list[CustomDomain] | None: return None records = [ - CustomDomain( - host=d["host"], - backend_kind=d.get("backend_kind"), - enabled=bool(d.get("enabled", False)), - validated=bool(d.get("validated", False)), - ) + _record_from_payload(d, d.get("org", "")) for d in domains if isinstance(d, dict) and d.get("host") ] @@ -129,15 +182,35 @@ def read_cache(cache_path: Path) -> list[CustomDomain] | None: return None +def read_all_cached_domains() -> list[CustomDomain]: + """Return every custom domain in a currently-valid cache entry. + + Lets a run with no configured organisation list what earlier runs already + fetched, at no API cost. Each file goes through :func:`read_cache`, so the + TTL and format-version gates apply and a stale or unreadable entry is + skipped rather than reported. + """ + records: list[CustomDomain] = [] + for cache_path in sorted(get_cache_dir().glob("*.json")): + cached = read_cache(cache_path) + if cached: + records.extend(cached) + return records + + def write_cache(cache_path: Path, domains: list[CustomDomain]) -> None: """Write custom domains to cache file.""" data = { + "format_version": CACHE_FORMAT_VERSION, "domains": [ { "host": d.host, "backend_kind": d.backend_kind, "enabled": d.enabled, "validated": d.validated, + "org": d.org, + "repository": d.repository, + "repository_only": d.repository_only, } for d in domains ], @@ -156,11 +229,12 @@ def get_custom_domains( # pylint: disable=too-many-return-statements credential: CredentialResult | None = None, api_host: str | None = None, refresh: bool = False, + strict: bool = False, ) -> list[CustomDomain]: """ Fetch custom domains for a Cloudsmith organization. - Results are cached on the filesystem for 1 hour to avoid excessive API calls. + Results are cached on the filesystem for 7 days to avoid excessive API calls. Args: org: Organization slug @@ -170,17 +244,28 @@ def get_custom_domains( # pylint: disable=too-many-return-statements configuration default when not provided. refresh: When ``True``, skip the cache read and always fetch from the API. The fresh result is still written to the cache. + strict: When ``True``, a failed lookup re-raises its ``ApiException`` + instead of degrading to an empty list. For callers that present + results to a user (``credential-helper domains``), which must not + render a typo'd org, a missing permission or an unreachable API as + "no custom domains". A 402 still returns ``[]`` even in strict + mode — the feature being disabled genuinely means there are none, + and that answer is cached like any other result. Everything else + that could later succeed (auth failures, a missing org, a + transient error) is never cached, so a strict lookup can be + served from the cache without risking a cached failure being + reported as success. Returns: List of CustomDomain records. - Empty list if API call fails or org has no custom domains. + Empty list if the org has no custom domains, or (unless ``strict``) + if the API call fails. Note: - Only ``ApiException`` is handled here (per-status). Network-layer errors - (DNS/timeout/SSL/urllib3) are intentionally NOT caught — they propagate to - the caller. The credential-helper protocol boundary (the click command) and - the installer are responsible for catching broadly and refusing gracefully, - so the library stays free of bare ``except Exception`` (reviewer feedback). + The API layer wraps transport failures (DNS/timeout/SSL) into + ``ApiException`` too, so every failure mode lands in the handler + below. The default is best-effort (helpers degrade rather than break + the wrapped tool); pass ``strict=True`` to fail loudly instead. """ cache_path = get_cache_path(org) cached = None if refresh else read_cache(cache_path) @@ -195,6 +280,8 @@ def get_custom_domains( # pylint: disable=too-many-return-statements try: raw_domains = list_custom_domains(org) except ApiException as exc: + if strict and exc.status != 402: + raise if exc.status in (401, 403): # Don't cache auth failures - might work later once authenticated. logger.debug( @@ -204,13 +291,14 @@ def get_custom_domains( # pylint: disable=too-many-return-statements return [] if exc.status == 404: - # Cache empty result to avoid repeated lookups for a missing org. logger.debug("Organization %s not found or has no custom domains", org) - write_cache(cache_path, []) return [] if exc.status == 402: - # Custom domains product feature not enabled - treat as none. + # Unlike 404 (org not found - probably a typo, must fail loudly + # every time) a 402 is a stable, correct answer: the product + # feature isn't enabled for this org. Caching it avoids hitting + # the API on every invocation forever. logger.debug("Custom domains not enabled for %s", org) write_cache(cache_path, []) return [] @@ -218,16 +306,7 @@ def get_custom_domains( # pylint: disable=too-many-return-statements logger.debug("Failed to fetch custom domains for %s: HTTP %s", org, exc.status) return [] - records = [ - CustomDomain( - host=d["host"], - backend_kind=d.get("backend_kind"), - enabled=bool(d.get("enabled", False)), - validated=bool(d.get("validated", False)), - ) - for d in raw_domains - if d.get("host") - ] + records = [_record_from_payload(d, org) for d in raw_domains if d.get("host")] logger.debug("Fetched %d custom domains for %s", len(records), org) write_cache(cache_path, records) diff --git a/cloudsmith_cli/credential_helpers/default_domains.py b/cloudsmith_cli/credential_helpers/default_domains.py new file mode 100644 index 00000000..b847da06 --- /dev/null +++ b/cloudsmith_cli/credential_helpers/default_domains.py @@ -0,0 +1,363 @@ +# Copyright 2026 Cloudsmith Ltd +"""Built-in Cloudsmith service hosts, with a trusted-config override. + +The table below lists the standard ``*.cloudsmith.io`` service hosts and the +package format each one serves. A deployment with its own package-serving +domains — dedicated or on-premise — can replace the table via a ``[domains]`` +section in ``config.ini``, mapping hostname to package format. + +The override is honoured only from trusted locations. ``config.ini`` is +searched in the current working directory first (``cli/config.py``), so a +``config.ini`` committed to a repository is attacker-controlled input; this +module therefore never reads the domain table from a directory-relative +config, mirroring the split ``_guard_untrusted_endpoints`` +(``cli/decorators.py``) already applies to ``api_host``/``api_proxy``. An +explicit ``--config-file`` is a user's direct statement of trust regardless of +where it lives, so it is honoured ahead of the search-path locations. +""" + +from __future__ import annotations + +import configparser +import logging +import os +from dataclasses import dataclass +from enum import Enum +from pathlib import Path + +from ..cli.config import ConfigReader +from .backends import BackendKind + +logger = logging.getLogger(__name__) + + +class DomainType(str, Enum): + """What a Cloudsmith host is for. + + A host that serves a single package format speaks that format's native + protocol (``NATIVE_API``); the two format-less hosts are the download CDN + and the generic upload endpoint. + """ + + DOWNLOAD = "download" + UPLOAD = "upload" + NATIVE_API = "native_api" + + +class DomainScope(str, Enum): + """What a custom domain is bound to. + + A custom domain always belongs to one organisation; one that additionally + serves a single repository identifies that repository too, so neither + appears in its URLs. + """ + + ORGANIZATION = "organization" + REPOSITORY = "repository" + + +def domain_scope(value: str | None) -> DomainScope: + """Resolve a persisted scope string, defaulting to organisation. + + An unrecognised value degrades rather than raising: a hand-edited + ``package-managers.ini`` must not break every wrapped run. + """ + try: + return DomainScope(value) + except ValueError: + return DomainScope.ORGANIZATION + + +def format_for_backend_kind(backend_kind: int | None) -> str | None: + """Return the package format a backend kind serves, or None. + + A host with no backend kind serves no single format and resolves to + ``None``. :class:`BackendKind` is a hand-maintained mirror of the + server-side enum, so a format the CLI does not know yet renders as + ``unknown`` rather than breaking its caller. + """ + if backend_kind is None: + return None + + try: + return BackendKind(backend_kind).name.lower() + except ValueError: + return "unknown" + + +def domain_type_for_backend_kind(backend_kind: int | None) -> DomainType: + """Classify a host from its backend kind. + + Compares against ``None`` rather than testing truthiness: ``BackendKind.DEB`` + is ``0``, and a falsy check would classify it as a download host. + """ + if backend_kind is None: + return DomainType.DOWNLOAD + return DomainType.NATIVE_API + + +@dataclass(frozen=True) +class DefaultDomain: + """A built-in or config-declared default Cloudsmith host.""" + + host: str + backend_kind: int | None + domain_type: DomainType = DomainType.NATIVE_API + + def __post_init__(self) -> None: + if ( + self.backend_kind is not None + and self.domain_type is not DomainType.NATIVE_API + ): + raise ValueError( + f"A backend kind is only possible on a NATIVE_API host: " + f"{self.host} is {self.domain_type.value}" + ) + + @property + def format_label(self) -> str | None: + """The package format this host serves, or None for no single format.""" + return format_for_backend_kind(self.backend_kind) + + +BUILTIN_DOMAINS: tuple[DefaultDomain, ...] = ( + DefaultDomain("cargo.cloudsmith.io", BackendKind.CARGO), + DefaultDomain("composer.cloudsmith.io", BackendKind.COMPOSER), + DefaultDomain("conan.cloudsmith.io", BackendKind.CONAN), + DefaultDomain("conda.cloudsmith.io", BackendKind.CONDA), + DefaultDomain("dart.cloudsmith.io", BackendKind.DART), + DefaultDomain("dl.cloudsmith.io", None, DomainType.DOWNLOAD), + DefaultDomain("docker.cloudsmith.io", BackendKind.DOCKER), + DefaultDomain("generic.cloudsmith.io", BackendKind.GENERIC), + DefaultDomain("golang.cloudsmith.io", BackendKind.GO), + DefaultDomain("helm.oci.cloudsmith.io", BackendKind.HELM), + DefaultDomain("hex.cloudsmith.io", BackendKind.HEX), + DefaultDomain("huggingface.cloudsmith.io", BackendKind.HUGGINGFACE), + DefaultDomain("maven.cloudsmith.io", BackendKind.MAVEN), + DefaultDomain("nix.cloudsmith.io", BackendKind.NIX), + DefaultDomain("npm.cloudsmith.io", BackendKind.NPM), + DefaultDomain("nuget.cloudsmith.io", BackendKind.NUGET), + DefaultDomain("python.cloudsmith.io", BackendKind.PYTHON), + DefaultDomain("ruby.cloudsmith.io", BackendKind.RUBY), + DefaultDomain("swift.cloudsmith.io", BackendKind.SWIFT), + DefaultDomain("terraform.cloudsmith.io", BackendKind.TERRAFORM), + DefaultDomain("upload.cloudsmith.io", None, DomainType.UPLOAD), +) + + +def _host_for_backend_kind( + domains: tuple[DefaultDomain, ...] | list[DefaultDomain], backend_kind: int +) -> str | None: + """Return the first host in `domains` serving `backend_kind`, if any.""" + for domain in domains: + if domain.backend_kind == backend_kind: + return domain.host + return None + + +def _host_for_type( + domains: tuple[DefaultDomain, ...] | list[DefaultDomain], domain_type: DomainType +) -> str | None: + """Return the first host in `domains` of `domain_type`, if any.""" + if domain_type is DomainType.NATIVE_API: + raise ValueError( + "NATIVE_API is served by many hosts; resolve it with builtin_host()" + ) + for domain in domains: + if domain.domain_type is domain_type: + return domain.host + return None + + +def builtin_host(backend_kind: int) -> str: + """Return the built-in Cloudsmith service host for `backend_kind`. + + Raises ValueError for formats served only via the CDN, which have no + dedicated built-in host. + """ + host = _host_for_backend_kind(BUILTIN_DOMAINS, backend_kind) + if host is None: + raise ValueError(f"No built-in Cloudsmith host for backend kind {backend_kind}") + return host + + +def builtin_host_for_type(domain_type: DomainType) -> str: + """Return the single built-in Cloudsmith host serving `domain_type`. + + Raises ValueError for NATIVE_API, which many hosts share - resolve those by + backend kind with :func:`builtin_host` instead. + """ + host = _host_for_type(BUILTIN_DOMAINS, domain_type) + if host is None: + raise ValueError(f"No built-in Cloudsmith host of type {domain_type.value}") + return host + + +def default_host(backend_kind: int) -> str: + """Return the host for `backend_kind`, honouring a trusted override. + + A ``[domains]`` section replaces the built-in table wholesale, so a table + that declares no host for this kind falls back to the built-in one. + """ + host = _host_for_backend_kind(load_default_domains(), backend_kind) + return host if host is not None else builtin_host(backend_kind) + + +def default_host_for_type(domain_type: DomainType) -> str: + """Return the host of `domain_type`, honouring a trusted override.""" + host = _host_for_type(load_default_domains(), domain_type) + return host if host is not None else builtin_host_for_type(domain_type) + + +def _resolve_backend_kind(label: str) -> int | None: + """Resolve a config-declared label to a BackendKind member, if any.""" + if not label: + return None + return BackendKind.__members__.get(label.upper()) + + +def _domain_from_config_entry(host: str, label: str) -> DefaultDomain: + """Build a DefaultDomain from one ``[domains]`` ``host = label`` entry. + + A label that does not resolve to a BackendKind is dropped: a host without + a backend kind serves no single format, so it becomes a formatless + download host. + """ + backend_kind = _resolve_backend_kind(label) + return DefaultDomain( + host=host, + backend_kind=backend_kind, + domain_type=domain_type_for_backend_kind(backend_kind), + ) + + +def _config_candidates(*, trusted: bool) -> list[Path]: + """Return candidate config.ini paths from ConfigReader's search locations. + + The locations are click's, via ``ConfigReader``, rather than restated here. + Its search path lists the current directory first, so absolute entries are + the trusted ones and directory-relative entries are not. + + Absolute filenames are skipped: ``load_config`` prepends an explicit + ``--config-file`` to ``config_files``, and joining an absolute path onto a + search directory would yield that path under every location - handing a + trusted file to the untrusted scan. Such an entry is read directly as a + trusted candidate instead, by :func:`_explicit_config_candidate`. + """ + filenames = [ + filename + for filename in ConfigReader.config_files + if not os.path.isabs(filename) + ] + return [ + Path(directory) / filename + for directory in ConfigReader.config_searchpath + if os.path.isabs(directory) is trusted + for filename in filenames + ] + + +def _explicit_config_candidate() -> Path | None: + """Return the explicit ``--config-file`` entry in ``config_files``, if any. + + When ``--config-file`` names a file (rather than a directory), + ``ConfigReader.load_config`` prepends its absolute path to + ``config_files``; a directory is prepended to ``config_searchpath`` + instead, already covered by the trusted search-path scan in + :func:`_config_candidates`. Such an entry is explicit user intent, so it + is read directly as trusted rather than through the search-directory join + that ``_config_candidates`` deliberately excludes absolute filenames from. + """ + for filename in ConfigReader.config_files: + if os.path.isabs(filename): + return Path(filename) + return None + + +def _trusted_domains() -> list[DefaultDomain] | None: + """Return the [domains] table from the first trusted config declaring one. + + An explicit ``--config-file`` is checked first, ahead of the search-path + candidates - it is the most specific statement of trust available. + Existence alone is not enough: a trusted ``config.ini`` holding only + ``[default]`` api settings must not mask a ``[domains]`` section in a + later candidate. + """ + candidates = [] + explicit = _explicit_config_candidate() + if explicit is not None: + candidates.append(explicit) + candidates.extend(_config_candidates(trusted=True)) + + for candidate in candidates: + domains = _domains_from_config(candidate) + if domains is not None: + return domains + return None + + +def _read_domains_section(path: Path) -> configparser.SectionProxy | None: + """Return the ``[domains]`` section at `path`, or None if absent/unreadable. + + ``[domains]`` maps arbitrary hostnames to formats, so its keys cannot be + declared as a click_configfile section schema the way ``[default]`` is; it + is read with configparser directly. + """ + parser = configparser.ConfigParser() + try: + read_files = parser.read(path, encoding="utf-8") + except (OSError, configparser.Error) as exc: + logger.debug("Failed to read config file %s: %s", path, exc) + return None + + if not read_files or not parser.has_section("domains"): + return None + return parser["domains"] + + +def _domains_from_config(path: Path) -> list[DefaultDomain] | None: + """Parse a [domains] section at `path`, or None if absent/unreadable.""" + section = _read_domains_section(path) + if section is None: + return None + + domains = [ + _domain_from_config_entry(host, label) for host, label in section.items() + ] + return sorted(domains, key=lambda domain: domain.host) + + +def load_default_domains(config_path: Path | str | None = None) -> list[DefaultDomain]: + """Return the default domain table, honouring a trusted config override. + + When `config_path` is given, that file is read directly. Otherwise, an + explicit ``--config-file`` naming a file is read first if present, + followed by `config.ini` in ConfigReader's absolute search locations - + never the current working directory. A malformed or unreadable file falls + back to the built-in table. + """ + if config_path is not None: + domains = _domains_from_config(Path(config_path)) + else: + domains = _trusted_domains() + + if domains is None: + return list(BUILTIN_DOMAINS) + + return domains + + +def untrusted_config_declares_domains() -> bool: + """True if a directory-relative config.ini declares a [domains] section. + + Such a section is deliberately ignored: config.ini is searched in the + current working directory first, so a repository can ship one, and this + command decides which hosts may receive a Cloudsmith token. Honouring it + would let a malicious repo harvest a live credential - the same vector + ``_guard_untrusted_endpoints`` closes for api_host. The caller warns so + the omission is visible rather than silent. + """ + return any( + _read_domains_section(candidate) is not None + for candidate in _config_candidates(trusted=False) + )