diff --git a/codecov-cli/codecov_cli/helpers/args.py b/codecov-cli/codecov_cli/helpers/args.py index ea30891c..7d35b455 100644 --- a/codecov-cli/codecov_cli/helpers/args.py +++ b/codecov-cli/codecov_cli/helpers/args.py @@ -17,6 +17,8 @@ def get_cli_args(ctx: click.Context): args.update(ctx.params) if "token" in args: del args["token"] + if "http_header" in args: + del args["http_header"] filtered_args = {} for k in args.keys(): diff --git a/codecov-cli/codecov_cli/helpers/request.py b/codecov-cli/codecov_cli/helpers/request.py index bd1c9a17..d36f05d4 100644 --- a/codecov-cli/codecov_cli/helpers/request.py +++ b/codecov-cli/codecov_cli/helpers/request.py @@ -16,25 +16,33 @@ USER_AGENT = f"codecov-cli/{__version__}" +_extra_headers: dict = {} -def _set_user_agent(headers: Optional[dict] = None) -> dict: + +def set_extra_headers(headers: dict): + global _extra_headers + _extra_headers = dict(headers) + + +def _prepare_headers(headers: Optional[dict] = None) -> dict: headers = headers or {} - headers.setdefault("User-Agent", USER_AGENT) - return headers + merged = {**_extra_headers, **headers} + merged["User-Agent"] = USER_AGENT + return merged def patch(url: str, headers: dict = None, json: dict = None) -> requests.Response: - headers = _set_user_agent(headers) + headers = _prepare_headers(headers) return requests.patch(url, json=json, headers=headers) def get(url: str, headers: dict = None, params: dict = None) -> requests.Response: - headers = _set_user_agent(headers) + headers = _prepare_headers(headers) return requests.get(url, params=params, headers=headers) def put(url: str, data: dict = None, headers: dict = None) -> requests.Response: - headers = _set_user_agent(headers) + headers = _prepare_headers(headers) return requests.put(url, data=data, headers=headers) @@ -44,7 +52,7 @@ def post( headers: Optional[dict] = None, params: Optional[dict] = None, ) -> requests.Response: - headers = _set_user_agent(headers) + headers = _prepare_headers(headers) return requests.post(url, json=data, headers=headers, params=params) diff --git a/codecov-cli/codecov_cli/main.py b/codecov-cli/codecov_cli/main.py index 2568ef99..ab49caa5 100644 --- a/codecov-cli/codecov_cli/main.py +++ b/codecov-cli/codecov_cli/main.py @@ -22,6 +22,7 @@ from codecov_cli.helpers.ci_adapters import get_ci_adapter, get_ci_providers_list from codecov_cli.helpers.config import load_cli_config from codecov_cli.helpers.logging_utils import configure_logger +from codecov_cli.helpers.request import set_extra_headers from codecov_cli.helpers.versioning_systems import get_versioning_system from codecov_cli.opentelemetry import init_telem @@ -48,6 +49,11 @@ @click.option( "--disable-telem", help="Disable sending telemetry data to Codecov", is_flag=True ) +@click.option( + "--http-header", + multiple=True, + help="Extra HTTP header to send with every request (format: Header-Name:Value). Can be specified multiple times.", +) @click.pass_context @click.version_option(__version__, prog_name="codecovcli") def cli( @@ -57,6 +63,7 @@ def cli( enterprise_url: str, verbose: bool = False, disable_telem: bool = False, + http_header: typing.Tuple[str, ...] = (), ): ctx.obj["cli_args"] = ctx.params ctx.obj["cli_args"]["version"] = f"cli-{__version__}" @@ -72,6 +79,23 @@ def cli( ctx.obj["enterprise_url"] = enterprise_url ctx.obj["disable_telem"] = disable_telem ctx.obj["branding"] = [Branding.CODECOV] + if http_header: + extra = {} + for h in http_header: + if ":" not in h: + raise click.BadParameter( + f"Invalid header format: '{h}'. Expected 'Header-Name:Value'.", + param_hint="'--http-header'", + ) + name, value = h.split(":", 1) + name = name.strip() + if not name: + raise click.BadParameter( + f"Invalid header format: '{h}'. Header name cannot be empty.", + param_hint="'--http-header'", + ) + extra[name] = value.strip() + set_extra_headers(extra) init_telem(ctx.obj) diff --git a/codecov-cli/tests/helpers/test_request.py b/codecov-cli/tests/helpers/test_request.py index 769e3f3a..23baeee0 100644 --- a/codecov-cli/tests/helpers/test_request.py +++ b/codecov-cli/tests/helpers/test_request.py @@ -5,11 +5,14 @@ from requests import Response from codecov_cli import __version__ +from codecov_cli.helpers import request as request_module from codecov_cli.helpers.request import ( + _prepare_headers, get, get_token_header, get_token_header_or_fail, log_warnings_and_errors_if_any, + set_extra_headers, ) from codecov_cli.helpers.request import logger as req_log from codecov_cli.helpers.request import ( @@ -186,3 +189,65 @@ def mock_request(*args, headers={}, **kwargs): side_effect=mock_request, ) patch("my_url") + + +class TestExtraHeaders: + @pytest.fixture(autouse=True) + def reset_extra_headers(self): + set_extra_headers({}) + yield + set_extra_headers({}) + + def test_prepare_headers_without_extra(self): + headers = _prepare_headers() + assert headers == {"User-Agent": f"codecov-cli/{__version__}"} + + def test_prepare_headers_with_extra(self): + set_extra_headers({"CF-Access-Client-Id": "abc123"}) + headers = _prepare_headers() + assert headers["CF-Access-Client-Id"] == "abc123" + assert headers["User-Agent"] == f"codecov-cli/{__version__}" + + def test_extra_headers_dont_overwrite_authorization(self): + set_extra_headers({"Authorization": "evil"}) + headers = _prepare_headers({"Authorization": "token real-token"}) + assert headers["Authorization"] == "token real-token" + + def test_extra_headers_dont_overwrite_user_agent(self): + set_extra_headers({"User-Agent": "custom-agent"}) + headers = _prepare_headers() + assert headers["User-Agent"] == f"codecov-cli/{__version__}" + + def test_extra_headers_merged_into_post(self, mocker): + set_extra_headers({"X-Custom": "value"}) + + def mock_post(*args, headers=None, **kwargs): + assert headers["X-Custom"] == "value" + assert headers["User-Agent"] == f"codecov-cli/{__version__}" + resp = Response() + resp.status_code = 200 + resp._content = b"ok" + return resp + + mocker.patch.object(requests, "post", side_effect=mock_post) + send_post_request("my_url") + + def test_extra_headers_merged_into_get(self, mocker): + set_extra_headers({"X-Custom": "value"}) + + def mock_get(*args, headers=None, **kwargs): + assert headers["X-Custom"] == "value" + resp = Response() + resp.status_code = 200 + resp._content = b"ok" + return resp + + mocker.patch.object(requests, "get", side_effect=mock_get) + get("my_url") + + def test_set_extra_headers_replaces_previous(self): + set_extra_headers({"A": "1"}) + set_extra_headers({"B": "2"}) + headers = _prepare_headers() + assert "A" not in headers + assert headers["B"] == "2" diff --git a/codecov-cli/tests/test_codecov_cli.py b/codecov-cli/tests/test_codecov_cli.py index 6d3a81c3..2ca62b99 100644 --- a/codecov-cli/tests/test_codecov_cli.py +++ b/codecov-cli/tests/test_codecov_cli.py @@ -1,4 +1,8 @@ +import pytest +from click.testing import CliRunner + from codecov_cli import main +from codecov_cli.helpers import request as request_module def test_existing_commands(): @@ -17,3 +21,61 @@ def test_existing_commands(): "upload-coverage", "upload-process", ] + + +class TestHttpHeaderOption: + @pytest.fixture(autouse=True) + def reset_extra_headers(self): + request_module._extra_headers = {} + yield + request_module._extra_headers = {} + + def test_http_header_valid(self): + runner = CliRunner() + result = runner.invoke( + main.cli, + [ + "--http-header", + "CF-Access-Client-Id:abc123", + "--http-header", + "CF-Access-Client-Secret:xyz789", + "do-upload", + "--help", + ], + obj={}, + ) + assert result.exit_code == 0 + assert request_module._extra_headers == { + "CF-Access-Client-Id": "abc123", + "CF-Access-Client-Secret": "xyz789", + } + + def test_http_header_invalid_format(self): + runner = CliRunner() + result = runner.invoke( + main.cli, + ["--http-header", "InvalidHeader", "do-upload", "--help"], + obj={}, + ) + assert result.exit_code != 0 + assert "Invalid header format" in result.output + + def test_http_header_value_with_colon(self): + runner = CliRunner() + result = runner.invoke( + main.cli, + ["--http-header", "X-Test:value:with:colons", "do-upload", "--help"], + obj={}, + ) + assert result.exit_code == 0 + assert request_module._extra_headers == {"X-Test": "value:with:colons"} + + def test_http_header_empty_name(self): + runner = CliRunner() + result = runner.invoke( + main.cli, + ["--http-header", ":value", "do-upload", "--help"], + obj={}, + ) + assert result.exit_code != 0 + assert "Header name cannot be empty" in result.output