← Back
Editing: 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 clcommon.clpwd import drop_privileges 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 (walk + read under user privileges to # prevent symlink-based file exfiltration as root). 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'): with drop_privileges(username): 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'): with drop_privileges(username): 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()
Save File
Cancel