#!/usr/bin/env python3 """ CVE-2026-42826 Detection & Hardening Checker Checks Azure DevOps instances for information disclosure vulnerability indicators """ import argparse import json import sys from datetime import datetime from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError import base64 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, "OK": Colors.GREEN, "WARN": Colors.YELLOW, "VULN": Colors.RED, "BOLD": Colors.BOLD} c = colors.get(level, Colors.BLUE) print(f"[{ts}] {c}[{level}]{Colors.RESET} {msg}") class DetectionChecker: def __init__(self, org_url, pat=None, api_version="7.1", timeout=15): self.org_url = org_url.rstrip('/') self.pat = pat self.api_version = api_version self.timeout = timeout self.results = [] def _request(self, url): headers = {"Accept": "application/json"} if self.pat: auth = base64.b64encode(f":{self.pat}".encode()).decode() headers["Authorization"] = f"Basic {auth}" req = Request(url, headers=headers) try: with urlopen(req, timeout=self.timeout) as resp: return resp.getcode(), json.loads(resp.read().decode()) except HTTPError as e: try: return e.code, json.loads(e.read().decode()) except: return e.code, None except Exception as e: return None, str(e) def record(self, check, status, detail=""): icon = {"VULN": f"{Colors.RED}✗", "OK": f"{Colors.GREEN}✓", "WARN": f"{Colors.YELLOW}!"}.get(status, "?") print(f" {icon}{Colors.RESET} {check}: {detail}") self.results.append({"check": check, "status": status, "detail": detail}) def check_unauthenticated_access(self): """Check 1: Unauthenticated API access""" log("Check 1: Unauthenticated API Access", "BOLD") url = f"{self.org_url}/_apis/projects?api-version={self.api_version}" # Test without auth req = Request(url, headers={"Accept": "application/json"}) try: with urlopen(req, timeout=self.timeout) as resp: data = json.loads(resp.read().decode()) projects = data.get("value", []) public = [p for p in projects if p.get("visibility") == "public"] if public: self.record("Unauthenticated API access", "VULN", f"{len(public)} public project(s) accessible without auth") else: self.record("Unauthenticated API access", "OK", f"{len(projects)} projects accessible but none public") except HTTPError as e: if e.code == 401: self.record("Unauthenticated API access", "OK", "Properly returns 401") else: self.record("Unauthenticated API access", "WARN", f"HTTP {e.code}") except Exception as e: self.record("Unauthenticated API access", "WARN", str(e)) def check_public_projects(self): """Check 2: Public project visibility""" log("Check 2: Public Project Visibility", "BOLD") if not self.pat: log(" Skipping — requires PAT", "WARN") return url = f"{self.org_url}/_apis/projects?api-version={self.api_version}" status, data = self._request(url) if status == 200 and data: projects = data.get("value", []) public = [p for p in projects if p.get("visibility") == "public"] if public: self.record("Public projects", "VULN", f"{len(public)} public: {', '.join(p['name'] for p in public)}") else: self.record("Public projects", "OK", f"All {len(projects)} projects are private") def check_org_policy(self): """Check 3: Organization policy for public projects""" log("Check 3: Organization Policy", "BOLD") if not self.pat: log(" Skipping — requires PAT", "WARN") return url = f"{self.org_url}/_apis/orgpolicy/policies/ProjectVisibility?api-version={self.api_version}" status, data = self._request(url) if status == 200 and data: policy = data.get("value", [{}])[0] if data.get("value") else data allows_public = policy.get("allowValue", True) if allows_public: self.record("Org allows public projects", "WARN", "Public projects are allowed") else: self.record("Org allows public projects", "OK", "Public projects are blocked") else: self.record("Org policy check", "WARN", f"Cannot determine (status={status})") def check_variable_groups(self): """Check 4: Non-secret variables in variable groups""" log("Check 4: Variable Group Secrets", "BOLD") if not self.pat: log(" Skipping — requires PAT", "WARN") return url = f"{self.org_url}/_apis/projects?api-version={self.api_version}" status, data = self._request(url) if status != 200 or not data: return for project in data.get("value", [])[:5]: # Check first 5 projects proj_name = project.get("name") vg_url = f"{self.org_url}/{proj_name}/_apis/distributedtask/variablegroups?api-version={self.api_version}" s2, d2 = self._request(vg_url) if s2 == 200 and d2: for vg in d2.get("value", []): variables = vg.get("variables", {}) non_secret = [k for k, v in variables.items() if not (v.get("isSecret", False) or v.get("secret", False)) and v.get("value")] if non_secret: self.record(f"VG '{vg.get('name')}' in {proj_name}", "WARN", f"{len(non_secret)} non-secret variable(s) with values") def check_service_connections(self): """Check 5: Service connection exposure""" log("Check 5: Service Connection Exposure", "BOLD") if not self.pat: log(" Skipping — requires PAT", "WARN") return url = f"{self.org_url}/_apis/projects?api-version={self.api_version}" status, data = self._request(url) if status != 200 or not data: return for project in data.get("value", [])[:5]: proj_name = project.get("name") sc_url = f"{self.org_url}/{proj_name}/_apis/serviceendpoint/endpoints?api-version={self.api_version}" s2, d2 = self._request(sc_url) if s2 == 200 and d2: for sc in d2.get("value", []): allow_all = sc.get("data", {}).get("allowAllPipelines", False) if isinstance(sc.get("data"), dict) else False if allow_all: self.record(f"Service connection '{sc.get('name')}' in {proj_name}", "WARN", "Allow all pipelines — any pipeline can use this connection") def run(self): log("=" * 60, "BOLD") log(" CVE-2026-42826 Detection & Hardening Checker", "BOLD") log(f" Target: {self.org_url}", "BOLD") log("=" * 60, "BOLD") self.check_unauthenticated_access() self.check_public_projects() self.check_org_policy() self.check_variable_groups() self.check_service_connections() # Summary vuln_count = sum(1 for r in self.results if r["status"] == "VULN") warn_count = sum(1 for r in self.results if r["status"] == "WARN") ok_count = sum(1 for r in self.results if r["status"] == "OK") log("\n" + "=" * 60, "BOLD") log(f" Results: {Colors.RED}{vuln_count} VULN{Colors.RESET}, " f"{Colors.YELLOW}{warn_count} WARN{Colors.RESET}, " f"{Colors.GREEN}{ok_count} OK{Colors.RESET}", "BOLD") log("=" * 60, "BOLD") if vuln_count > 0: log(f"\n {Colors.RED}VULNERABLE — immediate remediation required{Colors.RESET}", "VULN") elif warn_count > 0: log(f"\n {Colors.YELLOW}WARN — hardening recommended{Colors.RESET}", "WARN") else: log(f"\n {Colors.GREEN}OK — no issues detected{Colors.RESET}", "OK") return vuln_count > 0 def main(): parser = argparse.ArgumentParser( description="CVE-2026-42826 Detection & Hardening Checker for Azure DevOps") parser.add_argument("-u", "--url", required=True, help="Azure DevOps organization URL") parser.add_argument("--pat", help="Personal Access Token (for authenticated checks)") parser.add_argument("--api-version", default="7.1") parser.add_argument("--timeout", type=int, default=15) args = parser.parse_args() checker = DetectionChecker(args.url, args.pat, args.api_version, args.timeout) is_vuln = checker.run() sys.exit(0 if is_vuln else 1) if __name__ == "__main__": main()