Edit

[EDIT] FILE: 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()

CANCEL