diff --git a/CHANGELOG.md b/CHANGELOG.md index 35a8328..0c0e2cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,40 @@ All notable changes to the Threat.Zone Python SDK are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## Unreleased + +### Added + +- `create_url_submission()` accepts `metafields` and `configurations`. Use + `metafields={"timeout": ..., "device_preset": ...}` to set the interactive session + duration and the emulated device. Use + `configurations={"networkConfig": ..., "sessionDurationSeconds": ...}` to select an + egress profile and the session duration (30-600 seconds). +- `get_url_analysis_session(uuid)` returns the interactive browser session state and the + viewer link as a `UrlAnalysisSession`. +- `start_url_analysis_session(uuid)` opens a session on an existing URL submission. +- `restart_url_analysis_session(uuid)` ends the current session and starts a clean one + with the same device and network choice. +- `restart_url_analysis(uuid)` re-runs a completed or errored URL-analysis report. +- `get_device_presets()` lists the selectable browser device presets as + `DevicePresetOption` values. +- `list_media_files()` and `get_media_file()` accept `source="dynamic"` or + `source="url_analysis"`. The parameter is omitted from the request when unset. +- `MediaFile.kind` reports `"video"`, `"screenshot"`, or `None`. +- `UrlAnalysisResponse` carries the scoring fields: `verdict_state`, `threat_score`, + `risk_level`, `privacy_score`, `privacy_risk_level`, `privacy_breakdown`, + `verdict_provenance`, `impersonation_target`, `completeness`, `intel_detections`, + `metafields`, and `scored_at`. Each is `None` or empty until scoring finishes. +- New types: `Message`, `DevicePresetOption`, `UrlAnalysisSession`, + `UrlAnalysisSessionConfig`, `UrlAnalysisPrivacyFactor`, `UrlAnalysisVerdictProvenance`, + `UrlAnalysisImpersonationTarget`, `UrlAnalysisCompleteness`, and + `UrlAnalysisIntelDetection`. +- The bundled `FakeThreatZoneAPI` serves the session, restart, and device-preset routes, + and returns the extended URL-analysis report. The new + `scenarios.seed_interactive_url_session()` seeds a submission with a running session. + +Every method above is mirrored on `AsyncThreatZone`. + ## 1.1.1 ### Fixed diff --git a/README.md b/README.md index 1128d73..48858a3 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,72 @@ See [Recipe 13](docs/RECIPES.md#13-control-dynamic_mimetype_check) for the full breakdown including reading back `FileInfo.is_mimetype_checked` after the submission completes. +## URL analysis + +Submit a URL, open an interactive browser session, then read the scored report. + +```python +# Pick an emulated device. Omit the metafield to use the plan default. +presets = client.get_device_presets() +mobile = next(p for p in presets if p.category == "mobile") + +created = client.create_url_submission( + "https://suspicious-landing.example.com/update.php", + private=True, + safe_browsing=True, + metafields={"timeout": 240, "device_preset": mobile.id}, + configurations={"networkConfig": ""}, +) + +# Poll for the viewer link. A session over workspace capacity is queued, not rejected. +session = client.get_url_analysis_session(created.uuid) +if session.state == "queued": + print(f"Queued at position {session.queue_position}") +if session.vnc_available: + print(f"Viewer: {session.link}") + +client.wait_for_completion(created.uuid, timeout=300) +report = client.get_url_analysis(created.uuid) + +print(report.threat_score, report.risk_level) # 88 Critical +print(report.privacy_score, report.privacy_risk_level) # 35 Medium +if report.impersonation_target is not None: + print(report.impersonation_target.brand_name) +for hit in report.intel_detections: + print(hit.source, hit.severity) +``` + +Session control: + +- `start_url_analysis_session(uuid)` opens a session on an existing submission. +- `restart_url_analysis_session(uuid)` ends the current session and starts a clean one + with the same device and network choice. +- Neither call consumes daily submission quota. +- `restart_url_analysis(uuid)` re-runs the analysis itself. It counts against the + concurrent-submission quota. + +Scoring fields (`threat_score`, `risk_level`, `privacy_score`, `privacy_risk_level`, +`privacy_breakdown`, `verdict_state`, `verdict_provenance`, `impersonation_target`, +`completeness`, `intel_detections`, `metafields`, `scored_at`) are `None` or empty until +scoring finishes. Read them after the report reaches `completed`. + +Media from the session: + +```python +media = client.list_media_files(created.uuid, source="url_analysis") +for item in media: + print(item.id, item.kind) # kind: "video", "screenshot", or None + +data = client.get_media_file(created.uuid, media[0].id, source="url_analysis") +``` + +`source` accepts `"dynamic"` or `"url_analysis"`. When omitted, the server picks +`dynamic` if the submission has a dynamic report, otherwise `url_analysis`. A submission +with neither report returns 409. + +See [recipe 2](./docs/RECIPES.md#2-submit-a-url-and-get-the-url-analysis-report) for the +full flow. + ## Core concepts - **Submissions** — one analysis target (file or URL) with a stable `uuid`, a workspace diff --git a/docs/RECIPES.md b/docs/RECIPES.md index 97c8e27..0bd3e06 100644 --- a/docs/RECIPES.md +++ b/docs/RECIPES.md @@ -118,11 +118,55 @@ def analyse_url(url: str) -> None: analyse_url("https://example-phish.test/login") ``` +**Interactive session and scoring.** + +Pass `safe_browsing=True` to run an interactive browser session alongside the analysis. +Poll `get_url_analysis_session()` for the viewer link, then read the scored report. + +```python +presets = client.get_device_presets() +mobile = next(p for p in presets if p.category == "mobile") + +created = client.create_url_submission( + url, + private=True, + safe_browsing=True, + metafields={"timeout": 240, "device_preset": mobile.id}, + configurations={"networkConfig": ""}, +) + +session = client.get_url_analysis_session(created.uuid) +if session.state == "queued": + print(f"Queued at position {session.queue_position}") +if session.vnc_available: + print(f"Viewer: {session.link}") + +client.wait_for_completion(created.uuid, timeout=300) +report = client.get_url_analysis(created.uuid) + +print(f"Score: {report.threat_score} ({report.risk_level})") +print(f"Privacy: {report.privacy_score} ({report.privacy_risk_level})") +if report.impersonation_target is not None: + print(f"Brand: {report.impersonation_target.brand_name}") +for hit in report.intel_detections: + print(f"Intel: {hit.source} {hit.severity}") + +for item in client.list_media_files(created.uuid, source="url_analysis"): + print(item.id, item.kind) +``` + **Notes.** - `screenshot.available` is `True` when a page screenshot is available. Fetch the bytes via `client.get_screenshot(uuid)`. - `whois` and `ssl_certificate` are both `Optional` — always null-check. +- The scoring fields are `None` (or `[]` for `intel_detections`) until scoring finishes. +- `start_url_analysis_session(uuid)` opens a session on an existing submission. + `restart_url_analysis_session(uuid)` gives a clean session with the same configuration. + Neither consumes daily submission quota. +- `restart_url_analysis(uuid)` re-runs the analysis. It counts against the + concurrent-submission quota. +- A session over workspace capacity is queued, not rejected. Read `queue_position`. - Use `create_open_in_browser_submission(url, environment="w10_x64")` instead if you need the full dynamic + browser capture on top of the URL analysis. diff --git a/docs/TYPES.md b/docs/TYPES.md index cfb8ab8..532e904 100644 --- a/docs/TYPES.md +++ b/docs/TYPES.md @@ -134,7 +134,7 @@ peer report type to dynamic/static/cdr — it is guarded separately by | Type | Purpose | |------|---------| -| `UrlAnalysisResponse` | Top-level report: `level`, `status`, `general_info`, `screenshot`, optional `ip_info`, `dns_records`, optional `whois`, optional `ssl_certificate`, optional `response_headers`, optional `extracted_file`, optional `threat_analysis`, `pages`. | +| `UrlAnalysisResponse` | Top-level report: `level`, `status`, `general_info`, `screenshot`, optional `ip_info`, `dns_records`, optional `whois`, optional `ssl_certificate`, optional `response_headers`, optional `extracted_file`, optional `threat_analysis`, `pages`, plus the scoring fields `verdict_state`, `threat_score`, `risk_level`, `privacy_score`, `privacy_risk_level`, `privacy_breakdown`, `verdict_provenance`, `impersonation_target`, `completeness`, `intel_detections`, `metafields`, `scored_at`. | | `UrlAnalysisGeneralInfo` | `{url, domain, website_title}`. | | `UrlAnalysisScreenshot` | `{available: bool}`. Fetch actual bytes with `client.get_screenshot(uuid)`. | | `UrlAnalysisIpInfo` | `{ip, asn, city, country, isp, organization, threat_status}`. | @@ -146,11 +146,21 @@ peer report type to dynamic/static/cdr — it is guarded separately by | `UrlAnalysisThreatAnalysis` | `{overview, blacklist, threat_details}`. | | `UrlAnalysisThreatOverviewItem` | `{source, title, description, verdict}`. Per-intel-source verdict. | | `UrlAnalysisThreatDetailItem` | `{source, details}`. Per-intel-source raw detail payload. | +| `UrlAnalysisPrivacyFactor` | `{factor, label, delta, detail}`. One contributor to the privacy score. | +| `UrlAnalysisVerdictProvenance` | `{raw_verdict, quorum_applied, allowlist_applied, collection_state, pass_}`. How the verdict was produced. `pass_` maps to the wire field `pass`. | +| `UrlAnalysisImpersonationTarget` | `{brand_name, canonical_domain, signals, confidence}`. Brand the page most likely impersonates. | +| `UrlAnalysisCompleteness` | `{measured_sources, skipped_sources, unavailable_sources, browser_collection}`. Which sources produced evidence. | +| `UrlAnalysisIntelDetection` | `{source, detection_type, severity, details}`. One threat-intelligence hit. | +| `UrlAnalysisSession` | Interactive browser session state: `state`, `collection_status`, `recording_status`, `queue_position`, `vnc_available`, `report_available`, `started_at`, `ready_at`, `expires_at`, `finished_at`, `failure_code`, `extensions_used`, `extensions_max`, `session_config`, `link`, `expired`. Returned by `get_url_analysis_session()`. | +| `UrlAnalysisSessionConfig` | `{device_profile_id, network_config_id, session_duration_seconds, device_preset_id, device_preset_name}`. Device and network choice the session runs with. | `UrlAnalysisResponse.level` is `ThreatLevel`, shared with submission level. Everything `Optional` is intentional — a submission may have no WHOIS (e.g. raw-IP URL), no TLS cert (plain HTTP), etc. +The scoring fields are `None` (or `[]` for `intel_detections`) until scoring finishes. +Read them after the report reaches `completed`. + --- ## Account @@ -222,6 +232,7 @@ Types shared across multiple surfaces. | `FileEntrypoint` | `{filename}`. Entrypoint inside an archive submission. | | `Tag` | `{type, value}`. | | `ReportStatus` | Per-report state — see Submissions section. | +| `Message` | `{message}`. One-line acknowledgement returned by an action endpoint. | | `ReportOperatingSystem` | `{name, platform}`. `platform` is one of `windows`/`linux`/`android`/`macos`. | | `ThreatLevel` | Literal union: `unknown` / `benign` / `suspicious` / `malicious`. | | `ReportType` | Literal union: `dynamic` / `static` / `cdr` / `url_analysis` / `open_in_browser`. | @@ -237,7 +248,8 @@ Types shared across multiple surfaces. | `Metafields` | Metafields grouped by scan type: `sandbox`, `static`, `cdr`, `url`, `open_in_browser`. | `GET /config/metafields` | | `MetafieldOption` | Single metafield: `key`, `label`, `description`, `type`, `default`, optional `options`. | same | | `EnvironmentOption` | `{key, name, platform, default}`. One available OS environment. | `GET /config/environments` | -| `MediaFile` | `{id, name, content_type, size}`. | `GET /submissions/:uuid/media` | +| `DevicePresetOption` | `{id, name, category, base_key, default}`. One selectable browser device preset. Pass `id` as the `device_preset` metafield. | `GET /config/device-presets` | +| `MediaFile` | `{id, name, content_type, size, kind}`. `kind` is `"video"`, `"screenshot"`, or `None`. | `GET /submissions/:uuid/media` | Download endpoints return `DownloadResponse` (sync) or `AsyncDownloadResponse` (async) from `threatzone._streaming`. Both are context managers; both expose `filename`, diff --git a/examples/recipe_02_url_analysis.py b/examples/recipe_02_url_analysis.py index 43227cf..12f9abe 100644 --- a/examples/recipe_02_url_analysis.py +++ b/examples/recipe_02_url_analysis.py @@ -1,4 +1,4 @@ -"""Recipe 2: Submit a URL, wait, print phishing findings. +"""Recipe 2: Submit a URL with an interactive session, wait, print the score. Runnable two ways: @@ -15,16 +15,59 @@ import os import sys +import time from threatzone import ReportUnavailableError, ThreatZone from threatzone.types import UrlAnalysisResponse +SESSION_POLL_ATTEMPTS = 30 +SESSION_POLL_INTERVAL = 2.0 + + +def wait_for_session_link( + client: ThreatZone, + uuid: str, + *, + attempts: int = SESSION_POLL_ATTEMPTS, + interval: float = SESSION_POLL_INTERVAL, +) -> str | None: + """Poll the interactive session until it runs. Return the viewer link.""" + for _ in range(attempts): + session = client.get_url_analysis_session(uuid) + if session.state == "running" and session.link is not None: + return session.link + if session.state in {"error", "finished"}: + print(f"Session ended early: {session.state} ({session.failure_code})") + return None + if session.state == "queued": + print(f"Session queued at position {session.queue_position}") + if interval > 0: + time.sleep(interval) + return None + def main(client: ThreatZone, url: str) -> UrlAnalysisResponse: - """Submit URL, wait, fetch URL analysis, print phishing-relevant fields.""" - created = client.create_url_submission(url, private=True) + """Submit a URL, open a session, wait, then print the scored report.""" + presets = client.get_device_presets() + mobile = next((p for p in presets if p.category == "mobile"), None) + + metafields: dict[str, object] = {"timeout": 240} + if mobile is not None: + metafields["device_preset"] = mobile.id + print(f"Using device preset {mobile.name}") + + created = client.create_url_submission( + url, + private=True, + safe_browsing=True, + metafields=metafields, + ) print(f"Created URL submission {created.uuid}") + link = wait_for_session_link(client, created.uuid, interval=0.0) + if link is not None: + print(f"Viewer: {link}") + final = client.wait_for_completion(created.uuid, timeout=300) try: @@ -36,6 +79,10 @@ def main(client: ThreatZone, url: str) -> UrlAnalysisResponse: print(f"URL: {report.general_info.url}") print(f"Domain: {report.general_info.domain}") print(f"Verdict: {report.level}") + print(f"Score: {report.threat_score} ({report.risk_level})") + if report.impersonation_target is not None: + target = report.impersonation_target + print(f"Brand: impersonates {target.brand_name} ({target.confidence} confidence)") if report.ip_info is not None: print(f"IP: {report.ip_info.ip} ({report.ip_info.country})") if report.threat_analysis is not None and report.threat_analysis.blacklist: diff --git a/src/threatzone/__init__.py b/src/threatzone/__init__.py index f6dce49..3e2853f 100644 --- a/src/threatzone/__init__.py +++ b/src/threatzone/__init__.py @@ -39,6 +39,7 @@ CdrSanitizedFileInfo, CdrSanitizedFileInfoDetail, Connection, + DevicePresetOption, DnsQuery, EmlAnalysis, EnvironmentOption, @@ -54,6 +55,7 @@ IoCsResponse, LimitsCount, MediaFile, + Message, MetafieldOption, Metafields, MitreResponse, @@ -82,6 +84,7 @@ SyscallsResponse, Tag, UrlAnalysisResponse, + UrlAnalysisSession, UserInfo, UserInfoDetails, WorkspaceBasicInfo, @@ -118,10 +121,12 @@ "FileInfo", "Tag", "ReportStatus", + "Message", # Types - Config "MetafieldOption", "Metafields", "EnvironmentOption", + "DevicePresetOption", # Types - Downloads "MediaFile", # Types - Indicators @@ -180,6 +185,7 @@ "SignatureCheckResult", # Types - URL analysis "UrlAnalysisResponse", + "UrlAnalysisSession", # Types - Submissions "SubmissionCreated", "SubmissionListItem", diff --git a/src/threatzone/_async_client.py b/src/threatzone/_async_client.py index b9563f1..4f64042 100644 --- a/src/threatzone/_async_client.py +++ b/src/threatzone/_async_client.py @@ -20,6 +20,7 @@ BehavioursResponse, CdrResponse, Connection, + DevicePresetOption, DnsQuery, EmlAnalysis, EnvironmentOption, @@ -30,6 +31,7 @@ IoCsResponse, IoCType, MediaFile, + Message, MetafieldOption, Metafields, MitreResponse, @@ -46,6 +48,7 @@ SubmissionCreated, SyscallsResponse, UrlAnalysisResponse, + UrlAnalysisSession, UserInfo, YaraRuleCategory, YaraRulesResponse, @@ -187,6 +190,15 @@ async def list_network_configs(self) -> list[NetworkConfigListItem]: data = response.json() return [NetworkConfigListItem.model_validate(item) for item in data.get("items", [])] + async def get_device_presets(self) -> list[DevicePresetOption]: + """List the browser device presets your plan can select. + + Pass a preset `id` as the `device_preset` metafield on + create_url_submission(). + """ + response = await self._http.get("/config/device-presets") + return [DevicePresetOption.model_validate(item) for item in response.json()] + # ========================================================================= # Submissions - Create # ========================================================================= @@ -321,24 +333,39 @@ async def create_url_submission( *, private: bool = False, safe_browsing: bool = False, + metafields: dict[str, Any] | None = None, + configurations: dict[str, Any] | None = None, ) -> SubmissionCreated: """ Create a new URL analysis submission. Args: - url: URL to analyze. + url: URL to analyze. A missing scheme is normalised to http:// server-side. private: If True, submission is private to your workspace. - safe_browsing: If True, spawn an isolated safe-browsing session - (a sandboxed Chromium pod accessible over noVNC) alongside - URL analysis. Default: False. + safe_browsing: If True, start an interactive browser session alongside + the analysis. Retrieve the viewer link with get_url_analysis_session(). + Sessions are queued when the workspace is at capacity. Default: False. + metafields: URL-analysis metafields. Discover the accepted keys with + get_metafields("url"). `timeout` sets the session duration in seconds; + `device_preset` selects the emulated device (get_device_presets()). + configurations: Per-scan configuration. `networkConfig` selects an egress + profile (list_network_configs()); `sessionDurationSeconds` sets the + interactive session duration (30-600, clamped by your plan). Returns: SubmissionCreated with the new submission UUID. """ - response = await self._http.post( - "/submissions/url_analysis", - json={"url": url, "private": private, "safeBrowsing": safe_browsing}, - ) + payload: dict[str, Any] = { + "url": url, + "private": private, + "safeBrowsing": safe_browsing, + } + if metafields is not None: + payload["metafields"] = metafields + if configurations is not None: + payload["configurations"] = configurations + + response = await self._http.post("/submissions/url_analysis", json=payload) return SubmissionCreated.model_validate(response.json()) async def create_open_in_browser_submission( @@ -922,6 +949,70 @@ async def get_url_analysis(self, uuid: str) -> UrlAnalysisResponse: response = await self._http.get(f"/submissions/{uuid}/url-analysis") return UrlAnalysisResponse.model_validate(response.json()) + async def get_url_analysis_session(self, uuid: str) -> UrlAnalysisSession: + """ + Get the interactive browser session state and viewer link for a URL submission. + + Args: + uuid: Submission UUID (must be a URL submission). + + Returns: + UrlAnalysisSession. `link` is None until `vnc_available` is True, and + again after the session expires. `queue_position` is set only while + `state` is "queued". + """ + response = await self._http.get(f"/submissions/{uuid}/url-analysis/session") + return UrlAnalysisSession.model_validate(response.json()) + + async def start_url_analysis_session(self, uuid: str) -> Message: + """ + Start an interactive browser session on an existing URL submission. + + The response confirms acceptance only. Poll get_url_analysis_session() + for the viewer link. A request over workspace capacity is queued, not + rejected. This call consumes no daily submission quota. + + Args: + uuid: Submission UUID (must be a URL submission). + + Returns: + Message acknowledging the request. + """ + response = await self._http.post(f"/submissions/{uuid}/url-analysis/session") + return Message.model_validate(response.json()) + + async def restart_url_analysis_session(self, uuid: str) -> Message: + """ + Tear down the current interactive session and start a fresh one. + + The stored session configuration is re-supplied, so the new session keeps + the original device and network choice. This call consumes no daily + submission quota. + + Args: + uuid: Submission UUID (must be a URL submission). + + Returns: + Message acknowledging the request. + """ + response = await self._http.post(f"/submissions/{uuid}/url-analysis/session/restart") + return Message.model_validate(response.json()) + + async def restart_url_analysis(self, uuid: str) -> Message: + """ + Restart a completed or errored URL-analysis report. + + Counts against the concurrent-submission quota. + + Args: + uuid: Submission UUID (must be a URL submission). + + Returns: + Message acknowledging the request. + """ + response = await self._http.post(f"/submissions/{uuid}/restart/url_analysis") + return Message.model_validate(response.json()) + # ========================================================================= # Network # ========================================================================= @@ -1207,29 +1298,37 @@ async def get_screenshot(self, uuid: str) -> bytes: response = await self._http.get(f"/submissions/{uuid}/screenshot") return response.content - async def list_media_files(self, uuid: str) -> list[MediaFile]: + async def list_media_files(self, uuid: str, *, source: str | None = None) -> list[MediaFile]: """ List available media files (screenshots, videos) for a submission. Args: uuid: Submission UUID. + source: Report family to read from: "dynamic" or "url_analysis". + Defaults server-side to "dynamic" when the submission has a + dynamic report, otherwise "url_analysis". Returns: - List of available media files. + List of available media files. Each item carries a `kind` of + "video", "screenshot", or None. """ - response = await self._http.get(f"/submissions/{uuid}/media") + response = await self._http.get(f"/submissions/{uuid}/media", params={"source": source}) return [MediaFile.model_validate(item) for item in response.json()] - async def get_media_file(self, uuid: str, file_id: str) -> bytes: + async def get_media_file(self, uuid: str, file_id: str, *, source: str | None = None) -> bytes: """ Get a specific media file. Args: uuid: Submission UUID. file_id: Media file ID from list_media_files(). + source: Report family to read from: "dynamic" or "url_analysis". + Defaults server-side as described on list_media_files(). Returns: Media file content as bytes. """ - response = await self._http.get(f"/submissions/{uuid}/media/{file_id}") + response = await self._http.get( + f"/submissions/{uuid}/media/{file_id}", params={"source": source} + ) return response.content diff --git a/src/threatzone/_sync_client.py b/src/threatzone/_sync_client.py index a31d0e3..86fe851 100644 --- a/src/threatzone/_sync_client.py +++ b/src/threatzone/_sync_client.py @@ -19,6 +19,7 @@ BehavioursResponse, CdrResponse, Connection, + DevicePresetOption, DnsQuery, EmlAnalysis, EnvironmentOption, @@ -29,6 +30,7 @@ IoCsResponse, IoCType, MediaFile, + Message, MetafieldOption, Metafields, MitreResponse, @@ -45,6 +47,7 @@ SubmissionCreated, SyscallsResponse, UrlAnalysisResponse, + UrlAnalysisSession, UserInfo, YaraRuleCategory, YaraRulesResponse, @@ -184,6 +187,15 @@ def list_network_configs(self) -> list[NetworkConfigListItem]: data = response.json() return [NetworkConfigListItem.model_validate(item) for item in data.get("items", [])] + def get_device_presets(self) -> list[DevicePresetOption]: + """List the browser device presets your plan can select. + + Pass a preset `id` as the `device_preset` metafield on + create_url_submission(). + """ + response = self._http.get("/config/device-presets") + return [DevicePresetOption.model_validate(item) for item in response.json()] + # ========================================================================= # Submissions - Create # ========================================================================= @@ -318,24 +330,39 @@ def create_url_submission( *, private: bool = False, safe_browsing: bool = False, + metafields: dict[str, Any] | None = None, + configurations: dict[str, Any] | None = None, ) -> SubmissionCreated: """ Create a new URL analysis submission. Args: - url: URL to analyze. + url: URL to analyze. A missing scheme is normalised to http:// server-side. private: If True, submission is private to your workspace. - safe_browsing: If True, spawn an isolated safe-browsing session - (a sandboxed Chromium pod accessible over noVNC) alongside - URL analysis. Default: False. + safe_browsing: If True, start an interactive browser session alongside + the analysis. Retrieve the viewer link with get_url_analysis_session(). + Sessions are queued when the workspace is at capacity. Default: False. + metafields: URL-analysis metafields. Discover the accepted keys with + get_metafields("url"). `timeout` sets the session duration in seconds; + `device_preset` selects the emulated device (get_device_presets()). + configurations: Per-scan configuration. `networkConfig` selects an egress + profile (list_network_configs()); `sessionDurationSeconds` sets the + interactive session duration (30-600, clamped by your plan). Returns: SubmissionCreated with the new submission UUID. """ - response = self._http.post( - "/submissions/url_analysis", - json={"url": url, "private": private, "safeBrowsing": safe_browsing}, - ) + payload: dict[str, Any] = { + "url": url, + "private": private, + "safeBrowsing": safe_browsing, + } + if metafields is not None: + payload["metafields"] = metafields + if configurations is not None: + payload["configurations"] = configurations + + response = self._http.post("/submissions/url_analysis", json=payload) return SubmissionCreated.model_validate(response.json()) def create_open_in_browser_submission( @@ -919,6 +946,70 @@ def get_url_analysis(self, uuid: str) -> UrlAnalysisResponse: response = self._http.get(f"/submissions/{uuid}/url-analysis") return UrlAnalysisResponse.model_validate(response.json()) + def get_url_analysis_session(self, uuid: str) -> UrlAnalysisSession: + """ + Get the interactive browser session state and viewer link for a URL submission. + + Args: + uuid: Submission UUID (must be a URL submission). + + Returns: + UrlAnalysisSession. `link` is None until `vnc_available` is True, and + again after the session expires. `queue_position` is set only while + `state` is "queued". + """ + response = self._http.get(f"/submissions/{uuid}/url-analysis/session") + return UrlAnalysisSession.model_validate(response.json()) + + def start_url_analysis_session(self, uuid: str) -> Message: + """ + Start an interactive browser session on an existing URL submission. + + The response confirms acceptance only. Poll get_url_analysis_session() + for the viewer link. A request over workspace capacity is queued, not + rejected. This call consumes no daily submission quota. + + Args: + uuid: Submission UUID (must be a URL submission). + + Returns: + Message acknowledging the request. + """ + response = self._http.post(f"/submissions/{uuid}/url-analysis/session") + return Message.model_validate(response.json()) + + def restart_url_analysis_session(self, uuid: str) -> Message: + """ + Tear down the current interactive session and start a fresh one. + + The stored session configuration is re-supplied, so the new session keeps + the original device and network choice. This call consumes no daily + submission quota. + + Args: + uuid: Submission UUID (must be a URL submission). + + Returns: + Message acknowledging the request. + """ + response = self._http.post(f"/submissions/{uuid}/url-analysis/session/restart") + return Message.model_validate(response.json()) + + def restart_url_analysis(self, uuid: str) -> Message: + """ + Restart a completed or errored URL-analysis report. + + Counts against the concurrent-submission quota. + + Args: + uuid: Submission UUID (must be a URL submission). + + Returns: + Message acknowledging the request. + """ + response = self._http.post(f"/submissions/{uuid}/restart/url_analysis") + return Message.model_validate(response.json()) + # ========================================================================= # Network # ========================================================================= @@ -1204,29 +1295,35 @@ def get_screenshot(self, uuid: str) -> bytes: response = self._http.get(f"/submissions/{uuid}/screenshot") return response.content - def list_media_files(self, uuid: str) -> list[MediaFile]: + def list_media_files(self, uuid: str, *, source: str | None = None) -> list[MediaFile]: """ List available media files (screenshots, videos) for a submission. Args: uuid: Submission UUID. + source: Report family to read from: "dynamic" or "url_analysis". + Defaults server-side to "dynamic" when the submission has a + dynamic report, otherwise "url_analysis". Returns: - List of available media files. + List of available media files. Each item carries a `kind` of + "video", "screenshot", or None. """ - response = self._http.get(f"/submissions/{uuid}/media") + response = self._http.get(f"/submissions/{uuid}/media", params={"source": source}) return [MediaFile.model_validate(item) for item in response.json()] - def get_media_file(self, uuid: str, file_id: str) -> bytes: + def get_media_file(self, uuid: str, file_id: str, *, source: str | None = None) -> bytes: """ Get a specific media file. Args: uuid: Submission UUID. file_id: Media file ID from list_media_files(). + source: Report family to read from: "dynamic" or "url_analysis". + Defaults server-side as described on list_media_files(). Returns: Media file content as bytes. """ - response = self._http.get(f"/submissions/{uuid}/media/{file_id}") + response = self._http.get(f"/submissions/{uuid}/media/{file_id}", params={"source": source}) return response.content diff --git a/src/threatzone/testing/_responses.py b/src/threatzone/testing/_responses.py index 30ed557..9b312fa 100644 --- a/src/threatzone/testing/_responses.py +++ b/src/threatzone/testing/_responses.py @@ -20,6 +20,7 @@ BehavioursResponse, CdrResponse, Connection, + DevicePresetOption, DnsQuery, EmlAnalysis, EnvironmentOption, @@ -28,6 +29,7 @@ IndicatorsResponse, IoCsResponse, MediaFile, + Message, MetafieldOption, Metafields, MitreResponse, @@ -44,6 +46,7 @@ SubmissionListItem, SyscallsResponse, UrlAnalysisResponse, + UrlAnalysisSession, UserInfo, YaraRulesResponse, ) @@ -778,10 +781,160 @@ def build_url_analysis_response(state: SubmissionState) -> UrlAnalysisResponse: "threatDetails": [], }, "pages": [state.final_url] if state.final_url else [target_url], + "verdictState": "final" if state.current_status() == "completed" else "provisional", + "threatScore": _THREAT_SCORE_BY_LEVEL[state.level], + "riskLevel": _RISK_LEVEL_BY_LEVEL[state.level], + "privacyScore": 35, + "privacyRiskLevel": "Medium", + "privacyBreakdown": [ + { + "factor": "trackers", + "label": "Third-party trackers", + "delta": 20, + "detail": "5 trackers", + } + ], + "verdictProvenance": { + "rawVerdict": state.level, + "quorumApplied": False, + "allowlistApplied": False, + "collectionState": "ok", + "pass": "final" if state.current_status() == "completed" else "provisional", + }, + "impersonationTarget": ( + { + "brandName": "PayPal", + "canonicalDomain": "paypal.com", + "signals": ["favicon_match"], + "confidence": "high", + } + if state.level == "malicious" + else None + ), + "completeness": { + "measuredSources": ["dns", "tls", "intel"], + "skippedSources": [] if state.session_active() else ["browser"], + "unavailableSources": [], + "browserCollection": "complete" if state.session_active() else None, + }, + "intelDetections": ( + [ + { + "source": "urlhaus", + "detectionType": "malware_url", + "severity": "high", + "details": state.threat_analysis_summary or "listed", + } + ] + if state.level == "malicious" + else [] + ), + "metafields": dict(state.metafields) or None, + "scoredAt": state.updated_at.isoformat(), } return UrlAnalysisResponse.model_validate(payload) +_THREAT_SCORE_BY_LEVEL: dict[str, int] = { + "unknown": 0, + "benign": 5, + "suspicious": 45, + "malicious": 88, +} + +_RISK_LEVEL_BY_LEVEL: dict[str, str] = { + "unknown": "Unknown", + "benign": "Low", + "suspicious": "Medium", + "malicious": "Critical", +} + + +def build_url_analysis_session(state: SubmissionState) -> UrlAnalysisSession: + """Build the interactive browser session state for a URL submission.""" + active = state.session_active() + started_at = int(state.created_at.timestamp() * 1000) + duration = state.session_duration_seconds or 120 + payload: dict[str, Any] = { + "state": "running" if active else "unavailable", + "collectionStatus": "collecting" if active else "unavailable", + "recordingStatus": "recording" if active else "unavailable", + "queuePosition": None, + "vncAvailable": active, + "reportAvailable": state.current_status() == "completed", + "startedAt": started_at if active else None, + "readyAt": started_at + 4000 if active else None, + "expiresAt": started_at + duration * 1000 if active else None, + "finishedAt": None, + "failureCode": None, + "extensionsUsed": 0, + "extensionsMax": 3, + "sessionConfig": ( + { + "deviceProfileId": device_preset_base_key_for(state.session_device_preset_id), + "networkConfigId": state.session_network_config_id, + "sessionDurationSeconds": state.session_duration_seconds, + "devicePresetId": state.session_device_preset_id, + "devicePresetName": state.session_device_preset_name, + } + if active + else None + ), + "link": (f"wss://fake.threat.zone/vnc/{state.uuid}?token=fake-session-token") + if active + else None, + "expired": not active, + } + return UrlAnalysisSession.model_validate(payload) + + +def build_message(text: str) -> Message: + """Build a one-line acknowledgement returned by an action endpoint.""" + return Message.model_validate({"message": text}) + + +DEVICE_PRESETS: list[dict[str, Any]] = [ + { + "id": "68b0000000000000000000a2", + "name": "Desktop", + "category": "desktop", + "baseKey": "desktop-chrome", + "default": True, + }, + { + "id": "68b0000000000000000000a1", + "name": "iPhone 15 Pro", + "category": "mobile", + "baseKey": "iphone-15-pro", + "default": False, + }, +] + + +def build_device_presets() -> list[DevicePresetOption]: + """Build the device-preset catalog served by /config/device-presets.""" + return [DevicePresetOption.model_validate(item) for item in DEVICE_PRESETS] + + +def device_preset_name_for(preset_id: str | None) -> str | None: + """Resolve a device-preset id to its display name.""" + return _device_preset_field(preset_id, "name") + + +def device_preset_base_key_for(preset_id: str | None) -> str | None: + """Resolve a device-preset id to the base browser profile it derives from.""" + return _device_preset_field(preset_id, "baseKey") + + +def _device_preset_field(preset_id: str | None, key: str) -> str | None: + if preset_id is None: + return None + for item in DEVICE_PRESETS: + if item["id"] == preset_id: + return str(item[key]) + return None + + def _domain_from_url(url: str) -> str: stripped = url.split("://", 1)[-1] return stripped.split("/", 1)[0] @@ -796,6 +949,7 @@ def build_media_files(state: SubmissionState) -> list[MediaFile]: "name": f"{mid}.png", "contentType": "image/png", "size": 68, + "kind": "screenshot", } ) for mid in state.media_ids diff --git a/src/threatzone/testing/fake_api.py b/src/threatzone/testing/fake_api.py index 450fdaf..67cdca6 100644 --- a/src/threatzone/testing/fake_api.py +++ b/src/threatzone/testing/fake_api.py @@ -34,6 +34,8 @@ ThreatLevelStr = Literal["unknown", "benign", "suspicious", "malicious"] ReportTypeStr = Literal["dynamic", "static", "cdr", "url_analysis"] +_MEDIA_SOURCES: frozenset[str] = frozenset({"dynamic", "url_analysis"}) + _MIN_PNG = ( b"\x89PNG\r\n\x1a\n" @@ -275,6 +277,9 @@ def register_url_analysis( final_url: str | None = None, screenshot_available: bool = True, threat_analysis_summary: str = "Phishing kit detected", + safe_browsing: bool = False, + device_preset_id: str | None = None, + session_duration_seconds: int | None = None, ) -> None: """Seed a URL-submission scenario keyed by its URL string.""" submission_uuid = _deterministic_uuid(f"url::{url}") @@ -294,6 +299,10 @@ def register_url_analysis( threat_analysis_summary=threat_analysis_summary, artifact_ids=[f"art-{submission_uuid[:8]}-1"], media_ids=[f"media-{submission_uuid[:8]}-1"], + safe_browsing=safe_browsing, + session_device_preset_id=device_preset_id, + session_device_preset_name=_responses.device_preset_name_for(device_preset_id), + session_duration_seconds=session_duration_seconds, ) self._url_to_uuid[url] = submission_uuid self._states[submission_uuid] = state @@ -427,6 +436,7 @@ def _handlers(self) -> dict[str, Callable[[httpx.Request, RouteMatch], httpx.Res "get_metafields_all": self._handle_metafields_all, "get_metafields_by_type": self._handle_metafields_by_type, "get_environments": self._handle_environments, + "get_device_presets": self._handle_device_presets, "list_network_configs": self._handle_list_network_configs, "list_submissions": self._handle_list_submissions, "get_submission": self._handle_get_submission, @@ -465,6 +475,10 @@ def _handlers(self) -> dict[str, Callable[[httpx.Request, RouteMatch], httpx.Res "download_html_report": self._handle_download_html, "download_cdr": self._handle_download_cdr, "get_url_analysis": self._handle_get_url_analysis, + "get_url_analysis_session": self._handle_get_url_analysis_session, + "start_url_analysis_session": self._handle_start_url_analysis_session, + "restart_url_analysis_session": self._handle_restart_url_analysis_session, + "restart_url_analysis": self._handle_restart_url_analysis, "get_screenshot": self._handle_get_screenshot, "list_media": self._handle_list_media, "get_media_file": self._handle_get_media_file, @@ -565,6 +579,23 @@ def _require_cdr(self, state: SubmissionState) -> httpx.Response | None: ) return None + def _require_url_submission(self, state: SubmissionState) -> httpx.Response | None: + """Interactive sessions exist only on URL submissions, report or not.""" + if state.type != "url": + return _error( + 409, + "Conflict", + "This submission is not a URL submission, so it has no interactive session.", + "URL_ANALYSIS_REPORT_UNAVAILABLE", + { + "submissionUuid": state.uuid, + "requiredReport": "url_analysis", + "currentStatus": "not_started", + "availableReports": state.available_reports(), + }, + ) + return None + def _require_url_analysis(self, state: SubmissionState) -> httpx.Response | None: if not state.has_url_analysis_report: return _error( @@ -609,6 +640,10 @@ def _handle_environments(self, request: httpx.Request, match: RouteMatch) -> htt del request, match return _json_list(list(_responses.build_environments())) + def _handle_device_presets(self, request: httpx.Request, match: RouteMatch) -> httpx.Response: + del request, match + return _json_list(list(_responses.build_device_presets())) + def _handle_list_network_configs( self, request: httpx.Request, match: RouteMatch ) -> httpx.Response: @@ -758,8 +793,34 @@ def _handle_create_url(self, request: httpx.Request, match: RouteMatch) -> httpx else: self.register_url_analysis(url=target_url, verdict="unknown") state = self._states[self._url_to_uuid[target_url]] + self._apply_url_options(state, payload) return _json(_responses.build_submission_created(state), status_code=200) + @staticmethod + def _apply_url_options(state: SubmissionState, payload: dict[str, Any]) -> None: + """Record the create-time session options the SDK sent.""" + metafields = payload.get("metafields") + if isinstance(metafields, dict): + state.metafields = dict(metafields) + preset_id = metafields.get("device_preset") + if isinstance(preset_id, str): + state.session_device_preset_id = preset_id + state.session_device_preset_name = _responses.device_preset_name_for(preset_id) + timeout = metafields.get("timeout") + if isinstance(timeout, int): + state.session_duration_seconds = timeout + + configurations = payload.get("configurations") + if isinstance(configurations, dict): + network_config = configurations.get("networkConfig") + if isinstance(network_config, str): + state.session_network_config_id = network_config + duration = configurations.get("sessionDurationSeconds") + if isinstance(duration, int): + state.session_duration_seconds = duration + + state.safe_browsing = payload.get("safeBrowsing") is True + def _handle_create_open_in_browser( self, request: httpx.Request, match: RouteMatch ) -> httpx.Response: @@ -1195,6 +1256,59 @@ def _handle_get_url_analysis( return guard return _json(_responses.build_url_analysis_response(state)) + def _handle_get_url_analysis_session( + self, _request: httpx.Request, match: RouteMatch + ) -> httpx.Response: + state, err = self._require_state(match.params["uuid"]) + if err is not None or state is None: + assert err is not None + return err + guard = self._require_url_submission(state) + if guard is not None: + return guard + return _json(_responses.build_url_analysis_session(state)) + + def _handle_start_url_analysis_session( + self, _request: httpx.Request, match: RouteMatch + ) -> httpx.Response: + state, err = self._require_state(match.params["uuid"]) + if err is not None or state is None: + assert err is not None + return err + guard = self._require_url_submission(state) + if guard is not None: + return guard + state.session_started = True + return _json(_responses.build_message("Interactive browser session start initiated")) + + def _handle_restart_url_analysis_session( + self, _request: httpx.Request, match: RouteMatch + ) -> httpx.Response: + state, err = self._require_state(match.params["uuid"]) + if err is not None or state is None: + assert err is not None + return err + guard = self._require_url_submission(state) + if guard is not None: + return guard + state.session_started = True + state.session_restarts += 1 + return _json(_responses.build_message("Interactive browser session restart initiated")) + + def _handle_restart_url_analysis( + self, _request: httpx.Request, match: RouteMatch + ) -> httpx.Response: + state, err = self._require_state(match.params["uuid"]) + if err is not None or state is None: + assert err is not None + return err + guard = self._require_url_analysis(state) + if guard is not None: + return guard + state.report_restarts += 1 + state.polls_seen = 0 + return _json(_responses.build_message("URL report restart initiated successfully")) + def _handle_get_screenshot(self, _request: httpx.Request, match: RouteMatch) -> httpx.Response: state, err = self._require_state(match.params["uuid"]) if err is not None or state is None: @@ -1205,18 +1319,63 @@ def _handle_get_screenshot(self, _request: httpx.Request, match: RouteMatch) -> return guard return _binary(_MIN_PNG, "image/png") - def _handle_list_media(self, _request: httpx.Request, match: RouteMatch) -> httpx.Response: + def _handle_list_media(self, request: httpx.Request, match: RouteMatch) -> httpx.Response: state, err = self._require_state(match.params["uuid"]) if err is not None or state is None: assert err is not None return err + source, source_err = self._resolve_media_source(request, state) + if source_err is not None: + return source_err + del source return _json_list(list(_responses.build_media_files(state))) - def _handle_get_media_file(self, _request: httpx.Request, match: RouteMatch) -> httpx.Response: + def _resolve_media_source( + self, request: httpx.Request, state: SubmissionState + ) -> tuple[str | None, httpx.Response | None]: + """Validate the `source` query param against the submission's reports.""" + requested = self._first(self._query(request), "source") + if requested is not None and requested not in _MEDIA_SOURCES: + return None, _error( + 400, + "Bad Request", + "source is neither dynamic nor url_analysis", + "INVALID_QUERY_PARAM", + {"param": "source"}, + ) + + if state.has_dynamic_report: + available = "dynamic" + elif state.has_url_analysis_report: + available = "url_analysis" + else: + return None, _error( + 409, + "Conflict", + "The submission has no media-bearing report", + "MEDIA_NOT_FOUND", + {"submissionUuid": state.uuid}, + ) + + if requested is not None and requested != available: + return None, _error( + 409, + "Conflict", + f"The submission has no {requested} media", + "MEDIA_NOT_FOUND", + {"submissionUuid": state.uuid, "source": requested}, + ) + return available, None + + def _handle_get_media_file(self, request: httpx.Request, match: RouteMatch) -> httpx.Response: state, err = self._require_state(match.params["uuid"]) if err is not None or state is None: assert err is not None return err + source, source_err = self._resolve_media_source(request, state) + if source_err is not None: + return source_err + del source file_id = match.params["file_id"] if file_id not in state.media_ids: return _error( diff --git a/src/threatzone/testing/routes.py b/src/threatzone/testing/routes.py index 1fff742..d234738 100644 --- a/src/threatzone/testing/routes.py +++ b/src/threatzone/testing/routes.py @@ -31,6 +31,7 @@ class RouteMatch: re.compile(r"^/config/metafields/(?P[A-Za-z_]+)$"), ), ("get_environments", re.compile(r"^/config/environments$")), + ("get_device_presets", re.compile(r"^/config/device-presets$")), ("list_network_configs", re.compile(r"^/v1/network-configs$")), ("list_submissions", re.compile(r"^/submissions$")), ( @@ -113,6 +114,10 @@ class RouteMatch: "download_cdr", re.compile(rf"^/submissions/(?P{_UUID})/download/cdr$"), ), + ( + "get_url_analysis_session", + re.compile(rf"^/submissions/(?P{_UUID})/url-analysis/session$"), + ), ( "get_url_analysis", re.compile(rf"^/submissions/(?P{_UUID})/url-analysis$"), @@ -136,6 +141,18 @@ class RouteMatch: ("create_cdr", re.compile(r"^/submissions/cdr$")), ("create_url_analysis", re.compile(r"^/submissions/url_analysis$")), ("create_open_in_browser", re.compile(r"^/submissions/open_in_browser$")), + ( + "restart_url_analysis_session", + re.compile(rf"^/submissions/(?P{_UUID})/url-analysis/session/restart$"), + ), + ( + "start_url_analysis_session", + re.compile(rf"^/submissions/(?P{_UUID})/url-analysis/session$"), + ), + ( + "restart_url_analysis", + re.compile(rf"^/submissions/(?P{_UUID})/restart/url_analysis$"), + ), ] diff --git a/src/threatzone/testing/scenarios.py b/src/threatzone/testing/scenarios.py index a0d3c59..fc73daf 100644 --- a/src/threatzone/testing/scenarios.py +++ b/src/threatzone/testing/scenarios.py @@ -97,6 +97,32 @@ def seed_phishing_url(fake: FakeThreatZoneAPI, *, url: str = "https://phishing.e return url +def seed_interactive_url_session( + fake: FakeThreatZoneAPI, + *, + url: str = "https://interactive.example.com", + device_preset_id: str = "68b0000000000000000000a1", + session_duration_seconds: int = 240, +) -> str: + """Seed a URL submission that already runs an interactive browser session. + + ``GET /submissions/{uuid}/url-analysis/session`` reports ``state == "running"`` + and returns a viewer link. + """ + fake.register_url_analysis( + url=url, + verdict="suspicious", + advance_after_polls=1, + final_url=url, + screenshot_available=True, + threat_analysis_summary="Credential harvesting form", + safe_browsing=True, + device_preset_id=device_preset_id, + session_duration_seconds=session_duration_seconds, + ) + return url + + def seed_static_only_submission(fake: FakeThreatZoneAPI, *, sha256: str | None = None) -> str: """Seed a submission with ONLY a static report. Dynamic/CDR endpoints return ``409 DYNAMIC_REPORT_UNAVAILABLE``/``CDR_REPORT_UNAVAILABLE``.""" diff --git a/src/threatzone/testing/state.py b/src/threatzone/testing/state.py index 4d7d178..dc741f4 100644 --- a/src/threatzone/testing/state.py +++ b/src/threatzone/testing/state.py @@ -81,6 +81,15 @@ class SubmissionState: updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) artifact_ids: list[str] = field(default_factory=list) media_ids: list[str] = field(default_factory=list) + metafields: dict[str, Any] = field(default_factory=dict) + safe_browsing: bool = False + session_started: bool = False + session_restarts: int = 0 + report_restarts: int = 0 + session_device_preset_id: str | None = None + session_device_preset_name: str | None = None + session_network_config_id: str | None = None + session_duration_seconds: int | None = None extra: dict[str, Any] = field(default_factory=dict) def current_status(self) -> StatusStr: @@ -127,3 +136,7 @@ def report_status_for(self, report_type: str) -> str: if report_type in self.report_status_overrides: return self.report_status_overrides[report_type] return self.current_status() + + def session_active(self) -> bool: + """True when an interactive browser session was requested or started.""" + return self.safe_browsing or self.session_started diff --git a/src/threatzone/types/__init__.py b/src/threatzone/types/__init__.py index f5f1d77..32a0bd4 100644 --- a/src/threatzone/types/__init__.py +++ b/src/threatzone/types/__init__.py @@ -7,6 +7,7 @@ FileInfo, FileSource, Hashes, + Message, OperatingSystemPlatform, ReportOperatingSystem, ReportStatus, @@ -16,6 +17,7 @@ ThreatLevel, ) from .config import ( + DevicePresetOption, EnvironmentOption, MetafieldOption, Metafields, @@ -110,17 +112,24 @@ ) from .syscalls import SyscallsResponse from .url_analysis import ( + UrlAnalysisCompleteness, UrlAnalysisDnsRecord, UrlAnalysisExtractedFile, UrlAnalysisGeneralInfo, + UrlAnalysisImpersonationTarget, + UrlAnalysisIntelDetection, UrlAnalysisIpInfo, + UrlAnalysisPrivacyFactor, UrlAnalysisResponse, UrlAnalysisScreenshot, + UrlAnalysisSession, + UrlAnalysisSessionConfig, UrlAnalysisSslCertificate, UrlAnalysisThreatAnalysis, UrlAnalysisThreatDetailItem, UrlAnalysisThreatOverviewItem, UrlAnalysisThreatStatus, + UrlAnalysisVerdictProvenance, UrlAnalysisWhoisInfo, ) @@ -130,6 +139,7 @@ "FileInfo", "FileSource", "Hashes", + "Message", "OperatingSystemPlatform", "ReportOperatingSystem", "ReportStatus", @@ -141,6 +151,7 @@ "ApiError", "ApiErrorCode", # Config + "DevicePresetOption", "EnvironmentOption", "MetafieldOption", "Metafields", @@ -229,17 +240,24 @@ "SignatureCheckResponse", "SignatureCheckResult", # URL analysis + "UrlAnalysisCompleteness", "UrlAnalysisDnsRecord", "UrlAnalysisExtractedFile", "UrlAnalysisGeneralInfo", + "UrlAnalysisImpersonationTarget", + "UrlAnalysisIntelDetection", "UrlAnalysisIpInfo", + "UrlAnalysisPrivacyFactor", "UrlAnalysisResponse", "UrlAnalysisScreenshot", + "UrlAnalysisSession", + "UrlAnalysisSessionConfig", "UrlAnalysisSslCertificate", "UrlAnalysisThreatAnalysis", "UrlAnalysisThreatDetailItem", "UrlAnalysisThreatOverviewItem", "UrlAnalysisThreatStatus", + "UrlAnalysisVerdictProvenance", "UrlAnalysisWhoisInfo", # Submissions "PaginatedSubmissions", diff --git a/src/threatzone/types/common.py b/src/threatzone/types/common.py index 8361836..250e393 100644 --- a/src/threatzone/types/common.py +++ b/src/threatzone/types/common.py @@ -14,6 +14,14 @@ OperatingSystemPlatform = Literal["windows", "linux", "android", "macos"] +class Message(BaseModel): + """A one-line acknowledgement returned by an action endpoint.""" + + model_config = ConfigDict(populate_by_name=True) + + message: str + + class Hashes(BaseModel): """File hash values.""" diff --git a/src/threatzone/types/config.py b/src/threatzone/types/config.py index 6894ff4..1e87f15 100644 --- a/src/threatzone/types/config.py +++ b/src/threatzone/types/config.py @@ -4,7 +4,7 @@ from typing import Literal -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, Field class MetafieldOptionValue(BaseModel): @@ -55,3 +55,15 @@ class EnvironmentOption(BaseModel): default: bool active: bool accessible: bool + + +class DevicePresetOption(BaseModel): + """A browser device preset selectable for the interactive session.""" + + model_config = ConfigDict(populate_by_name=True) + + id: str + name: str + category: str + base_key: str = Field(alias="baseKey") + default: bool = False diff --git a/src/threatzone/types/downloads.py b/src/threatzone/types/downloads.py index 9950c99..892c86d 100644 --- a/src/threatzone/types/downloads.py +++ b/src/threatzone/types/downloads.py @@ -14,3 +14,4 @@ class MediaFile(BaseModel): name: str content_type: str | None = Field(default=None, alias="contentType") size: int | None = None + kind: str | None = None diff --git a/src/threatzone/types/url_analysis.py b/src/threatzone/types/url_analysis.py index 59e7f63..ab70193 100644 --- a/src/threatzone/types/url_analysis.py +++ b/src/threatzone/types/url_analysis.py @@ -5,6 +5,7 @@ from __future__ import annotations +from datetime import datetime from typing import Any from pydantic import BaseModel, ConfigDict, Field @@ -133,6 +134,97 @@ class UrlAnalysisThreatAnalysis(BaseModel): threat_details: list[UrlAnalysisThreatDetailItem] = Field(alias="threatDetails") +class UrlAnalysisPrivacyFactor(BaseModel): + """One contributor to the privacy risk score.""" + + model_config = ConfigDict(populate_by_name=True) + + factor: str + label: str + delta: int + detail: str | None = None + + +class UrlAnalysisVerdictProvenance(BaseModel): + """How the final verdict was produced.""" + + model_config = ConfigDict(populate_by_name=True) + + raw_verdict: str = Field(alias="rawVerdict") + quorum_applied: bool = Field(alias="quorumApplied") + allowlist_applied: bool = Field(alias="allowlistApplied") + collection_state: str | None = Field(default=None, alias="collectionState") + pass_: str = Field(alias="pass") + + +class UrlAnalysisImpersonationTarget(BaseModel): + """Brand the analysed page most likely impersonates.""" + + model_config = ConfigDict(populate_by_name=True) + + brand_name: str = Field(alias="brandName") + canonical_domain: str | None = Field(default=None, alias="canonicalDomain") + signals: list[str] = Field(default_factory=list) + confidence: str + + +class UrlAnalysisCompleteness(BaseModel): + """Which sources produced evidence for the score.""" + + model_config = ConfigDict(populate_by_name=True) + + measured_sources: list[str] = Field(default_factory=list, alias="measuredSources") + skipped_sources: list[str] = Field(default_factory=list, alias="skippedSources") + unavailable_sources: list[str] = Field(default_factory=list, alias="unavailableSources") + browser_collection: str | None = Field(default=None, alias="browserCollection") + + +class UrlAnalysisIntelDetection(BaseModel): + """One threat-intelligence hit recorded during analysis.""" + + model_config = ConfigDict(populate_by_name=True) + + source: str + detection_type: str = Field(alias="detectionType") + severity: str + details: str | None = None + + +class UrlAnalysisSessionConfig(BaseModel): + """Device and network selection the interactive session ran with.""" + + model_config = ConfigDict(populate_by_name=True) + + device_profile_id: str | None = Field(default=None, alias="deviceProfileId") + network_config_id: str | None = Field(default=None, alias="networkConfigId") + session_duration_seconds: int | None = Field(default=None, alias="sessionDurationSeconds") + device_preset_id: str | None = Field(default=None, alias="devicePresetId") + device_preset_name: str | None = Field(default=None, alias="devicePresetName") + + +class UrlAnalysisSession(BaseModel): + """State of the interactive browser session for a URL submission.""" + + model_config = ConfigDict(populate_by_name=True) + + state: str + collection_status: str = Field(alias="collectionStatus") + recording_status: str = Field(alias="recordingStatus") + queue_position: int | None = Field(default=None, alias="queuePosition") + vnc_available: bool = Field(alias="vncAvailable") + report_available: bool = Field(alias="reportAvailable") + started_at: int | None = Field(default=None, alias="startedAt") + ready_at: int | None = Field(default=None, alias="readyAt") + expires_at: int | None = Field(default=None, alias="expiresAt") + finished_at: int | None = Field(default=None, alias="finishedAt") + failure_code: str | None = Field(default=None, alias="failureCode") + extensions_used: int = Field(default=0, alias="extensionsUsed") + extensions_max: int = Field(default=0, alias="extensionsMax") + session_config: UrlAnalysisSessionConfig | None = Field(default=None, alias="sessionConfig") + link: str | None = None + expired: bool + + class UrlAnalysisResponse(BaseModel): """Full URL analysis report payload.""" @@ -151,3 +243,23 @@ class UrlAnalysisResponse(BaseModel): extracted_file: UrlAnalysisExtractedFile | None = Field(default=None, alias="extractedFile") threat_analysis: UrlAnalysisThreatAnalysis | None = Field(default=None, alias="threatAnalysis") pages: list[str] + verdict_state: str | None = Field(default=None, alias="verdictState") + threat_score: int | None = Field(default=None, alias="threatScore") + risk_level: str | None = Field(default=None, alias="riskLevel") + privacy_score: int | None = Field(default=None, alias="privacyScore") + privacy_risk_level: str | None = Field(default=None, alias="privacyRiskLevel") + privacy_breakdown: list[UrlAnalysisPrivacyFactor] | None = Field( + default=None, alias="privacyBreakdown" + ) + verdict_provenance: UrlAnalysisVerdictProvenance | None = Field( + default=None, alias="verdictProvenance" + ) + impersonation_target: UrlAnalysisImpersonationTarget | None = Field( + default=None, alias="impersonationTarget" + ) + completeness: UrlAnalysisCompleteness | None = None + intel_detections: list[UrlAnalysisIntelDetection] = Field( + default_factory=list, alias="intelDetections" + ) + metafields: dict[str, Any] | None = None + scored_at: datetime | None = Field(default=None, alias="scoredAt") diff --git a/tests/conftest.py b/tests/conftest.py index dcdf7e2..437dfb1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -874,9 +874,99 @@ def sample_url_analysis_response() -> dict[str, Any]: "threatDetails": [{"source": "intel-1", "details": {"family": "EvilKit"}}], }, "pages": ["https://evil.com/login"], + "verdictState": "final", + "threatScore": 72, + "riskLevel": "High", + "privacyScore": 35, + "privacyRiskLevel": "Medium", + "privacyBreakdown": [ + { + "factor": "trackers", + "label": "Third-party trackers", + "delta": 20, + "detail": "5 trackers", + } + ], + "verdictProvenance": { + "rawVerdict": "malicious", + "quorumApplied": False, + "allowlistApplied": False, + "collectionState": "ok", + "pass": "final", + }, + "impersonationTarget": { + "brandName": "PayPal", + "canonicalDomain": "paypal.com", + "signals": ["favicon_match"], + "confidence": "high", + }, + "completeness": { + "measuredSources": ["dns", "tls", "intel"], + "skippedSources": ["browser"], + "unavailableSources": [], + "browserCollection": "partial", + }, + "intelDetections": [ + { + "source": "urlhaus", + "detectionType": "malware_url", + "severity": "high", + "details": "listed", + } + ], + "metafields": {"timeout": 240, "record_session": True}, + "scoredAt": "2026-08-27T10:00:00.000Z", + } + + +@pytest.fixture +def url_analysis_session_payload() -> dict[str, Any]: + return { + "state": "running", + "collectionStatus": "collecting", + "recordingStatus": "recording", + "queuePosition": None, + "vncAvailable": True, + "reportAvailable": False, + "startedAt": 1756288800000, + "readyAt": 1756288805000, + "expiresAt": 1756288920000, + "finishedAt": None, + "failureCode": None, + "extensionsUsed": 0, + "extensionsMax": 3, + "sessionConfig": { + "deviceProfileId": "iphone-15-pro", + "networkConfigId": None, + "sessionDurationSeconds": 240, + "devicePresetId": "68b0000000000000000000a1", + "devicePresetName": "iPhone 15 Pro", + }, + "link": "wss://gateway.example/?token=abc", + "expired": False, } +@pytest.fixture +def device_presets_payload() -> list[dict[str, Any]]: + return [ + { + "id": "68b0000000000000000000a2", + "name": "Desktop", + "category": "desktop", + "baseKey": "desktop-chrome", + "default": True, + }, + { + "id": "68b0000000000000000000a1", + "name": "iPhone 15 Pro", + "category": "mobile", + "baseKey": "iphone-15-pro", + "default": False, + }, + ] + + # --------------------------------------------------------------------------- # Misc payloads # --------------------------------------------------------------------------- @@ -890,12 +980,14 @@ def sample_media_files() -> list[dict[str, Any]]: "name": "screenshot_001.png", "contentType": "image/png", "size": 50000, + "kind": "screenshot", }, { "id": "media-2", "name": "video.mp4", "contentType": "video/mp4", "size": 5000000, + "kind": "video", }, ] diff --git a/tests/integration/flows/test_flow_url_lifecycle.py b/tests/integration/flows/test_flow_url_lifecycle.py index a4e1b21..a5dc2fa 100644 --- a/tests/integration/flows/test_flow_url_lifecycle.py +++ b/tests/integration/flows/test_flow_url_lifecycle.py @@ -4,9 +4,15 @@ import pytest -from threatzone import ReportUnavailableError, ThreatZone +from threatzone import BadRequestError, ReportUnavailableError, ThreatZone from threatzone.testing import FakeThreatZoneAPI, scenarios -from threatzone.types import Submission, UrlAnalysisResponse +from threatzone.types import ( + DevicePresetOption, + Message, + Submission, + UrlAnalysisResponse, + UrlAnalysisSession, +) def test_url_submission_roundtrip(fake_api: FakeThreatZoneAPI, sync_client: ThreatZone) -> None: @@ -50,3 +56,123 @@ def test_dynamic_endpoints_unavailable_on_url( created = sync_client.create_url_submission(target) with pytest.raises(ReportUnavailableError): sync_client.get_processes(created.uuid) + + +def test_url_analysis_report_carries_the_scoring_fields( + fake_api: FakeThreatZoneAPI, sync_client: ThreatZone +) -> None: + target = scenarios.seed_phishing_url(fake_api) + created = sync_client.create_url_submission(target) + report = sync_client.get_url_analysis(created.uuid) + assert report.threat_score is not None + assert report.risk_level == "Critical" + assert report.verdict_provenance is not None + assert report.verdict_provenance.raw_verdict == "malicious" + assert report.impersonation_target is not None + assert report.completeness is not None + assert report.intel_detections[0].source == "urlhaus" + assert report.scored_at is not None + + +def test_url_submission_records_the_session_options( + fake_api: FakeThreatZoneAPI, sync_client: ThreatZone +) -> None: + target = scenarios.seed_phishing_url(fake_api) + created = sync_client.create_url_submission( + target, + safe_browsing=True, + metafields={"timeout": 240, "device_preset": "68b0000000000000000000a1"}, + configurations={"networkConfig": "68b0000000000000000000bb"}, + ) + session = sync_client.get_url_analysis_session(created.uuid) + assert isinstance(session, UrlAnalysisSession) + assert session.state == "running" + assert session.vnc_available is True + assert session.link is not None + assert session.session_config is not None + assert session.session_config.device_preset_name == "iPhone 15 Pro" + assert session.session_config.session_duration_seconds == 240 + assert session.session_config.network_config_id == "68b0000000000000000000bb" + + report = sync_client.get_url_analysis(created.uuid) + assert report.metafields == {"timeout": 240, "device_preset": "68b0000000000000000000a1"} + + +def test_url_session_is_unavailable_before_it_starts( + fake_api: FakeThreatZoneAPI, sync_client: ThreatZone +) -> None: + target = scenarios.seed_phishing_url(fake_api) + created = sync_client.create_url_submission(target) + session = sync_client.get_url_analysis_session(created.uuid) + assert session.state == "unavailable" + assert session.vnc_available is False + assert session.link is None + assert session.expired is True + + +def test_start_and_restart_the_url_session( + fake_api: FakeThreatZoneAPI, sync_client: ThreatZone +) -> None: + target = scenarios.seed_phishing_url(fake_api) + created = sync_client.create_url_submission(target) + + started = sync_client.start_url_analysis_session(created.uuid) + assert isinstance(started, Message) + assert sync_client.get_url_analysis_session(created.uuid).state == "running" + + restarted = sync_client.restart_url_analysis_session(created.uuid) + assert isinstance(restarted, Message) + assert sync_client.get_url_analysis_session(created.uuid).vnc_available is True + + +def test_seeded_interactive_session_is_running( + fake_api: FakeThreatZoneAPI, sync_client: ThreatZone +) -> None: + target = scenarios.seed_interactive_url_session(fake_api) + created = sync_client.create_url_submission(target, safe_browsing=True) + session = sync_client.get_url_analysis_session(created.uuid) + assert session.state == "running" + assert session.session_config is not None + assert session.session_config.device_preset_name == "iPhone 15 Pro" + + +def test_session_endpoints_reject_a_file_submission( + fake_api: FakeThreatZoneAPI, sync_client: ThreatZone +) -> None: + sha256 = scenarios.seed_malicious_pe(fake_api) + submissions = sync_client.search_by_sha256(sha256) + with pytest.raises(ReportUnavailableError): + sync_client.get_url_analysis_session(submissions[0].uuid) + + +def test_restart_url_analysis_report(fake_api: FakeThreatZoneAPI, sync_client: ThreatZone) -> None: + target = scenarios.seed_phishing_url(fake_api) + created = sync_client.create_url_submission(target) + result = sync_client.restart_url_analysis(created.uuid) + assert "restart" in result.message + + +def test_url_media_list_with_the_source_filter( + fake_api: FakeThreatZoneAPI, sync_client: ThreatZone +) -> None: + target = scenarios.seed_phishing_url(fake_api) + created = sync_client.create_url_submission(target) + media = sync_client.list_media_files(created.uuid, source="url_analysis") + assert len(media) >= 1 + assert media[0].kind == "screenshot" + + +def test_url_media_list_rejects_a_bad_source( + fake_api: FakeThreatZoneAPI, sync_client: ThreatZone +) -> None: + target = scenarios.seed_phishing_url(fake_api) + created = sync_client.create_url_submission(target) + with pytest.raises(BadRequestError): + sync_client.list_media_files(created.uuid, source="static") + + +def test_device_preset_catalog(sync_client: ThreatZone) -> None: + presets = sync_client.get_device_presets() + assert all(isinstance(p, DevicePresetOption) for p in presets) + assert any(p.default for p in presets) + assert {p.category for p in presets} == {"desktop", "mobile"} diff --git a/tests/test_async_client.py b/tests/test_async_client.py index 7231fd2..7ee130b 100644 --- a/tests/test_async_client.py +++ b/tests/test_async_client.py @@ -31,6 +31,7 @@ BehavioursResponse, CdrResponse, Connection, + DevicePresetOption, DnsQuery, EmlAnalysis, EnvironmentOption, @@ -39,6 +40,7 @@ IndicatorsResponse, IoCsResponse, MediaFile, + Message, MetafieldOption, Metafields, MitreResponse, @@ -55,6 +57,7 @@ SubmissionCreated, SyscallsResponse, UrlAnalysisResponse, + UrlAnalysisSession, UserInfo, YaraRulesResponse, ) @@ -371,6 +374,23 @@ async def test_get_metafields_404(self, api_key: str) -> None: # --------------------------------------------------------------------------- +class TestAsyncDevicePresets: + async def test_get_device_presets( + self, api_key: str, device_presets_payload: list[dict[str, Any]] + ) -> None: + ctx, mock = _patched_async_client(_make_response(200, device_presets_payload)) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + presets = await client.get_device_presets() + assert all(isinstance(p, DevicePresetOption) for p in presets) + assert [p.id for p in presets] == [ + "68b0000000000000000000a2", + "68b0000000000000000000a1", + ] + assert presets[0].default is True + assert _call_url(mock).endswith("/config/device-presets") + + class TestAsyncCreateSubmissions: async def test_create_sandbox_submission( self, api_key: str, sample_submission_created: dict[str, Any] @@ -475,6 +495,38 @@ async def test_create_url_submission_safe_browsing_defaults_false( _, kwargs = mock.call_args assert kwargs["json"]["safeBrowsing"] is False + async def test_create_url_submission_sends_metafields_and_configurations( + self, api_key: str, sample_submission_created: dict[str, Any] + ) -> None: + ctx, mock = _patched_async_client(_make_response(200, sample_submission_created)) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + await client.create_url_submission( + "https://example.com", + metafields={"timeout": 240, "device_preset": "68b0000000000000000000a1"}, + configurations={ + "networkConfig": "68b0000000000000000000bb", + "sessionDurationSeconds": 240, + }, + ) + _, kwargs = mock.call_args + assert kwargs["json"]["metafields"] == { + "timeout": 240, + "device_preset": "68b0000000000000000000a1", + } + assert kwargs["json"]["configurations"]["sessionDurationSeconds"] == 240 + + async def test_create_url_submission_omits_absent_option_blocks( + self, api_key: str, sample_submission_created: dict[str, Any] + ) -> None: + ctx, mock = _patched_async_client(_make_response(200, sample_submission_created)) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + await client.create_url_submission("https://example.com") + _, kwargs = mock.call_args + assert "metafields" not in kwargs["json"] + assert "configurations" not in kwargs["json"] + async def test_create_open_in_browser_submission( self, api_key: str, sample_submission_created: dict[str, Any] ) -> None: @@ -1240,6 +1292,70 @@ async def test_get_url_analysis_404(self, api_key: str) -> None: with pytest.raises(NotFoundError): await client.get_url_analysis("ghost") + async def test_get_url_analysis_session( + self, + api_key: str, + submission_uuid: str, + url_analysis_session_payload: dict[str, Any], + ) -> None: + ctx, mock = _patched_async_client(_make_response(200, url_analysis_session_payload)) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + session = await client.get_url_analysis_session(submission_uuid) + assert isinstance(session, UrlAnalysisSession) + assert session.state == "running" + assert session.vnc_available is True + assert session.link == "wss://gateway.example/?token=abc" + assert session.session_config is not None + assert session.session_config.device_preset_name == "iPhone 15 Pro" + assert _call_url(mock).endswith(f"/submissions/{submission_uuid}/url-analysis/session") + + async def test_get_url_analysis_session_503(self, api_key: str, submission_uuid: str) -> None: + ctx, _ = _patched_async_client( + _make_response(503, {"message": "session service unavailable"}) + ) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + with pytest.raises(InternalServerError): + await client.get_url_analysis_session(submission_uuid) + + async def test_start_url_analysis_session(self, api_key: str, submission_uuid: str) -> None: + ctx, mock = _patched_async_client( + _make_response(200, {"message": "Interactive browser session start initiated"}) + ) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + result = await client.start_url_analysis_session(submission_uuid) + assert isinstance(result, Message) + assert _call_method(mock) == "POST" + assert _call_url(mock).endswith(f"/submissions/{submission_uuid}/url-analysis/session") + + async def test_restart_url_analysis_session(self, api_key: str, submission_uuid: str) -> None: + ctx, mock = _patched_async_client( + _make_response(200, {"message": "Interactive browser session restart initiated"}) + ) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + result = await client.restart_url_analysis_session(submission_uuid) + assert isinstance(result, Message) + assert _call_method(mock) == "POST" + assert _call_url(mock).endswith( + f"/submissions/{submission_uuid}/url-analysis/session/restart" + ) + + async def test_restart_url_analysis_posts_to_the_restart_route( + self, api_key: str, submission_uuid: str + ) -> None: + ctx, mock = _patched_async_client( + _make_response(200, {"message": "URL report restart initiated successfully"}) + ) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + result = await client.restart_url_analysis(submission_uuid) + assert _call_method(mock) == "POST" + assert _call_url(mock).endswith(f"/submissions/{submission_uuid}/restart/url_analysis") + assert "restart" in result.message + # --------------------------------------------------------------------------- # Network endpoints @@ -1584,6 +1700,40 @@ async def test_list_media_files_404(self, api_key: str) -> None: with pytest.raises(NotFoundError): await client.list_media_files("ghost") + async def test_list_media_files_passes_the_source( + self, api_key: str, sample_media_files: list[dict[str, Any]] + ) -> None: + ctx, mock = _patched_async_client(_make_response(200, sample_media_files)) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + await client.list_media_files("sub-789", source="url_analysis") + assert _call_params(mock)["source"] == "url_analysis" + + async def test_list_media_files_omits_the_source_when_absent( + self, api_key: str, sample_media_files: list[dict[str, Any]] + ) -> None: + ctx, mock = _patched_async_client(_make_response(200, sample_media_files)) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + await client.list_media_files("sub-789") + assert "source" not in _call_params(mock) + + async def test_list_media_files_carries_the_kind( + self, api_key: str, sample_media_files: list[dict[str, Any]] + ) -> None: + ctx, _ = _patched_async_client(_make_response(200, sample_media_files)) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + files = await client.list_media_files("sub-789") + assert [f.kind for f in files] == ["screenshot", "video"] + + async def test_get_media_file_passes_the_source(self, api_key: str) -> None: + ctx, mock = _patched_async_client(_make_response(200, content=b"binary")) + with ctx: + async with AsyncThreatZone(api_key=api_key) as client: + await client.get_media_file("sub-789", "1.png", source="url_analysis") + assert _call_params(mock)["source"] == "url_analysis" + async def test_get_media_file_happy(self, api_key: str) -> None: ctx, mock = _patched_async_client(_make_response(200, content=b"binary")) with ctx: diff --git a/tests/test_sync_client.py b/tests/test_sync_client.py index 45a6762..4064747 100644 --- a/tests/test_sync_client.py +++ b/tests/test_sync_client.py @@ -36,6 +36,7 @@ BehavioursResponse, CdrResponse, Connection, + DevicePresetOption, DnsQuery, EmlAnalysis, EnvironmentOption, @@ -44,6 +45,7 @@ IndicatorsResponse, IoCsResponse, MediaFile, + Message, MetafieldOption, Metafields, MitreResponse, @@ -60,6 +62,7 @@ SubmissionCreated, SyscallsResponse, UrlAnalysisResponse, + UrlAnalysisSession, UserInfo, YaraRulesResponse, ) @@ -357,6 +360,28 @@ def test_get_metafields_404(self, api_key: str) -> None: # --------------------------------------------------------------------------- +class TestDevicePresets: + def test_get_device_presets( + self, api_key: str, device_presets_payload: list[dict[str, Any]] + ) -> None: + ctx, mock = _patched_client(_make_response(200, device_presets_payload)) + with ctx, ThreatZone(api_key=api_key) as client: + presets = client.get_device_presets() + assert all(isinstance(p, DevicePresetOption) for p in presets) + assert [p.id for p in presets] == [ + "68b0000000000000000000a2", + "68b0000000000000000000a1", + ] + assert presets[0].default is True + assert presets[1].base_key == "iphone-15-pro" + assert _call_url(mock).endswith("/config/device-presets") + + def test_get_device_presets_403(self, api_key: str) -> None: + ctx, _ = _patched_client(_make_response(403, {"message": "no plan"})) + with ctx, ThreatZone(api_key=api_key) as client, pytest.raises(PermissionDeniedError): + client.get_device_presets() + + class TestCreateSubmissions: def test_create_sandbox_submission( self, api_key: str, sample_submission_created: dict[str, Any] @@ -448,6 +473,37 @@ def test_create_url_submission_safe_browsing_defaults_false( _, kwargs = mock.call_args assert kwargs["json"]["safeBrowsing"] is False + def test_create_url_submission_sends_metafields_and_configurations( + self, api_key: str, sample_submission_created: dict[str, Any] + ) -> None: + ctx, mock = _patched_client(_make_response(200, sample_submission_created)) + with ctx, ThreatZone(api_key=api_key) as client: + client.create_url_submission( + "https://example.com", + metafields={"timeout": 240, "device_preset": "68b0000000000000000000a1"}, + configurations={ + "networkConfig": "68b0000000000000000000bb", + "sessionDurationSeconds": 240, + }, + ) + _, kwargs = mock.call_args + assert kwargs["json"]["metafields"] == { + "timeout": 240, + "device_preset": "68b0000000000000000000a1", + } + assert kwargs["json"]["configurations"]["sessionDurationSeconds"] == 240 + assert kwargs["json"]["configurations"]["networkConfig"] == "68b0000000000000000000bb" + + def test_create_url_submission_omits_absent_option_blocks( + self, api_key: str, sample_submission_created: dict[str, Any] + ) -> None: + ctx, mock = _patched_client(_make_response(200, sample_submission_created)) + with ctx, ThreatZone(api_key=api_key) as client: + client.create_url_submission("https://example.com") + _, kwargs = mock.call_args + assert "metafields" not in kwargs["json"] + assert "configurations" not in kwargs["json"] + def test_create_open_in_browser_submission( self, api_key: str, sample_submission_created: dict[str, Any] ) -> None: @@ -1136,6 +1192,63 @@ def test_get_url_analysis_404(self, api_key: str) -> None: with ctx, ThreatZone(api_key=api_key) as client, pytest.raises(NotFoundError): client.get_url_analysis("ghost") + def test_get_url_analysis_session( + self, + api_key: str, + submission_uuid: str, + url_analysis_session_payload: dict[str, Any], + ) -> None: + ctx, mock = _patched_client(_make_response(200, url_analysis_session_payload)) + with ctx, ThreatZone(api_key=api_key) as client: + session = client.get_url_analysis_session(submission_uuid) + assert isinstance(session, UrlAnalysisSession) + assert session.state == "running" + assert session.vnc_available is True + assert session.expired is False + assert session.link == "wss://gateway.example/?token=abc" + assert session.session_config is not None + assert session.session_config.device_preset_name == "iPhone 15 Pro" + assert _call_url(mock).endswith(f"/submissions/{submission_uuid}/url-analysis/session") + + def test_get_url_analysis_session_503(self, api_key: str, submission_uuid: str) -> None: + ctx, _ = _patched_client(_make_response(503, {"message": "session service unavailable"})) + with ctx, ThreatZone(api_key=api_key) as client, pytest.raises(InternalServerError): + client.get_url_analysis_session(submission_uuid) + + def test_start_url_analysis_session(self, api_key: str, submission_uuid: str) -> None: + ctx, mock = _patched_client( + _make_response(200, {"message": "Interactive browser session start initiated"}) + ) + with ctx, ThreatZone(api_key=api_key) as client: + result = client.start_url_analysis_session(submission_uuid) + assert isinstance(result, Message) + assert _call_method(mock) == "POST" + assert _call_url(mock).endswith(f"/submissions/{submission_uuid}/url-analysis/session") + + def test_restart_url_analysis_session(self, api_key: str, submission_uuid: str) -> None: + ctx, mock = _patched_client( + _make_response(200, {"message": "Interactive browser session restart initiated"}) + ) + with ctx, ThreatZone(api_key=api_key) as client: + result = client.restart_url_analysis_session(submission_uuid) + assert isinstance(result, Message) + assert _call_method(mock) == "POST" + assert _call_url(mock).endswith( + f"/submissions/{submission_uuid}/url-analysis/session/restart" + ) + + def test_restart_url_analysis_posts_to_the_restart_route( + self, api_key: str, submission_uuid: str + ) -> None: + ctx, mock = _patched_client( + _make_response(200, {"message": "URL report restart initiated successfully"}) + ) + with ctx, ThreatZone(api_key=api_key) as client: + result = client.restart_url_analysis(submission_uuid) + assert _call_method(mock) == "POST" + assert _call_url(mock).endswith(f"/submissions/{submission_uuid}/restart/url_analysis") + assert "restart" in result.message + # --------------------------------------------------------------------------- # Network endpoints @@ -1441,6 +1554,36 @@ def test_list_media_files_404(self, api_key: str) -> None: with ctx, ThreatZone(api_key=api_key) as client, pytest.raises(NotFoundError): client.list_media_files("ghost") + def test_list_media_files_passes_the_source( + self, api_key: str, sample_media_files: list[dict[str, Any]] + ) -> None: + ctx, mock = _patched_client(_make_response(200, sample_media_files)) + with ctx, ThreatZone(api_key=api_key) as client: + client.list_media_files("sub-789", source="url_analysis") + assert _call_params(mock)["source"] == "url_analysis" + + def test_list_media_files_omits_the_source_when_absent( + self, api_key: str, sample_media_files: list[dict[str, Any]] + ) -> None: + ctx, mock = _patched_client(_make_response(200, sample_media_files)) + with ctx, ThreatZone(api_key=api_key) as client: + client.list_media_files("sub-789") + assert "source" not in _call_params(mock) + + def test_list_media_files_carries_the_kind( + self, api_key: str, sample_media_files: list[dict[str, Any]] + ) -> None: + ctx, _ = _patched_client(_make_response(200, sample_media_files)) + with ctx, ThreatZone(api_key=api_key) as client: + files = client.list_media_files("sub-789") + assert [f.kind for f in files] == ["screenshot", "video"] + + def test_get_media_file_passes_the_source(self, api_key: str) -> None: + ctx, mock = _patched_client(_make_response(200, content=b"binary")) + with ctx, ThreatZone(api_key=api_key) as client: + client.get_media_file("sub-789", "1.png", source="url_analysis") + assert _call_params(mock)["source"] == "url_analysis" + def test_get_media_file_happy(self, api_key: str) -> None: ctx, mock = _patched_client(_make_response(200, content=b"binary")) with ctx, ThreatZone(api_key=api_key) as client: diff --git a/tests/test_types.py b/tests/test_types.py index 040e84b..9a289a8 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -26,6 +26,7 @@ CdrResponse, Connection, ConnectionPackets, + DevicePresetOption, DnsQuery, EmlAnalysis, EmlAnalysisAttachment, @@ -44,6 +45,7 @@ IoCsResponse, LimitsCount, MediaFile, + Message, MetafieldOption, Metafields, MitreResponse, @@ -72,6 +74,7 @@ SyscallsResponse, Tag, UrlAnalysisResponse, + UrlAnalysisSession, UserInfo, UserInfoDetails, WorkspaceBasicInfo, @@ -705,13 +708,143 @@ def test_url_analysis_round_trip(self, sample_url_analysis_response: dict[str, A second = UrlAnalysisResponse.model_validate(dumped) assert second.general_info.url == first.general_info.url + def test_url_analysis_response_keeps_the_scoring_fields( + self, sample_url_analysis_response: dict[str, Any] + ) -> None: + report = UrlAnalysisResponse.model_validate(sample_url_analysis_response) + assert report.threat_score == 72 + assert report.risk_level == "High" + assert report.verdict_state == "final" + assert report.privacy_score == 35 + assert report.privacy_risk_level == "Medium" + assert report.privacy_breakdown is not None + assert report.privacy_breakdown[0].factor == "trackers" + assert report.privacy_breakdown[0].delta == 20 + assert report.verdict_provenance is not None + assert report.verdict_provenance.raw_verdict == "malicious" + assert report.verdict_provenance.quorum_applied is False + assert report.verdict_provenance.pass_ == "final" + assert report.impersonation_target is not None + assert report.impersonation_target.brand_name == "PayPal" + assert report.impersonation_target.canonical_domain == "paypal.com" + assert report.completeness is not None + assert report.completeness.measured_sources == ["dns", "tls", "intel"] + assert report.completeness.browser_collection == "partial" + assert report.intel_detections[0].source == "urlhaus" + assert report.intel_detections[0].detection_type == "malware_url" + assert report.metafields == {"timeout": 240, "record_session": True} + assert report.scored_at is not None + + def test_url_analysis_response_tolerates_an_unscored_report(self) -> None: + report = UrlAnalysisResponse.model_validate( + { + "level": "unknown", + "status": "completed", + "generalInfo": {"url": "https://x", "domain": "x", "websiteTitle": None}, + "screenshot": {"available": False}, + "dnsRecords": [], + "pages": [], + } + ) + assert report.threat_score is None + assert report.risk_level is None + assert report.verdict_provenance is None + assert report.impersonation_target is None + assert report.completeness is None + assert report.intel_detections == [] + assert report.metafields is None + assert report.scored_at is None + + def test_url_analysis_scoring_round_trip( + self, sample_url_analysis_response: dict[str, Any] + ) -> None: + first = UrlAnalysisResponse.model_validate(sample_url_analysis_response) + dumped = first.model_dump(by_alias=True) + assert "threatScore" in dumped + assert "verdictProvenance" in dumped + assert dumped["verdictProvenance"]["pass"] == "final" + assert "intelDetections" in dumped + second = UrlAnalysisResponse.model_validate(dumped) + assert second.threat_score == first.threat_score + + +class TestUrlAnalysisSessionTypes: + def test_url_analysis_session(self, url_analysis_session_payload: dict[str, Any]) -> None: + session = UrlAnalysisSession.model_validate(url_analysis_session_payload) + assert session.state == "running" + assert session.collection_status == "collecting" + assert session.recording_status == "recording" + assert session.queue_position is None + assert session.vnc_available is True + assert session.report_available is False + assert session.started_at == 1756288800000 + assert session.ready_at == 1756288805000 + assert session.expires_at == 1756288920000 + assert session.finished_at is None + assert session.failure_code is None + assert session.extensions_used == 0 + assert session.extensions_max == 3 + assert session.link == "wss://gateway.example/?token=abc" + assert session.expired is False + assert session.session_config is not None + assert session.session_config.device_profile_id == "iphone-15-pro" + assert session.session_config.network_config_id is None + assert session.session_config.session_duration_seconds == 240 + assert session.session_config.device_preset_id == "68b0000000000000000000a1" + + def test_url_analysis_session_without_config( + self, url_analysis_session_payload: dict[str, Any] + ) -> None: + url_analysis_session_payload["sessionConfig"] = None + url_analysis_session_payload["link"] = None + session = UrlAnalysisSession.model_validate(url_analysis_session_payload) + assert session.session_config is None + assert session.link is None + + def test_url_analysis_session_round_trip( + self, url_analysis_session_payload: dict[str, Any] + ) -> None: + first = UrlAnalysisSession.model_validate(url_analysis_session_payload) + dumped = first.model_dump(by_alias=True) + assert "vncAvailable" in dumped + assert "sessionConfig" in dumped + second = UrlAnalysisSession.model_validate(dumped) + assert second.state == first.state + # --------------------------------------------------------------------------- # Config / downloads / me # --------------------------------------------------------------------------- +class TestMessageType: + def test_message(self) -> None: + msg = Message.model_validate({"message": "URL report restart initiated successfully"}) + assert msg.message == "URL report restart initiated successfully" + + def test_message_missing_field_raises(self) -> None: + with pytest.raises(ValidationError): + Message.model_validate({}) + + class TestConfigTypes: + def test_device_preset_option(self, device_presets_payload: list[dict[str, Any]]) -> None: + preset = DevicePresetOption.model_validate(device_presets_payload[1]) + assert preset.id == "68b0000000000000000000a1" + assert preset.name == "iPhone 15 Pro" + assert preset.category == "mobile" + assert preset.base_key == "iphone-15-pro" + assert preset.default is False + + def test_device_preset_option_round_trip( + self, device_presets_payload: list[dict[str, Any]] + ) -> None: + first = DevicePresetOption.model_validate(device_presets_payload[0]) + dumped = first.model_dump(by_alias=True) + assert "baseKey" in dumped + second = DevicePresetOption.model_validate(dumped) + assert second.base_key == first.base_key + def test_metafield_option(self, sample_metafields: dict[str, list[dict[str, Any]]]) -> None: opt = MetafieldOption.model_validate(sample_metafields["sandbox"][0]) assert opt.key == "timeout" @@ -754,6 +887,16 @@ def test_media_file_round_trip(self, sample_media_files: list[dict[str, Any]]) - dumped = first.model_dump(by_alias=True) assert "contentType" in dumped + def test_media_file_kind(self, sample_media_files: list[dict[str, Any]]) -> None: + screenshot = MediaFile.model_validate(sample_media_files[0]) + video = MediaFile.model_validate(sample_media_files[1]) + assert screenshot.kind == "screenshot" + assert video.kind == "video" + + def test_media_file_kind_defaults_to_none(self) -> None: + media = MediaFile.model_validate({"id": "report.txt", "name": "report.txt"}) + assert media.kind is None + class TestMeTypes: def test_workspace_basic_info(self, sample_user_info: dict[str, Any]) -> None: