22

✏️ Editing: sentry.py

from pathlib import Path
from subprocess import DEVNULL, CalledProcessError, check_output
from typing import Any

from defence360agent.utils import stub_unexpected_error

def _run_cmd(cmd):
    try:
        out = check_output(cmd, stderr=DEVNULL)
    except (FileNotFoundError, CalledProcessError):
        return None

return out.decode("utf-8", errors="ignore").strip()

@stub_unexpected_error
def _get_virtualization_type():
    systemd_virt = _run_cmd(["systemd-detect-virt"])
    if systemd_virt:
        return systemd_virt

virt_what = _run_cmd(["virt-what"])
    if virt_what:
        return virt_what

demicode = _run_cmd(["dmidecode", "-s", "system-manufacturer"])
    if demicode:
        return demicode

return "fail to detect"

@stub_unexpected_error
def _get_total_ram():
    import psutil

return psutil.virtual_memory().total // 2**20

_TAGS = None

def _tags():
    global _TAGS
    if _TAGS is None:
        from defence360agent.utils import OsReleaseInfo

_TAGS = {
            "av_version": None,
            "version": None,
            "os_details": stub_unexpected_error(OsReleaseInfo.pretty_name)(),
            "ip": None,
            "hosting_panel": None,
            "total_ram": _get_total_ram(),
            "firewall": None,
            "strategy": None,
            "virtualization": _get_virtualization_type(),
            "server_id": None,
            "iaid": None,
            "name": None,
            "test_build_id": None,
            "test_build_job_id": None,
            "test_parent_build_id": None,
        }
    return _TAGS

def set_firewall_type(firewall: str) -> None:
    _tags()["firewall"] = firewall

def set_hosting_panel(panel: str) -> None:
    _tags()["hosting_panel"] = panel

def set_strategy(strategy: str) -> None:
    _tags()["strategy"] = strategy

def set_ip(ip: str) -> None:
    _tags()["ip"] = ip

def set_product_name(product: str) -> None:
    _tags()["name"] = product

def set_server_id(id: str | None) -> None:
    _tags()["server_id"] = id

def set_iaid(iaid: str | None) -> None:
    _tags()["iaid"] = iaid

def set_version(version: str) -> None:
    _tags()["version"] = version

def set_av_version(version: str) -> None:
    _tags()["av_version"] = version

def tags() -> dict:
    return _tags().copy()

def tag(name: str) -> Any:
    return _tags()[name]

def set_test_env() -> None:
    """Set tags for sentry events about test environment."""
    for file_name, tag in [
        (
            Path("/var/imunify360/TEST_BUILD_ID"),
            "test_build_id",
        ),
        (
            Path("/var/imunify360/TEST_BUILD_JOB_ID"),
            "test_build_job_id",
        ),
        (
            Path("/var/imunify360/TEST_PARENT_BUILD_ID"),
            "test_parent_build_id",
        ),
    ]:
        if file_name.exists():
            try:
                _tags()[tag] = file_name.read_text().strip()
            except Exception:
                # Ignore errors on loading test env tags
                pass

MR SKB