64 lines
1.5 KiB
Python
64 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from contextvars import ContextVar, Token
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
|
|
CORRELATION_HEADER = "X-Correlation-ID"
|
|
_correlation_id: ContextVar[str | None] = ContextVar(
|
|
"cloud_correlation_id",
|
|
default=None,
|
|
)
|
|
_SENSITIVE_KEYS = {
|
|
"authorization",
|
|
"token",
|
|
"bearer_token",
|
|
"password",
|
|
"screenshot",
|
|
"ui_tree",
|
|
"typed_text",
|
|
"text_input",
|
|
}
|
|
|
|
|
|
def new_correlation_id() -> str:
|
|
return uuid4().hex
|
|
|
|
|
|
def normalize_correlation_id(value: str | None) -> str:
|
|
if value is None:
|
|
return new_correlation_id()
|
|
normalized = value.strip()
|
|
if not normalized or len(normalized) > 128:
|
|
return new_correlation_id()
|
|
return normalized
|
|
|
|
|
|
def bind_correlation_id(correlation_id: str) -> Token[str | None]:
|
|
return _correlation_id.set(correlation_id)
|
|
|
|
|
|
def reset_correlation_id(token: Token[str | None]) -> None:
|
|
_correlation_id.reset(token)
|
|
|
|
|
|
def current_correlation_id() -> str:
|
|
correlation_id = _correlation_id.get()
|
|
return correlation_id or new_correlation_id()
|
|
|
|
|
|
def redact_sensitive_fields(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
return {
|
|
key: "[REDACTED]"
|
|
if key.lower() in _SENSITIVE_KEYS
|
|
else redact_sensitive_fields(item)
|
|
for key, item in value.items()
|
|
}
|
|
if isinstance(value, list):
|
|
return [redact_sensitive_fields(item) for item in value]
|
|
if isinstance(value, tuple):
|
|
return tuple(redact_sensitive_fields(item) for item in value)
|
|
return value
|