diff --git a/.oss-scanner/threat_model.md b/.oss-scanner/threat_model.md index fc3bf5d5..538e35bd 100644 --- a/.oss-scanner/threat_model.md +++ b/.oss-scanner/threat_model.md @@ -16,7 +16,7 @@ Untrusted: Trusted: command-line flags, environment variables, `config.ini` and `credentials.ini` that the user owns, and the OS keyring. ## Components that matter most / least -- Credential scoping. A credential must only go to the host that it belongs to. This covers `core/credentials/chain.py`, the providers in `core/credentials/providers/`, the credential helper runtimes, and the custom-domain lookup (`credential_helpers/custom_domains.py`, `credential_helpers/common.py`). Host-name parsing and matching bugs are in scope. +- Credential scoping. A credential must only go to the host that it belongs to. This covers `core/credentials/chain.py`, the providers in `core/credentials/providers/`, the credential helper runtimes, and the custom-domain lookup (`credential_helpers/domains.py`, `credential_helpers/common.py`). Host-name parsing and matching bugs are in scope. - Token storage: `core/keyring.py`, `core/macos_keychain.py`, and the OIDC token cache (`core/credentials/oidc/cache.py`). File permissions and cross-profile leaks are in scope. - The SSO callback server: CSRF, token injection, and open redirects. - File writes from untrusted data: download paths from API filenames, tar extraction in `core/self_update.py`, and edits to package-manager and MCP client config files (`cli/commands/mcp.py`). diff --git a/CHANGELOG.md b/CHANGELOG.md index 89add60d..10a54dc9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ## [Unreleased] +### Changed + +- `cloudsmith domains list` and the credential helpers now get the domains from `GET /v2/workspaces/{workspace}/domains/` on the configured API host. This list includes the Cloudsmith-owned hosts, so the CLI no longer contains a list of Cloudsmith hostnames. The CLI keeps a separate domain cache for each API host. +- `cloudsmith domains list` keeps version 1 of its JSON document. The API does not give `primary` and `created_at`, so these fields are now always `null`. The hosts are in the preference order of the API. `domain_type` can also be `cloudsmith_api`. +- Without a Workspace, `cloudsmith domains list` lists the domains that earlier runs cached. If no domains are cached, it gets the domains of a Workspace that the credential is a member of. Without a credential, the list is empty. +- `cloudsmith credential-helper install` now configures the Cloudsmith-owned hosts that the API lists for the Workspace. It uses its built-in default host, such as `docker.cloudsmith.io`, only when it cannot get the domains from the API. + +### Removed + +- The CLI no longer reads a `[domains]` section from `config.ini`. The configured API host supplies the domains. + ## [1.29.0] - 2026-10-15 ### Added diff --git a/cloudsmith_cli/cli/commands/domains.py b/cloudsmith_cli/cli/commands/domains.py index 9c9fa28b..fc1b9cb0 100644 --- a/cloudsmith_cli/cli/commands/domains.py +++ b/cloudsmith_cli/cli/commands/domains.py @@ -1,34 +1,27 @@ # Copyright 2026 Cloudsmith Ltd -"""List the hosts Cloudsmith can authenticate: built-in and custom domains. +"""List the hosts Cloudsmith can authenticate: Cloudsmith-owned 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. +The configured API host lists the domains that serve a Workspace, so the CLI +holds no table of Cloudsmith hostnames and works against any Cloudsmith. """ from __future__ import annotations import json -import os import click from click.core import ParameterSource from ...core.api.exceptions import ApiException +from ...core.api.orgs import list_orgs from ...core.pagination import PageInfo -from ...credential_helpers.custom_domains import ( - get_custom_domains, - order_by_precedence, - read_all_cached_domains, -) -from ...credential_helpers.default_domains import ( +from ...credential_helpers.domains import ( DomainType, - format_for_backend_kind, - load_default_domains, - untrusted_config_declares_domains, + WorkspaceDomain, + get_workspace_domains, + read_all_cached_domains, ) from .. import command -from ..config import get_default_config_path from ..decorators import ( common_api_auth_options, common_cli_config_options, @@ -55,22 +48,86 @@ def _is_domain_type(domain_type: DomainType, wanted: str | None) -> bool: return wanted is None or domain_type.value == wanted -def _custom_entries(records) -> list[dict]: - """Render custom-domain records as document entries, in the order given.""" - return [ - { - "host": record.host, - "format": format_for_backend_kind(record.backend_kind), - "type": "custom", - "domain_type": record.domain_type.value, - "workspace": record.org, - "org": record.org, - "repository": record.repository, - "primary": record.primary, - "created_at": record.created_at, - } - for record in records - ] +def _entry(record: WorkspaceDomain) -> dict: + """Render a domain record as a document entry.""" + return { + "host": record.host, + "format": record.format_label, + "type": "custom" if record.is_custom else "default", + "domain_type": record.domain_type.value, + "workspace": record.workspace, + "org": record.workspace, + "repository": record.repository, + "primary": None, + "created_at": None, + } + + +def _merge_cached(records: list[WorkspaceDomain]) -> list[WorkspaceDomain]: + """Merge the cached answers for several Workspaces into one list. + + No single API order covers several Workspaces, so custom domains are + grouped by Workspace ahead of the Cloudsmith-owned ones. Each group keeps + the API order. Every cached answer repeats the Cloudsmith-owned domains, + so each one is listed once. + """ + custom = [record for record in records if record.is_custom] + cloudsmith_owned = [record for record in records if not record.is_custom] + return sorted(custom, key=lambda record: record.workspace or "") + list( + dict.fromkeys(cloudsmith_owned) + ) + + +def _member_workspace(opts) -> str | None: + """Return a Workspace that the credential is a member of, or None.""" + if opts.credential is None: + return None + workspaces = list_orgs(page=1, page_size=1) + return workspaces[0]["slug"] if workspaces else None + + +def _get_domains(opts, refresh: bool) -> list[WorkspaceDomain]: + """Get the domains of the configured Workspace, or of a cached or member one. + + Every Workspace answer holds the Cloudsmith-owned domains. Without a + configured Workspace, the cached answers supply them, or else the answer + for any Workspace that the credential is a member of. + """ + workspace = opts.org + if not workspace: + cached = read_all_cached_domains(opts.api_host) + if cached: + if refresh: + click.secho( + "Warning: --refresh needs a Workspace to fetch from, so the " + "cached domains below are unchanged. Set --workspace, " + "CLOUDSMITH_WORKSPACE, or workspace in config.ini.", + fg="yellow", + err=True, + ) + return _merge_cached(cached) + try: + workspace = _member_workspace(opts) + except ApiException as exc: + raise click.ClickException( + f"Failed to find a Workspace to fetch domains for: {exc}" + ) from exc + if workspace is None: + return [] + + try: + return get_workspace_domains( + workspace, + credential=opts.credential, + api_host=opts.api_host, + refresh=refresh, + strict=True, + configure_api=False, + ) + except ApiException as exc: + raise click.ClickException( + f"Failed to fetch domains for {workspace!r}: {exc}" + ) from exc @main.group(name="domains", cls=command.AliasGroup, aliases=["domain"]) @@ -87,7 +144,7 @@ def domains_(): "--refresh", is_flag=True, default=False, - help="Bypass the custom-domain cache and fetch fresh data from the API.", + help="Bypass the domain cache and fetch fresh data from the API.", ) @click.option( "--format", @@ -101,7 +158,8 @@ def domains_(): "--domain-type", "domain_type", default=None, - help="Only list hosts with this purpose: download, upload, or native_api. " + help="Only list hosts with this purpose: download, upload, cloudsmith_api, " + "or native_api. " "Every host has exactly one, so this is an exact match.", ) @click.option( @@ -130,30 +188,30 @@ def list_domains( # pylint: disable=too-many-arguments ) -> None: """Emit the Cloudsmith hosts and their package formats as JSON. - Machine-readable output for programmatic consumers. It lists the built-in - ``*.cloudsmith.io`` service hosts alongside custom domains, with ``type`` - distinguishing ``default`` from ``custom`` and ``domain_type`` saying what - each host is for: ``download``, ``upload``, or ``native_api`` for a host - speaking one package format's own protocol. A Workspace's own custom - domains are listed ahead of the built-in hosts, which serve as the - fallback. Only usable hosts are listed: a custom domain that is disabled or - not yet validated serves nothing, so it is left out rather than offered as - somewhere to publish to. Check it in the Cloudsmith UI if one you expect is - missing here. - - Built-in hosts are always listed and need no Workspace or authentication. - A Workspace from ``--workspace``, CLOUDSMITH_WORKSPACE, or ``workspace`` in - ``config.ini`` adds its custom domains, and a failed lookup exits non-zero - rather than rendering as "no domains". With no Workspace the command lists - whatever earlier runs cached and makes no API call; ``--refresh`` bypasses - that cache for a configured Workspace. - - Where two custom domains could serve the same request Cloudsmith picks the - one bound to the repository in hand, then ``primary`` over secondary, then - the oldest — so ``created_at`` is reported to make that order reproducible. - The custom domains are always listed in that order, so the first is the one - Cloudsmith would bind; ``--repo`` is what lets the repository-bound hosts - take part in it. + Machine-readable output for programmatic consumers. The configured API host + lists the domains that serve a Workspace, so the hosts match whichever + Cloudsmith the CLI talks to. ``type`` distinguishes Cloudsmith-owned + (``default``) from ``custom`` domains, and ``domain_type`` says what each + host is for: ``download``, ``upload``, ``cloudsmith_api``, or + ``native_api`` for a host speaking one package format's own protocol. Only + usable hosts are listed: the API leaves out a custom domain that is + disabled or not yet validated. Check it in the Cloudsmith UI if one you + expect is missing here. + + The lookup uses the Workspace from ``--workspace``, CLOUDSMITH_WORKSPACE, + or ``workspace`` in ``config.ini``, and needs authentication as a member + of it. A failed lookup exits non-zero rather than rendering as "no + domains". With no Workspace the command lists whatever earlier runs cached + for the configured API host. If nothing is cached, it fetches the domains + of a Workspace that the credential is a member of, which include all the + Cloudsmith-owned hosts. ``--refresh`` bypasses the cache for a configured + Workspace. + + Hosts are listed in the API's preference order: domains restricted to a + repository, then the Workspace's own, then the Cloudsmith-owned ones. + Within each group the primary domain comes first, then the oldest, so the + first host usable for a request is the one Cloudsmith would choose. + ``--repo`` drops the domains restricted to other repositories. Results support the same --page/--page-size (-p/-l) options as other `list` commands, selecting a page of the combined, filtered list. Unlike @@ -165,8 +223,12 @@ def list_domains( # pylint: disable=too-many-arguments \b Output (stdout): JSON: {"version": 1, "domains": [{"host": ..., "format": ..., - "type": ..., "domain_type": ..., "workspace": ..., "repository": ..., - "primary": ..., "created_at": ...}], "meta": {"pagination": {...}}} + "type": ..., "domain_type": ..., "workspace": ..., "org": ..., + "repository": ..., "primary": null, "created_at": null}], + "meta": {"pagination": {...}}} + + "org" repeats "workspace". The API does not give "primary" and + "created_at", so they are always null. Use the list order instead. "meta" is only present when the result is paginated (i.e. not --page-all), and mirrors the pagination metadata other `list` @@ -175,11 +237,11 @@ def list_domains( # pylint: disable=too-many-arguments Examples: \b - # List built-in hosts plus any custom domains already cached + # List the Cloudsmith-owned hosts and any cached custom domains $ cloudsmith domains list \b - # List built-in hosts plus a Workspace's custom domains + # List the domains that serve a Workspace $ cloudsmith domains list --workspace my-workspace \b @@ -208,79 +270,18 @@ def list_domains( # pylint: disable=too-many-arguments # cache, so paging by default would hide entries for no benefit. page_all = True - 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") - - 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, - ) - wanted_format = _label(format_) wanted_type = _label(domain_type) repository = _label(repo) - default_entries = [ - { - "host": domain.host, - "format": domain.format_label, - "type": "default", - "domain_type": domain.domain_type.value, - "workspace": None, - "org": None, - "repository": None, - "primary": True, - "created_at": None, - } - for domain in load_default_domains(config_path=explicit_config) - if _serves_format(domain.format_label, wanted_format) - and _is_domain_type(domain.domain_type, wanted_type) - ] - - org = opts.org - if org: - try: - records = get_custom_domains( - org, - credential=opts.credential, - api_host=opts.api_host, - refresh=refresh, - strict=True, - configure_api=False, - ) - except ApiException as exc: - raise click.ClickException( - f"Failed to fetch custom domains for {org!r}: {exc}" - ) from exc - else: - if refresh: - click.secho( - "Warning: --refresh needs a Workspace to fetch from, so the " - "cached custom domains below are unchanged. Set --workspace, " - "CLOUDSMITH_WORKSPACE, or workspace in config.ini.", - fg="yellow", - err=True, - ) - records = read_all_cached_domains() - records = [ record - for record in records - if _serves_format(format_for_backend_kind(record.backend_kind), wanted_format) + for record in _get_domains(opts, refresh) + if _serves_format(record.format_label, wanted_format) and _is_domain_type(record.domain_type, wanted_type) - and record.is_active and (repository is None or record.serves_repository(repository)) ] - records = sorted( - order_by_precedence(records, repository), key=lambda record: record.org - ) - - data = _custom_entries(records) + default_entries + data = [_entry(record) for record in records] document = {"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 9e073bb6..9308d9ce 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper.py @@ -18,17 +18,15 @@ is_cloudsmith_domain, is_standard_cloudsmith_domain, ) -from ....credential_helpers.custom_domains import ( +from ....credential_helpers.domains import ( CACHE_FORMAT_VERSION, - CustomDomain, get_cache_path, - get_custom_domains, get_format_domains, + get_workspace_domains, read_all_cached_domains, read_cache, write_cache, ) -from ....credential_helpers.default_domains import DomainType from ....credential_helpers.docker.runtime import ( _REFUSAL_MESSAGE, execute, @@ -40,6 +38,16 @@ API_HOST = "https://api.cloudsmith.io" +def _server_domain(name, domain_type, scope="WORKSPACE"): + """Build one entry of the v2 Workspace domains listing.""" + return { + "domain_name": name, + "domain_type": domain_type, + "domain_scope": scope, + "repository": None, + } + + # --------------------------------------------------------------------------- # 1. runtime.execute — protocol matrix # --------------------------------------------------------------------------- @@ -219,14 +227,14 @@ def test_backend_kind_values(): # --------------------------------------------------------------------------- -# 6. get_custom_domains — HTTP status matrix +# 6. get_workspace_domains — HTTP status matrix # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def _cache_dir(tmp_path, monkeypatch): monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) monkeypatch.setattr( @@ -258,42 +266,34 @@ def _no_retry_sleep(monkeypatch): ], ) @httpretty.activate(allow_net_connect=False) -def test_get_custom_domains_status_matrix( +def test_get_workspace_domains_status_matrix( tmp_path, monkeypatch, status, expect_domains, expect_cached ): - """get_custom_domains() caches or not based on HTTP status.""" + """get_workspace_domains() caches or not based on HTTP status.""" # redirect per-test (autouse fixture already set module-level path) monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) if status == 200: body = json.dumps( - [ - { - "host": "docker.acme.com", - "backend_kind": 6, - "domain_type": 1, - "enabled": True, - "validated": True, - } - ] + {"results": [_server_domain("docker.acme.com", "NATIVE_DOCKER")]} ) else: body = json.dumps({"detail": "error"}) httpretty.register_uri( httpretty.GET, - f"{API_HOST}/orgs/acme/custom-domains/", + f"{API_HOST}/v2/workspaces/acme/domains/", body=body, status=status, content_type="application/json", ) credential = CredentialResult(api_key="k_abc", source_name="test") - result = get_custom_domains("acme", credential=credential, api_host=API_HOST) - cache = read_cache(get_cache_path("acme")) + result = get_workspace_domains("acme", credential=credential, api_host=API_HOST) + cache = read_cache(get_cache_path("acme", API_HOST)) if expect_domains: assert len(result) == 1 @@ -322,7 +322,7 @@ def test_get_custom_domains_status_matrix( ], ) @httpretty.activate(allow_net_connect=False) -def test_get_custom_domains_strict_raises_on_failure( +def test_get_workspace_domains_strict_raises_on_failure( tmp_path, monkeypatch, status, expect_raise ): """strict=True re-raises failures instead of degrading to []. @@ -331,13 +331,17 @@ def test_get_custom_domains_strict_raises_on_failure( 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", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) - body = json.dumps([]) if status == 200 else json.dumps({"detail": "error"}) + body = ( + json.dumps({"results": []}) + if status == 200 + else json.dumps({"detail": "error"}) + ) httpretty.register_uri( httpretty.GET, - f"{API_HOST}/orgs/acme/custom-domains/", + f"{API_HOST}/v2/workspaces/acme/domains/", body=body, status=status, content_type="application/json", @@ -346,47 +350,47 @@ def test_get_custom_domains_strict_raises_on_failure( credential = CredentialResult(api_key="k_abc", source_name="test") if expect_raise: with pytest.raises(ApiException): - get_custom_domains( + get_workspace_domains( "acme", credential=credential, api_host=API_HOST, strict=True ) else: - result = get_custom_domains( + result = get_workspace_domains( "acme", credential=credential, api_host=API_HOST, strict=True ) assert result == [] @httpretty.activate(allow_net_connect=False) -def test_get_custom_domains_caches_an_empty_listing(tmp_path, monkeypatch): - """An org with no custom domains answers 200 with an empty array. +def test_get_workspace_domains_caches_an_empty_listing(tmp_path, monkeypatch): + """A Workspace with no domains answers 200 with an empty array. That is a real answer rather than a failure, so it is cached like any other: a second lookup within the TTL must be served from the cache instead of asking again. """ monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) httpretty.register_uri( httpretty.GET, - f"{API_HOST}/orgs/acme/custom-domains/", - body=json.dumps([]), + f"{API_HOST}/v2/workspaces/acme/domains/", + body=json.dumps({"results": []}), status=200, 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 get_workspace_domains("acme", credential=credential, api_host=API_HOST) == [] + assert read_cache(get_cache_path("acme", API_HOST)) == [] assert len(httpretty.latest_requests()) == 1 - assert get_custom_domains("acme", credential=credential, api_host=API_HOST) == [] + assert get_workspace_domains("acme", credential=credential, api_host=API_HOST) == [] assert len(httpretty.latest_requests()) == 1 @httpretty.activate(allow_net_connect=False) -def test_get_custom_domains_bearer_credential_sends_authorization_header( +def test_get_workspace_domains_bearer_credential_sends_authorization_header( tmp_path, monkeypatch ): """A bearer credential authenticates the lookup with its own header scheme. @@ -395,13 +399,13 @@ def test_get_custom_domains_bearer_credential_sends_authorization_header( SSO access token goes out as ``Authorization: Bearer``, never ``X-Api-Key``. """ monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) httpretty.register_uri( httpretty.GET, - f"{API_HOST}/orgs/acme/custom-domains/", - body=json.dumps([]), + f"{API_HOST}/v2/workspaces/acme/domains/", + body=json.dumps({"results": []}), status=200, content_type="application/json", ) @@ -413,7 +417,7 @@ def test_get_custom_domains_bearer_credential_sends_authorization_header( credential = CredentialResult( api_key="jwt_token", source_name="test", auth_type="bearer" ) - get_custom_domains("acme", credential=credential, api_host=API_HOST) + get_workspace_domains("acme", credential=credential, api_host=API_HOST) headers = httpretty.last_request().headers assert headers.get("Authorization") == "Bearer jwt_token" @@ -421,7 +425,7 @@ def test_get_custom_domains_bearer_credential_sends_authorization_header( # --------------------------------------------------------------------------- -# 7. get_custom_domains — cache edge cases +# 7. get_workspace_domains — cache edge cases # --------------------------------------------------------------------------- @@ -434,10 +438,10 @@ def test_get_custom_domains_bearer_credential_sends_authorization_header( ("empty_list", []), ], ) -def test_get_custom_domains_cache_edge(tmp_path, monkeypatch, scenario, expected): +def test_get_workspace_domains_cache_edge(tmp_path, monkeypatch, scenario, expected): """Cache format edge cases: legacy string-list is a miss; empty list is a hit.""" monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) @@ -462,7 +466,7 @@ def test_get_custom_domains_cache_edge(tmp_path, monkeypatch, scenario, expected def test_cache_that_is_not_utf8_reads_as_a_miss(tmp_path, monkeypatch): """A cache file of undecodable bytes is a miss, not a crash.""" monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) @@ -478,7 +482,7 @@ def test_unwritable_config_dir_reads_as_no_cached_domains(tmp_path, monkeypatch) read_only = tmp_path / "read-only" read_only.mkdir(mode=0o500) monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(read_only), ) @@ -492,60 +496,26 @@ def test_unwritable_config_dir_reads_as_no_cached_domains(tmp_path, monkeypatch) @httpretty.activate(allow_net_connect=False) def test_get_format_domains_filters_correctly(tmp_path, monkeypatch): - """get_format_domains returns only enabled+validated Docker hosts. + """get_format_domains returns only the Docker hosts, in the server's order. - Mixed body: docker enabled+validated (included), npm (excluded), - disabled docker (excluded), unvalidated docker (excluded), backend_kind=None (excluded). + Mixed body: a custom and a Cloudsmith-owned Docker host (included), an npm + host and the download CDN (excluded). """ monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) - body = [ - # Included: Docker, enabled, validated - { - "host": "docker.acme.com", - "backend_kind": 6, - "domain_type": 1, - "enabled": True, - "validated": True, - }, - # Excluded: NPM backend - { - "host": "npm.acme.com", - "backend_kind": 9, - "domain_type": 1, - "enabled": True, - "validated": True, - }, - # Excluded: disabled - { - "host": "docker2.acme.com", - "backend_kind": 6, - "domain_type": 1, - "enabled": False, - "validated": True, - }, - # Excluded: unvalidated - { - "host": "docker3.acme.com", - "backend_kind": 6, - "domain_type": 1, - "enabled": True, - "validated": False, - }, - # Excluded: backend_kind=None (download domain) - { - "host": "dl.acme.com", - "backend_kind": None, - "domain_type": 0, - "enabled": True, - "validated": True, - }, - ] + body = { + "results": [ + _server_domain("docker.acme.com", "NATIVE_DOCKER"), + _server_domain("npm.acme.com", "NATIVE_NPM"), + _server_domain("dl.acme.com", "DOWNLOAD"), + _server_domain("docker.cloudsmith.io", "NATIVE_DOCKER", "GLOBAL"), + ] + } httpretty.register_uri( httpretty.GET, - f"{API_HOST}/orgs/acme/custom-domains/", + f"{API_HOST}/v2/workspaces/acme/domains/", body=json.dumps(body), status=200, content_type="application/json", @@ -558,20 +528,20 @@ def test_get_format_domains_filters_correctly(tmp_path, monkeypatch): api_host=API_HOST, ) - assert hosts == ["docker.acme.com"] + assert hosts == ["docker.acme.com", "docker.cloudsmith.io"] def test_get_format_domains_forwards_configure_api(monkeypatch): """A caller that has already configured the SDK can say so.""" recorded = {} - def _fake_get_custom_domains(org, **kwargs): + def _fake_get_workspace_domains(workspace, **kwargs): recorded.update(kwargs) return [] monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_custom_domains", - _fake_get_custom_domains, + "cloudsmith_cli.credential_helpers.domains.get_workspace_domains", + _fake_get_workspace_domains, ) get_format_domains("acme", BackendKind.DOCKER, configure_api=False) @@ -594,54 +564,19 @@ def _fake_get_custom_domains(org, **kwargs): ("docker.cloudsmith.com", None, None, None, True), # Non-cloudsmith → False ("evil.example.com", None, None, None, False), - # Custom enabled+validated → True + # Custom domain the server listed → True ( "docker.acme.com", "acme", - [ - CustomDomain( - host="docker.acme.com", - backend_kind=6, - domain_type=DomainType.NATIVE_API, - enabled=True, - validated=True, - org="acme", - ) - ], + [_server_domain("docker.acme.com", "NATIVE_DOCKER")], None, True, ), - # Custom disabled → False - ( - "docker.acme.com", - "acme", - [ - CustomDomain( - host="docker.acme.com", - backend_kind=6, - domain_type=DomainType.NATIVE_API, - enabled=False, - validated=True, - org="acme", - ) - ], - None, - False, - ), - # Custom enabled but unvalidated → False + # Host the server did not list → False ( "docker.acme.com", "acme", - [ - CustomDomain( - host="docker.acme.com", - backend_kind=6, - domain_type=DomainType.NATIVE_API, - enabled=True, - validated=False, - org="acme", - ) - ], + [_server_domain("npm.acme.com", "NATIVE_NPM")], None, False, ), @@ -649,16 +584,7 @@ def _fake_get_custom_domains(org, **kwargs): ( "docker.acme.com", "acme", - [ - CustomDomain( - host="docker.acme.com", - backend_kind=6, - domain_type=DomainType.NATIVE_API, - enabled=True, - validated=True, - org="acme", - ) - ], + [_server_domain("docker.acme.com", "NATIVE_DOCKER")], BackendKind.DOCKER, True, ), @@ -666,16 +592,7 @@ def _fake_get_custom_domains(org, **kwargs): ( "npm.acme.com", "acme", - [ - CustomDomain( - host="npm.acme.com", - backend_kind=9, - domain_type=DomainType.NATIVE_API, - enabled=True, - validated=True, - org="acme", - ) - ], + [_server_domain("npm.acme.com", "NATIVE_NPM")], BackendKind.DOCKER, False, ), @@ -684,16 +601,7 @@ def _fake_get_custom_domains(org, **kwargs): ( "DOCKER.ACME.COM", "acme", - [ - CustomDomain( - host="docker.acme.com", - backend_kind=6, - domain_type=DomainType.NATIVE_API, - enabled=True, - validated=True, - org="acme", - ) - ], + [_server_domain("docker.acme.com", "NATIVE_DOCKER")], BackendKind.DOCKER, True, ), @@ -704,12 +612,12 @@ def test_is_cloudsmith_domain( ): """is_cloudsmith_domain returns correct bool for standard, custom, and edge cases.""" monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) if cached_domains is not None: - write_cache(get_cache_path(org), cached_domains) + write_cache(get_cache_path(org, API_HOST), org, cached_domains) kwargs = { "credential": CredentialResult(api_key="k_abc", source_name="test"), @@ -759,24 +667,15 @@ def test_get_credentials_refuses_npm_custom_domain(tmp_path, monkeypatch): Proves the runtime passes backend_kind=DOCKER to is_cloudsmith_domain. """ monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) monkeypatch.setenv("CLOUDSMITH_ORG", "acme") - cache_path = get_cache_path("acme") write_cache( - cache_path, - [ - CustomDomain( - host="npm.acme.com", - backend_kind=9, - domain_type=DomainType.NATIVE_API, - enabled=True, - validated=True, - org="acme", - ) - ], + get_cache_path("acme", API_HOST), + "acme", + [_server_domain("npm.acme.com", "NATIVE_NPM")], ) credential = CredentialResult(api_key="k_xyz", source_name="test") 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..1b1a6d21 --- /dev/null +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper_domains.py @@ -0,0 +1,220 @@ +# Copyright 2026 Cloudsmith Ltd +"""Tests for reading the server's Workspace domain records and caching them.""" + +import subprocess +import sys +from unittest.mock import patch + +import cloudsmith_api +import pytest + +from ....credential_helpers.backends import BackendKind +from ....credential_helpers.domains import ( + DomainScope, + DomainType, + domain_from_payload, + get_cache_path, + get_format_domains, + read_all_cached_domains, + read_cache, + write_cache, +) + + +@pytest.fixture(autouse=True) +def _cache_dir(tmp_path, monkeypatch): + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", + lambda: str(tmp_path), + ) + + +def _payload(name, domain_type, scope="GLOBAL", repository=None): + return { + "domain_name": name, + "domain_type": domain_type, + "domain_scope": scope, + "repository": repository, + } + + +def _record(*args, **kwargs): + """Parse one server record that this CLI is expected to understand.""" + record = domain_from_payload(_payload(*args, **kwargs), "acme") + assert record is not None + return record + + +@pytest.mark.parametrize( + "server_type,domain_type,backend_kind", + [ + ("DOWNLOAD", DomainType.DOWNLOAD, None), + ("UPLOAD", DomainType.UPLOAD, None), + ("CLOUDSMITH_API", DomainType.CLOUDSMITH_API, None), + ("NATIVE_DOCKER", DomainType.NATIVE_API, BackendKind.DOCKER), + ("NATIVE_HELM_OCI", DomainType.NATIVE_API, BackendKind.HELM), + ("NATIVE_GO", DomainType.NATIVE_API, BackendKind.GO), + ], +) +def test_server_domain_types_map_to_a_purpose_and_format( + server_type, domain_type, backend_kind +): + """A native type names its package format; the others serve no single one.""" + record = _record("host.example.com", server_type) + + assert record.domain_type is domain_type + assert record.backend_kind == backend_kind + + +@pytest.mark.parametrize("server_type", ["NATIVE_SOMETHING_NEW", "MIRROR", None]) +def test_a_domain_type_this_cli_does_not_know_is_skipped(server_type): + """The server adds types over time; an unknown one is not guessed at.""" + assert domain_from_payload(_payload("x.example.com", server_type), "acme") is None + + +def test_a_domain_scope_this_cli_does_not_know_is_skipped(): + """An unknown scope cannot say whom the domain serves, so it is not guessed.""" + payload = _payload("x.example.com", "DOWNLOAD", scope="ORGANIZATION_UNIT") + + assert domain_from_payload(payload, "acme") is None + + +def test_a_cloudsmith_owned_domain_belongs_to_no_workspace(): + """A GLOBAL domain serves every Workspace, so it is not attributed to one.""" + record = _record("dl.cloudsmith.io", "DOWNLOAD") + + assert record.scope is DomainScope.GLOBAL + assert record.is_custom is False + assert record.workspace is None + + +def test_a_custom_domain_keeps_its_workspace_and_repository(): + """A REPOSITORY domain says which repository it is restricted to.""" + record = _record("dl.acme.com", "DOWNLOAD", "REPOSITORY", "prod") + + assert record.is_custom is True + assert record.workspace == "acme" + assert record.serves_repository("prod") + assert not record.serves_repository("dev") + + +def test_cache_round_trip_keeps_the_server_order(tmp_path): + """The cache holds the server's answer, so a cached run sees the same order.""" + payload = [ + _payload("docker.acme.com", "NATIVE_DOCKER", "WORKSPACE"), + _payload("docker.cloudsmith.io", "NATIVE_DOCKER"), + ] + cache_path = tmp_path / "acme.json" + write_cache(cache_path, "acme", payload) + + assert [d.host for d in read_cache(cache_path) or []] == [ + "docker.acme.com", + "docker.cloudsmith.io", + ] + + +def test_each_api_host_has_its_own_cache(): + """A Workspace slug on one API host is a different Workspace on another. + + Sharing one cache entry would hand one host's domains to a run on the other. + """ + production = get_cache_path("acme", "https://api.cloudsmith.io") + other = get_cache_path("acme", "https://api.example.com") + + assert production != other + + +@pytest.mark.parametrize( + "other_base", + [ + "http://api.example.com", + "https://api.example.com:8443", + "https://api.example.com/tenant", + ], +) +def test_api_bases_on_one_hostname_have_their_own_cache(other_base): + """A different scheme, port or path is a different Cloudsmith.""" + assert get_cache_path("acme", "https://api.example.com") != get_cache_path( + "acme", other_base + ) + + +def test_a_trailing_slash_does_not_change_the_cache(): + """The same API base with a trailing slash uses the same cache.""" + assert get_cache_path("acme", "https://api.example.com/") == get_cache_path( + "acme", "https://api.example.com" + ) + + +def test_no_api_host_uses_the_cache_of_the_default_api_host(): + """Without an API host, the cache is the one for the default API host.""" + assert get_cache_path("acme") == get_cache_path( + "acme", cloudsmith_api.Configuration().host + ) + + +def test_no_api_host_uses_the_cache_of_the_configured_sdk_host(monkeypatch): + """Without an API host, the cache is the one for the host the SDK calls. + + A request then never fills the cache of another API host. + """ + config = cloudsmith_api.Configuration() + config.host = "https://api.example.com" + monkeypatch.setattr(cloudsmith_api.Configuration, "_default", config) + + assert get_cache_path("acme") == get_cache_path("acme", "https://api.example.com") + + +def test_a_cache_lookup_does_not_import_the_sdk(): + """A credential helper that hits the cache does not pay for the SDK import.""" + code = ( + "import sys\n" + "from cloudsmith_cli.credential_helpers.domains import " + "get_cache_path\n" + "get_cache_path('acme')\n" + "print('cloudsmith_api' in sys.modules)\n" + ) + result = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True, check=True + ) + + assert result.stdout.strip() == "False" + + +def test_cached_domains_are_read_for_the_current_api_host_only(): + """A no-Workspace listing on one API host never shows another host's domains.""" + write_cache( + get_cache_path("acme", "https://api.cloudsmith.io"), + "acme", + [_payload("dl.cloudsmith.io", "DOWNLOAD")], + ) + write_cache( + get_cache_path("acme", "https://api.example.com"), + "acme", + [_payload("dl.example.com", "DOWNLOAD")], + ) + + hosts = [d.host for d in read_all_cached_domains("https://api.example.com")] + + assert hosts == ["dl.example.com"] + + +def test_format_domains_keep_the_server_preference_order(): + """The first host is the one the server prefers, Cloudsmith-owned last.""" + records = [ + _record(name, server_type, scope) + for name, server_type, scope in [ + ("docker.acme.com", "NATIVE_DOCKER", "WORKSPACE"), + ("npm.acme.com", "NATIVE_NPM", "WORKSPACE"), + ("dl.cloudsmith.io", "DOWNLOAD", "GLOBAL"), + ("docker.cloudsmith.io", "NATIVE_DOCKER", "GLOBAL"), + ] + ] + + with patch( + "cloudsmith_cli.credential_helpers.domains.get_workspace_domains", + return_value=records, + ): + hosts = get_format_domains("acme", BackendKind.DOCKER) + + assert hosts == ["docker.acme.com", "docker.cloudsmith.io"] 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 9723032d..eef149b2 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py @@ -18,7 +18,6 @@ from cloudsmith_cli.credential_helpers.pnpm.installer import PNPMInstaller from ....core.credentials.models import CredentialResult -from ....credential_helpers.default_domains import DomainType from ....credential_helpers.docker.installer import DockerInstaller from ....credential_helpers.launchers import ( _launcher_content, @@ -633,6 +632,54 @@ def test_pnpm_installer_status_type_contract(tmp_path: Path, monkeypatch: Monkey ) +def test_docker_install_uses_the_cloudsmith_owned_host_the_api_lists( + tmp_path, monkeypatch +): + """Another API host lists its own Docker host, so the default is not added.""" + docker_dir = tmp_path / ".docker" + monkeypatch.setenv("DOCKER_CONFIG", str(docker_dir)) + bin_dir = tmp_path / "bin" + monkeypatch.setenv("PATH", str(bin_dir)) + monkeypatch.setattr( + _INSTALLER_GET_FORMAT_DOMAINS, + lambda *_a, **_kw: ["docker.acme.com", "docker.example.com"], + ) + + DockerInstaller().install( + bin_dir=str(bin_dir), + org="acme", + credential=CredentialResult(api_key="k_test", source_name="test"), + ) + + cfg = json.loads((docker_dir / "config.json").read_text()) + assert sorted(cfg["credHelpers"]) == ["docker.acme.com", "docker.example.com"] + + +def test_pnpm_install_uses_the_cloudsmith_owned_host_the_api_lists( + tmp_path, monkeypatch +): + """Another API host lists its own npm host, so the default is not added.""" + npm_path = tmp_path / ".npmrc" + monkeypatch.setenv("NPM_CONFIG_USERCONFIG", str(npm_path)) + bin_dir = tmp_path / "bin" + monkeypatch.setenv("PATH", str(bin_dir)) + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.pnpm.installer.get_format_domains", + lambda *_a, **_kw: ["npm.acme.com", "npm.example.com"], + ) + + PNPMInstaller().install( + bin_dir=str(bin_dir), + org="acme", + credential=CredentialResult(api_key="k_test", source_name="test"), + ) + + assert npm_path.read_text() == ( + f"//npm.acme.com/:tokenHelper={bin_dir}/pnpm-credential-cloudsmith\n" + f"//npm.example.com/:tokenHelper={bin_dir}/pnpm-credential-cloudsmith" + ) + + @pytest.mark.parametrize( "scenario", [ @@ -662,7 +709,7 @@ def test_autodiscovery(tmp_path, monkeypatch, scenario): def _fake_get_format_domains(*_a, **kwargs): captured.update(kwargs) - return ["docker.acme.com"] + return ["docker.acme.com", "docker.cloudsmith.io"] monkeypatch.setattr(_INSTALLER_GET_FORMAT_DOMAINS, _fake_get_format_domains) installer = DockerInstaller() @@ -765,58 +812,41 @@ def test_refresh_flag(tmp_path, monkeypatch, refresh): """refresh=False uses the on-disk cache; refresh=True bypasses it and hits the API.""" import time - from ....credential_helpers.custom_domains import ( - CustomDomain, + from ....credential_helpers.domains import ( + domain_from_payload, get_cache_path, - get_custom_domains, + get_workspace_domains, write_cache, ) monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) + def _payload(name): + return { + "domain_name": name, + "domain_type": "NATIVE_DOCKER", + "domain_scope": "WORKSPACE", + "repository": None, + } + cache_path = get_cache_path("acme") - cached_domain = CustomDomain( - host="docker.acme.com", - backend_kind=6, - enabled=True, - validated=True, - org="acme", - domain_type=DomainType.NATIVE_API, - ) - write_cache(cache_path, [cached_domain]) + write_cache(cache_path, "acme", [_payload("docker.acme.com")]) os.utime(cache_path, (time.time(), time.time())) - fresh_domain = CustomDomain( - host="new.acme.com", - backend_kind=6, - enabled=True, - validated=True, - org="acme", - domain_type=DomainType.NATIVE_API, - ) api_calls = [] def _fake_list(*_a, **_kw): api_calls.append(True) - return [ - { - "host": "new.acme.com", - "backend_kind": 6, - "enabled": True, - "validated": True, - "domain_type": 3, - "primary": True, - } - ] + return [_payload("new.acme.com")] with patch( - "cloudsmith_cli.core.api.orgs.list_custom_domains", + "cloudsmith_cli.core.api.workspaces.list_domains", _fake_list, ): - result = get_custom_domains( + result = get_workspace_domains( "acme", credential=CredentialResult(api_key="k", source_name="test"), refresh=refresh, @@ -825,13 +855,13 @@ def _fake_list(*_a, **_kw): if refresh: # API must have been called assert api_calls, "API must be called when refresh=True" - assert result == [fresh_domain] + assert result == [domain_from_payload(_payload("new.acme.com"), "acme")] else: # API must NOT have been called assert not api_calls, ( "API must not be called when refresh=False with valid cache" ) - assert result == [cached_domain] + assert result == [domain_from_payload(_payload("docker.acme.com"), "acme")] # --------------------------------------------------------------------------- diff --git a/cloudsmith_cli/cli/tests/commands/test_custom_domain_precedence.py b/cloudsmith_cli/cli/tests/commands/test_custom_domain_precedence.py deleted file mode 100644 index 1cd99011..00000000 --- a/cloudsmith_cli/cli/tests/commands/test_custom_domain_precedence.py +++ /dev/null @@ -1,115 +0,0 @@ -# Copyright 2026 Cloudsmith Ltd -"""Tests for custom-domain precedence and the record fields it ranks on. - -Those fields have to survive the on-disk cache: a run served from the cache -must rank domains identically to the run that populated it. -""" - -from datetime import UTC, datetime -from unittest.mock import patch - -from ....credential_helpers.backends import BackendKind -from ....credential_helpers.custom_domains import ( - CustomDomain, - _record_from_payload, - get_format_domains, - read_cache, - write_cache, -) -from ....credential_helpers.default_domains import DomainType - - -def _domain( - host, *, backend_kind=None, primary=True, created_at="2025-01-01T00:00:00Z" -): - """Build a CustomDomain record for a test.""" - return CustomDomain( - host=host, - backend_kind=backend_kind, - enabled=True, - validated=True, - org="acme", - domain_type=( - DomainType.DOWNLOAD if backend_kind is None else DomainType.NATIVE_API - ), - primary=primary, - created_at=created_at, - ) - - -def test_format_domains_are_returned_in_precedence_order(): - """The host Cloudsmith treats as active comes first. - - Callers take the first host, so the order is the answer: primary beats - secondary, and the oldest breaks a tie between equals. - """ - records = [ - _domain("second.acme.com", backend_kind=BackendKind.DOCKER, primary=False), - _domain( - "third.acme.com", - backend_kind=BackendKind.DOCKER, - created_at="2026-01-01T00:00:00Z", - ), - _domain( - "first.acme.com", - backend_kind=BackendKind.DOCKER, - created_at="2024-01-01T00:00:00Z", - ), - ] - - with patch( - "cloudsmith_cli.credential_helpers.custom_domains.get_custom_domains", - return_value=records, - ): - hosts = get_format_domains("acme", BackendKind.DOCKER) - - assert hosts == ["first.acme.com", "third.acme.com", "second.acme.com"] - - -def test_domain_with_no_creation_time_ranks_last_not_first(): - """An unknown creation time is a last resort, not the oldest domain.""" - records = [ - _domain("nodate.acme.com", backend_kind=BackendKind.DOCKER, created_at=None), - _domain( - "dated.acme.com", - backend_kind=BackendKind.DOCKER, - created_at="2020-01-01T00:00:00Z", - ), - ] - - with patch( - "cloudsmith_cli.credential_helpers.custom_domains.get_custom_domains", - return_value=records, - ): - hosts = get_format_domains("acme", BackendKind.DOCKER) - - assert hosts == ["dated.acme.com", "nodate.acme.com"] - - -def test_precedence_fields_survive_the_api_and_the_cache(tmp_path): - """A cached run must rank domains the way the fetching run did. - - The SDK deserialises ``created_at`` into a datetime that JSON cannot hold, - so it is normalised on the way in; dropping any of these fields on the way - to disk would silently fall back to API order for the whole seven-day TTL. - """ - record = _record_from_payload( - { - "host": "dl.acme.com", - "enabled": True, - "validated": True, - "primary": False, - "repository": {"name": "Prod", "slug": "prod"}, - "created_at": datetime(2025, 3, 4, 9, 22, 30, tzinfo=UTC), - }, - "acme", - ) - - assert record.created_at.startswith("2025-03-04T09:22:30") - assert record.primary is False - assert record.repository == "prod" - - cache_path = tmp_path / "acme.json" - write_cache(cache_path, [record]) - - assert read_cache(cache_path) == [record] diff --git a/cloudsmith_cli/cli/tests/commands/test_default_domains.py b/cloudsmith_cli/cli/tests/commands/test_default_domains.py deleted file mode 100644 index 20010697..00000000 --- a/cloudsmith_cli/cli/tests/commands/test_default_domains.py +++ /dev/null @@ -1,189 +0,0 @@ -# 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, - 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} - - -@pytest.fixture -def no_trusted_config(tmp_path, monkeypatch): - """Pin the trusted search path at an empty directory. - - ``load_default_domains`` falls back to the trusted lookup, so without this - the developer's own ``config.ini`` decides what these tests observe. - """ - empty = tmp_path / "empty" - empty.mkdir() - monkeypatch.setattr(cli_config.ConfigReader, "config_files", ["config.ini"]) - monkeypatch.setattr(cli_config.ConfigReader, "config_searchpath", [str(empty)]) - - -@pytest.mark.parametrize( - "host,label,kind", - [ - ("python.cloudsmith.io", "python", BackendKind.PYTHON), - ("docker.cloudsmith.io", "docker", BackendKind.DOCKER), - ("golang.cloudsmith.io", "go", BackendKind.GO), - ("dl.cloudsmith.io", None, None), - ("upload.cloudsmith.io", None, None), - ], -) -def test_builtin_entries_map_to_expected_kinds(host, label, kind): - """Each built-in host carries the right label and backend kind.""" - assert len(BUILTIN_DOMAINS) == 21 - - domain = _by_host(BUILTIN_DOMAINS)[host] - assert domain.format_label == label - assert domain.backend_kind == kind - - -@pytest.mark.parametrize("exists", [False, True], ids=["absent", "no-domains-section"]) -def test_a_config_declaring_no_table_falls_back_to_builtins( - tmp_path, no_trusted_config, exists -): - """Only a readable [domains] section replaces the built-in table.""" - config = tmp_path / "config.ini" - if exists: - 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_config_section_replaces_builtins_wholesale(tmp_path, no_trusted_config): - """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 = download\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"].domain_type is DomainType.NATIVE_API - assert indexed["cdn.internal.example.com"].format_label is None - assert indexed["cdn.internal.example.com"].domain_type is DomainType.DOWNLOAD - - -@pytest.mark.parametrize( - "label,domain_type", - [("download", DomainType.DOWNLOAD), ("upload", DomainType.UPLOAD)], -) -def test_config_reserved_labels_declare_formatless_hosts( - tmp_path, no_trusted_config, label, domain_type -): - """`download` and `upload` name the two hosts no package format names. - - A dedicated deployment has to redirect its upload endpoint, and no - BackendKind identifies it - without a reserved label the table cannot name - it at all. - """ - config = tmp_path / "config.ini" - config.write_text( - f"[domains]\nhost.internal.example.com = {label}\n", encoding="utf-8" - ) - - domain = _by_host(load_default_domains(config_path=config))[ - "host.internal.example.com" - ] - - assert domain.domain_type is domain_type - assert domain.backend_kind is None - - -@pytest.mark.parametrize("label", ["", "widget"], ids=["empty", "unrecognised"]) -def test_an_entry_naming_no_format_is_skipped(tmp_path, no_trusted_config, label): - """A label is required, and a bad one is dropped rather than guessed at. - - Reading it as the download CDN would leave a typo'd format quietly serving - the wrong thing, which is far harder to spot than a missing host. - """ - config = tmp_path / "config.ini" - config.write_text( - f"[domains]\npackages.internal.example.com = python\n" - f"broken.internal.example.com = {label}\n", - encoding="utf-8", - ) - - hosts = _by_host(load_default_domains(config_path=config)) - - assert set(hosts) == {"packages.internal.example.com"} - - -def test_a_config_that_is_not_utf8_falls_back_to_builtins(tmp_path, no_trusted_config): - """An undecodable config.ini falls back rather than aborting the command.""" - config = tmp_path / "config.ini" - config.write_bytes(b"[domains]\nmaven.acme.example.com = maven\n# \xe9\xe8\xf1\n") - - 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 = download\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_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) - 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 diff --git a/cloudsmith_cli/cli/tests/commands/test_domains.py b/cloudsmith_cli/cli/tests/commands/test_domains.py index bc877130..7585cf8b 100644 --- a/cloudsmith_cli/cli/tests/commands/test_domains.py +++ b/cloudsmith_cli/cli/tests/commands/test_domains.py @@ -9,15 +9,14 @@ from ....cli import config as cli_config from ....cli.commands.domains import list_domains from ....core.api.exceptions import ApiException -from ....credential_helpers.backends import BackendKind -from ....credential_helpers.custom_domains import ( - CustomDomain, +from ....credential_helpers.domains import ( + domain_from_payload, get_cache_path, write_cache, ) -from ....credential_helpers.default_domains import DomainType -_PATCH_TARGET = "cloudsmith_cli.cli.commands.domains.get_custom_domains" +_PATCH_TARGET = "cloudsmith_cli.cli.commands.domains.get_workspace_domains" +_ORGS_PATCH_TARGET = "cloudsmith_cli.cli.commands.domains.list_orgs" HERMETIC_ARGS = ["--api-key", "fake-api-key"] @@ -26,63 +25,62 @@ def hermetic_environment(monkeypatch, tmp_path): """Keep the developer's real environment out of these tests. - An inherited organisation, config file, [domains] table or custom-domain - cache would each replace part of what these tests assert on - and the CLI's - Options object and ConfigReader's search path are both process-wide, so - they are pinned per test rather than merely cleared. + An inherited Workspace, config file or domain cache would each replace part + of what these tests assert on - and the CLI's Options object and + ConfigReader's search path are both process-wide, so they are pinned per + test rather than merely cleared. """ monkeypatch.delenv("CLOUDSMITH_ORG", raising=False) + monkeypatch.delenv("CLOUDSMITH_WORKSPACE", raising=False) + monkeypatch.delenv("CLOUDSMITH_API_HOST", raising=False) monkeypatch.delenv("CLOUDSMITH_CONFIG_FILE", raising=False) + monkeypatch.delenv("CLOUDSMITH_API_KEY", raising=False) + monkeypatch.setenv("CLOUDSMITH_NO_KEYRING", "1") monkeypatch.delattr(cli_config.OPTIONS, "value", raising=False) monkeypatch.setattr(cli_config.ConfigReader, "config_files", ["config.ini"]) monkeypatch.setattr(cli_config.ConfigReader, "config_searchpath", ["."]) monkeypatch.setattr( - "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", lambda: str(tmp_path), ) -def _domain( - host, - backend_kind, - org="acme", - repository=None, - created_at=None, - domain_type=None, - enabled=True, - validated=True, - primary=True, -): - """Build a CustomDomain record for a test. - - The server classifies every domain, so a record stands in for one it sent: - a host serving a package format speaks that format's protocol, and the - format-less ones default to the CDN unless the caller says otherwise. - """ - if domain_type is None: - domain_type = ( - DomainType.NATIVE_API if backend_kind is not None else DomainType.DOWNLOAD - ) - return CustomDomain( - host=host, - backend_kind=backend_kind, - enabled=enabled, - validated=validated, - org=org, - repository=repository, - domain_type=domain_type, - created_at=created_at, - primary=primary, +def _payload(name, domain_type, scope="WORKSPACE", repository=None): + return { + "domain_name": name, + "domain_type": domain_type, + "domain_scope": scope, + "repository": repository, + } + + +def _domain(name, domain_type, scope="WORKSPACE", repository=None, workspace="acme"): + """Build a record as the server would send it for *workspace*.""" + return domain_from_payload( + _payload(name, domain_type, scope, repository), workspace ) -def _invoke(runner, *args, custom_domains=()): - """Run the command with the custom-domain lookup stubbed; assert a clean exit. +SERVER_DOMAINS = [ + _domain("dl-prod.acme.example.com", "DOWNLOAD", "REPOSITORY", "prod"), + _domain("dl-dev.acme.example.com", "DOWNLOAD", "REPOSITORY", "dev"), + _domain("pypi.acme.example.com", "NATIVE_PYTHON"), + _domain("dl.acme.example.com", "DOWNLOAD"), + _domain("dl.cloudsmith.io", "DOWNLOAD", "GLOBAL"), + _domain("upload.cloudsmith.io", "UPLOAD", "GLOBAL"), + _domain("api.cloudsmith.io", "CLOUDSMITH_API", "GLOBAL"), + _domain("maven.cloudsmith.io", "NATIVE_MAVEN", "GLOBAL"), + _domain("python.cloudsmith.io", "NATIVE_PYTHON", "GLOBAL"), +] + + +def _invoke(runner, *args, domains=SERVER_DOMAINS): + """Run the command with the 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: + with patch(_PATCH_TARGET, return_value=list(domains)) as mock_get: result = runner.invoke( list_domains, [*args, *HERMETIC_ARGS], catch_exceptions=False ) @@ -91,25 +89,24 @@ def _invoke(runner, *args, custom_domains=()): return result, mock_get +def _hosts(result): + return [entry["host"] for entry in json.loads(result.stdout)["domains"]] + + 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_one_schema_for_both_kinds_of_host(runner, monkeypatch): - """Built-in and custom hosts share one entry schema, under a versioned document. + """Cloudsmith-owned and custom hosts share one entry schema. - A key present on custom entries alone raises KeyError on the first built-in - host for any consumer filtering the document on it. + A key present on custom entries alone raises KeyError on the first + Cloudsmith-owned host for any consumer filtering the document on it. """ - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") - result, _ = _invoke( - runner, - custom_domains=[ - _domain("pypi.acme.example.com", 3, created_at="2026-02-06T00:00:00Z") - ], - ) + result, _ = _invoke(runner) document = json.loads(result.stdout) assert set(document) == {"version", "domains"} @@ -125,8 +122,8 @@ def test_document_has_one_schema_for_both_kinds_of_host(runner, monkeypatch): "workspace": "acme", "org": "acme", "repository": None, - "primary": True, - "created_at": "2026-02-06T00:00:00Z", + "primary": None, + "created_at": None, } assert by_host["python.cloudsmith.io"] == { "host": "python.cloudsmith.io", @@ -136,7 +133,7 @@ def test_document_has_one_schema_for_both_kinds_of_host(runner, monkeypatch): "workspace": None, "org": None, "repository": None, - "primary": True, + "primary": None, "created_at": None, } @@ -146,11 +143,14 @@ def test_document_has_one_schema_for_both_kinds_of_host(runner, monkeypatch): [ ("dl.cloudsmith.io", None, "download"), ("upload.cloudsmith.io", None, "upload"), + ("api.cloudsmith.io", None, "cloudsmith_api"), ("maven.cloudsmith.io", "maven", "native_api"), ], ) -def test_builtin_hosts_say_what_they_serve(runner, host, format_, domain_type): - """The two format-less hosts are download/upload; the rest are native API.""" +def test_hosts_say_what_they_serve(runner, monkeypatch, host, format_, domain_type): + """Each host carries the purpose the server gave it.""" + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") + entry = _by_host(_invoke(runner)[0])[host] assert entry["format"] == format_ @@ -158,44 +158,101 @@ def test_builtin_hosts_say_what_they_serve(runner, host, format_, domain_type): @pytest.mark.parametrize("source", ["env", "config", "flag"]) -def test_the_resolved_organisation_is_looked_up(runner, monkeypatch, tmp_path, source): - """However the organisation is configured, that is the one looked up. +def test_the_resolved_workspace_is_looked_up(runner, monkeypatch, tmp_path, source): + """However the Workspace is configured, that is the one looked up. The accepted spellings of the option are covered in test_org_option.py; this only proves the command hands what it resolved to the lookup. """ args = [] if source == "env": - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") elif source == "config": config = tmp_path / "config.ini" - config.write_text("[default]\norg = acme\n", encoding="utf-8") + config.write_text("[default]\nworkspace = acme\n", encoding="utf-8") args = ["--config-file", str(config)] else: - args = ["--org", "acme"] + args = ["--workspace", "acme"] _, mock_get = _invoke(runner, *args) assert mock_get.call_args.args[0] == "acme" -def test_without_org_no_custom_domain_lookup(runner): - """With no organisation configured, no custom-domain lookup is attempted.""" - _, mock_get = _invoke(runner) +def test_without_workspace_or_credential_no_lookup(runner): + """With no Workspace and no credential there is nothing to ask the API.""" + with patch(_PATCH_TARGET) as mock_get, patch(_ORGS_PATCH_TARGET) as mock_orgs: + result = runner.invoke(list_domains, catch_exceptions=False) + + assert result.exit_code == 0, result.output + mock_get.assert_not_called() + mock_orgs.assert_not_called() + assert json.loads(result.stdout) == {"version": 1, "domains": []} + + +def test_without_workspace_a_member_workspace_supplies_the_domains(runner): + """With nothing cached, any Workspace of the credential lists the hosts. + + Every Workspace answer holds the Cloudsmith-owned hosts, so consumers that + set no Workspace still get them. + """ + with patch(_ORGS_PATCH_TARGET, return_value=[{"slug": "acme"}]) as mock_orgs: + result, mock_get = _invoke(runner) + + mock_orgs.assert_called_once_with(page=1, page_size=1) + assert mock_get.call_args.args[0] == "acme" + assert mock_get.call_args.kwargs["strict"] is True + assert result.stderr == "" + assert _hosts(result) == [domain.host for domain in SERVER_DOMAINS] + + +def test_without_workspace_a_credential_with_no_workspace_lists_nothing(runner): + """A credential that is a member of no Workspace has no domains to list.""" + with patch(_ORGS_PATCH_TARGET, return_value=[]): + result, mock_get = _invoke(runner) mock_get.assert_not_called() + assert _hosts(result) == [] + + +def test_without_workspace_a_failed_member_lookup_exits_non_zero(runner): + """A failed lookup is not shown as "no domains".""" + with patch( + _ORGS_PATCH_TARGET, side_effect=ApiException(401, detail="Invalid API key") + ): + result = runner.invoke(list_domains, HERMETIC_ARGS, catch_exceptions=False) + + assert result.exit_code != 0 + assert result.stdout.strip() == "" + assert "Invalid API key" in result.stderr + + +def test_refresh_without_workspace_warns_that_the_cache_is_unchanged(runner): + """--refresh needs a Workspace to fetch from.""" + write_cache( + get_cache_path("acme"), + "acme", + [_payload("dl.cloudsmith.io", "DOWNLOAD", "GLOBAL")], + ) + + result, mock_get = _invoke(runner, "--refresh") + + mock_get.assert_not_called() + assert "--refresh needs a Workspace" in result.stderr + assert _hosts(result) == ["dl.cloudsmith.io"] def test_lookup_is_handed_the_commands_own_settings(runner, monkeypatch): """The lookup runs on the command's terms, not its own defaults. - ``strict`` so a typo'd org or unreachable API cannot render as "no custom + ``strict`` so a typo'd Workspace or unreachable API cannot render as "no domains"; ``configure_api=False`` because `initialise_api` has already applied the proxy, TLS, user-agent and header settings that a narrower - re-initialisation inside the lookup would discard; the resolved credential - and ``--refresh`` because neither reaches the lookup any other way. + re-initialisation inside the lookup would discard; the resolved credential, + API host and ``--refresh`` because none reaches the lookup any other way. """ - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") + monkeypatch.setenv("CLOUDSMITH_API_HOST", "https://api.example.com") _, mock_get = _invoke(runner, "--refresh") @@ -203,6 +260,7 @@ def test_lookup_is_handed_the_commands_own_settings(runner, monkeypatch): assert kwargs["strict"] is True assert kwargs["configure_api"] is False assert kwargs["refresh"] is True + assert kwargs["api_host"] == "https://api.example.com" assert kwargs["credential"].api_key == "fake-api-key" @@ -212,74 +270,47 @@ def test_lookup_failure_says_what_went_wrong(runner, monkeypatch): ApiException builds Exception with no args, so interpolating one used to yield a message that stopped dead at the colon. """ - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") with patch(_PATCH_TARGET, side_effect=ApiException(401, detail="Invalid API key")): result = runner.invoke(list_domains, HERMETIC_ARGS, catch_exceptions=False) assert result.exit_code != 0 assert result.stdout.strip() == "" - assert "Failed to fetch custom domains" in result.stderr + assert "Failed to fetch domains" in result.stderr assert "401" in result.stderr assert "Invalid API key" in result.stderr -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)) - - assert set(_by_host(result)) == {"index.internal.example.com"} - - -def test_untrusted_config_section_is_ignored_with_a_warning( - runner, monkeypatch, tmp_path -): - """A cwd config.ini cannot replace the host table, and says so on stderr. - - That file travels with whatever repository is checked out, and this table - decides which hosts may receive a credential. - """ - 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 - assert "python.cloudsmith.io" in _by_host(result) - assert "index.internal.example.com" not in _by_host(result) - +def test_lists_cached_domains_when_no_workspace_is_configured(runner, monkeypatch): + """A run with no Workspace lists what earlier runs cached, without any API call. -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. + Each Workspace's custom domains keep their attribution, and the + Cloudsmith-owned hosts that every cached answer repeats are listed once. """ called = [] monkeypatch.setattr(_PATCH_TARGET, lambda *a, **k: called.append("api") or []) - write_cache(get_cache_path("acme"), [_domain("dl.acme.com", None, org="acme")]) + shared = _payload("dl.cloudsmith.io", "DOWNLOAD", "GLOBAL") write_cache( - get_cache_path("widgets"), [_domain("dl.widgets.com", None, org="widgets")] + get_cache_path("widgets"), + "widgets", + [_payload("dl.widgets.com", "DOWNLOAD"), shared], + ) + write_cache( + get_cache_path("acme"), "acme", [_payload("dl.acme.com", "DOWNLOAD"), shared] ) result = runner.invoke(list_domains, HERMETIC_ARGS, catch_exceptions=False) 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} == { + assert result.stderr == "" + document = json.loads(result.stdout)["domains"] + assert [(d["host"], d["workspace"]) for d in document] == [ ("dl.acme.com", "acme"), ("dl.widgets.com", "widgets"), - } + ("dl.cloudsmith.io", None), + ] def test_format_filter_keeps_that_format_and_the_formatless_hosts(runner, monkeypatch): @@ -289,135 +320,72 @@ def test_format_filter_keeps_that_format_and_the_formatless_hosts(runner, monkey every format, so excluding it would hide the very host Maven resolves dependencies from. """ - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") - records = [ - _domain("mvn.acme.example.com", int(BackendKind.MAVEN)), - _domain("pypi.acme.example.com", int(BackendKind.PYTHON)), - _domain("dl.acme.example.com", None), - ] - - by_host = _by_host(_invoke(runner, "--format", "maven", custom_domains=records)[0]) + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") - assert "mvn.acme.example.com" in by_host - assert "dl.acme.example.com" in by_host - assert "pypi.acme.example.com" not in by_host - # Built-ins are filtered on the same rule. - assert "maven.cloudsmith.io" in by_host - assert "python.cloudsmith.io" not in by_host - assert "dl.cloudsmith.io" in by_host + result, _ = _invoke(runner, "--format", "maven") + assert _hosts(result) == [ + "dl-prod.acme.example.com", + "dl-dev.acme.example.com", + "dl.acme.example.com", + "dl.cloudsmith.io", + "upload.cloudsmith.io", + "api.cloudsmith.io", + "maven.cloudsmith.io", + ] -def test_custom_domains_are_listed_before_the_builtin_hosts(runner, monkeypatch): - """An organisation's own hosts come first; the built-ins are the fallback. - A caller reading the document top-down should reach the host bound to its - organisation before the public one that merely also works. - """ - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") +def test_hosts_keep_the_server_preference_order(runner, monkeypatch): + """The server lists domains most-preferred first, and so does the command.""" + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") - result, _ = _invoke(runner, custom_domains=[_domain("dl.acme.example.com", None)]) - - types = [entry["type"] for entry in json.loads(result.stdout)["domains"]] - assert types == ["custom"] + ["default"] * (len(types) - 1) + result, _ = _invoke(runner) + assert _hosts(result) == [domain.host for domain in SERVER_DOMAINS] -def test_a_domain_that_cannot_serve_traffic_is_not_listed(runner, monkeypatch): - """The listing answers what is usable, so a host serving nothing is out. - It needs both flags: a domain is usable only when it is enabled *and* - validated. Under ``--repo`` this also keeps a dead host off the top of the - list, since one bound to the repository outranks an organisation-wide one - whatever its state. - """ - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") - records = [ - _domain("dl.acme.example.com", None), - _domain("disabled.acme.example.com", None, enabled=False), - _domain("unvalidated.acme.example.com", None, validated=False), +def test_the_api_order_is_never_rearranged(runner, monkeypatch): + """The API decides the priority, even where the CLI might expect another.""" + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") + domains = [ + _domain("dl.cloudsmith.io", "DOWNLOAD", "GLOBAL"), + _domain("dl.acme.example.com", "DOWNLOAD"), ] - by_host = _by_host(_invoke(runner, custom_domains=records)[0]) + result, _ = _invoke(runner, domains=domains) - assert "dl.acme.example.com" in by_host - assert "disabled.acme.example.com" not in by_host - assert "unvalidated.acme.example.com" not in by_host + assert _hosts(result) == ["dl.cloudsmith.io", "dl.acme.example.com"] def test_domain_type_filter_selects_hosts_by_purpose(runner, monkeypatch): """--domain-type answers "which host do I upload to?". Unlike --format, every host has exactly one purpose, so this is an exact - match: an upload endpoint cannot stand in for the download CDN, and a - native-API host speaks one package format's protocol and neither. + match: an upload endpoint cannot stand in for the download CDN. """ - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") - records = [ - _domain("up.acme.example.com", None, domain_type=DomainType.UPLOAD), - _domain("dl.acme.example.com", None, domain_type=DomainType.DOWNLOAD), - _domain("mvn.acme.example.com", int(BackendKind.MAVEN)), - ] + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") - by_host = _by_host( - _invoke(runner, "--domain-type", "upload", custom_domains=records)[0] - ) + result, _ = _invoke(runner, "--domain-type", "upload") - assert set(by_host) == {"upload.cloudsmith.io", "up.acme.example.com"} + assert _hosts(result) == ["upload.cloudsmith.io"] -def test_repo_filter_drops_other_repositories_and_ranks_the_rest(runner, monkeypatch): +def test_repo_filter_drops_other_repositories(runner, monkeypatch): """--repo lists the hosts usable for one repository, most-preferred first. A repository-scoped host names its own repository in its URLs, so it can - never stand in for another; an organisation-wide one serves every - repository but ranks behind the domain bound to this one. The first custom - entry is what Cloudsmith would actually bind. + never stand in for another. """ - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") - records = [ - _domain("dl.acme.example.com", None, created_at="2020-01-01T00:00:00Z"), - _domain("dl-dev.acme.example.com", None, repository="dev"), - _domain("dl-prod.acme.example.com", None, repository="prod"), - ] + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") - result, _ = _invoke(runner, "--repo", "prod", custom_domains=records) + result, _ = _invoke(runner, "--repo", "prod", "--domain-type", "download") - document = json.loads(result.stdout)["domains"] - customs = [entry["host"] for entry in document if entry["type"] == "custom"] - assert customs == ["dl-prod.acme.example.com", "dl.acme.example.com"] - # A built-in host serves every repository, so it survives the filter. - assert "dl.cloudsmith.io" in {entry["host"] for entry in document} - - -def test_custom_domains_are_ranked_by_precedence_without_repo(runner, monkeypatch): - """Precedence orders the custom entries whether or not --repo is given. - - The first custom entry is the host Cloudsmith would bind, so sorting by host - without --repo would disagree with the credential path about which is active. - """ - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") - records = [ - _domain( - "alpha.acme.example.com", - BackendKind.DOCKER, - created_at="2020-01-01T00:00:00Z", - primary=False, - ), - _domain( - "zeta.acme.example.com", - BackendKind.DOCKER, - created_at="2024-01-01T00:00:00Z", - ), + assert _hosts(result) == [ + "dl-prod.acme.example.com", + "dl.acme.example.com", + "dl.cloudsmith.io", ] - result, _ = _invoke(runner, "--format", "docker", custom_domains=records) - - customs = [ - entry["host"] - for entry in json.loads(result.stdout)["domains"] - if entry["type"] == "custom" - ] - assert customs == ["zeta.acme.example.com", "alpha.acme.example.com"] - def test_default_shows_everything_with_no_pagination_metadata(runner, monkeypatch): """With no --page/--page-size given, the command defaults to --page-all. @@ -426,32 +394,21 @@ def test_default_shows_everything_with_no_pagination_metadata(runner, monkeypatc `list` commands backed by a live paginated endpoint, so there is no upside to hiding entries by default. """ - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") - records = [_domain(f"h{i}.acme.example.com", None) for i in range(5)] + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") - result, _ = _invoke(runner, custom_domains=records) + result, _ = _invoke(runner) document = json.loads(result.stdout) assert "meta" not in document - assert len([e for e in document["domains"] if e["type"] == "custom"]) == 5 + assert len(document["domains"]) == len(SERVER_DOMAINS) def test_page_size_selects_a_page_and_reports_pagination_metadata(runner, monkeypatch): """Passing --page-size switches the command into paged output.""" - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") - records = [ - _domain( - f"h{i}.acme.example.com", None, created_at=f"2020-01-{i + 1:02d}T00:00:00Z" - ) - for i in range(5) - ] - - unpaged, _ = _invoke(runner, custom_domains=records) - total = len(json.loads(unpaged.stdout)["domains"]) + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") + total = len(SERVER_DOMAINS) - result, _ = _invoke( - runner, "--page-size", "2", "--page", "1", custom_domains=records - ) + result, _ = _invoke(runner, "--page-size", "2", "--page", "1") document = json.loads(result.stdout) assert len(document["domains"]) == 2 @@ -464,42 +421,29 @@ def test_page_size_selects_a_page_and_reports_pagination_metadata(runner, monkey def test_second_page_continues_where_the_first_left_off(runner, monkeypatch): """Successive pages walk through the full list without gaps or overlap.""" - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") - records = [ - _domain( - f"h{i}.acme.example.com", None, created_at=f"2020-01-{i + 1:02d}T00:00:00Z" - ) - for i in range(5) - ] + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") - page_one, _ = _invoke( - runner, "--page-size", "2", "--page", "1", custom_domains=records - ) - page_two, _ = _invoke( - runner, "--page-size", "2", "--page", "2", custom_domains=records - ) + page_one, _ = _invoke(runner, "--page-size", "2", "--page", "1") + page_two, _ = _invoke(runner, "--page-size", "2", "--page", "2") - hosts_one = [e["host"] for e in json.loads(page_one.stdout)["domains"]] - hosts_two = [e["host"] for e in json.loads(page_two.stdout)["domains"]] - assert not set(hosts_one) & set(hosts_two) + assert not set(_hosts(page_one)) & set(_hosts(page_two)) assert json.loads(page_two.stdout)["meta"]["pagination"]["page"] == 2 def test_page_all_shows_everything_even_with_explicit_page_size(runner, monkeypatch): """--page-all still means "everything", overriding any --page-size given.""" - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") - records = [_domain(f"h{i}.acme.example.com", None) for i in range(5)] + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") - result, _ = _invoke(runner, "--page-all", custom_domains=records) + result, _ = _invoke(runner, "--page-all") document = json.loads(result.stdout) assert "meta" not in document - assert len([e for e in document["domains"] if e["type"] == "custom"]) == 5 + assert len(document["domains"]) == len(SERVER_DOMAINS) def test_page_and_page_all_are_mutually_exclusive(runner, monkeypatch): """Explicit --page together with --page-all is rejected, as elsewhere in the CLI.""" - monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + monkeypatch.setenv("CLOUDSMITH_WORKSPACE", "acme") with patch(_PATCH_TARGET, return_value=[]): result = runner.invoke( @@ -510,36 +454,3 @@ def test_page_and_page_all_are_mutually_exclusive(runner, monkeypatch): assert result.exit_code != 0 assert "--page-all" in result.output or "page-all" in result.output - - -def test_cached_domains_from_several_orgs_are_grouped_by_org(runner, tmp_path): - """Cross-org cached records are never interleaved into one ranking. - - A no-org listing spans whatever earlier runs cached, and ranking one - organisation's hosts against another's expresses a preference the server - never made. - """ - write_cache( - get_cache_path("acme"), - [_domain("mvn.acme.example.com", BackendKind.MAVEN, org="acme")], - ) - write_cache( - get_cache_path("bravo"), - [ - _domain( - "mvn.bravo.example.com", - BackendKind.MAVEN, - org="bravo", - created_at="2001-01-01T00:00:00Z", - ) - ], - ) - - result, _ = _invoke(runner, "--format", "maven") - - customs = [ - entry["host"] - for entry in json.loads(result.stdout)["domains"] - if entry["type"] == "custom" - ] - assert customs == ["mvn.acme.example.com", "mvn.bravo.example.com"] diff --git a/cloudsmith_cli/cli/tests/test_credential_helper_terraform.py b/cloudsmith_cli/cli/tests/test_credential_helper_terraform.py index c551ed49..07dc267a 100644 --- a/cloudsmith_cli/cli/tests/test_credential_helper_terraform.py +++ b/cloudsmith_cli/cli/tests/test_credential_helper_terraform.py @@ -20,6 +20,7 @@ import click.testing import pytest +from ...core.api.exceptions import ApiException from ...core.credentials.models import CredentialResult from ...credential_helpers.backends import BackendKind from ...credential_helpers.terraform.runtime import ( @@ -29,6 +30,7 @@ execute, get_token, ) +from ...credential_helpers.domains import domain_from_payload from ..commands.credential_helper.terraform import terraform CLOUDSMITH_HOST = "terraform.cloudsmith.io" @@ -114,12 +116,25 @@ def test_execute_get_returns_token_object_for_cloudsmith_host(credential): assert json.loads(stdout) == {"token": "acme/myrepo/k_abc"} +def _terraform_domain(host, scope): + return domain_from_payload( + {"domain_name": host, "domain_type": "NATIVE_TERRAFORM", "domain_scope": scope}, + "acme", + ) + + def test_execute_get_omits_org_from_token_for_a_custom_domain(credential): """A custom domain is already bound to one organisation, so its token must omit the org and be scoped as ``{repo}/{token}`` (not ``{org}/{repo}/...``).""" - with patch( - "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", - return_value=True, + with ( + patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", + return_value=True, + ), + patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.get_workspace_domains", + return_value=[_terraform_domain("tf.acme.com", "WORKSPACE")], + ), ): exit_code, stdout, stderr = execute( "get", @@ -133,6 +148,64 @@ def test_execute_get_omits_org_from_token_for_a_custom_domain(credential): assert json.loads(stdout) == {"token": "myrepo/k_abc"} +def test_execute_get_keeps_the_org_in_the_token_for_a_cloudsmith_owned_host( + credential, +): + """A Cloudsmith-owned host serves every Workspace, so its token names the org. + + The API, not the hostname suffix, says which hosts Cloudsmith owns. + """ + with ( + patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", + return_value=True, + ), + patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.get_workspace_domains", + return_value=[_terraform_domain("terraform.example.com", "GLOBAL")], + ), + ): + exit_code, stdout, stderr = execute( + "get", + "terraform.example.com", + credential=credential, + org="acme", + repo="myrepo", + ) + + assert (exit_code, stderr) == (0, None) + assert json.loads(stdout) == {"token": "acme/myrepo/k_abc"} + + +def test_execute_get_refuses_when_the_domain_scope_lookup_fails( + credential, tmp_path, monkeypatch +): + """A failed scope lookup is a refusal, never a token in the wrong format.""" + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.domains.get_default_config_path", + lambda: str(tmp_path), + ) + with ( + patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", + return_value=True, + ), + patch( + "cloudsmith_cli.core.api.workspaces.list_domains", + side_effect=ApiException(503), + ), + ): + exit_code, stdout, stderr = execute( + "get", + "terraform.example.com", + credential=credential, + org="acme", + repo="myrepo", + ) + + assert (exit_code, stdout, stderr) == (1, None, _REFUSAL_MESSAGE) + + def test_execute_get_custom_domain_does_not_require_an_org_in_the_token(credential): """The org is only needed to *resolve* a custom domain, never in its token: once matched, a custom-domain token is ``{repo}/{token}`` with no org.""" diff --git a/cloudsmith_cli/core/api/orgs.py b/cloudsmith_cli/core/api/orgs.py index 1b730401..7dea731b 100644 --- a/cloudsmith_cli/core/api/orgs.py +++ b/cloudsmith_cli/core/api/orgs.py @@ -13,21 +13,17 @@ def get_orgs_api(): return get_api_client(cloudsmith_api.OrgsApi) -def list_custom_domains(owner): - """List custom domains for an organization (raw SDK model dicts).""" +def list_orgs(page, page_size): + """List the organizations that the credential is a member of.""" client = get_orgs_api() with catch_raise_api_exception(): - ( - domains, - _, - headers, - ) = client.orgs_custom_domains_list_with_http_info( # pylint: disable=no-member - org=owner + orgs, _, headers = client.orgs_list_with_http_info( + page=page, page_size=page_size ) ratelimits.maybe_rate_limit(client, headers) - return [domain.to_dict() for domain in domains] + return [org.to_dict() for org in orgs] def list_vulnerability_policies(owner, page, page_size): diff --git a/cloudsmith_cli/core/api/workspaces.py b/cloudsmith_cli/core/api/workspaces.py new file mode 100644 index 00000000..3e83c612 --- /dev/null +++ b/cloudsmith_cli/core/api/workspaces.py @@ -0,0 +1,22 @@ +# Copyright 2026 Cloudsmith Ltd +"""API - Workspace (v2) endpoints.""" + +from ..credentials.models import CredentialResult +from . import v2 + + +def list_domains(workspace: str, *, credential: CredentialResult | None) -> list[dict]: + """List the domains that serve a Workspace, most-preferred first. + + The list holds Cloudsmith-owned and custom domains. The endpoint pages + with a ``Link`` header, so this follows each ``next`` link to the end. + """ + client = v2.get_client(credential) + response = v2.request(client, "GET", "workspaces", workspace, "domains") + domains = v2.response_json(response).get("results", []) + + while next_url := response.response.links.get("next", {}).get("url"): + response = v2.request_url(client, "GET", next_url) + domains.extend(v2.response_json(response).get("results", [])) + + return domains diff --git a/cloudsmith_cli/core/tests/test_workspaces.py b/cloudsmith_cli/core/tests/test_workspaces.py new file mode 100644 index 00000000..284cf3fe --- /dev/null +++ b/cloudsmith_cli/core/tests/test_workspaces.py @@ -0,0 +1,109 @@ +# Copyright 2026 Cloudsmith Ltd +"""Tests for the v2 Workspace API client.""" + +import json + +import httpretty +import httpretty.core +import pytest + +from .. import keyring +from ..api import workspaces +from ..api.exceptions import ApiException +from ..api.init import initialise_api +from ..credentials.models import CredentialResult + +API_HOST = "https://api.example.com" +DOMAINS_URL = f"{API_HOST}/v2/workspaces/acme/domains/" +CREDENTIAL = CredentialResult(api_key="test-api-key", source_name="test") + + +@pytest.fixture(autouse=True) +def _setup_api(monkeypatch): + """Initialise the SDK Configuration and stub keyring lookups.""" + monkeypatch.setattr(keyring, "get_access_token", lambda host, profile=None: None) + monkeypatch.setattr(keyring, "get_refresh_token", lambda host, profile=None: None) + monkeypatch.setattr( + keyring, "should_refresh_access_token", lambda host, profile=None: False + ) + monkeypatch.setattr( + httpretty.core.fakesock.socket, + "shutdown", + lambda self, how: None, + raising=False, + ) + monkeypatch.setattr("time.sleep", lambda _: None) + initialise_api(host=API_HOST) + + +def _domain(name, domain_type="DOWNLOAD", scope="GLOBAL", repository=None): + return { + "domain_name": name, + "domain_type": domain_type, + "domain_scope": scope, + "repository": repository, + } + + +@httpretty.activate(allow_net_connect=False) +def test_list_domains_asks_the_configured_api_host(): + """The lookup goes to whichever API host the CLI is configured for.""" + body = {"results": [_domain("dl.example.com")]} + httpretty.register_uri( + httpretty.GET, DOMAINS_URL, body=json.dumps(body), status=200 + ) + + assert workspaces.list_domains("acme", credential=CREDENTIAL) == body["results"] + assert httpretty.last_request().headers.get("X-Api-Key") == "test-api-key" + + +@httpretty.activate(allow_net_connect=False) +def test_list_domains_follows_the_link_header_to_every_page(): + """The endpoint pages with a Link header, so each next page is fetched.""" + first = [_domain("dl.example.com")] + second = [_domain("upload.example.com", "UPLOAD")] + + def _respond(request, uri, headers): + if "page=2" in uri: + return 200, headers, json.dumps({"results": second}) + headers["Link"] = f'<{DOMAINS_URL}?page=2>; rel="next"' + return 200, headers, json.dumps({"results": first}) + + httpretty.register_uri(httpretty.GET, DOMAINS_URL, body=_respond) + + assert workspaces.list_domains("acme", credential=CREDENTIAL) == first + second + assert len(httpretty.latest_requests()) == 2 + + +@httpretty.activate(allow_net_connect=False) +def test_list_domains_raises_the_api_status_on_failure(): + """A failure reaches the caller as an ApiException carrying the status.""" + httpretty.register_uri( + httpretty.GET, + DOMAINS_URL, + body=json.dumps({"detail": "Not found."}), + status=404, + ) + + with pytest.raises(ApiException) as raised: + workspaces.list_domains("acme", credential=CREDENTIAL) + + assert raised.value.status == 404 + + +@httpretty.activate(allow_net_connect=False) +def test_list_domains_does_not_send_credentials_to_another_origin(): + """A next link on another scheme or host does not get the API key.""" + body = {"results": [_domain("dl.example.com")]} + httpretty.register_uri( + httpretty.GET, + DOMAINS_URL, + body=json.dumps(body), + status=200, + adding_headers={"Link": '; rel="next"'}, + ) + + with pytest.raises(ApiException): + workspaces.list_domains("acme", credential=CREDENTIAL) + + assert len(httpretty.latest_requests()) == 1 diff --git a/cloudsmith_cli/credential_helpers/cargo/installer.py b/cloudsmith_cli/credential_helpers/cargo/installer.py index 32893505..5859eef2 100644 --- a/cloudsmith_cli/credential_helpers/cargo/installer.py +++ b/cloudsmith_cli/credential_helpers/cargo/installer.py @@ -28,7 +28,7 @@ from ...core.cache_utils import merge_config_file from ..backends import BackendKind from ..common import extract_hostname, path_warning -from ..custom_domains import get_format_domains +from ..domains import get_format_domains from ..launchers import ( is_on_path, is_windows, @@ -233,8 +233,8 @@ def install( actions: list[str] = [] - # Start with the default host plus any explicitly requested domains. - hosts: list[str] = [self.DEFAULT_HOST, *domains] + hosts: list[str] = [*domains] + discovered: list[str] = [] # --- Custom-domain auto-discovery (best-effort) --- if discover: @@ -265,7 +265,9 @@ def install( f"WARNING: custom-domain auto-discovery failed: {exc}" ) discovered = [] - new_hosts = [h for h in discovered if h not in hosts] + new_hosts = [ + h for h in discovered if h not in (*hosts, self.DEFAULT_HOST) + ] hosts.extend(discovered) actions.append( f"discovered {len(new_hosts)} new Cargo custom domain(s)" @@ -276,6 +278,11 @@ def install( " (no organization/credentials; pass --no-discover to silence)" ) + # The API lists the Cloudsmith-owned hosts with the custom domains, so + # the default host is needed only when discovery gave no hosts. + if not discovered: + hosts.insert(0, self.DEFAULT_HOST) + # De-duplicate while preserving order seen: set[str] = set() deduped: list[str] = [] diff --git a/cloudsmith_cli/credential_helpers/common.py b/cloudsmith_cli/credential_helpers/common.py index bc8dd54c..5db71994 100644 --- a/cloudsmith_cli/credential_helpers/common.py +++ b/cloudsmith_cli/credential_helpers/common.py @@ -8,7 +8,7 @@ import logging from pathlib import Path -from .custom_domains import get_custom_domains, get_format_domains +from .domains import get_format_domains, get_workspace_domains logger = logging.getLogger(__name__) @@ -135,7 +135,8 @@ def is_cloudsmith_domain( else: hosts = { d.host.lower() - for d in get_custom_domains(org, credential=credential, api_host=api_host) - if d.is_active + for d in get_workspace_domains( + org, credential=credential, api_host=api_host + ) } return hostname in hosts diff --git a/cloudsmith_cli/credential_helpers/custom_domains.py b/cloudsmith_cli/credential_helpers/custom_domains.py deleted file mode 100644 index 1cdb53bc..00000000 --- a/cloudsmith_cli/credential_helpers/custom_domains.py +++ /dev/null @@ -1,378 +0,0 @@ -# Copyright 2026 Cloudsmith Ltd -""" -Helper for discovering Cloudsmith custom domains. - -This module provides functions to fetch custom domains from the Cloudsmith API -for use in credential helpers. - -Results are cached on the filesystem. Custom domains change rarely, so the -cache is long-lived and ``--refresh`` bypasses it when one has just been added -or validated. Bump ``CACHE_FORMAT_VERSION`` when the cached record shape -changes, so an older document reads as a miss rather than as a record with the -new fields defaulted. -""" - -import json -import logging -import time -from dataclasses import dataclass -from pathlib import Path - -from ..cli.config import get_default_config_path -from ..core.cache_utils import atomic_write_json -from ..core.credentials.models import CredentialResult -from .default_domains import DomainType, domain_type_from_server - -logger = logging.getLogger(__name__) - -CACHE_TTL_SECONDS = 7 * 24 * 3600 - -CACHE_FORMAT_VERSION = 2 - - -@dataclass(frozen=True) -class CustomDomain: - """A structured Cloudsmith custom domain record.""" - - host: str - backend_kind: int | None - enabled: bool - validated: bool - org: str - domain_type: DomainType - repository: str | None = None - primary: bool = True - created_at: str | None = None - - @property - def is_active(self) -> bool: - """Whether this domain can serve traffic at all.""" - return self.enabled and self.validated - - def serves_repository(self, repository: str | None) -> bool: - """Whether this domain may be used for `repository`.""" - return not self.repository or self.repository == repository - - -def get_cache_dir() -> Path: - """ - Get the cache directory for custom domains, creating it where possible. - - A directory that cannot be created is not an error: every read path then - resolves to a miss, and :func:`write_cache` already degrades on ``OSError``. - """ - cache_dir = Path(get_default_config_path()) / "custom_domains_cache" - try: - cache_dir.mkdir(mode=0o700, parents=True, exist_ok=True) - except OSError as exc: - logger.debug("Could not create cache directory %s: %s", cache_dir, exc) - return cache_dir - - -def get_cache_path(org: str) -> Path: - """ - Get the cache file path for an organization's custom domains. - - Args: - org: Organization slug - - Returns: - Path to cache file - """ - cache_dir = get_cache_dir() - safe_org = "".join(c if c.isalnum() or c in "-_" else "_" for c in org) - return cache_dir / f"{safe_org}.json" - - -def is_cache_valid(cache_path: Path) -> bool: - """ - Check if a cache file exists and is still valid. - - Args: - cache_path: Path to cache file - - Returns: - bool: True if cache exists and hasn't expired - """ - if not cache_path.exists(): - return False - - try: - mtime = cache_path.stat().st_mtime - age = time.time() - mtime - return age < CACHE_TTL_SECONDS - except OSError: - return False - - -def _repository_slug(raw) -> str | None: - """Read a repository slug: the API nests it, the cache stores it flat.""" - if isinstance(raw, dict): - raw = raw.get("slug") - return raw or None - - -def _timestamp(raw) -> str | None: - """Read a creation time as a string: the SDK gives a datetime, the cache a str. - - ``json.dump`` cannot write a datetime, so it is normalised on the way in - rather than at every cache write. - """ - if raw is None or isinstance(raw, str): - return raw or None - return raw.isoformat() - - -def _record_from_payload(raw: dict, org: str) -> CustomDomain: - """Build a CustomDomain from an API or cache payload for *org*.""" - backend_kind = raw.get("backend_kind") - return CustomDomain( - host=raw["host"], - backend_kind=backend_kind, - enabled=bool(raw.get("enabled")), - validated=bool(raw.get("validated")), - org=org, - repository=_repository_slug(raw.get("repository")), - domain_type=domain_type_from_server(raw.get("domain_type")), - primary=bool(raw.get("primary")), - created_at=_timestamp(raw.get("created_at")), - ) - - -def _precedence_key(domain: CustomDomain, repository: str | None) -> tuple: - """Rank a candidate domain, lowest first. - - A domain bound to *repository* beats an organisation-wide one, primary - beats secondary, the oldest wins, and the host breaks any remaining tie. A - record with no ``created_at`` sorts after every dated one. - """ - bound_to_repository = repository is not None and domain.repository == repository - return ( - not bound_to_repository, - not domain.primary, - domain.created_at is None, - domain.created_at or "", - domain.host, - ) - - -def order_by_precedence( - domains: list[CustomDomain], repository: str | None = None -) -> list[CustomDomain]: - """Return *domains* ranked as Cloudsmith would choose among them. - - The first entry is the one that serves *repository*, so a caller listing - candidates shows the same answer the credential path would bind. - """ - return sorted(domains, key=lambda domain: _precedence_key(domain, repository)) - - -def read_cache(cache_path: Path) -> list[CustomDomain] | None: - """Return the cached custom domains at `cache_path`, or None for a miss. - - ``format_version`` is what separates a document this CLI can read from one - an older build wrote, so anything failing that check needs no further - handling of its shape - including the versionless documents that stored - domains as plain strings. - """ - if not is_cache_valid(cache_path): - return None - - try: - with open(cache_path, encoding="utf-8") as f: - data = json.load(f) - except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: - logger.debug("Failed to read cache %s: %s", cache_path, exc) - return None - - if not isinstance(data, dict) or data.get("format_version") != CACHE_FORMAT_VERSION: - logger.debug( - "Cache %s is not format version %s - treating as miss", - cache_path, - CACHE_FORMAT_VERSION, - ) - return None - - domains = data.get("domains") - if not isinstance(domains, list): - return None - - records = [ - _record_from_payload(d, d.get("org", "")) - for d in domains - if isinstance(d, dict) and d.get("host") - ] - logger.debug("Read %d domains from cache: %s", len(records), cache_path) - return records - - -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, - "domain_type": d.domain_type.value, - "primary": d.primary, - "created_at": d.created_at, - } - for d in domains - ], - "cached_at": time.time(), - } - try: - atomic_write_json(cache_path, data) - logger.debug("Wrote %d domains to cache: %s", len(domains), cache_path) - except OSError as exc: - logger.debug("Failed to write cache %s: %s", cache_path, exc) - - -def get_custom_domains( - org: str, - *, - credential: CredentialResult | None = None, - api_host: str | None = None, - refresh: bool = False, - strict: bool = False, - configure_api: bool = True, -) -> list[CustomDomain]: - """ - Fetch custom domains for a Cloudsmith organization. - - Results are cached on the filesystem for 7 days to avoid excessive API calls. - - Args: - org: Organization slug - credential: Optional resolved credential for authentication; it carries - its own auth scheme (X-Api-Key vs Authorization: Bearer) - api_host: Cloudsmith API host URL (including version). Taken from the SDK - 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. Use it for callers that show - results to a user, which must not render a typo'd org, a missing - permission or an unreachable API as "no custom domains". No failure - is ever cached, so a strict lookup can still be served from the - cache. - configure_api: When ``True`` (default), configure the SDK from - ``api_host`` and ``credential``. Pass ``False`` when the caller has - already done so, since the CLI's ``initialise_api`` also carries - proxy, TLS-verification, user-agent and header settings that this - narrower configuration would discard. - - Returns: - List of CustomDomain records. Empty if the org has no custom domains, - which the API answers as 200 with an empty array and is cached like any - other result, or if the call fails and ``strict`` is False. - - Note: - The API layer wraps transport failures (DNS/timeout/SSL) into - ``ApiException``, so every failure mode lands in the handler below. - Helpers degrade rather than break the tool wrapping them; pass - ``strict=True`` to fail loudly instead. - """ - cache_path = get_cache_path(org) - cached = None if refresh else read_cache(cache_path) - if cached is not None: - logger.debug("Using cached custom domains for %s", org) - return cached - - logger.debug("Fetching custom domains from API for %s", org) - - from ..core.api.exceptions import ApiException - from ..core.api.init import initialise_api - from ..core.api.orgs import list_custom_domains - - if configure_api: - initialise_api(host=api_host, credential=credential) - - try: - raw_domains = list_custom_domains(org) - except ApiException as exc: - if strict: - raise - if exc.status in (401, 403): - # Don't cache auth failures - might work later once authenticated. - logger.debug( - "Custom domains API requires auth - assuming no custom domains for %s", - org, - ) - return [] - - if exc.status == 404: - logger.debug("Organization %s not found or has no custom domains", org) - return [] - - logger.debug("Failed to fetch custom domains for %s: HTTP %s", org, exc.status) - return [] - - 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) - return records - - -def get_format_domains( - org: str, - backend_kind: int, - *, - credential: CredentialResult | None = None, - api_host: str | None = None, - refresh: bool = False, - configure_api: bool = True, -) -> list[str]: - """ - Return enabled and validated custom domain hostnames for a specific backend format. - - Args: - org: Organization slug - backend_kind: BackendKind int value (e.g. BackendKind.DOCKER == 6) - credential: Optional resolved credential for authentication - api_host: Cloudsmith API host URL - refresh: When ``True``, bypass the cache and fetch fresh data from the API. - configure_api: When ``True`` (default), configure the SDK from - ``api_host`` and ``credential``. Pass ``False`` when the caller has - already done so, for the reasons :func:`get_custom_domains` gives. - - Returns: - Hostnames that are usable and match the given backend_kind, in - precedence order - the host Cloudsmith would treat as active first. - """ - domains = get_custom_domains( - org, - credential=credential, - api_host=api_host, - refresh=refresh, - configure_api=configure_api, - ) - matching = [ - domain - for domain in domains - if domain.backend_kind == int(backend_kind) and domain.is_active - ] - matching.sort(key=lambda domain: _precedence_key(domain, None)) - return [domain.host for domain in matching] diff --git a/cloudsmith_cli/credential_helpers/default_domains.py b/cloudsmith_cli/credential_helpers/default_domains.py deleted file mode 100644 index b56b7cf3..00000000 --- a/cloudsmith_cli/credential_helpers/default_domains.py +++ /dev/null @@ -1,305 +0,0 @@ -# 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 dedicated deployment, serving packages from -its own domains, can replace the table via a ``[domains]`` section in -``config.ini``, mapping each hostname to the package format it serves, or to -``download``/``upload`` for the two endpoints no format names. - -A declared table replaces the built-ins wholesale rather than layering over -them: a host it omits has no default at all, so resolution raises instead of -handing back a ``*.cloudsmith.io`` host the operator never listed. - -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 format-less hosts are the download CDN and - the generic upload endpoint. - """ - - DOWNLOAD = "download" - UPLOAD = "upload" - NATIVE_API = "native_api" - - -SERVER_DOMAIN_TYPES: dict[int, DomainType] = { - 0: DomainType.DOWNLOAD, - 1: DomainType.UPLOAD, - 3: DomainType.NATIVE_API, -} - - -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 hand-maintained, 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_from_server(value: int | str) -> DomainType: - """Resolve a custom domain's type as the server reported it. - - The server classifies every domain and sends the answer as an integer; the - on-disk cache stores this enum's own string value. A type this CLI does not - know yet reads as a download host rather than breaking the listing. - """ - try: - if isinstance(value, str): - return DomainType(value) - return SERVER_DOMAIN_TYPES[int(value)] - except (KeyError, TypeError, ValueError): - logger.debug("Unrecognised domain type %r", value) - return DomainType.DOWNLOAD - - -@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 - - @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), -) - -RESERVED_LABELS: dict[str, DomainType] = { - "download": DomainType.DOWNLOAD, - "upload": DomainType.UPLOAD, -} - - -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 | None: - """Build a DefaultDomain from one ``[domains]`` ``host = label`` entry. - - A label names a package format, or one of the two endpoints no format - names: ``download`` for the CDN and ``upload`` for the generic upload - endpoint, neither of which has a BackendKind to name it by. - - The label is required. An entry naming nothing recognisable is skipped with - a warning rather than read as the CDN, since a typo'd format quietly - serving downloads is far harder to spot than a host that is simply absent. - """ - normalised = label.strip().lower() - - reserved_type = RESERVED_LABELS.get(normalised) - if reserved_type is not None: - return DefaultDomain(host=host, backend_kind=None, domain_type=reserved_type) - - backend_kind = _resolve_backend_kind(normalised) - if backend_kind is None: - logger.warning( - "Ignoring %s in the [domains] section: %r names no package format, " - "nor download or upload", - host, - label, - ) - return None - - return DefaultDomain(host=host, backend_kind=backend_kind) - - -def _config_candidates(*, trusted: bool) -> list[Path]: - """Return candidate config.ini paths from ConfigReader's search locations. - - The search path lists the current directory first, so absolute entries are - the trusted ones and directory-relative entries are not. - - Absolute filenames are skipped. Joining one onto a search directory would - yield that same path under every location, handing a trusted file to the - untrusted scan; :func:`_explicit_config_candidate` reads it instead. - """ - 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. - - ``ConfigReader.load_config`` prepends an absolute path here when - ``--config-file`` names a file; a directory goes to ``config_searchpath`` - instead and is covered by :func:`_config_candidates`. The entry is - explicit user intent, so it is trusted. - """ - 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 ahead of the search-path - candidates. Declaring a table is what counts, not existing: 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. - - A domain table is data, not a template, so interpolation is off and a ``%`` - in a format label is a literal. - """ - parser = configparser.ConfigParser(interpolation=None) - try: - read_files = parser.read(path, encoding="utf-8") - except (OSError, UnicodeDecodeError, 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 it declares no host. - - A section whose every entry was skipped reads as no table at all rather - than as an empty one, so a wholly unusable declaration falls back to the - built-in hosts instead of leaving the caller with nothing to publish to. - """ - section = _read_domains_section(path) - if section is None: - return None - - domains = [ - domain - for host, label in section.items() - if (domain := _domain_from_config_entry(host, label)) is not None - ] - return sorted(domains, key=lambda domain: domain.host) or None - - -def load_default_domains(config_path: Path | str | None = None) -> list[DefaultDomain]: - """Return the default domain table, honouring a trusted config override. - - `config_path` is read ahead of the trusted search locations rather than - instead of them, so a file holding only ``[default]`` api settings cannot - mask a deployment's ``[domains]`` table. The search locations are - ConfigReader's absolute ones, never the current working directory. A - malformed or unreadable file falls back to the built-in table. - """ - domains = None - if config_path is not None: - domains = _domains_from_config(Path(config_path)) - if domains is None: - 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 ignored. config.ini is searched in the current working - directory first, so a repository can ship one, and this table 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. - """ - return any( - _read_domains_section(candidate) is not None - for candidate in _config_candidates(trusted=False) - ) diff --git a/cloudsmith_cli/credential_helpers/docker/installer.py b/cloudsmith_cli/credential_helpers/docker/installer.py index 0382b9a9..54cc15e7 100644 --- a/cloudsmith_cli/credential_helpers/docker/installer.py +++ b/cloudsmith_cli/credential_helpers/docker/installer.py @@ -18,7 +18,7 @@ from ...core.cache_utils import merge_config_file from ..backends import BackendKind from ..common import path_warning -from ..custom_domains import get_format_domains +from ..domains import get_format_domains from ..launchers import ( is_on_path, is_windows, @@ -136,8 +136,8 @@ def install( actions: list[str] = [] - # Start with the default host plus any explicitly requested domains. - hosts: list[str] = [self.DEFAULT_HOST, *domains] + hosts: list[str] = [*domains] + discovered: list[str] = [] # --- Custom-domain auto-discovery (best-effort) --- if discover: @@ -168,7 +168,9 @@ def install( f"WARNING: custom-domain auto-discovery failed: {exc}" ) discovered = [] - new_hosts = [h for h in discovered if h not in hosts] + new_hosts = [ + h for h in discovered if h not in (*hosts, self.DEFAULT_HOST) + ] hosts.extend(discovered) actions.append( f"discovered {len(new_hosts)} new Docker custom domain(s)" @@ -179,6 +181,11 @@ def install( " (no organization/credentials; pass --no-discover to silence)" ) + # The API lists the Cloudsmith-owned hosts with the custom domains, so + # the default host is needed only when discovery gave no hosts. + if not discovered: + hosts.insert(0, self.DEFAULT_HOST) + # De-duplicate while preserving order seen: set[str] = set() deduped: list[str] = [] diff --git a/cloudsmith_cli/credential_helpers/domains.py b/cloudsmith_cli/credential_helpers/domains.py new file mode 100644 index 00000000..2b8b3ac0 --- /dev/null +++ b/cloudsmith_cli/credential_helpers/domains.py @@ -0,0 +1,399 @@ +# Copyright 2026 Cloudsmith Ltd +""" +Helper for discovering the domains that serve a Cloudsmith Workspace. + +The v2 API lists both the Cloudsmith-owned domains and a Workspace's own custom +domains, already in preference order. The CLI therefore holds no table of +Cloudsmith hostnames: whichever API host it is configured for supplies its own. + +Results are cached on the filesystem, one directory per API host, so that a +Workspace slug on one Cloudsmith never answers for the same slug on another. +Domains change rarely, so the cache is long-lived and ``--refresh`` bypasses +it when one has just been added or validated. The cache keeps the server's +records as sent; bump ``CACHE_FORMAT_VERSION`` when that shape changes. +""" + +import hashlib +import json +import logging +import sys +import time +from dataclasses import dataclass +from enum import Enum +from pathlib import Path +from urllib.parse import urlparse + +from ..cli.config import get_default_config_path +from ..core.cache_utils import atomic_write_json +from ..core.credentials.models import CredentialContext, CredentialResult +from .backends import BackendKind + +logger = logging.getLogger(__name__) + +CACHE_TTL_SECONDS = 7 * 24 * 3600 + +CACHE_FORMAT_VERSION = 3 + +NATIVE_TYPE_PREFIX = "NATIVE_" + +SERVER_FORMAT_NAMES = {"HELM_OCI": "HELM"} + + +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 format-less hosts are the download CDN, the + generic upload endpoint and the Cloudsmith API itself. + """ + + DOWNLOAD = "download" + UPLOAD = "upload" + CLOUDSMITH_API = "cloudsmith_api" + NATIVE_API = "native_api" + + +class DomainScope(str, Enum): + """Whom a domain serves: every Workspace, one Workspace, or one repository.""" + + GLOBAL = "global" + WORKSPACE = "workspace" + REPOSITORY = "repository" + + +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 hand-maintained, 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" + + +@dataclass(frozen=True) +class WorkspaceDomain: + """A domain that serves a Workspace, as the server reported it.""" + + host: str + domain_type: DomainType + backend_kind: int | None + scope: DomainScope + workspace: str | None + repository: str | None = None + + @property + def is_custom(self) -> bool: + """Whether a Workspace owns this domain, rather than Cloudsmith.""" + return self.scope is not DomainScope.GLOBAL + + @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) + + def serves_repository(self, repository: str | None) -> bool: + """Whether this domain may be used for `repository`.""" + return not self.repository or self.repository == repository + + +def _classify(server_type) -> tuple[DomainType, int | None] | None: + """Return the purpose and format a server domain type names, or None. + + The server adds domain types over time. A type this CLI does not know has + no safe reading, so it resolves to None and the caller skips the domain. + """ + if not isinstance(server_type, str): + return None + + if server_type.startswith(NATIVE_TYPE_PREFIX): + name = server_type.removeprefix(NATIVE_TYPE_PREFIX) + backend_kind = BackendKind.__members__.get(SERVER_FORMAT_NAMES.get(name, name)) + if backend_kind is None: + return None + return DomainType.NATIVE_API, int(backend_kind) + + try: + return DomainType(server_type.lower()), None + except ValueError: + return None + + +def _scope(server_scope) -> DomainScope | None: + try: + return DomainScope(str(server_scope).lower()) + except ValueError: + return None + + +def domain_from_payload(raw: dict, workspace: str) -> WorkspaceDomain | None: + """Build a WorkspaceDomain from one v2 API record, or None to skip it.""" + classified = _classify(raw.get("domain_type")) + scope = _scope(raw.get("domain_scope")) + if classified is None or scope is None or not raw.get("domain_name"): + logger.debug("Skipping domain record %r", raw) + return None + + domain_type, backend_kind = classified + return WorkspaceDomain( + host=raw["domain_name"], + domain_type=domain_type, + backend_kind=backend_kind, + scope=scope, + workspace=None if scope is DomainScope.GLOBAL else workspace, + repository=raw.get("repository") or None, + ) + + +def _domains_from_payload(payload: list, workspace: str) -> list[WorkspaceDomain]: + return [ + domain + for raw in payload + if isinstance(raw, dict) + and (domain := domain_from_payload(raw, workspace)) is not None + ] + + +def _safe_name(name: str) -> str: + return "".join(c if c.isalnum() or c in "-_" else "_" for c in name) + + +def _default_api_host() -> str: + """Return the API host that the SDK calls when no host is given. + + The SDK is read only if it is already loaded, so a cache hit does not + import it. + """ + sdk = sys.modules.get("cloudsmith_api") + return sdk.Configuration().host if sdk else CredentialContext.api_host + + +def _cache_dir_name(api_host: str | None) -> str: + """Name the cache directory after the complete API base URL. + + The scheme, port and path are part of the name, so two Cloudsmith + deployments on one hostname never share a cache. + """ + api_base = (api_host or _default_api_host()).rstrip("/") + digest = hashlib.sha256(api_base.encode("utf-8")).hexdigest()[:16] + return f"{_safe_name(urlparse(api_base).hostname or 'default')}-{digest}" + + +def get_cache_dir(api_host: str | None = None) -> Path: + """ + Get the domain cache directory for an API host, creating it where possible. + + ``None`` names the SDK's default API host. A directory that cannot be + created is not an error: every read path then resolves to a miss, and + :func:`write_cache` already degrades on ``OSError``. + """ + cache_dir = ( + Path(get_default_config_path()) / "domains_cache" / _cache_dir_name(api_host) + ) + try: + cache_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + except OSError as exc: + logger.debug("Could not create cache directory %s: %s", cache_dir, exc) + return cache_dir + + +def get_cache_path(workspace: str, api_host: str | None = None) -> Path: + """Get the cache file path for a Workspace's domains on an API host.""" + return get_cache_dir(api_host) / f"{_safe_name(workspace)}.json" + + +def is_cache_valid(cache_path: Path) -> bool: + """Check if a cache file exists and has not expired.""" + if not cache_path.exists(): + return False + + try: + mtime = cache_path.stat().st_mtime + age = time.time() - mtime + return age < CACHE_TTL_SECONDS + except OSError: + return False + + +def read_cache(cache_path: Path) -> list[WorkspaceDomain] | None: + """Return the cached domains at `cache_path`, or None for a miss. + + ``format_version`` is what separates a document this CLI can read from one + an older build wrote, so anything failing that check needs no further + handling of its shape. + """ + if not is_cache_valid(cache_path): + return None + + try: + with open(cache_path, encoding="utf-8") as f: + data = json.load(f) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + logger.debug("Failed to read cache %s: %s", cache_path, exc) + return None + + if not isinstance(data, dict) or data.get("format_version") != CACHE_FORMAT_VERSION: + logger.debug( + "Cache %s is not format version %s - treating as miss", + cache_path, + CACHE_FORMAT_VERSION, + ) + return None + + domains = data.get("domains") + if not isinstance(domains, list): + return None + + records = _domains_from_payload(domains, data.get("workspace", "")) + logger.debug("Read %d domains from cache: %s", len(records), cache_path) + return records + + +def read_all_cached_domains(api_host: str | None = None) -> list[WorkspaceDomain]: + """Return every domain in a currently-valid cache entry for an API host. + + Lets a run with no configured Workspace 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[WorkspaceDomain] = [] + for cache_path in sorted(get_cache_dir(api_host).glob("*.json")): + cached = read_cache(cache_path) + if cached: + records.extend(cached) + return records + + +def write_cache(cache_path: Path, workspace: str, payload: list[dict]) -> None: + """Write a Workspace's domain records, as the server sent them, to the cache.""" + data = { + "format_version": CACHE_FORMAT_VERSION, + "workspace": workspace, + "domains": payload, + "cached_at": time.time(), + } + try: + atomic_write_json(cache_path, data) + logger.debug("Wrote %d domains to cache: %s", len(payload), cache_path) + except OSError as exc: + logger.debug("Failed to write cache %s: %s", cache_path, exc) + + +def get_workspace_domains( + workspace: str, + *, + credential: CredentialResult | None = None, + api_host: str | None = None, + refresh: bool = False, + strict: bool = False, + configure_api: bool = True, +) -> list[WorkspaceDomain]: + """ + Fetch the domains that serve a Cloudsmith Workspace, most-preferred first. + + Results are cached on the filesystem for 7 days to avoid excessive API calls. + + Args: + workspace: Workspace slug + credential: Optional resolved credential for authentication; it carries + its own auth scheme (X-Api-Key vs Authorization: Bearer) + api_host: Cloudsmith API host URL. Taken from the SDK 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. Use it for callers that show + results to a user, which must not render a typo'd Workspace, a + missing permission or an unreachable API as "no domains". No + failure is ever cached, so a strict lookup can still be served from + the cache. + configure_api: When ``True`` (default), configure the SDK from + ``api_host`` and ``credential``. Pass ``False`` when the caller has + already done so, since the CLI's ``initialise_api`` also carries + proxy, TLS-verification, user-agent and header settings that this + narrower configuration would discard. + + Returns: + List of WorkspaceDomain records in the server's preference order: + repository domains, then Workspace domains, then Cloudsmith-owned + domains. Empty if the call fails and ``strict`` is False. + + Note: + The API layer wraps transport failures (DNS/timeout/SSL) into + ``ApiException``, so every failure mode lands in the handler below. + Helpers degrade rather than break the tool wrapping them; pass + ``strict=True`` to fail loudly instead. + """ + cache_path = get_cache_path(workspace, api_host) + cached = None if refresh else read_cache(cache_path) + if cached is not None: + logger.debug("Using cached domains for %s", workspace) + return cached + + logger.debug("Fetching domains from API for %s", workspace) + + from ..core.api.exceptions import ApiException + from ..core.api.init import initialise_api + from ..core.api.workspaces import list_domains + + if configure_api: + initialise_api(host=api_host, credential=credential) + + try: + payload = list_domains(workspace, credential=credential) + except ApiException as exc: + if strict: + raise + logger.debug("Failed to fetch domains for %s: HTTP %s", workspace, exc.status) + return [] + + records = _domains_from_payload(payload, workspace) + + logger.debug("Fetched %d domains for %s", len(records), workspace) + write_cache(cache_path, workspace, payload) + return records + + +def get_format_domains( + workspace: str, + backend_kind: int, + *, + credential: CredentialResult | None = None, + api_host: str | None = None, + refresh: bool = False, + configure_api: bool = True, +) -> list[str]: + """ + Return the hostnames that serve a package format for a Workspace. + + Args: + workspace: Workspace slug + backend_kind: BackendKind int value (e.g. BackendKind.DOCKER == 6) + credential: Optional resolved credential for authentication + api_host: Cloudsmith API host URL + refresh: When ``True``, bypass the cache and fetch fresh data from the API. + configure_api: When ``True`` (default), configure the SDK from + ``api_host`` and ``credential``. Pass ``False`` when the caller has + already done so, for the reasons :func:`get_workspace_domains` gives. + + Returns: + Hostnames that speak the given format's native protocol, in the + server's preference order. Cloudsmith-owned hosts come last. + """ + domains = get_workspace_domains( + workspace, + credential=credential, + api_host=api_host, + refresh=refresh, + configure_api=configure_api, + ) + return [ + domain.host for domain in domains if domain.backend_kind == int(backend_kind) + ] diff --git a/cloudsmith_cli/credential_helpers/pnpm/installer.py b/cloudsmith_cli/credential_helpers/pnpm/installer.py index f45cea8d..edc8f606 100644 --- a/cloudsmith_cli/credential_helpers/pnpm/installer.py +++ b/cloudsmith_cli/credential_helpers/pnpm/installer.py @@ -5,7 +5,7 @@ from pathlib import Path from cloudsmith_cli.credential_helpers.backends import BackendKind -from cloudsmith_cli.credential_helpers.custom_domains import get_format_domains +from cloudsmith_cli.credential_helpers.domains import get_format_domains from cloudsmith_cli.credential_helpers.generic import PartialInstallError from cloudsmith_cli.credential_helpers.launchers import ( remove_launcher, @@ -104,8 +104,8 @@ def install( actions: list[str] = [] - # Start with the default host plus any explicitly requested domains. - hosts: list[str] = [self.DEFAULT_HOST, *domains] + hosts: list[str] = [*domains] + discovered: list[str] = [] if discover: if dry_run: @@ -127,7 +127,9 @@ def install( f"WARNING: custom-domain auto-discovery failed: {exc}" ) discovered = [] - new_hosts = [h for h in discovered if h not in hosts] + new_hosts = [ + h for h in discovered if h not in (*hosts, self.DEFAULT_HOST) + ] hosts.extend(discovered) actions.append(f"discovered {len(new_hosts)} new pnpm custom domain(s)") else: @@ -136,6 +138,11 @@ def install( " (no organization/credentials; pass --no-discover to silence)" ) + # The API lists the Cloudsmith-owned hosts with the custom domains, so + # the default host is needed only when discovery gave no hosts. + if not discovered: + hosts.insert(0, self.DEFAULT_HOST) + # De-duplicate while preserving order seen: set[str] = set() deduped: list[str] = [] diff --git a/cloudsmith_cli/credential_helpers/terraform/runtime.py b/cloudsmith_cli/credential_helpers/terraform/runtime.py index d10ac82b..926796c2 100644 --- a/cloudsmith_cli/credential_helpers/terraform/runtime.py +++ b/cloudsmith_cli/credential_helpers/terraform/runtime.py @@ -32,7 +32,12 @@ import logging from ..backends import BackendKind -from ..common import is_cloudsmith_domain, is_standard_cloudsmith_domain +from ..common import ( + extract_hostname, + is_cloudsmith_domain, + is_standard_cloudsmith_domain, +) +from ..domains import get_workspace_domains logger = logging.getLogger(__name__) @@ -91,6 +96,20 @@ def get_token(hostname, credential=None, api_host=None, org=None): return credential.api_key +def _is_cloudsmith_owned(hostname, credential, api_host, org) -> bool: + """Whether the API lists `hostname` as a Cloudsmith-owned Terraform host.""" + if not org: + return False + host = extract_hostname(hostname) + return any( + domain.host.lower() == host and not domain.is_custom + for domain in get_workspace_domains( + org, credential=credential, api_host=api_host, strict=True + ) + if domain.backend_kind == BackendKind.TERRAFORM + ) + + def _execute_get( hostname, credential, api_host, org, repo ) -> tuple[int, str | None, str | None]: @@ -142,7 +161,7 @@ def _execute_get( # Scope the token to a single repository. Standard domains also carry the # org ("{org}/{repo}/{token}"); custom domains are already org-bound, so # they omit it ("{repo}/{token}"). - if standard_domain: + if standard_domain or _is_cloudsmith_owned(hostname, credential, api_host, org): scoped_token = f"{org}/{repo}/{token}" else: scoped_token = f"{repo}/{token}"