Issue #46: implement Python async client values errors and CLI
This commit is contained in:
@@ -0,0 +1,58 @@
|
||||
"""Authentication metadata helpers for MXAccess Gateway clients."""
|
||||
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
|
||||
AUTHORIZATION_HEADER = "authorization"
|
||||
REDACTED = "[redacted]"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ApiKey:
|
||||
"""API key wrapper that avoids leaking the secret through repr output."""
|
||||
|
||||
value: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.value:
|
||||
raise ValueError("api_key must not be empty")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{type(self).__name__}({REDACTED!r})"
|
||||
|
||||
def bearer_value(self) -> str:
|
||||
return f"Bearer {self.value}"
|
||||
|
||||
|
||||
def auth_metadata(api_key: str | ApiKey | None) -> tuple[tuple[str, str], ...]:
|
||||
"""Return gRPC metadata for API key auth."""
|
||||
|
||||
if api_key is None:
|
||||
return ()
|
||||
|
||||
key = api_key.value if isinstance(api_key, ApiKey) else api_key
|
||||
if not key:
|
||||
return ()
|
||||
|
||||
return ((AUTHORIZATION_HEADER, f"Bearer {key}"),)
|
||||
|
||||
|
||||
def merge_metadata(
|
||||
api_key: str | ApiKey | None,
|
||||
metadata: Sequence[tuple[str, str]] | None = None,
|
||||
) -> tuple[tuple[str, str], ...]:
|
||||
"""Merge caller metadata with API key metadata."""
|
||||
|
||||
merged = list(metadata or ())
|
||||
merged.extend(auth_metadata(api_key))
|
||||
return tuple(merged)
|
||||
|
||||
|
||||
def redact_secret(text: str, secrets: Sequence[str | None]) -> str:
|
||||
"""Replace known secret values with a stable redaction marker."""
|
||||
|
||||
redacted = text
|
||||
for secret in secrets:
|
||||
if secret:
|
||||
redacted = redacted.replace(secret, REDACTED)
|
||||
return redacted
|
||||
Reference in New Issue
Block a user