134 lines
6.6 KiB
Python
134 lines
6.6 KiB
Python
import io
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from rpcscope.console import compact, use_color
|
|
from rpcscope.models import HostEvidence, Report
|
|
from rpcscope.progress import ProgressEvent, ProgressReporter
|
|
from rpcscope.rules import evaluate
|
|
|
|
|
|
class Terminal(io.StringIO):
|
|
def isatty(self):
|
|
return True
|
|
|
|
|
|
def report(hosts, errors=None):
|
|
return Report("now", "test", [], None, None, hosts, evaluate(hosts), errors or [])
|
|
|
|
|
|
class ConsoleTests(unittest.TestCase):
|
|
def test_conflicting_results_are_not_counted_as_confirmed(self):
|
|
host = HostEvidence("192.0.2.1", checks={"ms17-010": "vulnerable"})
|
|
host.merge(HostEvidence(host.address, checks={"ms17-010": "not-vulnerable"}))
|
|
text = compact(report([host]))
|
|
self.assertIn("CONFLICT[MS17-010]", text)
|
|
self.assertNotIn("VULN[MS17-010]", text)
|
|
self.assertIn("0 confirmed", text)
|
|
self.assertIn("1 conflicting check", text)
|
|
finding = report([host]).findings[0]
|
|
self.assertEqual(finding.confidence, "low")
|
|
self.assertIn("conflict", finding.title)
|
|
|
|
def test_optional_signing_is_not_labelled_disabled(self):
|
|
host = HostEvidence("192.0.2.1", signing=False)
|
|
text = compact(report([host]))
|
|
self.assertIn("SIGNING-OPTIONAL", text)
|
|
self.assertNotIn("SIGNING-OFF", text)
|
|
|
|
def test_pending_and_interrupted_output_are_explicit(self):
|
|
host = HostEvidence("192.0.2.1", checks={"ms17-010": "pending"}, collection={"rpcdump": "pending"})
|
|
current = report([host])
|
|
current.mode = "interrupted"
|
|
text = compact(current)
|
|
self.assertIn("PENDING[rpcdump]", text)
|
|
self.assertNotIn("ERROR[rpcdump]", text)
|
|
self.assertIn("INTERRUPTED", text)
|
|
self.assertIn("UNKNOWN:1", text)
|
|
|
|
def test_compact_percentage_resets_explicitly_for_audit_phase(self):
|
|
output = Terminal()
|
|
reporter = ProgressReporter(output, mode="compact", color="never")
|
|
reporter(ProgressEvent("start", "tcp-discovery", total=509))
|
|
reporter(ProgressEvent("advance", "tcp-discovery", completed=508, total=509))
|
|
reporter(ProgressEvent("finish", "discovery", completed=509, total=649))
|
|
reporter(ProgressEvent("start", "batch:ms17-010", completed=509, total=649))
|
|
self.assertIn("0% audit [0/140]", reporter._dashboard(reporter.clock()))
|
|
reporter(ProgressEvent("finish", "ms17-010", "192.0.2.1", completed=579, total=649))
|
|
self.assertIn("50% audit [70/140]", reporter._dashboard(reporter.clock()))
|
|
|
|
def test_compact_separates_probe_results_and_collection_failures(self):
|
|
host = HostEvidence("192.0.2.1", hostname="HOST", domain="DOMAIN", checks={
|
|
"ms17-010": "not-vulnerable", "smbghost": "exposed",
|
|
"printnightmare": "skipped-access-denied", "zerologon": "inconclusive",
|
|
}, check_details={"zerologon": "checker exited with status 124"}, collection={"rpcdump": "failed"})
|
|
text = compact(report([host]))
|
|
self.assertIn("192.0.2.1 (HOST)", text)
|
|
self.assertIn("NOT-VULN[MS17-010]", text)
|
|
self.assertIn("EXPOSED[SMB3-COMPRESSION]", text)
|
|
self.assertIn("SKIP[access-denied:1]", text)
|
|
self.assertIn("UNKNOWN:1(timeout:1)", text)
|
|
self.assertIn("ERROR[rpcdump]", text)
|
|
self.assertNotIn("PARTIAL", text)
|
|
self.assertNotIn("PORTS:", text)
|
|
self.assertNotIn("4/4", text)
|
|
self.assertNotIn("\x1b", text)
|
|
|
|
def test_dead_hosts_are_hidden_and_uncertainty_is_not_a_clean_verdict(self):
|
|
text = compact(report([HostEvidence("192.0.2.2", services={"445": "timeout"}),
|
|
HostEvidence("192.0.2.1", checks={"ms17-010": "inconclusive"})]))
|
|
self.assertNotIn("192.0.2.2", text)
|
|
self.assertNotIn("NOT-VULN", text)
|
|
self.assertIn("UNKNOWN:1", text)
|
|
|
|
def test_verbose_contains_evidence_and_local_error_notes(self):
|
|
host = HostEvidence("192.0.2.1", checks={"ms17-010": "inconclusive"},
|
|
check_details={"ms17-010": "no SMB1 response"})
|
|
text = compact(report([host], ["rpcdump: ModuleNotFoundError"]), verbose=True)
|
|
self.assertIn("MS17-010: inconclusive", text)
|
|
self.assertIn("no SMB1 response", text)
|
|
self.assertIn("ModuleNotFoundError", text)
|
|
|
|
def test_color_detection_respects_redirects_and_no_color(self):
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
self.assertTrue(use_color(Terminal()))
|
|
self.assertFalse(use_color(io.StringIO()))
|
|
self.assertFalse(use_color(Terminal(), "never"))
|
|
with patch.dict("os.environ", {"NO_COLOR": "1"}):
|
|
self.assertFalse(use_color(Terminal()))
|
|
self.assertTrue(use_color(io.StringIO(), "always"))
|
|
|
|
def test_terminal_control_characters_in_host_and_evidence_are_removed(self):
|
|
host = HostEvidence("192.0.2.1", hostname="FAKE\x1b[2K\nNAME", checks={"ms17-010": "vulnerable"})
|
|
text = compact(report([host]))
|
|
self.assertIn("(FAKE NAME)", text)
|
|
self.assertNotIn("\x1b", text)
|
|
self.assertEqual(len(text.splitlines()), 2)
|
|
self.assertIn("\x1b[31;1mVULN[MS17-010]", compact(report([host]), color=True))
|
|
|
|
def test_compact_tty_uses_a_bar_without_completed_task_lines(self):
|
|
output = Terminal()
|
|
reporter = ProgressReporter(output, mode="compact", color="never")
|
|
reporter(ProgressEvent("start", "discovery", total=100))
|
|
reporter(ProgressEvent("finish", "discovery", completed=1, total=100))
|
|
reporter(ProgressEvent("start", "batch:ms17-010", total=100))
|
|
reporter(ProgressEvent("finish", "ms17-010", "192.0.2.1", "vulnerable", 99, 100))
|
|
reporter._render()
|
|
reporter(ProgressEvent("summary", "scan", status="complete", completed=100, total=100))
|
|
self.assertIn("[###############-]", output.getvalue())
|
|
self.assertIn("99%", output.getvalue())
|
|
self.assertNotIn("DONE", output.getvalue())
|
|
self.assertNotIn("192.0.2.1", output.getvalue())
|
|
self.assertIn("COMPLETE |", output.getvalue())
|
|
self.assertTrue(output.getvalue().endswith("\n"))
|
|
|
|
def test_compact_redirected_output_has_only_phase_and_final_lines(self):
|
|
output = io.StringIO()
|
|
reporter = ProgressReporter(output, mode="compact")
|
|
reporter(ProgressEvent("start", "discovery", total=4))
|
|
for i in range(4):
|
|
reporter(ProgressEvent("finish", "check", f"192.0.2.{i}", "unknown", i + 1, 4))
|
|
reporter(ProgressEvent("summary", "scan", status="incomplete", completed=4, total=4))
|
|
self.assertEqual(len(output.getvalue().splitlines()), 2)
|
|
self.assertNotIn("\x1b", output.getvalue())
|