portal/app/app/Console/Commands/VerifyAuditChains.php

<?php

declare(strict_types=1);

namespace App\Console\Commands;

use App\Mail\AuditChainBreachMail;
use App\Services\Audit\AuditChainConfig;
use Illuminate\Console\Command;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Mail;

/**
 * Проверяет целостность SHA-256 hash-chain во всех 6 audit-таблицах.
 *
 * Алгоритм на стороне PostgreSQL (не PHP) — чтобы воспроизвести ровно ту же
 * сериализацию ROW::text, что использует триггер audit_chain_hash():
 *
 *   digest(COALESCE(prev_log_hash,''::bytea) || ROW(col1,...,NULL::bytea,...col_n)::text::bytea, 'sha256')
 *
 * где NULL::bytea — позиция log_hash (она была NULL в момент срабатывания
 * BEFORE INSERT триггера). Список столбцов в порядке их ordinal_position
 * из information_schema жёстко закодирован для каждой таблицы.
 *
 * ──────────────────────────────────────────────────────────────────────────────
 * ВАЖНО: per-partition scan (hole #2 adaptation).
 *
 * После перевода таблиц на RANGE-партиционирование (v8.31) каждая партиция
 * содержит строки одного месяца. Триггер audit_chain_hash() при INSERT в
 * партицию видит строки только ЭТОЙ партиции (TG_TABLE_NAME = partition name,
 * SELECT LAG по partition → prev — последняя запись той же партиции).
 *
 * Поэтому валидатор проверяет hash-chain отдельно для каждой партиции:
 *   1. Получает список партиций через pg_inherits + pg_class.
 *   2. Для каждой партиции выполняет checkPartition().
 *   3. Несоответствие в ЛЮБОЙ партиции → инцидент с указанием partition_name.
 *
 * Пустые партиции (без строк) — OK, chain пустая = intact.
 *
 * ──────────────────────────────────────────────────────────────────────────────
 * ВАЖНО: per-scope RLS partitioning.
 *
 * Триггер audit_chain_hash() делает:
 *   SELECT log_hash FROM <table> ORDER BY id DESC LIMIT 1
 * Этот SELECT выполняется под ролью вставляющей сессии и подпадает под RLS.
 *
 * После партиционирования SELECT работает внутри текущей партиции — TG_TABLE_NAME.
 * RLS-scope воспроизводится так же, как до партиционирования, но область
 * видимости ограничена одной партицией → per-partition per-RLS-scope цепочка.
 *
 * Валидатор воспроизводит это через PARTITION BY RLS-scope ВНУТРИ каждой
 * partition-таблицы (те же partition_clause что раньше).
 *
 * ──────────────────────────────────────────────────────────────────────────────
 *
 * При разрыве: создаёт incidents_log (type='other', severity='high', через
 * pgsql_supplier BYPASSRLS), дедупликация 24ч, email на kdv1@bk.ru.
 * Возвращает self::FAILURE при ЛЮБОМ разрыве — независимо от успеха записи
 * инцидента (инцидент-запись best-effort, не влияет на exit code).
 *
 * Запускается daily 04:00 (routes/console.php).
 *
 * Ref: docs/superpowers/plans/2026-05-23-hole-1-hash-chain-validator.md
 *      docs/superpowers/plans/2026-05-23-hole-2-audit-partitioning-plan.md §A.4
 * Паттерн: IncidentsWatchFailures + SharesSupplierPdo.
 */
class VerifyAuditChains extends Command
{
    private const DB_CONNECTION = 'pgsql_supplier';

    /**
     * Monitoring email для критичных алертов audit-целостности.
     */
    private const MONITORING_EMAIL = 'kdv1@bk.ru';

    /**
     * Дедупликация инцидентов: не создавать повторный инцидент по той же таблице
     * если прошло менее DEDUP_HOURS часов.
     */
    private const DEDUP_HOURS = 24;

    protected $signature = 'audit:verify-chains';

    protected $description = 'Проверяет целостность SHA-256 hash-chain в 6 audit-таблицах (per-partition)';

    public function handle(): int
    {
        $anyBreach = false;
        $now = Carbon::now();

        foreach (AuditChainConfig::TABLES as $table => $config) {
            // Get all partitions for this table via pg_inherits.
            $partitions = $this->listPartitions($table);

            if (empty($partitions)) {
                // Table not yet partitioned or no partitions — check parent directly (fallback).
                $partitions = [$table];
            }

            $tableHadBreach = false;

            foreach ($partitions as $partitionName) {
                $breaches = $this->checkPartition($partitionName, $table, $config['partition']);

                if (empty($breaches)) {
                    $this->line("  ✓ {$partitionName}: chain intact");

                    continue;
                }

                $anyBreach = true;
                $tableHadBreach = true;
                $firstId = $breaches[0]->id;
                $count = count($breaches);

                $this->error("  ✗ {$partitionName}: {$count} mismatch(es), first broken id={$firstId}");

                // Incident write is best-effort: never let it suppress the breach signal.
                try {
                    $this->recordIncident($table, $partitionName, $firstId, $count, $now);
                } catch (\Throwable $e) {
                    $this->warn("    Incident write failed for {$partitionName}: {$e->getMessage()}");
                }

                $this->sendAlert($table, $partitionName, $firstId, $count);
            }

            // Auto-resolve: a table whose chain is intact across ALL partitions
            // closes any stale open chain incident left by a previous transient
            // breach (e.g. acceptance test-tenant rows since removed by teardown).
            // Best-effort: never let cleanup break the command or its exit code.
            if (! $tableHadBreach) {
                try {
                    $this->resolveOpenIncidents($table, $now);
                } catch (\Throwable $e) {
                    $this->warn("    Incident auto-resolve failed for {$table}: {$e->getMessage()}");
                }
            }
        }

        // Exit FAILURE on ANY breach regardless of incident-write success.
        if ($anyBreach) {
            return self::FAILURE;
        }

        $this->info('All audit chains intact.');

        return self::SUCCESS;
    }

    /**
     * Возвращает список дочерних партиций таблицы через pg_inherits.
     * Возвращает пустой массив если таблица не партиционирована или партиций нет.
     *
     * @return list<string>
     */
    private function listPartitions(string $table): array
    {
        $rows = DB::connection(self::DB_CONNECTION)->select(
            'SELECT c.relname
             FROM pg_inherits i
             JOIN pg_class c ON c.oid = i.inhrelid
             JOIN pg_class p ON p.oid = i.inhparent
             WHERE p.relname = ?
             ORDER BY c.relname',
            [$table],
        );

        return array_map(fn ($r) => $r->relname, $rows);
    }

    /**
     * Проверяет hash-chain одной партиции (или таблицы) через SQL на стороне PostgreSQL.
     *
     * Возвращает список строк, у которых stored log_hash ≠ recomputed hash.
     *
     * SQL-логика:
     *   1. Берёт все строки партиции.
     *   2. Через LAG(log_hash) OVER (<partition> ORDER BY id) получает prev_hash
     *      каждой строки в пределах её RLS-scope (partition).
     *   3. Пересчитывает: digest(COALESCE(prev_hash,''::bytea) || ROW(...)::text::bytea, 'sha256')
     *      где ROW(...) имеет NULL::bytea на позиции log_hash.
     *   4. Возвращает строки, где stored IS DISTINCT FROM recomputed.
     *
     * @return list<object>
     */
    private function checkPartition(string $partitionName, string $table, string $partition): array
    {
        $rowExpr = AuditChainConfig::rowExpression($table);

        // Build OVER clause: with or without PARTITION BY depending on table's RLS scope.
        $overClause = $partition !== ''
            ? "({$partition} ORDER BY id)"
            : '(ORDER BY id)';

        $sql = <<<SQL
            WITH ordered AS (
                SELECT
                    id,
                    log_hash AS stored_hash,
                    LAG(log_hash) OVER {$overClause} AS prev_hash
                FROM {$partitionName}
            )
            SELECT
                o.id,
                o.stored_hash,
                digest(
                    COALESCE(o.prev_hash, ''::bytea)
                    || (SELECT {$rowExpr}::text::bytea FROM {$partitionName} t WHERE t.id = o.id),
                    'sha256'
                ) AS recomputed_hash
            FROM ordered o
            WHERE o.stored_hash IS DISTINCT FROM
                digest(
                    COALESCE(o.prev_hash, ''::bytea)
                    || (SELECT {$rowExpr}::text::bytea FROM {$partitionName} t WHERE t.id = o.id),
                    'sha256'
                )
            ORDER BY o.id
        SQL;

        /** @var list<object> $results */
        $results = DB::connection(self::DB_CONNECTION)
            ->select($sql);

        return $results;
    }

    /**
     * Вставляет запись в incidents_log (через pgsql_supplier BYPASSRLS).
     * Дедупликация: не создаёт повторный инцидент для той же таблицы,
     * если за последние DEDUP_HOURS часов уже есть открытый инцидент.
     *
     * Вызывается внутри try/catch в handle() — исключение не подавляет
     * breach-сигнал (handle() всё равно вернёт self::FAILURE).
     *
     * @param  string  $table  Имя родительской таблицы (для дедупликации)
     * @param  string  $partitionName  Имя конкретной партиции где обнаружен разрыв
     */
    private function recordIncident(
        string $table,
        string $partitionName,
        int $firstBrokenId,
        int $count,
        Carbon $now
    ): void {
        $dedupSince = $now->copy()->subHours(self::DEDUP_HOURS);

        $alreadyOpen = DB::connection(self::DB_CONNECTION)
            ->table('incidents_log')
            ->where('type', 'other')
            ->where('severity', 'high')
            ->where('summary', 'like', '%chain%'.addcslashes($table, '%_\\').'%')
            ->whereNull('resolved_at')
            ->where('detected_at', '>=', $dedupSince)
            ->exists();

        if ($alreadyOpen) {
            $this->line("    Skipping incident (dedup): {$partitionName}");

            return;
        }

        // Для NOT NULL FK created_by_admin_id берём первого активного SaaS-admin.
        // Если нет активных admins — пишем предупреждение, но НЕ пропускаем:
        // бросаем исключение, чтобы caller (try/catch в handle()) его поймал
        // и залогировал. Breach-сигнал (FAILURE exit code) уже установлен выше.
        $adminId = DB::connection(self::DB_CONNECTION)
            ->table('saas_admin_users')
            ->where('is_active', true)
            ->whereNull('deleted_at')
            ->value('id');

        if ($adminId === null) {
            $this->warn("    No active saas_admin_users — incident not recorded for {$partitionName}");

            return;
        }

        DB::connection(self::DB_CONNECTION)->table('incidents_log')->insert([
            'type' => 'other',
            'severity' => 'high',
            'summary' => "Автоматически: разрыв hash-chain в партиции {$partitionName} (таблица {$table}). "
                                    ."Первый сломанный id={$firstBrokenId}, всего несовпадений={$count}. "
                                    .'Возможен tampering (UPDATE/DELETE в обход триггеров).',
            'root_cause' => null,
            'started_at' => $now,
            'detected_at' => $now,
            'resolved_at' => null,
            'created_by_admin_id' => $adminId,
            'created_at' => $now,
            'updated_at' => $now,
        ]);

        $this->warn("    Incident recorded for {$partitionName} (first broken id={$firstBrokenId})");
    }

    /**
     * Авто-закрытие устаревших открытых инцидентов разрыва цепочки для таблицы,
     * чья цепочка снова целостна во всех партициях.
     *
     * Закрывает класс «вечно-открытых» high-инцидентов после транзиентного
     * разрыва (строки удалены/исправлены вне прогона — напр. строки тест-тенантов
     * приёмки, убранные teardown): без этого verify-chains накапливал бы открытые
     * инциденты и слал бы по ним алёрты после истечения дедупа.
     *
     * Матчинг summary — тот же per-table шаблон, что в recordIncident()
     * (дедупликация и закрытие симметричны). Вызывается только когда таблица
     * чиста во ВСЕХ партициях (guard $tableHadBreach в handle()).
     */
    private function resolveOpenIncidents(string $table, Carbon $now): void
    {
        $resolved = DB::connection(self::DB_CONNECTION)
            ->table('incidents_log')
            ->where('type', 'other')
            ->where('severity', 'high')
            ->where('summary', 'like', '%chain%'.addcslashes($table, '%_\\').'%')
            ->whereNull('resolved_at')
            ->update([
                'resolved_at' => $now,
                'updated_at' => $now,
                'root_cause' => "Автоматически закрыт: audit:verify-chains подтвердил целостность hash-chain таблицы {$table}.",
            ]);

        if ($resolved > 0) {
            $this->info("  ↻ {$table}: auto-resolved {$resolved} stale chain incident(s).");
        }
    }

    /**
     * Отправляет email-алёрт на monitoring email.
     */
    private function sendAlert(string $table, string $partitionName, int $firstBrokenId, int $count): void
    {
        try {
            Mail::to(self::MONITORING_EMAIL)
                ->send(new AuditChainBreachMail($table, $firstBrokenId, $count, $partitionName));
        } catch (\Throwable $e) {
            // Не ломаем команду если почта недоступна — инцидент уже записан
            $this->warn("    Email failed: {$e->getMessage()}");
        }
    }
}