diff options
Diffstat (limited to 'src/netmonitor/front/scanner_tab.py')
| -rwxr-xr-x | src/netmonitor/front/scanner_tab.py | 277 |
1 files changed, 0 insertions, 277 deletions
diff --git a/src/netmonitor/front/scanner_tab.py b/src/netmonitor/front/scanner_tab.py deleted file mode 100755 index 868fa74..0000000 --- a/src/netmonitor/front/scanner_tab.py +++ /dev/null @@ -1,277 +0,0 @@ -from textual import on
-from textual.app import ComposeResult
-from textual.widgets import Checkbox, Input, Select, Button, Label, SelectionList, Pretty
-from textual.widgets.selection_list import Selection
-from textual.containers import Container, Vertical, Horizontal, VerticalScroll
-from textual.events import Mount
-
-import psutil, ipaddress, shlex
-from typing import List, Dict, Optional
-
-from ..back.scanner_profiles_manager import ScannerProfilesManager
-from .scanner_tab_pushscreens import SaveProfilePushScreen
-
-class ScannerTab(Container):
-
- def __init__(self, manager: ScannerProfilesManager):
- super().__init__()
- self.manager = manager
- self.manager.on_message = self.on_manager_message
-
- def compose(self) -> ComposeResult:
- friendly_section = Container(id="friendly-command-section", classes="section card")
- friendly_section.border_title = "Scanner Configuration"
-
- with friendly_section:
- yield Label("Interface:", classes="label")
- try:
- interfaces = list(psutil.net_if_addrs().keys())
- except Exception:
- interfaces = []
-
- yield Select.from_values(
- interfaces,
- id="interface-select",
- allow_blank=False,
- classes="input"
- )
-
- yield Label("IP address:", classes="label")
- yield Input(
- value="127.0.0.1/32",
- placeholder="192.168.0.0/24",
- name="ip-input",
- id="ip",
- classes="input"
- )
-
- yield Label("Port range (or single port)", classes="label")
- with Horizontal(id="port-range", classes="input-row"):
- yield Input(placeholder="Start / Single", id="low-port-range", classes="input half")
- yield Input(placeholder="End (Optional)", id="high-port-range", classes="input half")
-
- yield Checkbox("Check for CVE's", id="cve-checkbox", classes="checkbox")
-
- yield Label("Scan options:", classes="label")
- with Horizontal(classes="selection-row"):
- yield SelectionList[str](
- Selection("Service version (-sV)", "-sV", False),
- Selection("OS detection (-O)", "-O", False),
- Selection("SYN scan (-sS)", "-sS", False),
- Selection("UDP scan (-sU)", "-sU", False),
- Selection("Fast scan (-F)", "-F", False),
- Selection("Aggressive scan (-A)", "-A", False),
- Selection("Ping skip (-Pn)", "-Pn", False),
- Selection("Timing T4 (-T4)", "-T4", True),
- classes="selection-list"
- )
-
- user_section = Container(id="user-command-section", classes="section card")
- user_section.border_title = "Manual Command"
-
- with user_section:
- yield Input(
- placeholder="sudo nmap -sS 192.168.1.1",
- name="user-nmap-input",
- id="usercommand",
- classes="input full"
- )
-
-
- final_section = Container(id="final-section", classes="section card")
- final_section.border_title = "Execution"
-
- with final_section:
- with Vertical(id="final-command", classes="command-output"):
- self.final_command = Pretty([], id="final-command-output")
- yield self.final_command
- with Vertical(id="buttons", classes="buttons-column"):
- yield Button("Scan now", id="scan-button", variant="success")
- yield Button("Save profile", id="save-button", variant="primary")
-
- results_section = Container(id="results-section", classes="section card")
- results_section.border_title = "Scan Output"
-
- with results_section:
- with VerticalScroll(id="results-scroll"):
- self.results = Pretty([], id="results", classes="results")
- yield self.results
-
- def validate_inputs(self, nmap_input: Dict[str, str]) -> List[str]:
- errors: List[str] = []
-
- if self.query_one("#cve-checkbox", Checkbox).value:
- args = nmap_input.get("arguments", "")
- if "-sV" not in args.split():
- errors.append("To check for CVE's, '-sV' option is required")
- return errors
-
- targets = nmap_input.get("targets", "").strip()
- if not targets:
- errors.append("IP field cannot be empty")
- return errors
-
- if not self.query_one("#usercommand", Input).value.strip():
- try:
- ipaddress.ip_network(targets, strict=False)
- except ValueError:
- errors.append(f"Invalid IP address/CIDR: {targets}")
- return errors
-
- if "ports" in nmap_input:
- port_str = nmap_input["ports"]
- if "-" in port_str:
- try:
- low, high = map(int, port_str.split("-"))
- if not (1 <= low <= 65535 and 1 <= high <= 65535):
- errors.append("Ports must be in range 1-65535")
- if low > high:
- errors.append("Start port cannot be greater than end port")
- except ValueError:
- errors.append("Ports (range) must be numbers")
- else:
- try:
- port = int(port_str)
- if not (1 <= port <= 65535):
- errors.append("Port must be in range 1-65535")
- except ValueError:
- errors.append("Port must be a number")
-
- dry_run_err = self.dry_run_nmap_command(nmap_input)
- if dry_run_err:
- errors.append(dry_run_err)
- return errors
-
- def dry_run_nmap_command(self, nmap_input: Dict[str, str]) -> Optional[str]:
- try:
- cmd = ["nmap"]
- if "arguments" in nmap_input:
- cmd += shlex.split(nmap_input["arguments"])
- if "ports" in nmap_input:
- cmd += ["-p", nmap_input["ports"]]
- cmd.append(nmap_input["targets"])
- return None
- except Exception as e:
- return f"Błąd weryfikacji komendy: {e}"
-
- def get_nmap_input(self, return_dict: bool = False) -> list[str] | dict:
- user_command = self.query_one("#usercommand", Input).value.strip()
- low_port = self.query_one("#low-port-range", Input).value.strip()
- high_port = self.query_one("#high-port-range", Input).value.strip()
- interface = self.query_one("#interface-select", Select).value
- targets = self.query_one("#ip", Input).value.strip()
- options = self.query_one(SelectionList).selected
-
- if return_dict:
- if user_command:
- parts = shlex.split(user_command)
-
- if parts and parts[0] == "sudo": parts.pop(0)
- if parts and parts[0] == "nmap": parts.pop(0)
-
- if not parts:
- return {}
-
- target = parts[-1]
- args = parts[:-1]
-
- return {
- "targets": target,
- "arguments": " ".join(args)
- }
- else:
- nmap_dict = {"targets": targets}
-
- if low_port:
- if high_port:
- nmap_dict["ports"] = f"{low_port}-{high_port}"
- else:
- nmap_dict["ports"] = f"{low_port}"
-
- args: list[str] = []
- if options:
- args.extend(options)
- if interface:
- args.extend(["-e", str(interface)])
- if args:
- nmap_dict["arguments"] = " ".join(args)
-
- return {k: v for k, v in nmap_dict.items() if v is not None}
-
- if user_command:
- return shlex.split(user_command)
-
- parts = ["nmap"]
- if interface:
- parts.extend(["-e", str(interface)])
- if options:
- parts.extend(options)
-
- if low_port:
- if high_port:
- parts.extend(["-p", f"{low_port}-{high_port}"])
- else:
- parts.extend(["-p", f"{low_port}"])
-
- parts.append(targets)
- return parts
-
- @on(Mount)
- @on(SelectionList.SelectedChanged)
- @on(Input.Changed)
- @on(Select.Changed)
- def update_selected_view(self, event=None) -> None:
- self.final_command.update(" ".join(self.get_nmap_input()))
-
- @on(Button.Pressed, "#scan-button")
- async def handle_scan_button(self, event: Button.Pressed) -> None:
- nmap_input = self.get_nmap_input(return_dict=True)
- if not isinstance(nmap_input, dict):
- self.notify("Błąd: dane wejściowe nie są słownikiem", severity="error")
- return
- errors = self.validate_inputs(nmap_input)
- if errors:
- self.notify("\n".join(errors), title="Bad Input", severity="error")
- else:
- try:
- temp_prof_name = "temp_scan_profile"
- self.notify("Start scanning", title="Notification", severity="information")
-
- if not self.manager.add_profile(temp_prof_name, notify=False):
- self.notify("Błąd: Nie udało się utworzyć profilu (problem z zapisem pliku?)", severity="error")
- return
-
- if not self.manager.update_profile(temp_prof_name, "nmap_input", nmap_input, notify=False):
- self.notify("Błąd: Nie udało się zaktualizować parametrów skanowania", severity="error")
- self.manager.delete_profile(temp_prof_name, notify=False)
- return
-
- profile = self.manager.get_profile(temp_prof_name)
- if profile:
- scan_now_results = profile.scan()
- self.results.update(scan_now_results)
- self.manager.delete_profile(temp_prof_name, notify=False)
- self.notify("Scanning completed", title="Notification", severity="information")
- else:
- self.notify("Krytyczny błąd: Profil zniknął po utworzeniu", severity="error")
-
- except Exception as e:
- self.notify(f"Scanner error: {e}", title="Error", severity="error")
-
- @on(Button.Pressed, "#save-button")
- async def handle_save_button(self, event: Button.Pressed):
- nmap_input = self.get_nmap_input(return_dict=True)
- if not isinstance(nmap_input, dict):
- self.notify("Błąd: dane wejściowe nie są słownikiem", severity="error")
- return
-
- errors = self.validate_inputs(nmap_input)
- if errors:
- self.notify("\n".join(errors), title="Bad Input", severity="error")
- return
-
- cve_check = self.query_one("#cve-checkbox", Checkbox).value
- self.app.push_screen(SaveProfilePushScreen(self.manager, nmap_input, cve_check))
-
- def on_manager_message(self, msg: str, title: str, severity):
- self.app.notify(message=msg, title=title, severity=severity)
|