roberto
c8ef7b119b
- Updated `TestModuleSingletons` in `test_execution_plan.py` to reflect new agent templates and playbook names. - Changed assertions in playbook tests to match updated templates and agents. - Introduced `test_storage.py` to cover the storage layer, including encryption, BlobStore, and VectorStore functionalities. - Added tests for S3 interactions, ensuring upload, download, delete, and list operations work as expected. - Implemented mock tests for Pinecone and Qdrant vector stores to validate upsert, search, and delete operations.
33 lines
1017 B
Python
33 lines
1017 B
Python
"""Integrity verification only — the backend NEVER decrypts user data."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
|
|
from fastapi import HTTPException
|
|
|
|
|
|
def verify_checksum(blob: bytes, checksum: str) -> bool:
|
|
"""Return ``True`` if SHA-256(blob) matches *checksum*.
|
|
|
|
Uses ``hmac.compare_digest`` for constant-time comparison to prevent
|
|
timing-based side-channel attacks.
|
|
"""
|
|
computed = hashlib.sha256(blob).hexdigest()
|
|
return hmac.compare_digest(computed, checksum)
|
|
|
|
|
|
def reject_if_tampered(blob: bytes, checksum: str) -> None:
|
|
"""Raise ``HTTP 400`` if the blob does not match its checksum.
|
|
|
|
Call this before storing or forwarding any client-provided blob.
|
|
The backend never holds decryption keys — this check only verifies
|
|
that the opaque bytes arrived intact.
|
|
"""
|
|
if not verify_checksum(blob, checksum):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Checksum mismatch: blob integrity check failed",
|
|
)
|