diff --git a/CHANGELOG.md b/CHANGELOG.md index 33996e4..ce65722 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,59 @@ All notable changes to this project are documented here. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and the project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [3.2.0] - unreleased + +### Added + +- **On-demand scope API**, on both `Onyphe` and `AsyncOnyphe`: active scanning + of an IP, a CIDR or a domain, one target at a time or in bulk, plus result + retrieval. **Needs an On-demand subscription**; without it the call comes + back as a `PaymentRequiredError`. + + ```python + launch = api.ondemand_scope_ip("8.8.8.0/24", vulnscan=True, ports=[80, 443]) + page = api.ondemand_scope_result(scan_id) + ``` + + - `ondemand_scope_ip(ip, ...)` — `POST /ondemand/scope/ip/single` + - `ondemand_scope_domain(domain, ...)` — `POST /ondemand/scope/domain/single` + - `ondemand_scope_ip_bulk(ips, ...)` — `POST /dev/ondemand/scope/ip/bulk` + - `ondemand_scope_domain_bulk(domains, ...)` — `POST /dev/ondemand/scope/domain/bulk` + - `ondemand_scope_result(scan_id)` — `GET /ondemand/scope/result/{scan_id}` + + The two bulk endpoints live under ONYPHE's `/dev/` prefix and may move or + change shape without notice. `/ondemand/scope/domain/single` is not + documented upstream either: it is deduced by analogy with `ip/single` and + isolated in a single constant. + + The four launch methods share the optional `import_results`, `vulnscan`, + `urlscan`, `ports` and `maxscantime` arguments, and send a key only when one + is given. `import_results` is spelled out because `import` is a Python + keyword — **it publishes the results in the ONYPHE dataset, where every + ONYPHE user can see them.** + + A launch returns the raw envelope: ONYPHE documents that a Scan ID comes + back, but not the field it comes back in, so nothing is parsed out of it. + No polling helper either, for now. + + The four launches are never retried automatically — a POST replayed after a + 429, a 5xx or a transport failure would start a second scan, burning credits + or earning an error 9 (`Scan is already running`). `send()` grew a + keyword-only `retry: bool = True` for that; result retrieval, a GET, keeps + the usual retry behaviour. + +- `ScanInProgressError`, raised by `ondemand_scope_result` on ONYPHE error + codes 103 (`Scan ID is in progress`) and 111 (`Scan ID results are being + built`), whatever HTTP status carries them. It subclasses `APIError` and + carries `.scan_id`, so "retry later" is distinguishable from "failed". Every + other error code keeps the existing mapping. + +- CLI: `pyonyphe ondemand ip|domain|ip-bulk|domain-bulk|result`, with + `--import/--no-import`, `--vulnscan/--no-vulnscan`, + `--urlscan/--no-urlscan`, `--ports` and `--maxscantime`. The two bulk + commands read one target per line from a file. `ondemand result` exits with + 3 while the scan is still running. + ## [3.1.0] - 2026-08-04 ### Changed diff --git a/README.md b/README.md index 293f74d..287c358 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,10 @@ pyonyphe simple whois 8.8.8.8 --best pyonyphe resolve example.com pyonyphe bulk simple datascan ips.txt -o out.ndjson pyonyphe alert list + +# active scanning, On-demand subscription required +pyonyphe ondemand ip 8.8.8.0/24 --vulnscan --ports 80,443 +pyonyphe ondemand result SCAN_ID ``` ## Configuration diff --git a/docs/api.md b/docs/api.md index 5b796a4..1368088 100644 --- a/docs/api.md +++ b/docs/api.md @@ -53,6 +53,60 @@ Bulk Simple categories are the Simple ones minus `onionscan` and `onionshot`. `source` accepts a `Path`, a path string, a raw string, an iterable of assets, or bytes. +## On-demand scan + +Active scanning, **On-demand subscription required** — every other API on this +page only reads data ONYPHE already collected. + +| method | HTTP | endpoint | +| --- | --- | --- | +| `ondemand_scope_ip(ip, ...)` | POST | `/ondemand/scope/ip/single` | +| `ondemand_scope_domain(domain, ...)` | POST | `/ondemand/scope/domain/single` | +| `ondemand_scope_ip_bulk(ips, ...)` | POST | `/dev/ondemand/scope/ip/bulk` | +| `ondemand_scope_domain_bulk(domains, ...)` | POST | `/dev/ondemand/scope/domain/bulk` | +| `ondemand_scope_result(scan_id)` | GET | `/ondemand/scope/result/{scan_id}` | + +`ip` is `X.Y.Z.K` or `X.Y.Z.K/24`. The two bulk methods take any iterable of +strings and join it with commas themselves. + +The two bulk endpoints live under the `/dev/` prefix (`DEV_PREFIX`), which +ONYPHE serves from its development tree: they may move or change shape without +notice. `/ondemand/scope/domain/single` is **not** documented by ONYPHE +either — it is deduced by analogy with `ip/single`, and kept in a single +constant (`ONDEMAND_SCOPE_DOMAIN_PATH`) so that one edit fixes it if it turns +out to be wrong. + +The four launch methods share these optional arguments, and send a key only +when you pass one: + +| argument | sent as | value | +| --- | --- | --- | +| `import_results` | `import` | `"true"` / `"false"` | +| `vulnscan` | `vulnscan` | `"true"` / `"false"` | +| `urlscan` | `urlscan` | `"true"` / `"false"` | +| `ports` | `ports` | `[80, 443]` becomes `"80,443"` | +| `maxscantime` | `maxscantime` | an integer, in seconds | + +`import_results` is named that way because `import` is a Python keyword. +**Setting it publishes the scan results in the ONYPHE dataset, where every +ONYPHE user can see them.** + +A launch returns the raw `Response`: ONYPHE documents that it carries a Scan +ID but not under which field name, so nothing is parsed out of it. Read it +from `response.model_dump()` and feed it back to `ondemand_scope_result`. + +The four launch methods are sent exactly once and never retried automatically, +on a 429, a 5xx or a transport failure alike: replaying the POST would start a +second scan. `ondemand_scope_result` is a GET and keeps the usual retry +behaviour. + +`ondemand_scope_result` raises `ScanInProgressError` when ONYPHE answers with +error code 103 (`Scan ID is in progress`) or 111 (`Scan ID results are being +built`), whatever HTTP status carries it. That is not a failure: the same +`scan_id` is worth asking for again later. The exception carries `.scan_id`. +Every other error code goes through the usual mapping. There is no polling +helper yet. + ## Alerts | method | HTTP | endpoint | @@ -87,6 +141,7 @@ OnypheError ├── PaymentRequiredError 402 ├── NotFoundError 404 ├── RateLimitError 429 (.retry_after) + ├── ScanInProgressError ONYPHE error 103 / 111 (.scan_id) └── ServerError 5xx ``` @@ -94,4 +149,4 @@ OnypheError `SIMPLE_CATEGORIES`, `BEST_CATEGORIES`, `BULK_SIMPLE_CATEGORIES`, `SUMMARY_KINDS`, `SEARCH_MAX_RESULTS` (10000), `DEFAULT_BASE_URL`, -`UNRATED_BASE_URL`. +`UNRATED_BASE_URL`, `SCAN_IN_PROGRESS_CODES` (103, 111), `DEV_PREFIX`. diff --git a/docs/cli.md b/docs/cli.md index 03235f1..15bb723 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -111,6 +111,52 @@ pyonyphe bulk simple whois ips.txt --best One OQL query per line, all run against the given category. Griffin View only. +### `ondemand ip IP | ondemand domain DOMAIN` + +Launch an active scan. **Needs an On-demand subscription.** Prints the raw +ONYPHE envelope, which carries the Scan ID to pass to `ondemand result`. + +```bash +pyonyphe ondemand ip 8.8.8.8 +pyonyphe ondemand ip 8.8.8.0/24 --vulnscan --ports 80,443 --maxscantime 120 +pyonyphe ondemand domain example.com --urlscan +``` + +| option | meaning | +| --- | --- | +| `--import` / `--no-import` | import the results into the ONYPHE dataset | +| `--vulnscan` / `--no-vulnscan` | also run the vulnerability scan | +| `--urlscan` / `--no-urlscan` | also crawl the HTTP services found | +| `--ports 80,443` | ports to scan instead of the ONYPHE default | +| `--maxscantime 120` | scan budget, in seconds | + +Each option is sent only when given: passing none of them leaves every +default to ONYPHE. `--import` **makes the results PUBLIC** — they are added to +the ONYPHE dataset, where every ONYPHE user can see them. + +### `ondemand ip-bulk FILE | ondemand domain-bulk FILE` + +Same options, one target per line in `FILE`, all scanned under a single Scan +ID. Both endpoints sit under the ONYPHE `/dev/` prefix and may move without +notice. + +```bash +pyonyphe ondemand ip-bulk ips.txt --no-import +pyonyphe ondemand domain-bulk domains.txt +``` + +### `ondemand result SCAN_ID` + +Fetch the results of a scan launched earlier. + +```bash +pyonyphe ondemand result 0123456789abcdef +``` + +Exits with **3** when ONYPHE answers that the scan is still running or that +its results are being built (error codes 103 and 111) — retry later, nothing +failed. Other API errors still exit with 1. + ### `alert list | add | del` ```bash diff --git a/docs/usage.md b/docs/usage.md index f44f363..2d1a2a6 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -112,6 +112,69 @@ api.bulk_simple("datascan", ["1.1.1.1", "8.8.8.8"]) api.bulk_summary("domain", domains_from_your_database) ``` +## On-demand scan + +Everything else on this page reads what ONYPHE already collected. The +On-demand scope API makes ONYPHE *scan* an asset for you, which needs an +**On-demand subscription** — without it the call comes back as a +`PaymentRequiredError`. + +```python +from pyonyphe import Onyphe + +with Onyphe() as api: + launch = api.ondemand_scope_ip("8.8.8.0/24", vulnscan=True, ports=[80, 443]) + print(launch.model_dump()) # the Scan ID is in there +``` + +Then, with that Scan ID in hand: + +```python +from pyonyphe import ScanInProgressError + +try: + page = api.ondemand_scope_result(scan_id) +except ScanInProgressError as exc: + print(f"{exc.scan_id} is not done yet, try again later") +else: + for document in page: + print(document) +``` + +ONYPHE states that a launch returns a Scan ID, but does not document the field +it travels in, so the client returns the envelope as it came and parses +nothing out of it. There is no polling helper yet either: retry +`ondemand_scope_result` yourself. + +Four launch methods, two of them taking a list of targets and joining it for +you: + +```python +api.ondemand_scope_ip("8.8.8.8") # or "8.8.8.0/24" +api.ondemand_scope_domain("example.com") +api.ondemand_scope_ip_bulk(["1.1.1.1", "8.8.8.8", "10.0.0.0/24"]) +api.ondemand_scope_domain_bulk(["a.tld", "b.tld"]) +``` + +The two `_bulk` ones are served under the `/dev/` prefix, ONYPHE's development +tree: they can move without notice. + +All four take the same optional arguments, and send nothing at all for the ones +you leave out: + +| argument | sent as | meaning | +| --- | --- | --- | +| `import_results` | `import` | import the results into the ONYPHE dataset | +| `vulnscan` | `vulnscan` | also run the vulnerability scan | +| `urlscan` | `urlscan` | also crawl the HTTP services found | +| `ports` | `ports` | `[80, 443]` is sent as `"80,443"` | +| `maxscantime` | `maxscantime` | scan budget, in seconds | + +`import_results` spells out what `import` cannot: the name is a Python +keyword. **Importing makes the results PUBLIC** — they land in the ONYPHE +dataset and every ONYPHE user can see them. Leave it alone unless that is what +you want. + ## Alerts ```python @@ -141,11 +204,16 @@ Every exception derives from `OnypheError`: | `PaymentRequiredError` | 402 — credits exhausted, or API not in your license | | `NotFoundError` | 404 | | `RateLimitError` | 429, with `.retry_after` when ONYPHE says so | +| `ScanInProgressError` | an On-demand scan has no results yet, with `.scan_id` | | `ServerError` | 5xx | `AuthenticationError`, `PaymentRequiredError`, `NotFoundError`, -`RateLimitError` and `ServerError` all subclass `APIError`, which carries -`.status_code` and the decoded `.payload`. +`RateLimitError`, `ScanInProgressError` and `ServerError` all subclass +`APIError`, which carries `.status_code` and the decoded `.payload`. + +`ScanInProgressError` is the odd one: it reports ONYPHE error code 103 or 111 +on an On-demand scan, which means "not ready", not "failed". Catch it to retry +rather than to give up. 429 and 5xx are retried automatically (`max_retries`, exponential backoff, honouring `Retry-After`); the exception only surfaces once the retries are @@ -153,9 +221,9 @@ exhausted. ## Endpoints not wrapped yet -The Ondemand APIv3 (`scope`, `resolver`) and the beta ASD APIv1 are not -wrapped. Reach them with the escape hatch, which handles auth, retries and -error mapping like any other call: +The Ondemand `resolver` endpoints and the beta ASD APIv1 are not wrapped +(`ondemand/scope` is, see above). Reach them with the escape hatch, which +handles auth, retries and error mapping like any other call: ```python api.request("GET", "some/new/endpoint", params={"q": "..."}) diff --git a/src/pyonyphe/__init__.py b/src/pyonyphe/__init__.py index 31411c6..70ab87b 100644 --- a/src/pyonyphe/__init__.py +++ b/src/pyonyphe/__init__.py @@ -28,6 +28,7 @@ ParamError, PaymentRequiredError, RateLimitError, + ScanInProgressError, ServerError, TransportError, ) @@ -58,6 +59,7 @@ "PaymentRequiredError", "RateLimitError", "Response", + "ScanInProgressError", "ServerError", "Settings", "TransportError", diff --git a/src/pyonyphe/_base.py b/src/pyonyphe/_base.py index 09f1eb2..84b6ccb 100644 --- a/src/pyonyphe/_base.py +++ b/src/pyonyphe/_base.py @@ -5,6 +5,7 @@ import base64 import json as jsonlib +from collections.abc import Mapping from dataclasses import dataclass from typing import Any @@ -13,11 +14,13 @@ from ._specs import Spec from .config import Settings, load_settings from .errors import ( + SCAN_IN_PROGRESS_CODES, APIError, AuthenticationError, NotFoundError, PaymentRequiredError, RateLimitError, + ScanInProgressError, ServerError, ) from .models import Response @@ -154,6 +157,27 @@ def raise_for_status(self, response: httpx.Response, payload: dict[str, Any]) -> raise ServerError(message or "onyphe server error", status_code=status, payload=payload) raise APIError(message or "unknown error", status_code=status, payload=payload) + def raise_for_scan( + self, + scan_id: str, + payload: Mapping[str, Any], + *, + status_code: int | None = None, + ) -> None: + """Tell a pending On-demand scan apart from a genuine failure. + + :param payload: decoded body, successful or not + :raises ScanInProgressError: when ONYPHE reported code 103 or 111 + """ + code = payload.get("error") + if isinstance(code, int) and code in SCAN_IN_PROGRESS_CODES: + raise ScanInProgressError( + str(payload.get("text") or "scan in progress"), + scan_id=scan_id, + status_code=status_code, + payload=dict(payload), + ) + def to_response(self, response: httpx.Response) -> Response: """Validate a successful JSON body into a :class:`Response`.""" payload = self._decode(response) diff --git a/src/pyonyphe/_specs.py b/src/pyonyphe/_specs.py index 9215295..7639cbd 100644 --- a/src/pyonyphe/_specs.py +++ b/src/pyonyphe/_specs.py @@ -17,6 +17,12 @@ __all__ = [ "BEST_CATEGORIES", "BULK_SIMPLE_CATEGORIES", + "DEV_PREFIX", + "ONDEMAND_SCOPE_DOMAIN_BULK_PATH", + "ONDEMAND_SCOPE_DOMAIN_PATH", + "ONDEMAND_SCOPE_IP_BULK_PATH", + "ONDEMAND_SCOPE_IP_PATH", + "ONDEMAND_SCOPE_RESULT_PATH", "SEARCH_MAX_RESULTS", "SIMPLE_CATEGORIES", "SUMMARY_KINDS", @@ -31,6 +37,18 @@ #: Hard limit enforced by the Search API; beyond that you need Export. SEARCH_MAX_RESULTS = 10_000 +#: Prefix of the endpoints ONYPHE still serves from its development tree. The +#: On-demand bulk endpoints live there and may move without notice. +DEV_PREFIX = "dev" + +ONDEMAND_SCOPE_IP_PATH = "ondemand/scope/ip/single" +#: Deduced by analogy with :data:`ONDEMAND_SCOPE_IP_PATH`; the ONYPHE +#: documentation does not spell this one out. +ONDEMAND_SCOPE_DOMAIN_PATH = "ondemand/scope/domain/single" +ONDEMAND_SCOPE_IP_BULK_PATH = f"{DEV_PREFIX}/ondemand/scope/ip/bulk" +ONDEMAND_SCOPE_DOMAIN_BULK_PATH = f"{DEV_PREFIX}/ondemand/scope/domain/bulk" +ONDEMAND_SCOPE_RESULT_PATH = "ondemand/scope/result" + SimpleCategory = Literal[ "ctl", "datascan", @@ -253,6 +271,133 @@ def discovery(category: str, source: str | Path | Iterable[str] | bytes) -> Spec return Spec("POST", f"bulk/discovery/{category}/asset", content=to_payload(source), stream=True) +# -------------------------------------------------------------------------- +# On-demand scope API -- POST a JSON body, get a scan identifier back +# -------------------------------------------------------------------------- + + +def _scan_options( + import_results: bool | None, + vulnscan: bool | None, + urlscan: bool | None, + ports: Iterable[int] | None, + maxscantime: int | None, +) -> dict[str, Any]: + """Serialise the optional scan tuning, omitting whatever was left out. + + ``import_results`` travels as ``import``, a Python keyword, and the three + booleans travel as the strings ONYPHE documents rather than as JSON + booleans. + """ + body: dict[str, Any] = {} + if import_results is not None: + body["import"] = _flag(import_results) + if vulnscan is not None: + body["vulnscan"] = _flag(vulnscan) + if urlscan is not None: + body["urlscan"] = _flag(urlscan) + if ports is not None: + body["ports"] = ",".join(str(port) for port in ports) + if maxscantime is not None: + body["maxscantime"] = maxscantime + return body + + +def _scope_spec(path: str, field: str, value: str, options: dict[str, Any]) -> Spec: + if not value: + raise ParamError(f"a {field} is required") + return Spec("POST", path, json={field: value, **options}) + + +def _join(values: Iterable[str], label: str) -> str: + """Join scan targets into the comma-separated string ONYPHE expects.""" + # A bare string is iterable: wrap it, or it would be split per character. + items = [str(value).strip() for value in ([values] if isinstance(values, str) else values)] + items = [item for item in items if item] + if not items: + raise ParamError(f"empty {label} list") + return ",".join(items) + + +def ondemand_scope_ip( + ip: str, + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, +) -> Spec: + """On-demand scan of a single IP address or CIDR.""" + return _scope_spec( + ONDEMAND_SCOPE_IP_PATH, + "ip", + ip, + _scan_options(import_results, vulnscan, urlscan, ports, maxscantime), + ) + + +def ondemand_scope_domain( + domain: str, + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, +) -> Spec: + """On-demand scan of a single domain.""" + return _scope_spec( + ONDEMAND_SCOPE_DOMAIN_PATH, + "domain", + domain, + _scan_options(import_results, vulnscan, urlscan, ports, maxscantime), + ) + + +def ondemand_scope_ip_bulk( + ips: Iterable[str], + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, +) -> Spec: + """On-demand scan of several IP addresses or CIDRs at once.""" + return _scope_spec( + ONDEMAND_SCOPE_IP_BULK_PATH, + "ip", + _join(ips, "ip"), + _scan_options(import_results, vulnscan, urlscan, ports, maxscantime), + ) + + +def ondemand_scope_domain_bulk( + domains: Iterable[str], + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, +) -> Spec: + """On-demand scan of several domains at once.""" + return _scope_spec( + ONDEMAND_SCOPE_DOMAIN_BULK_PATH, + "domain", + _join(domains, "domain"), + _scan_options(import_results, vulnscan, urlscan, ports, maxscantime), + ) + + +def ondemand_scope_result(scan_id: str) -> Spec: + """Results of a scan launched earlier, by scan identifier.""" + if not scan_id: + raise ParamError("a scan id is required") + return Spec("GET", f"{ONDEMAND_SCOPE_RESULT_PATH}/{scan_id}") + + # -------------------------------------------------------------------------- # Alert API # -------------------------------------------------------------------------- diff --git a/src/pyonyphe/async_client.py b/src/pyonyphe/async_client.py index 42e4f5a..dc3d040 100644 --- a/src/pyonyphe/async_client.py +++ b/src/pyonyphe/async_client.py @@ -20,7 +20,7 @@ Spec, SummaryKind, ) -from .errors import TransportError +from .errors import APIError, TransportError from .models import Alert, Response __all__ = ["AsyncOnyphe"] @@ -69,21 +69,26 @@ def _kwargs(prepared: PreparedRequest) -> dict[str, Any]: kwargs["json"] = prepared.json return kwargs - async def send(self, spec: Spec) -> Response: - """Send a non-streaming spec, retrying transient failures.""" + async def send(self, spec: Spec, *, retry: bool = True) -> Response: + """Send a non-streaming spec, retrying transient failures. + + :param retry: ``False`` sends the request exactly once, for calls that + are not idempotent + """ prepared = self.prepare(spec) kwargs = self._kwargs(prepared) + max_retries = self.max_retries if retry else 0 last_error: Exception | None = None - for attempt in range(self.max_retries + 1): + for attempt in range(max_retries + 1): try: response = await self._client.request(prepared.method, prepared.url, **kwargs) except httpx.HTTPError as exc: last_error = TransportError(f"unable to reach ONYPHE: {exc}") - if attempt >= self.max_retries: + if attempt >= max_retries: raise last_error from exc await asyncio.sleep(self.retry_delay(attempt)) continue - if response.status_code in RETRY_STATUS and attempt < self.max_retries: + if response.status_code in RETRY_STATUS and attempt < max_retries: header = response.headers.get("Retry-After", "") after = float(header) if header.replace(".", "", 1).isdigit() else None await asyncio.sleep(self.retry_delay(attempt, after)) @@ -243,6 +248,150 @@ def discovery(self, category: str, source: BulkSource) -> AsyncIterator[dict[str """Discovery API: several OQL queries at once (Griffin View only).""" return self.stream(specs.discovery(category, source)) + # -- on-demand scope ---------------------------------------------------- + + async def ondemand_scope_ip( + self, + ip: str, + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, + ) -> Response: + """Launch an active scan of one IP address or CIDR. + + Needs an On-demand subscription. Sent once, never retried: replaying + the POST would start a second scan. + + :param ip: ``X.Y.Z.K`` or ``X.Y.Z.K/24`` + :param import_results: import the results into the ONYPHE dataset, + which makes them **PUBLIC** -- visible to every ONYPHE user + :param vulnscan: also run the vulnerability scan + :param urlscan: also crawl the HTTP services found + :param ports: ports to scan instead of the ONYPHE default + :param maxscantime: scan budget, in seconds + :returns: the raw envelope, which carries the scan identifier to hand + to :meth:`ondemand_scope_result` + """ + return await self.send( + specs.ondemand_scope_ip( + ip, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=ports, + maxscantime=maxscantime, + ), + retry=False, + ) + + async def ondemand_scope_domain( + self, + domain: str, + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, + ) -> Response: + """Launch an active scan of one domain. + + Needs an On-demand subscription. Options are the ones of + :meth:`ondemand_scope_ip`, ``import_results`` included: it makes the + results **PUBLIC** on ONYPHE. + """ + return await self.send( + specs.ondemand_scope_domain( + domain, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=ports, + maxscantime=maxscantime, + ), + retry=False, + ) + + async def ondemand_scope_ip_bulk( + self, + ips: Iterable[str], + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, + ) -> Response: + """Launch one active scan covering several IP addresses or CIDRs. + + Needs an On-demand subscription. Served from the ``/dev/`` prefix, so + ONYPHE may still move it. Options are the ones of + :meth:`ondemand_scope_ip`, ``import_results`` included: it makes the + results **PUBLIC** on ONYPHE. + + :param ips: any iterable of assets; joined with commas for you + """ + return await self.send( + specs.ondemand_scope_ip_bulk( + ips, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=ports, + maxscantime=maxscantime, + ), + retry=False, + ) + + async def ondemand_scope_domain_bulk( + self, + domains: Iterable[str], + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, + ) -> Response: + """Launch one active scan covering several domains. + + Needs an On-demand subscription. Served from the ``/dev/`` prefix, so + ONYPHE may still move it. Options are the ones of + :meth:`ondemand_scope_ip`, ``import_results`` included: it makes the + results **PUBLIC** on ONYPHE. + + :param domains: any iterable of domains; joined with commas for you + """ + return await self.send( + specs.ondemand_scope_domain_bulk( + domains, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=ports, + maxscantime=maxscantime, + ), + retry=False, + ) + + async def ondemand_scope_result(self, scan_id: str) -> Response: + """Fetch the results of a scan launched earlier. + + :param scan_id: identifier returned when the scan was launched + :raises ScanInProgressError: while the scan runs or its results are + being built -- retry later, nothing failed + """ + try: + response = await self.send(specs.ondemand_scope_result(scan_id)) + except APIError as exc: + self.raise_for_scan(scan_id, exc.payload, status_code=exc.status_code) + raise + self.raise_for_scan(scan_id, response.model_dump()) + return response + # -- alerts ------------------------------------------------------------- async def alerts(self) -> list[Alert]: diff --git a/src/pyonyphe/cli.py b/src/pyonyphe/cli.py index 8ada02d..d190280 100644 --- a/src/pyonyphe/cli.py +++ b/src/pyonyphe/cli.py @@ -22,7 +22,7 @@ from ._specs import BestCategory, BulkSimpleCategory, SimpleCategory, SummaryKind from .client import Onyphe from .config import load_settings -from .errors import OnypheError +from .errors import OnypheError, ScanInProgressError app = typer.Typer( name="pyonyphe", @@ -32,8 +32,12 @@ ) alert_app = typer.Typer(help="Manage ONYPHE alerts.", no_args_is_help=True) bulk_app = typer.Typer(help="Bulk endpoints, fed from a file of assets.", no_args_is_help=True) +ondemand_app = typer.Typer( + help="On-demand active scans; needs an On-demand subscription.", no_args_is_help=True +) app.add_typer(alert_app, name="alert") app.add_typer(bulk_app, name="bulk") +app.add_typer(ondemand_app, name="ondemand") out = Console() err = Console(stderr=True) @@ -147,6 +151,26 @@ def run(rows: Iterator[dict[str, Any]], output: Path | None) -> None: err.print(f"[dim]{count} document(s)[/dim]") +def _ports(value: str | None) -> list[int] | None: + """Parse ``--ports 80,443`` into the list the client expects.""" + if value is None: + return None + try: + return [int(part) for part in value.split(",") if part.strip()] + except ValueError as exc: + raise typer.BadParameter(f"not a comma-separated port list: {value!r}") from exc + + +def _targets(file: Path) -> list[str]: + """Read one scan target per line, ignoring blank lines.""" + try: + text = file.read_text(encoding="utf-8") + except OSError as exc: + err.print(f"[red]{exc}[/red]") + raise typer.Exit(code=2) from exc + return [line.strip() for line in text.splitlines() if line.strip()] + + def _version_callback(value: bool) -> None: if value: out.print(f"pyonyphe {__version__}") @@ -384,6 +408,194 @@ def bulk_discovery( raise typer.Exit(code=1) from exc +@ondemand_app.command("ip") +def ondemand_ip( + ip: Annotated[str, typer.Argument(help="IP address or CIDR, e.g. 8.8.8.8 or 8.8.8.0/24.")], + import_results: Annotated[ + bool | None, + typer.Option( + "--import/--no-import", + help="Import the results into ONYPHE, which makes them PUBLIC.", + ), + ] = None, + vulnscan: Annotated[ + bool | None, typer.Option("--vulnscan/--no-vulnscan", help="Also run the vuln scan.") + ] = None, + urlscan: Annotated[ + bool | None, typer.Option("--urlscan/--no-urlscan", help="Also crawl the HTTP services.") + ] = None, + ports: Annotated[ + str | None, typer.Option("--ports", help="Ports to scan, comma separated: 80,443.") + ] = None, + maxscantime: Annotated[ + int | None, typer.Option("--maxscantime", help="Scan budget, in seconds.") + ] = None, + output: Annotated[Path | None, typer.Option("--output", "-o")] = None, +) -> None: + """Launch an active scan of one IP address or CIDR.""" + with get_client() as client: + try: + response = client.ondemand_scope_ip( + ip, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=_ports(ports), + maxscantime=maxscantime, + ) + except OnypheError as exc: + err.print(f"[red]{exc}[/red]") + raise typer.Exit(code=1) from exc + emit_json(response.model_dump(), output) + + +@ondemand_app.command("domain") +def ondemand_domain( + domain: Annotated[str, typer.Argument(help="Domain to scan, e.g. example.com.")], + import_results: Annotated[ + bool | None, + typer.Option( + "--import/--no-import", + help="Import the results into ONYPHE, which makes them PUBLIC.", + ), + ] = None, + vulnscan: Annotated[ + bool | None, typer.Option("--vulnscan/--no-vulnscan", help="Also run the vuln scan.") + ] = None, + urlscan: Annotated[ + bool | None, typer.Option("--urlscan/--no-urlscan", help="Also crawl the HTTP services.") + ] = None, + ports: Annotated[ + str | None, typer.Option("--ports", help="Ports to scan, comma separated: 80,443.") + ] = None, + maxscantime: Annotated[ + int | None, typer.Option("--maxscantime", help="Scan budget, in seconds.") + ] = None, + output: Annotated[Path | None, typer.Option("--output", "-o")] = None, +) -> None: + """Launch an active scan of one domain.""" + with get_client() as client: + try: + response = client.ondemand_scope_domain( + domain, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=_ports(ports), + maxscantime=maxscantime, + ) + except OnypheError as exc: + err.print(f"[red]{exc}[/red]") + raise typer.Exit(code=1) from exc + emit_json(response.model_dump(), output) + + +@ondemand_app.command("ip-bulk") +def ondemand_ip_bulk( + file: Annotated[Path, typer.Argument(help="One IP address or CIDR per line.")], + import_results: Annotated[ + bool | None, + typer.Option( + "--import/--no-import", + help="Import the results into ONYPHE, which makes them PUBLIC.", + ), + ] = None, + vulnscan: Annotated[ + bool | None, typer.Option("--vulnscan/--no-vulnscan", help="Also run the vuln scan.") + ] = None, + urlscan: Annotated[ + bool | None, typer.Option("--urlscan/--no-urlscan", help="Also crawl the HTTP services.") + ] = None, + ports: Annotated[ + str | None, typer.Option("--ports", help="Ports to scan, comma separated: 80,443.") + ] = None, + maxscantime: Annotated[ + int | None, typer.Option("--maxscantime", help="Scan budget, in seconds.") + ] = None, + output: Annotated[Path | None, typer.Option("--output", "-o")] = None, +) -> None: + """Launch one active scan covering a list of IP addresses or CIDRs.""" + targets = _targets(file) + with get_client() as client: + try: + response = client.ondemand_scope_ip_bulk( + targets, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=_ports(ports), + maxscantime=maxscantime, + ) + except OnypheError as exc: + err.print(f"[red]{exc}[/red]") + raise typer.Exit(code=1) from exc + emit_json(response.model_dump(), output) + + +@ondemand_app.command("domain-bulk") +def ondemand_domain_bulk( + file: Annotated[Path, typer.Argument(help="One domain per line.")], + import_results: Annotated[ + bool | None, + typer.Option( + "--import/--no-import", + help="Import the results into ONYPHE, which makes them PUBLIC.", + ), + ] = None, + vulnscan: Annotated[ + bool | None, typer.Option("--vulnscan/--no-vulnscan", help="Also run the vuln scan.") + ] = None, + urlscan: Annotated[ + bool | None, typer.Option("--urlscan/--no-urlscan", help="Also crawl the HTTP services.") + ] = None, + ports: Annotated[ + str | None, typer.Option("--ports", help="Ports to scan, comma separated: 80,443.") + ] = None, + maxscantime: Annotated[ + int | None, typer.Option("--maxscantime", help="Scan budget, in seconds.") + ] = None, + output: Annotated[Path | None, typer.Option("--output", "-o")] = None, +) -> None: + """Launch one active scan covering a list of domains.""" + targets = _targets(file) + with get_client() as client: + try: + response = client.ondemand_scope_domain_bulk( + targets, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=_ports(ports), + maxscantime=maxscantime, + ) + except OnypheError as exc: + err.print(f"[red]{exc}[/red]") + raise typer.Exit(code=1) from exc + emit_json(response.model_dump(), output) + + +@ondemand_app.command("result") +def ondemand_result( + scan_id: Annotated[str, typer.Argument(help="Scan ID returned when the scan was launched.")], + fmt: Annotated[str, typer.Option("--format", "-f", help="table, json or ndjson.")] = "json", + output: Annotated[Path | None, typer.Option("--output", "-o")] = None, +) -> None: + """Fetch the results of a scan launched earlier. + + Exits with 3 when the scan is still running: retry later, nothing failed. + """ + with get_client() as client: + try: + response = client.ondemand_scope_result(scan_id) + except ScanInProgressError as exc: + err.print(f"[yellow]{exc}[/yellow]") + raise typer.Exit(code=3) from exc + except OnypheError as exc: + err.print(f"[red]{exc}[/red]") + raise typer.Exit(code=1) from exc + render(response.results, fmt, output, f"scan {scan_id}") + + @alert_app.command("list") def alert_list() -> None: """List the alerts configured on the account.""" diff --git a/src/pyonyphe/client.py b/src/pyonyphe/client.py index 036ecdd..dbe908b 100644 --- a/src/pyonyphe/client.py +++ b/src/pyonyphe/client.py @@ -20,7 +20,7 @@ Spec, SummaryKind, ) -from .errors import TransportError +from .errors import APIError, TransportError from .models import Alert, Response __all__ = ["Onyphe"] @@ -72,25 +72,29 @@ def _kwargs(prepared: PreparedRequest) -> dict[str, Any]: kwargs["json"] = prepared.json return kwargs - def send(self, spec: Spec) -> Response: + def send(self, spec: Spec, *, retry: bool = True) -> Response: """Send a non-streaming spec, retrying transient failures. + :param retry: ``False`` sends the request exactly once, for calls that + are not idempotent -- no replay on 429/5xx, none on a transport + failure either :raises TransportError: when the request never reached ONYPHE :raises APIError: on any non-2xx answer """ prepared = self.prepare(spec) kwargs = self._kwargs(prepared) + max_retries = self.max_retries if retry else 0 last_error: Exception | None = None - for attempt in range(self.max_retries + 1): + for attempt in range(max_retries + 1): try: response = self._client.request(prepared.method, prepared.url, **kwargs) except httpx.HTTPError as exc: last_error = TransportError(f"unable to reach ONYPHE: {exc}") - if attempt >= self.max_retries: + if attempt >= max_retries: raise last_error from exc time.sleep(self.retry_delay(attempt)) continue - if response.status_code in RETRY_STATUS and attempt < self.max_retries: + if response.status_code in RETRY_STATUS and attempt < max_retries: header = response.headers.get("Retry-After", "") after = float(header) if header.replace(".", "", 1).isdigit() else None time.sleep(self.retry_delay(attempt, after)) @@ -271,6 +275,150 @@ def discovery(self, category: str, source: BulkSource) -> Iterator[dict[str, Any """Discovery API: several OQL queries at once (Griffin View only).""" return self.stream(specs.discovery(category, source)) + # -- on-demand scope ---------------------------------------------------- + + def ondemand_scope_ip( + self, + ip: str, + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, + ) -> Response: + """Launch an active scan of one IP address or CIDR. + + Needs an On-demand subscription. Sent once, never retried: replaying + the POST would start a second scan. + + :param ip: ``X.Y.Z.K`` or ``X.Y.Z.K/24`` + :param import_results: import the results into the ONYPHE dataset, + which makes them **PUBLIC** -- visible to every ONYPHE user + :param vulnscan: also run the vulnerability scan + :param urlscan: also crawl the HTTP services found + :param ports: ports to scan instead of the ONYPHE default + :param maxscantime: scan budget, in seconds + :returns: the raw envelope, which carries the scan identifier to hand + to :meth:`ondemand_scope_result` + """ + return self.send( + specs.ondemand_scope_ip( + ip, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=ports, + maxscantime=maxscantime, + ), + retry=False, + ) + + def ondemand_scope_domain( + self, + domain: str, + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, + ) -> Response: + """Launch an active scan of one domain. + + Needs an On-demand subscription. Options are the ones of + :meth:`ondemand_scope_ip`, ``import_results`` included: it makes the + results **PUBLIC** on ONYPHE. + """ + return self.send( + specs.ondemand_scope_domain( + domain, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=ports, + maxscantime=maxscantime, + ), + retry=False, + ) + + def ondemand_scope_ip_bulk( + self, + ips: Iterable[str], + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, + ) -> Response: + """Launch one active scan covering several IP addresses or CIDRs. + + Needs an On-demand subscription. Served from the ``/dev/`` prefix, so + ONYPHE may still move it. Options are the ones of + :meth:`ondemand_scope_ip`, ``import_results`` included: it makes the + results **PUBLIC** on ONYPHE. + + :param ips: any iterable of assets; joined with commas for you + """ + return self.send( + specs.ondemand_scope_ip_bulk( + ips, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=ports, + maxscantime=maxscantime, + ), + retry=False, + ) + + def ondemand_scope_domain_bulk( + self, + domains: Iterable[str], + *, + import_results: bool | None = None, + vulnscan: bool | None = None, + urlscan: bool | None = None, + ports: Iterable[int] | None = None, + maxscantime: int | None = None, + ) -> Response: + """Launch one active scan covering several domains. + + Needs an On-demand subscription. Served from the ``/dev/`` prefix, so + ONYPHE may still move it. Options are the ones of + :meth:`ondemand_scope_ip`, ``import_results`` included: it makes the + results **PUBLIC** on ONYPHE. + + :param domains: any iterable of domains; joined with commas for you + """ + return self.send( + specs.ondemand_scope_domain_bulk( + domains, + import_results=import_results, + vulnscan=vulnscan, + urlscan=urlscan, + ports=ports, + maxscantime=maxscantime, + ), + retry=False, + ) + + def ondemand_scope_result(self, scan_id: str) -> Response: + """Fetch the results of a scan launched earlier. + + :param scan_id: identifier returned when the scan was launched + :raises ScanInProgressError: while the scan runs or its results are + being built -- retry later, nothing failed + """ + try: + response = self.send(specs.ondemand_scope_result(scan_id)) + except APIError as exc: + self.raise_for_scan(scan_id, exc.payload, status_code=exc.status_code) + raise + self.raise_for_scan(scan_id, response.model_dump()) + return response + # -- alerts ------------------------------------------------------------- def alerts(self) -> list[Alert]: diff --git a/src/pyonyphe/errors.py b/src/pyonyphe/errors.py index 9e19538..06efe0a 100644 --- a/src/pyonyphe/errors.py +++ b/src/pyonyphe/errors.py @@ -3,6 +3,7 @@ from __future__ import annotations __all__ = [ + "SCAN_IN_PROGRESS_CODES", "APIError", "AuthenticationError", "ConfigError", @@ -11,10 +12,15 @@ "ParamError", "PaymentRequiredError", "RateLimitError", + "ScanInProgressError", "ServerError", "TransportError", ] +#: ONYPHE error codes meaning "not ready yet" rather than "failed": 103 is +#: ``Scan ID is in progress``, 111 is ``Scan ID results are being built``. +SCAN_IN_PROGRESS_CODES: frozenset[int] = frozenset({103, 111}) + class OnypheError(Exception): """Base class for every error raised by this library.""" @@ -88,5 +94,25 @@ def __init__( self.retry_after = retry_after +class ScanInProgressError(APIError): + """The On-demand scan is still running, or its results are being built. + + Not a failure: the same ``scan_id`` is worth asking for again later. + + :param scan_id: identifier of the scan that has no results yet + """ + + def __init__( + self, + message: str, + *, + scan_id: str, + status_code: int | None = None, + payload: dict[str, object] | None = None, + ) -> None: + super().__init__(message, status_code=status_code, payload=payload) + self.scan_id = scan_id + + class ServerError(APIError): """5xx - ONYPHE is having a bad day.""" diff --git a/tests/test_ondemand.py b/tests/test_ondemand.py new file mode 100644 index 0000000..a58e10d --- /dev/null +++ b/tests/test_ondemand.py @@ -0,0 +1,501 @@ +"""On-demand scope API: specs, both clients, and the CLI. + +No integration test here: the feature needs an On-demand subscription, and a +scan costs credits and touches third-party hosts. +""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from pathlib import Path + +import httpx +import pytest +import respx +from typer.testing import CliRunner + +from pyonyphe import AsyncOnyphe, Onyphe +from pyonyphe import _specs as specs +from pyonyphe.cli import app +from pyonyphe.errors import ( + NotFoundError, + ParamError, + RateLimitError, + ScanInProgressError, + ServerError, + TransportError, +) + +from .conftest import API_KEY, BASE, envelope + +runner = CliRunner() + +SINGLE_IP = f"{BASE}/ondemand/scope/ip/single" +SINGLE_DOMAIN = f"{BASE}/ondemand/scope/domain/single" +BULK_IP = f"{BASE}/dev/ondemand/scope/ip/bulk" +BULK_DOMAIN = f"{BASE}/dev/ondemand/scope/domain/bulk" +RESULT = f"{BASE}/ondemand/scope/result/abc123" + +#: The four launch endpoints, with the call that reaches each of them. +LAUNCHES = [ + (SINGLE_IP, lambda client: client.ondemand_scope_ip("8.8.8.8")), + (SINGLE_DOMAIN, lambda client: client.ondemand_scope_domain("example.com")), + (BULK_IP, lambda client: client.ondemand_scope_ip_bulk(["8.8.8.8"])), + (BULK_DOMAIN, lambda client: client.ondemand_scope_domain_bulk(["a.tld"])), +] + + +def launched(scan_id: str = "abc123") -> dict[str, object]: + """Envelope of a successful launch. ONYPHE does not document the field + name carrying the scan id, hence the raw passthrough.""" + return envelope(text="Success", scan_id=scan_id) + + +# -- specs ----------------------------------------------------------------- + + +def test_single_paths() -> None: + assert specs.ondemand_scope_ip("8.8.8.8").path == "ondemand/scope/ip/single" + assert specs.ondemand_scope_domain("example.com").path == "ondemand/scope/domain/single" + + +def test_bulk_paths_live_under_the_dev_prefix() -> None: + assert specs.ondemand_scope_ip_bulk(["8.8.8.8"]).path == "dev/ondemand/scope/ip/bulk" + assert specs.ondemand_scope_domain_bulk(["a.tld"]).path == "dev/ondemand/scope/domain/bulk" + assert specs.DEV_PREFIX == "dev" + + +def test_result_path_carries_the_scan_id() -> None: + spec = specs.ondemand_scope_result("abc123") + assert spec.method == "GET" + assert spec.path == "ondemand/scope/result/abc123" + + +def test_single_body_is_posted_as_json() -> None: + spec = specs.ondemand_scope_ip("8.8.8.0/24") + assert spec.method == "POST" + assert spec.json == {"ip": "8.8.8.0/24"} + assert spec.content is None + assert spec.stream is False + + +def test_options_are_serialised_the_way_onyphe_wants_them() -> None: + spec = specs.ondemand_scope_ip( + "8.8.8.8", + import_results=True, + vulnscan=True, + urlscan=False, + ports=[80, 443], + maxscantime=120, + ) + assert spec.json == { + "ip": "8.8.8.8", + "import": "true", + "vulnscan": "true", + "urlscan": "false", + "ports": "80,443", + "maxscantime": 120, + } + + +def test_options_left_out_are_not_sent() -> None: + assert specs.ondemand_scope_domain("example.com").json == {"domain": "example.com"} + + +def test_false_is_sent_explicitly_when_asked_for() -> None: + # None means "say nothing"; False means "say false". + spec = specs.ondemand_scope_ip("8.8.8.8", import_results=False) + assert spec.json == {"ip": "8.8.8.8", "import": "false"} + + +def test_bulk_joins_the_targets_with_commas() -> None: + spec = specs.ondemand_scope_ip_bulk(["1.1.1.1", " 8.8.8.8 ", "", "10.0.0.0/24"]) + assert spec.json == {"ip": "1.1.1.1,8.8.8.8,10.0.0.0/24"} + + +def test_bulk_accepts_any_iterable() -> None: + spec = specs.ondemand_scope_domain_bulk(iter(("a.tld", "b.tld"))) + assert spec.json == {"domain": "a.tld,b.tld"} + + +def test_bulk_does_not_split_a_bare_string() -> None: + assert specs.ondemand_scope_ip_bulk("1.1.1.1").json == {"ip": "1.1.1.1"} + + +def test_bulk_rejects_an_empty_list() -> None: + with pytest.raises(ParamError): + specs.ondemand_scope_ip_bulk([]) + + +@pytest.mark.parametrize( + "call", + [ + lambda: specs.ondemand_scope_ip(""), + lambda: specs.ondemand_scope_domain(""), + lambda: specs.ondemand_scope_result(""), + ], +) +def test_empty_target_is_rejected_before_any_request(call: Callable[[], object]) -> None: + with pytest.raises(ParamError): + call() + + +# -- sync client ----------------------------------------------------------- + + +@respx.mock +def test_scan_ip_posts_the_json_body(client: Onyphe) -> None: + route = respx.post(SINGLE_IP).mock(return_value=httpx.Response(200, json=launched())) + response = client.ondemand_scope_ip("8.8.8.8", vulnscan=True, ports=[80, 443]) + request = route.calls.last.request + assert request.headers["Content-Type"] == "application/json" + assert request.headers["Authorization"] == f"bearer {API_KEY}" + assert json.loads(request.content) == { + "ip": "8.8.8.8", + "vulnscan": "true", + "ports": "80,443", + } + # The scan id is handed back untouched: ONYPHE does not document its name. + assert response.model_dump()["scan_id"] == "abc123" + + +@respx.mock +def test_scan_domain_posts_the_json_body(client: Onyphe) -> None: + route = respx.post(SINGLE_DOMAIN).mock(return_value=httpx.Response(200, json=launched())) + client.ondemand_scope_domain("example.com", import_results=True) + assert json.loads(route.calls.last.request.content) == { + "domain": "example.com", + "import": "true", + } + + +@respx.mock +def test_scan_ip_bulk_joins_and_uses_the_dev_prefix(client: Onyphe) -> None: + route = respx.post(BULK_IP).mock(return_value=httpx.Response(200, json=launched())) + client.ondemand_scope_ip_bulk(["1.1.1.1", "8.8.8.8", "10.0.0.0/24"], maxscantime=60) + assert json.loads(route.calls.last.request.content) == { + "ip": "1.1.1.1,8.8.8.8,10.0.0.0/24", + "maxscantime": 60, + } + + +@respx.mock +def test_scan_domain_bulk_joins_and_uses_the_dev_prefix(client: Onyphe) -> None: + route = respx.post(BULK_DOMAIN).mock(return_value=httpx.Response(200, json=launched())) + client.ondemand_scope_domain_bulk(["a.tld", "b.tld"]) + assert json.loads(route.calls.last.request.content) == {"domain": "a.tld,b.tld"} + + +@respx.mock +def test_result_returns_the_envelope(client: Onyphe) -> None: + respx.get(RESULT).mock( + return_value=httpx.Response(200, json=envelope([{"ip": "8.8.8.8", "port": 443}])) + ) + response = client.ondemand_scope_result("abc123") + assert response.results[0]["port"] == 443 + + +@respx.mock +@pytest.mark.parametrize("code", [103, 111]) +def test_result_in_progress_on_a_2xx_body(client: Onyphe, code: int) -> None: + respx.get(RESULT).mock( + return_value=httpx.Response(200, json={"error": code, "text": "Scan ID is in progress"}) + ) + with pytest.raises(ScanInProgressError) as info: + client.ondemand_scope_result("abc123") + assert info.value.scan_id == "abc123" + assert "in progress" in str(info.value) + + +@respx.mock +@pytest.mark.parametrize("code", [103, 111]) +def test_result_in_progress_on_an_error_status(client: Onyphe, code: int) -> None: + # Same two codes, this time carried by a non-2xx answer: still not a failure. + respx.get(RESULT).mock( + return_value=httpx.Response(404, json={"error": code, "text": "results being built"}) + ) + with pytest.raises(ScanInProgressError) as info: + client.ondemand_scope_result("abc123") + assert info.value.scan_id == "abc123" + assert info.value.status_code == 404 + + +@respx.mock +def test_other_error_codes_keep_the_existing_mapping(client: Onyphe) -> None: + respx.get(RESULT).mock( + return_value=httpx.Response(404, json={"error": 100, "text": "no such scan"}) + ) + with pytest.raises(NotFoundError): + client.ondemand_scope_result("abc123") + + +# -- launches are never replayed ------------------------------------------- + + +@respx.mock +@pytest.mark.parametrize(("url", "launch"), LAUNCHES) +def test_launch_is_not_replayed_on_a_server_error( + url: str, launch: Callable[[Onyphe], object] +) -> None: + # Replaying a POST would start a second scan and burn credits. + route = respx.post(url).mock(return_value=httpx.Response(502, json={"text": "bad gateway"})) + with ( + Onyphe(API_KEY, max_retries=3, backoff=0.0) as client, + pytest.raises(ServerError), + ): + launch(client) + assert route.call_count == 1 + + +@respx.mock +def test_launch_is_not_replayed_on_a_rate_limit() -> None: + route = respx.post(SINGLE_IP).mock( + return_value=httpx.Response(429, headers={"Retry-After": "0"}, json={"text": "slow down"}) + ) + with ( + Onyphe(API_KEY, max_retries=3, backoff=0.0) as client, + pytest.raises(RateLimitError), + ): + client.ondemand_scope_ip("8.8.8.8") + assert route.call_count == 1 + + +@respx.mock +def test_launch_is_not_replayed_on_a_transport_failure() -> None: + # The scan may well have started: ONYPHE just never got to answer. + route = respx.post(SINGLE_IP).mock(side_effect=httpx.ConnectError("boom")) + with ( + Onyphe(API_KEY, max_retries=3, backoff=0.0) as client, + pytest.raises(TransportError), + ): + client.ondemand_scope_ip("8.8.8.8") + assert route.call_count == 1 + + +@respx.mock +def test_result_is_still_replayed() -> None: + route = respx.get(RESULT).mock( + side_effect=[ + httpx.Response(502, json={"text": "bad gateway"}), + httpx.Response(200, json=envelope([{"ip": "8.8.8.8"}])), + ] + ) + with Onyphe(API_KEY, max_retries=3, backoff=0.0) as client: + response = client.ondemand_scope_result("abc123") + assert response.results == [{"ip": "8.8.8.8"}] + assert route.call_count == 2 + + +# -- async client ---------------------------------------------------------- + + +@respx.mock +async def test_async_scan_ip(async_client: AsyncOnyphe) -> None: + route = respx.post(SINGLE_IP).mock(return_value=httpx.Response(200, json=launched())) + async with async_client as client: + await client.ondemand_scope_ip("8.8.8.8", import_results=False, urlscan=True) + assert json.loads(route.calls.last.request.content) == { + "ip": "8.8.8.8", + "import": "false", + "urlscan": "true", + } + + +@respx.mock +async def test_async_scan_domain_bulk(async_client: AsyncOnyphe) -> None: + route = respx.post(BULK_DOMAIN).mock(return_value=httpx.Response(200, json=launched())) + async with async_client as client: + await client.ondemand_scope_domain_bulk(["a.tld", "b.tld"], ports=[80]) + assert route.calls.last.request.url.path == "/api/v2/dev/ondemand/scope/domain/bulk" + assert json.loads(route.calls.last.request.content) == { + "domain": "a.tld,b.tld", + "ports": "80", + } + + +@respx.mock +async def test_async_result(async_client: AsyncOnyphe) -> None: + respx.get(RESULT).mock(return_value=httpx.Response(200, json=envelope([{"ip": "8.8.8.8"}]))) + async with async_client as client: + response = await client.ondemand_scope_result("abc123") + assert response.results == [{"ip": "8.8.8.8"}] + + +@respx.mock +@pytest.mark.parametrize("code", [103, 111]) +async def test_async_result_in_progress(async_client: AsyncOnyphe, code: int) -> None: + respx.get(RESULT).mock( + return_value=httpx.Response(200, json={"error": code, "text": "being built"}) + ) + async with async_client as client: + with pytest.raises(ScanInProgressError) as info: + await client.ondemand_scope_result("abc123") + assert info.value.scan_id == "abc123" + + +@respx.mock +async def test_async_launch_is_not_replayed_on_a_server_error() -> None: + route = respx.post(BULK_IP).mock(return_value=httpx.Response(502, json={"text": "bad gateway"})) + async with AsyncOnyphe(API_KEY, max_retries=3, backoff=0.0) as client: + with pytest.raises(ServerError): + await client.ondemand_scope_ip_bulk(["8.8.8.8"]) + assert route.call_count == 1 + + +@respx.mock +async def test_async_launch_is_not_replayed_on_a_rate_limit() -> None: + route = respx.post(SINGLE_DOMAIN).mock( + return_value=httpx.Response(429, headers={"Retry-After": "0"}, json={"text": "slow down"}) + ) + async with AsyncOnyphe(API_KEY, max_retries=3, backoff=0.0) as client: + with pytest.raises(RateLimitError): + await client.ondemand_scope_domain("example.com") + assert route.call_count == 1 + + +@respx.mock +async def test_async_launch_is_not_replayed_on_a_transport_failure() -> None: + route = respx.post(SINGLE_IP).mock(side_effect=httpx.ConnectError("boom")) + async with AsyncOnyphe(API_KEY, max_retries=3, backoff=0.0) as client: + with pytest.raises(TransportError): + await client.ondemand_scope_ip("8.8.8.8") + assert route.call_count == 1 + + +@respx.mock +async def test_async_result_is_still_replayed() -> None: + route = respx.get(RESULT).mock( + side_effect=[ + httpx.Response(502, json={"text": "bad gateway"}), + httpx.Response(200, json=envelope([{"ip": "8.8.8.8"}])), + ] + ) + async with AsyncOnyphe(API_KEY, max_retries=3, backoff=0.0) as client: + response = await client.ondemand_scope_result("abc123") + assert response.results == [{"ip": "8.8.8.8"}] + assert route.call_count == 2 + + +# -- CLI ------------------------------------------------------------------- + + +@respx.mock +def test_cli_ip_passes_every_option() -> None: + route = respx.post(SINGLE_IP).mock(return_value=httpx.Response(200, json=launched())) + result = runner.invoke( + app, + [ + "--api-key", + API_KEY, + "ondemand", + "ip", + "8.8.8.0/24", + "--import", + "--no-vulnscan", + "--urlscan", + "--ports", + "80,443", + "--maxscantime", + "120", + ], + ) + assert result.exit_code == 0 + assert json.loads(route.calls.last.request.content) == { + "ip": "8.8.8.0/24", + "import": "true", + "vulnscan": "false", + "urlscan": "true", + "ports": "80,443", + "maxscantime": 120, + } + assert "abc123" in result.stdout + + +@respx.mock +def test_cli_ip_sends_no_option_by_default() -> None: + route = respx.post(SINGLE_IP).mock(return_value=httpx.Response(200, json=launched())) + result = runner.invoke(app, ["--api-key", API_KEY, "ondemand", "ip", "8.8.8.8"]) + assert result.exit_code == 0 + assert json.loads(route.calls.last.request.content) == {"ip": "8.8.8.8"} + + +@respx.mock +def test_cli_domain() -> None: + route = respx.post(SINGLE_DOMAIN).mock(return_value=httpx.Response(200, json=launched())) + result = runner.invoke( + app, ["--api-key", API_KEY, "ondemand", "domain", "example.com", "--no-import"] + ) + assert result.exit_code == 0 + assert json.loads(route.calls.last.request.content) == { + "domain": "example.com", + "import": "false", + } + + +@respx.mock +def test_cli_ip_bulk_reads_one_target_per_line(tmp_path: Path) -> None: + targets = tmp_path / "ips.txt" + targets.write_text("1.1.1.1\n\n 8.8.8.8 \n10.0.0.0/24\n", encoding="utf-8") + route = respx.post(BULK_IP).mock(return_value=httpx.Response(200, json=launched())) + result = runner.invoke( + app, ["--api-key", API_KEY, "ondemand", "ip-bulk", str(targets), "--vulnscan"] + ) + assert result.exit_code == 0 + assert json.loads(route.calls.last.request.content) == { + "ip": "1.1.1.1,8.8.8.8,10.0.0.0/24", + "vulnscan": "true", + } + + +@respx.mock +def test_cli_domain_bulk_reads_one_target_per_line(tmp_path: Path) -> None: + targets = tmp_path / "domains.txt" + targets.write_text("a.tld\nb.tld\n", encoding="utf-8") + route = respx.post(BULK_DOMAIN).mock(return_value=httpx.Response(200, json=launched())) + result = runner.invoke(app, ["--api-key", API_KEY, "ondemand", "domain-bulk", str(targets)]) + assert result.exit_code == 0 + assert json.loads(route.calls.last.request.content) == {"domain": "a.tld,b.tld"} + + +def test_cli_bulk_missing_file_exits_with_2(tmp_path: Path) -> None: + result = runner.invoke( + app, ["--api-key", API_KEY, "ondemand", "ip-bulk", str(tmp_path / "nope.txt")] + ) + assert result.exit_code == 2 + + +def test_cli_rejects_a_bad_port_list() -> None: + result = runner.invoke( + app, ["--api-key", API_KEY, "ondemand", "ip", "8.8.8.8", "--ports", "80,https"] + ) + assert result.exit_code == 2 + + +@respx.mock +def test_cli_result() -> None: + respx.get(RESULT).mock( + return_value=httpx.Response(200, json=envelope([{"ip": "8.8.8.8", "port": 443}])) + ) + result = runner.invoke(app, ["--api-key", API_KEY, "ondemand", "result", "abc123"]) + assert result.exit_code == 0 + assert "8.8.8.8" in result.stdout + + +@respx.mock +@pytest.mark.parametrize("code", [103, 111]) +def test_cli_result_exits_with_3_while_the_scan_runs(code: int) -> None: + respx.get(RESULT).mock( + return_value=httpx.Response(200, json={"error": code, "text": "Scan ID is in progress"}) + ) + result = runner.invoke(app, ["--api-key", API_KEY, "ondemand", "result", "abc123"]) + assert result.exit_code == 3 + + +@respx.mock +def test_cli_result_api_error_exits_with_1() -> None: + respx.get(RESULT).mock( + return_value=httpx.Response(403, json={"text": "no on-demand in your license"}) + ) + result = runner.invoke(app, ["--api-key", API_KEY, "ondemand", "result", "abc123"]) + assert result.exit_code == 1