Seldon Core - Detect

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Seldon Core, an open-source MLOps/LLMOps framework for deploying, monitoring, and scaling machine learning mod."""

import re

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


class Module(Scanner, Http_client):
    __info__ = {
        'name': 'Seldon Core - Detect',
        'description': 'Seldon Core, an open-source MLOps/LLMOps framework for deploying, monitoring, and scaling machine learning models on Kubernetes, was detected. Its Pipeline Gateway component, which routes inference requests between chained model pipelines, exposed an unauthenticated status endpoint returning a fixed JSON banner and a matching Server response header. Exposed instances may have revealed internal ML pipeline configuration and provided a foothold for further probing of the inference API.',
        'author': ['KittySploit Team'],
        'severity': 'info',
        'tags': ['web', 'scanner', 'technology', 'tech', 'seldon-core', 'ml', 'ai', 'mlops'],
        'agent': {
            'risk': 'active',
            'effects': ['network_probe'],
            'expected_requests': 1,
            'reversible': True,
            'approval_required': False,
            'produces': ['tech_hints', 'risk_signals', 'endpoints'],
            'cost': 1.0,
            'noise': 0.3,
            'value': 1.0,
            'requires': {
                'min_endpoints': 0,
                'min_params': 0,
                'tech_hints_any': [],
                'tech_hints_all': [],
                'specializations_any': [],
                'risk_signals_any': [],
                'auth_session': False,
                'capabilities_any': [],
                'capabilities_all': [],
                'confidence_min': {},
                'confidence_min_any': {},
                'endpoint_pattern_any': [],
                'param_any': [],
                'api_surface_ready': False,
            },
            'chain': {
                'produces_capabilities': [
                    {
                        'capability': 'admin_surface',
                        'from_detail': '',
                    },
                ],
                'consumes_capabilities': [],
                'option_bindings': {},
                'suggested_followups': ['auxiliary/scanner/http/login_page_detector'],
            },
        },
        'references': ['https://github.com/SeldonIO/seldon-core', 'https://docs.seldon.ai/seldon-core-2'],
    }

    def run(self):
        r = self.http_request(method="GET", path='/', allow_redirects=False)
        if not r or r.status_code != 200:
            return False
        body = r.text or ""
        body_regexes = ('"name":\\s*"seldon-pipelinegateway"',)
        body_re_hit = any(re.search(rx, body, 0) for rx in body_regexes)
        headers = "\n".join(f"{k}: {v}" for k, v in r.headers.items())
        header_markers = ('seldon-pipelinegateway',)
        if body_re_hit:
            self.set_info(
                severity='info',
                reason="Seldon Core detected",
                path='/',
            )
            return True
        return False