fix: harden scan resource limits

Author: sameeralam3127

requirements.txt

@@ -1,6 +1,4 @@
-pandas>=1.3.0
-tqdm>=4.45.0
-colorama>=0.4.4
-openpyxl>=3.0.0
-pythonping>=1.1.4
-argparse>=1.4.0
+pandas>=2.0
+tqdm>=4.65
+openpyxl>=3.1
+rich>=13.0

src/ipmg/core/ping.py

@@ -1,7 +1,7 @@
 import ipaddress
 import platform
 import re
-import subprocess
+import subprocess  # nosec B404
 from typing import Optional, Tuple
 
 from ipmg.exceptions import PingError
@@ -45,7 +45,8 @@ def ping_ip(ip: str, timeout: int, count: int) -> Tuple[str, Optional[float]]:
     cmd = ["ping", param, str(count), timeout_param, timeout_val, ip]
 
     try:
-        result = subprocess.run(
+        # Fixed argv list, no shell, and ip is validated before this call.
+        result = subprocess.run(  # nosec B603
             cmd,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,

src/ipmg/infrastructure/file_io.py

@@ -10,6 +10,7 @@
 
 
 SUPPORTED_INPUT_SUFFIXES = {".xlsx", ".xls", ".csv", ".txt", ".list"}
+MAX_EXPANDED_TARGETS = 65_536
 
 
 def _deduplicate(targets: Iterable[str]) -> list[str]:
@@ -31,6 +32,12 @@ def _expand_cidr(target: str) -> list[str]:
     if parsed.num_addresses == 1:
         return [str(parsed.network_address)]
 
+    if parsed.num_addresses - 2 > MAX_EXPANDED_TARGETS:
+        raise FileIOError(
+            f"CIDR target '{target}' expands to too many hosts. "
+            f"Maximum allowed hosts: {MAX_EXPANDED_TARGETS}."
+        )
+
     return [str(ip) for ip in parsed.hosts()]
 
 

src/ipmg/services/scan_service.py

@@ -32,6 +32,8 @@
 
 def run_scan(args) -> None:
     args.threads = clamp_int(args.threads, 1, 500)
+    args.timeout = clamp_int(args.timeout, 1, 60)
+    args.count = clamp_int(args.count, 1, 10)
 
     if not os.path.exists(args.input) and not args.discover:
         if Path(args.input).suffix.lower() in {".xlsx", ".xls", ".csv", ".txt", ".list"}:

tests/test_file_io.py

@@ -24,6 +24,11 @@ def test_load_targets_from_literal_cidr():
     assert load_targets("10.0.0.0/30") == ["10.0.0.1", "10.0.0.2"]
 
 
+def test_load_targets_rejects_large_cidr():
+    with pytest.raises(Exception):
+        load_targets("10.0.0.0/8")
+
+
 def test_load_targets_requires_expected_column(tmp_path):
     path = tmp_path / "targets.csv"
     pd.DataFrame({"Address": ["8.8.8.8"]}).to_csv(path, index=False)

tests/test_scan_service.py

@@ -41,9 +41,42 @@ def fake_print_summary(df, batch_timestamp, duration_seconds):
 
     run_scan(args)
 
+    assert args.timeout == 1
+    assert args.count == 1
+
     df: pd.DataFrame = captured["df"]
     statuses = dict(zip(df["IP Address"], df["Status"]))
     assert statuses["8.8.8.8"] == "Active"
     assert statuses["1.1.1.1"] == "Error"
     assert len(df["Batch Timestamp"].unique()) == 1
     assert (df["Scan Duration (s)"] >= 0).all()
+
+
+def test_run_scan_clamps_resource_limits(monkeypatch):
+    captured = {}
+
+    def fake_ping_ip(_ip, timeout, count):
+        captured["limits"] = (timeout, count)
+        return "Active", 1.0
+
+    monkeypatch.setattr("ipmg.services.scan_service.load_targets", lambda _source: ["8.8.8.8"])
+    monkeypatch.setattr("ipmg.services.scan_service.ping_ip", fake_ping_ip)
+    monkeypatch.setattr("ipmg.services.scan_service.save_results", lambda *_args: None)
+    monkeypatch.setattr("ipmg.services.scan_service.print_summary", lambda *_args: None)
+
+    args = SimpleNamespace(
+        input="targets.csv",
+        output="results",
+        timeout=999,
+        count=999,
+        threads=999,
+        formats=["csv"],
+        discover=False,
+        resolve=False,
+        interval=None,
+    )
+
+    run_scan(args)
+
+    assert args.threads == 500
+    assert captured["limits"] == (60, 10)