Google

Edit File: anomaly.py

# Copyright (c) Cloud Linux Software, Inc # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT import io import json import os import shlex import tarfile import time import uuid from tempfile import NamedTemporaryFile from kcarectl import auth, config, http_utils, ipv6_support, kcare, log_utils, utils from kcarectl.process_utils import run_command if False: # pragma: no cover from typing import Any, Dict, List, Optional # noqa: F401 from typing_extensions import Self # noqa: F401 class DataPackage(object): """based on DataPackage from eportal""" def __init__(self): # type: () -> None self._tar = None # type: Optional[tarfile.TarFile] self._errors_buffer = [] # type: List[str] self._total_payload_size = 0 @property def archive_path(self): # type: () -> str tar = self._ensure_tar_created() return str(tar.name) def add_stdout(self, arcname, cmd): # type: (str, str) -> None stdout = None stderr = None try: _, stdout, stderr = run_command(shlex.split(cmd), catch_stdout=True, catch_stderr=True) except Exception as e: stderr = str(e) if stderr: self.log_error('failed to dump stdout of {0}:\n{1}'.format(cmd, stderr)) if stdout is not None: self.add_file(arcname, data_bytes=utils.bstr(stdout, encoding='utf-8')) def add_file(self, arcname, src_path=None, data_bytes=None, skip_limit_check=False): # type: (str, Optional[str], Optional[bytes], bool) -> None if src_path is None and data_bytes is None: raise ValueError('No src_path or data_bytes provided') tar = self._ensure_tar_created() entry_size = 0 if src_path is not None: if not os.path.exists(src_path): self.log_error('file not found: {0}'.format(src_path)) return entry_size = os.path.getsize(src_path) else: entry_size = len(data_bytes) # type: ignore[arg-type] if not self._check_required_space(entry_size): self.log_error('no available space to store: {0}'.format(arcname)) return if not skip_limit_check and not self._check_total_payload_limit(entry_size): self.log_error('skip due to total payload size limit: {0}'.format(arcname)) return try: if src_path: tar.add(src_path, arcname=arcname) else: info = tarfile.TarInfo(arcname) info.size = entry_size tar.addfile(info, io.BytesIO(data_bytes)) # type: ignore[arg-type] self._total_payload_size += entry_size except Exception as e: self.log_error('failed to store {0}: {1}'.format(arcname, e)) def add_json(self, arcname, data): # type: (str, Dict[str, Any]) -> None try: data_bytes = utils.bstr(json.dumps(data, indent=4), encoding='utf-8') except TypeError as e: self.log_error('failed to dump {0}:\n{1}'.format(arcname, e)) return self.add_file(arcname, data_bytes=data_bytes) def _check_required_space(self, entry_size): # type: (int) -> bool # here we simplify the check and ignore that the compressed file size will be less statvfs = os.statvfs(self.archive_path) return statvfs.f_frsize * statvfs.f_bfree > entry_size def _check_total_payload_limit(self, entry_size): # type: (int) -> bool # here we simplify the check and ignore that the compressed file size will be less return config.KERNEL_ANOMALY_REPORT_MAX_SIZE_BYTES > self._total_payload_size + entry_size def log_error(self, error_msg): # type: (str) -> None error_msg = error_msg.strip() log_utils.logerror(error_msg, print_msg=False) self._errors_buffer.append(error_msg) def __enter__(self): # type: () -> Self for compression_mode in ('w:xz', 'w:bz2', 'w:gz'): # pragma: no branch tmpfile = NamedTemporaryFile(suffix='.tar.{0}'.format(compression_mode[2:]), delete=False) tmpfile.close() try: log_utils.loginfo('Creating DataPackage: {0}'.format(tmpfile.name), print_msg=False) self._tar = tarfile.open(name=tmpfile.name, mode=compression_mode) return self except Exception as err: # pragma: no cover if os.path.exists(tmpfile.name): os.unlink(tmpfile.name) self._tar = None if not isinstance(err, tarfile.CompressionError): raise raise tarfile.CompressionError('No supported compression method found') # pragma: no cover def __exit__(self, exc_type, exc_val, exc_tb): # type: (Optional[type[BaseException]], Optional[BaseException], Any) -> bool if self._errors_buffer: # pragma: no branch errors = '\n'.join(self._errors_buffer) + '\n' self.add_file('errors.log', data_bytes=utils.bstr(errors), skip_limit_check=True) if self._tar: # pragma: no branch self._tar.close() if exc_val: self.remove_archive() return False return True @utils.catch_errors(logger=log_utils.logwarn) def remove_archive(self): # type: () -> None if self._tar and os.path.exists(self.archive_path): # pragma: no branch os.unlink(self.archive_path) def _ensure_tar_created(self): # type: () -> tarfile.TarFile if not self._tar: raise RuntimeError('DataPackage should be used as a context manager') return self._tar @utils.catch_errors(logger=log_utils.logwarn) def send_data_package(data_package): # type: (DataPackage) -> str """Send the DataPackage archive to the patch server. :param data_package: DataPackage instance to send :return: Upload name (package identifier) """ # Generate a unique package name # Use find('.') to get extension from first dot to preserve .tar.xz/.tar.bz2/.tar.gz basename = os.path.basename(data_package.archive_path) ext = basename[basename.find('.') :] if '.' in basename else '' upload_name = str(uuid.uuid4()) + ext upload_url = ipv6_support.get_patch_server() + '/upload/kernel-anomaly/' + upload_name http_utils.upload_file( data_package.archive_path, upload_url=upload_url, auth_string=auth.get_http_auth_string(), ) return upload_name def make_manifest_json(): # pragma: no cover # type: () -> Dict[str, Any] return { "schema_version": 1, "type": "kernel-anomaly", "time_created": int(time.time()), } def copy_recent_files(files, data_package, archive_prefix): # type: (List[str], DataPackage, str) -> None """adds recent files for the last hour to the given data package starting from the newest one""" now = time.time() for path, ctime in kcare.sort_files_by_ctime(files): if now - ctime > 3600: break arcname = '{0}/{1}'.format(archive_prefix, path.replace('/', '_')) data_package.add_file(arcname, src_path=path) @utils.catch_errors(logger=log_utils.logwarn) def prepare_kernel_anomaly_report(server_info): # pragma: no cover # type: (Dict[str, Any]) -> DataPackage data_package = DataPackage() with data_package: if os.path.exists('/var/log/messages'): data_package.add_stdout('messages', 'tail -n10000 /var/log/messages') if os.path.exists('/var/log/syslog'): data_package.add_stdout('syslog', 'tail -n10000 /var/log/syslog') data_package.add_stdout('kcarectl.log', 'tail -n10000 /var/log/kcarectl.log') data_package.add_stdout('dmesg', 'dmesg') data_package.add_stdout('ls_var_cache_kcare', 'ls -lR /var/cache/kcare/') if os.path.exists('/usr/bin/rpm') or os.path.exists('/bin/rpm'): packages_cmd = r'rpm -q -a --queryformat="%{N}|%{V}-%{R}|%{arch}|%{INSTALLTIME:date}\n"' elif os.path.exists('/usr/bin/dpkg'): packages_cmd = r'/usr/bin/dpkg-query -W -f "${binary:Package}|${Version}|${Architecture}\n"' data_package.add_file('dpkg.log', src_path='/var/log/dpkg.log') else: packages_cmd = 'echo "unknown package manager"' data_package.add_stdout('packages.list', packages_cmd) # tar fails to add files from /proc directly so we first read them to memory with open('/proc/version') as f: data_package.add_file('proc_version', data_bytes=utils.bstr(f.read())) with open('/proc/modules') as f: data_package.add_file('proc_modules', data_bytes=utils.bstr(f.read())) data_package.add_file('kcare.conf', src_path='/etc/sysconfig/kcare/kcare.conf') data_package.add_json('server_info.json', server_info) data_package.add_json('manifest.json', make_manifest_json()) # kdump data_package.add_file('kdump.conf', src_path='/etc/kdump.conf') data_package.add_stdout('ls_kdump', 'ls -lR {0}'.format(kcare.get_kdump_root())) try: copy_recent_files(kcare.list_kdump_txt_files(), data_package, 'kdump') except Exception as e: data_package.log_error('failed to copy kdumps:\n{0}'.format(e)) try: copy_recent_files(kcare.list_crashreporter_log_files(), data_package, 'crashreporter') except Exception as e: data_package.log_error('failed to copy crashreporter artifacts:\n{0}'.format(e)) return data_package @utils.catch_errors(logger=log_utils.logwarn, default_return=False) def detect_anomaly(server_info): # type: (Dict[str, Any]) -> bool """taken from eportal - anomalies::detect_agent_reboot""" reason = server_info['reason'] # type: str uptime = int(server_info['uptime']) patch_level = int(server_info.get('patch_level') or '-1') last_stop = int(server_info['last_stop']) ts = server_info['ts'] # type: int try: state_ts = int(server_info['state']['ts']) except (KeyError, TypeError, ValueError): state_ts = 0 fields = [reason, uptime, patch_level, state_ts, last_stop, ts] if not all(fields): return False first_update_after_reboot_marker = False crash_soon_after_update_marker = False no_proper_shutdown_marker = False if uptime < 300 and patch_level == -1: first_update_after_reboot_marker = True if (ts - uptime) > state_ts > (ts - uptime - 1800) and reason == 'update': crash_soon_after_update_marker = True if last_stop < state_ts: # pragma: no branch no_proper_shutdown_marker = True markers = [ first_update_after_reboot_marker, crash_soon_after_update_marker, no_proper_shutdown_marker, ] if all(markers): log_utils.loginfo('Agent anomaly detected: {0}'.format(server_info)) return True return False if __name__ == '__main__': # pragma: no cover prepare_kernel_anomaly_report({})