| from __future__ import annotations |
|
|
| import re |
| import shlex |
| from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit |
|
|
|
|
| REDACTED = "<redacted>" |
|
|
| SENSITIVE_KEYS = ( |
| "token", |
| "secret", |
| "password", |
| "passwd", |
| "api_key", |
| "apikey", |
| "authorization", |
| "auth", |
| "cookie", |
| "client_secret", |
| "access_token", |
| "refresh_token", |
| ) |
|
|
| TOKEN_PATTERNS = ( |
| re.compile(r"sk-[A-Za-z0-9_-]{12,}"), |
| re.compile(r"sk-ant-[A-Za-z0-9_-]{12,}"), |
| re.compile(r"ghp_[A-Za-z0-9_]{12,}"), |
| re.compile(r"github_pat_[A-Za-z0-9_]{12,}"), |
| re.compile(r"xox[baprs]-[A-Za-z0-9-]{12,}"), |
| re.compile(r"AIza[A-Za-z0-9_-]{12,}"), |
| re.compile(r"\bAKIA[A-Z0-9]{12,}\b"), |
| ) |
|
|
|
|
| def _is_sensitive_key(key: str) -> bool: |
| normalized = key.lower().replace("-", "_") |
| return any(item in normalized for item in SENSITIVE_KEYS) |
|
|
|
|
| def _mask_url(value: str) -> str: |
| try: |
| parts = urlsplit(value) |
| except ValueError: |
| return value |
| if not parts.scheme or not parts.netloc: |
| return value |
|
|
| netloc = parts.netloc |
| changed = False |
| if "@" in netloc: |
| host = netloc.rsplit("@", 1)[1] |
| netloc = f"{REDACTED}@{host}" |
| changed = True |
|
|
| query_items = [] |
| for key, item_value in parse_qsl(parts.query, keep_blank_values=True): |
| if _is_sensitive_key(key) or key.lower() in {"code", "key"}: |
| query_items.append((key, REDACTED)) |
| changed = True |
| else: |
| query_items.append((key, item_value)) |
|
|
| if not changed: |
| return value |
| return urlunsplit( |
| (parts.scheme, netloc, parts.path, urlencode(query_items), parts.fragment) |
| ) |
|
|
|
|
| def mask_value(value: str) -> str: |
| masked = _mask_url(value) |
| for pattern in TOKEN_PATTERNS: |
| masked = pattern.sub(REDACTED, masked) |
| return masked |
|
|
|
|
| def sanitize_argv(argv: list[str] | tuple[str, ...]) -> tuple[tuple[str, ...], int]: |
| result: list[str] = [] |
| masked_count = 0 |
| mask_next = False |
|
|
| for arg in argv: |
| if mask_next: |
| result.append(REDACTED) |
| masked_count += 1 |
| mask_next = False |
| continue |
|
|
| url_masked = _mask_url(arg) |
| if url_masked != arg: |
| masked = mask_value(url_masked) |
| result.append(masked) |
| masked_count += 1 |
| continue |
|
|
| if "=" in arg: |
| key, _value = arg.split("=", 1) |
| stripped_key = key.lstrip("-") |
| if _is_sensitive_key(stripped_key): |
| result.append(f"{key}={REDACTED}") |
| masked_count += 1 |
| continue |
|
|
| stripped = arg.lstrip("-") |
| if arg.startswith("-") and _is_sensitive_key(stripped): |
| result.append(arg) |
| mask_next = True |
| continue |
|
|
| masked = mask_value(arg) |
| if masked != arg: |
| masked_count += 1 |
| result.append(masked) |
|
|
| return tuple(result), masked_count |
|
|
|
|
| def render_argv(argv: list[str] | tuple[str, ...], *, max_len: int = 280) -> str: |
| rendered = " ".join(shlex.quote(arg) for arg in argv) |
| if len(rendered) <= max_len: |
| return rendered |
| if max_len <= 3: |
| return "." * max_len |
| return rendered[: max(0, max_len - 3)] + "..." |
|
|
