Steffen Skui
7336b17f79
Build and push soc-collector / build (push) Successful in 31s
A single Docker image, configured entirely by env vars (SOC_URL + SOC_TOKEN + targets). Runs once and exits; schedule with cron. Four self-skipping sources: - surface nmap open-port/service scan (SOC_TARGETS) - cve Trivy HIGH/CRITICAL scan of running images (Docker socket) - tls cert-expiry / weak-TLS / missing-headers (SOC_DOMAINS) - config 0.0.0.0 binds, --privileged, rw docker-socket mounts (Docker socket) Stable slugs per source so SOC Center versions/diffs/alerts. Pure stdlib; image adds nmap + trivy. MIT, intended as a public Ethica repo.
61 lines
2.5 KiB
Python
61 lines
2.5 KiB
Python
"""Attack-surface source: nmap the targets -> SOC surface[] + findings for risky open ports."""
|
|
import subprocess
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from soc import finding
|
|
|
|
# Ports / service-names that usually should NOT be world-reachable -> severity.
|
|
RISKY_PORT = {
|
|
"23": "high", "2375": "crit", "2376": "high", "3389": "high", "5432": "high",
|
|
"3306": "high", "6379": "high", "27017": "high", "9200": "high", "5984": "high",
|
|
"11211": "high", "2049": "high", "445": "high", "139": "med", "135": "med",
|
|
"5900": "med", "9000": "med", "8006": "high", "61616": "high",
|
|
}
|
|
RISKY_NAME = {
|
|
"telnet": "high", "ms-wbt-server": "high", "postgresql": "high", "mysql": "high",
|
|
"redis": "high", "mongodb": "high", "vnc": "med", "docker": "crit", "memcached": "high",
|
|
"elasticsearch": "high", "rdp": "high",
|
|
}
|
|
|
|
|
|
def run(targets):
|
|
if not targets:
|
|
return None
|
|
proc = subprocess.run(
|
|
["nmap", "-Pn", "-sV", "-T4", "--open", "-oX", "-"] + targets,
|
|
capture_output=True, text=True, timeout=1800)
|
|
if not proc.stdout.strip():
|
|
raise RuntimeError(proc.stderr.strip() or "nmap produced no output")
|
|
root = ET.fromstring(proc.stdout)
|
|
|
|
surface, findings = [], []
|
|
for host in root.findall("host"):
|
|
addr = host.find("address")
|
|
ip = addr.get("addr") if addr is not None else "?"
|
|
open_ports = []
|
|
for p in host.findall("./ports/port"):
|
|
st = p.find("state")
|
|
if st is None or st.get("state") != "open":
|
|
continue
|
|
num = p.get("portid")
|
|
svc = p.find("service")
|
|
name = svc.get("name") if svc is not None else ""
|
|
open_ports.append(num)
|
|
sev = RISKY_PORT.get(num) or RISKY_NAME.get(name)
|
|
if sev:
|
|
findings.append(finding(
|
|
f"surface-{ip}-{num}", sev,
|
|
f"Exposed {name or 'service'} on port {num}", host=ip,
|
|
evidence=f"{ip}:{num} ({name or 'unknown'})",
|
|
impact=["reachable over the network"],
|
|
fix=[f"restrict port {num} to trusted networks, bind to localhost, "
|
|
"or front it with the reverse proxy"]))
|
|
surface.append({"ip": ip, "role": "", "ports": ",".join(open_ports)})
|
|
|
|
return {
|
|
"scope": "nmap " + " ".join(targets),
|
|
"surface": surface,
|
|
"findings": findings,
|
|
"good": [] if findings else ["No risky ports found open on the scanned targets"],
|
|
}
|