File size: 3,264 Bytes
06357e7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
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)] + "..."