fix(security): strengthen sensitive hash derivation

This commit is contained in:
rookiestar28
2026-04-08 01:06:41 +08:00
parent d429c11018
commit 01f9828f99
3 changed files with 83 additions and 4 deletions
+20 -1
View File
@@ -4,9 +4,11 @@ Standardized, append-only audit events for sensitive operations.
"""
import hashlib
import hmac
import json
import logging
import os
import secrets
import threading
import time
import uuid
@@ -19,6 +21,7 @@ logger = logging.getLogger("ComfyUI-OpenClaw.services.audit")
_TRUTHY = {"1", "true", "yes", "on"}
_AUDIT_MAX_BYTES_DEFAULT = 5 * 1024 * 1024
_AUDIT_BACKUPS_DEFAULT = 3
_AUDIT_CHAIN_KEY: Optional[bytes] = None
def _default_audit_log_path() -> str:
@@ -150,11 +153,27 @@ def _sanitize_audit_details(details: Optional[Dict[str, Any]]) -> Any:
return redact_json(safe_details)
def _get_audit_chain_key() -> bytes:
global _AUDIT_CHAIN_KEY
if _AUDIT_CHAIN_KEY is None:
raw = os.environ.get("OPENCLAW_AUDIT_CHAIN_KEY") or os.environ.get(
"MOLTBOT_AUDIT_CHAIN_KEY"
)
_AUDIT_CHAIN_KEY = raw.encode("utf-8") if raw else secrets.token_bytes(32)
return _AUDIT_CHAIN_KEY
def _chain_hash(prev_hash: str, entry: Dict[str, Any]) -> str:
payload = json.dumps(
entry, sort_keys=True, separators=(",", ":"), ensure_ascii=True
)
return hashlib.sha256(f"{prev_hash}|{payload}".encode("utf-8")).hexdigest()
# IMPORTANT: keep audit-chain hashing keyed; plain SHA-256 on sensitive events
# triggers CodeQL and weakens correlation resistance.
return hmac.new(
_get_audit_chain_key(),
f"{prev_hash}|{payload}".encode("utf-8"),
hashlib.sha256,
).hexdigest()
def _rotate_if_needed(path: str) -> None:
+12 -3
View File
@@ -106,6 +106,7 @@ class BridgeTokenStore:
self._audit_trail: List[TokenAuditEvent] = []
self._state_dir = state_dir
self._store_path: Optional[Path] = None
self._token_index_key = self._build_token_index_key()
if state_dir:
self._store_path = Path(state_dir) / "bridge_tokens.json"
self._load()
@@ -171,11 +172,19 @@ class BridgeTokenStore:
# --- Token hashing ---
@staticmethod
def _hash_token(token_value: str) -> str:
def _build_token_index_key(self) -> bytes:
raw = os.environ.get("OPENCLAW_BRIDGE_TOKEN_INDEX_KEY") or os.environ.get(
"MOLTBOT_BRIDGE_TOKEN_INDEX_KEY"
)
if raw:
return raw.encode("utf-8")
return secrets.token_bytes(32)
def _hash_token(self, token_value: str) -> str:
"""Constant-time-safe hash for token lookup."""
# IMPORTANT: never fall back to a hardcoded token-index key here.
return hmac.new(
b"openclaw-bridge-token-index",
self._token_index_key,
token_value.encode("utf-8"),
"sha256",
).hexdigest()
+51
View File
@@ -0,0 +1,51 @@
import os
import unittest
from unittest.mock import patch
import services.audit as audit_module
from services.bridge_token_lifecycle import BridgeTokenStore
class TestS79AuditHashing(unittest.TestCase):
def test_chain_hash_is_keyed(self):
entry = {"action": "config.update", "target": "settings.json"}
with patch.object(audit_module, "_AUDIT_CHAIN_KEY", b"key-a"):
hash_a = audit_module._chain_hash("GENESIS", entry)
with patch.object(audit_module, "_AUDIT_CHAIN_KEY", b"key-b"):
hash_b = audit_module._chain_hash("GENESIS", entry)
self.assertNotEqual(hash_a, hash_b)
def test_chain_hash_is_stable_for_same_key(self):
entry = {"action": "config.update", "target": "settings.json"}
with patch.object(audit_module, "_AUDIT_CHAIN_KEY", b"fixed-key"):
hash_a = audit_module._chain_hash("GENESIS", entry)
hash_b = audit_module._chain_hash("GENESIS", entry)
self.assertEqual(hash_a, hash_b)
class TestS79BridgeTokenHashing(unittest.TestCase):
def tearDown(self):
os.environ.pop("OPENCLAW_BRIDGE_TOKEN_INDEX_KEY", None)
os.environ.pop("MOLTBOT_BRIDGE_TOKEN_INDEX_KEY", None)
def test_token_hash_differs_across_store_instances_without_override(self):
store_a = BridgeTokenStore()
store_b = BridgeTokenStore()
self.assertNotEqual(store_a._hash_token("secret-token"), store_b._hash_token("secret-token"))
def test_token_hash_can_be_pinned_via_env_override(self):
with patch.dict(
os.environ,
{"OPENCLAW_BRIDGE_TOKEN_INDEX_KEY": "fixed-token-index-key"},
clear=False,
):
store_a = BridgeTokenStore()
store_b = BridgeTokenStore()
self.assertEqual(store_a._hash_token("secret-token"), store_b._hash_token("secret-token"))