tobs-code/ghostAudit
GitHub: tobs-code/ghostAudit
GhostAudit是一款基于SQLite的隐蔽审计日志工具。
Stars: 1 | Forks: 0
# GhostAudit
```
from core.carrier_config import CarrierConfig
from core.ghost_audit_v9 import GhostAuditInterceptor
import secrets
carrier = CarrierConfig(
table="users",
id_field="id",
semantic_field="bio",
float_a_field="trust_score",
float_b_field="profile_score",
tilde_field="avatar_url",
)
ga = GhostAuditInterceptor(
db_path="app.db",
carrier_config=carrier,
secret_key=secrets.token_hex(32),
)
ga.log_event("user=alice action=login ip=10.0.0.1")
# App write hook: one UPDATE, with stego overlay folded into the app fields.
fields = {"bio": bio, "trust_score": trust, "profile_score": profile, "avatar_url": url}
fields = ga.intercept(row_id=user_id, fields=fields)
# UPDATE users SET bio=?, trust_score=?, profile_score=?, avatar_url=? WHERE id=?
# attacker deletes audit_log — recovery still reads the carrier table
events = ga.recover_events() # → [( 1, "user=alice action=login ip=10.0.0.1")]
```
**工作原理：** 事件被编码到生物文本的同义词选项、浮点分数列的LSB、机会性文本形状变体和URL波浪线后缀中，这些被分散在5个逻辑通道中，使用RAID-6擦除编码和Reed-Solomon ECC。不知道主密钥的攻击者无法找到、读取或不可检测地修改日志。
```
pip install reedsolo cryptography numpy
python quickstart.py # see it in action
```
请参阅[quickstart.py](../quickstart.py)以获取包括攻击模拟和恢复的遗留独立演示。
## 配置 & 环境变量
所有参数都有合理的默认值，并且可以通过环境变量覆盖：
| 环境变量 | 默认值 | 描述 |
|---------|---------|--------------|
| `GHOST_AUDIT_KEY` | — | 主密钥（相对于`secret_key`参数更受青睐） |
| `GHOST_AUDIT_SLOT_SIZE` | `1600` | 每个槽的载体行数 |
| `GHOST_AUDIT_PER_CHANNEL_MIN_REPS` | `5` | 最大隐写复制品 |
| `GHOST_AUDIT_ECC_SYMBOLS` | `36` | RS奇偶校验符号（基线） |
| `GHOST_AUDIT_REPLICA_COUNT` | `3` | 每个事件每个槽的副本数 |
| `GHOST_AUDIT_EXTERNAL_STATE` | `.evolve` | 回滚识别文件的路径 |
| `GHOST_AUDIT_REBUILD_ECC_SYMBOLS` | `52` | 重建nsym（下限，V8.4） |
| `GHOST_AUDIT_REBUILD_MIN_REPS` | `4` | 重建复制品（下限，V8.4） |
| `GHOST_AUDIT_REBUILD_THRESHOLD` | `0.35` | 重建触发器退化阈值（V8.4） |
| `GHOST_AUDIT_REBUILD_INTERVAL` | `50` | 两次空闲检查之间的事件数（V8.4） |
**示例（PowerShell）：**
```
# Höhere ECC-Stärke für sensible Umgebungen
$env:GHOST_AUDIT_ECC_SYMBOLS="48"
# Aggressiveres Self-Healing
$env:GHOST_AUDIT_REBUILD_THRESHOLD="0.20"
$env:GHOST_AUDIT_REBUILD_INTERVAL="25"
# Rollback-Schutz auf separatem Datenträger
$env:GHOST_AUDIT_EXTERNAL_STATE="E:\secure_mount\audit.evolve"
```
**检查点工作流程（Git作为见证人）：**
从V8.6开始，Git-Witness被作为标准实现。如果`*.evolve`文件（回滚保护）位于Git存储库中，GhostAudit将在每次内部状态更新后自动执行`git add`和`git commit`。
```
# Git-Witness funktioniert jetzt automatisch, solange
# sich die *.evolve-Datei in einem Git-Repo befindet.
```
## API 参考
### V9 构造函数
```
from core.carrier_config import CarrierConfig
from core.ghost_audit_v9 import GhostAuditInterceptor
carrier = CarrierConfig(
table="users",
id_field="id",
semantic_field="bio",
float_a_field="trust_score",
float_b_field="profile_score",
tilde_field="avatar_url",
slot_size=1600,
slot_count=5,
)
ga = GhostAuditInterceptor(
db_path="app.db",
carrier_config=carrier, # None nutzt Legacy-sys_cache-Layout
secret_key=None, # oder GHOST_AUDIT_KEY env-var
key_provider=None, # DPAPI / EnvKeyProvider
ecc_symbols=36,
verbose=True,
siem_export_path=None, # Auto-Export bei jedem log_event()
siem_export_format="jsonl",# "jsonl" oder "cef"
metronome_interval=0, # Heartbeat-Intervall in Sekunden (0=aus)
external_state_path=None, # Rollback-Erkennungsdatei
force_reinit=False, # Admin-Override: überspringt alle Rollback-Checks (für Tests / DB-Neuanlage)
)
```
### V9 应用程序写入钩子
```
ga.log_event("user=alice action=login")
# fields sind die Werte, die du in die DB schreiben willst.
# Der Caller muss sicherstellen, dass alle relevanten Felder vollständig
# übergeben werden — auch wenn die App nur ein Subset updatet.
fields = {
"bio": new_bio, # der Wert, den das UPDATE gleich setzen wird
"trust_score": new_score,
"profile_score": new_profile,
"avatar_url": new_avatar,
}
fields = ga.intercept(row_id=user_id, fields=fields)
# ↑ intercept() darf die Werte minimal transformieren (TextShape, Synonym,
# Float-LSB). Der Rückgabewert ist das, was tatsächlich geschrieben wird.
# Die App schreibt genau diesen Rückgabewert mit ihrem normalen UPDATE.
# Nach dem Commit kann optional explizit geflusht werden; recover_events()
# ruft flush_headers() ebenfalls automatisch auf.
ga.flush_headers()
```
**`intercept()` — 调用者合同（一句话）：**
**由此得出：**
- 调用者从*自己的*新值构建`fields`（或者如果更新只涉及子集，则首先读取数据库——这是应用程序的责任，而不是GhostAudit）。
- `intercept()`不知道任何先前的值。它仅基于传递的字典工作。如果`bio`没有携带文本形状载体形式，则跳过该行，并将挂起的位分配给另一个合适的行。
- 返回值始终是一个完整的字典，具有与`fields`相同的键。即使没有嵌入任何位，`returned == input`也是相同的值——应用程序可以无条件地写入。
- 没有提供`current_fields`参数。如果需要从先前的状态获取载体上下文，必须在`intercept()`之前进行数据库读取。
### 记录
```
ga.log_event("event message") # gibt Sequenznummer zurück
ga.log_event("event", immediate_commit=False) # deferred commit
ga.log_events(["msg1", "msg2"]) # Batch — gibt Liste von Sequenznummern zurück
```
### 恢复
```
events = ga.recover_events() # → [(seq, msg), ...]
gaps = ga.detect_truncation(events) # Heartbeat-Lücken
```
### 验证
```
ga.get_verification_digest() # Merkle-Root als hex-String
ga.verify_event_mac(seq) # einzelnes Event
ga.verify_all_event_macs() # alle Events
ga.verify_merkle_root() # aktueller Anchor
ga.list_merkle_anchors(limit=10)
```
### 检查点（外部见证人）
```
# Checkpoint exportieren — in externe, read-only Location speichern
cp = ga.export_checkpoint(path="checkpoint.json")
# cp = {"seq": 42, "root": "a3f9...", "entry_chain": "...",
# "anchor_chain": "...", "timestamp": "...", "mac": "..."}
# Später verifizieren (z.B. nach einem Incident)
result = ga.verify_checkpoint(cp) # aus dict
result = ga.verify_checkpoint(None, path="checkpoint.json") # aus Datei
# result["valid"] → True/False
# result["root_match"] → Carrier-Layer unverändert?
# result["entry_chain_match"] → Event-Kette unverändert?
# result["anchor_chain_match"] → Anchor-Kette unverändert?
# result["details"] → "OK" oder Fehlerbeschreibung
```
### 导出
```
ga.export_recovered_logs("out.jsonl", format="jsonl")
ga.export_recovered_logs("out.cef", format="cef")
```
## 执行测试
```
# Interaktives Menü (Einstiegspunkt)
python tests/quickstart_tests.py
# Resilienz-Benchmark (5 quantitative Tests)
python tests/resilience_benchmark_v7.py
# Angriffssimulation V8 (5 Vektoren)
python tests/attack_simulator_v8.py
# Gradual Decay Ramp (50 BER-Stufen 0–49%)
python tests/gradual_decay_ramp_v82.py
# MUX Row-Wipe Sweep (5/10/15/20%)
python tests/sweep_wipe_v8.py
# Rollback-Erkennung
python tests/test_rollback_v82.py
# V9 Interceptor / echter Carrier
python -m pytest tests/test_v9_interceptor.py -q
# Härtungs-Tests (LSB, Forward Security, Merkle, Export)
python tests/test_hardenings_v7.py
# Multi-Kanal-Degradation (7 Szenarien)
python tests/test_multichannel_degradation.py
# Master-Testsuite (alle Läufe, JSON-Report)
python tests/master_test_suite_v7.py
# Durchsatz-Benchmark
python tests/benchmark_throughput_v8.py
```
## 测试结果
### V8 攻击模拟
| 攻击 | 向量 | 结果 |
|------|--------|----------|
| MAC-Strip | `row_mac`从Manifest中删除 | ✅ 3/3 恢复 |
| MUX Row-Wipe 15% | 删除15%的有效载荷行 | ✅ 3/3 恢复 |
| Bio-Normalize | `bio`小写+trim+同义词（2/5载体） | ✅ 3/3 恢复 |
| Both-Floats Round | `trust_score`+`profile_score`四舍五入（2/5载体） | ✅ 3/3 恢复 |
| Gaussian Seed | 尝试种子恢复（阳性控制） | ✅ 3/3 恢复 |
### V7 弹性基准
| 测试 | 阈值 | 结果 |
|------|----------|----------|
| 擦除容错 | ≥90%在8%行擦除时 | ✅ 100% |
| 比特翻转抵抗 | ≥80%在10%BER时 | ✅ 100% |
| 通道隔离 | ≥90%在部分Ch0损坏时 | ✅ 100% |
| 多列擦除 | ≥70%在50%尾随空间擦除时 | ✅ 100% |
| 高BER容错 | ≥70%在10%浮点BER时 | ✅ 100% |
**5/5 — 100% 通过**
### 渐进衰减斜率（V8.2）
50级BER扫描（0-49%）：
| 指标 | 结果 |
|--------|----------|
| CORRUPTED_LEAK | **0%** |
| 第一个不可纠正的 | 步骤7（BER=7%）——RS数学界限 |
| nsym范围 | 36-56（自适应） |
| min_reps | 2-4（稳定，无振荡） |
## 已知限制
- **研究原型** — 没有产品化的审计系统
- V9需要足够的真实载体行和应用程序写入，以完全嵌入挂起的有效载荷位。
- 标题行当前以槽为单位直接通过GhostAudit引擎写入；有效载荷写入通过应用程序钩子运行。
- ORM文本规范化（TRIM、小写、同义词平坦化）破坏基于bio的载体。Bio-SPOF修复将载体杀伤从3/5减少到2/5。
- RAID-6容忍**任何**2/5载体杀伤。在3/5时无法恢复。
- 前向安全性保护槽之间，而不是防止主密钥丢失。
- 回滚保护有一个记录的限制：如果从同一个快照中同时恢复数据库和`.evolve`文件，则不会检测到回滚。
- `audit_archive`是有意为之的诱饵——真正的恢复来自载体表（V9中的`users`，Legacy模式中的`sys_cache`）。
## 项目结构
```
core/
├── ghost_audit_v9.py Interceptor-Architektur mit echtem App-Carrier
├── carrier_config.py Konfiguration für Carrier-Tabelle und Felder
├── ghost_audit_v7.py Engine und Legacy-sys_cache-Modus (V7–V8.x)
├── ecc_layer.py Reed-Solomon Utilities
├── key_provider.py DPAPI / EnvKeyProvider
├── worker_erasure.py Erasure-Recovery
└── security_suite_support.py Factory, CLI-Flags, Test-Gate-Bypass
tests/
├── test_v9_interceptor.py V9 Hook, echter Carrier, External-Carrier-Recovery
├── quickstart_tests.py Interaktives Testmenü
├── master_test_suite_v7.py Orchestrator, erzeugt JSON-Report
├── attack_simulator_v8.py 5 Angriffsvektoren (MITRE ATT&CK)
├── resilience_benchmark_v7.py 5 quantitative Robustheitstests
├── gradual_decay_ramp_v82.py 50-Stufen BER-Sweep
├── sweep_wipe_v8.py MUX Row-Wipe Sweep
├── test_rollback_v82.py Rollback-Erkennung
├── test_hardenings_v7.py LSB, Forward Security, Merkle, Export
├── test_multichannel_degradation.py 7 ORM-Szenarien
├── benchmark_throughput_v8.py Durchsatz-Benchmark
└── hardware_resilience_test.py FileCarrier (binäre Datei statt SQLite)
docs/
├── README.md Diese Datei
├── README_GHOST_AUDIT.md Ausführliche Vorgänger-Dokumentation
├── QUICK_REFERENCE.md Kurzbefehle auf einen Blick
├── TEST_SUITE_OVERVIEW.md Testfluss und Interpretation
└── HARDWARE_CARRIER_TEST.md FileCarrier-Architektur
analysis/ Sweep-Runner, Kapazitätsanalyse, Steganalyse
tools/ Slot-Größen-Vergleich, Sweep-Aggregation, Key-Utilities
```
标签：ECC, Python, RAID-6, Reed-Solomon, SQLite, URL发现, 安全检测, 审计日志, 攻击模拟, 数据加密, 数据恢复, 数据隐藏, 无后门, 环境变量配置, 用户行为分析, 纠错编码, 逆向工具, 隐写术, 驱动签名利用