WARDEN-WP-0026 T03: masking display filter (defense-in-depth)
- warden/mask.py: fingerprint()/mask_value() — presence, length, 8-char sha256 prefix; never the value. - proxy.proxy_fetch_fingerprint + `warden access --fingerprint`: masked status view (presence/length/hash) that emits no value, so it bypasses the T02 stdout guard. Lets two parties compare sha256 prefixes to confirm a shared value without seeing it (e.g. rotation landed). - documented as defense-in-depth (raw bao bypasses it) in OperatorAccessAssist.md and the module docstring. - tests: tests/test_mask.py + CLI fingerprint test. 299 pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
04c8b2ab1d
commit
fc0f18aa5c
7 changed files with 150 additions and 3 deletions
|
|
@ -960,6 +960,7 @@ def _access_proxy(
|
||||||
wrap: bool = False,
|
wrap: bool = False,
|
||||||
wrap_ttl: str = "5m",
|
wrap_ttl: str = "5m",
|
||||||
unsafe_stdout: bool = False,
|
unsafe_stdout: bool = False,
|
||||||
|
fingerprint: bool = False,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Proxy a non-SSH credential fetch as the caller (WP-0014 T3).
|
"""Proxy a non-SSH credential fetch as the caller (WP-0014 T3).
|
||||||
|
|
||||||
|
|
@ -973,6 +974,7 @@ def _access_proxy(
|
||||||
caller_auth_present,
|
caller_auth_present,
|
||||||
proxy_exec,
|
proxy_exec,
|
||||||
proxy_fetch,
|
proxy_fetch,
|
||||||
|
proxy_fetch_fingerprint,
|
||||||
proxy_fetch_to_file,
|
proxy_fetch_to_file,
|
||||||
proxy_fetch_wrapped,
|
proxy_fetch_wrapped,
|
||||||
resolve_fetch_command,
|
resolve_fetch_command,
|
||||||
|
|
@ -1061,7 +1063,7 @@ def _access_proxy(
|
||||||
# secret value on stdout. Streaming a value to stdout is the documented anti-pattern:
|
# secret value on stdout. Streaming a value to stdout is the documented anti-pattern:
|
||||||
# allowed only to an interactive terminal, and only with an explicit acknowledgment
|
# allowed only to an interactive terminal, and only with an explicit acknowledgment
|
||||||
# when stdout is captured/piped (the logged-context disclosure risk).
|
# when stdout is captured/piped (the logged-context disclosure risk).
|
||||||
if not is_login and not do_exec and not wrap and not out_path:
|
if not is_login and not do_exec and not wrap and not out_path and not fingerprint:
|
||||||
import sys as _sys
|
import sys as _sys
|
||||||
|
|
||||||
if not _sys.stdout.isatty() and not unsafe_stdout:
|
if not _sys.stdout.isatty() and not unsafe_stdout:
|
||||||
|
|
@ -1098,6 +1100,15 @@ def _access_proxy(
|
||||||
elif out_path:
|
elif out_path:
|
||||||
rc = proxy_fetch_to_file(resolved, Path(out_path))
|
rc = proxy_fetch_to_file(resolved, Path(out_path))
|
||||||
err.print(f"[dim]value written to {out_path} (mode 0600); not shown[/dim]")
|
err.print(f"[dim]value written to {out_path} (mode 0600); not shown[/dim]")
|
||||||
|
elif fingerprint:
|
||||||
|
fp = proxy_fetch_fingerprint(resolved)
|
||||||
|
# Masked fingerprint only — presence, length, short hash; never the value.
|
||||||
|
print(fp.render())
|
||||||
|
err.print(
|
||||||
|
"[dim]masked fingerprint (defense-in-depth; not the value). Compare "
|
||||||
|
"sha256 prefixes to confirm two parties hold the same secret.[/dim]"
|
||||||
|
)
|
||||||
|
rc = 0
|
||||||
else:
|
else:
|
||||||
rc = proxy_fetch(resolved)
|
rc = proxy_fetch(resolved)
|
||||||
except ProxyError as e:
|
except ProxyError as e:
|
||||||
|
|
@ -1160,6 +1171,10 @@ def access(
|
||||||
bool,
|
bool,
|
||||||
typer.Option("--unsafe-stdout", help="Acknowledge streaming a value to a captured/piped stdout (anti-pattern)"),
|
typer.Option("--unsafe-stdout", help="Acknowledge streaming a value to a captured/piped stdout (anti-pattern)"),
|
||||||
] = False,
|
] = False,
|
||||||
|
fingerprint: Annotated[
|
||||||
|
bool,
|
||||||
|
typer.Option("--fingerprint", help="Show a masked fingerprint (presence, length, short hash) — never the value"),
|
||||||
|
] = False,
|
||||||
no_policy: Annotated[
|
no_policy: Annotated[
|
||||||
bool,
|
bool,
|
||||||
typer.Option("--no-policy", help="Acknowledge proxying when the flex-auth gate is not enforced"),
|
typer.Option("--no-policy", help="Acknowledge proxying when the flex-auth gate is not enforced"),
|
||||||
|
|
@ -1195,7 +1210,7 @@ def access(
|
||||||
|
|
||||||
entry = matches[0]
|
entry = matches[0]
|
||||||
|
|
||||||
if do_fetch or do_exec or out_path or wrap:
|
if do_fetch or do_exec or out_path or wrap or fingerprint:
|
||||||
_access_proxy(
|
_access_proxy(
|
||||||
entry,
|
entry,
|
||||||
domain=domain,
|
domain=domain,
|
||||||
|
|
@ -1208,6 +1223,7 @@ def access(
|
||||||
wrap=wrap,
|
wrap=wrap,
|
||||||
wrap_ttl=wrap_ttl,
|
wrap_ttl=wrap_ttl,
|
||||||
unsafe_stdout=unsafe_stdout,
|
unsafe_stdout=unsafe_stdout,
|
||||||
|
fingerprint=fingerprint,
|
||||||
)
|
)
|
||||||
return
|
return
|
||||||
|
|
||||||
|
|
|
||||||
46
src/warden/mask.py
Normal file
46
src/warden/mask.py
Normal file
|
|
@ -0,0 +1,46 @@
|
||||||
|
"""Masking display filter for KV values (WARDEN-WP-0026 T03).
|
||||||
|
|
||||||
|
Defense-in-depth, **not a boundary**: any place warden would otherwise render a
|
||||||
|
secret value for a human (a status/listing view) shows a *fingerprint* instead —
|
||||||
|
presence, length, and a short non-reversible hash. Two operators can compare
|
||||||
|
fingerprints to confirm they hold the same value (e.g. that a rotation landed the
|
||||||
|
expected token) without either seeing it, and a fingerprint in a transcript
|
||||||
|
discloses nothing.
|
||||||
|
|
||||||
|
Limitation (documented, by design): raw `bao kv get <path>` bypasses this entirely
|
||||||
|
— warden only masks *warden-mediated* output. The real boundary is OpenBao policy
|
||||||
|
plus the T01 capabilities-safe verify and T02 no-stdout transports.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import hashlib
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
# Short, non-reversible hash: first 8 hex chars of SHA-256. Not a value, and a
|
||||||
|
# collision is irrelevant for the "same/different?" comparison this supports.
|
||||||
|
_HASH_PREFIX_LEN = 8
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Fingerprint:
|
||||||
|
present: bool
|
||||||
|
length: int
|
||||||
|
sha256_prefix: str # "" when the value is empty/absent
|
||||||
|
|
||||||
|
def render(self) -> str:
|
||||||
|
if not self.present:
|
||||||
|
return "‹absent›"
|
||||||
|
return f"‹hidden len={self.length} sha256:{self.sha256_prefix}›"
|
||||||
|
|
||||||
|
|
||||||
|
def fingerprint(value: str | None) -> Fingerprint:
|
||||||
|
"""Compute a non-reversible fingerprint of a value. Never returns the value."""
|
||||||
|
if not value:
|
||||||
|
return Fingerprint(present=False, length=0, sha256_prefix="")
|
||||||
|
digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:_HASH_PREFIX_LEN]
|
||||||
|
return Fingerprint(present=True, length=len(value), sha256_prefix=digest)
|
||||||
|
|
||||||
|
|
||||||
|
def mask_value(value: str | None) -> str:
|
||||||
|
"""Render a value as its masked fingerprint string. Never emits the value."""
|
||||||
|
return fingerprint(value).render()
|
||||||
|
|
@ -301,6 +301,22 @@ def proxy_fetch_wrapped(resolved: ResolvedFetch) -> str:
|
||||||
return str(token)
|
return str(token)
|
||||||
|
|
||||||
|
|
||||||
|
def proxy_fetch_fingerprint(resolved: ResolvedFetch):
|
||||||
|
"""Fetch the value and return a masked fingerprint — never the value (T03).
|
||||||
|
|
||||||
|
Defense-in-depth status view: lets an operator confirm presence/length and
|
||||||
|
compare a short non-reversible hash without disclosing the secret. The value
|
||||||
|
transits warden's memory only to be hashed, and is scrubbed immediately.
|
||||||
|
"""
|
||||||
|
from warden.mask import fingerprint
|
||||||
|
|
||||||
|
value = _capture_value(resolved)
|
||||||
|
try:
|
||||||
|
return fingerprint(value)
|
||||||
|
finally:
|
||||||
|
value = "" # noqa: F841 — best-effort scrub
|
||||||
|
|
||||||
|
|
||||||
def proxy_exec(resolved: ResolvedFetch, *, env_var: str, child_argv: List[str]) -> int:
|
def proxy_exec(resolved: ResolvedFetch, *, env_var: str, child_argv: List[str]) -> int:
|
||||||
"""Fetch the value and inject it into a child command's environment only.
|
"""Fetch the value and inject it into a child command's environment only.
|
||||||
|
|
||||||
|
|
|
||||||
33
tests/test_mask.py
Normal file
33
tests/test_mask.py
Normal file
|
|
@ -0,0 +1,33 @@
|
||||||
|
"""Tests for the masking display filter (WARDEN-WP-0026 T03)."""
|
||||||
|
from warden.mask import fingerprint, mask_value
|
||||||
|
from warden.proxy import ResolvedFetch, proxy_fetch_fingerprint
|
||||||
|
|
||||||
|
|
||||||
|
def test_mask_never_contains_the_value():
|
||||||
|
secret = "ghp_realtokenvalue1234567890abcdef"
|
||||||
|
masked = mask_value(secret)
|
||||||
|
assert secret not in masked
|
||||||
|
assert "hidden" in masked and "len=" in masked and "sha256:" in masked
|
||||||
|
|
||||||
|
|
||||||
|
def test_fingerprint_reports_presence_and_length():
|
||||||
|
fp = fingerprint("abcd")
|
||||||
|
assert fp.present is True and fp.length == 4
|
||||||
|
assert len(fp.sha256_prefix) == 8
|
||||||
|
|
||||||
|
|
||||||
|
def test_absent_value_renders_absent():
|
||||||
|
assert mask_value("") == "‹absent›"
|
||||||
|
assert mask_value(None) == "‹absent›"
|
||||||
|
assert fingerprint(None).present is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_fingerprint_is_stable_and_discriminating():
|
||||||
|
assert fingerprint("token-A").sha256_prefix == fingerprint("token-A").sha256_prefix
|
||||||
|
assert fingerprint("token-A").sha256_prefix != fingerprint("token-B").sha256_prefix
|
||||||
|
|
||||||
|
|
||||||
|
def test_proxy_fingerprint_returns_mask_not_value():
|
||||||
|
fp = proxy_fetch_fingerprint(ResolvedFetch(shell_cmd="printf 'the-secret-value'"))
|
||||||
|
assert fp.present and fp.length == len("the-secret-value")
|
||||||
|
assert "the-secret-value" not in fp.render()
|
||||||
|
|
@ -364,3 +364,22 @@ def test_access_fetch_to_nonterminal_stdout_is_refused(tmp_path, monkeypatch):
|
||||||
r = runner.invoke(app, ["access", "whynot-design-npm-publish", "--fetch", "--no-policy"])
|
r = runner.invoke(app, ["access", "whynot-design-npm-publish", "--fetch", "--no-policy"])
|
||||||
assert r.exit_code == 6
|
assert r.exit_code == 6
|
||||||
assert "sanctioned transport" in r.output.lower() or "refusing" in r.output.lower()
|
assert "sanctioned transport" in r.output.lower() or "refusing" in r.output.lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_access_fingerprint_masks_and_bypasses_stdout_guard(monkeypatch, tmp_path):
|
||||||
|
"""--fingerprint prints a masked fingerprint (never the value) even to captured stdout."""
|
||||||
|
_proxy_env(monkeypatch, tmp_path)
|
||||||
|
monkeypatch.setenv("VAULT_TOKEN", "caller-token")
|
||||||
|
|
||||||
|
class _Fake:
|
||||||
|
returncode = 0
|
||||||
|
stdout = "top-secret-token-value"
|
||||||
|
|
||||||
|
monkeypatch.setattr("warden.proxy.subprocess.run", lambda *a, **k: _Fake())
|
||||||
|
r = runner.invoke(
|
||||||
|
app,
|
||||||
|
["access", "whynot-design-npm-publish", "--fingerprint", "--no-policy"],
|
||||||
|
)
|
||||||
|
assert r.exit_code == 0
|
||||||
|
assert "top-secret-token-value" not in r.output # value never shown
|
||||||
|
assert "hidden" in r.output and "sha256:" in r.output
|
||||||
|
|
|
||||||
|
|
@ -31,12 +31,20 @@ $ warden access "npm token" --field NPM_AUTH_TOKEN --exec -- npm publish
|
||||||
$ warden access "npm token" --path <p> --wrap # then: bao unwrap <token>
|
$ warden access "npm token" --path <p> --wrap # then: bao unwrap <token>
|
||||||
# interactive login (login lane): no token required, no secret-read gate
|
# interactive login (login lane): no token required, no secret-read gate
|
||||||
$ warden access "login oidc" --domain coulomb_social --fetch
|
$ warden access "login oidc" --domain coulomb_social --fetch
|
||||||
|
# masked status: presence, length, short hash — never the value (WP-0026 T03)
|
||||||
|
$ warden access "npm token" --path <p> --fingerprint # ‹hidden len=40 sha256:1a2b3c4d›
|
||||||
```
|
```
|
||||||
|
|
||||||
> **Raw `--fetch` to stdout is the anti-pattern.** It is refused when stdout is
|
> **Raw `--fetch` to stdout is the anti-pattern.** It is refused when stdout is
|
||||||
> captured or piped (a logged-context disclosure risk); pass `--unsafe-stdout` only
|
> captured or piped (a logged-context disclosure risk); pass `--unsafe-stdout` only
|
||||||
> for an interactive human terminal. Prefer `--out` / `--exec` / `--wrap`.
|
> for an interactive human terminal. Prefer `--out` / `--exec` / `--wrap`.
|
||||||
|
|
||||||
|
> **`--fingerprint` is defense-in-depth, not a boundary.** It masks warden-mediated
|
||||||
|
> output so two parties can compare `sha256` prefixes to confirm they hold the same
|
||||||
|
> value (e.g. that a rotation landed) without either seeing it. It only masks
|
||||||
|
> *warden's* output — raw `bao kv get <path>` bypasses it entirely. The real boundary
|
||||||
|
> is OpenBao policy plus capabilities-safe verify (T01) and the no-stdout transports.
|
||||||
|
|
||||||
`--json` gives a stable, secret-free shape for agentic operators.
|
`--json` gives a stable, secret-free shape for agentic operators.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
|
||||||
|
|
@ -115,11 +115,20 @@ the anti-pattern is documented fleet-wide.
|
||||||
|
|
||||||
```task
|
```task
|
||||||
id: WARDEN-WP-0026-T03
|
id: WARDEN-WP-0026-T03
|
||||||
status: todo
|
status: done
|
||||||
priority: medium
|
priority: medium
|
||||||
state_hub_task_id: "21ab08d5-7782-4567-a08d-980211dd7851"
|
state_hub_task_id: "21ab08d5-7782-4567-a08d-980211dd7851"
|
||||||
```
|
```
|
||||||
|
|
||||||
|
Done 2026-07-16: `warden/mask.py` (`fingerprint`/`mask_value` — presence, length,
|
||||||
|
8-char sha256 prefix; never the value) + `proxy_fetch_fingerprint` and a
|
||||||
|
`warden access … --fingerprint` masked status view (bypasses the stdout guard
|
||||||
|
because it emits no value). Lets two parties compare sha256 prefixes to confirm a
|
||||||
|
shared value (e.g. rotation landed) without disclosure. Explicitly labelled
|
||||||
|
defense-in-depth — raw `bao kv get` bypasses it — in `wiki/OperatorAccessAssist.md`
|
||||||
|
and the module docstring. Tests in `tests/test_mask.py` + a CLI test in
|
||||||
|
`tests/test_proxy.py`.
|
||||||
|
|
||||||
In the warden wrapper, mask KV data values by default when any listing/status is
|
In the warden wrapper, mask KV data values by default when any listing/status is
|
||||||
shown — display presence, length, and a short non-reversible hash instead of the
|
shown — display presence, length, and a short non-reversible hash instead of the
|
||||||
value. Explicitly labelled as defense-in-depth (raw bao bypasses it).
|
value. Explicitly labelled as defense-in-depth (raw bao bypasses it).
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue