mcpField
A
audit-log-generator

audit-log-generator

github.com/dengineproblem安全64 次安装
代码仓库

关于 audit-log-generator

用于设计安全与合规审计日志的技能，面向需要记录和审查系统活动的团队。内容涵盖事件结构、合规上下文、完整性控制、保留策略和监控。

主要功能

  1. 01结构化 JSON 与 CEF 日志格式
  2. 02认证、数据访问和管理事件
  3. 03GDPR、SOX 与 HIPAA 上下文
  4. 04完整性与访问控制建议
  5. 05保留策略、告警和报告

适用场景

  1. 01定义统一的审计事件字段
  2. 02记录受监管数据的访问
  3. 03规划防篡改日志控制
  4. 04制定保留与告警策略

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/dengineproblem/agents-monorepo --skill audit-log-generator

安装后在 Claude Code 中输入/audit-log-generator确认已加载。

详细介绍

Эксперт по генерации журналов аудита

Эксперт по проектированию и реализации журналов аудита для обеспечения безопасности, соответствия требованиям и мониторинга.

Основные элементы журнала

Каждая запись должна содержать:

  • Временная метка: UTC с точностью до миллисекунд
  • Идентификатор события: уникальный идентификатор
  • Субъект: кто выполнил действие
  • Действие: что было выполнено（CREATE、READ、UPDATE、DELETE）
  • Ресурс: на что было воздействие
  • Источник: IP-адрес, приложение
  • Результат: успех/неудача
  • Уровень риска: классификация критичности

Формат JSON（рекомендуется）

{
  "timestamp": "2024-01-15T14:30:45.123Z",
  "event_id": "evt_7f4a9b2c8e1d",
  "version": "1.0",
  "actor": {
    "user_id": "[email protected]",
    "session_id": "sess_abc123",
    "role": "admin"
  },
  "action": "DELETE",
  "resource": {
    "type": "database_record",
    "id": "customer_12345",
    "table": "customers",
    "classification": "PII"
  },
  "context": {
    "source_ip": "192.168.1.100",
    "user_agent": "Mozilla/5.0...",
    "application": "customer_portal",
    "api_endpoint": "/api/v1/customers/12345"
  },
  "result": {
    "status": "SUCCESS",
    "response_code": 200,
    "affected_records": 1
  },
  "metadata": {
    "risk_level": "HIGH",
    "compliance_tags": ["GDPR", "SOX"],
    "retention_years": 7,
    "checksum": "sha256:a1b2c3d4..."
  }
}

CEF（Common Event Format）

CEF:0|CompanyName|CustomerPortal|2.1|1001|User Data Deletion|8|
rt=Jan 15 2024 14:30:45 UTC src=192.168.1.100
[email protected] act=DELETE dst=customer_db
cs1Label=Table cs1=customers cs2Label=RecordID cs2=12345

Классификация событий

События аутентификации
  • Попытки входа（успех/неудача）
  • Смена паролей
  • Активация MFA
  • Блокировки аккаунтов
  • Повышение привилегий
События доступа к данным
{
  "action": "READ",
  "data_classification": "SENSITIVE",
  "access_method": "API",
  "record_count": 150,
  "query_hash": "sha256:..."
}
Административные события
  • Изменения конфигурации
  • Предоставление/отзыв доступа
  • Изменения разрешений
  • Операции резервного копирования
События безопасности
  • Неудачные попытки авторизации
  • Аномальные шаблоны доступа
  • Нарушения политик безопасности

Требования соответствия

GDPR
{
  "gdpr_context": {
    "lawful_basis": "legitimate_interest",
    "data_subject_id": "ds_789",
    "processing_purpose": "customer_service",
    "retention_justified": true,
    "cross_border_transfer": false
  }
}
SOX
  • Отслеживание доступа к финансовым данным
  • Разделение обязанностей
  • Документирование изменений
  • Мониторинг доступа руководства
HIPAA
{
  "hipaa_context": {
    "phi_involved": true,
    "minimum_necessary": true,
    "covered_entity": "hospital_system",
    "patient_authorization": "auth_456"
  }
}

Отслеживание изменений

{
  "change_tracking": {
    "before_state": {
      "customer_tier": "silver",
      "credit_limit": 5000
    },
    "after_state": {
      "customer_tier": "gold",
      "credit_limit": 10000
    },
    "change_reason": "promotion_campaign",
    "approver": "[email protected]"
  }
}

Пакетные операции

{
  "batch_context": {
    "batch_id": "batch_2024_01_15_001",
    "total_records": 10000,
    "successful_records": 9987,
    "failed_records": 13,
    "processing_duration_ms": 45678,
    "error_summary": ["validation_failed: 13"]
  }
}

Средства контроля безопасности

Шифрование
  • AES-256 для данных в состоянии покоя
  • TLS 1.3 для данных при передаче
  • Отдельные ключи для разных типов журналов
  • Регулярная ротация ключей
Контроль доступа
{
  "log_access_policy": {
    "read_access": ["audit_team", "compliance_officer"],
    "search_access": ["security_analyst"],
    "export_access": ["legal_team"],
    "retention_management": ["data_governance"]
  }
}
Обнаружение вмешательства
  • Дерево Меркла для обеспечения целостности
  • Цифровые подписи с использованием PKI
  • Регулярная проверка целостности
  • Неизменяемые временные метки

Реализация на Python

import json
import hashlib
from datetime import datetime
from typing import Dict, Any, Optional
from enum import Enum

class AuditAction(Enum):
    CREATE = "CREATE"
    READ = "READ"
    UPDATE = "UPDATE"
    DELETE = "DELETE"
    LOGIN = "LOGIN"
    LOGOUT = "LOGOUT"

class RiskLevel(Enum):
    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"
    CRITICAL = "CRITICAL"

class AuditLogger:
    def __init__(self, service_name: str):
        self.service_name = service_name

    def log(
        self,
        action: AuditAction,
        actor_id: str,
        resource_type: str,
        resource_id: str,
        result: str = "SUCCESS",
        risk_level: RiskLevel = RiskLevel.LOW,
        context: Optional[Dict[str, Any]] = None,
        before_state: Optional[Dict] = None,
        after_state: Optional[Dict] = None
    ) -> Dict[str, Any]:
        event_id = self._generate_event_id()
        timestamp = datetime.utcnow().isoformat() + "Z"

        log_entry = {
            "timestamp": timestamp,
            "event_id": event_id,
            "service": self.service_name,
            "actor": {"user_id": actor_id},
            "action": action.value,
            "resource": {
                "type": resource_type,
                "id": resource_id
            },
            "result": {"status": result},
            "metadata": {
                "risk_level": risk_level.value
            }
        }

        if context:
            log_entry["context"] = context

        if before_state or after_state:
            log_entry["change_tracking"] = {
                "before_state": before_state,
                "after_state": after_state
            }

        log_entry["metadata"]["checksum"] = self._calculate_checksum(log_entry)

        self._persist_log(log_entry)
        return log_entry

    def _generate_event_id(self) -> str:
        import uuid
        return f"evt_{uuid.uuid4().hex[:12]}"

    def _calculate_checksum(self, entry: Dict) -> str:
        content = json.dumps(entry, sort_keys=True)
        return f"sha256:{hashlib.sha256(content.encode()).hexdigest()[:16]}"

    def _persist_log(self, entry: Dict):
        # Отправка в систему логирования
        print(json.dumps(entry, indent=2))

Политики хранения

retention_policies = {
    "authentication": {"years": 3, "hot_storage_days": 90},
    "data_access": {"years": 7, "hot_storage_days": 365},
    "administrative": {"years": 10, "hot_storage_days": 180},
    "security_incidents": {"years": 10, "hot_storage_days": 1095}
}

Мониторинг и оповещения

Оповещения в реальном времени
  • Всплески неудачных попыток аутентификации
  • Шаблоны привилегированного доступа
  • Индикаторы эксфильтрации данных
  • Нарушения политик соответствия
Регулярная отчетность
  • Ежедневные сводки доступа
  • Еженедельные панели мониторинга соответствия
  • Ежемесячный анализ тенденций
  • Ежеквартальные отчеты о готовности к аудиту

Рекомендации

  1. Неизменяемость — журналы только для добавления
  2. Криптографическая защита — подписи и контрольные суммы
  3. Отдельное хранение — изолированное от операционных систем
  4. Регулярная проверка — контроль целостности
  5. Документация — цепочка поставки данных

评论

支持 Markdown