VectaX - Mirror Security - Qdrant 向量数据库

文档
框架
VectaX - Mirror Security

VectaX 是由 Mirror Security 开发的一款以 AI 为中心的访问控制和加密系统，专为管理和保护向量嵌入而设计。它结合了保持相似性的加密技术和细粒度的 RBAC（基于角色的访问控制），以实现向量数据的安全存储、检索和操作。

它可以与 Qdrant 集成，以确保向量搜索的安全性。

我们将了解如何使用基本的 VectaX 向量加密和复杂的 RBAC 机制来实现这一点。您可以从 Mirror Security Platform 获取 API 密钥和 Mirror SDK。

让我们同时设置 VectaX 和 Qdrant 客户端。

from mirror_sdk.core.mirror_core import MirrorSDK, MirrorConfig
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

# Get your API key from
# https://platform.mirrorsecurity.io
config = MirrorConfig(
    api_key="<your_api_key>",
    server_url="https://mirrorapi.azure-api.net/v1",
    secret="<your_encrypt_secret>",
)
mirror_sdk = MirrorSDK(config)

# Connects to http://localhost:6333/ by default
qdrant = QdrantClient()

向量加密

现在，让我们使用 VectaX 加密来保护向量嵌入。

from qdrant_client.models import PointStruct
from mirror_sdk.core.models import VectorData

# Generate or retrieve vector embeddings
# embedding = generate_document_embedding()

vector_data = VectorData(vector=embedding, id="doc1")
encrypted = mirror_sdk.vectax.encrypt(vector_data)

point = PointStruct(
    id=0,
    vector=encrypted.ciphertext,
    payload={
        "content": "Document content",
        "iv": encrypted.iv,
        "auth_hash": encrypted.auth_hash
    }
)
qdrant.upsert(collection_name="vectax", points=[point])

# Encrypt a query vector for secure search
# query_embedding = generate_query_embedding(...)

encrypted_query = mirror_sdk.vectax.encrypt(
    VectorData(vector=query_embedding, id="query")
)

results = qdrant.query_points(
    collection_name="vectax",
    query=encrypted_query.ciphertext,
    limit=5
).points

使用 RBAC 进行向量搜索

RBAC 允许基于角色、组和部门对加密向量数据进行细粒度访问控制。

定义访问策略

app_policy = {
    "roles": ["admin", "analyst", "user"],
    "groups": ["team_a", "team_b"],
    "departments": ["research", "engineering"],
}
mirror_sdk.set_policy(app_policy)

生成访问密钥

# Generate a secret key for use by the 'admin' role holders.
admin_key = mirror_sdk.rbac.generate_user_secret_key(
    {"roles": ["admin"], "groups": ["team_a"], "departments": ["research"]}
)

存储带有 RBAC 策略的加密数据

现在我们可以存储只有拥有“admin”角色的用户才能访问的数据。

from mirror_sdk.core.models import RBACVectorData
from mirror_sdk.utils import encode_binary_data

policy = {
    "roles": ["admin"],
    "groups": ["team_a"],
    "departments": ["research"],
}
# vector_embedding = generate_vector_embedding(...)
vector_data = RBACVectorData(
    # Generate or retrieve vector embeddings
    vector=vector_embedding,
    id=1,
    access_policy=policy,
)
encrypted = mirror_sdk.rbac.encrypt(vector_data)

qdrant.upsert(
    collection_name="vectax",
    points=[
        models.PointStruct(
            id=1,
            vector=encrypted.crypto.ciphertext,
            payload={
                "encrypted_header": encrypted.encrypted_header,
                "encrypted_vector_metadata": encode_binary_data(
                    encrypted.crypto.serialize()
                ),
                "content": "My content",
            },
        )
    ],
)

使用基于角色的解密进行查询

使用 admin 密钥，只有可访问的数据才会被解密。

from mirror_sdk.core import MirrorError
from mirror_sdk.core.models import MirrorCrypto
from mirror_sdk.utils import decode_binary_data

# Encrypt a query vector for secure search
# query_embedding = generate_query_embedding(...)

query_data = RBACVectorData(vector=query_embedding, id="query", access_policy=policy)
encrypted_query = mirror_sdk.rbac.encrypt(query_data)

results = qdrant.query_points(
    collection_name="vectax", query=encrypted_query.crypto.ciphertext, limit=10
)

accessible_results = []
for point in results.points:
    try:
        encrypted_vector_metadata = decode_binary_data(
            point.payload["encrypted_vector_metadata"]
        )
        mirror_data = MirrorCrypto.deserialize(encrypted_vector_metadata)
        admin_decrypted = mirror_sdk.rbac.decrypt(
            mirror_data,
            point.payload["encrypted_header"],
            admin_key,
        )
        accessible_results.append(
            {
                "id": point.id,
                "content": point.payload["content"],
                "score": point.score,
                "accessible": True,
            }
        )

    except MirrorError as e:
        print(f"Access denied for point {point.id}: {e}")

# Proceed to only use results within `accessible_results`.