From dc9ffeb84285298f42da488ee9c2a5c90a31b697 Mon Sep 17 00:00:00 2001 From: netanelcyber <87965762+netanelcyber@users.noreply.github.com> Date: Thu, 8 Oct 2026 13:53:59 +0300 Subject: [PATCH] Add bounded PPTP control check and prepare 2.14.213.78.21.9 Add an explicit TCP/1723 protocol check with a pinned endpoint, a shared deadline, reply evidence, cleanup and domain-linked ORM/report results. Leave authentication, GRE and VPN tunnel connectivity untested. Validation: 755 tests and 271 subtests, strict package checks, Git/archive equality and isolated installed-wheel smoke checks. --- .github/workflows/publish.yml | 27 ++- CHANGELOG.md | 8 + README.md | 24 ++- adpentest/__init__.py | 2 +- adpentest/core.py | 2 +- adpentest/diagnostics.py | 108 +++++++++++- adpentest/pptp_check.py | 160 +++++++++++++++++ docs/DIAGNOSTIC_TOOLS.md | 43 ++++- docs/RELEASE_2.14.213.78.21.9.md | 18 ++ docs/SMBV1_DATA.md | 6 +- pyproject.toml | 4 +- tests/test_diagnostic_information.py | 2 +- tests/test_diagnostic_pptp.py | 255 +++++++++++++++++++++++++++ tests/test_diagnostic_storage.py | 2 +- tests/test_diagnostic_tools.py | 10 +- tests/test_release_version.py | 4 +- tools/add_diagnostic_tools.py | 3 + 17 files changed, 640 insertions(+), 38 deletions(-) create mode 100644 adpentest/pptp_check.py create mode 100644 docs/RELEASE_2.14.213.78.21.9.md create mode 100644 tests/test_diagnostic_pptp.py diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 09f661e..bfdbab1 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -3,7 +3,7 @@ name: Publish to PyPI on: push: branches: - - 'release/pypi-2.14.213.78.21.8' + - 'release/pypi-2.14.213.78.21.9' release: types: [published] workflow_dispatch: @@ -65,7 +65,7 @@ jobs: run: python -m build - name: Check distribution - run: twine check dist/* + run: twine check --strict dist/* - name: Check source distribution contents run: | @@ -75,7 +75,8 @@ jobs: with tarfile.open(next(pathlib.Path("dist").glob("*.tar.gz")), "r:gz") as source: names = {name.split("/", 1)[1] for name in source.getnames() if "/" in name} assert {"requirements.txt", "d2.aspx", "tools/add_diagnostic_tools.py", - "tests/conftest.py", "tests/test_domain_orm.py"}.issubset(names) + "tests/conftest.py", "tests/test_domain_orm.py", "tests/test_diagnostic_pptp.py", + "adpentest/pptp_check.py"}.issubset(names) PY - name: Check installed wheel @@ -93,7 +94,14 @@ jobs: import importlib.metadata requirements = importlib.metadata.metadata("adpentest").get_all("Requires-Dist") assert any(item.startswith("impacket") and ";" not in item for item in requirements) - assert adpentest.__version__ == "2.14.213.78.21.8" + assert adpentest.__version__ == "2.14.213.78.21.9" + from adpentest.pptp_check import parse_start_reply, start_request + import struct + reply = struct.pack("!HHIHHHBBIIHH64s64s", 156, 1, 0x1A2B3C4D, 2, 0, + 0x0100, 1, 0, 3, 3, 0, 42, b"fixture-vpn", b"Fixture Vendor") + assert len(start_request()) == 156 + assert parse_start_reply(reply)["control_status"] == "accepted" + assert parse_start_reply(reply)["gre"]["status"] == "not-tested" from adpentest.smbv1_data import SMBv1ExportParser, analyze_smbv1_data inventory = analyze_smbv1_data( [{"host": "fileserver.lab.example", "smbv1": True, @@ -117,6 +125,9 @@ jobs: assert session.scalar(select(SMBv1Host)).users[0].username == "alice" from adpentest.diagnostics import build_diagnostic_command args = build_diagnostic_command("nmap_smbv1", "192.0.2.10") + pptp_command = build_diagnostic_command("pptp_control", "vpn.example", resolved_address="192.0.2.10") + assert pathlib.Path(pptp_command[2]).is_file() + assert pptp_command[pptp_command.index("--target") + 1] == "192.0.2.10" assert "smb-protocols" in args and "445" in args and "-vv" in args ports = build_diagnostic_command("nmap_tcp_ports", "192.0.2.10", ports="1-65535", timeout=600) assert "1-65535" in ports and "--script" not in ports and "-vv" in ports @@ -150,15 +161,17 @@ jobs: [sys.executable, "-I", "-c", "import sys; sys.path.insert(0, " + repr(str(pathlib.Path("wheel-check").resolve())) + "); import adpentest; raise SystemExit(adpentest.main())", "--port-map", "--ports", "1-65535", "--port-map-timeout", "600", "--smbv1-check", "--smbv1-access", - "--smb-share", "SHARE", "--smb-user", "alice", "--target", "192.0.2.10", "--mode", "dry-run", + "--smb-share", "SHARE", "--smb-user", "alice", "--pptp-check", "--target", "192.0.2.10", "--mode", "dry-run", "--dns-server=192.0.2.53,2001:db8::53", "--db-path", str(pathlib.Path("wheel-check-history.sqlite").resolve())], check=True, capture_output=True, text=True, timeout=30, ) report = json.loads(result.stdout) - assert len(report["tools"]) == 3 + assert len(report["tools"]) == 4 assert all(result["status"] == "dry-run" for result in report["tools"]) assert all(result["persistence"]["status"] == "saved" for result in report["tools"]) + pptp_result = next(result for result in report["tools"] if result["tool"] == "pptp_control") + assert pptp_result["pptp"]["details"]["gre"]["status"] == "not-tested" PY - name: Publish to TestPyPI @@ -171,7 +184,7 @@ jobs: - name: Publish to PyPI # Manual dispatch must obey the release-branch boundary too. if: >- - github.ref == 'refs/heads/release/pypi-2.14.213.78.21.8' && + github.ref == 'refs/heads/release/pypi-2.14.213.78.21.9' && (github.event_name == 'push' || (github.event_name == 'workflow_dispatch' && github.event.inputs.pypi_environment == 'pypi')) uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/CHANGELOG.md b/CHANGELOG.md index a2225d6..3b88bc4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,14 @@ All notable changes to AdPentestAI-Python are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [2.14.213.78.21.9] - 2026-10-08 + +### Added +- Opt-in `--pptp-check` / `pptp_control` for a bounded standard control exchange on TCP/1723 with one pinned endpoint. +- Protocol reply validation, server metadata, accepted/rejected status and explicit untested authentication/GRE/tunnel coverage. +- Domain-linked ORM and configuration-report evidence for completed, failed, blocked and dry-run PPTP checks. +- Offline packet, socket, deadline, scope, address-binding and persistence tests; no new runtime dependency. + ## [2.14.213.78.21.8] - 2026-10-08 ### Changed diff --git a/README.md b/README.md index 335412b..9d23af2 100644 --- a/README.md +++ b/README.md @@ -1,15 +1,15 @@ # AdPentestAI-Python -Version `2.14.213.78.21.8` is prepared with the domain ORM, DNS, XML persistence, -required SMB dependency and information-report features released in `.7`. -See [release notes](docs/RELEASE_2.14.213.78.21.8.md) and [domain ORM](docs/DOMAIN_ORM.md). +Version `2.14.213.78.21.9` adds an opt-in PPTP control check on TCP/1723 and carries +forward domain ORM, DNS/XML history, required SMB dependency and information reports. +See [release notes](docs/RELEASE_2.14.213.78.21.9.md) and [domain ORM](docs/DOMAIN_ORM.md). The PyPI `2.14.213.78.21.6` files published on 2026-10-06 are the `cb4ce2426765704ddb53d14276e5aacb2cf238fa` snapshot. The features introduced in -`.7` and retained in `.8` are absent from those published `.6` files. See the +`.7` and retained in `.9` are absent from those published `.6` files. See the [verified artifact provenance](docs/PYPI_2.14.213.78.21.6_PROVENANCE.md). -SMBv1 host, user and UNC inventory from existing exports (2.14.213.78.21.8): +SMBv1 host, user and UNC inventory from existing exports (2.14.213.78.21.9): ```console python -m adpentest --smbv1-data existing-report.json @@ -50,6 +50,18 @@ PPTP control port 1723 is included; its GRE prerequisite is reported as untested Results are port states and candidate labels, without service or vulnerability verification. See [VPN port prerequisites and output](docs/DIAGNOSTIC_TOOLS.md#all-tcp-ports-and-vpn-port-states). +For a PPTP control check on one endpoint: + +```console +python -m adpentest --pptp-check --target vpn.example --mode dry-run --domain lab.example --info-report ./pptp-review.json --db-path ./pptp-history.sqlite +``` + +Choose `--mode active --scope-confirmed` for an authorized control exchange. It +uses Python's standard library on TCP/1723, validates the bounded reply, records +the server's result code and metadata, and closes its connection. Domain-linked +ORM history includes failures and dry runs. Authentication, GRE and VPN tunnel +connectivity are reported as untested. See [PPTP checks](docs/DIAGNOSTIC_TOOLS.md#pptp-control-check). + For fixed TLS certificate, SMB and SSH metadata checks with a configuration review report: @@ -89,7 +101,7 @@ per stream and flag truncated output. See [diagnostic profiles](docs/DIAGNOSTIC_ The Python SMB client (`impacket>=0.13,<0.14`) is a required package dependency: ```console -python -m pip install --upgrade --no-cache-dir --index-url https://pypi.org/simple "adpentest==2.14.213.78.21.8" +python -m pip install --upgrade --no-cache-dir --index-url https://pypi.org/simple "adpentest==2.14.213.78.21.9" ``` With Nmap on PATH, run the combined built-in diagnostics on one host you administer: diff --git a/adpentest/__init__.py b/adpentest/__init__.py index c3858d4..3d1a918 100644 --- a/adpentest/__init__.py +++ b/adpentest/__init__.py @@ -1,4 +1,4 @@ -__version__ = "2.14.213.78.21.8" +__version__ = "2.14.213.78.21.9" from .report_generator import generate_html_report, export_html_report diff --git a/adpentest/core.py b/adpentest/core.py index e2cc24b..a4f48ce 100644 --- a/adpentest/core.py +++ b/adpentest/core.py @@ -1189,7 +1189,7 @@ def _ldap_spn_cli_requested(argv): # Source: adpentest/__init__.py # ============================================================================ -__version__ = "2.14.213.78.21.8" +__version__ = "2.14.213.78.21.9" diff --git a/adpentest/diagnostics.py b/adpentest/diagnostics.py index f5be209..fa434fb 100644 --- a/adpentest/diagnostics.py +++ b/adpentest/diagnostics.py @@ -36,6 +36,7 @@ "nmap_smbv1": {"binary": "nmap", "port": 445, "description": "Single-host SMB dialect negotiation, including SMBv1"}, "nmap_tcp_ports": {"binary": "nmap", "port": None, "description": "Selected TCP port states on one host, without service probes"}, "nmap_vpn_udp_ports": {"binary": "nmap", "port": None, "description": "Fixed common VPN UDP port states on one host; requires --ALL-PORTS"}, + "pptp_control": {"binary": "python-stdlib", "port": 1723, "description": "One PPTP control reply on TCP/1723; explicit --pptp-check"}, "smbv1_access": {"binary": "python+impacket", "port": 445, "description": "One existing account and explicit share; SMBv1 root directory listing"}, "nmap_rdp_info": {"binary": "nmap", "port": 3389, "description": "RDP NTLM negotiation metadata"}, "nmap_ssh_algorithms": {"binary": "nmap", "port": 22, "description": "Advertised SSH algorithms"}, @@ -48,7 +49,8 @@ "nmap_rdp_info": "rdp-ntlm-info", "nmap_ssh_algorithms": "ssh2-enum-algos", } -_DIAG_INFO_TOOLS = ("nmap_tls_cert", "nmap_smb_signing", "nmap_smbv1", "nmap_ssh_algorithms") +_DIAG_DEFAULT_INFO_TOOLS = ("nmap_tls_cert", "nmap_smb_signing", "nmap_smbv1", "nmap_ssh_algorithms") +_DIAG_INFO_TOOLS = _DIAG_DEFAULT_INFO_TOOLS + ("pptp_control",) _DIAG_EXCLUDED_NETWORK = _diag_ipaddress.ip_network("172.16.199.0/24") _DIAG_VPN_UDP_PORTS = "500,1194,1701,4500,51820" _DIAG_VPN_PORT_HINTS = { @@ -265,6 +267,9 @@ def _diag_access_module(): def _diag_executable(tool: str) -> str | None: + if tool == "pptp_control": + return _diag_sys.executable if globals().get("_DIAGNOSTIC_PPTP_SOURCE") or ( + globals().get("__file__") and _diag_Path(__file__).with_name("pptp_check.py").is_file()) else None if tool == "smbv1_access": try: path = globals().get("__file__") @@ -297,6 +302,11 @@ def build_diagnostic_command(tool: str, host: str, *, domain: str | None = None, seconds = str(_diag_math.ceil(_diag_timeout(timeout, 600 if tool == "nmap_tcp_ports" else 120))) exe = executable or str(DIAGNOSTIC_TOOLS[tool]["binary"]) endpoint = "[" + address + "]" if ":" in address else address + if tool == "pptp_control": + source = globals().get("_DIAGNOSTIC_PPTP_SOURCE") + worker = ["-c", source] if source else [str(_diag_Path(__file__).with_name("pptp_check.py").resolve())] + return [executable or _diag_sys.executable, "-I"] + worker + [ + "--target", address, "--timeout", seconds, "--mode", "active", "--scope-confirmed"] if tool == "smbv1_access": share, username, account_domain = _diag_access_module().validate_options(smb_share, smb_user, smb_domain) return [executable or _diag_sys.executable, "-I", str(_diag_Path(__file__).with_name("smbv1_access.py").resolve()), @@ -556,9 +566,84 @@ def _diag_smbv1_observation(result): return observation +_DIAG_PPTP_MODULE = None + + +def _diag_pptp_module(): + """Load the pure parser from the installed worker or the standalone embedding.""" + global _DIAG_PPTP_MODULE + if _DIAG_PPTP_MODULE is None: + name = "_adpentest_pptp_check_" + str(id(globals())) + embedded = globals().get("_DIAGNOSTIC_PPTP_SOURCE") + if embedded: + module = _diag_types.ModuleType(name) + exec(compile(embedded, "", "exec"), module.__dict__) + else: + path = _diag_Path(__file__).with_name("pptp_check.py") + spec = _diag_importlib.spec_from_file_location(name, path) + module = _diag_importlib.module_from_spec(spec) + spec.loader.exec_module(module) + _DIAG_PPTP_MODULE = module + return _DIAG_PPTP_MODULE + + +def _diag_pptp_observation(result): + observation = {"status": "unknown", "source": "pptp:sccrq-sccrp", "port": 1723, + "reason": "no-completed-control-check", "details": { + "tcp_state": "not-tested", "check_status": "not-tested", "authentication": "not-tested", + "gre": {"ip_protocol": 47, "status": "not-tested"}, "vpn_tunnel": "not-tested"}} + if result.get("mode") != "active" or result.get("status") not in {"completed", "failed", "timeout"}: + return observation + if result.get("output_truncated"): + observation["reason"] = "truncated-output" + return observation + text = result.get("stdout", "") + try: + if not isinstance(text, str) or not text or len(text.encode("utf-8")) > 4096: + raise ValueError("missing-or-oversized-worker-output") + data = _diag_json.loads(text) + endpoint = result.get("resolved_address") or result.get("normalized_host") or result.get("host") + if (not isinstance(data, dict) or data.get("schema_version") != 1 + or data.get("source") != observation["source"] or data.get("port") != 1723 + or data.get("target") != str(_diag_ipaddress.ip_address(endpoint)) + or data.get("tcp_state") not in {"not-tested", "connected", "refused"}): + raise ValueError("worker-result-mismatch") + state = data.get("status") + if state != "observed": + if state not in {"connection-refused", "timeout", "invalid-reply", "connection-error"}: + raise ValueError("invalid-worker-status") + reason = data.get("reason") + if not isinstance(reason, str) or not _diag_re.fullmatch(r"[a-z0-9-]{1,96}", reason): + raise ValueError("invalid-worker-reason") + observation["details"]["tcp_state"] = data["tcp_state"] + observation["details"]["check_status"] = state + observation["reason"] = reason + return observation + if result.get("status") != "completed" or result.get("returncode") != 0 or data["tcp_state"] != "connected": + raise ValueError("uncompleted-control-check") + raw = data.get("reply_hex") + if not isinstance(raw, str) or not _diag_re.fullmatch(r"[0-9a-f]{312}", raw): + raise ValueError("invalid-reply-evidence") + module = _diag_pptp_module() + details = module.parse_start_reply(bytes.fromhex(raw)) + cleanup = data.get("cleanup") + if cleanup not in {"not-needed", "connection-closed", "stop-acknowledged"}: + raise ValueError("invalid-cleanup-status") + if cleanup == "stop-acknowledged" and data.get("stop_reply_hex") != ( + module._header(16, 4) + b"\x01\x00\x00\x00").hex(): + raise ValueError("invalid-stop-reply-evidence") + details.update(tcp_state="connected", check_status="observed", cleanup=cleanup) + observation.update(status="observed", reason="valid-control-reply", details=details) + except (ValueError, TypeError, AttributeError, UnicodeError) as exc: + observation["reason"] = str(exc) if isinstance(exc, ValueError) else "invalid-worker-output" + return observation + + def _diag_information_observation(result): """Read bounded metadata from one completed, address-bound NSE result.""" tool = result.get("tool") + if tool == "pptp_control": + return _diag_pptp_observation(result) if tool == "nmap_smbv1": smb = _diag_smbv1_observation(result) return {"status": "observed" if smb["status"] != "unknown" else "unknown", @@ -752,6 +837,10 @@ def add(code, title, evidence, remediation): if offered: add("ssh-legacy-algorithms-advertised", "SSH advertises legacy algorithms", {"algorithms": offered}, "Review dependent clients, update SSH and remove the listed legacy algorithms.") + elif result["tool"] == "pptp_control" and details["control_status"] == "accepted": + add("pptp-control-observed", "PPTP control endpoint accepted a connection", + {key: details[key] for key in ("protocol_version", "result_code", "vendor", "peer_hostname")}, + "Review whether this PPTP endpoint is still required and restrict its exposure to approved clients.") return findings @@ -779,6 +868,8 @@ def build_information_report(results): report["summary"] = {"checks": len(report["checks"]), "review_findings": len(report["findings"]), "observed_checks": sum(item["information"]["status"] == "observed" for item in report["checks"]), "unknown_checks": sum(item["information"]["status"] == "unknown" for item in report["checks"])} + if any(result.get("tool") == "pptp_control" for result in results): + report["limitations"].append("PPTP control replies do not establish authentication, GRE reachability or a working VPN tunnel.") return report @@ -1099,6 +1190,10 @@ def run_diagnostic_tool(tool: str, host: str, *, domain: str | None = None, if tool in _DIAG_INFO_TOOLS: result["information"] = _diag_information_observation(result) result["information_findings"] = _diag_information_findings(result, result["information"]) + if tool == "pptp_control": + result["pptp"] = result["information"] + if result["status"] == "failed" and result["information"]["reason"] == "pptp-deadline-exceeded": + result["status"] = "timeout" if result["status"] == "interrupted": result["run_interrupted"] = True result["finished_at"] = _diag_timestamp() @@ -1114,7 +1209,7 @@ def _diagnostic_cli_requested(argv: list[str]) -> bool: if arg.split("=", 1)[0] in {"--diagnostic-tools", "--list-diagnostic-tools", "--replay-diagnostic-results", "--port-map", "--ports", "--port-map-timeout", "--smbv1-check", "--smbv1-access", "--ALL-PORTS", "--all-ports", - "--info-checks", "--info-report", + "--info-checks", "--info-report", "--pptp-check", "--smb-share", "--smb-user", "--smb-domain"}: return True return False @@ -1133,6 +1228,7 @@ def diagnostic_main(argv: list[str] | None = None) -> int: parser.add_argument("--ports", help="TCP numbers/ranges for port mapping; default: 1-1024; full map: 1-65535") parser.add_argument("--port-map-timeout", type=float, help="Port-mapping deadline (1-600 seconds); defaults to --timeout") parser.add_argument("--smbv1-check", action="store_true", help="Add nmap_smbv1 protocol negotiation") + parser.add_argument("--pptp-check", action="store_true", help="Attempt one PPTP control exchange on TCP/1723") parser.add_argument("--smbv1-access", action="store_true", help="Add smbv1_access; hidden password prompt in active mode") parser.add_argument("--smb-share", help="One known share name, without a path") parser.add_argument("--smb-user", help="One existing account username; use --smb-domain separately") @@ -1150,9 +1246,9 @@ def diagnostic_main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) shortcuts = [tool for enabled, tool in ((args.port_map or args.all_ports, "nmap_tcp_ports"), (args.all_ports, "nmap_vpn_udp_ports"), (args.smbv1_check, "nmap_smbv1"), - (args.smbv1_access, "smbv1_access")) if enabled] + (args.smbv1_access, "smbv1_access"), (args.pptp_check, "pptp_control")) if enabled] if args.info_checks: - shortcuts.extend(_DIAG_INFO_TOOLS) + shortcuts.extend(_DIAG_DEFAULT_INFO_TOOLS) if args.replay_diagnostic_results: if args.list_diagnostic_tools or args.diagnostic_tools or shortcuts or args.info_report: parser.error("Replay cannot be combined with tool execution or listing") @@ -1177,7 +1273,7 @@ def diagnostic_main(argv: list[str] | None = None) -> int: if not (args.diagnostic_tools or shortcuts) or not args.target: parser.error("Select --diagnostic-tools or a diagnostic shortcut, and supply --target") # Authenticated access is deliberate opt-in; `all` never asks for credentials. - selected = ([tool for tool in DIAGNOSTIC_TOOLS if tool not in {"smbv1_access", "nmap_vpn_udp_ports"}] if args.diagnostic_tools == "all" + selected = ([tool for tool in DIAGNOSTIC_TOOLS if tool not in {"smbv1_access", "nmap_vpn_udp_ports", "pptp_control"}] if args.diagnostic_tools == "all" else [item.strip() for item in (args.diagnostic_tools or "").split(",") if item.strip()]) selected = list(dict.fromkeys(selected + shortcuts)) if not selected: @@ -1250,7 +1346,7 @@ def options(tool): "target": results[0]["target"], "input_target": args.target, "run_id": run_id, "assessment": "not-assessed", "tools": results} report_failed = False - if args.info_checks or args.info_report: + if args.info_checks or args.info_report or args.pptp_check or "pptp_control" in selected: report = build_information_report(results) output["information_report"] = report if report_path is not None: diff --git a/adpentest/pptp_check.py b/adpentest/pptp_check.py new file mode 100644 index 0000000..49c8364 --- /dev/null +++ b/adpentest/pptp_check.py @@ -0,0 +1,160 @@ +"""One bounded RFC 2637 control exchange; no PPP authentication or GRE traffic.""" +import argparse +import ipaddress +import json +import math +import socket +import struct +import time + +PORT = 1723 +COOKIE = 0x1A2B3C4D +HEADER = struct.Struct("!HHIHH") +START_LENGTH = 156 + + +def _header(length, control_type): + return HEADER.pack(length, 1, COOKIE, control_type, 0) + + +def start_request(): + # PNS: no requested user calls, no credentials and no local host disclosure. + return _header(START_LENGTH, 1) + struct.pack( + "!HHIIHH64s64s", 0x0100, 0, 3, 3, 0, 0, b"adpentest-check", b"AdPentest" + ) + + +def parse_start_reply(raw): + if not isinstance(raw, bytes) or len(raw) != START_LENGTH: + raise ValueError("invalid-reply-length") + if HEADER.unpack(raw[:HEADER.size]) != (START_LENGTH, 1, COOKIE, 2, 0): + raise ValueError("invalid-control-header") + version, result, error, framing, bearer, channels, firmware, host, vendor = struct.unpack( + "!HBBIIHH64s64s", raw[HEADER.size:] + ) + if result not in range(1, 6) or (result != 2 and error != 0): + raise ValueError("invalid-result-code") + + def label(value): + # Retain arbitrary peer bytes as hex evidence; display printable ASCII only. + return "".join(chr(b) if 32 <= b <= 126 else "?" for b in value.split(b"\0", 1)[0]) + + return { + "protocol_version": f"{version >> 8}.{version & 255}", + "result_code": result, "error_code": error, + "control_status": "accepted" if result == 1 and version == 0x0100 else + "unsupported-version" if result == 1 else "rejected", + "framing_capabilities": framing, "bearer_capabilities": bearer, + "maximum_channels": channels, "firmware_revision": firmware, + "peer_hostname": label(host), "vendor": label(vendor), + "authentication": "not-tested", "gre": {"ip_protocol": 47, "status": "not-tested"}, + "vpn_tunnel": "not-tested", + } + + +def _read_frame(connection, deadline, expected_length, control_type): + data = bytearray() + wanted = HEADER.size + while len(data) < wanted: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("pptp-deadline-exceeded") + connection.settimeout(remaining) + chunk = connection.recv(wanted - len(data)) + if not chunk: + raise ValueError("incomplete-reply") + data.extend(chunk) + if len(data) == HEADER.size: + if HEADER.unpack(data) != (expected_length, 1, COOKIE, control_type, 0): + raise ValueError("invalid-control-header") + wanted = expected_length + return bytes(data) + + +def _send(connection, packet, deadline): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("pptp-deadline-exceeded") + connection.settimeout(remaining) + connection.sendall(packet) + + +def check_pptp(target, timeout=15, *, mode="dry-run", scope_confirmed=False): + """Use one literal address and one shared deadline; close our control channel.""" + if not isinstance(target, str) or "%" in target: + raise ValueError("PPTP worker requires one literal IP without a scope ID") + address = ipaddress.ip_address(target) + if address.is_multicast or address.is_unspecified: + raise ValueError("PPTP worker requires a unicast endpoint") + if isinstance(timeout, bool) or not isinstance(timeout, (int, float)) or not math.isfinite(timeout) or not 1 <= timeout <= 120: + raise ValueError("PPTP timeout must be 1-120 seconds") + if mode not in {"active", "dry-run"}: + raise ValueError("Invalid PPTP mode") + result = {"schema_version": 1, "source": "pptp:sccrq-sccrp", "target": str(address), + "port": PORT, "status": "dry-run", "tcp_state": "not-tested", + "reply_hex": "", "cleanup": "not-needed", "stop_reply_hex": "", "reason": "dry-run"} + if mode != "active": + return result + if scope_confirmed is not True: + result.update(status="blocked", reason="scope-required") + return result + # Reserve a small part of the outer diagnostic deadline for process startup/output. + deadline = time.monotonic() + max(0.5, timeout - 0.5) + connection = None + try: + # socket.connect with a numeric address performs no second DNS lookup. + family = socket.AF_INET6 if address.version == 6 else socket.AF_INET + connection = socket.socket(family, socket.SOCK_STREAM) + connection.settimeout(max(0.001, deadline - time.monotonic())) + endpoint = (str(address), PORT, 0, 0) if address.version == 6 else (str(address), PORT) + connection.connect(endpoint) + result["tcp_state"] = "connected" + _send(connection, start_request(), deadline) + reply = _read_frame(connection, deadline, START_LENGTH, 2) + result["reply_hex"] = reply.hex() + details = parse_start_reply(reply) + result.update(status="observed", reason="valid-control-reply") + if details["result_code"] == 1: + # This newly opened channel has no user calls. Stop only our channel. + result["cleanup"] = "connection-closed" + try: + cleanup_deadline = min(deadline, time.monotonic() + 0.25) + _send(connection, _header(16, 3) + struct.pack("!BBH", 1, 0, 0), cleanup_deadline) + stop = _read_frame(connection, cleanup_deadline, 16, 4) + result["stop_reply_hex"] = stop.hex() + code, error, reserved = struct.unpack("!BBH", stop[HEADER.size:]) + if (code, error, reserved) == (1, 0, 0): + result["cleanup"] = "stop-acknowledged" + except (OSError, ValueError, TimeoutError): + pass # The valid start reply remains evidence if stop is not acknowledged. + except ConnectionRefusedError: + result.update(status="connection-refused", tcp_state="refused", reason="connection-refused") + except TimeoutError: + result.update(status="timeout", reason="pptp-deadline-exceeded") + except ValueError as exc: + result.update(status="invalid-reply", reason=str(exc)) + except OSError as exc: + result.update(status="connection-error", reason="socket-error-" + str(exc.errno if exc.errno is not None else "unknown")) + finally: + if connection is not None: + connection.close() + return result + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False) + parser.add_argument("--target", required=True) + parser.add_argument("--timeout", type=float, default=15) + parser.add_argument("--mode", choices=("dry-run", "active"), default="dry-run") + parser.add_argument("--scope-confirmed", action="store_true") + args = parser.parse_args(argv) + try: + result = check_pptp(args.target, args.timeout, mode=args.mode, scope_confirmed=args.scope_confirmed) + except ValueError as exc: + parser.error(str(exc)) + print(json.dumps(result)) + return 0 if result["status"] in {"observed", "dry-run"} else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/DIAGNOSTIC_TOOLS.md b/docs/DIAGNOSTIC_TOOLS.md index 5fb3ddc..b06363a 100644 --- a/docs/DIAGNOSTIC_TOOLS.md +++ b/docs/DIAGNOSTIC_TOOLS.md @@ -1,11 +1,11 @@ # Diagnostic tools and mandatory ORM history -The thirteen opt-in profiles in `adpentest/diagnostics.py` persist every terminal +The fourteen opt-in profiles in `adpentest/diagnostics.py` persist every terminal result using SQLAlchemy 2.x and SQLite. This covers the diagnostic CLI, direct `run_diagnostic_tool(...)` calls, and the optional standalone-core embedding. Calls without diagnostic flags still delegate to the existing legacy scanner; that scanner's sqlite3 implementation is **not converted by this change**. -Package version is `2.14.213.78.21.8`. +Package version is `2.14.213.78.21.9`. The dashboard d2.aspx and default scan selections are unchanged. ## Profiles @@ -22,6 +22,7 @@ The dashboard d2.aspx and default scan selections are unchanged. | `nmap_smbv1` | nmap | SMB dialect negotiation on 445, including SMBv1 | | `nmap_tcp_ports` | nmap | Selected TCP port states on one host; no service probes | | `nmap_vpn_udp_ports` | nmap | UDP 500, 1194, 1701, 4500 and 51820 on one host; requires `--ALL-PORTS` | +| `pptp_control` | Python standard library | One PPTP control exchange on TCP/1723, selected with `--pptp-check` | | `smbv1_access` | Python + required impacket | Existing account and explicit share; SMBv1 root directory listing on 445 | | `nmap_rdp_info` | nmap | RDP negotiation metadata on 3389 | | `nmap_ssh_algorithms` | nmap | Advertised SSH algorithms on 22 | @@ -31,7 +32,7 @@ Python 3.10+, `SQLAlchemy>=2.0,<3` and `impacket>=0.13,<0.14` are required. The `smb` extra remains an alias for older install commands. Install the release: ```powershell -python -m pip install --upgrade "adpentest==2.14.213.78.21.8" +python -m pip install --upgrade "adpentest==2.14.213.78.21.9" python -m adpentest --list-diagnostic-tools python -m adpentest --diagnostic-tools all --target 192.0.2.10 --domain lab.example --mode dry-run --db-path "C:\inetpub\wwwroot\App_Data\AdPentest\results.sqlite" ``` @@ -47,6 +48,42 @@ validation and dependency checks. It performs one NTLMv2 login and a read-only root listing; guest sessions are rejected. See [port mapping and SMB access](SMBV1_DATA.md#port-mapping-and-authenticated-share-access) for the combined shortcuts, TCP range selection, extended port-map timeout and results. +## PPTP control check + +`--pptp-check` (or `--diagnostic-tools pptp_control`) adds one standard +[RFC 2637](https://www.rfc-editor.org/rfc/rfc2637) Start-Control-Connection exchange +on TCP/1723. It uses Python's standard library, requires no PPTP client or Nmap, +and accepts one endpoint through the existing DNS resolution and address-pinning +rules. It can be combined explicitly with `--ALL-PORTS`. + +```console +python -m adpentest --pptp-check --target vpn.example --domain lab.example --mode dry-run --info-report ./pptp-review.json --db-path ./pptp-history.sqlite +``` + +For an endpoint within the approved scope, choose `--mode active --scope-confirmed`. +Dry runs and blocked requests perform no DNS queries or socket connections. +The check has one total deadline, performs no retries, and reads at most the fixed +156-byte start reply and 16-byte stop reply. Length, message type, cookie, control +type and result/error fields are validated. Peer names are displayed as bounded +printable labels; the raw reply is retained in hex as evidence. + +Results distinguish a valid accepted control channel, a valid rejection, an +unsupported version, a refused TCP connection, a timeout and invalid/incomplete +protocol data. A successful start reply is retained if the stop acknowledgement +is missing; the socket is always closed. Stop messages apply only to the newly +opened control channel, which has no user calls. + +`pptp` and `information` in each result expose the structured observation. +The ORM stores it with the raw worker JSON, run ID, pinned address and supplied +DNS domain; the JSON information report includes the same evidence and coverage. +Accepted control channels produce a configuration-review finding with +`confirmed_vulnerability: false`. Authentication, GRE protocol 47 and a working +VPN tunnel remain `not-tested`. A protocol reply is not a verified vulnerability. + +The profile is explicitly selected; existing `--info-checks` and `--diagnostic-tools all` +selections are unchanged. Python API callers can use +`run_diagnostic_tool("pptp_control", host, mode="active", scope_confirmed=True, ...)`. + ## Limited information checks and report `--info-checks` selects four fixed metadata profiles: `nmap_tls_cert`, diff --git a/docs/RELEASE_2.14.213.78.21.9.md b/docs/RELEASE_2.14.213.78.21.9.md new file mode 100644 index 0000000..91fc162 --- /dev/null +++ b/docs/RELEASE_2.14.213.78.21.9.md @@ -0,0 +1,18 @@ +# 2.14.213.78.21.9 + +Adds `--pptp-check` and the `pptp_control` diagnostic profile. An explicitly +authorized check sends one standard PPTP control request to TCP/1723 on the +single pinned endpoint, validates the bounded reply, records server metadata +and result codes, and closes its newly opened connection. Python's standard +library supplies the client; no new runtime dependency is added. + +Accepted and rejected replies are distinct. Refusals, timeouts, malformed +replies, dry runs and blocked attempts remain in domain-linked ORM history. +The information report includes the same protocol evidence and reports +authentication, GRE and VPN tunnels as untested. Control acceptance is a +manual configuration-review finding, not a confirmed vulnerability. + +Existing all-ports inventory, TLS/SMB/SSH information selections, DNS pinning, +SMBv1 inventory and access, required Impacket and domain/XML history are +preserved. PPTP checks remain a separate opt-in. Test coverage uses synthetic +packets, mocked sockets/tools and temporary SQLite; no remote target was used. diff --git a/docs/SMBV1_DATA.md b/docs/SMBV1_DATA.md index 15ece64..f329a97 100644 --- a/docs/SMBV1_DATA.md +++ b/docs/SMBV1_DATA.md @@ -1,4 +1,4 @@ -# SMBv1 observations and inventory — 2.14.213.78.21.8 +# SMBv1 observations and inventory — 2.14.213.78.21.9 ## Active single-host protocol check @@ -37,7 +37,7 @@ package dependency `impacket>=0.13,<0.14`; it does not require the external Samb `smbclient` binary. Existing `[smb]` install commands remain compatible: ```console -python -m pip install --upgrade --no-cache-dir --index-url https://pypi.org/simple "adpentest==2.14.213.78.21.8" +python -m pip install --upgrade --no-cache-dir --index-url https://pypi.org/simple "adpentest==2.14.213.78.21.9" ``` Preview the combined operations for one host and known share: @@ -123,7 +123,7 @@ References: [Nmap port specification](https://nmap.org/book/man-port-specificati ### Host, user and UNC inventory -Version `2.14.213.78.21.8` searches data already present in an export. It does +Version `2.14.213.78.21.9` searches data already present in an export. It does not discover live accounts, query remote shares or resolve host aliases. ```console diff --git a/pyproject.toml b/pyproject.toml index ffe77f1..7c34f76 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "adpentest" -version = "2.14.213.78.21.8" +version = "2.14.213.78.21.9" description = "Active Directory penetration testing framework with automatic Domain Controller detection" readme = "README.md" requires-python = ">=3.10" @@ -63,7 +63,7 @@ dev = [ Homepage = "https://github.com/netanelcyber/AdPentestAI-Python" Repository = "https://github.com/netanelcyber/AdPentestAI-Python.git" Issues = "https://github.com/netanelcyber/AdPentestAI-Python/issues" -Documentation = "https://github.com/netanelcyber/AdPentestAI-Python/blob/release/pypi-2.14.213.78.21.8/README.md" +Documentation = "https://github.com/netanelcyber/AdPentestAI-Python/blob/release/pypi-2.14.213.78.21.9/README.md" [project.scripts] adpentest = "adpentest:main" diff --git a/tests/test_diagnostic_information.py b/tests/test_diagnostic_information.py index 837adf2..1c99aeb 100644 --- a/tests/test_diagnostic_information.py +++ b/tests/test_diagnostic_information.py @@ -124,7 +124,7 @@ def test_info_dry_run_exports_unknown_report_without_dns_or_process(tmp_path, ca code = diag.diagnostic_main(["--info-checks", "--target", "vpn.example", "--info-report", str(destination)]) output = json.loads(capsys.readouterr().out) assert code == 0 and output["information_report_file"]["status"] == "saved" - assert [row["tool"] for row in output["tools"]] == list(diag._DIAG_INFO_TOOLS) + assert [row["tool"] for row in output["tools"]] == list(diag._DIAG_DEFAULT_INFO_TOOLS) assert json.loads(destination.read_text(encoding="utf-8")) == output["information_report"] assert output["information_report"]["summary"]["unknown_checks"] == 4 assert not output["information_report"]["findings"] diff --git a/tests/test_diagnostic_pptp.py b/tests/test_diagnostic_pptp.py new file mode 100644 index 0000000..fdce822 --- /dev/null +++ b/tests/test_diagnostic_pptp.py @@ -0,0 +1,255 @@ +"""PPTP control fixtures and mocked sockets only; no target connections.""" +import json +import sqlite3 +import struct +from unittest import mock + +import pytest + +from adpentest import diagnostics as diag +from adpentest import pptp_check as pptp + + +def start_reply(*, version=0x0100, code=1, error=0, host=b"fixture-vpn", vendor=b"Fixture Vendor"): + return pptp._header(156, 2) + struct.pack("!HBBIIHH64s64s", version, code, error, 3, 3, 8, 42, host, vendor) + + +STOP = pptp._header(16, 4) + b"\x01\x00\x00\x00" + + +class SocketFixture: + def __init__(self, response, chunk=11, error=None): + self.response = response + self.chunk = chunk + self.error = error + self.sent = [] + self.read_sizes = [] + self.timeouts = [] + self.closed = False + + def settimeout(self, value): + self.timeouts.append(value) + + def connect(self, endpoint): + self.endpoint = endpoint + if self.error: + raise self.error + + def sendall(self, data): + self.sent.append(data) + + def recv(self, size): + self.read_sizes.append(size) + take = min(size, self.chunk) + result, self.response = self.response[:take], self.response[take:] + return result + + def close(self): + self.closed = True + + +def exchange(response, **options): + connection = SocketFixture(response) + with mock.patch.object(pptp.socket, "socket", return_value=connection) as opened, mock.patch.object( + pptp.socket, "getaddrinfo", side_effect=AssertionError("worker must not resolve names")): + result = pptp.check_pptp("192.0.2.10", mode="active", scope_confirmed=True, **options) + assert opened.call_count == 1 and connection.closed + return result, connection + + +def worker_record(response=None, **changes): + data = {"schema_version": 1, "source": "pptp:sccrq-sccrp", "target": "192.0.2.10", "port": 1723, + "status": "observed", "tcp_state": "connected", "reply_hex": (response or start_reply()).hex(), + "cleanup": "stop-acknowledged", "stop_reply_hex": STOP.hex(), "reason": "valid-control-reply"} + data.update(changes) + return {"status": "completed", "returncode": 0, "stdout": json.dumps(data), "stderr": "", "output_truncated": False} + + +def result_record(**changes): + result = dict(worker_record(), tool="pptp_control", mode="active", resolved_address="192.0.2.10", + host="vpn.example", target="vpn.example", result_id="fixture-result", run_id="fixture-run") + result.update(changes) + return result + + +def test_fragmented_standard_exchange_closes_only_its_own_channel(): + result, connection = exchange(start_reply() + STOP) + assert result["status"] == "observed" and result["cleanup"] == "stop-acknowledged" + assert connection.endpoint == ("192.0.2.10", 1723) + assert [pptp.HEADER.unpack(data[:12])[3] for data in connection.sent] == [1, 3] + assert len(connection.sent[0]) == 156 and len(connection.sent[1]) == 16 + assert struct.unpack("!HHIIHH64s64s", connection.sent[0][12:])[:6] == (0x0100, 0, 3, 3, 0, 0) + assert max(connection.read_sizes) <= 144 + assert all(0 < timeout <= 15 for timeout in connection.timeouts) + details = pptp.parse_start_reply(bytes.fromhex(result["reply_hex"])) + assert details["control_status"] == "accepted" and details["vendor"] == "Fixture Vendor" + assert details["firmware_revision"] == 42 + assert details["authentication"] == details["vpn_tunnel"] == details["gre"]["status"] == "not-tested" + + +@pytest.mark.parametrize("code", [2, 3, 4, 5]) +def test_valid_rejection_is_protocol_evidence_without_an_accepted_channel(code): + result, connection = exchange(start_reply(code=code)) + assert result["status"] == "observed" and result["cleanup"] == "not-needed" + assert len(connection.sent) == 1 + assert pptp.parse_start_reply(bytes.fromhex(result["reply_hex"]))["control_status"] == "rejected" + + +@pytest.mark.parametrize("reply", [ + b"HTTP/1.1 200 OK\r\n", pptp._header(65535, 2), pptp._header(156, 1), + struct.pack("!HHIHH", 156, 1, 0, 2, 0), start_reply()[:80], start_reply(code=0), start_reply(error=1), +]) +def test_malformed_wrong_protocol_or_truncated_reply_never_becomes_observed(reply): + result, connection = exchange(reply) + assert result["status"] == "invalid-reply" and len(connection.sent) == 1 + assert max(connection.read_sizes) <= 144 + + +def test_missing_stop_ack_preserves_the_valid_start_reply_and_closes_socket(): + result, connection = exchange(start_reply()) + assert result["status"] == "observed" and result["cleanup"] == "connection-closed" + assert connection.closed + + +def test_unsupported_version_is_recorded_and_its_new_channel_is_stopped(): + result, connection = exchange(start_reply(version=0x0200) + STOP) + assert result["cleanup"] == "stop-acknowledged" and len(connection.sent) == 2 + assert pptp.parse_start_reply(bytes.fromhex(result["reply_hex"]))["control_status"] == "unsupported-version" + + +@pytest.mark.parametrize("status", ["failed", "timeout", "blocked", "dry-run"]) +def test_uncompleted_results_cannot_create_protocol_findings(status): + report = diag.build_information_report([result_record(status=status)]) + assert report["summary"]["unknown_checks"] == 1 and not report["findings"] + + +def test_server_labels_are_bounded_printable_and_raw_evidence_is_preserved(): + raw = start_reply(host=b"vpn\x1b\nname", vendor=b"V" * 64) + data = pptp.parse_start_reply(raw) + assert data["peer_hostname"] == "vpn??name" and len(data["vendor"]) == 64 + + +def test_combining_pptp_with_all_ports_keeps_the_protocol_check_explicit(tmp_path, capsys): + diag.diagnostic_main(["--ALL-PORTS", "--pptp-check", "--target", "192.0.2.10", "--db-path", str(tmp_path / "combined.sqlite")]) + assert [row["tool"] for row in json.loads(capsys.readouterr().out)["tools"]] == [ + "nmap_tcp_ports", "nmap_vpn_udp_ports", "pptp_control"] + + +def test_one_total_deadline_limits_slow_fragmented_input(): + connection = SocketFixture(start_reply(), chunk=1) + ticks = iter(i / 10 for i in range(100)) + with mock.patch.object(pptp.socket, "socket", return_value=connection), mock.patch.object( + pptp.time, "monotonic", side_effect=lambda: next(ticks)): + result = pptp.check_pptp("192.0.2.10", timeout=1, mode="active", scope_confirmed=True) + assert result["status"] == "timeout" and connection.closed + assert len(connection.read_sizes) <= 3 and len(connection.sent) == 1 + + +def test_connection_refusal_is_distinct_from_protocol_support(): + connection = SocketFixture(b"", error=ConnectionRefusedError()) + with mock.patch.object(pptp.socket, "socket", return_value=connection): + result = pptp.check_pptp("192.0.2.10", mode="active", scope_confirmed=True) + assert result["status"] == "connection-refused" and result["tcp_state"] == "refused" + assert connection.closed and connection.sent == [] + + +def test_ipv6_is_pinned_without_another_dns_lookup(): + connection = SocketFixture(start_reply(code=4)) + with mock.patch.object(pptp.socket, "socket", return_value=connection) as opened: + result = pptp.check_pptp("2001:db8::10", mode="active", scope_confirmed=True) + assert opened.call_args.args == (pptp.socket.AF_INET6, pptp.socket.SOCK_STREAM) + assert connection.endpoint == ("2001:db8::10", 1723, 0, 0) and result["status"] == "observed" + + +@pytest.mark.parametrize("target", ["vpn.example", "192.0.2.0/24", "https://vpn.example", "0.0.0.0", "224.0.0.1", "fe80::1%eth0"]) +def test_worker_rejects_ambiguous_endpoints_before_opening_socket(target): + with mock.patch.object(pptp.socket, "socket") as opened, pytest.raises(ValueError): + pptp.check_pptp(target, mode="active", scope_confirmed=True) + opened.assert_not_called() + + +def test_worker_dry_run_and_missing_scope_do_not_connect(): + with mock.patch.object(pptp.socket, "socket") as opened: + assert pptp.check_pptp("192.0.2.10")["status"] == "dry-run" + assert pptp.check_pptp("192.0.2.10", mode="active")["status"] == "blocked" + opened.assert_not_called() + + +@pytest.mark.parametrize("data", [ + {"target": "192.0.2.11"}, {"port": 443}, {"reply_hex": "00" * 156}, + {"reply_hex": "a" * 65536}, {"stop_reply_hex": "00" * 16}, {"tcp_state": "refused"}, +]) +def test_wrong_endpoint_or_unverified_worker_evidence_is_unknown(data): + record = dict(result_record(), **worker_record(**data)) + report = diag.build_information_report([record]) + assert report["summary"]["unknown_checks"] == 1 and report["findings"] == [] + + +def test_control_acceptance_is_a_review_finding_with_explicit_coverage_limits(): + report = diag.build_information_report([result_record()]) + assert report["summary"] == {"checks": 1, "review_findings": 1, "observed_checks": 1, "unknown_checks": 0} + assert report["findings"][0]["code"] == "pptp-control-observed" + assert report["findings"][0]["confirmed_vulnerability"] is False + assert report["checks"][0]["information"]["details"]["vpn_tunnel"] == "not-tested" + rejected = diag.build_information_report([dict(result_record(), **worker_record(start_reply(code=4)))]) + assert rejected["summary"]["observed_checks"] == 1 and rejected["findings"] == [] + + +def test_cli_dry_run_and_missing_scope_never_resolve_or_launch(tmp_path, capsys): + with mock.patch.object(diag, "_diag_target_context") as resolve, mock.patch.object(diag, "_diag_process") as launch: + assert diag.diagnostic_main(["--pptp-check", "--target", "vpn.example", "--db-path", str(tmp_path / "dry.sqlite")]) == 0 + output = json.loads(capsys.readouterr().out) + assert output["tools"][0]["tool"] == "pptp_control" + assert output["information_report"]["summary"]["unknown_checks"] == 1 + with pytest.raises(SystemExit): + diag.diagnostic_main(["--pptp-check", "--target", "vpn.example", "--mode", "active", + "--db-path", str(tmp_path / "blocked.sqlite")]) + resolve.assert_not_called() + launch.assert_not_called() + + +def test_cli_saves_address_bound_reply_with_domain_and_report(tmp_path, capsys): + db, destination = tmp_path / "history.sqlite", tmp_path / "review.json" + context = {"host": "vpn.example", "target": "vpn.example", "resolved_address": "192.0.2.10"} + with mock.patch.object(diag, "_diag_target_context", return_value=context) as resolve, mock.patch.object( + diag, "_diag_process", return_value=worker_record()) as launch: + code = diag.diagnostic_main(["--pptp-check", "--target", "vpn.example", "--domain", "lab.example", "--mode", "active", + "--scope-confirmed", "--info-report", str(destination), "--db-path", str(db)]) + output = json.loads(capsys.readouterr().out) + assert code == 0 and resolve.call_count == launch.call_count == 1 + command = launch.call_args.args[0] + assert command[command.index("--target") + 1] == "192.0.2.10" + assert output["tools"][0]["pptp"]["details"]["control_status"] == "accepted" + assert json.loads(destination.read_text()) == output["information_report"] + with sqlite3.connect(db) as connection: + row = json.loads(connection.execute("SELECT result_json FROM diagnostic_results").fetchone()[0]) + assert row["domain"] == "lab.example" and row["pptp"] == output["tools"][0]["pptp"] + assert connection.execute("SELECT domain_name FROM domain_run_links").fetchone()[0] == "lab.example" + assert connection.execute("PRAGMA foreign_key_check").fetchall() == [] + + +@pytest.mark.parametrize("state,tcp,reason,terminal", [ + ("connection-refused", "refused", "connection-refused", "failed"), + ("timeout", "connected", "pptp-deadline-exceeded", "timeout"), + ("invalid-reply", "connected", "invalid-control-header", "failed"), + ("connection-error", "not-tested", "socket-error-101", "failed"), +]) +def test_failed_attempts_preserve_their_distinct_outcomes_in_domain_history(tmp_path, state, tcp, reason, terminal): + db = tmp_path / "failed.sqlite" + process = worker_record(status=state, tcp_state=tcp, reason=reason, reply_hex="", cleanup="not-needed") + process.update(status="failed", returncode=2) + with mock.patch.object(diag, "_diag_process", return_value=process): + result = diag.run_diagnostic_tool("pptp_control", "192.0.2.10", domain="lab.example", + mode="active", scope_confirmed=True, db_path=db) + assert result["status"] == terminal and result["pptp"]["details"]["check_status"] == state + assert result["pptp"]["details"]["tcp_state"] == tcp and result["information_findings"] == [] + with sqlite3.connect(db) as connection: + saved = json.loads(connection.execute("SELECT result_json FROM diagnostic_results").fetchone()[0]) + assert saved["status"] == terminal and saved["pptp"] == result["pptp"] + assert connection.execute("SELECT domain_name FROM domain_run_links").fetchone()[0] == "lab.example" + + +def test_pptp_remains_explicit_when_all_information_profiles_are_selected(tmp_path, capsys): + diag.diagnostic_main(["--diagnostic-tools", "all", "--target", "192.0.2.10", "--domain", "lab.example", "--db-path", str(tmp_path / "all.sqlite")]) + assert "pptp_control" not in {row["tool"] for row in json.loads(capsys.readouterr().out)["tools"]} + assert "pptp_control" not in diag._DIAG_DEFAULT_INFO_TOOLS diff --git a/tests/test_diagnostic_storage.py b/tests/test_diagnostic_storage.py index d8d809d..9c8c08a 100644 --- a/tests/test_diagnostic_storage.py +++ b/tests/test_diagnostic_storage.py @@ -179,7 +179,7 @@ def test_inventory_output_saved(self): self.assertEqual(code, 0) self.assertEqual(output["persistence"]["status"], "saved") row = self.query("SELECT stdout FROM diagnostic_results")[0] - self.assertEqual(len(json.loads(row["stdout"])["diagnostic_tools"]), 13) + self.assertEqual(len(json.loads(row["stdout"])["diagnostic_tools"]), 14) def test_explicit_db_path_takes_precedence_over_environment(self): explicit = self.root / "chosen" / "db # with spaces.sqlite" diff --git a/tests/test_diagnostic_tools.py b/tests/test_diagnostic_tools.py index 8a316ec..77fa42d 100644 --- a/tests/test_diagnostic_tools.py +++ b/tests/test_diagnostic_tools.py @@ -66,12 +66,12 @@ def test_unknown_tool(self): class CommandTests(unittest.TestCase): def test_all_profiles_build(self): - self.assertEqual(len(diag.DIAGNOSTIC_TOOLS), 13) + self.assertEqual(len(diag.DIAGNOSTIC_TOOLS), 14) for tool in diag.DIAGNOSTIC_TOOLS: with self.subTest(tool=tool): args = diag.build_diagnostic_command(tool, "192.0.2.10", domain="lab.example", smb_share="SHARE", smb_user="alice") self.assertTrue(all(isinstance(arg, str) for arg in args)) - self.assertEqual(args[0], sys.executable if tool == "smbv1_access" else diag.DIAGNOSTIC_TOOLS[tool]["binary"]) + self.assertEqual(args[0], sys.executable if tool in {"smbv1_access", "pptp_control"} else diag.DIAGNOSTIC_TOOLS[tool]["binary"]) if diag.DIAGNOSTIC_TOOLS[tool]["binary"] == "nmap": self.assertIn("-vv", args) @@ -137,7 +137,7 @@ def test_nmap_ipv6_flag(self): class RunnerTests(unittest.TestCase): def test_inventory_does_not_execute(self): with mock.patch.object(diag, "_diag_process") as process: - self.assertEqual(len(diag.diagnostic_tool_inventory()), 13) + self.assertEqual(len(diag.diagnostic_tool_inventory()), 14) process.assert_not_called() def test_dry_run_does_not_execute(self): @@ -211,7 +211,7 @@ def test_inventory_cli(self): output = io.StringIO() with contextlib.redirect_stdout(output): self.assertEqual(diag.diagnostic_main(["--list-diagnostic-tools"]), 0) - self.assertEqual(len(json.loads(output.getvalue())["diagnostic_tools"]), 13) + self.assertEqual(len(json.loads(output.getvalue())["diagnostic_tools"]), 14) def test_all_dry_run(self): output = io.StringIO() @@ -336,7 +336,7 @@ def test_console_entrypoint_exists(self): def test_existing_exports_preserved(self): with package_fixture() as (package, _, report): - self.assertEqual(package.__version__, "2.14.213.78.21.8") + self.assertEqual(package.__version__, "2.14.213.78.21.9") self.assertIs(package.generate_html_report, report.generate_html_report) self.assertIs(package.export_html_report, report.export_html_report) diff --git a/tests/test_release_version.py b/tests/test_release_version.py index 399f537..7939991 100644 --- a/tests/test_release_version.py +++ b/tests/test_release_version.py @@ -6,7 +6,7 @@ from packaging.version import Version ROOT = Path(__file__).resolve().parents[1] -EXPECTED_VERSION = "2.14.213.78.21.8" +EXPECTED_VERSION = "2.14.213.78.21.9" def test_package_and_project_versions_match_requested_release(): @@ -27,7 +27,7 @@ def test_package_and_project_versions_match_requested_release(): def test_packaging_preserves_requested_version(): assert str(Version(EXPECTED_VERSION)) == EXPECTED_VERSION - assert Version(EXPECTED_VERSION) > Version("2.14.213.78.21.7") + assert Version(EXPECTED_VERSION) > Version("2.14.213.78.21.8") def test_orm_dependency_is_declared_in_both_install_paths(): diff --git a/tools/add_diagnostic_tools.py b/tools/add_diagnostic_tools.py index fbeb770..52112e6 100644 --- a/tools/add_diagnostic_tools.py +++ b/tools/add_diagnostic_tools.py @@ -52,8 +52,11 @@ def build_block(fragment: str) -> str: if isinstance(node, ast.Assign) and any(isinstance(target, ast.Name) and target.id == "_SMBV1_DATA_SOURCE" for target in node.targets)) compile(inventory, "smbv1_data.py", "exec") + pptp = (Path(__file__).resolve().parents[1] / "adpentest" / "pptp_check.py").read_text(encoding="utf-8") + compile(pptp, "pptp_check.py", "exec") return (BEGIN + "\n_DIAGNOSTIC_STORAGE_SOURCE = " + repr(storage) + "\n_DIAGNOSTIC_INVENTORY_SOURCE = " + repr(inventory) + + "\n_DIAGNOSTIC_PPTP_SOURCE = " + repr(pptp) + "\n" + body + "\n" + WRAPPER + END + "\n\n")