From af375cae8cfb5dabd0690ea39d85d8c65ed3ef5c Mon Sep 17 00:00:00 2001 From: batuhanisildak-malwation Date: Fri, 2 Oct 2026 07:52:30 +0000 Subject: [PATCH 1/4] feat: expose indicator score scale and network availability --- CHANGELOG.md | 6 +++ docs/TYPES.md | 17 ++++--- src/threatzone/__init__.py | 2 + src/threatzone/types/__init__.py | 8 +++ src/threatzone/types/indicators.py | 1 + src/threatzone/types/network.py | 26 ++++++++++ tests/test_types.py | 78 ++++++++++++++++++++++++++++++ 7 files changed, 132 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c0e2cf..5adeb50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `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`. All are `None` for legacy payloads. + 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..8159cd2 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`, `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,6 +74,7 @@ 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 `Literal[10, 100] | 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. - `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`, @@ -107,18 +108,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}`. The last three are `None` on legacy reports. | `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. +- 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..6931635 100644 --- a/src/threatzone/types/indicators.py +++ b/src/threatzone/types/indicators.py @@ -36,6 +36,7 @@ class Indicator(BaseModel): category: list[str] level: IndicatorLevel score: int + score_maximum: Literal[10, 100] | None = Field(default=None, alias="scoreMaximum") pids: list[int] attack_codes: list[str] = Field(alias="attackCodes") event_ids: list[int] = Field(alias="eventIds") diff --git a/src/threatzone/types/network.py b/src/threatzone/types/network.py index 74c6449..16eb4c1 100644 --- a/src/threatzone/types/network.py +++ b/src/threatzone/types/network.py @@ -55,6 +55,25 @@ ThreatAppProto = Literal["HTTP", "TLS", "DNS"] ThreatSeverity = Literal["high", "low"] +NetworkAvailabilityState = Literal["disabled", "unavailable", "verified_zero", "observed"] +NetworkCoverage = Literal["none", "partial", "complete"] +NetworkSourceScope = Literal[ + "sample_attributed", + "submission_ip_scoped", + "submission_unattributed", + "none", +] + + +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 +86,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 +102,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 +115,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 +142,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 +174,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..b99ba9d 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,18 @@ 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_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_category_is_list(self, sample_indicator: dict[str, Any]) -> None: ind = Indicator.model_validate(sample_indicator) assert ind.category == ["evasion", "persistence"] @@ -435,6 +448,71 @@ 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_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) From 2bf930e0233e3768c6228b3f816d29217cee1b85 Mon Sep 17 00:00:00 2001 From: batuhanisildak-malwation Date: Fri, 2 Oct 2026 07:53:50 +0000 Subject: [PATCH 2/4] fix: accept unknown network availability and score scale values --- CHANGELOG.md | 2 +- docs/TYPES.md | 6 +++--- src/threatzone/types/indicators.py | 2 +- src/threatzone/types/network.py | 19 +++++++++++-------- tests/test_types.py | 22 ++++++++++++++++++++++ 5 files changed, 38 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5adeb50..aa173b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `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`. All are `None` for legacy payloads. + `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. Enum-like fields accept unknown values, so a new server value does not fail the call. 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 diff --git a/docs/TYPES.md b/docs/TYPES.md index 8159cd2..ffe01f3 100644 --- a/docs/TYPES.md +++ b/docs/TYPES.md @@ -74,7 +74,7 @@ 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 `Literal[10, 100] | 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.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. - `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`, @@ -108,7 +108,7 @@ 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, counts_available, availability, source_scope}`. The last three are `None` on legacy reports. | `GET /submissions/:uuid/network/summary` | +| `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` | @@ -122,7 +122,7 @@ report. Every endpoint is guarded by `ReportStatusGuard('dynamic')`; read the co - `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. +- `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. 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!). diff --git a/src/threatzone/types/indicators.py b/src/threatzone/types/indicators.py index 6931635..d574a89 100644 --- a/src/threatzone/types/indicators.py +++ b/src/threatzone/types/indicators.py @@ -36,7 +36,7 @@ class Indicator(BaseModel): category: list[str] level: IndicatorLevel score: int - score_maximum: Literal[10, 100] | None = Field(default=None, alias="scoreMaximum") + 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") diff --git a/src/threatzone/types/network.py b/src/threatzone/types/network.py index 16eb4c1..a8d2ac1 100644 --- a/src/threatzone/types/network.py +++ b/src/threatzone/types/network.py @@ -55,14 +55,17 @@ ThreatAppProto = Literal["HTTP", "TLS", "DNS"] ThreatSeverity = Literal["high", "low"] -NetworkAvailabilityState = Literal["disabled", "unavailable", "verified_zero", "observed"] -NetworkCoverage = Literal["none", "partial", "complete"] -NetworkSourceScope = Literal[ - "sample_attributed", - "submission_ip_scoped", - "submission_unattributed", - "none", -] +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): diff --git a/tests/test_types.py b/tests/test_types.py index b99ba9d..3a97b8b 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -293,6 +293,10 @@ def test_indicator_score_maximum(self, sample_indicator: dict[str, Any]) -> None 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 @@ -491,6 +495,24 @@ def test_network_summary_legacy_explicit_nulls( 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"} From b184adb990b053ea1f318f34982f4286498e248b Mon Sep 17 00:00:00 2001 From: batuhanisildak-malwation Date: Fri, 2 Oct 2026 12:09:23 +0000 Subject: [PATCH 3/4] feat: expose network availability on the overview summary --- CHANGELOG.md | 3 +- docs/TYPES.md | 6 ++- src/threatzone/types/indicators.py | 5 +++ tests/test_types.py | 60 ++++++++++++++++++++++++++++++ 4 files changed, 71 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index aa173b8..273058b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `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. Enum-like fields accept unknown values, so a new server value does not fail the call. + `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 diff --git a/docs/TYPES.md b/docs/TYPES.md index ffe01f3..dc4b59d 100644 --- a/docs/TYPES.md +++ b/docs/TYPES.md @@ -80,7 +80,9 @@ in this group is envelope-wrapped (`items`, `total`) for server-side pagination. - `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` @@ -122,7 +124,7 @@ report. Every endpoint is guarded by `ReportStatusGuard('dynamic')`; read the co - `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. Unknown values from a newer server are kept as plain strings. +- `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!). diff --git a/src/threatzone/types/indicators.py b/src/threatzone/types/indicators.py index d574a89..943bd28 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"] @@ -220,6 +222,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/tests/test_types.py b/tests/test_types.py index 3a97b8b..8a58368 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -433,6 +433,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) From 144aa3ac6572c8c6129f129fcbfcf1cb0399d9a2 Mon Sep 17 00:00:00 2001 From: batuhanisildak-malwation Date: Sat, 3 Oct 2026 14:25:18 +0000 Subject: [PATCH 4/4] feat: expose the indicator analyst summary --- CHANGELOG.md | 1 + docs/TYPES.md | 3 ++- src/threatzone/types/indicators.py | 1 + tests/test_types.py | 11 +++++++++++ 4 files changed, 15 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 273058b..c04bb98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ 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`, diff --git a/docs/TYPES.md b/docs/TYPES.md index dc4b59d..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`, `score_maximum`, `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` | @@ -75,6 +75,7 @@ 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`, diff --git a/src/threatzone/types/indicators.py b/src/threatzone/types/indicators.py index 943bd28..b7a18d7 100644 --- a/src/threatzone/types/indicators.py +++ b/src/threatzone/types/indicators.py @@ -35,6 +35,7 @@ class Indicator(BaseModel): id: str name: str description: str + summary: str | None = None category: list[str] level: IndicatorLevel score: int diff --git a/tests/test_types.py b/tests/test_types.py index 8a58368..16bac81 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -302,6 +302,17 @@ def test_indicator_without_score_maximum(self, sample_indicator: dict[str, Any]) 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"]