asepharyana
f6389018f5
- Add `zesdex_sec_daemon` module with main entry point for running the security daemon. - Implement `TieredInstaller` for installing security tools from various sources (pip, binaries, gems). - Create a newline-delimited JSON frame protocol for communication between the daemon and tools. - Introduce a `ToolRegistry` for managing and dispatching tool executions. - Add various tools including HTTP, SQLMap, Nuclei, and more with their respective execution logic. - Establish health check and installation commands for tool management. - Include prompts for classifier and quality reviewer to enhance code review and safety checks. - Document the system's tools and guidelines for usage.
242 lines
9.2 KiB
Python
242 lines
9.2 KiB
Python
"""Tiered installer for security sidecar tools.
|
|
|
|
Installation tiers:
|
|
pip — Python packages (pycryptodome, pwntools, factordb-python, etc.)
|
|
binary — Pre-built GitHub release binaries (nuclei, ffuf, dalfox, etc.)
|
|
gem — Ruby gems (zap-cli, etc.)
|
|
detect — Manual-detect-only (z3, sage, etc. — user must install themselves)
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import platform
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import urllib.request
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
@dataclass
|
|
class InstallResult:
|
|
ok: bool = False
|
|
message: str = ""
|
|
installed: list[str] = field(default_factory=list)
|
|
failed: list[str] = field(default_factory=list)
|
|
skipped: list[str] = field(default_factory=list)
|
|
|
|
|
|
class TieredInstaller:
|
|
"""Multi-tier installer for security tools.
|
|
|
|
Detects the current platform and tries each tier in order of
|
|
preference: pip -> binary -> gem -> detect (manual only).
|
|
"""
|
|
|
|
def __init__(self, target_dir: Optional[str] = None):
|
|
self.target_dir = Path(target_dir or self._default_bin_dir())
|
|
self.target_dir.mkdir(parents=True, exist_ok=True)
|
|
self._pip_available = shutil.which("pip3") is not None or shutil.which("pip") is not None
|
|
self._gem_available = shutil.which("gem") is not None
|
|
self._arch = platform.machine()
|
|
self._os = platform.system().lower()
|
|
|
|
def _default_bin_dir(self) -> str:
|
|
if self._os == "linux":
|
|
return "/usr/local/bin"
|
|
return os.path.expanduser("~/.local/bin")
|
|
|
|
# ── pip tier ────────────────────────────────────────────────────
|
|
|
|
def _pip_install(self, pkg: str) -> bool:
|
|
pip = shutil.which("pip3") or shutil.which("pip")
|
|
if not pip:
|
|
return False
|
|
try:
|
|
result = subprocess.run(
|
|
[pip, "install", "--quiet", pkg],
|
|
capture_output=True, text=True, timeout=120,
|
|
)
|
|
return result.returncode == 0
|
|
except subprocess.TimeoutExpired:
|
|
return False
|
|
except Exception:
|
|
return False
|
|
|
|
# ── GitHub release binary tier ──────────────────────────────────
|
|
|
|
def _download_gh_release(
|
|
self, repo: str, asset_pattern: str,
|
|
) -> Optional[Path]:
|
|
api_url = f"https://api.github.com/repos/{repo}/releases/latest"
|
|
try:
|
|
req = urllib.request.Request(api_url, headers={
|
|
"Accept": "application/json",
|
|
"User-Agent": "zesdex-security-daemon/0.1.0",
|
|
})
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
data = json.loads(resp.read().decode())
|
|
assets = data.get("assets", [])
|
|
os_arch_tag = f"{self._os}_{self._arch}"
|
|
for asset in assets:
|
|
name = asset["name"]
|
|
if asset_pattern.replace("{os_arch}", os_arch_tag) in name:
|
|
download_url = asset["browser_download_url"]
|
|
break
|
|
else:
|
|
# Try without os_arch matching, just look for the pattern.
|
|
candidates = [a for a in assets if asset_pattern.split("/")[0] in a["name"]]
|
|
if not candidates:
|
|
return None
|
|
download_url = candidates[0]["browser_download_url"]
|
|
|
|
with tempfile.NamedTemporaryFile(suffix=".tmp", delete=False) as tmp:
|
|
with urllib.request.urlopen(download_url, timeout=120) as dl:
|
|
tmp.write(dl.read())
|
|
tmppath = tmp.name
|
|
|
|
if download_url.endswith(".tar.gz"):
|
|
extract_dir = tempfile.mkdtemp()
|
|
with tarfile.open(tmppath, "r:gz") as tar:
|
|
tar.extractall(path=extract_dir)
|
|
os.unlink(tmppath)
|
|
# Find the binary in extracted files.
|
|
for root, _dirs, files in os.walk(extract_dir):
|
|
for fname in files:
|
|
if fname == asset_pattern.split("/")[0].replace(".tar.gz", ""):
|
|
return Path(root) / fname
|
|
return None
|
|
else:
|
|
result = Path(tmppath)
|
|
result.chmod(0o755)
|
|
return result
|
|
except Exception:
|
|
return None
|
|
|
|
def _install_binary(self, name: str, repo: str, asset_pattern: str) -> bool:
|
|
downloaded = self._download_gh_release(repo, asset_pattern)
|
|
if downloaded is None:
|
|
return False
|
|
dest = self.target_dir / name
|
|
try:
|
|
shutil.move(str(downloaded), str(dest))
|
|
dest.chmod(0o755)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
# ── Gem tier ────────────────────────────────────────────────────
|
|
|
|
def _gem_install(self, gem_name: str) -> bool:
|
|
if not self._gem_available:
|
|
return False
|
|
try:
|
|
result = subprocess.run(
|
|
["gem", "install", "--quiet", gem_name],
|
|
capture_output=True, text=True, timeout=120,
|
|
)
|
|
return result.returncode == 0
|
|
except Exception:
|
|
return False
|
|
|
|
# ── Tool-specific installers ────────────────────────────────────
|
|
|
|
def _install_pip_tools(self) -> list[str]:
|
|
pkgs = [
|
|
"pycryptodome",
|
|
"pwntools",
|
|
"requests",
|
|
"factordb-python",
|
|
]
|
|
installed = []
|
|
for pkg in pkgs:
|
|
if self._pip_install(pkg):
|
|
installed.append(pkg)
|
|
return installed
|
|
|
|
def _install_binary_tools(self) -> list[str]:
|
|
targets = [
|
|
("nuclei", "projectdiscovery/nuclei", "nuclei_{os_arch}.tar.gz"),
|
|
("ffuf", "ffuf/ffuf", "ffuf_{os_arch}.tar.gz"),
|
|
("dalfox", "hahwul/dalfox", "dalfox_{os_arch}.tar.gz"),
|
|
("httpx", "projectdiscovery/httpx", "httpx_{os_arch}.tar.gz"),
|
|
]
|
|
installed = []
|
|
for name, repo, pattern in targets:
|
|
if shutil.which(name) is not None:
|
|
installed.append(name)
|
|
continue
|
|
if self._install_binary(name, repo, pattern):
|
|
installed.append(name)
|
|
return installed
|
|
|
|
def _install_gem_tools(self) -> list[str]:
|
|
installed = []
|
|
if self._gem_install("zap-cli"):
|
|
installed.append("zap-cli")
|
|
return installed
|
|
|
|
# ── Public API ──────────────────────────────────────────────────
|
|
|
|
def install_all(self) -> InstallResult:
|
|
result = InstallResult()
|
|
try:
|
|
pip_ok = self._install_pip_tools()
|
|
result.installed.extend(pip_ok)
|
|
except Exception as e:
|
|
result.failed.append(f"pip: {e}")
|
|
|
|
try:
|
|
bin_ok = self._install_binary_tools()
|
|
result.installed.extend(bin_ok)
|
|
except Exception as e:
|
|
result.failed.append(f"binary: {e}")
|
|
|
|
try:
|
|
gem_ok = self._install_gem_tools()
|
|
result.installed.extend(gem_ok)
|
|
except Exception as e:
|
|
result.failed.append(f"gem: {e}")
|
|
|
|
result.ok = True
|
|
result.message = (
|
|
f"Installed {len(result.installed)} tool(s): "
|
|
f"{', '.join(result.installed)}"
|
|
)
|
|
return result
|
|
|
|
def install_tool(self, tool_name: str) -> dict:
|
|
"""Install a single tool by name. Returns {"ok": bool, "message": str}."""
|
|
pip_map = {
|
|
"pycryptodome": "pycryptodome",
|
|
"pwntools": "pwntools",
|
|
"factordb": "factordb-python",
|
|
"requests": "requests",
|
|
}
|
|
binary_map = {
|
|
"nuclei": ("projectdiscovery/nuclei", "nuclei_{os_arch}.tar.gz"),
|
|
"ffuf": ("ffuf/ffuf", "ffuf_{os_arch}.tar.gz"),
|
|
"dalfox": ("hahwul/dalfox", "dalfox_{os_arch}.tar.gz"),
|
|
"httpx": ("projectdiscovery/httpx", "httpx_{os_arch}.tar.gz"),
|
|
"sqlmap": ("sqlmapproject/sqlmap", "sqlmap.tar.gz"),
|
|
}
|
|
gem_map = {
|
|
"zap-cli": "zap-cli",
|
|
}
|
|
|
|
if tool_name in pip_map:
|
|
ok = self._pip_install(pip_map[tool_name])
|
|
return {"ok": ok, "message": f"pip install {pip_map[tool_name]}: {'ok' if ok else 'failed'}"}
|
|
if tool_name in binary_map:
|
|
repo, pattern = binary_map[tool_name]
|
|
ok = self._install_binary(tool_name, repo, pattern)
|
|
return {"ok": ok, "message": f"binary install {tool_name}: {'ok' if ok else 'failed'}"}
|
|
if tool_name in gem_map:
|
|
ok = self._gem_install(gem_map[tool_name])
|
|
return {"ok": ok, "message": f"gem install {gem_map[tool_name]}: {'ok' if ok else 'failed'}"}
|
|
return {"ok": False, "message": f"no installer available for {tool_name}; may need manual install"}
|