diff --git a/AGENTS.md b/AGENTS.md index e588d38..c04e817 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -89,8 +89,9 @@ Mirror the existing patterns (e.g. `field-property`, `prompt-config`, `ruleset`) Details in [`specs/04-testing.md`](./specs/04-testing.md). The shape: - Tests live in `tests/` and drive command behaviour through `click.testing.CliRunner` — no live PolySwarm stack needed. -- Two styles for that: **SDK-boundary mocks** (`mock.patch('polyswarm_api.api.PolyswarmAPI.')`, e.g. `tests/field_property_test.py`) and **VCR cassettes** (`tests/cli_test.py`) that replay recorded HTTP for end-to-end CLI runs. Cassettes live in `tests/vcr/` (`.vcr` for the HTTP interactions, `.click` for the expected rendered output). +- Three styles for that: **SDK-boundary mocks** (`mock.patch('polyswarm_api.api.PolyswarmAPI.')`, e.g. `tests/field_property_test.py`), **SDK-transport mocks** (`PolyswarmSession.execute`, e.g. `tests/refang_test.py`, for behaviour that runs inside an SDK method; see the next bullet) and **VCR cassettes** (`tests/cli_test.py`) that replay recorded HTTP for end-to-end CLI runs. Cassettes live in `tests/vcr/` (`.vcr` for the HTTP interactions, `.click` for the expected rendered output). - Pure rendering logic — which line a given field set produces, no command-tree behaviour — is instead unit-tested against the formatter directly (e.g. `tests/known_good_field_test.py`); see the spec's *Style 3* for when that's the right choice. +- Behaviour that runs *inside* an SDK endpoint method (IoC refanging) is tested at the SDK transport instead — `PolyswarmSession.execute` — because an SDK-method mock would replace the code under test (e.g. `tests/refang_test.py`); see the spec's *Style 4*. - VCR is an **efficiency cache, not a requirement** — the suite must pass against a live e2e stack with VCR off. Re-record a cassette by deleting it and re-running the test against a live stack; never hand-edit a cassette or `cp` one from a sibling test. ## Commit + PR hygiene diff --git a/pyproject.toml b/pyproject.toml index 75dbe5f..11692aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,7 @@ classifiers = [ ] dependencies = [ - "polyswarm_api>=4.5.0,<5.0.0", + "polyswarm_api>=4.6.0,<5.0.0", "click>=7.1", "colorama>=0.4.6", "click-log>=0.4.0", diff --git a/specs/01-architecture.md b/specs/01-architecture.md index b888079..af5ec3b 100644 --- a/specs/01-architecture.md +++ b/specs/01-architecture.md @@ -16,7 +16,7 @@ The components of the CLI and how a command flows from `argv` to rendered output `polyswarm_cli` is the top-level `click.Group`, constructed with `cls=ExceptionHandlingGroup`. It: -1. Declares the **global options** — `--api-key` (env `POLYSWARM_API_KEY`), `--api-uri` (env `POLYSWARM_API_URI`), the **endpoint shortcuts** `--prod` / `--stage` / `--local` / `--prod-eu` / `--stage-eu`, `--output-file`, `--output-format`/`--fmt` (`text`|`json`|…), `--color/--no-color`, `--verbose`, `--community` (env `POLYSWARM_COMMUNITY`), `--parallel`, `--verify/--no-verify`, plus `--version` / `--api-version`. +1. Declares the **global options** — `--api-key` (env `POLYSWARM_API_KEY`), `--api-uri` (env `POLYSWARM_API_URI`), the **endpoint shortcuts** `--prod` / `--stage` / `--local` / `--prod-eu` / `--stage-eu`, `--output-file`, `--output-format`/`--fmt` (`text`|`json`|…), `--color/--no-color`, `--verbose`, `--community` (env `POLYSWARM_COMMUNITY`), `--parallel`, `--verify/--no-verify`, `--refang/--no-refang` (default on; see [`02-commands.md`](./02-commands.md) §Global `--refang/--no-refang`), plus `--version` / `--api-version`. **Endpoint resolution** (`resolve_api_uri`): the shortcuts are convenience aliases for known public endpoints (`API_URI_SHORTCUTS`); `--prod` is an explicit alias for the production endpoint (identical to the no-flag default, offered for symmetry). Precedence is **explicit command-line flag → `POLYSWARM_API_URI` env var → production default** (`PROD_API_URI` = `https://api.polyswarm.network/v3`). Specifically: a shortcut and an explicit *command-line* `--api-uri` are mutually exclusive (conflict → `click.UsageError`, exit 2), as are two shortcuts; a shortcut **wins over** an ambient `POLYSWARM_API_URI` (the env var is consulted only when no shortcut is given) — so `--prod` forces production even when the env var points elsewhere; and a command-line `--api-uri` wins over the env var (click's own source precedence). The command-line-vs-env distinction uses `ctx.get_parameter_source('api_uri') == ParameterSource.COMMANDLINE`. 2. **Seeds `ctx.obj`** — constructs a `Polyswarm(...)` client (the SDK wrapper) as `ctx.obj['api']` and the selected formatter as `ctx.obj['output']`. @@ -82,13 +82,13 @@ The catalogue of groups and the SDK methods each wraps is in [`02-commands.md`]( ## Support — `utils.py`, `exceptions.py` -- **`utils.py`** — `parallelize`/`parallel_executor` (thread-pool fan-out with per-item exception aggregation: collects results, logs per-item no-results, raises an aggregate `NoResultsException`/`NotFoundException`/`InternalFailureException` at the end) and `parallel_executor_iterable_results` (the same, for SDK methods that return generators — it materialises each generator inside the worker so per-item exception handling still fires), plus `collect_files` and the detection helpers (`is_valid_id`, `is_ip`, `is_domain`, `is_url`). +- **`utils.py`** — `parallelize`/`parallel_executor` (thread-pool fan-out with per-item exception aggregation: collects results, logs per-item no-results, raises an aggregate `NoResultsException`/`NotFoundException`/`InternalFailureException` at the end) and `parallel_executor_iterable_results` (the same, for SDK methods that return generators — it materialises each generator inside the worker so per-item exception handling still fires), plus `collect_files` and the detection helpers (`is_valid_id`, `is_ip`, `is_domain`, `is_url`), and `refang_input(api, value)` — not a detection helper but the refang one: it applies the SDK's `polyswarm_api.refang.refang_ioc` only when the client's `refang_iocs` is on, for the few places the CLI handles a value before (or instead of) an SDK endpoint method. - **`client/utils.py`** — `parse_hashes` and the click parameter validators (`validate_id`, `validate_hash(es)`, `validate_key`, …). Note the module is `client/utils.py`, not the top-level `utils.py` above — the two are distinct and easily confused. - **`exceptions.py`** — the CLI's own hierarchy, **distinct from the SDK's**: `PolyswarmException` → `NoResultsException`, `NotFoundException`, `InternalFailureException`, `PartialResultsException`. `ExceptionHandlingGroup` catches both these and the SDK's `api_exceptions.*`. ## Lifecycle of a command (end to end) -1. `polyswarm_cli` parses global options, builds `Polyswarm(api_key, uri=…, community=…, parallel=…, verify=…)` and the formatter into `ctx.obj`. +1. `polyswarm_cli` parses global options, builds `Polyswarm(api_key, uri=…, community=…, parallel=…, verify=…, refang_iocs=…)` and the formatter into `ctx.obj`. 2. The subcommand reads `api`/`output` from `ctx.obj`, parses its own args, and calls one or more SDK methods (directly or via a wrapper fan-out method). 3. Results are rendered through the formatter; collections are iterated. 4. Any exception propagates to `ExceptionHandlingGroup.invoke`, which logs it and raises `Exit()`. diff --git a/specs/02-commands.md b/specs/02-commands.md index 62a62aa..a41b8c7 100644 --- a/specs/02-commands.md +++ b/specs/02-commands.md @@ -62,6 +62,19 @@ The top-level command groups, what each is for, and the primary `polyswarm-api` > likeliest origin of the original mistake — check which endpoint you are on > before copying a default between them. +## Global `--refang/--no-refang` (IoC refanging) + +Threat-intel reports print indicators defanged (`hxxps[:]//evil[.]com`, `127[.]0[.]0[.]1`); pasted verbatim they never match a search, and a submitted one becomes a broken URL artifact. The root group's `--refang/--no-refang` (default on) is passed to the client as `Polyswarm(..., refang_iocs=…)`. The SDK's own default is off (its refanging is opt-in, so the SDK release stays a minor bump); the CLI opts in unless `--no-refang` is given, and the SDK refangs the URL / domain / IP inputs of its own endpoint methods (`polyswarm_api.refang`; rules and gate in the SDK's downstream-contract spec). So `search url`, `search metadata -p/-u/-d` (never the free-form query), `search ioc ip|domain`, `search known`, `known add/update`, and the `-r/--url-file` lines of `scan url` (which reach the wire only through the SDK's `submit`) need no CLI code. Positional `scan url` / `sandbox url` URLs are the exception below. + +**Write paths change too, not only searches.** With refanging on (the default), `known add` / `known update` store the live host (`known add domain evil[.]com feed` stores `evil.com`), so rows an earlier CLI wrote verbatim as `evil[.]com` are no longer reachable from `search known -d evil[.]com`, which now sends `evil.com`. And `scan url` / `sandbox url` given a defanged URL submit a different artifact (content, `artifact_name`, sha — and the quota it costs) than the same invocation did before. Pass `--no-refang` to keep the verbatim behaviour, e.g. to reach catalogue rows stored defanged. + +Two places handle the value in the CLI and call `utils.refang_input(api, value)` — the SDK's `refang_ioc`, gated on `api.refang_iocs`: + +- **Validation before the SDK sees the value.** `scan url` and `sandbox url` check positional URLs with `is_url`; they validate the refanged form, so `hxxps[:]//evil[.]com` is accepted instead of rejected as invalid. With `--no-refang` the defanged URL is still rejected, exactly as before. +- **CLI-owned requests.** `metadata analyze-ip` goes through `Polyswarm.submit_url`, which builds its request with `_single` rather than an SDK endpoint method, so it refangs explicitly. + +Hashes and ids are never touched. Neither is a `--qrcode-file` path on `scan url` / `sandbox url`, and that exemption lives in the SDK, not in CLI code: a `preprocessing={'type': 'qrcode'}` submission skips refanging entirely (pinned here by the two qrcode tests in `tests/refang_test.py`). Tests: `tests/refang_test.py`, mocking at the SDK transport so the SDK's own refang is exercised — see [`04-testing.md`](./04-testing.md) §Style 4. + ## Adding to the catalogue When you add or materially change a group, update its row (and add per-subcommand detail here if the behaviour is non-obvious). The `AGENTS.md` §"When adding a new command family" checklist covers the wiring + formatter + test steps. diff --git a/specs/04-testing.md b/specs/04-testing.md index f9ae294..6b1823d 100644 --- a/specs/04-testing.md +++ b/specs/04-testing.md @@ -2,12 +2,12 @@ ## Scope -How the CLI is tested: the `CliRunner` harness, the two mocking styles (SDK-boundary mocks vs VCR cassettes), the formatter-unit style for pure rendering, the cassette layout and record workflow, and how to run the suite. Files: `tests/`, `tests/vcr/`, `src/conftest.py`, `pyproject.toml` (`[project.optional-dependencies].tests`, `[tool.pytest.ini_options]`). +How the CLI is tested: the `CliRunner` harness, the mocking styles (SDK-boundary mocks, SDK-transport mocks, VCR cassettes), the formatter-unit style for pure rendering, the cassette layout and record workflow, and how to run the suite. Files: `tests/`, `tests/vcr/`, `src/conftest.py`, `pyproject.toml` (`[project.optional-dependencies].tests`, `[tool.pytest.ini_options]`). ## Invariants - **Anything that is command behaviour is driven through `click.testing.CliRunner`** — argument parsing, the SDK call, the wiring, the exit code: exercise the real command tree, never an internal function standing in for it. No live PolySwarm stack is required. The one sanctioned exception is pure rendering logic — see [Style 3](#style-3--formatter-unit-tests). -- **Mock at the SDK boundary, or replay HTTP with VCR — never both for the same path.** A test either patches `polyswarm_api.api.PolyswarmAPI.` (unit-style) or lets VCR replay recorded HTTP (end-to-end). The CLI's own code is exercised either way. +- **Mock at exactly one point per path — the SDK method, the SDK transport, or VCR.** A test patches `polyswarm_api.api.PolyswarmAPI.` (unit-style, the default), patches the SDK transport `PolyswarmSession.execute` when the behaviour under test runs *inside* the SDK method ([Style 4](#style-4--sdk-transport-mocks)), or lets VCR replay recorded HTTP (end-to-end). Never two of them for the same path. The CLI's own code is exercised either way. - **VCR is an efficiency cache, not a load-bearing requirement.** The suite must pass against a live e2e stack with VCR off. Don't hardcode `record_mode='none'`; if a test only works against its recorded cassette, that's a bug in the test. Note this is about a test's *logic*, not its fixtures: a `.click` snapshot pins server-generated ids and timestamps, so re-recording needs a stack in a particular state — see [Re-recording a cassette](#re-recording-a-cassette). @@ -84,6 +84,14 @@ For **rendering logic with no command-tree behaviour** — which labelled line a Use it **only** for that. Argument parsing, SDK calls, generator consumption, `ctx.obj` wiring and exit codes are command behaviour: a formatter unit test can't observe them, so those need Style 1 or Style 2. A command whose rendering is covered by Style 3 still needs at least one `CliRunner` test proving the command reaches the formatter at all. +## Style 4 — SDK-transport mocks + +For behaviour that happens **inside** an SDK endpoint method, a Style 1 mock proves nothing: patching `PolyswarmAPI.search_url` replaces the very code under test. IoC refanging is the case today — the SDK refangs `search_url`'s argument (and the other URL / domain / IP inputs) before it builds the request, so the observable effect is the request that reaches the transport. + +Patch `polyswarm_api.session.PolyswarmSession.execute` — the SDK's documented session customization point — and assert on the `PolyswarmRequest` it receives: `request.params` (query string) and `request.input_json` (JSON body), read-only. Example: `tests/refang_test.py`. + +Use this style only when Style 1 would bypass the behaviour under test. Moving these tests "up" to endpoint-method mocks would keep them green while dropping their coverage. The dependency it adds (the `execute(request)` seam and those two request fields) is recorded in [`05-sdk-contract.md`](./05-sdk-contract.md); an SDK rename there breaks these tests, which is the intended signal. + ## What to test for a new command 1. The command parses its arguments and calls the expected SDK method with the expected arguments. diff --git a/specs/05-sdk-contract.md b/specs/05-sdk-contract.md index ed2ae90..563764e 100644 --- a/specs/05-sdk-contract.md +++ b/specs/05-sdk-contract.md @@ -21,6 +21,7 @@ How the CLI depends on the `polyswarm-api` SDK: which parts of the SDK's public | `from polyswarm_api import exceptions as api_exceptions` | Caught in `ExceptionHandlingGroup` and `utils.parallel_executor` (`NoResultsException`, `NotFoundException`, `FailedInstanceException`, `PolyswarmException`). Also `RequestException`, caught by `rules favorite` (`client/rules.py`) to read the machine-readable `FAVORITE_LIMIT` refusal off `exc.request.errors['code']` — and, when the envelope carries no counters, `exc.request.json['result']` as the server's own message. Note the spelling: the request object exposes the response envelope as `.json` and keeps only a private `._result`, so `exc.request.result` is not a thing — reading it yields `None` silently. The SDK does not raise a typed exception for that refusal by design: `.request.errors` is a plain dict the server's error envelope populates. It is pinned without a recording — the SDK's stubbed-transport suite fixes the wire shape, and both CLI branches are unit-pinned against a real `PolyswarmRequest` (not a hand-built mock, which fabricates whatever attribute it is asked for and so cannot detect a rename). It cannot be cassette-pinned: the server sends the counters on every `FAVORITE_LIMIT`, so the envelope the fallback exists for is one no recording can produce. The fallback is defensive against a server that omits them, and the unit test is what fixes the spelling it reads. | | `from polyswarm_api.core import parse_isoformat` | Date rendering in `formatters/text.py`. | | `import polyswarm_api` (`__version__`) | `--api-version`. | +| `from polyswarm_api import refang` | `utils.refang_input` — the SDK's `refang_ioc`, for the values the CLI handles itself (pre-SDK URL validation, the CLI-owned `submit_url` request); gated on the client's public `refang_iocs` attribute, which `--refang/--no-refang` sets through the constructor. See [`02-commands.md`](./02-commands.md) §Global `--refang/--no-refang`. | All of the above are part of the SDK's documented public surface. If a future change needs something not on that list, that's a signal to add a method/export to the SDK rather than reach into internals. @@ -80,7 +81,9 @@ return self._single( ) ``` -`_single` builds the request descriptor, executes it via `self.session`, and returns the parsed resource. Do **not** import `PolyswarmRequest` and call `.execute()`/`.result()` — those are not part of the supported surface. +`_single` builds the request descriptor, executes it via `self.session`, and returns the parsed resource. Production code must **not** import `PolyswarmRequest` and call `.execute()`/`.result()` itself — those are not part of the supported surface. + +**Test-only dependency:** `tests/refang_test.py` (testing Style 4) patches `PolyswarmSession.execute(request)` and reads `request.params` / `request.input_json`. That is the SDK's session customization point, used here only as a test seam, never called from production code. An SDK rename of the seam or those fields breaks those tests, which is the intended signal. ## Coordinated changes (paired PRs) @@ -105,7 +108,7 @@ When a CLI feature needs an SDK surface that doesn't exist yet: **Read the declared version off the archive's own tree, and mind pre-release suffixes.** PEP 440 orders `4.2.0.dev1 < 4.2.0`, so a `develop` head carrying a dev suffix (the SDK's `pyproject.toml` has a `[tool.bumpversion.parts.dev]`) would *not* satisfy a `>=4.2.0` floor even though it looks like 4.2.0 — and the archive build would be silently replaced from PyPI. Check the version string in the SDK branch's `pyproject.toml` / `__init__.py`, not the last release tag. When the floor was last verified this way both were read from `origin/develop` as `4.2.0`, no suffix; the pin has since moved on (§Current floor is the one authoritative statement of its value), and every bump should be re-checked the same way. -### Current floor — `polyswarm_api>=4.5.0` +### Current floor — `polyswarm_api>=4.6.0` The floor is whatever `pyproject.toml` pins; this header follows it. It lives in ONE authoritative place for a reason — a copy here drifted behind the pin once already. The 4.2.0 rationale below still holds transitively; on 4.1.0 both behaviours fail *silently*, which is why the floor is a hard requirement rather than a preference: @@ -128,8 +131,12 @@ formatters render — are what moved the floor to 4.4.0, together with §Matched strings on hunt results). `rules list --sort active-first` forwards `ruleset_list(sort='active_first')`, a keyword 4.5.0 adds, and that is what moved the floor to 4.5.0 (the `tests/formatter_hunt_fields_test.py` autospec assertion is the -signature check: against a 4.4.0 SDK it fails at the mock, not at the server). Code and -tests use them directly. +signature check: against a 4.4.0 SDK it fails at the mock, not at the server). IoC +refanging — the `polyswarm_api.refang` module, the `refang_iocs=` constructor keyword and +the refang inside the SDK's endpoint methods, which `--refang/--no-refang` relies on — is +4.6.0, and that is what moved the floor to 4.6.0 (on 4.5.0 the constructor rejects +`refang_iocs=` and every command fails). This is a paired change: the SDK branch of the +same name declares 4.6.0. Code and tests use them directly. **Raising the floor is the whole procedure** when this repo needs something new from the SDK: diff --git a/src/polyswarm/client/metadata.py b/src/polyswarm/client/metadata.py index 70fb531..b309176 100644 --- a/src/polyswarm/client/metadata.py +++ b/src/polyswarm/client/metadata.py @@ -49,6 +49,10 @@ def analyze_ip(ctx, url): Creates an ArtifactInstance immediately (no S3 upload), triggers only the IP analyzer, and consumes no quota. + \b + A defanged URL or IP (127[.]0[.]0[.]1) is submitted in its live form; + pass --no-refang to the root command to submit it verbatim. + URL is the URL or IP address to submit for analysis. """ api = ctx.obj['api'] diff --git a/src/polyswarm/client/polyswarm.py b/src/polyswarm/client/polyswarm.py index 8db3c81..73c94fa 100644 --- a/src/polyswarm/client/polyswarm.py +++ b/src/polyswarm/client/polyswarm.py @@ -204,11 +204,14 @@ def invoke(self, ctx): help='Community to use.', show_envvar=True) @click.option('--parallel', default=8, help='Number of threads to be used in parallel http requests.') @click.option('--verify/--no-verify', default=True, help='Verify TLS connections.') +@click.option('--refang/--no-refang', default=True, + help='Refang defanged URL, domain and IP inputs (e.g. hxxps[:]//evil[.]com) ' + 'before sending them. On by default; free-form metadata queries and hashes are never changed.') @click.version_option(polyswarm.__version__, '--version', prog_name='polyswarm-cli') @click.version_option(polyswarm_api.__version__, '--api-version', prog_name='polyswarm-api') @click.pass_context def polyswarm_cli(ctx, api_key, api_uri, output_file, output_format, color, verbose, community, parallel, verify, - prod, stage, local, prod_eu, stage_eu): + refang, prod, stage, local, prod_eu, stage_eu): """ This is a PolySwarm CLI client, which allows you to interact directly with the PolySwarm network to scan files, search hashes, and more. @@ -229,7 +232,8 @@ def polyswarm_cli(ctx, api_key, api_uri, output_file, output_format, color, verb {'prod': prod, 'stage': stage, 'local': local, 'prod_eu': prod_eu, 'stage_eu': stage_eu}) - ctx.obj['api'] = Polyswarm(api_key, uri=api_uri, community=community, parallel=parallel, verify=verify) + ctx.obj['api'] = Polyswarm(api_key, uri=api_uri, community=community, parallel=parallel, verify=verify, + refang_iocs=refang) ctx.obj['output'] = formatters[output_format](color=color, output=output_file) diff --git a/src/polyswarm/client/sandbox.py b/src/polyswarm/client/sandbox.py index 100ca46..e6f8915 100644 --- a/src/polyswarm/client/sandbox.py +++ b/src/polyswarm/client/sandbox.py @@ -4,7 +4,7 @@ from polyswarm.client import utils -from polyswarm.utils import is_url +from polyswarm.utils import is_url, refang_input logger = logging.getLogger(__name__) @@ -116,18 +116,27 @@ def file(ctx, path, provider, vm_slug, internet_disabled, is_zip, zip_password, def url(ctx, url, qrcode_file, provider, vm_slug, browser): """ Submit an url to be sandboxed. + + \b + Defanged URLs (hxxps[:]//evil[.]com) are refanged before they are + validated and submitted, so the live form is what gets sandboxed; pass + --no-refang to the root command to send them verbatim. """ + api = ctx.obj['api'] + output = ctx.obj['output'] if qrcode_file: if url: raise click.BadArgumentUsage('--qrcode-file cannot be used with URL.') preprocessing = {'type': 'qrcode'} else: preprocessing = None + # Refang before validating (see ``refang_input``). + # The error quotes what was typed, not the rewrite. + typed = url + url = refang_input(api, url) if url else url if url and not is_url(url): - raise click.BadArgumentUsage(f'URL "{url}" is not valid. ' + raise click.BadArgumentUsage(f'URL "{typed}" is not valid. ' 'Make sure the protocol "https://" or "http://" is set.') - api = ctx.obj['api'] - output = ctx.obj['output'] output.sandbox_task(api.sandbox_url(url, provider, vm_slug, diff --git a/src/polyswarm/client/scan.py b/src/polyswarm/client/scan.py index b6f0e9e..c494da9 100644 --- a/src/polyswarm/client/scan.py +++ b/src/polyswarm/client/scan.py @@ -5,7 +5,7 @@ from polyswarm_api import settings from polyswarm.client import utils -from polyswarm.utils import is_url +from polyswarm.utils import is_url, refang_input logger = logging.getLogger(__name__) @@ -91,6 +91,11 @@ def file(ctx, recursive, timeout, nowait, path, scan_config, is_zip, zip_passwor def url_(ctx, qrcode_file, url_file, timeout, nowait, url, scan_config, expiration_window): """ Scan files or directories via PolySwarm + + \b + Defanged URLs (hxxps[:]//evil[.]com) are refanged before they are + validated and submitted, so the live form is what gets scanned; pass + --no-refang to the root command to send them verbatim. """ api = ctx.obj['api'] output = ctx.obj['output'] @@ -100,12 +105,16 @@ def url_(ctx, qrcode_file, url_file, timeout, nowait, url, scan_config, expirati urls = [qrcode_file] preprocessing = {'type': 'qrcode'} else: - urls = list(url) + # Refang before validating, so a defanged URL is judged in the form + # that will actually be submitted instead of being rejected. + # The error quotes what was typed, not the rewrite. + positional = [(u, refang_input(api, u)) for u in url] + urls = [live for _, live in positional] if url_file: urls.extend([u.strip() for u in url_file.readlines()]) - for _url in url: - if not is_url(_url): - raise click.BadArgumentUsage(f'URL "{_url}" is not valid. ' + for typed, live in positional: + if not is_url(live): + raise click.BadArgumentUsage(f'URL "{typed}" is not valid. ' 'Make sure the protocol "https://" or "http://" is set.') preprocessing = None for instance in api.scan_url(urls, timeout, nowait, scan_config, preprocessing, expiration_window): diff --git a/src/polyswarm/client/search.py b/src/polyswarm/client/search.py index 5e7a6cb..5f9cd65 100644 --- a/src/polyswarm/client/search.py +++ b/src/polyswarm/client/search.py @@ -108,6 +108,11 @@ def iocs_by_hash(ctx, type, value, hide_known_good): def search_known(ctx, ip, domain): """ Check if an ip address or domain is known. + + \b + A defanged host (evil[.]com) is looked up in its live form (evil.com), so + a row stored defanged by an earlier client is not matched; pass + --no-refang to the root command to look it up verbatim. """ api = ctx.obj['api'] output = ctx.obj['output'] @@ -124,6 +129,10 @@ def search_known(ctx, ip, domain): def add(ctx, type, host, source): """ Add a known good ip or domain. + + \b + A defanged host (evil[.]com) is stored in its live form (evil.com); pass + --no-refang to the root command to store it verbatim. """ api = ctx.obj['api'] output = ctx.obj['output'] @@ -141,6 +150,10 @@ def add(ctx, type, host, source): def update(ctx, id, type, host, source, good): """ Update a known ip address or domain. + + \b + A defanged host (evil[.]com) is stored in its live form (evil.com); pass + --no-refang to the root command to store it verbatim. """ api = ctx.obj['api'] output = ctx.obj['output'] diff --git a/src/polyswarm/polyswarm.py b/src/polyswarm/polyswarm.py index d29ad93..d440d53 100644 --- a/src/polyswarm/polyswarm.py +++ b/src/polyswarm/polyswarm.py @@ -285,6 +285,8 @@ def submit_url(self, url): :return: An ArtifactInstance resource. """ from polyswarm_api import resources + # A CLI-owned request: no SDK endpoint method refangs it for us. + url = utils.refang_input(self, url) logger.info('Submitting URL for IP analysis: %s', url) return self._single( { diff --git a/src/polyswarm/utils.py b/src/polyswarm/utils.py index f40e030..44d37f1 100644 --- a/src/polyswarm/utils.py +++ b/src/polyswarm/utils.py @@ -7,6 +7,7 @@ from itertools import zip_longest from polyswarm_api import exceptions as api_exceptions +from polyswarm_api import refang from polyswarm import exceptions @@ -133,3 +134,16 @@ def is_url(value): return value.startswith("https://") or value.startswith("http://") \ or is_domain(value) \ or is_ip(value) + + +def refang_input(api, value): + """Refang a URL / domain / IP argument the way the SDK does, if enabled. + + The SDK refangs the inputs of its own endpoint methods, so most commands + need nothing. This is for the two places the CLI handles the value itself: + validation that runs before the SDK sees it (``scan url``, ``sandbox url`` + would otherwise reject ``hxxps[:]//evil[.]com`` as invalid), and requests + the CLI builds without an SDK endpoint method (``Polyswarm.submit_url``). + Honours ``--no-refang`` through the client's ``refang_iocs``. + """ + return refang.refang_ioc(value) if api.refang_iocs else value diff --git a/tests/refang_test.py b/tests/refang_test.py new file mode 100644 index 0000000..076f3de --- /dev/null +++ b/tests/refang_test.py @@ -0,0 +1,228 @@ +"""IoC refanging through the CLI. + +Threat-intel reports print indicators defanged (``hxxps[:]//evil[.]com``, +``127[.]0[.]0[.]1``). The SDK refangs URL / domain / IP inputs before building +a request (``polyswarm_api.refang``); these tests pin that every CLI entry point +that takes such an input reaches the wire refanged, that ``--no-refang`` sends +it verbatim, and that live input is unchanged. + +The SDK is mocked at its transport: ``PolyswarmSession.execute``, the +documented session customization point, receives the fully built +``PolyswarmRequest`` descriptor, and the tests read its ``params`` / +``input_json`` directly. Mocking any higher — at ``search_url`` and friends — +would prove nothing, because the refang runs INSIDE those endpoint methods. +""" +from unittest import TestCase, mock + +from click.testing import CliRunner + +from polyswarm.client import polyswarm as client + + +_API_KEY = '1' * 32 +_API_URL = 'http://artifact-index-e2e:9696/v3' +_COMMUNITY = 'gamma' + + +class _Stop(Exception): + """Stops the command at its first request, which is the one carrying the IoC.""" + + +def _params(request): + params = request.params + if isinstance(params, dict): + pairs = [] + for key, value in params.items(): + if isinstance(value, (list, tuple)): + pairs.extend((key, v) for v in value) + else: + pairs.append((key, value)) + return pairs + return list(params) + + +class RefangCliTest(TestCase): + def setUp(self): + self.cli = CliRunner() + self.requests = [] + + def fake_execute(session, request): + self.requests.append(request) + raise _Stop() + + patcher = mock.patch('polyswarm_api.session.PolyswarmSession.execute', + autospec=True, side_effect=fake_execute) + patcher.start() + self.addCleanup(patcher.stop) + + def _run(self, *cmd, refang=True): + flags = [] if refang else ['--no-refang'] + return self.cli.invoke( + client.polyswarm_cli, + ['-a', _API_KEY, '-u', _API_URL, '-c', _COMMUNITY] + flags + list(cmd), + ) + + def _first_params(self): + self.assertTrue(self.requests, 'no request reached the SDK') + return _params(self.requests[0]) + + def _first_body(self): + self.assertTrue(self.requests, 'no request reached the SDK') + return self.requests[0].input_json + + # ── search ──────────────────────────────────────────────────────────── + + def test_search_url_refangs(self): + self._run('search', 'url', 'hxxps[:]//evil[.]com/x') + self.assertIn(('url', 'https://evil.com/x'), self._first_params()) + + def test_search_url_no_refang_sends_raw(self): + self._run('search', 'url', 'hxxps[:]//evil[.]com/x', refang=False) + self.assertIn(('url', 'hxxps[:]//evil[.]com/x'), self._first_params()) + + def test_search_url_live_value_is_unchanged(self): + self._run('search', 'url', 'https://example.com/a[.]b') + self.assertIn(('url', 'https://example.com/a[.]b'), self._first_params()) + + def test_search_metadata_refangs_ioc_options_but_not_the_query(self): + self._run('search', 'metadata', '-p', '127[.]0[.]0[.]1', '-u', 'hxxp://evil[.]com', + '-d', 'evil[dot]com', 'strings.domains:"bad[.]org"') + params = self._first_params() + self.assertIn(('ips', '127.0.0.1'), params) + self.assertIn(('urls', 'http://evil.com'), params) + self.assertIn(('domains', 'evil.com'), params) + self.assertIn(('query', 'strings.domains:"bad[.]org"'), params) + + def test_search_metadata_no_refang_sends_raw(self): + self._run('search', 'metadata', '-p', '127[.]0[.]0[.]1', 'x:*', refang=False) + self.assertIn(('ips', '127[.]0[.]0[.]1'), self._first_params()) + + def test_search_ioc_ip_refangs(self): + self._run('search', 'ioc', 'ip', '10[.]0[.]0[.]1') + self.assertIn(('ip', '10.0.0.1'), self._first_params()) + + def test_search_ioc_domain_refangs(self): + self._run('search', 'ioc', 'domain', 'evil(.)com') + self.assertIn(('domain', 'evil.com'), self._first_params()) + + def test_search_ioc_no_refang_sends_raw(self): + self._run('search', 'ioc', 'domain', 'evil(.)com', refang=False) + self.assertIn(('domain', 'evil(.)com'), self._first_params()) + + def test_search_known_refangs(self): + self._run('search', 'known', '-p', '8[.]8[.]8[.]8', '-d', 'good[.]example') + params = self._first_params() + self.assertIn(('ip', '8.8.8.8'), params) + self.assertIn(('domain', 'good.example'), params) + + # ── scan / sandbox / analyze-ip (submissions) ───────────────────────── + + def test_scan_url_accepts_and_refangs_a_defanged_url(self): + # The command validates URLs before submitting; a defanged URL must + # be validated in its refanged form, not rejected. + result = self._run('scan', 'url', '--nowait', 'hxxps[:]//evil[.]com/x') + self.assertNotIn('is not valid', result.output) + self.assertEqual(self._first_body()['artifact_name'], 'https://evil.com/x') + + def test_scan_url_file_lines_are_refanged(self): + with self.cli.isolated_filesystem(): + with open('urls.txt', 'w') as f: + f.write('hxxp://evil[.]com/a\n') + self._run('scan', 'url', '--nowait', '-r', 'urls.txt') + self.assertEqual(self._first_body()['artifact_name'], 'http://evil.com/a') + + def test_scan_url_no_refang_keeps_rejecting_a_defanged_url(self): + result = self._run('scan', 'url', '--nowait', 'hxxps[:]//evil[.]com/x', refang=False) + self.assertIn('is not valid', result.output) + self.assertEqual(self.requests, []) + + def test_sandbox_url_accepts_and_refangs_a_defanged_url(self): + result = self._run('sandbox', 'url', 'provider', 'hxxps[:]//evil[.]com/x', '--vm_slug', 'vm') + self.assertNotIn('is not valid', result.output) + self.assertEqual(self._first_body()['artifact_name'], 'https://evil.com/x') + + def test_sandbox_url_no_refang_keeps_rejecting_a_defanged_url(self): + result = self._run('sandbox', 'url', 'provider', 'hxxps[:]//evil[.]com/x', '--vm_slug', 'vm', + refang=False) + self.assertIn('is not valid', result.output) + self.assertEqual(self.requests, []) + + # ── QR-code submissions: the argument is an image path, never refanged ── + # + # The exemption lives in the SDK: a ``preprocessing={'type': 'qrcode'}`` + # submission skips refanging entirely. ``qr[.]png`` would otherwise pass + # the gate (``png`` has the shape of a TLD) and be rewritten to + # ``qr.png``, a path that does not exist. + + def test_scan_url_qrcode_file_path_is_not_refanged(self): + with self.cli.isolated_filesystem(): + with open('qr[.]png', 'wb') as f: + f.write(b'not really a png') + result = self._run('scan', 'url', '--nowait', '--qrcode-file', 'qr[.]png') + self.assertNotIsInstance(result.exception, TypeError) + self.assertEqual(self._first_body()['artifact_name'], 'qr[.]png') + + def test_sandbox_url_qrcode_file_submits_with_no_url(self): + # The qrcode branch hands the SDK ``url=None``; refanging must let it + # through untouched rather than fail on a non-string. + with self.cli.isolated_filesystem(): + with open('qr[.]png', 'wb') as f: + f.write(b'not really a png') + result = self._run('sandbox', 'url', 'provider', '--qrcode-file', 'qr[.]png', '--vm_slug', 'vm') + self.assertNotIsInstance(result.exception, TypeError) + self.assertEqual(self._first_body()['artifact_name'], 'qr[.]png') + + # ── known-host catalogue writes ─────────────────────────────────────── + + def test_known_add_refangs_the_host(self): + self._run('known', 'add', 'domain', 'good[.]example', 'feed') + self.assertEqual(self._first_body()['host'], 'good.example') + + def test_known_add_no_refang_sends_raw(self): + self._run('known', 'add', 'domain', 'good[.]example', 'feed', refang=False) + self.assertEqual(self._first_body()['host'], 'good[.]example') + + def test_known_update_refangs_the_host(self): + self._run('known', 'update', '7', 'domain', 'good[.]example', 'feed', '-g', 'true') + self.assertEqual(self._first_body()['host'], 'good.example') + + def test_known_update_no_refang_sends_raw(self): + self._run('known', 'update', '7', 'domain', 'good[.]example', 'feed', '-g', 'true', refang=False) + self.assertEqual(self._first_body()['host'], 'good[.]example') + + def test_metadata_analyze_ip_refangs(self): + # CLI-owned request (it bypasses the SDK endpoint methods), so the CLI + # applies the SDK's refang itself. + self._run('metadata', 'analyze-ip', '192(.)168(.)100(.)200') + self.assertEqual(self._first_body(), {'url': '192.168.100.200'}) + + def test_metadata_analyze_ip_no_refang_sends_raw(self): + self._run('metadata', 'analyze-ip', '192(.)168(.)100(.)200', refang=False) + self.assertEqual(self._first_body(), {'url': '192(.)168(.)100(.)200'}) + + # ── the flag itself ─────────────────────────────────────────────────── + + def test_invalid_defanged_url_error_quotes_what_was_typed(self): + # Validation runs on the refanged form, but the message must quote the + # input the user pasted, so the error is greppable against it. + # Refangs to ftp://files.example.org: a network IoC for the gate, but + # not a URL is_url accepts (http(s), a domain or an IP only). + typed = 'fxp://files[.]example[.]org' + for cmd in (['scan', 'url', '--nowait', typed], + ['sandbox', 'url', 'provider', typed, '--vm_slug', 'vm']): + result = self._run(*cmd) + self.assertIn(f'URL "{typed}" is not valid', result.output) + + def test_behaviour_change_is_in_each_affected_commands_help(self): + # A behaviour change is noted in the command's own --help, not only in + # the specs (specs/02-commands.md): scan url / sandbox url / analyze-ip + # submit, known add / update store, and search known looks up, the live + # form of a defanged input. + for cmd in (['scan', 'url'], ['sandbox', 'url'], ['known', 'add'], ['known', 'update'], + ['search', 'known'], ['metadata', 'analyze-ip']): + result = self._run(*cmd, '--help') + self.assertIn('--no-refang', result.output, cmd) + + def test_no_refang_flag_is_documented_in_help(self): + result = self.cli.invoke(client.polyswarm_cli, ['--help']) + self.assertIn('--no-refang', result.output)