mirror of
https://github.com/qdrant/qdrant.git
synced 2026-07-30 14:41:00 -05:00
6f2e19a43a
* Skip audit logging for telemetry endpoints Telemetry access checks (telemetry_memory, telemetry_requests, telemetry_cluster, cluster_telemetry) fire on every metrics scrape and produce very noisy audit logs with no security-relevant signal. Switch all telemetry callers to use `auth.unlogged_access()` to bypass audit logging, consistent with the existing metrics and prepare_data handlers that already did this. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * trigger CI * feat: add some tests --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com> Co-authored-by: Daniel Boros <dancixx@gmail.com>
55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
import random
|
|
import string
|
|
|
|
import jwt
|
|
from consensus_tests.utils import start_cluster
|
|
|
|
PORT_SEED = 10000
|
|
REST_URI = f"http://127.0.0.1:{PORT_SEED + 2}"
|
|
GRPC_URI = f"127.0.0.1:{PORT_SEED + 1}"
|
|
|
|
SECRET = "my_top_secret_key"
|
|
ALT_SECRET = "my_alternative_secret_key"
|
|
|
|
READ_ONLY_API_KEY = "boo-hoo, this can only read!"
|
|
|
|
API_KEY_HEADERS = {"Api-Key": SECRET}
|
|
API_KEY_METADATA = [("api-key", SECRET)]
|
|
READ_ONLY_API_KEY_METADATA = [("api-key", READ_ONLY_API_KEY)]
|
|
|
|
|
|
def start_jwt_protected_cluster(tmp_path, num_peers=1, extra_env=None):
|
|
base_env = {
|
|
"QDRANT__SERVICE__API_KEY": SECRET,
|
|
"QDRANT__SERVICE__ALT_API_KEY": ALT_SECRET,
|
|
"QDRANT__SERVICE__READ_ONLY_API_KEY": READ_ONLY_API_KEY,
|
|
"QDRANT__SERVICE__JWT_RBAC": "true",
|
|
"QDRANT__STORAGE__WAL__WAL_CAPACITY_MB": "1", # to speed up snapshot tests
|
|
}
|
|
extra_env = {
|
|
**base_env,
|
|
**(extra_env or {}),
|
|
}
|
|
|
|
peer_api_uris, peer_dirs, bootstrap_uri = start_cluster(
|
|
tmp_path,
|
|
num_peers=num_peers,
|
|
port_seed=PORT_SEED,
|
|
extra_env=extra_env,
|
|
headers=API_KEY_HEADERS,
|
|
)
|
|
|
|
assert REST_URI in peer_api_uris
|
|
|
|
return peer_api_uris, peer_dirs, bootstrap_uri
|
|
|
|
|
|
def encode_jwt(claims: dict, secret: str) -> str:
|
|
return jwt.encode(claims, secret, algorithm="HS256")
|
|
|
|
def decode_jwt(token: str, secret: str) -> dict:
|
|
return jwt.decode(token, secret, algorithms=["HS256"])
|
|
|
|
def random_str():
|
|
return "".join(random.choices(string.ascii_lowercase, k=10))
|