diff --git a/src/cmcp_gateway/audit/chain.py b/src/cmcp_gateway/audit/chain.py index 10df0ded..65be4e56 100644 --- a/src/cmcp_gateway/audit/chain.py +++ b/src/cmcp_gateway/audit/chain.py @@ -117,10 +117,15 @@ def append( workflow_id: str | None = None, ) -> AuditEntry: prev_hash = self._entries[-1].entry_hash if self._entries else "genesis" + now = datetime.now(tz=UTC) + if self._entries: + prev_ts = datetime.fromisoformat(self._entries[-1].timestamp_utc) + if now < prev_ts: + now = prev_ts entry = AuditEntry( entry_id=str(uuid4()), sequence_number=len(self._entries), - timestamp_utc=datetime.now(tz=UTC).isoformat(), + timestamp_utc=now.isoformat(), session_id=self._session_id, call_id=call_id, entry_type=entry_type, diff --git a/tests/unit/test_audit.py b/tests/unit/test_audit.py index cdc7a55e..c345eea9 100644 --- a/tests/unit/test_audit.py +++ b/tests/unit/test_audit.py @@ -2,6 +2,9 @@ from __future__ import annotations +from datetime import UTC, datetime, timedelta +from unittest.mock import patch + from cmcp_gateway.audit.chain import AuditChain from cmcp_gateway.audit.keys import SigningKey @@ -119,3 +122,37 @@ def test_audit_chain_entries_returns_copy(): entries = chain.entries entries.clear() assert chain.length == 1 # original not affected + + +# ── AUDIT-001: monotonic timestamp guard ────────────────────────────────────── + +def test_timestamp_clamped_when_clock_steps_backward(): + """AUDIT-001: if wall-clock steps back, timestamp is clamped to previous entry's time.""" + chain = AuditChain("sess-001") + first_ts = datetime.fromisoformat(chain.entries[0].timestamp_utc) + + # Simulate a clock that returns a time 5 seconds in the past on the next call + backward = first_ts - timedelta(seconds=5) + with patch("cmcp_gateway.audit.chain.datetime") as mock_dt: + mock_dt.now.return_value = backward + mock_dt.fromisoformat = datetime.fromisoformat + entry = chain.append("tool_call", call_id="c1", tool_name="t", policy_decision="allow") + + appended_ts = datetime.fromisoformat(entry.timestamp_utc) + assert appended_ts >= first_ts, ( + f"timestamp {appended_ts} must not be before previous entry {first_ts}" + ) + + +def test_timestamp_not_clamped_when_clock_moves_forward(): + """AUDIT-001: normal forward-moving timestamps are preserved as-is.""" + chain = AuditChain("sess-001") + first_ts = datetime.fromisoformat(chain.entries[0].timestamp_utc) + + forward = first_ts + timedelta(seconds=10) + with patch("cmcp_gateway.audit.chain.datetime") as mock_dt: + mock_dt.now.return_value = forward + mock_dt.fromisoformat = datetime.fromisoformat + entry = chain.append("tool_call", call_id="c1", tool_name="t", policy_decision="allow") + + assert datetime.fromisoformat(entry.timestamp_utc) == forward