Парсеры и чекеры для глубокого аудита безопасности. Автоматизация поиска уязвимостей и инвентаризации активов

#!/usr/bin/env python3
"""
privilege_auditor.py
Полный аудит опасных привилегий пользователей и групп
Security Audit Tool

зависимости: pip install pywin32
запускать: от Domain Admin или с правами на чтение AD
"""

import win32security
import win32api
import win32net
import win32con
import subprocess
import json
import sys
from datetime import datetime


# привилегии которые дают LPE или критический доступ
DANGEROUS_PRIVS = {
    "SeDebugPrivilege":           "CRITICAL — чтение любого процесса включая lsass",
    "SeImpersonatePrivilege":     "CRITICAL — Potato LPE, получение SYSTEM токена",
    "SeAssignPrimaryTokenPrivilege": "CRITICAL — назначение токена процессу",
    "SeBackupPrivilege":          "HIGH — обход ACL для чтения, включая NTDS.dit",
    "SeRestorePrivilege":         "HIGH — обход ACL для записи, замена файлов",
    "SeTakeOwnershipPrivilege":   "HIGH — смена владельца любого объекта",
    "SeLoadDriverPrivilege":      "HIGH — загрузка произвольного kernel driver",
    "SeCreateTokenPrivilege":     "CRITICAL — создание произвольного токена",
    "SeTcbPrivilege":             "CRITICAL — 'Act as part of OS', полный доступ",
    "SeSecurityPrivilege":        "HIGH — управление аудитом и security log",
    "SeNetworkLogonRight":        "MEDIUM — вход по сети, интересно для pivot",
    "SeRemoteInteractiveLogonRight": "MEDIUM — RDP доступ",
}


def get_user_privileges_via_secedit() -> dict:
    """получаем назначение привилегий через secedit экспорт"""
    import tempfile, os

    tmpfile = tempfile.mktemp(suffix=".inf")
    try:
        subprocess.run(
            ["secedit", "/export", "/cfg", tmpfile, "/quiet"],
            capture_output=True, check=True
        )

        priv_map = {}  # privilege -> [accounts]

        with open(tmpfile, 'r', encoding='utf-16') as f:
            in_privs = False
            for line in f:
                line = line.strip()
                if line == "[Privilege Rights]":
                    in_privs = True
                    continue
                if line.startswith("[") and in_privs:
                    break
                if in_privs and "=" in line:
                    priv, accounts_raw = line.split("=", 1)
                    priv = priv.strip()
                    accounts = [a.strip().lstrip("*") for a in accounts_raw.split(",")]
                    priv_map[priv] = accounts

        return priv_map
    finally:
        try: os.remove(tmpfile)
        except: pass


def resolve_sid_to_name(sid_str: str) -> str:
    """резолвим SID в имя аккаунта"""
    try:
        sid = win32security.ConvertStringSidToSid(sid_str)
        name, domain, _ = win32security.LookupAccountSid(None, sid)
        return f"{domain}\\{name}" if domain else name
    except:
        return sid_str


def audit_privileges():
    print("=" * 60)
    print("AUDIT: Dangerous Privilege Assignments")
    print(f"Время: {datetime.now().isoformat()}")
    print("=" * 60)

    priv_map = get_user_privileges_via_secedit()
    findings = []

    for priv, description in DANGEROUS_PRIVS.items():
        if priv not in priv_map:
            continue

        accounts = priv_map[priv]
        resolved = []

        for acc in accounts:
            if not acc:
                continue
            # SID или имя
            if acc.startswith("S-1-"):
                resolved.append(resolve_sid_to_name(acc))
            else:
                resolved.append(acc)

        # фильтруем ожидаемые системные аккаунты
        expected_system = {
            "NT AUTHORITY\\SYSTEM",
            "NT AUTHORITY\\LOCAL SERVICE",
            "NT AUTHORITY\\NETWORK SERVICE",
            "BUILTIN\\Administrators",
            "Window Manager\\Window Manager Group",
        }

        suspicious = [a for a in resolved if a not in expected_system]

        if suspicious:
            severity = "CRITICAL" if "CRITICAL" in description else "HIGH"
            findings.append({
                "privilege":   priv,
                "severity":    severity,
                "description": description,
                "accounts":    suspicious
            })
            print(f"\n[{severity}] {priv}")
            print(f"  Описание: {description}")
            print(f"  Назначено: {', '.join(suspicious)}")

    print(f"\n{'='*60}")
    print(f"Итого подозрительных назначений: {len(findings)}")
    if findings:
        print("[!!] все CRITICAL — в backlog немедленно")
    return findings


if __name__ == "__main__":
    audit_privileges()

#!/usr/bin/env python3
"""
kerberos_delegation_audit.py
Аудит Kerberos Delegation в Active Directory
Security Audit Tool

зависимости: pip install ldap3
запускать: с правами чтения AD (любой domain user достаточно для чтения)
"""

from ldap3 import Server, Connection, ALL, NTLM, ALL_ATTRIBUTES
from ldap3.core.exceptions import LDAPException
import argparse
import sys


# флаги userAccountControl
UAC_TRUSTED_FOR_DELEGATION           = 0x80000      # Unconstrained
UAC_NOT_DELEGATED                    = 0x100000
UAC_TRUSTED_TO_AUTH_FOR_DELEGATION  = 0x1000000     # Constrained (любой протокол)


def get_base_dn(domain: str) -> str:
    return ",".join(f"DC={part}" for part in domain.split("."))


def audit_delegation(server_addr: str, domain: str, username: str, password: str):
    base_dn = get_base_dn(domain)

    print(f"[*] подключаемся к {server_addr}")
    server = Server(server_addr, get_info=ALL)

    try:
        conn = Connection(
            server,
            user=f"{domain}\\{username}",
            password=password,
            authentication=NTLM,
            auto_bind=True
        )
    except LDAPException as e:
        print(f"[-] ошибка подключения: {e}")
        sys.exit(1)

    print(f"[+] подключились, base DN: {base_dn}\n")

    # ── 1. Unconstrained Delegation (самое опасное) ────────────────────────────
    print("[*] ищем Unconstrained Delegation объекты...")

    conn.search(
        base_dn,
        f"(userAccountControl:1.2.840.113556.1.4.803:={UAC_TRUSTED_FOR_DELEGATION})",
        attributes=["sAMAccountName", "userAccountControl", "operatingSystem",
                    "distinguishedName", "lastLogonTimestamp", "objectClass"]
    )

    unconstrained = conn.entries
    if unconstrained:
        print(f"\n[CRITICAL] UNCONSTRAINED DELEGATION — {len(unconstrained)} объектов:\n")
        for obj in unconstrained:
            obj_type = "COMPUTER" if "computer" in str(obj.objectClass).lower() else "USER"
            print(f"  [{obj_type}] {obj.sAMAccountName}")
            print(f"    DN: {obj.distinguishedName}")
            if obj.operatingSystem:
                print(f"    OS: {obj.operatingSystem}")
            print()
    else:
        print("  [OK] Unconstrained Delegation объектов нет\n")

    # ── 2. Constrained Delegation ───────────────────────────────────────────────
    print("[*] ищем Constrained Delegation объекты...")

    conn.search(
        base_dn,
        "(msDS-AllowedToDelegateTo=*)",
        attributes=["sAMAccountName", "msDS-AllowedToDelegateTo",
                    "userAccountControl", "distinguishedName", "objectClass"]
    )

    constrained = conn.entries
    if constrained:
        print(f"\n[HIGH] CONSTRAINED DELEGATION — {len(constrained)} объектов:\n")
        for obj in constrained:
            protocol_transition = bool(
                int(str(obj.userAccountControl)) & UAC_TRUSTED_TO_AUTH_FOR_DELEGATION
            )
            obj_type = "COMPUTER" if "computer" in str(obj.objectClass).lower() else "USER"

            print(f"  [{obj_type}] {obj.sAMAccountName}")
            print(f"    Protocol Transition (любой протокол): "
                  f"{'ДА — опаснее' if protocol_transition else 'нет'}")
            print(f"    Может делегировать к:")
            for target in obj["msDS-AllowedToDelegateTo"]:
                print(f"      - {target}")
            print()
    else:
        print("  [OK] Constrained Delegation объектов нет\n")

    # ── 3. Resource-Based Constrained Delegation (RBCD) ────────────────────────
    print("[*] ищем Resource-Based Constrained Delegation...")

    conn.search(
        base_dn,
        "(msDS-AllowedToActOnBehalfOfOtherIdentity=*)",
        attributes=["sAMAccountName", "distinguishedName",
                    "msDS-AllowedToActOnBehalfOfOtherIdentity", "objectClass"]
    )

    rbcd = conn.entries
    if rbcd:
        print(f"\n[HIGH] RESOURCE-BASED CONSTRAINED DELEGATION — {len(rbcd)} объектов:\n")
        for obj in rbcd:
            obj_type = "COMPUTER" if "computer" in str(obj.objectClass).lower() else "USER"
            print(f"  [{obj_type}] {obj.sAMAccountName}")
            print(f"    DN: {obj.distinguishedName}")
            print(f"    Имеет msDS-AllowedToActOnBehalfOfOtherIdentity — требует ручной проверки")
            print()
    else:
        print("  [OK] RBCD объектов нет\n")

    # ── 4. krbtgt последняя смена пароля ───────────────────────────────────────
    print("[*] проверяем krbtgt аккаунт...")

    conn.search(
        base_dn,
        "(sAMAccountName=krbtgt)",
        attributes=["pwdLastSet", "sAMAccountName"]
    )

    if conn.entries:
        krbtgt = conn.entries[0]
        pwd_last_set = krbtgt.pwdLastSet.value if krbtgt.pwdLastSet else None
        if pwd_last_set:
            from datetime import datetime, timezone
            now = datetime.now(timezone.utc)
            delta = now - pwd_last_set
            days = delta.days

            print(f"\n  krbtgt пароль последний раз менялся: {days} дней назад")
            if days > 180:
                print(f"  [HIGH] {days} дней — слишком давно, меняй (рекомендация: каждые 180 дней)")
                print(f"  [HIGH] после смены: сменить дважды с паузой в репликацию")
            else:
                print(f"  [OK] {days} дней — приемлемо")

    conn.unbind()

    print("\n" + "="*60)
    print("DELEGATION AUDIT ЗАВЕРШЁН")
    print("CRITICAL объекты с Unconstrained Delegation — немедленный приоритет")
    print("особенно если это не DC — это аномалия")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--dc",       required=True, help="IP или FQDN Domain Controller")
    parser.add_argument("--domain",   required=True, help="домен, например corp.local")
    parser.add_argument("--user",     required=True, help="имя пользователя")
    parser.add_argument("--password", required=True, help="пароль")
    args = parser.parse_args()

    audit_delegation(args.dc, args.domain, args.user, args.password)

#!/usr/bin/env python3
"""
patch_gap_scanner.py
Сканер отсутствующих патчей безопасности
Security Audit Tool

что делает:
  1. собирает установленные KB через WMI
  2. сравнивает с критическими KB из hardcoded списка (актуален 2024-2025)
  3. проверяет версию OS и UBR
  4. генерирует приоритизированный отчёт

зависимости: pip install pywin32 wmi requests
"""

import wmi
import subprocess
import json
import re
import sys
from datetime import datetime


# критические KB которые должны быть установлены (2024-2025, LPE и RCE)
# источник: Microsoft Security Update Guide + CISA KEV catalog
CRITICAL_KB = {
    # LPE патчи
    "KB5035853": {
        "cve": "CVE-2024-21338",
        "type": "LPE",
        "severity": "CRITICAL",
        "desc": "Windows Kernel Elevation via appid.sys UAF (Lazarus)",
        "affected_builds": [19045, 22631, 22000, 20348, 17763]
    },
    "KB5034441": {
        "cve": "CVE-2024-20666",
        "type": "LPE",
        "severity": "HIGH",
        "desc": "BitLocker Recovery Mode Bypass — EoP",
        "affected_builds": [19041, 19044, 19045]
    },
    "KB5037771": {
        "cve": "CVE-2024-30051",
        "type": "LPE",
        "severity": "CRITICAL",
        "desc": "DWM Core Library Heap Buffer Overflow — SYSTEM",
        "affected_builds": [19045, 22631, 20348, 17763]
    },
    "KB5040442": {
        "cve": "CVE-2024-38080",
        "type": "LPE",
        "severity": "CRITICAL",
        "desc": "Windows Hyper-V Elevation of Privilege — эксплуатируется in-the-wild",
        "affected_builds": [22631, 22000, 20348]
    },
    # RCE патчи
    "KB5034122": {
        "cve": "CVE-2024-21413",
        "type": "RCE",
        "severity": "CRITICAL",
        "desc": "Microsoft Outlook RCE через Preview Pane (MonikerLink)",
        "affected_builds": None  # все версии Office
    },
    "KB5037765": {
        "cve": "CVE-2024-29988",
        "type": "RCE",
        "severity": "HIGH",
        "desc": "SmartScreen Prompt Security Feature Bypass",
        "affected_builds": [19045, 22631, 22000]
    },
    # NTLM / Auth патчи
    "KB5040434": {
        "cve": "CVE-2024-38077",
        "type": "RCE",
        "severity": "CRITICAL",
        "desc": "Windows Remote Desktop Licensing Service RCE — wormable",
        "affected_builds": [20348, 17763, 14393]
    },
}


def get_installed_kbs() -> set:
    """собираем установленные KB через wmic qfe"""
    result = subprocess.run(
        ["wmic", "qfe", "get", "HotFixID", "/format:csv"],
        capture_output=True, text=True
    )

    kbs = set()
    for line in result.stdout.splitlines():
        # формат: Node,HotFixID
        parts = line.strip().split(",")
        if len(parts) >= 2:
            kb = parts[-1].strip()
            if kb.startswith("KB"):
                kbs.add(kb)

    return kbs


def get_os_build() -> int:
    """получаем номер билда OS"""
    c = wmi.WMI()
    for os_info in c.Win32_OperatingSystem():
        build_str = os_info.BuildNumber
        try:
            return int(build_str)
        except:
            return 0
    return 0


def scan_missing_patches():
    print("=" * 65)
    print("PATCH GAP SCANNER — Critical Security Updates Audit")
    print(f"Время: {datetime.now().isoformat()}")
    print("=" * 65)

    print("\n[*] собираем установленные KB...")
    installed = get_installed_kbs()
    print(f"  найдено установленных обновлений: {len(installed)}")

    os_build = get_os_build()
    print(f"  OS Build: {os_build}\n")

    missing = []

    for kb, info in CRITICAL_KB.items():
        # проверяем применимость к текущей OS
        if info["affected_builds"] is not None:
            if os_build not in info["affected_builds"]:
                continue

        if kb not in installed:
            missing.append({
                "kb":       kb,
                "cve":      info["cve"],
                "type":     info["type"],
                "severity": info["severity"],
                "desc":     info["desc"]
            })

    # вывод по приоритету
    critical_missing = [m for m in missing if m["severity"] == "CRITICAL"]
    high_missing     = [m for m in missing if m["severity"] == "HIGH"]

    if critical_missing:
        print(f"[!!!] КРИТИЧЕСКИЕ ПАТЧИ ОТСУТСТВУЮТ: {len(critical_missing)}\n")
        for m in critical_missing:
            print(f"  [CRITICAL] {m['kb']} — {m['cve']}")
            print(f"    Тип: {m['type']}")
            print(f"    Описание: {m['desc']}\n")

    if high_missing:
        print(f"[!] ВАЖНЫЕ ПАТЧИ ОТСУТСТВУЮТ: {len(high_missing)}\n")
        for m in high_missing:
            print(f"  [HIGH] {m['kb']} — {m['cve']}")
            print(f"    Тип: {m['type']}")
            print(f"    Описание: {m['desc']}\n")

    if not missing:
        print("[OK] все критические патчи установлены")

    print("=" * 65)
    print(f"ИТОГ: пропущено CRITICAL={len(critical_missing)}, HIGH={len(high_missing)}")
    if critical_missing:
        print("[!!!] машина уязвима к известным эксплойтам, патчить НЕМЕДЛЕННО")

    return missing


if __name__ == "__main__":
    scan_missing_patches()

#!/usr/bin/env python3
"""
smb_share_auditor.py
Аудит прав доступа на SMB-шарах домена
Security Audit Tool

что делает:
  1. перечисляет шары через net view / WMI
  2. проверяет доступность для Everyone / Authenticated Users
  3. ищет шары с write access для обычных пользователей
  4. флагирует административные шары (ADMIN$, C$) если доступны

зависимости: pip install pywin32 wmi impacket
"""

import subprocess
import os
import sys
import wmi
from pathlib import Path


DANGEROUS_PERMISSIONS = [
    "Everyone",
    "Authenticated Users",
    "Domain Users",
    "BUILTIN\\Users",
]


def enum_local_shares() -> list[dict]:
    """перечисляем локальные шары через WMI"""
    c = wmi.WMI()
    shares = []

    for share in c.Win32_Share():
        shares.append({
            "name":     share.Name,
            "path":     share.Path,
            "type":     share.Type,
            "caption":  share.Caption,
        })

    return shares


def check_share_acl(share_name: str) -> list[dict]:
    """получаем ACL шары через icacls на локальном пути"""
    c = wmi.WMI()
    findings = []

    for share in c.Win32_Share():
        if share.Name != share_name:
            continue

        if not share.Path:
            break

        path = share.Path
        if not os.path.exists(path):
            break

        result = subprocess.run(
            ["icacls", path],
            capture_output=True, text=True, encoding="cp866"
        )

        for line in result.stdout.splitlines():
            for dangerous_principal in DANGEROUS_PERMISSIONS:
                if dangerous_principal.lower() in line.lower():
                    # разбираем права
                    permissions = line.strip()
                    is_write = any(p in permissions for p in ["(W)", "(M)", "(F)", "(OI)(CI)(W)"])
                    findings.append({
                        "principal":   dangerous_principal,
                        "permissions": permissions,
                        "has_write":   is_write
                    })
        break

    return findings


def audit_smb_shares():
    print("=" * 60)
    print("SMB SHARE PERMISSION AUDIT")
    print("=" * 60)

    shares = enum_local_shares()
    print(f"\n[*] найдено шар: {len(shares)}\n")

    risky_shares = []

    for share in shares:
        name = share["name"]
        path = share["path"] or "(нет пути)"
        stype = share["type"]

        # 0 = Disk Share, 2147483648 = Admin Share
        is_admin_share = (stype == 2147483648 or name.endswith("$"))

        print(f"  [{name}] path={path} admin={is_admin_share}")

        acl_findings = check_share_acl(name)

        for finding in acl_findings:
            risk = "CRITICAL" if finding["has_write"] else "MEDIUM"
            print(f"    [{risk}] {finding['principal']}: {finding['permissions']}")
            risky_shares.append({
                "share":     name,
                "path":      path,
                "principal": finding["principal"],
                "write":     finding["has_write"],
                "risk":      risk
            })

    print(f"\n{'='*60}")
    write_access = [r for r in risky_shares if r["write"]]
    print(f"Шар с write-доступом для широких групп: {len(write_access)}")
    if write_access:
        print("[!!] шары с write для Everyone/Domain Users — вектор для DLL planting")
        print("[!!] и плацдарм для lateral movement через GPO/script замену")

    return risky_shares


if __name__ == "__main__":
    audit_smb_shares()

// детект массового LDAP-перечисления — признак AD разведки
// источник: Azure AD / Defender for Identity

IdentityQueryEvents
| where Timestamp > ago(1d)
| where QueryType in ("Ldap", "ActiveDirectory")
| where QueryTarget has_any (
    "userAccountControl", "msDS-AllowedToDelegateTo",
    "servicePrincipalName", "adminCount", "member"
)
| summarize QueryCount = count(),
            Targets = make_set(QueryTarget)
            by AccountName, DeviceName, bin(Timestamp, 5m)
| where QueryCount > 50  // порог — настроить под baseline
| order by QueryCount desc

// детект secedit /export — классический шаг аудита привилегий
// используется и атакующими для разведки

DeviceProcessEvents
| where Timestamp > ago(7d)
| where FileName =~ "secedit.exe"
| where ProcessCommandLine has "/export"
| where InitiatingProcessFileName !in~ (
    "msc",               // MMC Group Policy
    "gpupdate.exe",
    "gpsvc.dll"
)
| project Timestamp, DeviceName, AccountName,
          ProcessCommandLine, InitiatingProcessFileName

// детект wmic qfe — сканирование установленных обновлений
// атакующий делает это для patch gap разведки

DeviceProcessEvents
| where Timestamp > ago(7d)
| where FileName =~ "wmic.exe"
| where ProcessCommandLine has "qfe"
| where InitiatingProcessFileName !in~ (
    "wsus", "wusa.exe", "WindowsUpdateAgent.exe",
    "TiWorker.exe", "TrustedInstaller.exe"
)
| project Timestamp, DeviceName, AccountName,
          ProcessCommandLine, InitiatingProcessFileName