OpenNIT
Quelltext des aktuellen Standes
Fassung 1.1.1
AuditStorageFormatTest.php
13 KB<?php
declare(strict_types=1);
use OpenNIT\Audit\HashChain;
use OpenNIT\Audit\JsonCanonicalizer;
use OpenNIT\Core\Database;
/**
* Deckt die Einordnung von Einträgen ab, deren JSON-Felder die Datenbank
* umformatiert zurückgibt.
*
* Solche Einträge wurden über einen Text signiert, den es so nicht mehr gibt.
* Sie dürfen nicht als Manipulation gelten – aber die Nachsicht darf weder für
* neue Einträge greifen noch echte Manipulationen verdecken.
*/
return static function (): void {
$_ENV['APP_KEY'] = 'test-key-für-speicherformat';
/**
* Baut eine Eintragszeile.
*
* @param string|null $storedJson Text, den die Datenbank zurückliefert
* @param string|null $signedJson Text, über den signiert wurde (Standard: derselbe)
* @param bool $canonical Signatur nach dem kanonischen Verfahren bilden
*/
$makeEntry = function (
int $id,
string $prevChecksum,
?string $storedJson = null,
?string $signedJson = null,
bool $canonical = false
): array {
$entry = [
'id' => $id,
'event_id' => sprintf('%08d-0000-4000-8000-000000000000', $id),
'event_timestamp' => sprintf('2026-08-21 10:%02d:00.000000', $id),
'user_id' => 1,
'action_category' => 'admin',
'action_type' => 'assign_role',
'object_type' => 'user',
'object_id' => 1,
'object_name' => 'Administrator',
'details_json' => $storedJson,
'old_value_json' => null,
'new_value_json' => null,
'result' => 'success',
'risk_level' => 'low',
];
$json = $signedJson ?? $storedJson;
if ($canonical) {
$json = JsonCanonicalizer::canonicalize($json);
}
$data = implode('|', [
$entry['event_id'], $entry['event_timestamp'], $entry['user_id'],
$entry['action_category'], $entry['action_type'], $entry['object_type'],
$entry['object_id'], $json ?? '', $entry['result'],
$entry['object_name'], '', '', $entry['risk_level'], $prevChecksum,
]);
$entry['checksum'] = hash_hmac('sha256', $data, $_ENV['APP_KEY']);
$entry['prev_checksum'] = $prevChecksum;
return $entry;
};
/** Datenbank-Attrappe mit optionalem, signiertem Grenz-Anker. */
$db = function (array $rows, ?int $boundary = null): Database {
return new class ($rows, $boundary) extends Database {
public function __construct(private array $rows, private ?int $boundary)
{
parent::__construct([]);
}
public function fetchAll(string $sql, array $params = []): array
{
if (str_contains($sql, "action_type = 'setup_encryption'")) {
return [];
}
if (str_contains($sql, 'FROM audit_log') && str_contains($sql, 'ORDER BY id ASC')) {
return $this->rows;
}
if (str_contains($sql, 'FROM audit_log') && str_contains($sql, 'ORDER BY id DESC')) {
$before = array_filter($this->rows, fn($r) => $r['id'] < ($params[0] ?? PHP_INT_MAX));
return array_map(fn($r) => ['checksum' => $r['checksum']], array_reverse($before));
}
return [];
}
public function fetchOne(string $sql, array $params = []): ?array
{
if (str_contains($sql, 'audit_canonical_anchor')) {
if ($this->boundary === null) {
return null;
}
return [
'boundary_id' => $this->boundary,
'signature' => hash_hmac(
'sha256',
'audit_canonical_anchor|' . $this->boundary,
$_ENV['APP_KEY'] ?? ''
),
];
}
if (str_contains($sql, 'audit_chain_anchor') || str_contains($sql, 'audit_hmac_anchor')) {
return null;
}
foreach ($this->rows as $r) {
if (str_contains($sql, 'WHERE id = ?') && $r['id'] === ($params[0] ?? null)) {
return $r;
}
}
if (str_contains($sql, 'id < ?')) {
$before = array_filter($this->rows, fn($r) => $r['id'] < ($params[0] ?? 0));
return $before ? end($before) : null;
}
return null;
}
public function fetchColumn(string $sql, array $params = [], int $column = 0): mixed
{
if (str_contains($sql, 'COUNT(*)')) {
return count($this->rows);
}
if (str_contains($sql, 'WHERE checksum = ?')) {
foreach ($this->rows as $r) {
if ($r['checksum'] === ($params[0] ?? null)) {
return 1;
}
}
return false;
}
return false;
}
};
};
/** Baut einen Eintrag, dessen Signatur über die angegebene Kategorie gebildet wird. */
$makeEntryMitKategorie = function (string $kategorie, string $json): array {
$entry = [
'id' => 1,
'event_id' => '00000001-0000-4000-8000-000000000000',
'event_timestamp' => '2026-02-20 09:32:04.000000',
'user_id' => 1,
'action_category' => $kategorie,
'action_type' => 'create_service',
'object_type' => 'service',
'object_id' => 1,
'object_name' => 'WLAN',
'details_json' => $json,
'old_value_json' => null,
'new_value_json' => null,
'result' => 'success',
'risk_level' => 'low',
];
$data = implode('|', [
$entry['event_id'], $entry['event_timestamp'], $entry['user_id'],
$kategorie, $entry['action_type'], $entry['object_type'],
$entry['object_id'], JsonCanonicalizer::canonicalize($json), $entry['result'],
$entry['object_name'], '', '', $entry['risk_level'], 'GENESIS',
]);
$entry['checksum'] = hash_hmac('sha256', $data, $_ENV['APP_KEY']);
$entry['prev_checksum'] = 'GENESIS';
return $entry;
};
// So schreibt PHP (alter Wert zuerst), so liefert eine kanonisierende
// Datenbank denselben Inhalt zurück (Schlüssel umsortiert, mit Leerzeichen).
$vonPhp = '{"old_is_admin":0,"new_is_admin":true}';
$vonDb = '{"new_is_admin": true, "old_is_admin": 0}';
test('Nach dem kanonischen Verfahren signierte Einträge sind gültig', function () use ($makeEntry, $db, $vonPhp, $vonDb) {
// Signiert über die kanonische Form, gespeichert in DB-Schreibweise
$a = $makeEntry(1, 'GENESIS', $vonDb, $vonPhp, true);
$result = (new HashChain($db([$a], 0)))->verify();
assertTrue($result['valid']);
assertSame(0, count($result['storage_format']));
});
test('Vor der Grenze bleibt die frühere Rohtext-Signatur gültig', function () use ($makeEntry, $db, $vonPhp) {
// Datenbank ohne Umformatierung: gespeicherter Text = signierter Text
$a = $makeEntry(1, 'GENESIS', $vonPhp, $vonPhp, false);
$result = (new HashChain($db([$a], 5)))->verify();
assertTrue($result['valid']);
assertSame(0, count($result['storage_format']));
});
test('Umformatierte Alt-Einträge gelten nicht als Manipulation', function () use ($makeEntry, $db, $vonPhp, $vonDb) {
// Signiert über den PHP-Text, zurückgeliefert in DB-Schreibweise
$a = $makeEntry(1, 'GENESIS', $vonDb, $vonPhp, false);
$result = (new HashChain($db([$a], 5)))->verify();
assertTrue($result['valid']);
assertSame(1, count($result['storage_format']));
assertSame(0, count($result['invalid']));
});
test('Oberhalb der Grenze gibt es keine Nachsicht', function () use ($makeEntry, $db, $vonPhp, $vonDb) {
$a = $makeEntry(9, 'GENESIS', $vonDb, $vonPhp, false);
$result = (new HashChain($db([$a], 5)))->verify();
assertFalse($result['valid']);
assertSame(0, count($result['storage_format']));
assertSame(9, (int)$result['first_invalid_id']);
});
test('Ohne JSON-Inhalt bleibt der Nachweis auch vor der Grenze bestehen', function () use ($makeEntry, $db) {
$a = $makeEntry(1, 'GENESIS', null, null, false);
$a['object_name'] = 'Manipuliert';
$result = (new HashChain($db([$a], 5)))->verify();
assertFalse($result['valid']);
assertSame(0, count($result['storage_format']));
});
test('Ein manipulierter Eintrag mit kanonischem JSON wird weiterhin erkannt', function () use ($makeEntry, $db, $vonPhp) {
// Der gespeicherte Text ist bereits kanonisch – Umformatierung scheidet aus
$kanonisch = JsonCanonicalizer::canonicalize($vonPhp);
$a = $makeEntry(1, 'GENESIS', $kanonisch, $kanonisch, true);
$a['object_name'] = 'Manipuliert';
$result = (new HashChain($db([$a], 5)))->verify();
assertFalse($result['valid']);
assertSame(0, count($result['storage_format']));
});
test('Die Einzelprüfung ordnet gleich ein wie die Bereichsprüfung', function () use ($makeEntry, $db, $vonPhp, $vonDb) {
$a = $makeEntry(1, 'GENESIS', $vonDb, $vonPhp, false);
$einzeln = (new HashChain($db([$a], 5)))->verifyEntry(1);
assertTrue($einzeln['valid']);
assertTrue($einzeln['storage_format']);
$neu = $makeEntry(9, 'GENESIS', $vonDb, $vonPhp, false);
$einzelnNeu = (new HashChain($db([$neu], 5)))->verifyEntry(9);
assertFalse($einzelnNeu['valid']);
assertFalse($einzelnNeu['storage_format']);
});
test('Eine von der Datenbank verworfene Kategorie wird zurückgewonnen', function () use ($makeEntryMitKategorie, $db) {
// Signiert wurde über die Kategorie 'service'; die Spalte kannte den Wert
// damals nicht und hat ihn verworfen.
// Bereits kanonisches JSON: die Speicherformat-Nachsicht scheidet aus,
// geprüft wird allein die Rückgewinnung der Kategorie.
$a = $makeEntryMitKategorie('service', '{"aktiv":true}');
$a['action_category'] = '';
$chain = new HashChain($db([$a], 5));
$result = $chain->verify();
assertTrue($result['valid']);
assertSame(0, count($result['invalid']));
assertSame('service', $chain->recoveredCategory(1));
});
test('Eine gefüllte Kategorie wird nicht durchprobiert', function () use ($makeEntryMitKategorie, $db) {
// Inhaltlich verändert: die Rückgewinnung darf hier nicht greifen
// Bereits kanonisches JSON: die Speicherformat-Nachsicht scheidet aus,
// geprüft wird allein die Rückgewinnung der Kategorie.
$a = $makeEntryMitKategorie('service', '{"aktiv":true}');
$a['action_category'] = 'admin';
$result = (new HashChain($db([$a], 5)))->verify();
assertFalse($result['valid']);
});
test('Eine leere Kategorie rettet einen manipulierten Eintrag nicht', function () use ($makeEntryMitKategorie, $db) {
// Bereits kanonisches JSON: die Speicherformat-Nachsicht scheidet aus,
// geprüft wird allein die Rückgewinnung der Kategorie.
$a = $makeEntryMitKategorie('service', '{"aktiv":true}');
$a['action_category'] = '';
$a['object_name'] = 'Manipuliert';
$result = (new HashChain($db([$a], 5)))->verify();
assertFalse($result['valid']);
});
test('Ein gefälschter Grenz-Anker wirkt nicht', function () use ($makeEntry, $vonPhp, $vonDb) {
$a = $makeEntry(9, 'GENESIS', $vonDb, $vonPhp, false);
// Anker mit unpassender Signatur: die Grenze darf nicht gelten
$db = new class ([$a]) extends Database {
public function __construct(private array $rows) { parent::__construct([]); }
public function fetchAll(string $sql, array $params = []): array
{
if (str_contains($sql, 'ORDER BY id ASC')) { return $this->rows; }
return [];
}
public function fetchOne(string $sql, array $params = []): ?array
{
if (str_contains($sql, 'audit_canonical_anchor')) {
return ['boundary_id' => 0, 'signature' => str_repeat('0', 64)];
}
return null;
}
public function fetchColumn(string $sql, array $params = [], int $column = 0): mixed
{
return str_contains($sql, 'COUNT(*)') ? count($this->rows) : false;
}
};
// Ungültige Signatur => gesamte Kette zählt als "vor der Grenze"
$result = (new HashChain($db))->verify();
assertTrue($result['valid']);
assertSame(1, count($result['storage_format']));
});
};