91 lines
2.3 KiB
Python
91 lines
2.3 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Append-only hash-chain records for local forensic artifacts."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from . import schemas
|
|
|
|
CHAIN_NAME = "hash-chain.jsonl"
|
|
|
|
|
|
def chain_path(cfg) -> Path:
|
|
return cfg.log_dir / CHAIN_NAME
|
|
|
|
|
|
def _sha256_bytes(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def file_digest(path: Path) -> str:
|
|
return _sha256_bytes(path.read_bytes())
|
|
|
|
|
|
def last_chain_hash(cfg) -> str:
|
|
try:
|
|
lines = chain_path(cfg).read_text().splitlines()
|
|
except OSError:
|
|
return ""
|
|
for line in reversed(lines):
|
|
try:
|
|
record = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
value = record.get("chain_hash")
|
|
if isinstance(value, str):
|
|
return value
|
|
return ""
|
|
|
|
|
|
def append_artifact(cfg, kind: str, path: Path) -> dict:
|
|
"""Append a tamper-evident hash-chain record for ``path``."""
|
|
cfg.log_dir.mkdir(parents=True, exist_ok=True)
|
|
prev = last_chain_hash(cfg)
|
|
try:
|
|
digest = file_digest(path)
|
|
size = path.stat().st_size
|
|
except OSError:
|
|
digest = ""
|
|
size = 0
|
|
record = {
|
|
"schema": schemas.HASH_CHAIN_V1,
|
|
"time": time.time(),
|
|
"kind": kind,
|
|
"path": str(path),
|
|
"size": size,
|
|
"sha256": digest,
|
|
"prev_hash": prev,
|
|
}
|
|
payload = json.dumps(record, sort_keys=True, separators=(",", ":")).encode()
|
|
record["chain_hash"] = _sha256_bytes(payload + prev.encode())
|
|
with chain_path(cfg).open("a") as fh:
|
|
fh.write(json.dumps(record, sort_keys=True) + "\n")
|
|
try:
|
|
chain_path(cfg).chmod(0o640)
|
|
except OSError:
|
|
pass
|
|
return record
|
|
|
|
|
|
def summary(cfg) -> dict:
|
|
path = chain_path(cfg)
|
|
try:
|
|
lines = path.read_text().splitlines()
|
|
except OSError:
|
|
return {"path": str(path), "exists": False, "count": 0, "last_hash": ""}
|
|
last = ""
|
|
for line in reversed(lines):
|
|
try:
|
|
last = json.loads(line).get("chain_hash", "")
|
|
break
|
|
except ValueError:
|
|
continue
|
|
return {
|
|
"path": str(path),
|
|
"exists": True,
|
|
"count": len(lines),
|
|
"last_hash": last,
|
|
}
|