Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-27 02:38
📄 __init__.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from cl_website_collector.website_collector import WebsiteCollector from cl_website_collector.docroot_processor import DocrootProcessor from cl_website_collector.apache_processor import ApacheProcessor __all__ = [ "WebsiteCollector", "DocrootProcessor", "ApacheProcessor", ]	494 B	2026-02-17 10:40
📄 __main__.py # -- coding: utf-8 -- # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT """ CloudLinux website data collector utility Collects Apache configs, system info, and .htaccess files from websites. Note: Data collection and sending is integrated into cloudlinux-summary. """ import argparse import logging import sys from cl_website_collector.website_collector import WebsiteCollector def setup_logging(verbose: bool = False) -> None: """Setup logging configuration.""" log_level = logging.DEBUG if verbose else logging.INFO app_logger = logging.getLogger('cloudlinux-website-collector') app_logger.setLevel(log_level) stream_handler = logging.StreamHandler(sys.stdout) formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') stream_handler.setFormatter(formatter) app_logger.addHandler(stream_handler) def parse_arguments() -> argparse.Namespace: """Parse command line arguments.""" parser = argparse.ArgumentParser( description='CloudLinux website data collector', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__ ) parser.add_argument( '--send', action='store_true', help='Send data via API (requires --api-url)' ) parser.add_argument( '--api-url', type=str, default='https://stat-api.cloudlinux.com/api/clos-upload', help='API URL for sending data (default: %(default)s)' ) parser.add_argument( '--system-id', type=str, default='test-system-id', help='System identifier (default: %(default)s)' ) parser.add_argument( '--client-id', type=str, default='000000', help='Client identifier (default: %(default)s)' ) parser.add_argument( '--platform', type=str, default='rhel_like', help='Platform type (default: %(default)s)' ) parser.add_argument( '--panel', type=str, default='cPanel', help='Control panel name (default: %(default)s)' ) parser.add_argument( '--server', type=str, default='apache', help='Server name (default: %(default)s)' ) parser.add_argument( '--dry-run', action='store_true', help='Save data to dry-run.jsonl file instead of sending to server' ) parser.add_argument( '--verbose', '-v', action='store_true', help='Enable verbose logging' ) return parser.parse_args() def main() -> int: """Main entry point.""" args = parse_arguments() # Setup logging setup_logging(args.verbose) logger = logging.getLogger("cloudlinux-website-collector") if not args.send: logger.info("No action specified.") return 1 try: collector = WebsiteCollector(logger=logger) if args.send: collector.send_data( system_id=args.system_id, client_id=args.client_id, platform=args.platform, panel=args.panel, server=args.server, api_url=args.api_url, remote_config_version="dummy_version", # no sense to specify there some real version dry_run=args.dry_run ) logger.info("Data sending completed") return 0 except KeyboardInterrupt: logger.info("Operation interrupted by user") return 130 except Exception as e: logger.error("Operation failed: %s", e) return 1 if __name__ == "__main__": sys.exit(main())	3.7 KB	2026-02-17 10:40
📁 __pycache__	-	2026-03-27 02:38
📄 apache_processor.py # -- coding: utf-8 -- # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import logging import os import subprocess import shutil from typing import Dict, List, Optional, Any class ApacheProcessor: """ Collects Apache configuration and module information. """ def __init__(self, logger: logging.Logger): # General Apache configuration directories (exclude user-specific subdirs) self.apache_config_locations = [ '/etc/apache2/conf/', '/etc/apache2/conf-available/', '/etc/apache2/conf.d/', # Will be filtered to exclude userdata subdirs '/etc/apache2/conf.modules.d/', ] # User-specific configuration base paths (Apache versions 2.2 and 2.4) self.user_config_base_paths = [ '/etc/apache2/conf.d/userdata/ssl/2_4', # SSL configs for Apache 2.4 '/etc/apache2/conf.d/userdata/std/2_4', # Standard configs for Apache 2.4 '/etc/apache2/conf.d/userdata/ssl/2_2', # SSL configs for Apache 2.2 (legacy) '/etc/apache2/conf.d/userdata/std/2_2', # Standard configs for Apache 2.2 (legacy) ] self.logger = logger self._apache_binary_cache = None self._apache_version_cache = None self._apache_modules_cache = None def _find_apache_binary(self) -> Optional[str]: """ Find Apache binary executable using which command with caching. """ if self._apache_binary_cache is not None: return self._apache_binary_cache error = None try: for binary_name in ['httpd', 'apache2']: binary_path = shutil.which(binary_name) if binary_path and os.access(binary_path, os.X_OK): self.logger.debug("Found Apache binary: %s", binary_path) self._apache_binary_cache = binary_path return binary_path except Exception as e: error = e self.logger.debug("Apache binary not found in PATH %s", error) self._apache_binary_cache = None return None def _get_apache_version(self) -> str: """ Get Apache version information with caching. """ if self._apache_version_cache is not None: return self._apache_version_cache apache_binary = self._find_apache_binary() if not apache_binary: self._apache_version_cache = "Apache binary not found" return self._apache_version_cache try: result = subprocess.run( [apache_binary, '-v'], capture_output=True, text=True, timeout=10, check=False ) if result.returncode == 0: # Extract just the version line for cleaner metadata version_output = result.stdout.strip() if version_output: # Get first line which contains version info first_line = version_output.split('\n')[0] self._apache_version_cache = first_line else: self._apache_version_cache = "Version info empty" else: self._apache_version_cache = f"Error getting version: {result.stderr.strip()}" except subprocess.TimeoutExpired: self._apache_version_cache = "Version check timed out" except Exception as e: self._apache_version_cache = f"Version check failed: {e}" return self._apache_version_cache def _get_apache_modules(self) -> List[str]: """ Get Apache modules using httpd -M command with caching. Returns: List of module names (e.g., ['core_module', 'so_module', 'http_module', ...]) """ # Return cached modules if available if self._apache_modules_cache is not None: return self._apache_modules_cache apache_binary = self._find_apache_binary() if not apache_binary: self.logger.debug("Apache binary not found, cannot get modules") self._apache_modules_cache = [] return self._apache_modules_cache try: result = subprocess.run( [apache_binary, '-M'], capture_output=True, text=True, timeout=10, check=False ) if result.returncode == 0: modules = [] lines = result.stdout.split('\n') # Skip first line 'Loaded Modules:' and process module lines for line in lines[1:]: if not line.strip(): continue # Parse line like: "core_module (static)" or "so_module (shared)" try: module_name = line.strip().split(' ')[0] if module_name: modules.append(module_name) except (IndexError, AttributeError): continue # Skip malformed lines self.logger.debug("Found %d Apache modules via %s -M", len(modules), apache_binary) self._apache_modules_cache = modules return modules self.logger.debug("Apache modules command failed: %s", result.stderr.strip()) self._apache_modules_cache = [] return self._apache_modules_cache except subprocess.TimeoutExpired: self.logger.error("[WEBSITE-COLLECTOR] Apache modules command timed out") self._apache_modules_cache = [] return self._apache_modules_cache except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Failed to get Apache modules: %s", e) self._apache_modules_cache = [] return self._apache_modules_cache def get_apache_system_info(self) -> Dict[str, Any]: """ Get Apache system information for export as separate record. Returns: Dict with system information to be exported as single record """ modules = self._get_apache_modules() version = self._get_apache_version() system_info = { 'apache_version': version, 'loaded_modules': modules, } return system_info def _get_user_config_locations(self, usernames: List[str]) -> List[str]: """ Get Apache configuration directories for specific users only. Args: usernames: List of usernames to get configurations for Returns: List of directories containing user-specific Apache configurations """ user_locations = [] for username in usernames: for base_path in self.user_config_base_paths: user_path = os.path.join(base_path, username) if os.path.isdir(user_path): user_locations.append(user_path) self.logger.debug("Found user config directory: %s", user_path) return user_locations def collect_user_specific_config_paths(self, usernames: List[str]) -> Dict[str, Any]: """ Collect paths to user-specific Apache configuration files. Args: usernames: List of usernames to collect configs for Returns: Dict with collected config file paths as simple list """ config_data = { 'config_paths': [] } if not usernames: return config_data try: user_locations = self._get_user_config_locations(usernames) self.logger.debug("Scanning %d user-specific locations for %d users", len(user_locations), len(usernames)) for location in user_locations: if os.path.isdir(location): self._collect_config_paths_from_directory(location, config_data, exclude_user_dirs=False) except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Error in collect_user_specific_config_paths: %s", e) self.logger.debug("Found %d user-specific Apache config files", len(config_data['config_paths'])) return config_data def collect_global_config_paths(self) -> Dict[str, Any]: """ Collect paths to general Apache configuration files (non-user-specific). Returns: Dict with collected config file paths as simple list """ config_data = { 'config_paths': [] } try: for location in self.apache_config_locations: if os.path.isdir(location): try: # For conf.d directory, exclude userdata subdirs to avoid collecting user-specific configs exclude_users = location.rstrip('/').endswith('/conf.d') self._collect_config_paths_from_directory(location, config_data, exclude_user_dirs=exclude_users) self.logger.debug("Collected config paths from %s (exclude_user_dirs=%s)", location, exclude_users) except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Error scanning config directory %s: %s", location, e) self.logger.debug("Found %d general Apache config files", len(config_data['config_paths'])) except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Error in collect_global_config_paths: %s", e) return config_data def _collect_config_paths_from_directory(self, location: str, config_data: Dict[str, Any], exclude_user_dirs: bool = False) -> None: """ Helper method to collect configuration file paths from a specific directory. Args: location: Directory path to scan for .conf files config_data: Dictionary with 'config_paths' list to store collected paths exclude_user_dirs: If True, skip userdata subdirectories to avoid collecting user-specific configs """ max_depth = 10 # Reasonable limit for Apache configs for root, dirs, files in os.walk(location): # Calculate current depth relative to base location depth = root[len(location):].count(os.sep) if depth >= max_depth: dirs[:] = [] # Don't descend further continue # Exclude userdata directories when collecting general configs if exclude_user_dirs and 'userdata' in dirs: dirs.remove('userdata') self.logger.debug("Excluded userdata directory from general config collection in %s", root) for filename in files: if filename.endswith('.conf'): file_path = os.path.join(root, filename) # Check if file exists (including symlinks) and is readable if (os.path.exists(file_path) and (os.path.isfile(file_path) or os.path.islink(file_path)) and os.access(file_path, os.R_OK)): # Add to simple list - no grouping by directory needed config_data['config_paths'].append({ 'file_path': file_path, 'is_symlink': os.path.islink(file_path) }) else: self.logger.debug("Cannot access config file: %s", file_path)	11.65 KB	2026-02-17 10:40
📄 constants.py # -- coding: utf-8 -- # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from pathlib import Path # Collector version # once version is increased - collector invalidates cache # see sending_stats dict in website_collector.py MAJOR_VER = 2 VERSION = f"{MAJOR_VER}.0.0" # Storage dir STORAGE_DIR = Path("/var/lve/website_collector") # Opt-out control OPT_OUT_FILE = STORAGE_DIR / ".disabled" # Processing limits DOCROOT_SCAN_TIMEOUT = 30 # Timeout per docroot scan in seconds DOCROOT_LIMIT_PER_RUN = 500 # Maximum number of docroots to process in single run DOCROOT_MAX_DEPTH = 30 # Maximum depth of docroot scanning DOCROOT_EXCLUDE_DIRS = [ 'node_modules', 'vendor', '.idea', '.vscode', '.well-known', '.git', '.svn', '.hg'] # Directories to exclude from docroot scanning SENDING_TIMEOUT = 7200 # Total timeout 2 hours SENDING_COMPRESSION = True # Compress data before sending MAX_RECORD_SIZE = 1024 * 1024 * 1000 # 1000MB FILE_SIZE_LIMIT = 1024 * 1024 * 1000 # 1000MB	1.1 KB	2026-02-17 10:40
📄 docroot_processor.py # -- coding: utf-8 -- # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import logging import os import time from pathlib import Path from typing import Dict, List, Optional, Any from cl_website_collector.constants import DOCROOT_EXCLUDE_DIRS, DOCROOT_MAX_DEPTH class DocrootProcessor: """ Processes individual docroot to collect .htaccess files and metadata. """ def __init__(self, logger: logging.Logger): self.logger = logger def collect_htaccess_paths(self, docroot: str, domains: list, username: str, timeout: int = 30) -> Optional[ Dict[str, Any]]: """ Collect .htaccess file paths from a docroot without reading file contents. Args: docroot: Document root path domains: Domain names username: Owner username timeout: Processing timeout in seconds Returns: Dictionary with collected file paths or None if failed """ start_time = time.time() result = { 'docroot': docroot, 'domains': domains, 'username': username, 'htaccess_file_paths': [], 'symlinks': [], 'timeout_reached': False, 'processing_time_seconds': 0, 'htaccess_files_found': 0, } try: self.logger.debug("Finding .htaccess files in %s", docroot) htaccess_files = self._find_htaccess_files(docroot, max_depth=DOCROOT_MAX_DEPTH, timeout=timeout - 5) self.logger.debug("Found %d .htaccess files in %s", len(htaccess_files), docroot) for file_path in htaccess_files: self.logger.debug(" - %s", file_path) if not htaccess_files: self.logger.debug("No .htaccess files found in %s", docroot) else: # Process each found file path (no content reading) for file_path in htaccess_files: if time.time() - start_time > timeout: result['timeout_reached'] = True self.logger.error("[WEBSITE-COLLECTOR] Timeout reached while collecting paths in %s", docroot) break try: self.logger.debug("Collecting .htaccess path: %s", file_path) # Handle symlinks p = Path(file_path) is_symlink = p.is_symlink() real_path = str(p.resolve(strict=False)) if is_symlink else file_path if is_symlink: result['symlinks'].append({ 'link': self._normalize_path(file_path, docroot), 'target': real_path }) # Check if file is readable if Path(real_path).exists() and os.access(real_path, os.R_OK): # Store file path info for on-demand reading location = self._normalize_path(file_path, docroot) result['htaccess_file_paths'].append({ 'location': location, 'file_path': file_path, 'real_path': real_path, 'is_symlink': is_symlink }) else: self.logger.debug("Cannot read file: %s", file_path) except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Error collecting path %s: %s", file_path, e) result['htaccess_files_found'] = len(result['htaccess_file_paths']) result['processing_time_seconds'] = time.time() - start_time self.logger.debug("Collected %d .htaccess file paths from %s in %.2fs", result['htaccess_files_found'], docroot, result['processing_time_seconds']) except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Error processing docroot %s: %s", docroot, e) return result def _find_htaccess_files(self, docroot: str, max_depth: int = DOCROOT_MAX_DEPTH, timeout: int = 25) -> List[str]: """ Find .htaccess files. """ start_time = time.time() htaccess_files = [] try: for root, dirs, files in os.walk(docroot): # Check timeout if time.time() - start_time > timeout: self.logger.error("[WEBSITE-COLLECTOR] os.walk timeout for %s", docroot) break # Calculate current depth robustly regardless of trailing separators if root == docroot: depth = 0 else: depth = os.path.relpath(root, docroot).count(os.sep) if depth >= max_depth: dirs[:] = [] # Don't go deeper, but still process files at this level # Apply exclusion filters for directories dirs[:] = [d for d in dirs if not self._should_exclude_directory(root, d)] # Look for .htaccess files if '.htaccess' in files: file_path = Path(root) / '.htaccess' # Consider empty .htaccess files as valid as well if (file_path.is_file() and os.access(str(file_path), os.R_OK)): htaccess_files.append(str(file_path)) except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Error walking %s: %s", docroot, e) return htaccess_files def _should_exclude_directory(self, parent_path: str, dirname: str) -> bool: """ Check if directory should be excluded based on DOCROOT_EXCLUDE_DIRS. Supports both plain directory names (e.g. "node_modules") and nested paths (e.g. "wp-content/cache"). The check is performed against the full candidate path composed from parent_path and dirname. """ try: candidate = Path(parent_path) / dirname candidate_normalized = candidate.resolve(strict=False) for exclude_dir in DOCROOT_EXCLUDE_DIRS: pattern = Path(exclude_dir) # Match exact directory name or nested path suffix if (str(candidate_normalized).endswith(os.sep + str(pattern)) or candidate.name == pattern.name): return True except Exception: # Be conservative on errors and do not exclude return False return False def _normalize_path(self, file_path: str, docroot: str) -> str: """ Normalize file path relative to docroot. """ try: return str(Path(file_path).relative_to(Path(docroot))) except ValueError: # If relative path calculation fails, return filename only return Path(file_path).name	7.1 KB	2026-02-17 10:40
📄 feature_manager.py # -- coding: utf-8 -- # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import os from copy import deepcopy import requests class FeatureManagerApiException(Exception): pass class FeatureManager: """ Communicates with FeatureManager API to get feature settings based on server metadata """ def __init__(self, app_logger): self.api_url = self.base_url() + "/feature_manager/feature_settings" self.app_logger = app_logger @staticmethod def base_url() -> str: base_url = "https://x-ray-advice.cloudlinux.com" if os.path.exists("/opt/cloudlinux/staging_mode"): base_url = "https://x-ray-staging.cloudlinux.com" return base_url def request(self, payload): response = requests.get(self.api_url, json=payload, timeout=20) self.app_logger.info(f"FeatureManager request to {self.api_url} " f"with payload {payload} " f"returned {response.status_code}") if response.status_code != 200: raise FeatureManagerApiException(f"FeatureManager API returned {response.status_code}") return response.json() def get_decision(self, server_metadata, reason): payload = deepcopy(server_metadata) payload["reason"] = reason return self.request(payload)	1.45 KB	2026-02-17 10:40
📄 website_collector.py # -- coding: utf-8 -- # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import json import logging import os import time import gzip import base64 import hashlib from datetime import datetime, timezone from pathlib import Path from typing import Dict, Any, Optional from copy import deepcopy import requests from lve_utils import PKG_VERSION from clcommon.cpapi import cpusers, userdomains from cl_website_collector.apache_processor import ApacheProcessor from cl_website_collector.docroot_processor import DocrootProcessor from cl_website_collector.constants import ( VERSION, MAJOR_VER, STORAGE_DIR, OPT_OUT_FILE, DOCROOT_SCAN_TIMEOUT, DOCROOT_LIMIT_PER_RUN, SENDING_TIMEOUT, SENDING_COMPRESSION, MAX_RECORD_SIZE, FILE_SIZE_LIMIT, ) class WebsiteCollector: """ Website data collection. Collects and exports various Apache-related data: - Apache configuration files (.conf) from standard directories - Apache system information (version, loaded modules) - .htaccess files from website document roots - Maintains processing state to avoid duplicate scans """ def __init__(self, logger: logging.Logger, storage_dir: Optional[Path] = None, max_sites: Optional[int] = None): """ Initialize website collector for on-demand data collection and sending. Args: storage_dir: Directory for storing website collector state max_sites: Maximum number of sites to process (with hash-based selection) """ # Use default directories if not provided if storage_dir is None: storage_dir = STORAGE_DIR self.logger = logger self.max_sites = max_sites self.logger.debug("[WEBSITE-COLLECTOR] Initialized with max_sites=%s", str(self.max_sites)) # Ensure directories exist self._ensure_directories(storage_dir=storage_dir) # Initialize components for on-demand data collection self.docroot_processor = DocrootProcessor(logger=self.logger) self.apache_processor = ApacheProcessor(logger=self.logger) # Initialize sending stats tracking self.stats_file = storage_dir / "website_collector_stats.json" self.dry_run_file = storage_dir / "dry-run.jsonl" self.sending_stat_defaults = { 'global_apache_configs_sent': False, 'docroots': {}, "collected_version": 0 } self.sending_stats = deepcopy(self.sending_stat_defaults) # Load sending stats self._load_sending_stats() # reset cache to trigger collection with latest collector self.reset_cache_within_version() def _hash_and_sort_docroots(self, docroots_data: Dict) -> Dict: """ Hash and sort docroots, then return top N based on max_sites limit. Takes into account already sent sites to respect the global limit. Args: docroots_data: Dict with docroot as key and data as value (unprocessed only) Returns: Dict with selected docroots (limited and sorted by hash) """ if self.max_sites is None: return docroots_data # Count how many sites were already sent from this server already_sent_count = len(self.sending_stats.get('docroots', {})) # Calculate how many more sites we can send remaining_limit = max(0, self.max_sites - already_sent_count) if remaining_limit <= 0: self.logger.debug("Max sites limit reached: %d already sent, limit is %d", already_sent_count, self.max_sites) return {} if len(docroots_data) <= remaining_limit: return docroots_data # Create list of (hash, docroot, data) tuples for consistent selection hashed_docroots = [] for docroot, data in docroots_data.items(): # Hash the docroot path for consistent, random-like ordering docroot_hash = hashlib.sha256(docroot.encode()).hexdigest() hashed_docroots.append((docroot_hash, docroot, data)) # Sort by hash to get consistent but random-like ordering hashed_docroots.sort(key=lambda x: x[0]) # Take only the remaining limit selected_docroots = {} for _, docroot, data in hashed_docroots[:remaining_limit]: selected_docroots[docroot] = data self.logger.debug("Hash-based selection: %d already sent, %d remaining limit, " "selected %d from %d candidates", already_sent_count, remaining_limit, len(selected_docroots), len(docroots_data)) return selected_docroots def _get_docroots_for_sending(self) -> Dict: """ Get docroots for sending, with optional filtering and limit. Returns: Dict of docroots ready for processing (limited and filtered by sent status) """ selected_docroots = {} processed_count = 0 try: # Use Panel API, but optimize by checking processed status early panel_users = cpusers() for username in panel_users: try: user_domains = userdomains(username) user_docroots_found = {} # Collect ALL domains for ALL docroots for this user first for domain, docroot in user_domains: docroot_path = Path(docroot) if docroot_path.exists() and os.access(str(docroot_path), os.R_OK): # Check if already fully sent - skip processing if self.is_docroot_sent(docroot): processed_count += 1 continue # Add to user's docroots (collect ALL domains per docroot) if docroot not in user_docroots_found: user_docroots_found[docroot] = { "docroot": docroot, "username": username, "domains": [] } user_docroots_found[docroot]["domains"].append(domain) # Add user's docroots to global selection (check limit at docroot level) limit_reached = False for docroot, docroot_data in user_docroots_found.items(): if len(selected_docroots) >= DOCROOT_LIMIT_PER_RUN: limit_reached = True break selected_docroots[docroot] = docroot_data # If we reached the limit, stop processing more users if limit_reached: break except Exception as e: self.logger.debug("Failed to get domains for user %s: %s", username, e) except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Panel API failed: %s", e) self.logger.debug("Found %d unprocessed docroots (skipped %d already processed)", len(selected_docroots), processed_count) return self._hash_and_sort_docroots(selected_docroots) def _load_sending_stats(self) -> None: """Load sending statistics from tracking file.""" try: if self.stats_file.exists(): with open(self.stats_file, 'r', encoding='utf-8') as f: self.sending_stats = json.load(f) self.logger.debug("Loaded sending stats") except Exception as e: self.logger.debug("Failed to load sending stats: %s", e) def reset_cache_within_version(self) -> None: """ Sync sending stats cache within the same version to remove entries for docroots that no longer exist in the system. """ # keep cache, data was collected with latest collector version prev_collected_ver = self.sending_stats.get("collected_version", 0) if prev_collected_ver >= MAJOR_VER: return self.logger.info("[WEBSITE-COLLECTOR] New version of collector, " "current version=%s, " "previously collected=%s re-setting cache", str(VERSION), str(prev_collected_ver)) self.sending_stats = deepcopy(self.sending_stat_defaults) def read_file_content(self, file_path: str) -> Optional[str]: """ Read any file content on-demand for sending. Supports both Apache configs and .htaccess files. """ try: p = Path(file_path) is_symlink = p.is_symlink() real_p = p.resolve(strict=False) if is_symlink else p if not (real_p.exists() and real_p.is_file() and os.access(str(real_p), os.R_OK)): self.logger.error("[WEBSITE-COLLECTOR] Cannot read file: %s", file_path) return None if real_p.stat().st_size > FILE_SIZE_LIMIT: self.logger.error("[WEBSITE-COLLECTOR] File too big: %s (size: %d bytes)", file_path, real_p.stat().st_size) return None return real_p.read_text(encoding='utf-8', errors='replace') except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Error reading file %s: %s", file_path, e) return None def _save_sending_stats(self) -> None: """Save sending statistics to tracking file.""" try: self.sending_stats['last_updated'] = datetime.now(timezone.utc).isoformat() # keep last collected version self.sending_stats["collected_version"] = MAJOR_VER with open(self.stats_file, 'w', encoding='utf-8') as f: json.dump(self.sending_stats, f, indent=2, ensure_ascii=False) self.logger.debug("Saved sending stats") except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Failed to save sending stats: %s", e) def is_docroot_sent(self, docroot: str) -> bool: """Check if docroot data was fully sent (both htaccess and user apache configs).""" docroot_stats = self.sending_stats.get('docroots', {}).get(docroot, False) return bool(docroot_stats) def mark_docroot_sent(self, docroot: str, username: str) -> None: """Mark docroot data as sent.""" if docroot not in self.sending_stats['docroots']: self.sending_stats['docroots'][docroot] = { 'username': username } self.sending_stats['docroots'][docroot]['last_processed'] = datetime.now(timezone.utc).isoformat() def mark_global_apache_sent(self) -> None: """Mark global apache configs as sent.""" self.sending_stats['global_apache_configs_sent'] = True @staticmethod def is_collection_disabled() -> bool: """ Check if website collection is disabled via opt-out file. Returns: True if collection is disabled, False otherwise """ return OPT_OUT_FILE.exists() def _ensure_directories(self, storage_dir: Path) -> None: """ Create necessary directory structure. Args: storage_dir: Directory for storing website collector state """ storage_dir.mkdir(mode=0o750, parents=True, exist_ok=True) def _create_base_record(self, client_id: str, system_id: str, platform: str, panel: str, server: str) -> Dict[ str, Any]: """Create base record with common fields.""" return { 'collection_timestamp': int(time.time()), 'collection_version': VERSION, 'collected_package_version': PKG_VERSION, 'client_id': client_id, 'system_id': system_id, 'platform': platform, 'panel': panel, 'server': server, 'data': {}, 'meta_data': {} } def _serialize_record(self, record: Dict[str, Any]) -> str: """ Serialize record to JSON. Returns: Serialized JSON string """ return json.dumps(record, ensure_ascii=False, separators=(',', ':')) def _create_new_record_part(self, base_record: Dict[str, Any]) -> Dict[str, Any]: """ Create new record part with base data but empty data/meta_data sections. Args: base_record: Base record to copy from Returns: New record with empty data sections """ new_record = base_record.copy() new_record['data'] = {} new_record['meta_data'] = {} return new_record def _add_file_to_records(self, file_path: str, file_content: str, is_symlink: bool, current_record: Dict[str, Any], base_record: Dict[str, Any], records_to_send: list) -> Dict[str, Any]: """ Add file to current record, splitting if size limit exceeded. Args: file_path: Path of the file to add file_content: Content of the file is_symlink: Whether the file is a symlink current_record: Current record being built base_record: Base record template for new parts records_to_send: List to append completed records to Returns: Updated current record (may be new if split occurred) """ # Create temporary copy to check size temp_record = current_record.copy() temp_record['data'] = current_record['data'].copy() temp_record['meta_data'] = current_record['meta_data'].copy() # Add new file to temporary copy temp_record['data'][file_path] = file_content if is_symlink: temp_record['meta_data'][file_path] = {'is_symlink': True} # Check size with actual serialization temp_serialized = self._serialize_record(temp_record) temp_size = len(temp_serialized.encode('utf-8')) if temp_size > MAX_RECORD_SIZE and current_record['data']: # Current record is full, save it and create new one # Store serialized data before appending current_serialized = self._serialize_record(current_record) records_to_send.append(current_serialized) current_record = self._create_new_record_part(base_record) # Add file to new record current_record['data'][file_path] = file_content if is_symlink: current_record['meta_data'][file_path] = {'is_symlink': True} else: # File fits, use temporary copy current_record = temp_record return current_record def _send_or_collect_record(self, session, api_url: str, dry_run: bool, serialized_data: str, remote_conf_ver: str) -> bool: """Send record or save to JSONL file for dry run. Returns True if successful.""" if dry_run: try: # Save to dry-run.jsonl file (append mode) with open(self.dry_run_file, 'a', encoding='utf-8') as f: f.write(serialized_data + '\n') return True except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Error saving dry-run record: %s", e) return False else: headers = { "Content-Type": "text/plain", "X-Website-Collector-Version": f"{VERSION}_{remote_conf_ver}", } data_to_send = serialized_data if SENDING_COMPRESSION: data_to_send = serialized_data.encode('utf-8') compressed_data = gzip.compress(data_to_send, compresslevel=6) original_bytes = len(serialized_data.encode('utf-8')) compressed_bytes = len(compressed_data) reduction = (1 - compressed_bytes / original_bytes) * 100 self.logger.debug("Applied gzip compression: %s -> %s bytes (%.1f%% reduction)", f"{original_bytes:,}", f"{compressed_bytes:,}", reduction) data_to_send = base64.b64encode(compressed_data).decode("ascii") base64_bytes = len(data_to_send) self.logger.debug("Base64 encoded: %s -> %s bytes", f"{compressed_bytes:,}", f"{base64_bytes:,}") # Try sending with retries max_retries = 3 last_error = None last_status_code = None for attempt in range(max_retries): try: response = session.post(api_url, data=data_to_send, headers=headers, timeout=30) if response.status_code == 200: return True last_status_code = response.status_code if attempt < max_retries - 1: self.logger.debug("Non-200 response %s, attempt %d/%d", response.status_code, attempt + 1, max_retries) # Exponential backoff: 1s, 2s, 4s time.sleep(2 attempt) else: self.logger.debug("Final attempt failed: Non-200 response %s", response.status_code) except Exception as e: last_error = str(e) if attempt < max_retries - 1: self.logger.debug("HTTP error on attempt %d/%d: %s", attempt + 1, max_retries, e) # Exponential backoff: 1s, 2s, 4s time.sleep(2 attempt) else: self.logger.error("[WEBSITE-COLLECTOR] Final HTTP error while sending record: %s", e) # All retries failed - determine failure reason if last_status_code: reason = f"HTTP {last_status_code}" elif last_error: reason = f"Exception: {last_error}" else: reason = "Unknown error" self.logger.error("[WEBSITE-COLLECTOR] Failed to send record after %d retries. Reason: %s", max_retries, reason) return False def _send_global_configs(self, session, api_url: str, dry_run: bool, client_id: str, system_id: str, platform: str, panel: str, server: str, remote_config_version: str) -> bool: """ Send global Apache configs and system info, splitting into multiple records if needed. Returns: True if all records sent successfully, False otherwise """ base_record = self._create_base_record(client_id, system_id, platform, panel, server) base_record.update({ 'username': '', 'domains': [], 'document_root': '', }) current_record = self._create_new_record_part(base_record) records_to_send = [] # Collect global Apache config files global_configs = self.apache_processor.collect_global_config_paths() config_paths = global_configs.get('config_paths', []) self.logger.debug("Found %d global Apache config files", len(config_paths)) for i, path_info in enumerate(config_paths, 1): self.logger.debug("Processing global config file: %s", path_info['file_path']) file_content = self.read_file_content(path_info['file_path']) if file_content is None: file_content = '# ERROR: File unreadable' current_record = self._add_file_to_records( path_info['file_path'], file_content, path_info.get('is_symlink', False), current_record, base_record, records_to_send ) # Add system info to the last record system_info = self.apache_processor.get_apache_system_info() current_record['data']['modules'] = system_info.get('loaded_modules', []) current_record['data']['version'] = system_info.get('apache_version', '') # Add last record current_serialized = self._serialize_record(current_record) records_to_send.append(current_serialized) # Send all records for i, serialized_data in enumerate(records_to_send): self.logger.debug("Sending global config record %d/%d", i + 1, len(records_to_send)) if not self._send_or_collect_record(session, api_url, dry_run, serialized_data, remote_config_version): return False return True def _send_docroot_data(self, docroot: str, domains: list, username: str, session, api_url: str, dry_run: bool, client_id: str, system_id: str, platform: str, panel: str, server: str, remote_config_version: str) -> bool: """ Send all docroot data, splitting into multiple records if needed. Returns: True if all records sent successfully, False otherwise """ base_record = self._create_base_record(client_id, system_id, platform, panel, server) base_record.update({ 'username': username, 'domains': domains, 'document_root': docroot, }) current_record = self._create_new_record_part(base_record) records_to_send = [] # Collect htaccess files htaccess_data = self.docroot_processor.collect_htaccess_paths(docroot, domains, username, timeout=DOCROOT_SCAN_TIMEOUT) if htaccess_data and htaccess_data.get('htaccess_file_paths'): for htaccess_path_info in htaccess_data['htaccess_file_paths']: content = self.read_file_content(htaccess_path_info['real_path']) if content is None: content = '# ERROR: File unreadable' current_record = self._add_file_to_records( htaccess_path_info['file_path'], content, htaccess_path_info.get('is_symlink', False), current_record, base_record, records_to_send ) # Collect user Apache configs user_configs = self.apache_processor.collect_user_specific_config_paths([username]) if user_configs and user_configs.get('config_paths'): for path_info in user_configs['config_paths']: file_content = self.read_file_content(path_info['file_path']) if file_content is None: file_content = '# ERROR: File unreadable' current_record = self._add_file_to_records( path_info['file_path'], file_content, path_info.get('is_symlink', False), current_record, base_record, records_to_send ) # Add last record if it has data or if no records were created if current_record['data'] or not records_to_send: current_serialized = self._serialize_record(current_record) records_to_send.append(current_serialized) # Send all records for i, serialized_data in enumerate(records_to_send): self.logger.debug("Sending docroot %s record %d/%d", docroot, i + 1, len(records_to_send)) if not self._send_or_collect_record(session, api_url, dry_run, serialized_data, remote_config_version): return False return True def send_data(self, system_id: str, client_id: str, platform: str, panel: str, server: str, api_url: str, remote_config_version: str, dry_run: bool = False) -> None: """ Send data directly or save to JSONL file in dry-run mode. Read htaccess and apache configs on-demand and track sending status. Args: dry_run: If True, saves data to dry-run.jsonl file instead of sending to API """ start_time = time.time() session = requests.Session() if not dry_run else None try: # Step 1: Send global Apache configs if not sent yet if not self.sending_stats.get('global_apache_configs_sent', False): self.logger.debug("%s global Apache configs", "Collecting" if dry_run else "Sending") if self._send_global_configs(session, api_url, dry_run, client_id, system_id, platform, panel, server, remote_config_version): if not dry_run: self.mark_global_apache_sent() self._save_sending_stats() self.logger.debug("Global Apache configs sent successfully") # Step 2: Process docroots unprocessed_docroots = self._get_docroots_for_sending() for docroot, domain_data in unprocessed_docroots.items(): if time.time() - start_time > SENDING_TIMEOUT: self.logger.error("[WEBSITE-COLLECTOR] Timeout reached during docroot processing") break self.logger.debug("Processing docroot %s for user %s", docroot, domain_data['username']) if self._send_docroot_data(docroot, domain_data['domains'], domain_data['username'], session, api_url, dry_run, client_id, system_id, platform, panel, server, remote_config_version): if not dry_run: self.mark_docroot_sent(docroot, domain_data['username']) self._save_sending_stats() self.logger.debug("Docroot %s sent successfully", docroot) execution_time = time.time() - start_time if dry_run: self.logger.info("DRY RUN completed: saved to %s in %.1f seconds", self.dry_run_file, execution_time) else: self.logger.debug("Sequential sending completed: %.1f seconds", execution_time) except Exception as e: self.logger.error("[WEBSITE-COLLECTOR] Failed during %s: %s", "dry run" if dry_run else "sequential sending", e) finally: if session: session.close()	27.34 KB	2026-02-17 10:40

Terminal