MCPcopy Create free account
hub / github.com/TheSecuredAnalyst/security-suite / run

Method run

modules/webscanner/nuclei.py:40–137  ·  view source on GitHub ↗

Run Nuclei scan on target.

(self, target: Target)

Source from the content-addressed store, hash-verified

38
39 async def run(self, target: Target) -> ScanResult:
40 """Run Nuclei scan on target."""
41 result = self.create_result(target)
42
43 if not self.nuclei_available:
44 result.errors.append("Nuclei not found. Install from https://github.com/projectdiscovery/nuclei")
45 result.success = False
46 result.complete()
47 return result
48
49 url = self._build_url(target)
50 self.logger.info(f"Starting Nuclei scan on {url}")
51
52 with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
53 output_file = f.name
54
55 try:
56 cmd = [
57 "nuclei",
58 "-u", url,
59 "-json-export", output_file,
60 "-silent",
61 ]
62
63 # Add severity filter
64 if self.severity_filter:
65 cmd.extend(["-severity", ",".join(self.severity_filter)])
66
67 # Add specific templates
68 if self.templates:
69 for t in self.templates:
70 cmd.extend(["-t", t])
71
72 # Add tags
73 if self.tags:
74 cmd.extend(["-tags", ",".join(self.tags)])
75
76 self.logger.info(f"Running: {' '.join(cmd)}")
77
78 proc = await asyncio.create_subprocess_exec(
79 *cmd,
80 stdout=asyncio.subprocess.PIPE,
81 stderr=asyncio.subprocess.PIPE,
82 )
83
84 stdout, stderr = await proc.communicate()
85
86 # Parse results
87 findings = []
88 if os.path.exists(output_file) and os.path.getsize(output_file) > 0:
89 with open(output_file) as f:
90 for line in f:
91 try:
92 finding = json.loads(line.strip())
93 findings.append(finding)
94 except json.JSONDecodeError:
95 continue
96
97 result.raw_data["findings"] = findings

Callers 1

scan_nucleiFunction · 0.95

Calls 5

_build_urlMethod · 0.95
completeMethod · 0.80
add_findingMethod · 0.80
create_resultMethod · 0.45
getMethod · 0.45

Tested by

no test coverage detected