diff --git a/cloudsmith_cli/conftest.py b/cloudsmith_cli/conftest.py index b3341fc5..d99ca9b9 100644 --- a/cloudsmith_cli/conftest.py +++ b/cloudsmith_cli/conftest.py @@ -1,3 +1,5 @@ +import json + import cloudsmith_api import pytest @@ -105,3 +107,74 @@ def usage_state_dir(monkeypatch, tmp_path_factory): path = tmp_path_factory.mktemp("usage-state") monkeypatch.setattr("cloudsmith_cli.core.usage.state.config_dir", lambda: str(path)) return path + + +@pytest.fixture(autouse=True) +def blocked_usage_posts(monkeypatch): + """Never send a real usage event from the test suite. + + Removes the developer's usage settings (AGENTS.md tells them to export a + test key) and replaces the POST with one that records the call and fails. + The sender swallows that failure in its thread, so the fixture fails the + test at teardown if any call was recorded. A test that expects a refused + call asserts on the returned list and clears it. Tests that exercise + sending opt back in with ``usage_events``. + """ + from cloudsmith_cli.core.usage import sender, settings + + calls = [] + + def refuse(*args, **kwargs): # pylint: disable=unused-argument + calls.append(args) + raise AssertionError("unit tests must not send usage telemetry") + + for name in ( + settings.KEY_ENV, + settings.HOST_ENV, + settings.MODE_ENV, + settings.INSTALL_ID_ENV, + settings.INTERNAL_ENV, + settings.SAMPLE_RATE_ENV, + ): + monkeypatch.delenv(name, raising=False) + monkeypatch.setattr(sender, "_post", refuse) + yield calls + if calls: + pytest.fail(f"unit test attempted {len(calls)} usage telemetry send(s)") + + +@pytest.fixture() +def usage_events(monkeypatch, telemetry_config_dir, blocked_usage_posts): # pylint: disable=unused-argument + """Enable usage telemetry into a list instead of the network.""" + from cloudsmith_cli.core import telemetry_common + from cloudsmith_cli.core.usage import environment, sender + + events = [] + for name in telemetry.OPT_OUT_ENVS: + monkeypatch.delenv(name, raising=False) + ci_names = [name for _, names in environment._CI_PROVIDERS for name in names] + [ + "KUBERNETES_SERVICE_HOST" + ] + for name in ci_names + [ + "CLOUDSMITH_TELEMETRY", + "CLOUDSMITH_TELEMETRY_CI_SAMPLE_RATE", + "CLOUDSMITH_TELEMETRY_INSTALL_ID", + "CLOUDSMITH_TELEMETRY_POSTHOG_HOST", + "CLOUDSMITH_TELEMETRY_INTERNAL", + "CLOUDSMITH_API_HOST", + "CLOUDSMITH_SSL_VERIFY", + "CLOUDSMITH_WITHOUT_API_SSL_VERIFY", + ]: + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + # Later tests that call the real main() must never run the update check's + # network fetch. + monkeypatch.setenv("CLOUDSMITH_NO_UPDATE_CHECK", "1") + monkeypatch.setattr(environment, "is_container", lambda: False) + monkeypatch.setattr( + sender, + "_post", + lambda session, url, body, timeout: events.append(json.loads(body)) or True, + ) + telemetry_common.WaitBudget.shared().reset(1.0) + return events diff --git a/cloudsmith_cli/core/tests/test_usage_identity.py b/cloudsmith_cli/core/tests/test_usage_identity.py index 10d3973c..15807e2a 100644 --- a/cloudsmith_cli/core/tests/test_usage_identity.py +++ b/cloudsmith_cli/core/tests/test_usage_identity.py @@ -68,6 +68,24 @@ def test_reset_raises_when_delete_fails(usage_state_dir, monkeypatch): identity.reset() +def test_forget_deletes_the_id_and_keeps_other_state(usage_state_dir): + identity.resolve(ephemeral=False, install_id_enabled=True) + state.write({"notice_version": 1}) + identity.forget() + assert state.read() == {"notice_version": 1} + + +def test_forget_without_an_id_writes_nothing(usage_state_dir): + identity.forget() + assert not usage_state_dir.joinpath(state.STATE_FILE_NAME).exists() + + +def test_forget_never_raises(usage_state_dir, monkeypatch): + identity.resolve(ephemeral=False, install_id_enabled=True) + monkeypatch.setattr(state, "delete", lambda: False) + identity.forget() + + def test_reset_raises_when_restoring_other_state_fails(usage_state_dir, monkeypatch): state.write({"install_id": "x", "notice_version": 1}) monkeypatch.setattr(state, "write", lambda updates: False) diff --git a/cloudsmith_cli/core/tests/test_usage_run.py b/cloudsmith_cli/core/tests/test_usage_run.py new file mode 100644 index 00000000..5268dbed --- /dev/null +++ b/cloudsmith_cli/core/tests/test_usage_run.py @@ -0,0 +1,441 @@ +# pylint: disable=unused-argument +import pytest + +from cloudsmith_cli.core.usage import notice, run, state +from cloudsmith_cli.core.usage.event import RunFacts + + +def facts(**kw): + return RunFacts(command=kw.pop("command", "whoami"), **kw) + + +@pytest.fixture() +def interactive(): + """A human at a terminal whose notice was already shown.""" + state.write({"notice_version": notice.NOTICE_VERSION}) + + +def test_sends_one_event(usage_events, interactive): + run.finish(facts()) + assert len(usage_events) == 1 + assert usage_events[0]["properties"]["command"] == "whoami" + + +@pytest.mark.parametrize( + "command", + [ + "credential-helper docker", + "credential-helper terraform", + "mcp start", + "telemetry reset", + ], +) +def test_excluded_commands_send_nothing_and_touch_no_state(usage_events, command): + run.finish(facts(command=command)) + assert usage_events == [] and state.read() == {} + + +def test_opt_out_sends_nothing(usage_events, interactive, monkeypatch): + monkeypatch.setenv("DO_NOT_TRACK", "1") + run.finish(facts()) + assert usage_events == [] + + +def test_custom_api_host_sends_nothing(usage_events, interactive): + run.finish(facts(api_host="https://artifacts.example.com")) + assert usage_events == [] + + +def test_ssl_verify_disabled_sends_nothing(usage_events, interactive): + run.finish(facts(ssl_verify=False)) + assert usage_events == [] + + +def test_dev_build_without_override_sends_nothing( + usage_events, interactive, monkeypatch +): + monkeypatch.delenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY") + run.finish(facts()) + assert usage_events == [] + + +def test_first_interactive_run_shows_notice_and_sends_nothing(usage_events, capsys): + run.finish(facts(stdout_tty=True, stdin_tty=True)) + assert usage_events == [] + assert "usage data" in capsys.readouterr().err + run.finish(facts(stdout_tty=True, stdin_tty=True)) + assert len(usage_events) == 1 + + +def test_ci_sends_from_first_run_without_notice(usage_events, monkeypatch, capsys): + monkeypatch.setenv("GITHUB_ACTIONS", "true") + # TTYs, so only CI (not a missing terminal) keeps the notice away. + run.finish(facts(stdout_tty=True, stdin_tty=True)) + assert len(usage_events) == 1 + props = usage_events[0]["properties"] + assert props["ci_provider"] == "github_actions" + assert props["$process_person_profile"] is False + assert capsys.readouterr().err == "" + assert state.read() == {} + + +def test_ci_sampling_zero_sends_nothing(usage_events, monkeypatch): + monkeypatch.setenv("GITHUB_ACTIONS", "true") + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_CI_SAMPLE_RATE", "0") + run.finish(facts()) + assert usage_events == [] + + +def test_ci_sample_rate_never_samples_runs_outside_ci(usage_events, monkeypatch): + # Unwritable state makes the run ephemeral, but it is not CI. + monkeypatch.setattr(state, "writable", lambda: False) + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_CI_SAMPLE_RATE", "0") + run.finish(facts()) + assert len(usage_events) == 1 + props = usage_events[0]["properties"] + assert props["sample_rate"] == 100 and props["ci"] is False + + +def test_container_runs_are_not_ci_sampled(usage_events, monkeypatch): + from cloudsmith_cli.core.usage import environment + + monkeypatch.setattr(environment, "is_container", lambda: True) + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_CI_SAMPLE_RATE", "0") + run.finish(facts()) + assert len(usage_events) == 1 + + +def test_resolved_ssl_strictness_reaches_the_sender( + usage_events, interactive, monkeypatch +): + from cloudsmith_cli.core.usage import sender + + seen = [] + monkeypatch.setattr( + sender, "send", lambda payload, strict=None: seen.append(strict) or True + ) + run.finish(facts(ssl_verify_strict=False)) + assert seen == [False] + + +def test_failed_send_backs_off_for_a_day(usage_events, interactive, monkeypatch): + from cloudsmith_cli.core.usage import sender + + monkeypatch.setattr(sender, "send", lambda payload, *_: False) + run.finish(facts()) + assert state.read().get("last_failure_at") + monkeypatch.setattr( + sender, "send", lambda payload, *_: usage_events.append(payload) or True + ) + run.finish(facts()) + assert usage_events == [] + + +def test_skipped_send_does_not_back_off(usage_events, interactive, monkeypatch): + from cloudsmith_cli.core.usage import sender + + monkeypatch.setattr(sender, "send", lambda payload, *_: None) + run.finish(facts()) + assert "last_failure_at" not in state.read() + + +def test_log_mode_prints_and_sends_nothing( + usage_events, interactive, monkeypatch, capsys +): + monkeypatch.setenv("CLOUDSMITH_TELEMETRY", "log") + run.finish(facts()) + assert usage_events == [] + assert "would send" in capsys.readouterr().err + + +def test_nested_run_sends_nothing(usage_events, interactive): + run.finish(facts(nested=True)) + assert usage_events == [] + + +def test_finish_never_raises(usage_events, interactive, monkeypatch): + from cloudsmith_cli.core.usage import event + + monkeypatch.setattr(event, "build_event", lambda **kw: 1 / 0) + run.finish(facts()) # must not raise + + +def test_unwritable_state_is_ephemeral(usage_events, monkeypatch, capsys): + monkeypatch.setattr(state, "writable", lambda: False) + run.finish(facts(stdout_tty=True, stdin_tty=True)) + assert len(usage_events) == 1 + assert usage_events[0]["properties"]["$process_person_profile"] is False + assert capsys.readouterr().err == "" + assert state.read() == {} + + +def test_dev_build_gate_alone_blocks_log_mode(usage_events, monkeypatch, capsys): + monkeypatch.delenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY") + monkeypatch.setenv("CLOUDSMITH_TELEMETRY", "log") + run.finish(facts()) + assert usage_events == [] + assert "would send" not in capsys.readouterr().err + + +def test_failed_ephemeral_send_does_no_state_io(usage_events, monkeypatch): + from cloudsmith_cli.core.usage import sender + + monkeypatch.setenv("GITHUB_ACTIONS", "true") + monkeypatch.setattr(sender, "send", lambda payload, *_: False) + run.finish(facts()) + assert state.read() == {} + + +def test_no_telemetry_env_sends_nothing(usage_events, interactive, monkeypatch): + monkeypatch.setenv("CLOUDSMITH_NO_TELEMETRY", "1") + run.finish(facts()) + assert usage_events == [] + + +def test_config_file_opt_out_sends_nothing( + usage_events, interactive, telemetry_config_dir +): + (telemetry_config_dir / "config.ini").write_text("[default]\ntelemetry = false\n") + run.finish(facts()) + assert usage_events == [] + + +def test_credential_helper_install_still_sends(usage_events, interactive): + run.finish(facts(command="credential-helper install")) + assert len(usage_events) == 1 + + +class _Loader: + """A deferred-sources loader that records how often it ran.""" + + def __init__(self, result=None, error=None): + self.calls = 0 + self.result = result or {} + self.error = error + + def __call__(self): + self.calls += 1 + if self.error: + raise self.error + return self.result + + +def test_excluded_command_never_loads_deferred_sources(usage_events): + loader = _Loader({"workspace": "config_file"}) + run.finish(facts(command="credential-helper docker", deferred_sources=loader)) + assert loader.calls == 0 and usage_events == [] + + +def test_unresolved_config_sends_nothing_and_reads_nothing(usage_events, interactive): + # A typed --config-file or --profile that no command loaded. + loader = _Loader({"workspace": "config_file"}) + run.finish(facts(config_unresolved=True, deferred_sources=loader)) + assert loader.calls == 0 and usage_events == [] + + +def test_opted_out_run_never_loads_deferred_sources( + usage_events, interactive, monkeypatch +): + monkeypatch.setenv("DO_NOT_TRACK", "1") + loader = _Loader({"workspace": "config_file"}) + run.finish(facts(deferred_sources=loader)) + assert loader.calls == 0 and usage_events == [] + + +def test_deferred_sources_load_once_and_reach_the_event(usage_events, interactive): + loader = _Loader({"workspace": "config_file"}) + run.finish(facts(deferred_sources=loader)) + assert loader.calls == 1 + assert usage_events[0]["properties"]["source_workspace"] == "config_file" + + +def test_click_source_wins_over_the_loaded_one(usage_events, interactive): + loader = _Loader({"workspace": "config_file", "api_host": "config_file"}) + run.finish(facts(setting_sources={"workspace": "flag"}, deferred_sources=loader)) + props = usage_events[0]["properties"] + assert props["source_workspace"] == "flag" + assert props["source_api_host"] == "config_file" + + +def test_failing_loader_still_sends_the_event(usage_events, interactive): + run.finish( + facts( + setting_sources={"workspace": "env"}, + deferred_sources=_Loader(error=RuntimeError("boom")), + ) + ) + assert len(usage_events) == 1 + assert usage_events[0]["properties"]["source_workspace"] == "env" + + +@pytest.fixture() +def config_reads(monkeypatch): + """Count config.ini reads; no command loaded config (no options yet).""" + from cloudsmith_cli.cli import config + + monkeypatch.delattr(config.OPTIONS, "value", raising=False) + calls = [] + real = config.ConfigReader.load_config.__func__ + + def counting(cls, opts, path=None, profile=None): + calls.append(profile) + return real(cls, opts, path=path, profile=profile) + + monkeypatch.setattr(config.ConfigReader, "load_config", classmethod(counting)) + return calls + + +def test_config_file_is_read_once_per_finish( + usage_events, interactive, telemetry_config_dir, config_reads +): + (telemetry_config_dir / "config.ini").write_text( + "[default]\napi_host = https://api.cloudsmith.io\nssl_verify = true\n" + ) + run.finish(facts()) + assert len(usage_events) == 1 + assert len(config_reads) == 1 + + +@pytest.mark.parametrize( + "setting", ["api_host = https://artifacts.example.com", "ssl_verify = false"] +) +def test_config_file_gate_blocks_with_one_read( + usage_events, interactive, telemetry_config_dir, config_reads, setting +): + (telemetry_config_dir / "config.ini").write_text(f"[default]\n{setting}\n") + run.finish(facts()) + assert usage_events == [] + assert len(config_reads) == 1 + + +def test_resolved_host_wins_over_the_config_file( + usage_events, interactive, telemetry_config_dir, config_reads +): + (telemetry_config_dir / "config.ini").write_text( + "[default]\napi_host = https://artifacts.example.com\n" + ) + run.finish(facts(api_host="https://api.cloudsmith.io")) + assert len(usage_events) == 1 + + +def test_loaded_config_skips_the_config_file_gate( + usage_events, interactive, telemetry_config_dir, config_reads, monkeypatch +): + # A command that loaded config already put config.ini's host and SSL + # setting into the facts, through opts. + from cloudsmith_cli.cli import config + + monkeypatch.setattr(config.OPTIONS, "value", config.Options(), raising=False) + (telemetry_config_dir / "config.ini").write_text("[default]\nssl_verify = false\n") + run.finish(facts()) + assert len(usage_events) == 1 + + +def test_opted_out_run_reads_no_config( + usage_events, interactive, telemetry_config_dir, config_reads, monkeypatch +): + monkeypatch.setenv("DO_NOT_TRACK", "1") + (telemetry_config_dir / "config.ini").write_text( + "[default]\napi_host = https://artifacts.example.com\n" + ) + run.finish(facts()) + assert usage_events == [] and config_reads == [] + + +def test_config_opt_out_reads_config_only_once( + usage_events, interactive, telemetry_config_dir, config_reads +): + (telemetry_config_dir / "config.ini").write_text("[default]\ntelemetry = false\n") + run.finish(facts()) + assert usage_events == [] and len(config_reads) == 1 + + +def test_config_file_gate_fails_closed(config_reads): + def unreadable(): + raise OSError("unreadable") + + assert run._config_file_blocks(facts(), unreadable) is True # pylint: disable=protected-access + + +def test_laptop_run_keeps_an_install_id_by_default(usage_events, interactive): + # No CLOUDSMITH_TELEMETRY_INSTALL_ID needed (the fixture clears it). + run.finish(facts()) + run.finish(facts()) + first, second = usage_events + assert first["distinct_id"] == second["distinct_id"] + assert first["properties"]["install_id_persisted"] is True + assert first["properties"]["first_run"] is True + assert second["properties"]["first_run"] is False + assert second["properties"]["$process_person_profile"] is False + assert state.read()["install_id"] == first["distinct_id"] + + +@pytest.fixture() +def stored_id(usage_events, interactive): + run.finish(facts()) + (sent,) = usage_events + assert state.read()["install_id"] == sent["distinct_id"] + usage_events.clear() + return sent["distinct_id"] + + +@pytest.mark.parametrize("name", ["DO_NOT_TRACK", "CLOUDSMITH_NO_TELEMETRY"]) +def test_env_opt_out_deletes_the_install_id(stored_id, usage_events, monkeypatch, name): + monkeypatch.setenv(name, "1") + run.finish(facts()) + assert usage_events == [] + assert "install_id" not in state.read() and "created_at" not in state.read() + # The notice was already shown; opting out does not forget that. + assert state.read()["notice_version"] == notice.NOTICE_VERSION + + +def test_config_opt_out_deletes_the_install_id( + stored_id, usage_events, telemetry_config_dir +): + (telemetry_config_dir / "config.ini").write_text("[default]\ntelemetry = false\n") + run.finish(facts()) + assert usage_events == [] and "install_id" not in state.read() + + +def test_unreadable_config_keeps_the_install_id( + stored_id, usage_events, telemetry_config_dir +): + # Fails closed (nothing sent), but a broken file is not an opt-out. + (telemetry_config_dir / "config.ini").write_text("[default\ntelemetry = x\n") + run.finish(facts()) + assert usage_events == [] and state.read()["install_id"] == stored_id + + +def test_opted_out_run_creates_no_state_file( + usage_events, usage_state_dir, monkeypatch +): + monkeypatch.setenv("DO_NOT_TRACK", "1") + run.finish(facts()) + assert not usage_state_dir.joinpath(state.STATE_FILE_NAME).exists() + + +def test_opting_back_in_creates_a_new_install_id(stored_id, usage_events, monkeypatch): + monkeypatch.setenv("DO_NOT_TRACK", "1") + run.finish(facts()) + monkeypatch.delenv("DO_NOT_TRACK") + run.finish(facts()) + (sent,) = usage_events + assert sent["distinct_id"] != stored_id + assert sent["properties"]["first_run"] is True + + +def test_excluded_command_keeps_the_install_id_when_opted_out(stored_id, monkeypatch): + monkeypatch.setenv("DO_NOT_TRACK", "1") + run.finish(facts(command="credential-helper docker")) + assert state.read()["install_id"] == stored_id + + +def test_install_id_off_switch_gives_one_off_ids( + usage_events, interactive, monkeypatch +): + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_INSTALL_ID", "0") + run.finish(facts()) + run.finish(facts()) + first, second = usage_events + assert first["distinct_id"] != second["distinct_id"] + assert "install_id" not in state.read() diff --git a/cloudsmith_cli/core/tests/test_usage_sender.py b/cloudsmith_cli/core/tests/test_usage_sender.py new file mode 100644 index 00000000..734c30b1 --- /dev/null +++ b/cloudsmith_cli/core/tests/test_usage_sender.py @@ -0,0 +1,225 @@ +import json +import threading +import time + +import pytest + +from cloudsmith_cli.core import telemetry_common +from cloudsmith_cli.core.usage import sender + +# Captured at import: the autouse conftest guard replaces sender._post. +REAL_POST = sender._post + +PAYLOAD = { + "event": "cli command executed", + "distinct_id": "x", + "timestamp": "t", + "properties": {}, +} + + +def test_send_posts_capture_body(monkeypatch): + seen = {} + + def fake_post(session, url, body, timeout): + seen.update( + url=url, + body=json.loads(body), + timeout=timeout, + headers=dict(session.headers), + ) + return True + + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + monkeypatch.setattr(sender, "_post", fake_post) + telemetry_common.WaitBudget.shared().reset(1.0) + assert sender.send(PAYLOAD) is True + assert seen["url"] == "https://eu.i.posthog.com/i/v0/e/" + assert seen["body"]["api_key"] == "phc_test" + assert seen["timeout"] == sender.REQUEST_TIMEOUT_SECONDS + assert seen["headers"]["User-Agent"] == sender.USER_AGENT + assert not any(h.lower().startswith("x-api-key") for h in seen["headers"]) + + +def test_send_session_is_isolated_from_user_settings(monkeypatch): + import cloudsmith_api + + config = cloudsmith_api.Configuration() + config.cert_file = "/x/c.pem" + config.key_file = "/x/k.pem" + config.verify_ssl = False + # Set on the SDK Configuration at runtime by core.api.init, not declared. + setattr(config, "error_retry_max", 7) + cloudsmith_api.Configuration.set_default(config) + seen = {} + + def fake_post(session, url, body, timeout): + seen.update( + cert=session.cert, + verify=session.verify, + retries=session.get_adapter(url).max_retries.total, + ) + return True + + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + monkeypatch.setattr(sender, "_post", fake_post) + telemetry_common.WaitBudget.shared().reset(1.0) + assert sender.send(PAYLOAD) is True + assert seen == {"cert": None, "verify": True, "retries": 0} + + +def test_send_skips_when_budget_is_spent(monkeypatch): + calls = [] + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + monkeypatch.setattr(sender, "_post", lambda *a: calls.append(a) or True) + telemetry_common.WaitBudget.shared().reset(0.0) + assert sender.send(PAYLOAD) is None + assert calls == [] + + +def test_send_skips_when_too_little_budget_is_left(monkeypatch): + # A POST that cannot finish in time would count as a failure and start + # the 24 h backoff, so a nearly spent budget sends nothing instead. + calls = [] + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + monkeypatch.setattr(sender, "_post", lambda *a: calls.append(a) or True) + budget = telemetry_common.WaitBudget.shared() + budget.reset(sender.MIN_SEND_SECONDS / 2) + assert sender.send(PAYLOAD) is None + assert calls == [] + assert budget.remaining() == sender.MIN_SEND_SECONDS / 2 + + +def test_session_setup_is_charged_to_the_budget(monkeypatch): + # A slow cold start (importing requests and ssl) must not get a full + # budget on top for the POST. + real_session = sender._session + + def slow_session(*args): + time.sleep(0.2) + return real_session(*args) + + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + monkeypatch.setattr(sender, "_session", slow_session) + monkeypatch.setattr(sender, "_post", lambda *a: True) + budget = telemetry_common.WaitBudget.shared() + budget.reset(1.0) + assert sender.send(PAYLOAD) is True + assert budget.remaining() <= 0.8 + + +def test_send_skips_when_setup_used_up_the_budget(monkeypatch): + calls = [] + real_session = sender._session + + def slow_session(*args): + time.sleep(0.3) + return real_session(*args) + + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + monkeypatch.setattr(sender, "_session", slow_session) + monkeypatch.setattr(sender, "_post", lambda *a: calls.append(a) or True) + telemetry_common.WaitBudget.shared().reset(0.4) + assert sender.send(PAYLOAD) is None + assert calls == [] + + +@pytest.mark.parametrize("strict, x509_strict_cleared", [(False, True), (None, False)]) +def test_session_follows_the_resolved_ssl_strictness(strict, x509_strict_cleared): + import ssl + + session = sender._session(strict) + assert session.verify is True # certificates are always verified + adapter = session.get_adapter("https://eu.i.posthog.com/") + has_factory = adapter.ssl_context_factory is not None + assert has_factory is x509_strict_cleared + if has_factory: + assert not adapter.ssl_context_factory().verify_flags & ssl.VERIFY_X509_STRICT + + +def test_send_reports_failed_post(monkeypatch): + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + monkeypatch.setattr(sender, "_post", lambda *a: False) + telemetry_common.WaitBudget.shared().reset(1.0) + assert sender.send(PAYLOAD) is False + + +def test_send_never_waits_beyond_budget(monkeypatch): + gate = threading.Event() + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + monkeypatch.setattr(sender, "_post", lambda *a: gate.wait(10)) + telemetry_common.WaitBudget.shared().reset(0.3) + started = time.monotonic() + assert sender.send(PAYLOAD) is False + assert time.monotonic() - started < 0.6 + gate.set() + + +def test_send_spends_the_shared_budget(monkeypatch): + gate = threading.Event() + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + monkeypatch.setattr(sender, "_post", lambda *a: gate.wait(10)) + budget = telemetry_common.WaitBudget.shared() + budget.reset(0.3) + assert sender.send(PAYLOAD) is False + assert budget.remaining() == 0.0 + gate.set() + + +def test_send_failure_is_silent(monkeypatch, capsys): + def boom(*a): + raise RuntimeError("network down") + + monkeypatch.setenv("CLOUDSMITH_TELEMETRY_POSTHOG_KEY", "phc_test") + monkeypatch.setattr(sender, "_post", boom) + telemetry_common.WaitBudget.shared().reset(1.0) + assert sender.send(PAYLOAD) is False + assert capsys.readouterr() == ("", "") + + +@pytest.mark.parametrize( + "status, ok", [(200, True), (204, True), (307, False), (308, False), (400, False)] +) +def test_post_succeeds_on_2xx_only_and_never_follows_redirects(status, ok): + seen = {} + + class FakeSession: + def post(self, url, **kwargs): + seen.update(kwargs) + return type("Response", (), {"status_code": status})() + + assert REAL_POST(FakeSession(), "https://h/i/v0/e/", "{}", 1.0) is ok + assert seen["allow_redirects"] is False + + +def test_log_prints_payload_to_stderr_without_key(capsys): + sender.log(PAYLOAD) + out, err = capsys.readouterr() + assert out == "" + assert json.loads(err.split("\n", 1)[1])["event"] == "cli command executed" + + +def test_session_never_sends_netrc_credentials_but_keeps_env_proxies( + monkeypatch, tmp_path +): + import requests + + from cloudsmith_cli.core import session as session_module + from cloudsmith_cli.core.usage import settings + + netrc = tmp_path / ".netrc" + netrc.write_text("default login nuser password NETRCSECRET\n") + netrc.chmod(0o600) + monkeypatch.setenv("NETRC", str(netrc)) + monkeypatch.setenv("HOME", str(tmp_path)) + for name in ("NO_PROXY", "no_proxy", "https_proxy", "HTTPS_PROXY"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("HTTPS_PROXY", "http://proxy.example:3128") + monkeypatch.setitem(session_module._default_proxy, "url", None) + url = settings.host() + sender._CAPTURE_PATH + + session = sender._session() + prepared = session.prepare_request(requests.Request("POST", url, data="{}")) + assert "Authorization" not in prepared.headers + env = session.merge_environment_settings(url, {}, None, None, None) + assert env["proxies"]["https"] == "http://proxy.example:3128" diff --git a/cloudsmith_cli/core/usage/event.py b/cloudsmith_cli/core/usage/event.py index caab878c..0d7de3bb 100644 --- a/cloudsmith_cli/core/usage/event.py +++ b/cloudsmith_cli/core/usage/event.py @@ -79,6 +79,9 @@ class RunFacts: auth_source: str | None = None api_host: str | None = None ssl_verify: bool = True + #: The resolved ``--without-ssl-verify-strict`` setting, for the sender's + #: own session (None: the default). Not an event property. + ssl_verify_strict: bool | None = None stdout_tty: bool = False stdin_tty: bool = False nested: bool = False diff --git a/cloudsmith_cli/core/usage/identity.py b/cloudsmith_cli/core/usage/identity.py index e78af0a2..e2d07468 100644 --- a/cloudsmith_cli/core/usage/identity.py +++ b/cloudsmith_cli/core/usage/identity.py @@ -68,3 +68,17 @@ def reset(): raise IdentityResetError( "removed the install ID but could not restore other telemetry state" ) + + +def forget(): + """Delete a stored install ID after an opt-out; silent, never raises. + + Writes nothing when no ID is stored, so an opted-out run creates no file. + """ + data = state.read() + if "install_id" not in data and "created_at" not in data: + return + try: + reset() + except IdentityResetError: + return diff --git a/cloudsmith_cli/core/usage/run.py b/cloudsmith_cli/core/usage/run.py new file mode 100644 index 00000000..2ba616b4 --- /dev/null +++ b/cloudsmith_cli/core/usage/run.py @@ -0,0 +1,171 @@ +# Copyright 2026 Cloudsmith Ltd +"""Decide whether to send a usage event for this run, and send it. + +Checks run cheapest-first and stop at the first "no", so excluded commands +(credential helpers on every pull) pay almost nothing. Never raises. +""" + +import functools +import random +import time +from datetime import UTC, datetime + +EXCLUDED_COMMANDS = frozenset({"mcp start"}) +#: Machine-invoked or telemetry-managing commands. ``install``, +#: ``uninstall`` and ``list`` under credential-helper are run by people and +#: do send. +EXCLUDED_PREFIXES = ( + "credential-helper cargo", + "credential-helper docker", + "credential-helper generic", + "credential-helper pnpm", + "credential-helper terraform", + "telemetry", +) +BACKOFF_SECONDS = 24 * 60 * 60 + + +def _excluded(command): + if command in EXCLUDED_COMMANDS: + return True + return any(command == p or command.startswith(p + " ") for p in EXCLUDED_PREFIXES) + + +def _load_deferred_sources(facts): + """Merge config-file setting sources in; Click flag/env sources win.""" + if facts.deferred_sources is None: + return + try: + loaded = facts.deferred_sources() + facts.setting_sources = {**loaded, **facts.setting_sources} + except Exception: # pylint: disable=broad-exception-caught + return + + +def _config_file_blocks(facts, read_config): + """Apply the host and SSL gates to config.ini for a run that never loaded it. + + Root ``--help`` and root usage errors end before the root callback loads + config, so a host or ``ssl_verify`` set only in config.ini never reached + the facts. A host from the env var or a flag still wins over the file. + Fails closed if the file cannot be read. + """ + from ...cli import config + from .. import telemetry_common + + if getattr(config.OPTIONS, "value", None) is not None: + return False + try: + probe = read_config() + except Exception: # pylint: disable=broad-exception-caught + return True + if not facts.api_host and not telemetry_common.is_cloudsmith_host(probe.api_host): + return True + return probe.ssl_verify is False + + +def _config_opted_out(read_config): + """Tell whether config.ini really says ``telemetry = false``. + + Unlike ``config_disabled``, an unreadable file is not an opt-out, so it + must not delete the install ID. + """ + from ...cli import config + + try: + opts = getattr(config.OPTIONS, "value", None) + if opts is not None and opts.telemetry is False: + return True + return read_config().telemetry is False + except Exception: # pylint: disable=broad-exception-caught + return False + + +def finish(facts): + try: + _finish(facts) + except BaseException: # pylint: disable=broad-exception-caught + return + + +def _finish(facts): # pylint: disable=too-many-return-statements,too-many-locals + if facts.nested or facts.config_unresolved or _excluded(facts.command): + return + + # Imported late so an excluded command pays for none of these modules. + from .. import installation, telemetry_common, version + from . import environment, event, identity, notice, sender, settings, state + + if telemetry_common.telemetry_disabled(): + identity.forget() + return + # The opt-out check and the config.ini gate share one read of the file. + read_config = functools.cache(telemetry_common.read_config_file) + if telemetry_common.config_disabled(read_config): + if _config_opted_out(read_config): + identity.forget() + return + if not telemetry_common.is_cloudsmith_host(facts.api_host) or not facts.ssl_verify: + return + if _config_file_blocks(facts, read_config): + return + env_name = telemetry_common.get_environment() + if env_name == "development" and not settings.override_set(): + return + if not settings.project_key() and not settings.log_mode(): + return + + channel = installation.detect_channel() + provider = environment.ci_provider() + ephemeral = environment.is_ephemeral(channel=channel, state_writable=True) + if not ephemeral: + # Ephemeral runs must do no file I/O, so the writability probe only + # runs for otherwise non-ephemeral runs. An unwritable state file + # cannot record the notice or the backoff, so treat it as ephemeral. + ephemeral = environment.is_ephemeral( + channel=channel, state_writable=state.writable() + ) + + if not ephemeral: + last_failure = state.read().get("last_failure_at") + if ( + isinstance(last_failure, (int, float)) + and time.time() - last_failure < BACKOFF_SECONDS + ): + return + + # Only CI runs are sampled; containers and unwritable state are ephemeral + # too but are not CI (their events carry ci=false). + sample_rate = settings.ci_sample_rate() if provider is not None else 100 + if sample_rate < 100 and random.uniform(0, 100) >= sample_rate: + return + + interactive = facts.stdout_tty and facts.stdin_tty and provider is None + if interactive and not ephemeral and notice.needs_notice(): + notice.show() + return + + ident = identity.resolve( + ephemeral=ephemeral, install_id_enabled=settings.install_id_enabled() + ) + _load_deferred_sources(facts) + payload = event.build_event( + facts=facts, + identity=ident, + runtime=environment.runtime_properties(), + channel=channel, + environment=env_name, + cli_version=version.get_version(), + ci_provider=provider, + sample_rate=sample_rate, + internal=settings.internal(), + timestamp=datetime.now(UTC).isoformat(), + ) + + if settings.log_mode(): + sender.log(payload) + return + # send() returns None when too little of the shared wait budget is left + # (nothing was attempted); only a real failure (False) starts the backoff. + if sender.send(payload, facts.ssl_verify_strict) is False and not ephemeral: + state.write({"last_failure_at": time.time()}) diff --git a/cloudsmith_cli/core/usage/sender.py b/cloudsmith_cli/core/usage/sender.py new file mode 100644 index 00000000..1c305439 --- /dev/null +++ b/cloudsmith_cli/core/usage/sender.py @@ -0,0 +1,103 @@ +# Copyright 2026 Cloudsmith Ltd +"""Send one usage event with a plain POST. + +The session is built on the main thread: no user headers, no client +certificate, no retries, the CLI's proxy and CA settings only. The thread +only POSTs; the caller records failures. +""" + +import json +import threading +import time + +from . import settings + +REQUEST_TIMEOUT_SECONDS = 2.0 +#: Below this much budget a POST rarely finishes, and a timeout would start +#: the 24 h backoff, so nothing is sent. +MIN_SEND_SECONDS = 0.25 +USER_AGENT = "cloudsmith-cli-telemetry" +_CAPTURE_PATH = "/i/v0/e/" + + +def _post(session, url, body, timeout): + # A followed 307/308 would re-POST the event and key to another host. + response = session.post( + url, + data=body, + timeout=timeout, + headers={"Content-Type": "application/json"}, + allow_redirects=False, + ) + return 200 <= response.status_code < 300 + + +def _no_auth(request): + return request + + +def _session(ssl_verify_strict=None): + from ..session import create_requests_session, get_default_proxy + + # Certificates are always verified; only X.509 strictness follows the + # command (--without-ssl-verify-strict, for TLS-intercepting proxies). + session = create_requests_session( + retries=0, + ssl_verify=True, + ssl_verify_strict=ssl_verify_strict, + proxy=get_default_proxy(), + user_agent=USER_AGENT, + ) + session.cert = None # never present the user's API client certificate + # Any session auth stops requests adding ~/.netrc credentials; a no-op + # keeps trust_env (proxy and CA bundle env vars) working. + session.auth = _no_auth + return session + + +def send(payload, ssl_verify_strict=None): + """POST ``payload`` within the shared wait budget. + + Returns None, with no request made, when less than ``MIN_SEND_SECONDS`` + of the budget is left, before or after building the session. Otherwise + returns True only if the POST finished with a 2xx within the budget, + else False. Building the session counts against the budget. + """ + from ..telemetry_common import WaitBudget + + budget = WaitBudget.shared() + if budget.remaining() < MIN_SEND_SECONDS: + return None + + setup_started = time.monotonic() + body = json.dumps({**payload, "api_key": settings.project_key()}) + url = settings.host() + _CAPTURE_PATH + session = _session(ssl_verify_strict) # imports requests on the main thread + budget.spend(time.monotonic() - setup_started) + if budget.remaining() < MIN_SEND_SECONDS: + return None + result = {"ok": False} + + def target(): + try: + result["ok"] = bool(_post(session, url, body, REQUEST_TIMEOUT_SECONDS)) + except Exception: # pylint: disable=broad-exception-caught + result["ok"] = False + + thread = threading.Thread(target=target, name="cloudsmith-usage", daemon=True) + started = time.monotonic() + thread.start() + thread.join(budget.remaining()) + budget.spend(time.monotonic() - started) + return (not thread.is_alive()) and result["ok"] + + +def log(payload): + """Print the event to stderr instead of sending it (log mode).""" + import click + + click.echo( + "[telemetry] would send (log mode, nothing sent):\n" + + json.dumps(payload, indent=2, sort_keys=True), + err=True, + )