From b665233e4e462a376eacb5e1f37f8d3ced460ef4 Mon Sep 17 00:00:00 2001 From: "omer.roth" Date: Tue, 28 Jul 2026 09:35:17 +0300 Subject: [PATCH 1/2] CM-69684 add api proxy endpoint --- README.md | 36 ++- cycode/cli/apps/api/__init__.py | 11 +- cycode/cli/apps/api/raw_api_command.py | 205 +++++++++++++ cycode/cli/utils/get_api_client.py | 18 +- cycode/cyclient/client_creator.py | 12 + tests/cli/apps/api/test_raw_api_command.py | 320 +++++++++++++++++++++ 6 files changed, 595 insertions(+), 7 deletions(-) create mode 100644 cycode/cli/apps/api/raw_api_command.py create mode 100644 tests/cli/apps/api/test_raw_api_command.py diff --git a/README.md b/README.md index 6ccbee88..8e962c39 100644 --- a/README.md +++ b/README.md @@ -24,8 +24,9 @@ This guide walks you through both installation and usage. 5. [Advanced Configuration](#advanced-configuration) 5. [Platform Command](#platform-command-beta) 1. [Discovering Commands](#discovering-commands) - 2. [Examples](#platform-examples) - 3. [Notes & Limitations](#platform-notes--limitations) + 2. [Raw API Requests](#raw-api-requests) + 3. [Examples](#platform-examples) + 4. [Notes & Limitations](#platform-notes--limitations) 6. [Scan Command](#scan-command) 1. [Running a Scan](#running-a-scan) 1. [Options](#options) @@ -672,6 +673,35 @@ cycode platform projects --help # list actions on a resource cycode platform projects list --help # list options/arguments for an action ``` +## Raw API Requests + +`cycode platform api` sends a raw authenticated request to any Cycode REST endpoint — including endpoints the spec does not expose as a generated command, and writes (`post`, `put`). It is the escape hatch for scripts: you supply the method, path, and body, and the CLI handles authentication. + +```bash +cycode platform api [-q KEY=VALUE]... [-H "KEY: VALUE"]... [-d JSON] +``` + +Credentials come from the CLI itself — `cycode auth`, the `CYCODE_CLIENT_ID` / `CYCODE_CLIENT_SECRET` (or `CYCODE_ID_TOKEN`) environment variables, or the global `--client-id` / `--client-secret` / `--id-token` options. The CLI mints and refreshes the access token for you, so scripts never handle tokens — your credentials and access token are never printed. Note that `-v` / `--verbose` logs full response bodies to stderr, so avoid it in shared CI logs when the endpoint returns sensitive data. + +```bash +# GET with query parameters (repeat a key to send multiple values) +cycode platform api get v4/projects -q page-size=5 +cycode platform api get v4/violations -q severity=High -q severity=Critical + +# POST a JSON body — inline, from a file, or from stdin +cycode platform api post v4/some/resource -d '{"name": "example"}' +cycode platform api post v4/some/resource -d @body.json +cat body.json | cycode platform api put v4/some/resource -d - + +# Pipe through jq like any other platform command +cycode platform api get v4/projects -q page-size=100 | jq '.items[].name' +``` + +The response body is printed as formatted JSON (non-JSON bodies are printed as-is). On an HTTP error the status and response body go to stderr and the command exits non-zero, so `set -e` scripts fail as expected. `PATH` must be a v4 API path such as `v4/projects` (or `api/v4/...`), not a full URL — use `cycode configure` to point the CLI at a different Cycode installation. The `Authorization` header is owned by the CLI and cannot be overridden. Unlike the generated commands, `cycode platform api` never fetches the OpenAPI spec. + +> [!NOTE] +> `-v` / `--verbose` logs the outgoing request and the response, which is the quickest way to debug an unexpected status code. + ## Platform Examples ```bash @@ -696,7 +726,7 @@ cycode platform projects list --page-size 100 | jq '.items[].name' ## Platform Notes & Limitations -- **Read-only today.** Only `GET` endpoints are exposed in this beta. +- **Generated commands are read-only.** Only `GET` endpoints are turned into commands in this beta. For writes, use [`cycode platform api`](#raw-api-requests) with `post` or `put`. - **Spec-driven.** Adding a new endpoint to the API surfaces it automatically the next time the cache is refreshed. - **No bundled spec.** The first `cycode platform` invocation after install (or after the 24h cache expires) performs a network fetch. On slow connections this first call may take a few seconds; subsequent calls are near-instant until the cache expires. - **Override the cache TTL** with `CYCODE_SPEC_CACHE_TTL=`. diff --git a/cycode/cli/apps/api/__init__.py b/cycode/cli/apps/api/__init__.py index e65f9c6f..cb560b11 100644 --- a/cycode/cli/apps/api/__init__.py +++ b/cycode/cli/apps/api/__init__.py @@ -60,10 +60,19 @@ def list_commands(self, ctx: click.Context) -> list[str]: return super().list_commands(ctx) def get_command(self, ctx: click.Context, cmd_name: str) -> Optional[click.Command]: + # Statically registered commands (like `api`) must not trigger a spec fetch. + if cmd_name in self.commands: + return super().get_command(ctx, cmd_name) + self._ensure_loaded(ctx) return super().get_command(ctx, cmd_name) def get_platform_group() -> click.Group: """Return the top-level `platform` Click group (lazy-loading).""" - return PlatformGroup(name='platform', help=_PLATFORM_HELP, no_args_is_help=True) + from cycode.cli.apps.api.raw_api_command import build_raw_api_command + + group = PlatformGroup(name='platform', help=_PLATFORM_HELP, no_args_is_help=True) + # The raw request escape hatch is registered statically: it needs no OpenAPI spec. + group.add_command(build_raw_api_command(), 'api') + return group diff --git a/cycode/cli/apps/api/raw_api_command.py b/cycode/cli/apps/api/raw_api_command.py new file mode 100644 index 00000000..6aff1ee4 --- /dev/null +++ b/cycode/cli/apps/api/raw_api_command.py @@ -0,0 +1,205 @@ +"""Raw REST passthrough: `cycode platform api `. + +Lets scripts call any Cycode REST endpoint without handling credentials themselves. +The CLI resolves credentials (client id/secret or OIDC, from flags, environment +variables, or `cycode auth`), mints and refreshes the access token, and prints the +response body. Tokens and secrets are never printed. +""" + +import json +import re +import sys +from typing import TYPE_CHECKING, Any, Optional, Union + +import click + +from cycode.logger import get_logger + +if TYPE_CHECKING: + from requests import Response + +logger = get_logger('Raw API Command') + +_SUPPORTED_METHODS = ('get', 'post', 'put') +_METHODS_WITH_BODY = ('post', 'put') +# Allowlist: a versioned API path such as `v4/projects` or `api/v4/auth/api-token`. +# Anything else (full URLs, protocol-relative paths, paths with whitespace) is rejected. +_API_PATH_RE = re.compile(r'^(api/)?v4/\S+$') + +_HELP = """[BETA] Send a raw authenticated request to the Cycode API. + +METHOD is one of get, post, put. PATH is an API path such as `v4/projects` +(a leading slash is optional). + +Credentials come from the CLI: `cycode auth`, the CYCODE_CLIENT_ID/CYCODE_CLIENT_SECRET +(or CYCODE_ID_TOKEN) environment variables, or the global `--client-id`/`--client-secret`/ +`--id-token` options. Your credentials and access token are never printed. + +\b +Examples: + cycode platform api get v4/projects -q page-size=5 + cycode platform api get v4/violations -q severity=High -q severity=Critical + cycode platform api post v4/sbom/import -d @body.json + cat body.json | cycode platform api put v4/some/resource -d - +""" + + +def _validate_path(path: str) -> str: + """Validate that PATH is a versioned API path, so credentials only reach the configured Cycode host.""" + url_path = path.lstrip('/') + if not _API_PATH_RE.match(url_path): + raise click.ClickException(f'PATH must be a versioned API path such as `v4/projects`, not `{path}`.') + + return url_path + + +def _parse_query(query: tuple[str, ...]) -> dict[str, Union[str, list[str]]]: + """Parse repeatable `key=value` pairs. Repeated keys collapse into a list.""" + params: dict[str, Union[str, list[str]]] = {} + for item in query: + key, sep, value = item.partition('=') + if not sep or not key: + raise click.ClickException(f'Invalid query parameter "{item}". Expected format: key=value') + + if key in params: + existing = params[key] + if isinstance(existing, list): + existing.append(value) + else: + params[key] = [existing, value] + else: + params[key] = value + + return params + + +def _parse_headers(header: tuple[str, ...]) -> dict[str, str]: + """Parse repeatable `Key: Value` pairs.""" + headers: dict[str, str] = {} + for item in header: + key, sep, value = item.partition(':') + key = key.strip() + if not sep or not key: + raise click.ClickException(f'Invalid header "{item}". Expected format: "Key: Value"') + + if key.lower() == 'authorization': + raise click.ClickException('The Authorization header is managed by the CLI and cannot be overridden.') + + headers[key] = value.strip() + + return headers + + +def _read_body(data: str) -> Any: + """Read the request body from an inline JSON string, `@file`, or `-` (stdin).""" + if data == '-': + raw = sys.stdin.read() + source = 'stdin' + elif data.startswith('@'): + file_path = data[1:] + try: + with open(file_path, encoding='utf-8') as f: + raw = f.read() + except OSError as e: + raise click.ClickException(f'Could not read request body file "{file_path}": {e}') from e + source = file_path + else: + raw = data + source = 'the --data value' + + try: + return json.loads(raw) + except json.JSONDecodeError as e: + raise click.ClickException(f'Could not parse JSON from {source}: {e}') from e + + +def _echo_response_body(response: 'Response') -> None: + try: + click.echo(json.dumps(response.json(), indent=2)) + except ValueError: + # Not a JSON body (empty response, plain text, file download, etc.) + click.echo(response.text) + + +def _callback( + method: str, + path: str, + query: tuple[str, ...], + header: tuple[str, ...], + data: Optional[str], + timeout: Optional[int], +) -> None: + from cycode.cli.exceptions.custom_exceptions import RequestHttpError + from cycode.cli.utils.get_api_client import get_raw_api_client + + method = method.lower() + url_path = _validate_path(path) + params = _parse_query(query) + headers = _parse_headers(header) + + body = None + if data is not None: + if method not in _METHODS_WITH_BODY: + raise click.ClickException(f'--data is not supported for the {method} method.') + body = _read_body(data) + + ctx = click.get_current_context() + client = get_raw_api_client(ctx.find_root()) + + kwargs: dict[str, Any] = {'headers': headers or None, 'params': params or None} + if timeout is not None: + kwargs['timeout'] = timeout + + logger.debug('Sending raw API request, %s', {'method': method, 'path': url_path}) + + try: + if method == 'get': + response = client.get(url_path, **kwargs) + elif method == 'post': + response = client.post(url_path, body=body, **kwargs) + else: + response = client.put(url_path, body=body, **kwargs) + except RequestHttpError as e: + click.echo(f'HTTP {e.status_code}: {e.error_message}', err=True) + raise click.exceptions.Exit(1) from e + except Exception as e: + click.echo(f'Error: {e}', err=True) + raise click.exceptions.Exit(1) from e + + _echo_response_body(response) + + +def build_raw_api_command() -> click.Command: + """Build the `cycode platform api` raw request command.""" + return click.Command( + name='api', + callback=_callback, + help=_HELP, + short_help='[BETA] Send a raw authenticated request to the Cycode API.', + params=[ + click.Argument(['method'], type=click.Choice(_SUPPORTED_METHODS, case_sensitive=False), required=True), + click.Argument(['path'], type=click.STRING, required=True), + click.Option( + ['-q', '--query'], + multiple=True, + metavar='KEY=VALUE', + help='Query parameter. Repeatable; repeating a key sends multiple values.', + ), + click.Option( + ['-H', '--header'], + multiple=True, + metavar='"KEY: VALUE"', + help='Additional request header. Repeatable. Authorization is managed by the CLI.', + ), + click.Option( + ['-d', '--data'], + metavar='JSON', + help='JSON request body for post/put. Use @file to read a file, or - to read stdin.', + ), + click.Option( + ['--timeout'], + type=click.INT, + help='Request timeout in seconds. Defaults to the CLI request timeout.', + ), + ], + ) diff --git a/cycode/cli/utils/get_api_client.py b/cycode/cli/utils/get_api_client.py index b69666d3..5f99d250 100644 --- a/cycode/cli/utils/get_api_client.py +++ b/cycode/cli/utils/get_api_client.py @@ -1,4 +1,4 @@ -from typing import TYPE_CHECKING, Optional, Union +from typing import TYPE_CHECKING, Callable, Optional, TypeVar, Union import click @@ -6,6 +6,7 @@ from cycode.cyclient.client_creator import ( create_ai_security_manager_client, create_import_sbom_client, + create_raw_api_client, create_report_client, create_scan_client, ) @@ -14,18 +15,21 @@ import typer from cycode.cyclient.ai_security_manager_client import AISecurityManagerClient + from cycode.cyclient.cycode_client_base import CycodeClientBase from cycode.cyclient.import_sbom_client import ImportSbomClient from cycode.cyclient.report_client import ReportClient from cycode.cyclient.scan_client import ScanClient +_ClientT = TypeVar('_ClientT') + def _get_cycode_client( - create_client_func: callable, + create_client_func: Callable[..., _ClientT], client_id: Optional[str], client_secret: Optional[str], hide_response_log: bool, id_token: Optional[str] = None, -) -> Union['ScanClient', 'ReportClient', 'ImportSbomClient', 'AISecurityManagerClient']: +) -> _ClientT: if client_id and id_token: return create_client_func(client_id, None, hide_response_log, id_token) @@ -75,6 +79,14 @@ def get_ai_security_manager_client(ctx: 'typer.Context', hide_response_log: bool return _get_cycode_client(create_ai_security_manager_client, client_id, client_secret, hide_response_log, id_token) +def get_raw_api_client(ctx: Union['typer.Context', click.Context]) -> 'CycodeClientBase': + client_id = ctx.obj.get('client_id') if ctx.obj else None + client_secret = ctx.obj.get('client_secret') if ctx.obj else None + id_token = ctx.obj.get('id_token') if ctx.obj else None + # hide_response_log is unused by create_raw_api_client: raw responses are the command's output + return _get_cycode_client(create_raw_api_client, client_id, client_secret, True, id_token) + + def _get_configured_credentials() -> tuple[str, str]: credentials_manager = CredentialsManager() return credentials_manager.get_credentials() diff --git a/cycode/cyclient/client_creator.py b/cycode/cyclient/client_creator.py index c26795c7..fd1454d2 100644 --- a/cycode/cyclient/client_creator.py +++ b/cycode/cyclient/client_creator.py @@ -7,6 +7,7 @@ ) from cycode.cyclient.config import dev_mode from cycode.cyclient.config_dev import DEV_CYCODE_API_URL +from cycode.cyclient.cycode_client_base import CycodeClientBase from cycode.cyclient.cycode_dev_based_client import CycodeDevBasedClient from cycode.cyclient.cycode_oidc_based_client import CycodeOidcBasedClient from cycode.cyclient.cycode_token_based_client import CycodeTokenBasedClient @@ -56,6 +57,17 @@ def create_import_sbom_client( return ImportSbomClient(client) +def create_raw_api_client( + client_id: str, client_secret: Optional[str] = None, _: bool = False, id_token: Optional[str] = None +) -> CycodeClientBase: + """Create an authenticated client without any service wrapper, for raw API requests.""" + if dev_mode: + return CycodeDevBasedClient(DEV_CYCODE_API_URL) + if id_token: + return CycodeOidcBasedClient(client_id, id_token) + return CycodeTokenBasedClient(client_id, client_secret) + + def create_ai_security_manager_client( client_id: str, client_secret: Optional[str] = None, _: bool = False, id_token: Optional[str] = None ) -> AISecurityManagerClient: diff --git a/tests/cli/apps/api/test_raw_api_command.py b/tests/cli/apps/api/test_raw_api_command.py new file mode 100644 index 00000000..6678c734 --- /dev/null +++ b/tests/cli/apps/api/test_raw_api_command.py @@ -0,0 +1,320 @@ +"""Tests for the raw API passthrough command (`cycode platform api`).""" + +import json +from pathlib import Path +from typing import TYPE_CHECKING + +import click +import pytest +import responses +from typer.testing import CliRunner + +from cycode.cli.app import app +from cycode.cli.apps.api.raw_api_command import ( + _parse_headers, + _parse_query, + _read_body, + _validate_path, +) +from tests.conftest import CLI_ENV_VARS + +if TYPE_CHECKING: + from cycode.cyclient.cycode_token_based_client import CycodeTokenBasedClient + +# --- _validate_path --- + + +def test_validate_path_strips_leading_slash() -> None: + assert _validate_path('/v4/projects') == 'v4/projects' + + +@pytest.mark.parametrize('path', ['v4/projects', 'api/v4/projects', 'v4/api-docs/cycode-api-swagger.json']) +def test_validate_path_accepts_versioned_api_path(path: str) -> None: + assert _validate_path(path) == path + + +@pytest.mark.parametrize( + 'path', + [ + 'https://evil.example/v4/x', + 'http://localhost/v4/x', + '//evil.example/v4/x', + 'v4/projects and more', + 'projects', + 'v4', + 'v4/', + 'api/projects', + 'v1/auth/api-token', + ], +) +def test_validate_path_rejects_non_api_path(path: str) -> None: + with pytest.raises(click.ClickException): + _validate_path(path) + + +# --- _parse_query --- + + +def test_parse_query_single_pair() -> None: + assert _parse_query(('page-size=5',)) == {'page-size': '5'} + + +def test_parse_query_value_with_equals_sign() -> None: + assert _parse_query(('filter=a=b',)) == {'filter': 'a=b'} + + +def test_parse_query_empty_value() -> None: + assert _parse_query(('name=',)) == {'name': ''} + + +def test_parse_query_repeated_key_collapses_to_list() -> None: + assert _parse_query(('severity=High', 'severity=Critical', 'severity=Low')) == { + 'severity': ['High', 'Critical', 'Low'] + } + + +@pytest.mark.parametrize('item', ['page-size', '=5', '']) +def test_parse_query_invalid_pair(item: str) -> None: + with pytest.raises(click.ClickException): + _parse_query((item,)) + + +# --- _parse_headers --- + + +def test_parse_headers_strips_whitespace() -> None: + assert _parse_headers(('X-Foo: bar ',)) == {'X-Foo': 'bar'} + + +def test_parse_headers_value_with_colon() -> None: + assert _parse_headers(('X-Url: https://example.com',)) == {'X-Url': 'https://example.com'} + + +@pytest.mark.parametrize('item', ['Authorization: Bearer x', 'authorization: Bearer x']) +def test_parse_headers_rejects_authorization(item: str) -> None: + with pytest.raises(click.ClickException): + _parse_headers((item,)) + + +@pytest.mark.parametrize('item', ['X-Foo', ': bar']) +def test_parse_headers_invalid_pair(item: str) -> None: + with pytest.raises(click.ClickException): + _parse_headers((item,)) + + +# --- _read_body --- + + +def test_read_body_inline_json() -> None: + assert _read_body('{"a": 1}') == {'a': 1} + + +def test_read_body_from_file(tmp_path: Path) -> None: + body_file = tmp_path.joinpath('body.json') + body_file.write_text('{"a": [1, 2]}', encoding='utf-8') + + assert _read_body(f'@{body_file}') == {'a': [1, 2]} + + +def test_read_body_missing_file(tmp_path: Path) -> None: + with pytest.raises(click.ClickException): + _read_body(f'@{tmp_path.joinpath("nope.json")}') + + +def test_read_body_invalid_json() -> None: + with pytest.raises(click.ClickException): + _read_body('not json') + + +# --- end-to-end --- + + +@responses.activate +def test_raw_api_get_prints_json( + token_based_client: 'CycodeTokenBasedClient', api_token_response: responses.Response +) -> None: + # The OpenAPI spec URL is deliberately not mocked: `platform api` must not fetch the spec + responses.add(api_token_response) + responses.add( + responses.Response( + method=responses.GET, + url=f'{token_based_client.api_url}/v4/projects', + json={'items': [{'id': '1'}]}, + status=200, + ) + ) + + result = CliRunner().invoke(app, ['platform', 'api', 'get', 'v4/projects'], env=CLI_ENV_VARS) + + assert result.exit_code == 0, result.output + assert json.loads(result.output) == {'items': [{'id': '1'}]} + + +@responses.activate +def test_raw_api_get_sends_query_params( + token_based_client: 'CycodeTokenBasedClient', api_token_response: responses.Response +) -> None: + responses.add(api_token_response) + responses.add( + responses.Response( + method=responses.GET, + url=f'{token_based_client.api_url}/v4/violations', + json={'items': []}, + status=200, + ) + ) + + args = ['platform', 'api', 'get', '/v4/violations', '-q', 'severity=High', '-q', 'severity=Critical'] + result = CliRunner().invoke(app, args, env=CLI_ENV_VARS) + + assert result.exit_code == 0, result.output + request_url = responses.calls[-1].request.url + assert 'severity=High' in request_url + assert 'severity=Critical' in request_url + + +@responses.activate +def test_raw_api_post_sends_body( + token_based_client: 'CycodeTokenBasedClient', api_token_response: responses.Response +) -> None: + responses.add(api_token_response) + responses.add( + responses.Response( + method=responses.POST, + url=f'{token_based_client.api_url}/v4/sbom/import', + json={'id': 'abc'}, + status=200, + ) + ) + + args = ['platform', 'api', 'post', 'v4/sbom/import', '-d', '{"name": "test"}'] + result = CliRunner().invoke(app, args, env=CLI_ENV_VARS) + + assert result.exit_code == 0, result.output + assert json.loads(responses.calls[-1].request.body) == {'name': 'test'} + + +@responses.activate +def test_raw_api_sends_additional_header( + token_based_client: 'CycodeTokenBasedClient', api_token_response: responses.Response +) -> None: + responses.add(api_token_response) + responses.add( + responses.Response( + method=responses.GET, + url=f'{token_based_client.api_url}/v4/projects', + json={}, + status=200, + ) + ) + + args = ['platform', 'api', 'get', 'v4/projects', '-H', 'X-Foo: bar'] + result = CliRunner().invoke(app, args, env=CLI_ENV_VARS) + + assert result.exit_code == 0, result.output + request_headers = responses.calls[-1].request.headers + assert request_headers['X-Foo'] == 'bar' + assert request_headers['Authorization'].startswith('Bearer ') + + +@responses.activate +def test_raw_api_put_reads_body_from_stdin( + token_based_client: 'CycodeTokenBasedClient', api_token_response: responses.Response +) -> None: + responses.add(api_token_response) + responses.add( + responses.Response( + method=responses.PUT, + url=f'{token_based_client.api_url}/v4/some/resource', + json={}, + status=200, + ) + ) + + args = ['platform', 'api', 'put', 'v4/some/resource', '-d', '-'] + result = CliRunner().invoke(app, args, env=CLI_ENV_VARS, input='{"from": "stdin"}') + + assert result.exit_code == 0, result.output + assert json.loads(responses.calls[-1].request.body) == {'from': 'stdin'} + + +@responses.activate +def test_raw_api_timeout_option_is_forwarded( + token_based_client: 'CycodeTokenBasedClient', api_token_response: responses.Response +) -> None: + responses.add(api_token_response) + responses.add( + responses.Response( + method=responses.GET, + url=f'{token_based_client.api_url}/v4/projects', + json={}, + status=200, + ) + ) + + args = ['platform', 'api', 'get', 'v4/projects', '--timeout', '7'] + result = CliRunner().invoke(app, args, env=CLI_ENV_VARS) + + assert result.exit_code == 0, result.output + assert responses.calls[-1].request.req_kwargs['timeout'] == 7 + + +@responses.activate +def test_raw_api_non_json_response_prints_text( + token_based_client: 'CycodeTokenBasedClient', api_token_response: responses.Response +) -> None: + responses.add(api_token_response) + responses.add( + responses.Response( + method=responses.GET, + url=f'{token_based_client.api_url}/v4/plain', + body='plain text', + status=200, + ) + ) + + result = CliRunner().invoke(app, ['platform', 'api', 'get', 'v4/plain'], env=CLI_ENV_VARS) + + assert result.exit_code == 0, result.output + assert 'plain text' in result.output + + +@responses.activate +def test_raw_api_http_error_exits_non_zero( + token_based_client: 'CycodeTokenBasedClient', api_token_response: responses.Response +) -> None: + responses.add(api_token_response) + responses.add( + responses.Response( + method=responses.GET, + url=f'{token_based_client.api_url}/v4/does-not-exist', + json={'message': 'not found'}, + status=404, + ) + ) + + result = CliRunner().invoke(app, ['platform', 'api', 'get', 'v4/does-not-exist'], env=CLI_ENV_VARS) + + assert result.exit_code != 0 + assert '404' in result.output + + +@pytest.mark.parametrize('path', ['https://evil.example/v4/x', 'projects']) +def test_raw_api_rejects_non_api_path(path: str) -> None: + result = CliRunner().invoke(app, ['platform', 'api', 'get', path], env=CLI_ENV_VARS) + + assert result.exit_code != 0 + assert 'must be a versioned API path' in result.output + + +def test_raw_api_rejects_data_with_get() -> None: + result = CliRunner().invoke(app, ['platform', 'api', 'get', 'v4/projects', '-d', '{}'], env=CLI_ENV_VARS) + + assert result.exit_code != 0 + assert '--data is not supported' in result.output + + +def test_raw_api_rejects_unsupported_method() -> None: + result = CliRunner().invoke(app, ['platform', 'api', 'delete', 'v4/projects'], env=CLI_ENV_VARS) + + assert result.exit_code != 0 From dea4898ccf3dc3646b756b894928c63c58c45979 Mon Sep 17 00:00:00 2001 From: "omer.roth" Date: Tue, 28 Jul 2026 11:34:58 +0300 Subject: [PATCH 2/2] CM-69684 fixed tests --- tests/cli/apps/api/test_raw_api_command.py | 70 ++++++++++++++++------ 1 file changed, 51 insertions(+), 19 deletions(-) diff --git a/tests/cli/apps/api/test_raw_api_command.py b/tests/cli/apps/api/test_raw_api_command.py index 6678c734..9061e523 100644 --- a/tests/cli/apps/api/test_raw_api_command.py +++ b/tests/cli/apps/api/test_raw_api_command.py @@ -1,5 +1,6 @@ """Tests for the raw API passthrough command (`cycode platform api`).""" +import contextlib import json from pathlib import Path from typing import TYPE_CHECKING @@ -19,8 +20,35 @@ from tests.conftest import CLI_ENV_VARS if TYPE_CHECKING: + from click.testing import Result + from cycode.cyclient.cycode_token_based_client import CycodeTokenBasedClient + +# The on-close update check requests PyPI (VersionChecker.PYPI_REQUEST_TIMEOUT = 1), +# which would otherwise show up in responses.calls and pollute request assertions. +_ROOT_ARGS = ['--no-update-notifier'] + + +def _all_output(result: 'Result') -> str: + """Combine stdout and stderr. + + Click 8.1 mixes stderr into `output`, while Click 8.2+ captures it separately. + """ + parts = [result.output] + with contextlib.suppress(AttributeError, ValueError): + parts.append(result.stderr) # raises when stderr was mixed into output already + + return ''.join(parts) + + +def _find_call(url: str) -> responses.Call: + """Find the recorded request for a URL, ignoring auth and version-check traffic.""" + matching = [call for call in responses.calls if call.request.url.split('?')[0] == url] + assert matching, f'no recorded request for {url}. Recorded: {[c.request.url for c in responses.calls]}' + return matching[-1] + + # --- _validate_path --- @@ -144,7 +172,7 @@ def test_raw_api_get_prints_json( ) ) - result = CliRunner().invoke(app, ['platform', 'api', 'get', 'v4/projects'], env=CLI_ENV_VARS) + result = CliRunner().invoke(app, [*_ROOT_ARGS, 'platform', 'api', 'get', 'v4/projects'], env=CLI_ENV_VARS) assert result.exit_code == 0, result.output assert json.loads(result.output) == {'items': [{'id': '1'}]} @@ -164,11 +192,11 @@ def test_raw_api_get_sends_query_params( ) ) - args = ['platform', 'api', 'get', '/v4/violations', '-q', 'severity=High', '-q', 'severity=Critical'] + args = [*_ROOT_ARGS, 'platform', 'api', 'get', '/v4/violations', '-q', 'severity=High', '-q', 'severity=Critical'] result = CliRunner().invoke(app, args, env=CLI_ENV_VARS) assert result.exit_code == 0, result.output - request_url = responses.calls[-1].request.url + request_url = _find_call(f'{token_based_client.api_url}/v4/violations').request.url assert 'severity=High' in request_url assert 'severity=Critical' in request_url @@ -187,11 +215,12 @@ def test_raw_api_post_sends_body( ) ) - args = ['platform', 'api', 'post', 'v4/sbom/import', '-d', '{"name": "test"}'] + args = [*_ROOT_ARGS, 'platform', 'api', 'post', 'v4/sbom/import', '-d', '{"name": "test"}'] result = CliRunner().invoke(app, args, env=CLI_ENV_VARS) assert result.exit_code == 0, result.output - assert json.loads(responses.calls[-1].request.body) == {'name': 'test'} + request = _find_call(f'{token_based_client.api_url}/v4/sbom/import').request + assert json.loads(request.body) == {'name': 'test'} @responses.activate @@ -208,11 +237,11 @@ def test_raw_api_sends_additional_header( ) ) - args = ['platform', 'api', 'get', 'v4/projects', '-H', 'X-Foo: bar'] + args = [*_ROOT_ARGS, 'platform', 'api', 'get', 'v4/projects', '-H', 'X-Foo: bar'] result = CliRunner().invoke(app, args, env=CLI_ENV_VARS) assert result.exit_code == 0, result.output - request_headers = responses.calls[-1].request.headers + request_headers = _find_call(f'{token_based_client.api_url}/v4/projects').request.headers assert request_headers['X-Foo'] == 'bar' assert request_headers['Authorization'].startswith('Bearer ') @@ -231,11 +260,12 @@ def test_raw_api_put_reads_body_from_stdin( ) ) - args = ['platform', 'api', 'put', 'v4/some/resource', '-d', '-'] + args = [*_ROOT_ARGS, 'platform', 'api', 'put', 'v4/some/resource', '-d', '-'] result = CliRunner().invoke(app, args, env=CLI_ENV_VARS, input='{"from": "stdin"}') assert result.exit_code == 0, result.output - assert json.loads(responses.calls[-1].request.body) == {'from': 'stdin'} + request = _find_call(f'{token_based_client.api_url}/v4/some/resource').request + assert json.loads(request.body) == {'from': 'stdin'} @responses.activate @@ -252,11 +282,12 @@ def test_raw_api_timeout_option_is_forwarded( ) ) - args = ['platform', 'api', 'get', 'v4/projects', '--timeout', '7'] + args = [*_ROOT_ARGS, 'platform', 'api', 'get', 'v4/projects', '--timeout', '7'] result = CliRunner().invoke(app, args, env=CLI_ENV_VARS) assert result.exit_code == 0, result.output - assert responses.calls[-1].request.req_kwargs['timeout'] == 7 + request = _find_call(f'{token_based_client.api_url}/v4/projects').request + assert request.req_kwargs['timeout'] == 7 @responses.activate @@ -273,7 +304,7 @@ def test_raw_api_non_json_response_prints_text( ) ) - result = CliRunner().invoke(app, ['platform', 'api', 'get', 'v4/plain'], env=CLI_ENV_VARS) + result = CliRunner().invoke(app, [*_ROOT_ARGS, 'platform', 'api', 'get', 'v4/plain'], env=CLI_ENV_VARS) assert result.exit_code == 0, result.output assert 'plain text' in result.output @@ -293,28 +324,29 @@ def test_raw_api_http_error_exits_non_zero( ) ) - result = CliRunner().invoke(app, ['platform', 'api', 'get', 'v4/does-not-exist'], env=CLI_ENV_VARS) + result = CliRunner().invoke(app, [*_ROOT_ARGS, 'platform', 'api', 'get', 'v4/does-not-exist'], env=CLI_ENV_VARS) assert result.exit_code != 0 - assert '404' in result.output + assert '404' in _all_output(result) @pytest.mark.parametrize('path', ['https://evil.example/v4/x', 'projects']) def test_raw_api_rejects_non_api_path(path: str) -> None: - result = CliRunner().invoke(app, ['platform', 'api', 'get', path], env=CLI_ENV_VARS) + result = CliRunner().invoke(app, [*_ROOT_ARGS, 'platform', 'api', 'get', path], env=CLI_ENV_VARS) assert result.exit_code != 0 - assert 'must be a versioned API path' in result.output + assert 'must be a versioned API path' in _all_output(result) def test_raw_api_rejects_data_with_get() -> None: - result = CliRunner().invoke(app, ['platform', 'api', 'get', 'v4/projects', '-d', '{}'], env=CLI_ENV_VARS) + args = [*_ROOT_ARGS, 'platform', 'api', 'get', 'v4/projects', '-d', '{}'] + result = CliRunner().invoke(app, args, env=CLI_ENV_VARS) assert result.exit_code != 0 - assert '--data is not supported' in result.output + assert '--data is not supported' in _all_output(result) def test_raw_api_rejects_unsupported_method() -> None: - result = CliRunner().invoke(app, ['platform', 'api', 'delete', 'v4/projects'], env=CLI_ENV_VARS) + result = CliRunner().invoke(app, [*_ROOT_ARGS, 'platform', 'api', 'delete', 'v4/projects'], env=CLI_ENV_VARS) assert result.exit_code != 0