TaintPropagator (Главный движок)
├── DataflowTracker       - Прямое распространение dataflow
├── InterProcTracker      - Межпроцедурное отслеживание
├── ControlFlowAnalyzer   - Анализ зависимостей по управлению
├── FieldSensitiveTracker - Отслеживание полей структур
├── PointerAliasAnalyzer  - Обнаружение алиасов
└── PathConstraintTracker - Символьное выполнение

1. Загрузка domain-специфичных sources/sinks
2. Поиск узлов-источников (пользовательский ввод и т.д.)
3. Поиск узлов-приёмников (SQL, файловые операции и т.д.)
4. Распространение taint через dataflow рёбра
5. [НОВОЕ] Построение графа алиасов указателей (если включено)
6. [НОВОЕ] Фильтрация невыполнимых путей с помощью Z3 (если включено)
7. Расчёт оценок рисков и уверенности
8. Возврат отсортированных путей уязвимостей

pip install -r requirements.txt

pip install z3-solver

from src.analysis.dataflow import analyze_sql_injections
from src.services.cpg import CPGQueryService

with CPGQueryService() as cpg:
    # Анализ уязвимостей SQL-инъекций
    paths = analyze_sql_injections(
        cpg,
        max_paths=100,      # Максимум путей для возврата
        max_hops=15,        # Максимальная глубина dataflow
        min_confidence=0.7  # Минимальный порог уверенности
    )

    # Отображение результатов
    for path in paths:
        print(f"SQL-инъекция: {path.source_function} → {path.sink_function}")
        print(f"  Риск: {path.risk_level} ({path.risk_score:.2f})")
        print(f"  Уверенность: {path.confidence:.2f}")
        print(f"  Местоположение: {path.source_location} → {path.sink_location}")

from src.analysis.dataflow import analyze_sql_injections

paths = analyze_sql_injections(cpg)

for path in paths:
    # Source: откуда приходят ненадёжные данные
    print(f"Источник: {path.source_function} в {path.source_location}")

    # Sink: где они используются опасным образом
    print(f"Приёмник: {path.sink_function} в {path.sink_location}")

    # Sanitization: была ли правильная санитизация?
    if path.sanitizers:
        print(f"Санитайзеры: {', '.join(path.sanitizers)}")
        print(f"Оценка санитизации: {path.sanitization_score:.2f}")
    else:
        print("⚠️ Санитизация не обнаружена!")

    # Детали пути
    print(f"Длина пути: {path.path_length} переходов")
    if path.inter_procedural:
        print(f"Пройденные функции: {', '.join(path.functions_crossed)}")

from src.analysis.dataflow import analyze_command_injections

paths = analyze_command_injections(cpg)

for path in paths:
    print(f"Command Injection: {path.source_function} → {path.sink_function}")
    print(f"  Риск: {path.risk_level}")
    print(f"  Рекомендация: {path.recommendation}")

from src.analysis.dataflow.taint_analysis import TaintPropagator

with CPGQueryService() as cpg:
    # Создание propagator с пользовательскими настройками
    propagator = TaintPropagator(
        cpg,
        enable_inter_proc=True,        # Межпроцедурное отслеживание
        enable_control_flow=True,      # Анализ потока управления
        enable_field_sensitive=True,   # Field-sensitive отслеживание
        enable_symbolic_execution=True # Проверка выполнимости путей
    )

    # Поиск taint-путей
    paths = propagator.find_taint_paths(
        source_category='sql',
        max_paths=100,
        max_depth=15
    )

from src.analysis.dataflow import analyze_use_after_free
from src.services.cpg import CPGQueryService

with CPGQueryService() as cpg:
    # Анализ с включённым анализом алиасов указателей
    from src.analysis.dataflow.memory_lifetime import MemoryLifetimeAnalyzer

    analyzer = MemoryLifetimeAnalyzer(
        cpg,
        enable_alias_analysis=True  # НОВОЕ: Отслеживание алиасов указателей
    )

    paths = analyzer.analyze_use_after_free(
        max_paths=100,
        max_hops=15,
        min_confidence=0.6
    )

    for path in paths:
        print(f"Use-After-Free:")
        print(f"  Аллокация: {path.allocation_function} в {path.allocation_location}")
        print(f"  Освобождение: {path.free_function} в {path.free_location}")
        print(f"  Использование: {path.use_type} в {path.use_location}")
        print(f"  Указатель: {path.pointer_name}")
        print(f"  Риск: {path.risk_score:.2f}")
        print(f"  Уверенность: {path.confidence:.2f}")

from src.analysis.dataflow import analyze_double_free

with CPGQueryService() as cpg:
    analyzer = MemoryLifetimeAnalyzer(cpg)

    paths = analyzer.analyze_double_free(
        max_paths=50,
        min_confidence=0.6
    )

    for path in paths:
        print(f"Double-Free:")
        print(f"  Аллокация: {path.allocation_function} в {path.allocation_location}")
        print(f"  Первое освобождение: {path.first_free_location}")
        print(f"  Второе освобождение: {path.second_free_location}")
        print(f"  Риск: {path.risk_score:.2f}")

from src.analysis.dataflow.pointer_alias import PointerAliasAnalyzer

with CPGQueryService() as cpg:
    analyzer = PointerAliasAnalyzer(cpg)

    # Анализ функции на наличие алиасов указателей
    points_to = analyzer.analyze(function_id=123)

    # Проверка, могут ли два указателя быть алиасами
    if analyzer.may_alias(ptr1_id=456, ptr2_id=457):
        print("Указатели могут быть алиасами!")

        # Получение всех алиасов
        aliases = analyzer.get_aliases(ptr1_id=456)
        print(f"Алиасы: {aliases}")

        # Получение мест аллокации
        sites = analyzer.points_to_allocation(ptr1_id=456)
        for site in sites:
            print(f"Указывает на: {site.alloc_type} в {site.location}")

    # Получение статистики
    stats = analyzer.get_statistics()
    print(f"Проанализировано {stats['total_pointers']} указателей")
    print(f"Найдено {stats['total_allocations']} аллокаций")

from src.analysis.dataflow import analyze_null_dereferences

with CPGQueryService() as cpg:
    paths = analyze_null_dereferences(
        cpg,
        max_paths=100,
        max_hops=10,
        min_confidence=0.7
    )

    for path in paths:
        print(f"NULL-разыменование:")
        print(f"  Источник: {path.source_function}() в {path.source_location}")
        print(f"  Разыменование: {path.dereference_type} в {path.dereference_location}")
        print(f"  Есть проверка на NULL: {'✅' if path.has_null_check else '❌'}")
        print(f"  Риск: {path.risk_score:.2f}")
        print(f"  Переменная: {path.variable_name}")

from src.analysis.dataflow import analyze_info_disclosure

with CPGQueryService() as cpg:
    paths = analyze_info_disclosure(
        cpg,
        max_paths=100,
        max_hops=10,
        min_risk_score=0.6
    )

    for path in paths:
        print(f"Утечка информации:")
        print(f"  Категория данных: {path.data_category}")  # credentials, pii, secrets
        print(f"  Источник: {path.source_function} в {path.source_location}")
        print(f"  Приёмник: {path.sink_function} в {path.sink_location}")
        print(f"  Тип раскрытия: {path.disclosure_type}")  # error_message, debug_log
        print(f"  Серьёзность: {path.severity}")
        print(f"  Риск: {path.risk_score:.2f}")

from src.analysis.dataflow import analyze_race_conditions

with CPGQueryService() as cpg:
    paths = analyze_race_conditions(
        cpg,
        max_paths=100,
        max_hops=15,
        min_confidence=0.7
    )

    for path in paths:
        print(f"Состояние гонки:")
        print(f"  Тип: {path.race_type}")  # toctou, data_race, missing_lock
        print(f"  Проверка: {path.check_function} в {path.check_location}")
        print(f"  Использование: {path.use_function} в {path.use_location}")
        print(f"  Ресурс: {path.resource_name}")
        print(f"  Есть блокировка: {'✅' if path.has_lock else '❌'}")
        print(f"  Риск: {path.risk_score:.2f}")

from src.analysis.dataflow.symbolic_execution import PathConstraintTracker

with CPGQueryService() as cpg:
    tracker = PathConstraintTracker(cpg, max_constraints=20)

    # Проверка, выполним ли конкретный путь
    is_feasible, confidence = tracker.is_path_feasible(
        start_node=100,
        end_node=200
    )

    if not is_feasible and confidence > 0.9:
        print("Путь определённо невыполним (ложное срабатывание)")

    # Фильтрация невыполнимых путей из списка
    all_paths = analyze_sql_injections(cpg)
    feasible_paths = tracker.filter_infeasible_paths(
        all_paths,
        min_confidence=0.8
    )

    print(f"Отфильтровано {len(all_paths) - len(feasible_paths)} невыполнимых путей")

from src.domains import DomainRegistry

# Активация домена для вашего языка
DomainRegistry.activate("python_django")  # или: javascript, go, postgresql

# Анализ автоматически адаптируется к домену
paths = analyze_info_disclosure(cpg)
# Использует Python-специфичные sources/sinks (os.getenv, logger.error и т.д.)

paths = propagator.find_taint_paths(source_category='sql')

for path in paths:
    if path.field_sensitive:
        print(f"Заражённые поля: {', '.join(path.tainted_fields)}")
        # Пример: ['user.email', 'user.password']

for path in paths:
    if path.is_conditional:
        print(f"Зависимости по управлению: {len(path.control_dependencies)}")
        print(f"Вероятность выполнения: {path.execution_probability:.2f}")

        for dep in path.control_dependencies:
            print(f"  Зависит от: {dep.condition} в {dep.location}")

domain:
  name: python_django  # или: javascript, go, postgresql

# Отключение конкретных функций для производительности
propagator = TaintPropagator(
    cpg,
    enable_inter_proc=False,        # Отключить межпроцедурный (быстрее)
    enable_control_flow=False,      # Отключить поток управления (быстрее)
    enable_field_sensitive=False,   # Отключить field tracking (быстрее)
    enable_symbolic_execution=False # Отключить выполнимость путей (быстрее)
)

# Настройка глубины анализа
paths = propagator.find_taint_paths(
    max_paths=50,    # Меньше путей (быстрее)
    max_depth=10     # Меньшая глубина поиска (быстрее)
)

# Сравнение точности с и без улучшений
baseline_analyzer = MemoryLifetimeAnalyzer(cpg, enable_alias_analysis=False)
enhanced_analyzer = MemoryLifetimeAnalyzer(cpg, enable_alias_analysis=True)

baseline_paths = baseline_analyzer.analyze_use_after_free()
enhanced_paths = enhanced_analyzer.analyze_use_after_free()

print(f"Базовый: {len(baseline_paths)} багов найдено")
print(f"Улучшенный: {len(enhanced_paths)} багов найдено")
print(f"Дополнительно: {len(enhanced_paths) - len(baseline_paths)} (+{(len(enhanced_paths) - len(baseline_paths)) / len(baseline_paths) * 100:.1f}%)")

pip install z3-solver

propagator = TaintPropagator(cpg, enable_symbolic_execution=False)

# 1. Уменьшить max_paths
paths = analyzer.analyze_use_after_free(max_paths=50)  # по умолчанию: 100

# 2. Уменьшить max_depth
paths = analyzer.analyze_use_after_free(max_hops=10)  # по умолчанию: 15

# 3. Отключить дорогие функции
analyzer = MemoryLifetimeAnalyzer(cpg, enable_alias_analysis=False)

# 1. Отключить символьное выполнение (экономия 20% времени)
propagator = TaintPropagator(cpg, enable_symbolic_execution=False)

# 2. Отключить анализ алиасов указателей (экономия 50% времени)
analyzer = MemoryLifetimeAnalyzer(cpg, enable_alias_analysis=False)

# 3. Уменьшить глубину анализа
paths = propagator.find_taint_paths(max_depth=10)  # по умолчанию: 15

# Понизить порог уверенности
paths = analyze_sql_injections(cpg, min_confidence=0.5)  # по умолчанию: 0.7

# Проверить домен
from src.domains import get_active_domain
print(f"Активный домен: {get_active_domain().name}")

# Проверить sources/sinks
propagator._load_domain_data()
print(f"Источники: {len(propagator._taint_sources)}")
print(f"Приёмники: {len(propagator._taint_sinks)}")

# Taint-анализ
analyze_sql_injections(cpg, max_paths=100, max_hops=15, min_confidence=0.7)
analyze_command_injections(cpg, ...)
analyze_path_traversal(cpg, ...)

# Безопасность памяти
analyze_use_after_free(cpg, max_paths=100, max_hops=15, min_confidence=0.6)
analyze_double_free(cpg, ...)

# NULL-указатели
analyze_null_dereferences(cpg, max_paths=100, max_hops=10, min_confidence=0.7)

# Утечка информации
analyze_info_disclosure(cpg, max_paths=100, max_hops=10, min_risk_score=0.6)

# Состояния гонки
analyze_race_conditions(cpg, max_paths=100, max_hops=15, min_confidence=0.7)