Как отслеживать актуальность софта: чек-лист + автоматические алерты

<h2 id="telegram-bot-py">telegram_bot.py</h2>

from telegram import Bot

import os



async def send_critical_alert(component: str, cve_id: str, cvss: float, description: str):

    bot = Bot(token=os.environ['TELEGRAM_BOT_TOKEN'])

    chat_id = os.environ['TELEGRAM_CHAT_ID']

    

    message = f"🚨 *Критическое обновление*\n"

    message += f"*Компонент:* {component}\n"

    message += f"*CVE:* {cve_id}\n"

    message += f"*CVSS:* {cvss}\n"

    message += f"*Описание:* {description}\n"

    message += f"\n<a href="https://monitoring.example.com" target="_blank" rel="nofollow noopener noreferrer">Дашборд</a>\n"

    message += f"<a href="https://nvd.nist.gov/vuln/detail/{cve_id}" target="_blank" rel="nofollow noopener noreferrer">Advisory</a>"

    

    await bot.send_message(chat_id=chat_id, text=message, parse_mode='Markdown')

<h2 id="escalation-py">escalation.py</h2>

import time

from datetime import datetime, timedelta



class AlertEscalation:

    def __init__(self, alert_id: str, initial_recipients: list, escalation_delay_minutes: int = 30):

        self.alert_id = alert_id

        self.recipients = initial_recipients

        self.escalation_delay = timedelta(minutes=escalation_delay_minutes)

        self.created_at = datetime.now()

        self.confirmed = False

    

    def check_escalation(self):

        if not self.confirmed and datetime.now() - self.created_at >= self.escalation_delay:

            # Эскалация: добавить руководителя в список получателей

            self.recipients.append('manager@example.com')

            return True

        return False

    

    def confirm(self):

        self.confirmed = True

<h2 id="filters-yaml">filters.yaml</h2>

filters:

  # Игнорировать пре-релизы

  ignore_prereleases: true

  

  # Игнорировать обновления с незначительными изменениями

  ignore_patch_only:

    enabled: true

    min_cvss_for_patch: 7.0

  

  # Исключить компоненты в maintenance mode

  exclude_components:

    - "legacy-app-v1"

    - "internal-tool-deprecated"

  

  # Требовать подтверждение от ответственного перед эскалацией

  require_confirmation:

    enabled: true

    timeout_minutes: 60

<h2 id="test-alert-sh">test_alert.sh</h2>

curl -X POST http://alertmanager:9093/api/v2/alerts \

  -H 'Content-Type: application/json' \

  -d '[{

    "labels": {

      "alertname": "TestAlert",

      "severity": "critical",

      "component": "test-component"

    },

    "annotations": {

      "description": "Это тестовый алерт для проверки системы уведомлений",

      "summary": "Test Alert"

    }

  }]'

curl -s "https://services.nvd.nist.gov/rest/json/cves/2.0?pubStartDate=2026-01-01T00:00:00.000&cvssV3Severity=CRITICAL" \

  -H "apiKey: YOUR_NVD_API_KEY" \

  | jq '.vulnerabilities[] | {cveId: .cve.id, description: .cve.descriptions[0].value, cvss: .cve.metrics.cvssMetricV31[0].cvssData.baseScore}'

{

  securityVulnerabilities(first: 10, ecosystem: NPM, severity: CRITICAL) {

    nodes {

      package { name }

      advisory { ghsId, summary, severity }

      vulnerableVersionRange

      firstPatchedVersion { identifier }

    }

  }

}

<h2 id="sources-base-py">sources/base.py</h2>

from abc import ABC, abstractmethod



class VulnerabilitySource(ABC):

    @abstractmethod

    def fetch_updates(self, since: datetime) -> list:

        """Получить обновления уязвимостей с указанной даты"""

        pass

    

    @abstractmethod

    def normalize_entry(self, raw: dict) -> dict:

        """Привести запись к единому формату"""

        pass



<h2 id="sources-nvd-py">sources/nvd.py</h2>

class NVDSource(VulnerabilitySource):

    def __init__(self, api_key: str):

        self.api_key = api_key

        self.base_url = "https://services.nvd.nist.gov/rest/json/cves/2.0"

    

    def fetch_updates(self, since: datetime) -> list:

        params = {

            "pubStartDate": since.isoformat(),

            "cvssV3Severity": "CRITICAL,HIGH"

        }

        headers = {"apiKey": self.api_key}

        response = requests.get(self.base_url, params=params, headers=headers)

        return response.json().get("vulnerabilities", [])

    

    def normalize_entry(self, raw: dict) -> dict:

        cve = raw["cve"]

        return {

            "source": "NVD",

            "cve_id": cve["id"],

            "description": cve["descriptions"][0]["value"],

            "cvss_score": cve["metrics"]["cvssMetricV31"][0]["cvssData"]["baseScore"],

            "published": cve["published"],

            "cpe_configs": cve["configurations"]

        }

<h2 id="cache-py">cache.py</h2>

import redis

import json

from datetime import datetime, timedelta



class SourceCache:

    def __init__(self, redis_url: str, ttl_hours: int = 6):

        self.redis = redis.from_url(redis_url)

        self.ttl = timedelta(hours=ttl_hours)

    

    def get(self, key: str) -> Optional[list]:

        data = self.redis.get(f"source:{key}")

        if data:

            return json.loads(data)

        return None

    

    def set(self, key: str, data: list):

        self.redis.setex(f"source:{key}", self.ttl, json.dumps(data))

    

    def invalidate(self, key: str):

        self.redis.delete(f"source:{key}")

<h2 id="utils-retry-py">utils/retry.py</h2>

import time

from requests.exceptions import RequestException



def retry_request(func, max_attempts: int = 3, delay_seconds: int = 5):

    for attempt in range(max_attempts):

        try:

            return func()

        except RequestException as e:

            if attempt == max_attempts - 1:

                raise

            time.sleep(delay_seconds * (attempt + 1))  # exponential backoff

<h2 id="crontab">crontab</h2>

<h2 id="obnovlyat-nvd-kazhdye-4-chasa">Обновлять NVD каждые 4 часа</h2>

0 */4 * * * /opt/monitoring/sync_nvd.sh



<h2 id="obnovlyat-github-advisories-ezhednevno-v-2-00">Обновлять GitHub advisories ежедневно в 2:00</h2>

0 2 * * * /opt/monitoring/sync_ghsa.sh



<h2 id="generirovat-svodnyy-otchyot-kazhdoe-utro-v-8-00">Генерировать сводный отчёт каждое утро в 8:00</h2>

0 8 * * * /opt/monitoring/generate_report.sh

<h2 id="validators-py">validators.py</h2>

def validate_cve_entry(entry: dict) -> bool:

    required_fields = ["cve_id", "description", "cvss_score", "published"]

    return all(field in entry for field in required_fields) and \

           isinstance(entry["cvss_score"], (int, float)) and \

           0 <= entry["cvss_score"] <= 10

<h2 id="sources-registry-yaml">sources_registry.yaml</h2>

sources:

  nvd:

    url: https://nvd.nist.gov

    api_docs: https://nvd.nist.gov/developers

    update_frequency: "4 hours"

    authentication: "API key"

    contact: "nvd@nist.gov"

  

  ghsa:

    url: https://github.com/advisories

    api_docs: https://docs.github.com/en/graphql/reference/objects#securityvulnerability

    update_frequency: "daily"

    authentication: "GitHub token"

    contact: "https://github.com/contact"

<h2 id="parsers-vendor-parser-py">parsers/vendor_parser.py</h2>

from bs4 import BeautifulSoup

import requests

import re



def parse_vendor_releases(url: str, version_pattern: str) -> list:

    response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"})

    soup = BeautifulSoup(response.text, 'html.parser')

    

    releases = []

    for item in soup.select('.release-item'):  # адаптируйте селектор под структуру сайта

        title = item.select_one('.title').text.strip()

        version_match = re.search(version_pattern, title)

        if version_match:

            releases.append({

                "version": version_match.group(1),

                "title": title,

                "url": item.select_one('a')['href'],

                "published": item.select_one('.date').text

            })

    

    return releases

<h2 id="parsers-hidden-api-py">parsers/hidden_api.py</h2>

def fetch_hidden_api_data(product_id: str) -> dict:

    url = f"https://vendor.com/api/internal/products/{product_id}/releases"

    headers = {

        "Authorization": "Bearer GENERATED_TOKEN",  # может требовать аутентификации

        "X-Client-Version": "1.0"

    }

    response = requests.get(url, headers=headers)

    return response.json()

#!/bin/bash

<h2 id="check-internal-software-sh">check_internal_software.sh</h2>




<h2 id="proverka-vnutrenney-java-prilozheniya">Проверка внутренней Java-приложения</h2>

APP_DIR="/opt/internal-app"

CURRENT_VERSION=$(grep "app.version" $APP_DIR/config.properties | cut -d= -f2)

LATEST_VERSION=$(curl -s https://artifactory.internal/api/maven/internal-app/maven-metadata.xml \

  | xmllint --xpath "/*[local-name()='metadata']/*[local-name()='versioning']/*[local-name()='latest']/text()" -)



if [ "$CURRENT_VERSION" != "$LATEST_VERSION" ]; then

    echo "ALERT: internal-app requires update: $CURRENT_VERSION → $LATEST_VERSION"

    # Отправить уведомление

    /opt/monitoring/send_alert.sh "internal-app" "$CURRENT_VERSION" "$LATEST_VERSION"

fi

<h2 id="github-workflows-check-dependencies-yml">.github/workflows/check-dependencies.yml</h2>

name: Check Dependencies



on:

  schedule:

    - cron: '0 6 * * *'  # ежедневно в 6:00

  workflow_dispatch:



jobs:

  check:

    runs-on: ubuntu-latest

    steps:

      - uses: actions/checkout@v4

      

      - name: Check for outdated dependencies

        run: |

          npm outdated --json > outdated.json || true

          pip list --outdated --format=json > outdated-python.json || true

      

      - name: Create issue if updates available

        if: ${{ hashFiles('outdated.json') != '' || hashFiles('outdated-python.json') != '' }}

        uses: actions/github-script@v7

        with:

          script: |

            await github.rest.issues.create({

              owner: context.repo.owner,

              repo: context.repo.repo,

              title: 'Dependencies require update',

              body: 'Automated check found outdated dependencies. See artifacts for details.',

              labels: ['dependencies', 'security']

            })

<h2 id="utils-version-compare-py">utils/version_compare.py</h2>

import re

from packaging import version as pkg_version



def parse_version(v: str) -> pkg_version.Version:

    # Удалить префиксы: v1.2.3 → 1.2.3, release-2.0 → 2.0

    cleaned = re.sub(r'^(v|release-|version-)', '', v, flags=re.I)

    # Обработать суффиксы: 1.2.3-beta → 1.2.3b0

    cleaned = re.sub(r'-?(alpha|beta|rc)(\d*)', r'\1\2', cleaned, flags=re.I)

    return pkg_version.parse(cleaned)



def is_newer(current: str, available: str) -> bool:

    try:

        return parse_version(available) > parse_version(current)

    except:

        # Fallback на строковое сравнение

        return available > current

<h2 id="utils-logging-py">utils/logging.py</h2>

import logging

import sys



def setup_logger(name: str, log_file: str = None, level=logging.INFO):

    logger = logging.getLogger(name)

    logger.setLevel(level)

    

    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    

    # Консольный вывод

    console = logging.StreamHandler(sys.stdout)

    console.setFormatter(formatter)

    logger.addHandler(console)

    

    # Файловый лог (опционально)

    if log_file:

        file_handler = logging.FileHandler(log_file)

        file_handler.setFormatter(formatter)

        logger.addHandler(file_handler)

    

    return logger

<h2 id="tests-test-parsers-py">tests/test_parsers.py</h2>

import unittest

from parsers.vendor_parser import parse_vendor_releases



class TestVendorParser(unittest.TestCase):

    def test_parse_releases(self):

        # Моковый ответ

        mock_html = """

        <div class="release-item">

            <span class="title">Release v2.1.0</span>

            <span class="date">2026-01-15</span>

        </div>

        """

        with patch('requests.get') as mock_get:

            mock_get.return_value.text = mock_html

            result = parse_vendor_releases("http://test", r"v?(\d+\.\d+\.\d+)")

            self.assertEqual(len(result), 1)

            self.assertEqual(result[0]["version"], "2.1.0")

<h2 id="risk-scorer-py">risk/scorer.py</h2>

def calculate_risk_score(cve: dict, context: dict) -> dict:

    base_score = cve.get("cvss_score", 0)

    epss = cve.get("epss_score", 0)

    in_kev = cve.get("in_kev_catalog", False)

    

    # Базовый скор из CVSS

    score = base_score * 10  # масштабировать до 0-100

    

    # Бонус за EPSS

    if epss > 0.1:

        score += 20

    elif epss > 0.01:

        score += 10

    

    # Бонус за KEV

    if in_kev:

        score += 30

    

    # Контекстные факторы

    if context.get("internet_facing", False):

        score += 15

    if not context.get("workaround_available", True):

        score += 10

    

    # Нормализация

    final_score = min(100, score)

    

    # Определение уровня

    if final_score >= 80:

        level = "critical"

    elif final_score >= 60:

        level = "high"

    elif final_score >= 40:

        level = "medium"

    else:

        level = "low"

    

    return {

        "score": final_score,

        "level": level,

        "factors": {

            "cvss": base_score,

            "epss": epss,

            "kev": in_kev,

            "internet_facing": context.get("internet_facing", False)

        }

    }

<h2 id="queue-prioritizer-py">queue/prioritizer.py</h2>

def prioritize_updates(updates: list) -> list:

    return sorted(

        updates,

        key=lambda u: (

            -u["risk_score"],  # сначала высокий риск

            u["component"]     # затем по алфавиту для стабильности

        )

    )

<h2 id="integrations-jira-py">integrations/jira.py</h2>

def create_prioritized_ticket(component: str, risk: dict):

    priority_map = {

        "critical": "Highest",

        "high": "High",

        "medium": "Medium",

        "low": "Low"

    }

    

    payload = {

        "fields": {

            "project": {"key": "SEC"},

            "summary": f"Update {component} (Risk: {risk['level'].upper()})",

            "priority": {"name": priority_map[risk["level"]]},

            "customfield_10001": risk["score"],  # кастомное поле для скоринга

            "description": f"""

Риск: {risk['score']}/100 ({risk['level']})



Факторы:

- CVSS: {risk['factors']['cvss']}

- EPSS: {risk['factors']['epss']}

- KEV: {'Yes' if risk['factors']['kev'] else 'No'}

- Internet-facing: {'Yes' if risk['factors']['internet_facing'] else 'No'}



Рекомендуемые действия:

1. Протестировать обновление в staging

2. Применить в production в ближайшее окно обслуживания

3. Верифицировать функциональность

            """

        }

    }

    # Отправить в Jira...

<h2 id="reports-backlog-py">reports/backlog.py</h2>

def generate_backlog_report(pending_updates: list) -> str:

    report = "## Backlog of Pending Updates\n\n"

    

    # Группировка по уровню риска

    by_level = {}

    for update in pending_updates:

        level = update["risk_level"]

        by_level.setdefault(level, []).append(update)

    

    for level in ["critical", "high", "medium", "low"]:

        if level in by_level:

            report += f"### {level.upper()}\n"

            for update in by_level[level]:

                report += f"- {update['component']}: {update['current_version']} → {update['new_version']} (Risk: {update['risk_score']})\n"

            report += "\n"

    

    return report

<h2 id="crontab">crontab</h2>

<h2 id="pereschityvat-riski-kazhdye-6-chasov">Пересчитывать риски каждые 6 часов</h2>

0 */6 * * * /opt/monitoring/recalculate_risks.sh

<h2 id="playbooks-update-software-yml">playbooks/update_software.yml</h2>

- name: Update critical software

  hosts: production

  become: yes

  vars:

    component: "{{ ansible_facts['component'] }}"

    target_version: "{{ ansible_facts['target_version'] }}"

  

  tasks:

    - name: Check current version

      command: "{{ component }} --version"

      register: current_version

      changed_when: false

    

    - name: Skip if already up to date

      meta: end_play

      when: current_version.stdout == target_version

    

    - name: Backup current state

      command: /opt/backup/create.sh {{ component }}

    

    - name: Apply update

      package:

        name: "{{ component }}"

        state: "{{ target_version }}"

    

    - name: Verify service health

      uri:

        url: http://localhost/health

        status_code: 200

      register: health_check

      retries: 5

      delay: 10

      until: health_check.status == 200

    

    - name: Rollback if health check fails

      command: /opt/backup/restore.sh {{ component }}

      when: health_check.failed

<h2 id="k8s-rolling-update-yaml">k8s/rolling-update.yaml</h2>

apiVersion: apps/v1

kind: Deployment

metadata:

  name: my-app

spec:

  strategy:

    type: RollingUpdate

    rollingUpdate:

      maxSurge: 1

      maxUnavailable: 0  # нулевой даунтайм

  template:

    spec:

      containers:

      - name: app

        image: my-registry/app:v2.1.0  # обновить версию здесь

        readinessProbe:

          httpGet:

            path: /ready

            port: 8080

          initialDelaySeconds: 10

          periodSeconds: 5

<h2 id="terraform-update-ami-tf">terraform/update_ami.tf</h2>

resource "aws_launch_template" "app" {

  name = "app-template"

  

  image_id = data.aws_ami.ubuntu_latest.id  # обновить при выходе новой AMI

  

  update_default_version = true

  

  lifecycle {

    create_before_destroy = true

  }

}



data "aws_ami" "ubuntu_latest" {

  most_recent = true

  owners      = ["099720109477"]  # Canonical

  

  filter {

    name   = "name"

    values = ["ubuntu/images/hvm-ssd/ubuntu-jammy-22.04-amd64-server-*"]

  }

}

<h2 id="tests-smoke-test-py">tests/smoke_test.py</h2>

def test_after_update(component: str, endpoint: str):

    """Базовые проверки после обновления"""

    # Проверка доступности

    response = requests.get(f"{endpoint}/health", timeout=10)

    assert response.status_code == 200

    

    # Проверка версии

    version_response = requests.get(f"{endpoint}/version")

    assert version_response.json()["version"] == expected_version

    

    # Проверка базовой функциональности

    test_payload = {"test": "data"}

    result = requests.post(f"{endpoint}/api/test", json=test_payload)

    assert result.status_code == 200

    assert result.json()["status"] == "ok"

#!/bin/bash

<h2 id="rollback-sh">rollback.sh</h2>




COMPONENT=$1

PREVIOUS_VERSION=$2



echo "Rolling back $COMPONENT to $PREVIOUS_VERSION..."



<h2 id="ostanovit-servis">Остановить сервис</h2>

systemctl stop $COMPONENT



<h2 id="vosstanovit-predyduschuyu-versiyu">Восстановить предыдущую версию</h2>

apt-get install -y $COMPONENT=$PREVIOUS_VERSION



<h2 id="vosstanovit-konfiguratsiyu-iz-bekapa">Восстановить конфигурацию из бэкапа</h2>

cp /opt/backup/$COMPONENT/config.bak /etc/$COMPONENT/config



<h2 id="zapustit-servis">Запустить сервис</h2>

systemctl start $COMPONENT



<h2 id="proverit-status">Проверить статус</h2>

if systemctl is-active --quiet $COMPONENT; then

    echo "Rollback successful"

    exit 0

else

    echo "Rollback failed - service not running"

    exit 1

fi

<h2 id="prometheus-post-update-alerts-yaml">prometheus/post_update_alerts.yaml</h2>

groups:

  - name: post-update-monitoring

    rules:

      - alert: HighErrorRateAfterUpdate

        expr: rate(http_requests_total{status=~"5.."}[5m]) > 0.1

        for: 2m

        labels:

          severity: critical

          context: post-update

        annotations:

          summary: "High error rate detected after update"

          description: "Error rate is {{ $value }} errors/sec, threshold is 0.1"

      

      - alert: IncreasedLatencyAfterUpdate

        expr: histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m])) > 2

        for: 5m

        labels:

          severity: warning

          context: post-update

        annotations:

          summary: "Latency increased after update"

          description: "P95 latency is {{ $value }}s, threshold is 2s"

<h2 id="runbook-automated-software-updates">Runbook: Automated Software Updates</h2>




<h2 id="predusloviya">Предусловия</h2>

- [ ] Обновление протестировано в staging

- [ ] План отката проверен

- [ ] Окно обслуживания согласовано



<h2 id="protsess">Процесс</h2>

1. Запустить playbook: `ansible-playbook update_software.yml -e component=nginx -e target_version=1.24.0`

2. Мониторить дашборд Grafana на предмет аномалий

3. Если ошибки > 1% или latency > 2s — выполнить откат



<h2 id="otkat">Откат</h2>

1. Запустить: `/opt/monitoring/rollback.sh nginx 1.22.1`

2. Проверить восстановление функциональности

3. Создать инцидент для анализа



<h2 id="post-mortem">Пост-мортем</h2>

- Задокументировать причину проблемы

- Обновить тесты для предотвращения повторения

<h2 id="reports-generator-py">reports/generator.py</h2>

from datetime import datetime, timedelta

import pandas as pd



def generate_monthly_report(start_date: datetime, end_date: datetime) -> dict:

    # Загрузить данные из БД

    updates = fetch_updates_from_db(start_date, end_date)

    alerts = fetch_alerts_from_db(start_date, end_date)

    

    # Рассчитать метрики

    metrics = {

        "period": f"{start_date.date()} to {end_date.date()}",

        "total_components": len(get_all_components()),

        "monitored_components": len(get_monitored_components()),

        "coverage_percent": len(get_monitored_components()) / len(get_all_components()) * 100,

        

        "alerts_total": len(alerts),

        "alerts_confirmed": len([a for a in alerts if a["confirmed"]]),

        "false_positive_rate": 1 - len([a for a in alerts if a["confirmed"]]) / len(alerts),

        

        "updates_applied": len([u for u in updates if u["status"] == "applied"]),

        "updates_pending": len([u for u in updates if u["status"] == "pending"]),

        "avg_response_time_hours": calculate_avg_response_time(updates),

        

        "backlog_critical": len([u for u in updates if u["status"] == "pending" and u["risk_level"] == "critical"]),

        "backlog_high": len([u for u in updates if u["status"] == "pending" and u["risk_level"] == "high"]),

    }

    

    return metrics



def format_report(metrics: dict) -> str:

    report = f"""# Monthly Software Currency Report

Период: {metrics['period']}



<h2 id="pokrytie-monitoringa">Покрытие мониторинга</h2>

- Отслеживаемые компоненты: {metrics['monitored_components']}/{metrics['total_components']} ({metrics['coverage_percent']:.1f}%)



<h2 id="effektivnost-alertov">Эффективность алертов</h2>

- Всего алертов: {metrics['alerts_total']}

- Подтверждённые: {metrics['alerts_confirmed']}

- Ложные срабатывания: {metrics['false_positive_rate']*100:.1f}%



<h2 id="obnovleniya">Обновления</h2>

- Применено: {metrics['updates_applied']}

- В ожидании: {metrics['updates_pending']}

- Среднее время реакции: {metrics['avg_response_time_hours']:.1f} ч



<h2 id="beklog-po-riskam">Бэклог по рискам</h2>

- Критические: {metrics['backlog_critical']}

- Высокие: {metrics['backlog_high']}



<h2 id="rekomendatsii">Рекомендации</h2>

{generate_recommendations(metrics)}

"""

    return report

<h2 id="reports-exporters-py">reports/exporters.py</h2>

def export_to_pdf(report: str, output_path: str):

    from fpdf import FPDF

    pdf = FPDF()

    pdf.add_page()

    pdf.set_font("Arial", size=12)

    pdf.multi_cell(0, 10, report)

    pdf.output(output_path)



def export_to_html(report: str, output_path: str):

    import markdown

    html = markdown.markdown(report)

    with open(output_path, 'w') as f:

        f.write(f"<html><body>{html}</body></html>")



def export_to_json(metrics: dict, output_path: str):

    import json

    with open(output_path, 'w') as f:

        json.dump(metrics, f, indent=2, default=str)

<h2 id="integrations-grc-py">integrations/grc.py</h2>

def sync_to_servicenow(metrics: dict):

    """Синхронизация метрик с ServiceNow GRC"""

    url = "https://your-instance.service-now.com/api/now/table/sn_grc_metric"

    headers = {

        "Authorization": f"Bearer {os.environ['SNOW_TOKEN']}",

        "Content-Type": "application/json"

    }

    

    payload = {

        "metric_name": "software_currency",

        "reporting_period": metrics["period"],

        "coverage_percent": metrics["coverage_percent"],

        "critical_backlog": metrics["backlog_critical"],

        "avg_response_time": metrics["avg_response_time_hours"]

    }

    

    response = requests.post(url, json=payload, headers=headers)

    return response.status_code == 201

<h2 id="scripts-archive-report-sh">scripts/archive_report.sh</h2>

#!/bin/bash



REPORT_DATE=$(date +%Y-%m)

REPORT_DIR="/opt/reports/monthly"



<h2 id="sgenerirovat-otchyot">Сгенерировать отчёт</h2>

python3 /opt/monitoring/reports/generate.py --month $REPORT_DATE > $REPORT_DIR/report-$REPORT_DATE.md



<h2 id="eksport-v-formaty">Экспорт в форматы</h2>

python3 /opt/monitoring/reports/export.py --input $REPORT_DIR/report-$REPORT_DATE.md --formats pdf,html,json



<h2 id="kommit-v-git">Коммит в Git</h2>

cd /opt/reports

git add monthly/report-$REPORT_DATE.*

git commit -m "Add report for $REPORT_DATE"

git push



<h2 id="otpravit-po-email">Отправить по email</h2>

mail -s "Monthly Report $REPORT_DATE" -a $REPORT_DIR/report-$REPORT_DATE.pdf team@example.com < /dev/null

<h2 id="audit-checklist">Audit Checklist</h2>




<h2 id="pokrytie">Покрытие</h2>

- [ ] Все production-компоненты добавлены в мониторинг

- [ ] Внутренние разработки отслеживаются

- [ ] Legacy-системы имеют обоснованные исключения



<h2 id="svoevremennost">Своевременность</h2>

- [ ] Среднее время реакции на критические уязвимости < 24 ч

- [ ] Алерты доставляются в течение 5 минут

- [ ] Эскалация срабатывает при отсутствии подтверждения



<h2 id="kachestvo-dannyh">Качество данных</h2>

- [ ] Источники данных обновляются согласно расписанию

- [ ] Парсеры обработали изменения в структуре источников

- [ ] Нет дубликатов в базе уязвимостей



<h2 id="bezopasnost">Безопасность</h2>

- [ ] Доступ к системе мониторинга ограничен по ролям

- [ ] Логи аудита включены и защищёны от модификации

- [ ] Секреты хранятся в vault, не в коде

<h2 id="env-dobavit-v-gitignore">.env (добавить в .gitignore)</h2>

NVD_API_KEY=your_nvd_key_here

TELEGRAM_BOT_TOKEN=your_bot_token

SLACK_WEBHOOK_URL=https://hooks.slack.com/services/XXX/YYY/ZZZ

<h2 id="config-loader-py">config/loader.py</h2>

import os

from dotenv import load_dotenv



load_dotenv()  # загрузить .env



def get_config():

    return {

        "nvd_api_key": os.environ.get("NVD_API_KEY"),

        "telegram_bot_token": os.environ.get("TELEGRAM_BOT_TOKEN"),

        "slack_webhook": os.environ.get("SLACK_WEBHOOK_URL"),

        # ... другие настройки

    }

<h2 id="config-vault-loader-py">config/vault_loader.py</h2>

import hvac



def get_secret(path: str) -> str:

    client = hvac.Client(url=os.environ["VAULT_ADDR"], token=os.environ["VAULT_TOKEN"])

    secret = client.secrets.kv.v2.read_secret_version(path=path)

    return secret["data"]["data"]["value"]

<h2 id="utils-safe-logger-py">utils/safe_logger.py</h2>

import logging

import re



SENSITIVE_PATTERNS = [

    r'api[_-]?key[=:]\s*[\'"]?([a-zA-Z0-9_-]+)[\'"]?',

    r'token[=:]\s*[\'"]?([a-zA-Z0-9_-]+)[\'"]?',

    r'password[=:]\s*[\'"]?([^\'"\s]+)[\'"]?',

]



class SensitiveDataFilter(logging.Filter):

    def filter(self, record):

        if isinstance(record.msg, str):

            for pattern in SENSITIVE_PATTERNS:

                record.msg = re.sub(pattern, r'\1=*REDACTED*', record.msg, flags=re.I)

        return True



def setup_safe_logger(name: str):

    logger = logging.getLogger(name)

    logger.addFilter(SensitiveDataFilter())

    return logger

<h2 id="rbac-yaml">rbac.yaml</h2>

roles:

  viewer:

    permissions:

      - read:dashboard

      - read:reports

    resources: ["*"]

  

  operator:

    permissions:

      - read:*

      - update:components

      - trigger:alerts

    resources: ["*"]

  

  admin:

    permissions:

      - "*"

    resources: ["*"]



users:

  - name: "analyst@company.com"

    role: viewer

  - name: "admin@company.com"

    role: admin

<h2 id="validators-input-py">validators/input.py</h2>

def validate_component_name(name: str) -> bool:

    """Разрешить только безопасные символы"""

    return bool(re.match(r'^[a-zA-Z0-9._-]+$', name)) and len(name) <= 100



def validate_version(version: str) -> bool:

    """Проверить формат версии"""

    return bool(re.match(r'^v?(\d+\.){1,3}\d+(-[a-zA-Z0-9.]+)?$', version))

<h2 id="scripts-security-scan-sh">scripts/security_scan.sh</h2>

#!/bin/bash



<h2 id="skanirovanie-zavisimostey-na-uyazvimosti">Сканирование зависимостей на уязвимости</h2>

pip-audit -r requirements.txt

npm audit --audit-level=high



<h2 id="staticheskiy-analiz-koda">Статический анализ кода</h2>

bandit -r /opt/monitoring/

semgrep --config=auto /opt/monitoring/



<h2 id="proverka-konfiguratsiy">Проверка конфигураций</h2>

checkov -d /opt/monitoring/config/



<h2 id="otchyot">Отчёт</h2>

echo "Security scan completed $(date)" >> /var/log/monitoring/security_scans.log

<h2 id="incident-response-monitoring-system-compromise">Incident Response: Monitoring System Compromise</h2>




<h2 id="priznaki-komprometatsii">Признаки компрометации</h2>

- Необычная активность в логах доступа

- Изменения в конфигурации без согласования

- Утечка секретов в публичные репозитории



<h2 id="nemedlennye-deystviya">Немедленные действия</h2>

1. Изолировать систему мониторинга от сети

2. Сменить все секреты, используемые системой

3. Проверить логи на предмет несанкционированного доступа

4. Уведомить команду безопасности



<h2 id="vosstanovlenie">Восстановление</h2>

1. Развернуть чистую версию из бэкапа

2. Проверить целостность конфигураций

3. Повторно настроить интеграции с новыми секретами

4. Провести пост-мортем анализ



<h2 id="predotvraschenie">Предотвращение</h2>

- Включить MFA для доступа к системе

- Настроить алерты на подозрительную активность

- Регулярно ротировать секреты

<h2 id="docker-compose-yml-dlya-open-source-monitoringa">docker-compose.yml для open-source мониторинга</h2>

version: '3.8'

services:

  trivy:

    image: aquasec/trivy:latest

    command: server --listen 0.0.0.0:4954

    volumes:

      - trivy-cache:/root/.cache

  

  postgres:

    image: postgres:15

    environment:

      POSTGRES_DB: monitoring

      POSTGRES_PASSWORD_FILE: /run/secrets/db_password

    secrets:

      - db_password

    volumes:

      - pgdata:/var/lib/postgresql/data

  

  grafana:

    image: grafana/grafana:latest

    ports:

      - "3000:3000"

    environment:

      - GF_SECURITY_ADMIN_PASSWORD_FILE=/run/secrets/grafana_password

    secrets:

      - grafana_password

  

  alertmanager:

    image: prom/alertmanager:latest

    command: --config.file=/etc/alertmanager/config.yml

    volumes:

      - ./alertmanager:/etc/alertmanager



volumes:

  trivy-cache:

  pgdata:



secrets:

  db_password:

    file: ./secrets/db_password.txt

  grafana_password:

    file: ./secrets/grafana_password.txt