Untitled
unknown
plain_text
10 months ago
9.1 kB
18
Indexable
#!/usr/bin/env python3
"""
Exim CVE-2025-26794 Safe Scanner
Purpose: Remote, non-exploitative scanner that assesses *indicators of exposure* for CVE-2025-26794.
It never sends harmful payloads. It only checks banner/version, STARTTLS capability,
and whether ETRN appears to be accepted. SQLite hints backend cannot be confirmed remotely.
Usage examples:
python exim_cve_2025_26794_scanner.py --targets targets.txt
python exim_cve_2025_26794_scanner.py --host mx.example.com --ports 25,587,465
python exim_cve_2025_26794_scanner.py --host mx.example.com --timeout 6 --no-starttls
Outputs:
- scan_results.json / scan_results.csv in the current directory
"""
import argparse
import csv
import json
import re
import socket
import ssl
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from typing import Dict, List, Optional, Tuple
DEFAULT_PORTS = [25, 587, 465]
SMTP_EOL = b"\r\n"
BANNER_EXIM_RE = re.compile(r"Exim\s+(\d+)\.(\d+)(?:\.(\d+))?", re.IGNORECASE)
def parse_args():
p = argparse.ArgumentParser(description="Safe scanner for Exim CVE-2025-26794 (no exploit).")
g = p.add_mutually_exclusive_group(required=True)
g.add_argument("--host", help="Single host to scan (hostname or IP)")
g.add_argument("--targets", help="File with hosts (one per line; # for comments)")
p.add_argument("--ports", default="25,587,465", help="Comma-separated ports to try (default: 25,587,465)")
p.add_argument("--timeout", type=float, default=5.0, help="Socket timeout seconds (default: 5)")
p.add_argument("--threads", type=int, default=32, help="Max concurrent threads (default: 32)")
p.add_argument("--no-starttls", action="store_true", help="Do not attempt STARTTLS upgrade on 25/587")
p.add_argument("--json", default="scan_results.json", help="JSON output file (default: scan_results.json)")
p.add_argument("--csv", default="scan_results.csv", help="CSV output file (default: scan_results.csv)")
return p.parse_args()
def read_line(fobj, timeout: float) -> Optional[str]:
try:
line = fobj.readline()
if not line:
return None
return line.decode(errors="ignore").rstrip("\r\n")
except Exception:
return None
def read_multiline(fobj, first_line: str) -> List[str]:
lines = [first_line]
m = re.match(r"^(\d{3})-(.*)", first_line or "")
if not m:
return lines
code = m.group(1)
while True:
nxt = read_line(fobj, timeout=0.5)
if not nxt:
break
lines.append(nxt)
if re.match(rf"^{code}\s", nxt):
break
return lines
def smtp_connect(host: str, port: int, timeout: float, use_implicit_tls: bool = False):
s = socket.create_connection((host, port), timeout=timeout)
if use_implicit_tls:
ctx = ssl.create_default_context()
s = ctx.wrap_socket(s, server_hostname=host if not re.match(r'^\d+\.\d+\.\d+\.\d+$', host) else None)
f = s.makefile("rb", buffering=0)
return s, f
def smtp_send(sock: socket.socket, data: str):
sock.sendall(data.encode() + SMTP_EOL)
def parse_exim_version_from_banner(lines: List[str]) -> Optional[Tuple[int, int, Optional[int]]]:
joined = " ".join(lines)
m = BANNER_EXIM_RE.search(joined)
if not m:
return None
major, minor, patch = m.group(1), m.group(2), m.group(3)
return int(major), int(minor), int(patch) if patch is not None else None
def version_lt(a: Tuple[int, int, Optional[int]], b: Tuple[int, int, Optional[int]]) -> bool:
a_major, a_minor, a_patch = a[0], a[1], a[2] if a[2] is not None else -1
b_major, b_minor, b_patch = b[0], b[1], b[2] if b[2] is not None else -1
return (a_major, a_minor, a_patch) < (b_major, b_minor, b_patch)
def need_attention_for_cve(exim_version: Optional[Tuple[int, int, Optional[int]]]) -> Optional[bool]:
if not exim_version:
return None
fixed = (4, 98, 1)
return version_lt(exim_version, fixed)
def do_scan(host: str, port: int, timeout: float, attempt_starttls: bool) -> Dict:
result = {
"host": host,
"port": port,
"implicit_tls": (port == 465),
"banner": None,
"exim_version": None,
"supports_starttls": None,
"etrn_supported": None,
"etrn_response": None,
"cve_risk_hint": None,
"notes": [],
"error": None,
}
try:
s, f = smtp_connect(host, port, timeout, use_implicit_tls=(port == 465))
first = read_line(f, timeout)
if not first:
raise RuntimeError("No SMTP banner")
lines = read_multiline(f, first)
result["banner"] = " | ".join(lines)
exim_ver = parse_exim_version_from_banner(lines)
result["exim_version"] = ".".join(str(x) for x in exim_ver if x is not None) if exim_ver else None
smtp_send(s, "EHLO scanner.local")
ehlo_first = read_line(f, timeout) or ""
ehlo_lines = read_multiline(f, ehlo_first)
ehlo_text = " ".join(ehlo_lines).lower()
result["supports_starttls"] = ("starttls" in ehlo_text)
if attempt_starttls and (port in (25, 587)) and result["supports_starttls"]:
smtp_send(s, "STARTTLS")
starttls_resp = read_line(f, timeout) or ""
if starttls_resp.startswith("220"):
ctx = ssl.create_default_context()
s = ctx.wrap_socket(s, server_hostname=host if not re.match(r'^\d+\.\d+\.\d+\.\d+$', host) else None)
f = s.makefile("rb", buffering=0)
smtp_send(s, "EHLO scanner.local")
ehlo_first = read_line(f, timeout) or ""
ehlo_lines = read_multiline(f, ehlo_first)
smtp_send(s, "ETRN test.example")
etrn_resp = read_line(f, timeout) or ""
result["etrn_response"] = etrn_resp
result["etrn_supported"] = etrn_resp.startswith("250") or etrn_resp.startswith("251") or etrn_resp.startswith("252")
vuln_hint = need_attention_for_cve(exim_ver)
if vuln_hint is None and result["etrn_supported"] is None:
risk = "UNKNOWN"
elif vuln_hint and result["etrn_supported"]:
risk = "HIGH"
elif (vuln_hint and not result["etrn_supported"]) or ((vuln_hint is None) and result["etrn_supported"]):
risk = "MEDIUM"
else:
risk = "LOW"
result["cve_risk_hint"] = risk
try:
smtp_send(s, "QUIT")
read_line(f, 0.5)
except Exception:
pass
s.close()
except Exception as e:
result["error"] = str(e)
return result
def load_targets(args) -> List[str]:
if args.host:
return [args.host.strip()]
hosts = []
with open(args.targets, "r", encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line or line.startswith("#"):
continue
hosts.append(line)
return hosts
def main():
args = parse_args()
ports = []
try:
for p in args.ports.split(","):
p = p.strip()
if not p:
continue
ports.append(int(p))
except Exception:
print("Invalid --ports value", file=sys.stderr)
sys.exit(2)
hosts = load_targets(args)
attempt_starttls = not args.no_starttls
jobs = []
results = []
started = datetime.utcnow().isoformat() + "Z"
with ThreadPoolExecutor(max_workers=args.threads) as ex:
for h in hosts:
for port in ports:
jobs.append(ex.submit(do_scan, h, port, args.timeout, attempt_starttls))
for fut in as_completed(jobs):
results.append(fut.result())
results.sort(key=lambda r: (r["host"], r["port"]))
with open(args.json, "w", encoding="utf-8") as jf:
json.dump({
"started": started,
"finished": datetime.utcnow().isoformat() + "Z",
"targets": hosts,
"results": results
}, jf, indent=2, ensure_ascii=False)
with open(args.csv, "w", newline="", encoding="utf-8") as cf:
writer = csv.writer(cf)
writer.writerow([
"host","port","implicit_tls","banner","exim_version",
"supports_starttls","etrn_supported","etrn_response","cve_risk_hint","error"
])
for r in results:
writer.writerow([
r["host"], r["port"], r["implicit_tls"], r["banner"] or "",
r["exim_version"] or "", r["supports_starttls"], r["etrn_supported"],
r["etrn_response"] or "", r["cve_risk_hint"] or "", r["error"] or ""
])
print(f"[+] Done. Wrote {args.json} and {args.csv}")
print("Risk legend: HIGH (version<4.98.1 && ETRN accepted), MEDIUM (one signal), LOW (fixed or ETRN disabled), UNKNOWN (insufficient signals).")
print("Note: SQLite hints backend cannot be verified remotely; confirm locally (exim -bV, config).")
if __name__ == "__main__":
main()Editor is loading...
Leave a Comment