diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c0e2cf..c04bb98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `Indicator.summary` (`str | None`) carries the one-sentence analyst summary of Windows V2 schema 2.1 rows. It is `None` on legacy rows and when an older server omits it. On rows with a summary, `Indicator.description` holds the long analyst text. +- `Indicator.score_maximum` (`10` legacy, `100` Windows V2) gives the scale of `Indicator.score`. +- `NetworkSummary` adds `counts_available`, `availability` (new `NetworkAvailability` model + with `state`, `reason`, `coverage`) and `source_scope`. `DnsQuery`, `HttpRequest`, + `Connection` and `NetworkThreat` add `source_scope`. Legacy reports send `counts_available=True`, `availability=None` and `source_scope=None`. Every new field is `None` when an older server omits it. Enum-like fields accept unknown values, so a new server value does not fail the call. + `SummaryNetwork` (`OverviewSummary.network`) adds the same three fields. + The `X-Network-Evidence-Availability` header of the list endpoints is not exposed, because + the list methods return plain lists. Use `get_network_summary().availability`. - `create_url_submission()` accepts `metafields` and `configurations`. Use `metafields={"timeout": ..., "device_preset": ...}` to set the interactive session duration and the emulated device. Use diff --git a/docs/TYPES.md b/docs/TYPES.md index 532e904..d44d4a5 100644 --- a/docs/TYPES.md +++ b/docs/TYPES.md @@ -48,7 +48,7 @@ in this group is envelope-wrapped (`items`, `total`) for server-side pagination. | Type | Purpose | Endpoint | |------|---------|----------| -| `Indicator` | A single rule hit with cross-reference IDs. Fields: `id`, `name`, `description`, `category`, `level`, `score`, `pids`, `attack_codes`, `event_ids`, `syscall_line_numbers`, `author`. | `GET /submissions/:uuid/indicators` | +| `Indicator` | A single rule hit with cross-reference IDs. Fields: `id`, `name`, `description`, `summary`, `category`, `level`, `score`, `score_maximum`, `pids`, `attack_codes`, `event_ids`, `syscall_line_numbers`, `author`. | `GET /submissions/:uuid/indicators` | | `IndicatorsResponse` | `{items, total, levels}` envelope. `levels` is a `{malicious, suspicious, benign}` severity rollup. | same | | `IndicatorLevels` | Severity counts: `malicious`, `suspicious`, `benign`. | nested | | `IoC` | `{type, value, artifacts}`. `type` is the 11-value literal below. | `GET /submissions/:uuid/iocs` | @@ -74,12 +74,16 @@ in this group is envelope-wrapped (`items`, `total`) for server-side pagination. - `Indicator.level` is `IndicatorLevel = Literal["malicious", "suspicious", "benign"]`. - `Indicator.author` is `Literal["system", "user"]`. +- `Indicator.score_maximum` is `int | None`: the scale of `score` (10 legacy, 100 Windows V2). `None` when the API does not send it. Divide `score` by `score_maximum` before comparing rows from different scales. +- `Indicator.summary` is `str | None`: a one-sentence plain-language summary, sent for Windows V2 schema 2.1 rows only. `None` on legacy rows and when the API does not send it. On rows with a summary, `description` holds the long analyst text. - `IoC.type` is `IoCType = Literal["ip", "domain", "url", "email", "sha512", "sha256", "sha1", "md5", "registry", "path", "uuid"]`. - `YaraRule.category` is `YaraRuleCategory = Literal["malicious", "suspicious", "benign"]`. - `Artifact.type` is `ArtifactType` (14-value literal including `sample`, `dropped_file`, `memory_dump`, `pcap`, `yara_rule`, …). - `OverviewSummary.network` is `Optional` — `None` on submissions without a - completed dynamic report. + completed dynamic report. `OverviewSummary.network` also carries `counts_available`, + `availability` and `source_scope` (same types as `NetworkSummary`). When `counts_available` + is `False` the counts are 0 and do not mean the sample had no network activity. Cross-references: `Indicator.pids` points at `Process.pid` (see Dynamic Scan), `Indicator.attack_codes` at `MitreResponse.techniques`, `Indicator.syscall_line_numbers` @@ -107,18 +111,22 @@ report. Every endpoint is guarded by `ReportStatusGuard('dynamic')`; read the co | `BehavioursResponse` | `{items, total, page, limit, total_pages}`. | same | | `BehaviourOs` | Deprecated compatibility alias; the endpoint derives the OS from the submission. | same | | `SyscallsResponse` | `{items: List[str], total}`. `items` are raw unparsed log lines. | `GET /submissions/:uuid/syscalls` | -| `NetworkSummary` | `{dns_count, http_count, tcp_count, udp_count, threat_count, pcap_available}`. | `GET /submissions/:uuid/network/summary` | -| `DnsQuery` | `{id, host, type, status, records, timeshift}`. | `GET /submissions/:uuid/network/dns` | -| `HttpRequest` | `{id, host, ip, port, country}` — slimmed-down shape, no method/URL/headers. | `GET /submissions/:uuid/network/http` | -| `Connection` | `{id, protocol, destination_ip, destination_port, domain, asn, country, packets, timeshift}`. Used for both TCP and UDP. | `GET /submissions/:uuid/network/tcp`, `GET /submissions/:uuid/network/udp` | +| `NetworkSummary` | `{dns_count, http_count, tcp_count, udp_count, threat_count, pcap_available, counts_available, availability, source_scope}`. Legacy reports send `counts_available=True`, `availability=None`, `source_scope=None`; all three are `None` when an older server omits them. | `GET /submissions/:uuid/network/summary` | +| `NetworkAvailability` | `{state, reason, coverage}`. Network evidence state of a Windows V2 report. | nested in `NetworkSummary` | +| `DnsQuery` | `{id, host, type, status, records, timeshift, source_scope}`. | `GET /submissions/:uuid/network/dns` | +| `HttpRequest` | `{id, host, ip, port, country, source_scope}` — slimmed-down shape, no method/URL/headers. | `GET /submissions/:uuid/network/http` | +| `Connection` | `{id, protocol, destination_ip, destination_port, domain, asn, country, packets, timeshift, source_scope}`. Used for both TCP and UDP. | `GET /submissions/:uuid/network/tcp`, `GET /submissions/:uuid/network/udp` | | `ConnectionPackets` | `{sent, received, empty}`. | nested | -| `NetworkThreat` | Suricata alert: `signature`, `description`, `severity`, `protocol`, `app_proto`, `destination_ip`, `destination_port`, `timeshift`, `metadata`, `details`, optional `tls`. | `GET /submissions/:uuid/network/threats` | +| `NetworkThreat` | Suricata alert: `signature`, `description`, `severity`, `protocol`, `app_proto`, `destination_ip`, `destination_port`, `timeshift`, `metadata`, `details`, optional `tls`, `source_scope`. | `GET /submissions/:uuid/network/threats` | | `NetworkThreatTls` | TLS fingerprint block: `version`, `sni`, `ja3_hash`, `ja3s_hash`, `ja3_string`, `ja3s_string`. Present iff `app_proto == "TLS"`. | nested in `NetworkThreat` | **Discriminative fields.** - `DnsQuery.type` is `DnsRecordType` (16-value literal: `A`, `AAAA`, `CNAME`, `MX`, …). - `DnsQuery.status` is `DnsStatus` (DNS RCODE literal, e.g. `NOERROR`, `NXDOMAIN`, `TIMEOUT`). +- `NetworkAvailability.state` is `NetworkAvailabilityState` (`disabled`, `unavailable`, `verified_zero`, `observed`); `coverage` is `NetworkCoverage` (`none`, `partial`, `complete`). When `NetworkSummary.counts_available` is `False`, all counts are 0 and do not mean the sample had no network activity. +- `source_scope` is `NetworkSourceScope` (`sample_attributed`, `submission_ip_scoped`, `submission_unattributed`, `none`) or `None` on legacy reports. Unknown values from a newer server are kept as plain strings. +- The list endpoints send the `X-Network-Evidence-Availability` response header on Windows V2. The `get_*` list methods return plain lists, so the SDK does not surface it. Read `get_network_summary().availability` instead. - `Connection.protocol` is `ConnectionProtocol = Literal["tcp", "udp"]`. - `NetworkThreat.protocol` is `ThreatProtocol = Literal["TCP", "UDP"]` (uppercase!). - `NetworkThreat.app_proto` is `ThreatAppProto = Literal["HTTP", "TLS", "DNS"]`. diff --git a/src/threatzone/__init__.py b/src/threatzone/__init__.py index 3e2853f..22fbe49 100644 --- a/src/threatzone/__init__.py +++ b/src/threatzone/__init__.py @@ -60,6 +60,7 @@ Metafields, MitreResponse, ModuleInfo, + NetworkAvailability, NetworkConfigListItem, NetworkConfigListResponse, NetworkSummary, @@ -155,6 +156,7 @@ "SubmissionLimits", "ModuleInfo", # Types - Network + "NetworkAvailability", "NetworkSummary", "DnsQuery", "HttpRequest", diff --git a/src/threatzone/types/__init__.py b/src/threatzone/types/__init__.py index 32a0bd4..35eba40 100644 --- a/src/threatzone/types/__init__.py +++ b/src/threatzone/types/__init__.py @@ -72,6 +72,10 @@ DnsRecordType, DnsStatus, HttpRequest, + NetworkAvailability, + NetworkAvailabilityState, + NetworkCoverage, + NetworkSourceScope, NetworkSummary, NetworkThreat, NetworkThreatTls, @@ -203,6 +207,10 @@ "DnsRecordType", "DnsStatus", "HttpRequest", + "NetworkAvailability", + "NetworkAvailabilityState", + "NetworkCoverage", + "NetworkSourceScope", "NetworkSummary", "NetworkThreat", "NetworkThreatTls", diff --git a/src/threatzone/types/indicators.py b/src/threatzone/types/indicators.py index ebc0f2a..b7a18d7 100644 --- a/src/threatzone/types/indicators.py +++ b/src/threatzone/types/indicators.py @@ -11,6 +11,8 @@ from pydantic import BaseModel, ConfigDict, Field +from .network import NetworkAvailability, NetworkSourceScope + IndicatorLevel = Literal["malicious", "suspicious", "benign"] IndicatorAuthor = Literal["system", "user"] @@ -33,9 +35,11 @@ class Indicator(BaseModel): id: str name: str description: str + summary: str | None = None category: list[str] level: IndicatorLevel score: int + score_maximum: int | None = Field(default=None, alias="scoreMaximum") pids: list[int] attack_codes: list[str] = Field(alias="attackCodes") event_ids: list[int] = Field(alias="eventIds") @@ -219,6 +223,9 @@ class SummaryNetwork(BaseModel): tcp_count: int = Field(alias="tcpCount") udp_count: int = Field(alias="udpCount") threat_count: int = Field(alias="threatCount") + counts_available: bool | None = Field(default=None, alias="countsAvailable") + availability: NetworkAvailability | None = None + source_scope: NetworkSourceScope | None = Field(default=None, alias="sourceScope") class OverviewSummary(BaseModel): diff --git a/src/threatzone/types/network.py b/src/threatzone/types/network.py index 74c6449..a8d2ac1 100644 --- a/src/threatzone/types/network.py +++ b/src/threatzone/types/network.py @@ -55,6 +55,28 @@ ThreatAppProto = Literal["HTTP", "TLS", "DNS"] ThreatSeverity = Literal["high", "low"] +NetworkAvailabilityState = Literal["disabled", "unavailable", "verified_zero", "observed"] | str +NetworkCoverage = Literal["none", "partial", "complete"] | str +NetworkSourceScope = ( + Literal[ + "sample_attributed", + "submission_ip_scoped", + "submission_unattributed", + "none", + ] + | str +) + + +class NetworkAvailability(BaseModel): + """Network evidence state of a Windows V2 report.""" + + model_config = ConfigDict(populate_by_name=True) + + state: NetworkAvailabilityState + reason: str | None = None + coverage: NetworkCoverage + class NetworkSummary(BaseModel): """Summary of network activity captured during dynamic analysis.""" @@ -67,6 +89,9 @@ class NetworkSummary(BaseModel): udp_count: int = Field(alias="udpCount") threat_count: int = Field(alias="threatCount") pcap_available: bool = Field(alias="pcapAvailable") + counts_available: bool | None = Field(default=None, alias="countsAvailable") + availability: NetworkAvailability | None = None + source_scope: NetworkSourceScope | None = Field(default=None, alias="sourceScope") class DnsQuery(BaseModel): @@ -80,6 +105,7 @@ class DnsQuery(BaseModel): status: DnsStatus records: list[str] timeshift: float + source_scope: NetworkSourceScope | None = Field(default=None, alias="sourceScope") class HttpRequest(BaseModel): @@ -92,6 +118,7 @@ class HttpRequest(BaseModel): ip: str port: int country: str | None = None + source_scope: NetworkSourceScope | None = Field(default=None, alias="sourceScope") class ConnectionPackets(BaseModel): @@ -118,6 +145,7 @@ class Connection(BaseModel): country: str | None = None packets: ConnectionPackets timeshift: float + source_scope: NetworkSourceScope | None = Field(default=None, alias="sourceScope") class NetworkThreatTls(BaseModel): @@ -149,3 +177,4 @@ class NetworkThreat(BaseModel): metadata: dict[str, list[str]] details: dict[str, str | int] tls: NetworkThreatTls | None = None + source_scope: NetworkSourceScope | None = Field(default=None, alias="sourceScope") diff --git a/tests/test_types.py b/tests/test_types.py index 9a289a8..16bac81 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -50,6 +50,7 @@ Metafields, MitreResponse, ModuleInfo, + NetworkAvailability, NetworkConfigListItem, NetworkConfigListResponse, NetworkSummary, @@ -285,6 +286,33 @@ def test_indicator(self, sample_indicator: dict[str, Any]) -> None: assert ind.syscall_line_numbers == [12, 100] assert ind.author == "system" + def test_indicator_score_maximum(self, sample_indicator: dict[str, Any]) -> None: + v2 = Indicator.model_validate({**sample_indicator, "score": 45, "scoreMaximum": 100}) + assert v2.score == 45 + assert v2.score_maximum == 100 + legacy = Indicator.model_validate({**sample_indicator, "scoreMaximum": 10}) + assert legacy.score_maximum == 10 + + def test_indicator_unknown_score_maximum(self, sample_indicator: dict[str, Any]) -> None: + ind = Indicator.model_validate({**sample_indicator, "scoreMaximum": 1000}) + assert ind.score_maximum == 1000 + + def test_indicator_without_score_maximum(self, sample_indicator: dict[str, Any]) -> None: + ind = Indicator.model_validate(sample_indicator) + assert ind.score_maximum is None + assert ind.score == sample_indicator["score"] + + def test_indicator_summary(self, sample_indicator: dict[str, Any]) -> None: + ind = Indicator.model_validate( + {**sample_indicator, "summary": "word.exe hollowed svchost.exe.", "scoreMaximum": 100} + ) + assert ind.summary == "word.exe hollowed svchost.exe." + assert ind.description == sample_indicator["description"] + + def test_indicator_without_summary(self, sample_indicator: dict[str, Any]) -> None: + ind = Indicator.model_validate(sample_indicator) + assert ind.summary is None + def test_indicator_category_is_list(self, sample_indicator: dict[str, Any]) -> None: ind = Indicator.model_validate(sample_indicator) assert ind.category == ["evasion", "persistence"] @@ -416,6 +444,66 @@ def test_overview_summary_without_network( summary = OverviewSummary.model_validate(sample_overview_summary) assert summary.network is None + def test_overview_summary_network_v2_unavailable( + self, sample_overview_summary: dict[str, Any] + ) -> None: + sample_overview_summary["network"] = { + "dnsCount": 0, + "httpCount": 0, + "tcpCount": 0, + "udpCount": 0, + "threatCount": 0, + "countsAvailable": False, + "availability": { + "state": "unavailable", + "reason": "capture_unavailable", + "coverage": "none", + }, + "sourceScope": "none", + } + network = OverviewSummary.model_validate(sample_overview_summary).network + assert network is not None + assert network.dns_count == 0 + assert network.counts_available is False + assert network.availability == NetworkAvailability( + state="unavailable", reason="capture_unavailable", coverage="none" + ) + assert network.source_scope == "none" + + def test_overview_summary_network_legacy_has_no_availability( + self, sample_overview_summary: dict[str, Any] + ) -> None: + network = OverviewSummary.model_validate(sample_overview_summary).network + assert network is not None + assert network.dns_count == 15 + assert network.counts_available is None + assert network.availability is None + assert network.source_scope is None + + def test_overview_summary_network_explicit_null_and_unknown_values( + self, sample_overview_summary: dict[str, Any] + ) -> None: + sample_overview_summary["network"].update( + {"countsAvailable": None, "availability": None, "sourceScope": None} + ) + network = OverviewSummary.model_validate(sample_overview_summary).network + assert network is not None + assert network.counts_available is None + assert network.availability is None + assert network.source_scope is None + + sample_overview_summary["network"].update( + { + "availability": {"state": "future_state", "reason": None, "coverage": "future_cov"}, + "sourceScope": "future_scope", + } + ) + network = OverviewSummary.model_validate(sample_overview_summary).network + assert network is not None + assert network.availability is not None + assert network.availability.state == "future_state" + assert network.source_scope == "future_scope" + def test_overview_summary_round_trip(self, sample_overview_summary: dict[str, Any]) -> None: first = OverviewSummary.model_validate(sample_overview_summary) dumped = first.model_dump(by_alias=True) @@ -435,6 +523,89 @@ def test_network_summary(self, sample_network_summary: dict[str, Any]) -> None: assert s.dns_count == 15 assert s.pcap_available is True + def test_network_summary_v2_unavailable(self, sample_network_summary: dict[str, Any]) -> None: + payload = { + **sample_network_summary, + "pcapAvailable": False, + "countsAvailable": False, + "availability": { + "state": "unavailable", + "reason": "capture_unavailable", + "coverage": "none", + }, + "sourceScope": "none", + } + s = NetworkSummary.model_validate(payload) + assert s.counts_available is False + assert s.availability == NetworkAvailability( + state="unavailable", reason="capture_unavailable", coverage="none" + ) + assert s.source_scope == "none" + assert s.pcap_available is False + + def test_network_summary_legacy_has_no_availability( + self, sample_network_summary: dict[str, Any] + ) -> None: + s = NetworkSummary.model_validate(sample_network_summary) + assert s.counts_available is None + assert s.availability is None + assert s.source_scope is None + assert s.dns_count == 15 + assert s.pcap_available is True + + def test_network_summary_legacy_explicit_nulls( + self, sample_network_summary: dict[str, Any] + ) -> None: + payload = { + **sample_network_summary, + "countsAvailable": True, + "availability": None, + "sourceScope": None, + } + s = NetworkSummary.model_validate(payload) + assert s.counts_available is True + assert s.availability is None + + def test_unknown_enum_values_parse( + self, + sample_network_summary: dict[str, Any], + sample_dns_query: dict[str, Any], + ) -> None: + s = NetworkSummary.model_validate( + { + **sample_network_summary, + "availability": {"state": "future_state", "reason": None, "coverage": "future_cov"}, + "sourceScope": "future_scope", + } + ) + assert s.availability is not None + assert s.availability.state == "future_state" + assert s.availability.coverage == "future_cov" + assert s.source_scope == "future_scope" + assert DnsQuery.model_validate({**sample_dns_query, "sourceScope": "x"}).source_scope == "x" + + def test_network_availability_null_reason(self) -> None: + a = NetworkAvailability.model_validate( + {"state": "observed", "reason": None, "coverage": "complete"} + ) + assert a.reason is None + + def test_list_rows_source_scope( + self, + sample_dns_query: dict[str, Any], + sample_http_request: dict[str, Any], + sample_network_threat: dict[str, Any], + ) -> None: + assert DnsQuery.model_validate(sample_dns_query).source_scope is None + dns = DnsQuery.model_validate({**sample_dns_query, "sourceScope": "sample_attributed"}) + assert dns.source_scope == "sample_attributed" + http = HttpRequest.model_validate( + {**sample_http_request, "sourceScope": "submission_ip_scoped"} + ) + assert http.source_scope == "submission_ip_scoped" + threat = NetworkThreat.model_validate({**sample_network_threat, "sourceScope": "none"}) + assert threat.source_scope == "none" + def test_network_summary_round_trip(self, sample_network_summary: dict[str, Any]) -> None: first = NetworkSummary.model_validate(sample_network_summary) dumped = first.model_dump(by_alias=True)