|
5 | 5 | import logging
|
6 | 6 | import re
|
7 | 7 | from base64 import b64decode, b64encode
|
| 8 | +from collections.abc import Mapping |
8 | 9 | from datetime import datetime, time, timezone
|
| 10 | +from typing import TypeVar, cast, overload |
9 | 11 | from urllib.parse import urljoin as _urljoin
|
10 | 12 | from warnings import warn
|
11 | 13 |
|
12 | 14 | _LOGGER = logging.getLogger(__name__)
|
| 15 | +_T = TypeVar("_T") |
13 | 16 |
|
14 | 17 | ENCODING = "utf-8"
|
| 18 | +REDACTED = "**REDACTED**" |
| 19 | +REDACT_FIELDS = [ |
| 20 | + "token", |
| 21 | + "idToken", |
| 22 | + "refreshToken", |
| 23 | + "userId", |
| 24 | + "userEmail", |
| 25 | + "sessionId", |
| 26 | + "oneSignalPlayerId", |
| 27 | + "deviceId", |
| 28 | + "id", |
| 29 | + "litterRobotId", |
| 30 | + "unitId", |
| 31 | + "litterRobotSerial", |
| 32 | + "serial", |
| 33 | +] |
15 | 34 |
|
16 | 35 |
|
17 | 36 | def decode(value: str) -> str:
|
@@ -79,3 +98,38 @@ def send_deprecation_warning(
|
79 | 98 | message = f"{old_name} has been deprecated{'' if new_name is None else f' in favor of {new_name}'} and will be removed in a future release"
|
80 | 99 | warn(message, DeprecationWarning, stacklevel=2)
|
81 | 100 | _LOGGER.warning(message)
|
| 101 | + |
| 102 | + |
| 103 | +@overload |
| 104 | +def redact(data: Mapping) -> dict: # type: ignore[misc] |
| 105 | + ... |
| 106 | + |
| 107 | + |
| 108 | +@overload |
| 109 | +def redact(data: _T) -> _T: |
| 110 | + ... |
| 111 | + |
| 112 | + |
| 113 | +def redact(data: _T) -> _T: |
| 114 | + """Redact sensitive data in a dict.""" |
| 115 | + if not isinstance(data, (Mapping, list)): |
| 116 | + return data |
| 117 | + |
| 118 | + if isinstance(data, list): |
| 119 | + return cast(_T, [redact(val) for val in data]) |
| 120 | + |
| 121 | + redacted = {**data} |
| 122 | + |
| 123 | + for key, value in redacted.items(): |
| 124 | + if value is None: |
| 125 | + continue |
| 126 | + if isinstance(value, str) and not value: |
| 127 | + continue |
| 128 | + if key in REDACT_FIELDS: |
| 129 | + redacted[key] = REDACTED |
| 130 | + elif isinstance(value, Mapping): |
| 131 | + redacted[key] = redact(value) |
| 132 | + elif isinstance(value, list): |
| 133 | + redacted[key] = [redact(item) for item in value] |
| 134 | + |
| 135 | + return cast(_T, redacted) |
0 commit comments