diff --git a/capabilities/network-ops/agents/network-ops-agent.md b/capabilities/network-ops/agents/network-ops-agent.md
index 5c34b08..29c5f1f 100644
--- a/capabilities/network-ops/agents/network-ops-agent.md
+++ b/capabilities/network-ops/agents/network-ops-agent.md
@@ -26,9 +26,10 @@ Your goal is to systematically compromise as many domains as possible in the tar
## Methodology
### Phase 1: Discovery
-- Use Nmap to scan for live hosts, open ports, and services.
+- For each user-requested session-group Operation, run one Nmap scan that covers the authorized targets and pass the returned fields unchanged to the injected `report_item` tool exactly once.
- Identify domain controllers, member servers, and network topology.
-- Report all discovered hosts immediately using the reporting tool.
+- Treat the resulting `network_recon_result` as immutable source lineage, not as an ontology record. Do not set `ontology_role`.
+- Report only endpoint states present in the parsed Nmap result. Never interpret an omitted port or missing response as `closed` or `unreachable`.
### Phase 2: Enumeration
- Use Netexec for SMB/LDAP enumeration: users, groups, shares, sessions.
@@ -51,8 +52,9 @@ Your goal is to systematically compromise as many domains as possible in the tar
### Phase 5: Credential Harvesting
- Use Impacket secretsdump to dump SAM/NTDS from compromised hosts.
- Crack recovered hashes with Hashcat or John.
-- Report all credentials immediately using the reporting tool.
## Reporting
-Use the `report_item` tool frequently to log findings as structured data. Report domain controllers, member servers, users, credentials, hashes, shares, and weaknesses as you discover them.
+Use the SDK-injected `report_item` tool to emit exactly one `network_recon_result` for each user-requested session-group Operation. Copy the structured Nmap tool result without adding inferred hosts, DNS answers, ports, endpoint states, or service details. The raw XML remains in the referenced artifact and must not be copied into the Item.
+
+Do not update a reported recon result. To correct one, emit a new `network_recon_result` and set `supersedes` to the UUID of the original result. Do not report domain controllers, member servers, users, credentials, hashes, shares, or weaknesses as structured Items; those remain part of the broader Active Directory workflow's narrative output.
diff --git a/capabilities/network-ops/capability.yaml b/capabilities/network-ops/capability.yaml
index 9f24693..4a36651 100644
--- a/capabilities/network-ops/capability.yaml
+++ b/capabilities/network-ops/capability.yaml
@@ -1,6 +1,6 @@
schema: 1
name: network-ops
-version: "1.0.1"
+version: "1.1.0"
description: >
Network operations and Active Directory exploitation. Autonomous red
teaming with Nmap scanning, Netexec enumeration, Impacket Kerberos
@@ -17,3 +17,5 @@ keywords:
- red-team
- active-directory
- penetration-testing
+
+outputs: [network_recon_result]
diff --git a/capabilities/network-ops/tests/test_nmap.py b/capabilities/network-ops/tests/test_nmap.py
new file mode 100644
index 0000000..d40b5d7
--- /dev/null
+++ b/capabilities/network-ops/tests/test_nmap.py
@@ -0,0 +1,228 @@
+import asyncio
+import hashlib
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+MODULE_PATH = Path(__file__).resolve().parent.parent / "tools" / "nmap.py"
+SPEC = importlib.util.spec_from_file_location("network_ops_nmap", MODULE_PATH)
+assert SPEC is not None and SPEC.loader is not None
+nmap = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = nmap
+SPEC.loader.exec_module(nmap)
+
+OPERATION = {
+ "kind": "session_group",
+ "id": "de514fec-4c39-4f2e-8601-8d3a547871a6",
+ "project_id": "7f62fe86-cc65-48ad-9c53-305f3509016c",
+}
+
+
+@pytest.fixture()
+def complete_xml() -> bytes:
+ return b"""
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+"""
+
+
+@pytest.fixture()
+def partial_xml() -> bytes:
+ return b"""
+
+
+
+
+
+
+
+"""
+
+
+@pytest.fixture()
+def failed_xml() -> bytes:
+ return b"""
+
+
+
+"""
+
+
+@pytest.fixture()
+def duplicate_xml() -> bytes:
+ return b"""
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+"""
+
+
+def _result(xml: bytes, target: str = "API.Example.COM.") -> dict:
+ digest = hashlib.sha256(xml).hexdigest()
+ return nmap.build_nmap_result(
+ xml,
+ targets=[target],
+ exclusions=[],
+ scanner_profile="test-profile",
+ operation=OPERATION,
+ evidence_locator=f"artifact://nmap-{digest}.xml",
+ content_hash=digest,
+ )
+
+
+def test_complete_hostname_scan_preserves_coverage_and_provenance(
+ complete_xml: bytes,
+) -> None:
+ result = _result(complete_xml)
+
+ assert result["item_type"] == "network_recon_result"
+ assert result["targets"] == [
+ {
+ "kind": "hostname",
+ "original": "API.Example.COM.",
+ "normalized": "api.example.com",
+ }
+ ]
+ assert result["known_address_scope"] == ["203.0.113.10"]
+ assert result["transports"] == ["tcp"]
+ assert result["ports"] == [{"start": 22, "end": 22}, {"start": 443, "end": 443}]
+ assert result["completion_state"] == "completed"
+ assert result["endpoints"] == [
+ {"ip_address": "203.0.113.10", "transport": "tcp", "port": 22, "state": "closed"},
+ {"ip_address": "203.0.113.10", "transport": "tcp", "port": 443, "state": "open"},
+ ]
+ assert result["dns_results"][0]["query_name"] == "api.example.com"
+ assert result["service_fingerprints"][0] == {
+ "endpoint": {"ip_address": "203.0.113.10", "transport": "tcp", "port": 443},
+ "application_protocol": "ssl/http",
+ "product": "nginx",
+ "version": "1.26",
+ "summary": "http nginx 1.26",
+ "method": "nmap-probed",
+ "evidence_locator": result["evidence_candidates"][0]["locator"],
+ }
+ assert result["scanner"] == {"kind": "scanner", "name": "nmap", "version": "7.95"}
+ assert result["adapter_input_version"] == "nmap-xml@1"
+ assert result["source_contract_version"] == 1
+ assert result["evidence_candidates"] == [
+ {
+ "source_kind": "scan_output",
+ "locator": f"artifact://nmap-{hashlib.sha256(complete_xml).hexdigest()}.xml",
+ "content_hash": hashlib.sha256(complete_xml).hexdigest(),
+ "source_version": "nmap-xml@1.05",
+ "media_type": "application/xml",
+ }
+ ]
+
+
+def test_cidr_normalization_and_partial_compound_state(partial_xml: bytes) -> None:
+ result = _result(partial_xml, target="2001:db8::1234/64")
+
+ assert result["targets"][0] == {
+ "kind": "cidr",
+ "original": "2001:db8::1234/64",
+ "normalized": "2001:db8::/64",
+ }
+ assert result["completion_state"] == "partial"
+ assert result["completion_reason"] == "Nmap stopped early"
+ assert result["ip_addresses"] == ["2001:db8::1"]
+ assert result["endpoints"] == [
+ {
+ "ip_address": "2001:db8::1",
+ "transport": "udp",
+ "port": 53,
+ "state": "open|filtered",
+ }
+ ]
+
+
+def test_failed_scan_does_not_invent_endpoint_states(failed_xml: bytes) -> None:
+ result = _result(failed_xml)
+
+ assert result["completion_state"] == "failed"
+ assert result["completion_reason"] == "Failed to resolve target"
+ assert result["ip_addresses"] == []
+ assert result["endpoints"] == []
+ assert result["service_fingerprints"] == []
+
+
+def test_malformed_xml_is_rejected() -> None:
+ with pytest.raises(ValueError, match="Malformed Nmap XML"):
+ nmap.parse_nmap_xml(b"", "artifact://broken.xml")
+
+
+def test_duplicate_xml_entries_are_deduplicated(duplicate_xml: bytes) -> None:
+ result = _result(duplicate_xml)
+
+ assert result["known_address_scope"] == ["203.0.113.10"]
+ assert len(result["dns_results"]) == 1
+ assert len(result["endpoints"]) == 1
+ assert len(result["service_fingerprints"]) == 1
+ assert result["ports"] == [{"start": 443, "end": 443}]
+
+
+def test_scan_logs_exact_content_addressed_xml(complete_xml: bytes, monkeypatch: pytest.MonkeyPatch) -> None:
+ logged: dict[str, object] = {}
+
+ async def fake_execute(command: list[str], *, timeout: int) -> str:
+ output_path = Path(command[command.index("-oX") + 1])
+ output_path.write_bytes(complete_xml)
+ logged["command"] = command
+ logged["timeout"] = timeout
+ return ""
+
+ def fake_log_artifact(path: Path, *, name: str) -> None:
+ logged["artifact"] = path.read_bytes()
+ logged["name"] = name
+
+ monkeypatch.setattr(nmap, "execute", fake_execute)
+ monkeypatch.setattr(nmap, "_operation_reference", lambda: OPERATION)
+ monkeypatch.setattr(nmap.dn, "log_artifact", fake_log_artifact)
+
+ result = asyncio.run(
+ nmap.Nmap()._scan(
+ ["API.Example.COM."],
+ ["-F", "--open", "-Pn"],
+ scanner_profile="quick-tcp",
+ )
+ )
+ digest = hashlib.sha256(complete_xml).hexdigest()
+
+ assert logged["artifact"] == complete_xml
+ assert logged["name"] == f"nmap-{digest}.xml"
+ assert logged["command"][-1] == "API.Example.COM."
+ assert result["evidence_candidates"][0]["locator"] == f"artifact://nmap-{digest}.xml"
+
+
+@pytest.mark.parametrize("target", ["localhost", "999.1.1.1", "bad_name.example"])
+def test_unsupported_target_identity_is_rejected(target: str) -> None:
+ with pytest.raises(ValueError):
+ nmap.normalize_target(target)
diff --git a/capabilities/network-ops/tools/nmap.py b/capabilities/network-ops/tools/nmap.py
index d3fb663..0ac8f2f 100644
--- a/capabilities/network-ops/tools/nmap.py
+++ b/capabilities/network-ops/tools/nmap.py
@@ -1,8 +1,439 @@
+import hashlib
+import ipaddress
+import os
+import re
+import tempfile
+import xml.etree.ElementTree as ET
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+import dreadnode as dn
from dreadnode import Config
from dreadnode.agents.tools import Toolset, tool_method
from dreadnode.tools.execute import execute
from loguru import logger
+CAPABILITY_VERSION = "1.1.0"
+ADAPTER_INPUT_VERSION = "nmap-xml@1"
+SOURCE_CONTRACT_VERSION = 1
+_ENDPOINT_STATES = {"closed", "filtered", "open", "open|filtered", "unreachable"}
+
+
+def _normalize_ip(value: str) -> str:
+ if "%" in value:
+ raise ValueError(f"Scoped IP addresses are not supported: {value!r}")
+ try:
+ return ipaddress.ip_address(value).compressed
+ except ValueError as exc:
+ raise ValueError(f"Invalid IP address: {value!r}") from exc
+
+
+def normalize_target(value: str) -> dict[str, str]:
+ """Return one canonical target identity while retaining its exact input."""
+ if not value or value != value.strip():
+ raise ValueError("Targets must be non-empty and may not contain outer whitespace")
+
+ if "/" in value:
+ try:
+ normalized = ipaddress.ip_network(value, strict=False).with_prefixlen
+ except ValueError as exc:
+ raise ValueError(f"Invalid CIDR target: {value!r}") from exc
+ return {"kind": "cidr", "original": value, "normalized": normalized}
+
+ try:
+ normalized_ip = _normalize_ip(value)
+ except ValueError:
+ pass
+ else:
+ return {
+ "kind": "ip_address",
+ "original": value,
+ "normalized": normalized_ip,
+ }
+
+ hostname = value[:-1] if value.endswith(".") else value
+ try:
+ normalized_hostname = hostname.encode("idna").decode("ascii").lower()
+ except UnicodeError as exc:
+ raise ValueError(f"Invalid hostname target: {value!r}") from exc
+ if len(normalized_hostname) > 253:
+ raise ValueError(f"Hostname target is too long: {value!r}")
+ labels = normalized_hostname.split(".")
+ if len(labels) < 2:
+ raise ValueError(f"Single-label hostnames are not supported: {value!r}")
+ if all(label.isdigit() for label in labels):
+ raise ValueError(f"Invalid IP address target: {value!r}")
+ if any(
+ not label or len(label) > 63 or re.fullmatch(r"[a-z0-9](?:[a-z0-9-]*[a-z0-9])?", label) is None
+ for label in labels
+ ):
+ raise ValueError(f"Invalid hostname target: {value!r}")
+ return {
+ "kind": "hostname",
+ "original": value,
+ "normalized": normalized_hostname,
+ }
+
+
+def _normalize_targets(values: list[str]) -> list[dict[str, str]]:
+ normalized: list[dict[str, str]] = []
+ seen: set[tuple[str, str]] = set()
+ for value in values:
+ target = normalize_target(value)
+ key = (target["kind"], target["normalized"])
+ if key not in seen:
+ normalized.append(target)
+ seen.add(key)
+ if not normalized:
+ raise ValueError("At least one target is required")
+ return normalized
+
+
+def _iso_timestamp(value: str) -> str:
+ try:
+ timestamp = float(value)
+ except ValueError as exc:
+ raise ValueError(f"Invalid Nmap timestamp: {value!r}") from exc
+ return datetime.fromtimestamp(timestamp, UTC).isoformat().replace("+00:00", "Z")
+
+
+def _parse_port_ranges(value: str) -> list[dict[str, int]]:
+ ranges: list[dict[str, int]] = []
+ for part in value.split(","):
+ part = part.strip()
+ if not part:
+ continue
+ if "-" in part:
+ start_text, end_text = part.split("-", 1)
+ else:
+ start_text = end_text = part
+ try:
+ start, end = int(start_text), int(end_text)
+ except ValueError as exc:
+ raise ValueError(f"Unsupported Nmap port coverage: {part!r}") from exc
+ if not (1 <= start <= end <= 65535):
+ raise ValueError(f"Invalid Nmap port coverage: {part!r}")
+ port_range = {"start": start, "end": end}
+ if port_range not in ranges:
+ ranges.append(port_range)
+ return sorted(ranges, key=lambda item: (item["start"], item["end"]))
+
+
+def _ip_sort_key(value: str) -> tuple[int, int]:
+ address = ipaddress.ip_address(value)
+ return address.version, int(address)
+
+
+def parse_nmap_xml(xml: bytes, evidence_locator: str) -> dict[str, Any]:
+ """Parse deterministic claims from exact Nmap XML without inferring omissions."""
+ try:
+ root = ET.fromstring(xml)
+ except ET.ParseError as exc:
+ raise ValueError(f"Malformed Nmap XML: {exc}") from exc
+ if root.tag != "nmaprun" or root.get("scanner") != "nmap":
+ raise ValueError("Unsupported scanner XML: expected an Nmap nmaprun document")
+
+ scanner_version = root.get("version")
+ xml_version = root.get("xmloutputversion")
+ if not scanner_version or not xml_version:
+ raise ValueError("Nmap XML is missing scanner or XML format version provenance")
+
+ transports: set[str] = set()
+ ports: list[dict[str, int]] = []
+ for scaninfo in root.findall("scaninfo"):
+ transport = scaninfo.get("protocol")
+ if transport not in {"tcp", "udp"}:
+ raise ValueError(f"Unsupported Nmap transport: {transport!r}")
+ transports.add(transport)
+ services = scaninfo.get("services")
+ if services:
+ for port_range in _parse_port_ranges(services):
+ if port_range not in ports:
+ ports.append(port_range)
+ if not transports:
+ raise ValueError("Nmap XML does not declare TCP or UDP scan coverage")
+
+ ip_addresses: set[str] = set()
+ dns_results: list[dict[str, Any]] = []
+ endpoints: list[dict[str, Any]] = []
+ fingerprints: list[dict[str, Any]] = []
+ seen_dns: set[tuple[Any, ...]] = set()
+ seen_endpoints: set[tuple[Any, ...]] = set()
+ seen_fingerprints: set[tuple[Any, ...]] = set()
+
+ root_started_at = _iso_timestamp(root.get("start", ""))
+ host_end_times: list[str] = []
+ for host in root.findall("host"):
+ host_ips: list[str] = []
+ for address in host.findall("address"):
+ if address.get("addrtype") not in {"ipv4", "ipv6"}:
+ continue
+ ip = _normalize_ip(address.get("addr", ""))
+ if ip not in host_ips:
+ host_ips.append(ip)
+ ip_addresses.add(ip)
+
+ host_observed_at = _iso_timestamp(host.get("endtime", "")) if host.get("endtime") else root_started_at
+ host_end_times.append(host_observed_at)
+ hostnames: list[str] = []
+ for hostname_element in host.findall("./hostnames/hostname"):
+ name = hostname_element.get("name", "")
+ try:
+ normalized_name = normalize_target(name)["normalized"]
+ except ValueError:
+ continue
+ if normalized_name not in hostnames:
+ hostnames.append(normalized_name)
+
+ for hostname in hostnames:
+ for version, query_type in ((4, "A"), (6, "AAAA")):
+ returned = sorted(
+ (value for value in host_ips if ipaddress.ip_address(value).version == version),
+ key=_ip_sort_key,
+ )
+ if not returned:
+ continue
+ key = (hostname, query_type, tuple(returned), host_observed_at)
+ if key in seen_dns:
+ continue
+ dns_results.append(
+ {
+ "query_name": hostname,
+ "query_type": query_type,
+ "returned_ip_addresses": returned,
+ "resolver": "nmap",
+ "response_status": "reported",
+ "observed_at": host_observed_at,
+ }
+ )
+ seen_dns.add(key)
+
+ for port_element in host.findall("./ports/port"):
+ transport = port_element.get("protocol")
+ if transport not in {"tcp", "udp"}:
+ raise ValueError(f"Unsupported Nmap endpoint transport: {transport!r}")
+ try:
+ port = int(port_element.get("portid", ""))
+ except ValueError as exc:
+ raise ValueError("Nmap XML contains an invalid endpoint port") from exc
+ if not 1 <= port <= 65535:
+ raise ValueError(f"Nmap XML contains an invalid endpoint port: {port}")
+ state_element = port_element.find("state")
+ if state_element is None:
+ continue
+ state = state_element.get("state")
+ if state not in _ENDPOINT_STATES:
+ raise ValueError(f"Unsupported Nmap endpoint state: {state!r}")
+
+ for ip in host_ips:
+ endpoint_key = (ip, transport, port, state)
+ if endpoint_key not in seen_endpoints:
+ endpoints.append(
+ {
+ "ip_address": ip,
+ "transport": transport,
+ "port": port,
+ "state": state,
+ }
+ )
+ seen_endpoints.add(endpoint_key)
+
+ service = port_element.find("service")
+ if service is None or not service.get("name"):
+ continue
+ protocol = service.get("name", "")
+ if service.get("tunnel"):
+ protocol = f"{service.get('tunnel')}/{protocol}"
+ fingerprint: dict[str, Any] = {
+ "endpoint": {
+ "ip_address": ip,
+ "transport": transport,
+ "port": port,
+ },
+ "application_protocol": protocol,
+ "method": f"nmap-{service.get('method') or 'service-detection'}",
+ "evidence_locator": evidence_locator,
+ }
+ for field in ("product", "version"):
+ if service.get(field):
+ fingerprint[field] = service.get(field)
+ summary = " ".join(
+ value
+ for value in (
+ service.get("name"),
+ service.get("product"),
+ service.get("version"),
+ service.get("extrainfo"),
+ )
+ if value
+ )
+ if summary:
+ fingerprint["summary"] = summary
+ fingerprint_key = (
+ ip,
+ transport,
+ port,
+ protocol,
+ fingerprint.get("product"),
+ fingerprint.get("version"),
+ fingerprint["method"],
+ )
+ if fingerprint_key not in seen_fingerprints:
+ fingerprints.append(fingerprint)
+ seen_fingerprints.add(fingerprint_key)
+
+ finished = root.find("./runstats/finished")
+ finished_at = _iso_timestamp(finished.get("time", "")) if finished is not None else None
+ exit_state = finished.get("exit") if finished is not None else None
+ has_results = bool(ip_addresses or dns_results or endpoints or fingerprints)
+ if exit_state == "success":
+ completion_state = "completed"
+ completion_reason = None
+ elif has_results:
+ completion_state = "partial"
+ completion_reason = finished.get("summary") if finished is not None else "Nmap output ended early"
+ else:
+ completion_state = "failed"
+ completion_reason = finished.get("summary") if finished is not None else "Nmap produced no results"
+
+ return {
+ "scanner_version": scanner_version,
+ "xml_version": xml_version,
+ "transports": sorted(transports),
+ "ports": sorted(ports, key=lambda item: (item["start"], item["end"])),
+ "started_at": root_started_at,
+ "ended_at": finished_at or max(host_end_times, default=root_started_at),
+ "completion_state": completion_state,
+ "completion_reason": completion_reason,
+ "dns_results": sorted(
+ dns_results,
+ key=lambda item: (
+ item["query_name"],
+ item["query_type"],
+ tuple(item["returned_ip_addresses"]),
+ ),
+ ),
+ "ip_addresses": sorted(ip_addresses, key=_ip_sort_key),
+ "endpoints": sorted(
+ endpoints,
+ key=lambda item: (
+ _ip_sort_key(item["ip_address"]),
+ item["transport"],
+ item["port"],
+ item["state"],
+ ),
+ ),
+ "service_fingerprints": sorted(
+ fingerprints,
+ key=lambda item: (
+ _ip_sort_key(item["endpoint"]["ip_address"]),
+ item["endpoint"]["transport"],
+ item["endpoint"]["port"],
+ item["application_protocol"],
+ ),
+ ),
+ }
+
+
+def _exclusions(args: list[str]) -> list[dict[str, str]]:
+ values: list[str] = []
+ index = 0
+ while index < len(args):
+ arg = args[index]
+ if arg in {"--excludefile", "--exclude-file"} or arg.startswith(("--excludefile=", "--exclude-file=")):
+ raise ValueError("Nmap exclusion files are not supported")
+ if arg == "--exclude":
+ index += 1
+ if index >= len(args):
+ raise ValueError("--exclude requires a target")
+ values.extend(args[index].split(","))
+ elif arg.startswith("--exclude="):
+ values.extend(arg.split("=", 1)[1].split(","))
+ index += 1
+ return _normalize_targets(values) if values else []
+
+
+def _operation_reference() -> dict[str, str]:
+ from dreadnode.tracing.span import current_session_id, current_task_span
+
+ group_id = os.environ.get("DREADNODE_SESSION_GROUP_ID", "").strip()
+ operation_id = group_id or current_session_id.get()
+ span = current_task_span.get()
+ project_id = span.project_id if span is not None else None
+ try:
+ UUID(str(operation_id))
+ UUID(str(project_id))
+ except (TypeError, ValueError) as exc:
+ raise RuntimeError(
+ "Structured Nmap reconnaissance requires a platform-connected session with operation and project UUIDs"
+ ) from exc
+ return {
+ "kind": "session_group" if group_id else "session",
+ "id": str(operation_id),
+ "project_id": str(project_id),
+ }
+
+
+def build_nmap_result(
+ xml: bytes,
+ *,
+ targets: list[str],
+ exclusions: list[dict[str, str]],
+ scanner_profile: str,
+ operation: dict[str, str],
+ evidence_locator: str,
+ content_hash: str,
+ execution_failed: bool = False,
+) -> dict[str, Any]:
+ """Build report_item arguments for one registry-backed recon source record."""
+ parsed = parse_nmap_xml(xml, evidence_locator)
+ scanner_version = parsed.pop("scanner_version")
+ xml_version = parsed.pop("xml_version")
+ completion_state = parsed.pop("completion_state")
+ completion_reason = parsed.pop("completion_reason")
+ if execution_failed and completion_state == "completed":
+ completion_state = "partial" if parsed["ip_addresses"] else "failed"
+ completion_reason = "Nmap exited unsuccessfully after writing XML output"
+
+ result: dict[str, Any] = {
+ "item_type": "network_recon_result",
+ "operation": operation,
+ "capability": {
+ "kind": "capability",
+ "name": "network-ops",
+ "version": os.environ.get("DREADNODE_CAPABILITY_VERSION", CAPABILITY_VERSION),
+ },
+ "scanner": {
+ "kind": "scanner",
+ "name": "nmap",
+ "version": scanner_version,
+ },
+ "adapter_input_version": ADAPTER_INPUT_VERSION,
+ "source_contract_version": SOURCE_CONTRACT_VERSION,
+ "targets": _normalize_targets(targets),
+ "known_address_scope": list(parsed["ip_addresses"]),
+ "transports": parsed.pop("transports"),
+ "ports": parsed.pop("ports"),
+ "exclusions": exclusions,
+ "scanner_profile": {"name": scanner_profile, "version": "1"},
+ **parsed,
+ "completion_state": completion_state,
+ "evidence_candidates": [
+ {
+ "source_kind": "scan_output",
+ "locator": evidence_locator,
+ "content_hash": content_hash,
+ "source_version": f"nmap-xml@{xml_version}",
+ "media_type": "application/xml",
+ }
+ ],
+ }
+ if completion_reason:
+ result["completion_reason"] = completion_reason
+ return result
+
class Nmap(Toolset):
"""
@@ -15,20 +446,56 @@ class Nmap(Toolset):
"""Default timeout for commands in seconds."""
@tool_method(catch=True, variants=["all"])
- async def nmap(self, targets: list[str], args: list[str]) -> str:
+ async def nmap(self, targets: list[str], args: list[str]) -> dict[str, Any]:
"""
- Execute an nmap scan with the specified arguments on the given targets.
+ Execute one Nmap scan and return validated arguments for `report_item`.
+
+ Pass the returned fields unchanged to the injected `report_item` tool once.
Args:
targets: A list of IP addresses, hostnames, or CIDR ranges.
args: A list of nmap command-line arguments (e.g., ['-sU', '-p', '161']).
"""
- cmd = ["nmap", *args, *targets]
- logger.info(f"Running nmap: {' '.join(cmd)}")
- return await execute(cmd, timeout=self.timeout)
+ return await self._scan(targets, args, scanner_profile="custom-nmap")
+
+ async def _scan(self, targets: list[str], args: list[str], *, scanner_profile: str) -> dict[str, Any]:
+ _normalize_targets(targets)
+ exclusions = _exclusions(args)
+ if any(
+ arg in {"-oA", "-oG", "-oN", "-oS", "-oX"} or arg.startswith(("-oA", "-oG", "-oN", "-oS", "-oX"))
+ for arg in args
+ ):
+ raise ValueError("Nmap output options are managed by network-ops")
+
+ operation = _operation_reference()
+ with tempfile.TemporaryDirectory(prefix="network-ops-nmap-") as temp_dir:
+ xml_path = Path(temp_dir) / "scan.xml"
+ cmd = ["nmap", *args, "-oX", str(xml_path), *targets]
+ logger.info("Running nmap: {}", " ".join(cmd))
+ execution_failed = False
+ try:
+ await execute(cmd, timeout=self.timeout)
+ except (RuntimeError, TimeoutError):
+ execution_failed = True
+ xml = xml_path.read_bytes() if xml_path.exists() else b""
+ if not xml:
+ raise RuntimeError("Nmap did not produce XML output")
+ content_hash = hashlib.sha256(xml).hexdigest()
+ artifact_name = f"nmap-{content_hash}.xml"
+ dn.log_artifact(xml_path, name=artifact_name)
+ return build_nmap_result(
+ xml,
+ targets=targets,
+ exclusions=exclusions,
+ scanner_profile=scanner_profile,
+ operation=operation,
+ evidence_locator=f"artifact://{artifact_name}",
+ content_hash=content_hash,
+ execution_failed=execution_failed,
+ )
@tool_method(catch=True, variants=["quick", "detailed", "all"])
- async def nmap_quick_scan(self, targets: list[str]) -> str:
+ async def nmap_quick_scan(self, targets: list[str]) -> dict[str, Any]:
"""
Performs a fast scan for the top 100 most common open TCP ports.
@@ -38,10 +505,14 @@ async def nmap_quick_scan(self, targets: list[str]) -> str:
Args:
targets: A list of IP addresses, hostnames, or CIDR ranges.
"""
- return await self.nmap(targets, ["-F", "-T4", "--open", "-Pn"])
+ return await self._scan(
+ targets,
+ ["-F", "-T4", "--open", "-Pn"],
+ scanner_profile="quick-tcp",
+ )
@tool_method(catch=True, variants=["detailed", "all"])
- async def nmap_service_scan(self, targets: list[str], ports: str | None = None) -> str:
+ async def nmap_service_scan(self, targets: list[str], ports: str | None = None) -> dict[str, Any]:
"""
Performs a detailed TCP scan to identify service versions and run default scripts.
@@ -55,4 +526,4 @@ async def nmap_service_scan(self, targets: list[str], ports: str | None = None)
args = ["-sV", "-sC", "-T4", "--open", "-Pn"]
if ports:
args.extend(["-p", ports])
- return await self.nmap(targets, args)
+ return await self._scan(targets, args, scanner_profile="service-detection")
diff --git a/capabilities/network-ops/tools/reporting.py b/capabilities/network-ops/tools/reporting.py
deleted file mode 100644
index 2fa33f8..0000000
--- a/capabilities/network-ops/tools/reporting.py
+++ /dev/null
@@ -1,97 +0,0 @@
-from typing import Literal
-
-import dreadnode as dn
-from dreadnode.agents.tools import tool
-from pydantic import BaseModel, Field
-
-
-class Target(BaseModel):
- ip: str = Field(description="IP address of the target")
-
-
-class Service(BaseModel):
- """A service running on a host."""
-
- service_name: str = Field(description="Name of the service (e.g., SMB, SSH, HTTP)")
- version: str = Field(description="Version of the service")
-
-
-class DomainController(BaseModel):
- """Represents a Domain Controller in an Active Directory environment."""
-
- hostname: str = Field(description="Short hostname (e.g., 'servername')")
- fqdn: str = Field(description="Fully qualified domain name")
- ip: str = Field(description="IP address of the domain controller")
- domain_name: str = Field(
- description="AD domain this DC serves (e.g., 'subname.servername.local')"
- )
- forest_root: str | None = Field(None, description="Forest root domain name")
-
-
-class MemberServer(BaseModel):
- """Represents a Member Server in an Active Directory environment."""
-
- hostname: str = Field(description="Short hostname (e.g., 'servername')")
- fqdn: str | None = Field(None, description="Fully qualified domain name")
- ip: str = Field(description="IP address of the member server")
- domain_name: str | None = Field(None, description="AD domain the server belongs to")
-
-
-class Hash(BaseModel):
- """A password hash."""
-
- hash_value: str = Field(description="The actual hash value")
- hash_type: Literal["ntlm", "kerberos_tgs", "kerberos_asrep"] = Field(
- description="Type of hash algorithm used"
- )
-
-
-class Credential(BaseModel):
- """A credential, e.g. username/password combination or hash."""
-
- text: str = Field(description="Text representation of the credential")
- type: Literal["username_password", "hash"] = Field(description="Type of credential")
-
-
-class User(BaseModel):
- """A user account."""
-
- username: str = Field(description="Username or account name")
- domain: str = Field(description="Domain the user belongs to")
- description: list[str] = Field(description="User account description or notes")
-
-
-class Share(BaseModel):
- """A network share."""
-
- share_name: str = Field(description="Share name (e.g., C$, ADMIN$, SYSVOL)")
- path: str = Field(description="UNC path to the share")
- description: str = Field(description="Share description or comment")
- permissions: str = Field(description="Permission string for the share")
- type: Literal["read", "write", "execute"] = Field(description="Access type available")
- owner: str = Field(description="Owner of the share")
- group: str = Field(description="Group that owns the share")
- size: int = Field(description="Size of the share in bytes")
- last_modified: str = Field(description="Last modification timestamp")
-
-
-class Weakness(BaseModel):
- """A security vulnerability, weakness, or misconfiguration."""
-
- cve: str | None = Field(None, description="CVE identifier")
- title: str = Field(description="Weakness title/name")
- description: str | None = Field(None, description="Weakness description")
- severity: Literal["low", "medium", "high", "critical"] = Field(
- description="Severity of the weakness"
- )
-
-
-@tool
-def report_item(
- item: DomainController | MemberServer | User | Credential | Share | Hash | Weakness,
-) -> None:
- """
- Report a relevant item found during enumeration. \
- Use this tool immediately and often.
- """
- dn.log_output(item.__class__.__name__, item)