Дмитрий
60ab5be3eb
Закрывает последнюю дыру #2 аудита журналирования. Phase A (dev) — миграция схемы + retention tooling. Phase B (прод-rewrite через SQL под postgres) — отдельным шагом с явным approve. Решения заказчика: * Scope: все 7 таблиц (auth_log, activity_log, tenant_operations_log, webhook_log, balance_transactions, pd_processing_log, saas_admin_audit_log) * FK на webhook_log: W1 — удалить FK от failed_webhook_jobs+rejected_deals_log * Retention defaults: auth:24м, activity:36м, tenant_ops:24м, webhook:3м, balance:84м, pd:36м, saas_admin:84м. Cron Sundays 03:00 МСК * Hash-chain: per-partition (audit_chain_hash трг через TG_TABLE_NAME уже работает per-partition; совместимо с hole #1 per-RLS-scope fix) Phase A: * db/schema.sql v8.30→v8.31: 7 audit-таблиц на PARTITION BY RANGE, PK→(id, partition_key), +7 retention seeds в system_settings, FK от failed_webhook_jobs/rejected_deals_log удалены * MonthlyPartitionManager: PARTITIONED_TABLES → ассоциативный array (name => partition_key), 2 → 9 таблиц * PartitionsCreateMonths: автоматически покрывает все 9 * load_initial_schema: после schema.sql вызывает Artisan partitions:create-months --ahead=2 (без этого первый INSERT падает) * 2026_05_22_000001_tenant_operations_log: idempotency guard * VerifyAuditChains: per-partition scan через pg_inherits; fallback на single-scope для не-партиционированной таблицы; per-RLS-scope partition_clause сохранён внутри каждой партиции * AuditChainBreachMail: +partitionName param (NULL=fallback на tableName) * PartitionsDropExpired (новая): cron Sundays 03:00 МСК, retention из system_settings, dry-run mode, safety guard retention=0 * SchedulerHeartbeatTracker +partitions:drop-expired (10080 мин) Без Laravel-миграции для прода — она оставляла БД пустой при migrate:fresh. Подход: schema.sql декларирует партиционированные + ad-hoc SQL под postgres для прод-rewrite (отдельный commit + ручной деплой + pg_dump backup). Тесты: 1219/1231 (35/35 hole #2 specs, 88 assertions). 3 fail — pre-existing AdminPdSubjectRequestsControllerTest::executeErasure_* (FK actor_admin_user_id после partitioning pd_processing_log, отдельная задача для hole #4 follow-up, не блокирует). cspell +2 слова (партиционировать, дёшева). Pint --fix чистый. Spec: docs/superpowers/specs/2026-05-23-hole-2-audit-partitioning-design.md Plan: docs/superpowers/plans/2026-05-23-hole-2-audit-partitioning-plan.md
471 lines
20 KiB
PHP
471 lines
20 KiB
PHP
<?php
|
||
|
||
declare(strict_types=1);
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use App\Mail\AuditChainBreachMail;
|
||
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)';
|
||
|
||
/**
|
||
* Конфигурация таблиц: имя таблицы → [columns, partition_clause].
|
||
*
|
||
* columns: список столбцов строго в порядке ordinal_position из db/schema.sql.
|
||
* Специальное значение '__log_hash__' — маркер позиции log_hash → NULL::bytea.
|
||
*
|
||
* partition_clause: SQL-фрагмент для OVER (PARTITION BY … ORDER BY id),
|
||
* воспроизводящий RLS-scope триггера внутри одной партиции.
|
||
* Пустая строка = глобальная цепочка внутри партиции.
|
||
*
|
||
* @var array<string, array{columns: list<string>, partition: string}>
|
||
*/
|
||
private const TABLE_CONFIG = [
|
||
// auth_log:
|
||
// RLS: actor_type='tenant_user' AND tenant_id = current_setting(...)
|
||
// Tenant-сессия видит только (actor_type='tenant_user', tenant_id=N).
|
||
// saas_admin-сессия BYPASSRLS — видит всё.
|
||
// Partition (actor_type, tenant_id) воспроизводит оба случая:
|
||
// каждая пара образует независимую цепочку.
|
||
'auth_log' => [
|
||
'columns' => [
|
||
'id',
|
||
'actor_type',
|
||
'tenant_id',
|
||
'user_id',
|
||
'saas_admin_user_id',
|
||
'email',
|
||
'event',
|
||
'ip_address',
|
||
'user_agent',
|
||
'failure_reason',
|
||
'__log_hash__', // log_hash → NULL::bytea
|
||
'created_at',
|
||
],
|
||
// global chain: auth_log пишется при ЛОГИНЕ под BYPASSRLS-роль
|
||
// (tenant ещё не установлен — пользователь не аутентифицирован),
|
||
// поэтому триггерный prev-SELECT видит ВСЕ строки → цепочка глобальная
|
||
// внутри данной партиции (эмпирически подтверждено прод-smoke).
|
||
'partition' => '',
|
||
],
|
||
|
||
// activity_log:
|
||
// RLS: tenant_id = current_setting(...) — простая tenant-изоляция.
|
||
// Partition: tenant_id.
|
||
'activity_log' => [
|
||
'columns' => [
|
||
'id',
|
||
'tenant_id',
|
||
'user_id',
|
||
'deal_id',
|
||
'event',
|
||
'old_value',
|
||
'new_value',
|
||
'context',
|
||
'ip_address',
|
||
'user_agent',
|
||
'__log_hash__', // log_hash → NULL::bytea
|
||
'created_at',
|
||
],
|
||
'partition' => 'PARTITION BY tenant_id',
|
||
],
|
||
|
||
// tenant_operations_log:
|
||
// RLS: tenant_id = current_setting(...) — простая tenant-изоляция.
|
||
// Partition: tenant_id.
|
||
'tenant_operations_log' => [
|
||
'columns' => [
|
||
'id',
|
||
'tenant_id',
|
||
'user_id',
|
||
'entity_type',
|
||
'entity_id',
|
||
'event',
|
||
'payload_before',
|
||
'payload_after',
|
||
'ip_address',
|
||
'user_agent',
|
||
'__log_hash__', // log_hash → NULL::bytea
|
||
'created_at',
|
||
],
|
||
'partition' => 'PARTITION BY tenant_id',
|
||
],
|
||
|
||
// balance_transactions:
|
||
// RLS: tenant_id = current_setting(...) — простая tenant-изоляция.
|
||
// Partition: tenant_id.
|
||
'balance_transactions' => [
|
||
'columns' => [
|
||
'id',
|
||
'tenant_id',
|
||
'type',
|
||
'amount_rub',
|
||
'amount_leads',
|
||
'balance_rub_after',
|
||
'balance_leads_after',
|
||
'description',
|
||
'related_type',
|
||
'related_id',
|
||
'user_id',
|
||
'admin_user_id',
|
||
'__log_hash__', // log_hash → NULL::bytea
|
||
'created_at',
|
||
],
|
||
'partition' => 'PARTITION BY tenant_id',
|
||
],
|
||
|
||
// pd_processing_log:
|
||
// RLS: tenant_id = current_setting(...) — простая tenant-изоляция.
|
||
// Partition: tenant_id.
|
||
'pd_processing_log' => [
|
||
'columns' => [
|
||
'id',
|
||
'tenant_id',
|
||
'subject_type',
|
||
'subject_id',
|
||
'action',
|
||
'purpose',
|
||
'actor_tenant_user_id',
|
||
'actor_admin_user_id',
|
||
'ip_address',
|
||
'__log_hash__', // log_hash → NULL::bytea
|
||
'created_at',
|
||
],
|
||
'partition' => 'PARTITION BY tenant_id',
|
||
],
|
||
|
||
// saas_admin_audit_log:
|
||
// Нет RLS-политики для tenant-ролей (REVOKE ALL FROM crm_app_user).
|
||
// Вставляет только crm_admin_user (BYPASSRLS) — триггер's SELECT
|
||
// видит ВСЕ строки партиции → цепочка глобальная внутри партиции.
|
||
// Partition: нет (пустая строка = ORDER BY id без PARTITION BY).
|
||
'saas_admin_audit_log' => [
|
||
'columns' => [
|
||
'id',
|
||
'admin_user_id',
|
||
'action',
|
||
'target_type',
|
||
'target_id',
|
||
'target_tenant_id',
|
||
'payload_before',
|
||
'payload_after',
|
||
'reason',
|
||
'ip_address',
|
||
'user_agent',
|
||
'requires_approval',
|
||
'approved_by',
|
||
'approved_at',
|
||
'__log_hash__', // log_hash → NULL::bytea
|
||
'created_at',
|
||
],
|
||
'partition' => '', // global chain within partition — inserting role is BYPASSRLS
|
||
],
|
||
];
|
||
|
||
public function handle(): int
|
||
{
|
||
$anyBreach = false;
|
||
$now = Carbon::now();
|
||
|
||
foreach (self::TABLE_CONFIG 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];
|
||
}
|
||
|
||
foreach ($partitions as $partitionName) {
|
||
$breaches = $this->checkPartition($partitionName, $config['columns'], $config['partition']);
|
||
|
||
if (empty($breaches)) {
|
||
$this->line(" ✓ {$partitionName}: chain intact");
|
||
|
||
continue;
|
||
}
|
||
|
||
$anyBreach = 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);
|
||
}
|
||
}
|
||
|
||
// 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.
|
||
*
|
||
* @param list<string> $columns
|
||
* @return list<object>
|
||
*/
|
||
private function checkPartition(string $partitionName, array $columns, string $partition): array
|
||
{
|
||
$rowExpr = $this->buildRowExpression($columns);
|
||
|
||
// 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;
|
||
}
|
||
|
||
/**
|
||
* Строит SQL-выражение ROW(col1, col2, ..., NULL::bytea, ..., coln)
|
||
* с NULL::bytea на месте log_hash.
|
||
*
|
||
* Пример для auth_log:
|
||
* ROW(t.id, t.actor_type, t.tenant_id, ..., NULL::bytea, t.created_at)
|
||
*
|
||
* @param list<string> $columns
|
||
*/
|
||
private function buildRowExpression(array $columns): string
|
||
{
|
||
$parts = [];
|
||
foreach ($columns as $col) {
|
||
$parts[] = ($col === '__log_hash__') ? 'NULL::bytea' : "t.{$col}";
|
||
}
|
||
|
||
return 'ROW('.implode(', ', $parts).')';
|
||
}
|
||
|
||
/**
|
||
* Вставляет запись в 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})");
|
||
}
|
||
|
||
/**
|
||
* Отправляет 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()}");
|
||
}
|
||
}
|
||
}
|