fix: provider() unicode-digit crash + phantom() short-value false-mask (v0.3.2)

provider() used str.isdigit(), which accepts unicode digit lookalikes (e.g. superscript
²) that int() can't parse, crashing on digit-only-looking input; _digits() now filters
on isdecimal()+isascii() so only real ASCII digits pass through. phantom() revealed the
entire value for len <= 10 while still looking masked (first6+last4 fully covers or
double-covers short strings); those now fully mask instead. Also compresses several
essay-length docstrings (retry, masking, paths) to a line plus the real nuance, with
zero behavior change, and bumps the README install pin.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
2026-07-02 23:21:51 -04:00
parent 19e6f4aa06
commit 4f835ff003
6 changed files with 27 additions and 34 deletions
+5 -3
View File
@@ -12,15 +12,15 @@ Small sync helpers shared across projects. Base is stdlib only — **no dependen
## Install ## Install
``` ```
commons @ git+ssh://git@git.rethinkstudios.io/rethink-public/commons.git@v0.3.1 commons @ git+ssh://git@git.rethinkstudios.io/rethink-public/commons.git@v0.3.2
# async address/geo lookups (fetch_ip / ip_location / fetch_location) need the extra: # async address/geo lookups (fetch_ip / ip_location / fetch_location) need the extra:
commons[addr] @ git+ssh://git@git.rethinkstudios.io/rethink-public/commons.git@v0.3.1 commons[addr] @ git+ssh://git@git.rethinkstudios.io/rethink-public/commons.git@v0.3.2
``` ```
The base install pulls **nothing** (stdlib). Only `commons[addr]` adds `aiohttp`, and The base install pulls **nothing** (stdlib). Only `commons[addr]` adds `aiohttp`, and
only for the geo lookups — the pure `commons.addr.ip` utilities ship in base. only for the geo lookups — the pure `commons.addr.ip` utilities ship in base.
Drop the `@v0.3.1` suffix from the line above to install the latest unpinned. Drop the `@v0.3.2` suffix from the line above to install the latest unpinned.
## timing ## timing
@@ -121,7 +121,9 @@ from commons import credit, cvv, phantom, provider, mask_url, mask_proxy
credit("4111 1111 1111 1234") # "•••• •••• •••• 1234" credit("4111 1111 1111 1234") # "•••• •••• •••• 1234"
cvv("123") # "•••" cvv("123") # "•••"
phantom("abcdef1234567890") # "abcdef...7890" phantom("abcdef1234567890") # "abcdef...7890"
phantom("1234567890") # "••••••••••" (len <= 10 fully masks, never a false reveal)
provider("4111111111111111") # "VISA" (VISA/MC/AMEX/UPAY/DISC/JCB/DNRS/UNKW) 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 # redact credentials in a connection string before logging it
mask_url("https://u:pw@api.x/v2?apiKey=SECRET&ip=8.8.8.8") mask_url("https://u:pw@api.x/v2?apiKey=SECRET&ip=8.8.8.8")
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "commons" name = "commons"
version = "0.3.1" version = "0.3.2"
description = "small stdlib-based sync helpers: time/timezone deltas, dotted-path dict access, display masking, ip/address tooling, and retry/backoff" description = "small stdlib-based sync helpers: time/timezone deltas, dotted-path dict access, display masking, ip/address tooling, and retry/backoff"
requires-python = ">=3.10" requires-python = ">=3.10"
dependencies = [] dependencies = []
+1 -1
View File
@@ -68,4 +68,4 @@ __all__ = [
"aretry", "aretry",
] ]
__version__ = "0.3.1" __version__ = "0.3.2"
+14 -13
View File
@@ -17,8 +17,8 @@ SENSITIVE_QUERY_KEYS = frozenset(
def _digits(value: str) -> str: def _digits(value: str) -> str:
"""keep only the digit characters of a string""" """keep only ascii-decimal characters of a string (excludes unicode digit lookalikes)"""
return "".join(c for c in value if c.isdigit()) return "".join(c for c in value if c.isdecimal() and c.isascii())
def credit(card_number: str) -> str: def credit(card_number: str) -> str:
@@ -33,7 +33,13 @@ def cvv(value: str) -> str:
def phantom(value: str) -> str: def phantom(value: str) -> str:
"""show a long token as first-six...last-four (e.g. a hash or id)""" """show a long token as first-six...last-four (e.g. a hash or id)
values of length <= 10 fully mask instead — first6+last4 would otherwise
reveal (or double-reveal) every character while still looking masked.
"""
if len(value) <= 10:
return "" * len(value)
return f"{value[:6]}...{value[-4:]}" return f"{value[:6]}...{value[-4:]}"
@@ -64,13 +70,10 @@ def provider(card_number: str) -> str:
def mask_url(url: str, *, keys: "frozenset[str] | None" = None) -> str: def mask_url(url: str, *, keys: "frozenset[str] | None" = None) -> str:
"""redact credentials in a url for logging """redact credentials in a url for logging: drop userinfo, ``***`` sensitive query values
strips userinfo (``user:pass@host`` -> ``host``) and replaces the values of param-name matching is case-insensitive; ``keys`` overrides the default sensitive-name
sensitive query parameters (apiKey, token, password, secret, ...) with set. non-url or unparseable input is returned unchanged.
``***``, keeping the parameter name so the shape stays readable. matching is
case-insensitive on the parameter name; pass ``keys`` to override the default
sensitive-name set. non-url or unparseable input is returned unchanged.
""" """
sensitive = SENSITIVE_QUERY_KEYS if keys is None else frozenset(k.lower() for k in keys) sensitive = SENSITIVE_QUERY_KEYS if keys is None else frozenset(k.lower() for k in keys)
try: try:
@@ -93,11 +96,9 @@ def mask_url(url: str, *, keys: "frozenset[str] | None" = None) -> str:
def mask_proxy(spec: str) -> str: def mask_proxy(spec: str) -> str:
"""redact the password of a ``host:port:user:password`` proxy string """redact the password of a ``host:port:user:password`` proxy string to ``:****``
renders ``host:port`` when there is no auth, or ``host:port:user:****`` when a plain ``host:port`` (no auth) and any other non-conforming input pass through unchanged.
a user/password is present — never the password itself. leaves a plain
``host:port`` untouched and returns non-conforming input unchanged.
""" """
parts = spec.split(":") parts = spec.split(":")
if len(parts) == 2: if len(parts) == 2:
+4 -8
View File
@@ -36,15 +36,11 @@ def deep_get(data: Any, path: str, default: Any = None, *, sep: str = ".") -> An
def deep_set(data: dict, path: str, value: Any, *, sep: str = ".") -> dict: def deep_set(data: dict, path: str, value: Any, *, sep: str = ".") -> dict:
"""set a nested value by dotted path, creating intermediate dicts """set a nested value by dotted path, creating intermediate dicts; returns data for chaining
returns the same top-level dict for chaining. mirrors deep_get: a numeric segment mirrors deep_get's list indexing (a numeric segment updates an existing list/tuple
indexes into an existing list/tuple instead of being treated as a dict key, so a element rather than corrupting it); unlike deep_get, an out-of-range index raises
read-modify-write round-trip using the same path never corrupts a list. overwrites IndexError instead of silently mis-storing, since there's no safe default to fall back to.
any non-dict/non-list value sitting where an intermediate container is needed. an
out-of-range list index raises IndexError rather than silently mis-storing (deep_get
returns default on out-of-range; deep_set has no safe default to fall back to, so it
fails loud instead).
""" """
segments = path.split(sep) segments = path.split(sep)
cur = data cur = data
+2 -8
View File
@@ -72,11 +72,8 @@ def retry(
sleep: Callable[[float], None] = time.sleep, sleep: Callable[[float], None] = time.sleep,
rand: Callable[[], float] = random.random, rand: Callable[[], float] = random.random,
): ):
"""retry a sync callable with exponential backoff; call form or decorator """retry a sync callable with exponential backoff; call form (`retry(fn, ...)`) or decorator
`retry(fn, ...)` runs immediately; `@retry(...)` wraps a function. retries on the
`on` exceptions, stops early if `give_up(exc)` is true, re-raises the last
exception once `attempts` are exhausted. `sleep`/`rand` are injectable for tests.
`attempts` is floored at 1 so the callable always runs at least once. `attempts` is floored at 1 so the callable always runs at least once.
""" """
types = _as_types(on) types = _as_types(on)
@@ -125,11 +122,8 @@ def aretry(
sleep: Callable[[float], "asyncio.Future"] = asyncio.sleep, sleep: Callable[[float], "asyncio.Future"] = asyncio.sleep,
rand: Callable[[], float] = random.random, rand: Callable[[], float] = random.random,
): ):
"""retry an async callable with exponential backoff; call form or decorator """async twin of `retry`; call form (`await aretry(coro_fn, ...)`) or decorator, same semantics
async twin of `retry`. `await aretry(coro_fn, ...)` runs immediately;
`@aretry(...)` wraps a coroutine function. same semantics: retry on `on`, stop on
`give_up`, re-raise the last exception after `attempts`. `sleep`/`rand` injectable.
`attempts` is floored at 1 so the callable always runs at least once. `attempts` is floored at 1 so the callable always runs at least once.
""" """
types = _as_types(on) types = _as_types(on)