150 lines
5.3 KiB
Python
150 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
import hashlib
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.encryption import (
|
|
CAPABILITY_ENCRYPTION_CONTENT_CIPHER,
|
|
ContentProtectionEnvelope,
|
|
ProtectedContent,
|
|
)
|
|
from govoplan_files.backend.db.models import FileBlob
|
|
from govoplan_files.backend.runtime import configure_runtime
|
|
from govoplan_files.backend.storage.common import FileStorageError
|
|
from govoplan_files.backend.storage.content_protection import protect_blob_content
|
|
from govoplan_files.backend.storage.integrity import read_verified_blob_bytes
|
|
|
|
|
|
class Registry:
|
|
def __init__(self, capability=None) -> None:
|
|
self.capability_value = capability
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return (
|
|
name == CAPABILITY_ENCRYPTION_CONTENT_CIPHER
|
|
and self.capability_value is not None
|
|
)
|
|
|
|
def capability(self, _name: str):
|
|
return self.capability_value
|
|
|
|
|
|
class Cipher:
|
|
def protect_content(self, _session, *, request):
|
|
ciphertext = b"protected:" + request.plaintext
|
|
envelope = ContentProtectionEnvelope(
|
|
envelope_id="envelope-file-1",
|
|
tenant_id=request.tenant_id,
|
|
owner_module=request.owner_module,
|
|
resource_type=request.resource_type,
|
|
resource_id=request.resource_id,
|
|
profile_kind="server_envelope",
|
|
profile_id=request.profile_id,
|
|
provider_id="test",
|
|
vault_id=request.vault_id,
|
|
key_version=1,
|
|
algorithm_suite="AES-256-GCM",
|
|
ciphertext_ref=request.ciphertext_ref,
|
|
ciphertext_digest=f"sha256:{hashlib.sha256(ciphertext).hexdigest()}",
|
|
authenticated_context_digest="sha256:context",
|
|
state="active",
|
|
created_at=datetime.now(tz=UTC),
|
|
wrapped_key_refs=("wrapped:1",),
|
|
)
|
|
return ProtectedContent(envelope=envelope, ciphertext=ciphertext)
|
|
|
|
def unprotect_content(self, _session, *, request):
|
|
if not request.ciphertext.startswith(b"protected:"):
|
|
raise ValueError("invalid fixture ciphertext")
|
|
return request.ciphertext.removeprefix(b"protected:")
|
|
|
|
def execute_rewrap(self, *_args, **_kwargs):
|
|
raise NotImplementedError
|
|
|
|
def prepare_reencryption(self, *_args, **_kwargs):
|
|
raise NotImplementedError
|
|
|
|
|
|
class Backend:
|
|
def __init__(self, data: bytes) -> None:
|
|
self.data = data
|
|
|
|
def get_bytes(self, _key: str) -> bytes:
|
|
return self.data
|
|
|
|
|
|
class FileContentProtectionTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
FileBlob.__table__.create(self.engine)
|
|
self.session = Session(self.engine)
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_encrypted_blob_is_opened_after_stored_object_integrity_check(self) -> None:
|
|
configure_runtime(registry=Registry(Cipher()), settings=object())
|
|
plaintext = b"monthly records"
|
|
protected = protect_blob_content(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
blob_id="blob-1",
|
|
vault_id="vault-1",
|
|
ciphertext_ref="tenants/tenant-1/files/blob-1",
|
|
plaintext=plaintext,
|
|
actor_id="account-1",
|
|
content_type="text/plain",
|
|
)
|
|
blob = FileBlob(
|
|
id="blob-1",
|
|
tenant_id="tenant-1",
|
|
storage_backend="test",
|
|
storage_key="tenants/tenant-1/files/blob-1",
|
|
checksum_sha256=hashlib.sha256(plaintext).hexdigest(),
|
|
size_bytes=len(plaintext),
|
|
protection_discriminator="vault:vault-1",
|
|
encryption_envelope_id=protected.envelope.envelope_id,
|
|
storage_checksum_sha256=hashlib.sha256(protected.ciphertext).hexdigest(),
|
|
storage_size_bytes=len(protected.ciphertext),
|
|
content_type="text/plain",
|
|
ref_count=1,
|
|
integrity_status="verified",
|
|
)
|
|
self.session.add(blob)
|
|
self.session.flush()
|
|
self.assertEqual(
|
|
plaintext,
|
|
read_verified_blob_bytes(blob, backend=Backend(protected.ciphertext)),
|
|
)
|
|
|
|
def test_encrypted_blob_fails_closed_without_encryption_capability(self) -> None:
|
|
configure_runtime(registry=Registry(), settings=object())
|
|
ciphertext = b"protected:monthly records"
|
|
blob = FileBlob(
|
|
id="blob-1",
|
|
tenant_id="tenant-1",
|
|
storage_backend="test",
|
|
storage_key="tenants/tenant-1/files/blob-1",
|
|
checksum_sha256=hashlib.sha256(b"monthly records").hexdigest(),
|
|
size_bytes=len(b"monthly records"),
|
|
protection_discriminator="vault:vault-1",
|
|
encryption_envelope_id="envelope-file-1",
|
|
storage_checksum_sha256=hashlib.sha256(ciphertext).hexdigest(),
|
|
storage_size_bytes=len(ciphertext),
|
|
ref_count=1,
|
|
integrity_status="verified",
|
|
)
|
|
self.session.add(blob)
|
|
self.session.flush()
|
|
with self.assertRaisesRegex(FileStorageError, "Encryption is unavailable"):
|
|
read_verified_blob_bytes(blob, backend=Backend(ciphertext))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|