rethink-public/commons
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: simplify masking to dumb mask_url/mask_proxy + add stable_id

Browse Source 
The selective-key URL/proxy machinery (_mask_qs, SENSITIVE_QUERY_KEYS,
_split_host, _bullet_trailing_fields) tried to redact tokens inside arbitrary
URL structure and leaked on shapes it didn't anticipate (SPA-route fragments,
colon-passwords). Replace with purpose-honest functions that cannot leak or
over-mask:

- mask_url: strips query + fragment, keeps scheme/host/path. Does not parse or
  hunt for sensitive params - a secret in a URL is the caller's bug.
- mask_proxy: partial user-reportable proxy id - drops any userinfo credentials
  and masks an IPv4 host's middle octets (A.***.***.D, port kept); non-IPv4
  host passes through; no parseable host raises ValueError.
- stable_id: deterministic sha256-derived id from ordered str parts (\x00-joined,
  regenerable); empty/non-str part or non-positive length raises ValueError.

credit/cvv/phantom/provider unchanged. Export stable_id from the facade.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
dsql
2026-07-06 18:25:06 -04:00
parent e6e655335e
commit 0030daeb7b
3 changed files with 87 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+27 -24
View File
@@ -144,7 +144,7 @@ Display helpers only — they format a value for showing; they are **not a secur
control** (the underlying value is unchanged and still needs proper handling).
```python
from commons import credit, cvv, phantom, provider, mask_url, mask_proxy
from commons import credit, cvv, phantom, provider, mask_url, mask_proxy, stable_id
credit("4111 1111 1111 1234") # "•••• •••• •••• 1234"
cvv("123") # "•••"
@@ -153,32 +153,35 @@ phantom("1234567890") # "••••••••••" (len <= 10
provider("4111111111111111") # "VISA" (VISA/MC/AMEX/UPAY/DISC/JCB/DNRS/UNKW)
provider("4111²111111111234") # "VISA" (unicode digit lookalikes ignored, never raises)
# redact credentials in a connection string before logging it
mask_url("https://u:pw@api.x/v2?apiKey=SECRET&ip=8.8.8.8")
# -> "https://api.x/v2?apiKey=***&ip=8.8.8.8" (userinfo dropped, secret query masked)
mask_url("redis://:pw@127.0.0.1:6379/0") # -> "redis://127.0.0.1:6379/0"
mask_url("https://x/cb#access_token=SECRET&token_type=bearer")
# -> "https://x/cb#access_token=***&token_type=bearer" (oauth implicit-grant fragment masked)
mask_proxy("1.2.3.4:8080:user:supersecret") # -> "1.2.3.4:8080:user:****"
mask_proxy("1.2.3.4:8080") # -> "1.2.3.4:8080" (no auth, untouched)
mask_proxy("user:supersecret@1.2.3.4:8080") # -> "1.2.3.4:8080" (userinfo shape, also masked)
# strip query + fragment off a url before logging it
mask_url("https://u:pw@api.x/v2?apiKey=SECRET&ip=8.8.8.8") # -> "https://u:pw@api.x/v2"
mask_url("https://api.x/v2") # -> "https://api.x/v2" (unchanged)
# partial, user-reportable proxy id: creds stripped, ipv4 middle octets masked
mask_proxy("172.58.32.1:8080") # -> "172.***.***.1:8080"
mask_proxy("user:supersecret@172.58.32.1:8080") # -> "172.***.***.1:8080" (creds gone)
mask_proxy("proxy.host.net:8080") # -> "proxy.host.net:8080" (hostname untouched)
# deterministic, regenerable id from ordered parts
stable_id("profit_lounge", "user_123") # -> "d1f4…" (same 16 hex chars every call)
```
`mask_url` strips `user:pass@` userinfo and replaces the values of sensitive query
params (`apiKey`, `token`, `password`, `secret`, …; override via `keys=`) with `***`.
Non-sensitive query values are re-percent-encoded on the way out, so a value with a
reserved character (`&`, `=`, a space, …) round-trips correctly instead of corrupting
the rebuilt URL. The URL fragment is masked the same way as the query when it is
genuinely `key=value&key=value` shaped (e.g. an OAuth implicit-grant callback) — a
fragment that only incidentally contains `=` (an SPA hash route like `#/page?x=1`) is
left untouched rather than risking a lossy rewrite of a non-secret value, and a plain
anchor (`#section`) always passes through unchanged.
`mask_url` drops the query and fragment, keeping scheme/host/path. It does **not** parse
or hunt for sensitive params — a secret in a URL is the caller's bug, not this function's
to detect — so it structurally cannot leak a query param and cannot over-mask a legit one.
A URL with no query/fragment is returned unchanged. (Note: `user:pass@` userinfo is part of
the authority and is **not** stripped; keep credentials out of the URL you log.)
`mask_proxy` bullets the password of a `host:port:user:password` spec, and a plain
`host:port` (no auth, including a bracketed IPv6 host) passes through unchanged. Any
other credential-bearing shape — `user:pass@host:port` userinfo, a `scheme://`-prefixed
URL, or a colon spec with more than 4 parts — is masked rather than ever returned
verbatim; it never logs a password in the clear.
`mask_proxy` yields a partial identifier a user can safely report for debugging. Any
`user:pass@` userinfo (or a `scheme://` prefix) is dropped entirely, then an IPv4 host
`A.B.C.D` is masked to `A.***.***.D` with the port kept in full; a non-IPv4 host (hostname
or IPv6) passes through unchanged with only its credentials stripped. A spec with no
parseable host raises `ValueError`.
`stable_id` derives a fixed-length hex id from ordered string parts via SHA-256 — same
inputs always produce the same id, no state, so it's regenerable anywhere. Parts join with
a `\x00` separator so `("ab","c")` and `("a","bc")` differ; an empty or non-str part, or a
non-positive `length`, raises `ValueError`.
## retry
src/commons/__init__.py
+2 -1
View File
@@ -6,7 +6,7 @@ base is stdlib only, no dependencies (the addr geo lookups add aiohttp via the
from importlib.metadata import PackageNotFoundError, version
from . import addr, masking, paths, timing
from .masking import credit, cvv, mask_proxy, mask_url, phantom, provider
from .masking import credit, cvv, mask_proxy, mask_url, phantom, provider, stable_id
from .paths import deep_get, deep_gets, deep_set, deep_sets
from .retry import aretry, retry
from .timing import (
@@ -53,6 +53,7 @@ __all__ = [
"provider",
"mask_url",
"mask_proxy",
"stable_id",
"retry",
"aretry",
]
src/commons/masking.py
+58 -88
View File
@@ -2,14 +2,17 @@
DISPLAY only, not a security control: the underlying value is unchanged and still
needs proper handling (encryption at rest, etc.).
"""
from urllib.parse import urlsplit, urlunsplit
SENSITIVE_QUERY_KEYS = frozenset(
{"apikey", "api_key", "key", "token", "access_token", "refresh_token",
"auth", "password", "passwd", "pwd", "pass", "secret", "client_secret",
"sig", "signature", "session"}
)
- ``credit``/``cvv``/``phantom``/``provider`` mask card-shaped values for display.
- ``mask_url`` strips the query and fragment off a url (keeping scheme/host/path),
so a url can be logged without leaking whatever a caller stuck in its params - it
does NOT hunt for tokens; a secret in a url is the caller's bug, not this to detect.
- ``mask_proxy`` gives a partial, user-reportable proxy identifier: credentials are
stripped and an ipv4 host's middle octets are masked (first/last octet + port kept).
- ``stable_id`` derives a deterministic, regenerable id from ordered string parts.
"""
import hashlib
from urllib.parse import urlsplit, urlunsplit
def _digits(value: str) -> str:
@@ -65,98 +68,65 @@ def provider(card_number: str) -> str:
return "UNKW"
def _mask_qs(qs: str, sensitive: "frozenset[str]") -> str:
"""mask sensitive ``key=value`` values in a query/fragment blob, structure-preserving
def mask_url(url: str) -> str:
"""strip query and fragment from a url, keeping scheme, host and path
splits on ``&`` and each pair on its FIRST ``=``; a sensitive key's value (everything
after that first ``=``, so a ``=``-padded base64 token is covered whole) becomes ``***``.
non-sensitive segments are left BYTE-IDENTICAL - no re-encoding, so a value with a
reserved char, an embedded ``=``, or a non-kv segment (a spa hash route like ``/page``)
is never corrupted. a sensitive token in the fragment is therefore always masked, never
passed through verbatim.
does NOT parse or hunt for sensitive params - it simply drops everything after the
path, so it structurally cannot leak a query param and cannot over-mask a legit one.
a url with no query/fragment is returned unchanged.
"""
out = []
for segment in qs.split("&"):
key, sep, _ = segment.partition("=")
if sep and key.lower() in sensitive:
out.append(f"{key}=***")
else:
out.append(segment)
return "&".join(out)
parts = urlsplit(url)
return urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))
def mask_url(url: str, *, keys: "frozenset[str] | None" = None) -> str:
"""redact credentials in a url for logging: drop userinfo, ``***`` sensitive query
and fragment values
param-name matching is case-insensitive; ``keys`` overrides the default sensitive-name
set. non-url or unparseable input is returned unchanged. masking is structure-preserving
(see ``_mask_qs``): only a sensitive key's value is replaced, everything else is left
byte-identical, so a non-secret value is never corrupted and a non-kv fragment (a spa
hash route like ``#/page``, a plain anchor ``#section``) passes through unchanged. the
fragment is masked the same way as the query, covering oauth implicit-grant callbacks
that put ``access_token`` (including a ``=``-padded base64 token) in the fragment.
"""
sensitive = SENSITIVE_QUERY_KEYS if keys is None else frozenset(k.lower() for k in keys)
try:
parts = urlsplit(url)
except ValueError:
return url
if not parts.scheme and not parts.netloc:
return url
netloc = parts.netloc
if "@" in netloc:
netloc = netloc.rsplit("@", 1)[1]
query = _mask_qs(parts.query, sensitive) if parts.query else parts.query
fragment = _mask_qs(parts.fragment, sensitive) if parts.fragment else parts.fragment
return urlunsplit((parts.scheme, netloc, parts.path, query, fragment))
def _split_host(rest: str) -> "tuple[str, list[str]]":
"""split a ``host:port[:...]`` tail into its host token and the remaining colon fields
a leading ``[bracketed-ipv6-literal]`` is kept intact as the host token, since its
own internal colons are not field separators; anything else splits on the first colon.
"""
if rest.startswith("["):
end = rest.find("]")
if end != -1:
return rest[: end + 1], rest[end + 1:].lstrip(":").split(":")
host, _, tail = rest.partition(":")
return host, tail.split(":")
def _bullet_trailing_fields(rest: str) -> str:
"""redact any credential fields of a ``host:port[:user:password...]`` tail
keeps ``host`` and the ``port`` (the first colon field after host) in the clear; every
field AFTER the port is credential material and is replaced with a single ``****``,
regardless of how many there are (so a colon-bearing password can't leak a fragment).
a bare ``host`` or ``host:port`` (no trailing fields) passes through unchanged - fails
safe: anything with a credential field is bulleted, never returned verbatim.
"""
host, fields = _split_host(rest)
if fields == [""]:
return host
if len(fields) == 1:
return f"{host}:{fields[0]}"
port = fields[0]
return f"{host}:{port}:****"
def _is_ipv4(host: str) -> bool:
"""true if host is a dotted-quad ipv4 literal (four 0-255 octets)"""
octets = host.split(".")
if len(octets) != 4:
return False
for octet in octets:
if not (octet.isdigit() and octet.isascii()):
return False
if not 0 <= int(octet) <= 255:
return False
return True
def mask_proxy(spec: str) -> str:
"""redact credentials in a proxy spec for logging
"""mask the middle octets of a proxy ip, keeping first/last octet and full port; strips any credentials
drops a ``scheme://`` prefix and any ``user:pass@`` userinfo, then keeps only
``host:port`` in the clear and bullets EVERY trailing colon field (user/password/etc.)
to a single ``:****`` - so ``host:port:user:password`` and any longer spec (a
colon-bearing password, extra fields) all collapse to ``host:port:****``. a plain
``host:port`` or bare ``host`` (no auth) passes through unchanged. fails safe: whenever
a credential field is present it is bulleted, never returned verbatim.
drops a ``scheme://`` prefix and any ``user:pass@`` userinfo entirely (a credential is
removed, never partial-masked). an ipv4 host ``A.B.C.D`` becomes ``A.***.***.D`` with the
port kept as-is (``172.58.32.1:8080`` -> ``172.***.***.1:8080``); a non-ipv4 host (hostname,
ipv6) passes through unchanged with only its credentials stripped. raises ValueError on a
spec with no parseable host.
"""
rest = spec
if "://" in rest:
rest = rest.split("://", 1)[1]
if "@" in rest:
rest = rest.rsplit("@", 1)[1]
return _bullet_trailing_fields(rest)
host, sep, port = rest.partition(":")
if not host:
raise ValueError(f"mask_proxy: no host in proxy spec {spec!r}")
if _is_ipv4(host):
octets = host.split(".")
host = f"{octets[0]}.***.***.{octets[3]}"
return f"{host}:{port}" if sep else host
def stable_id(*parts: str, length: int = 16) -> str:
"""deterministic id from ordered parts; same inputs always produce the same id
joins the parts with a ``\\x00`` separator (so ``("ab","c")`` != ``("a","bc")``) and
returns the leading ``length`` hex chars of their sha256. raises ValueError on an empty
part, a non-str part, or a non-positive length - fail loud rather than emit a weak id.
"""
if length <= 0:
raise ValueError(f"stable_id: length must be positive, got {length}")
for part in parts:
if not isinstance(part, str):
raise ValueError(f"stable_id: parts must be str, got {type(part).__name__}")
if part == "":
raise ValueError("stable_id: parts must be non-empty")
return hashlib.sha256("\x00".join(parts).encode()).hexdigest()[:length]