Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a73f2e3883 | |||
| 7ff4f6e5c0 | |||
| 7517e16b15 | |||
| 9a0433225e | |||
| fa82d30087 | |||
| 05b00a40c0 | |||
| 3068e74e5c | |||
| da20eb2fdb |
@@ -10,6 +10,7 @@ Run: python mcp_manager.py
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
@@ -141,6 +142,10 @@ QScrollBar::handle:vertical {{ background: {p.border}; border-radius: 5px; min-h
|
||||
QScrollBar::add-line, QScrollBar::sub-line {{ height: 0; }}
|
||||
QLabel#statusbar {{ color: {p.muted}; padding: 4px 2px; }}
|
||||
QLabel#warnBanner {{ color: {p.on_accent}; background: {p.warn}; border-radius: 8px; padding: 8px 10px; font-weight: 600; }}
|
||||
QFrame#noticeBanner {{ background: {p.panel_2}; border: 1px solid {p.accent}; border-radius: 8px; }}
|
||||
QLabel#noticeText {{ color: {p.text}; }}
|
||||
QPushButton#noticeClose {{ background: transparent; border: none; color: {p.muted}; font-size: 14px; padding: 2px; }}
|
||||
QPushButton#noticeClose:hover {{ color: {p.text}; }}
|
||||
QLabel#section {{ color: {p.muted}; font-weight: 600; font-size: 12px; padding: 2px 2px; }}
|
||||
QLabel#sectionDisabled {{ color: {p.muted}; font-weight: 600; font-size: 12px; padding: 2px 2px; }}
|
||||
QLabel#placeholder {{ color: {p.muted}; padding: 12px; background: {p.panel_2}; border: 1px dashed {p.border}; border-radius: 8px; }}
|
||||
@@ -263,7 +268,7 @@ class _SecretMaskDelegate(QStyledItemDelegate):
|
||||
if self.revealed or not option.text:
|
||||
return
|
||||
key_item = self._table.item(index.row(), 0)
|
||||
if key_item and core.is_secret_key(key_item.text()):
|
||||
if key_item and core.should_mask_value(key_item.text(), option.text):
|
||||
option.text = core.MASK
|
||||
|
||||
|
||||
@@ -1528,6 +1533,54 @@ class AboutDialog(QDialog):
|
||||
QDesktopServices.openUrl(QUrl(self._release_url or core.RELEASES_URL))
|
||||
|
||||
|
||||
class NoticeBanner(QFrame):
|
||||
"""A persistent, dismissible notice with an optional action button.
|
||||
|
||||
The status bar is the wrong home for anything the user needs to act on --
|
||||
21 call sites rewrite it, so a message posted there is gone by the next
|
||||
click. That wiped the MSIX warning (#35) and then the update notice (#78).
|
||||
This is the shared mechanism so it doesn't happen a third time.
|
||||
"""
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setObjectName("noticeBanner")
|
||||
row = QHBoxLayout(self)
|
||||
row.setContentsMargins(10, 8, 8, 8)
|
||||
row.setSpacing(8)
|
||||
self._label = QLabel("")
|
||||
self._label.setObjectName("noticeText")
|
||||
self._label.setWordWrap(True)
|
||||
row.addWidget(self._label, 1)
|
||||
self._action_btn = QPushButton("")
|
||||
self._action_btn.setCursor(Qt.CursorShape.PointingHandCursor)
|
||||
self._action_btn.hide()
|
||||
row.addWidget(self._action_btn)
|
||||
self._close_btn = QPushButton("\u2715")
|
||||
self._close_btn.setObjectName("noticeClose")
|
||||
self._close_btn.setCursor(Qt.CursorShape.PointingHandCursor)
|
||||
self._close_btn.setFixedWidth(26)
|
||||
self._close_btn.setToolTip("Dismiss")
|
||||
self._close_btn.clicked.connect(self.hide)
|
||||
row.addWidget(self._close_btn)
|
||||
self.hide()
|
||||
|
||||
def show_notice(self, text: str, action_label: str = "", on_action=None):
|
||||
self._label.setText(text)
|
||||
self._label.setToolTip(text)
|
||||
# Reconnect cleanly: a banner reused for a second notice would
|
||||
# otherwise fire the previous notice's action too.
|
||||
with contextlib.suppress(RuntimeError, TypeError):
|
||||
self._action_btn.clicked.disconnect()
|
||||
if action_label and on_action is not None:
|
||||
self._action_btn.setText(action_label)
|
||||
self._action_btn.clicked.connect(lambda _=False: on_action())
|
||||
self._action_btn.show()
|
||||
else:
|
||||
self._action_btn.hide()
|
||||
self.show()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Restart worker: core.restart_claude_desktop() blocks up to ~5 s on macOS
|
||||
# waiting for the old instance to exit, so it must run off the UI thread.
|
||||
@@ -1585,6 +1638,11 @@ class MainWindow(QMainWindow):
|
||||
self.warn_banner.hide()
|
||||
root.addWidget(self.warn_banner)
|
||||
|
||||
# Update availability gets its own persistent banner rather than a
|
||||
# status-line write, which the next UI action overwrites (#78).
|
||||
self.update_banner = NoticeBanner(self)
|
||||
root.addWidget(self.update_banner)
|
||||
|
||||
# User-draggable divider between the server list and the editor.
|
||||
split = QSplitter(Qt.Orientation.Horizontal)
|
||||
split.setChildrenCollapsible(False)
|
||||
@@ -1637,10 +1695,59 @@ class MainWindow(QMainWindow):
|
||||
theme_menu.addAction(act)
|
||||
|
||||
help_menu = self.menuBar().addMenu("&Help")
|
||||
|
||||
# "Check for updates" used to exist only as a button inside the About
|
||||
# dialog, which is not somewhere anyone looks for it (#79).
|
||||
update_action = QAction("Check for updates…", self)
|
||||
# Explicit role: macOS relocates actions it recognises by text, and
|
||||
# some Qt versions treat "update" as application-menu material. Pin it
|
||||
# so the item stays where the menu says it is on every platform.
|
||||
update_action.setMenuRole(QAction.MenuRole.ApplicationSpecificRole)
|
||||
update_action.triggered.connect(self.check_for_updates)
|
||||
help_menu.addAction(update_action)
|
||||
help_menu.addSeparator()
|
||||
|
||||
about_action = QAction("About Better Claude Config…", self)
|
||||
# Qt auto-assigns AboutRole to actions whose text starts with "About",
|
||||
# which moves this into the application menu on macOS. That is the
|
||||
# right home there -- state it explicitly rather than inheriting it by
|
||||
# accident, since the behaviour is invisible from this call site.
|
||||
about_action.setMenuRole(QAction.MenuRole.AboutRole)
|
||||
about_action.triggered.connect(self._show_about)
|
||||
help_menu.addAction(about_action)
|
||||
|
||||
def _show_update_notice(self, notice: dict):
|
||||
"""Surface an available update where it survives the next click."""
|
||||
url = notice["url"]
|
||||
self.update_banner.show_notice(
|
||||
notice["text"],
|
||||
action_label="Open releases page",
|
||||
on_action=lambda: QDesktopServices.openUrl(QUrl(url)),
|
||||
)
|
||||
|
||||
def check_for_updates(self):
|
||||
"""Menu-driven check. Unlike the startup check this is never throttled
|
||||
and always reports back -- the user asked, so silence would read as a
|
||||
broken button."""
|
||||
self.status.setText("Checking for updates…")
|
||||
self._menu_update_worker = UpdateCheckWorker()
|
||||
self._menu_update_worker.done.connect(self._on_menu_update_checked)
|
||||
self._menu_update_worker.start()
|
||||
|
||||
def _on_menu_update_checked(self, release: dict | None):
|
||||
self._menu_update_worker = None
|
||||
if release is None:
|
||||
self.status.setText("Couldn't check for updates (offline?).")
|
||||
return
|
||||
QSettings("BCC", "BetterClaudeConfig").setValue("update/lastCheck", time.time())
|
||||
notice = core.update_notice(core.__version__, release)
|
||||
if notice:
|
||||
self._show_update_notice(notice)
|
||||
self.status.setText(f"Update available: {notice['version']}")
|
||||
else:
|
||||
self.update_banner.hide()
|
||||
self.status.setText(f"You're up to date ({core.__version__}).")
|
||||
|
||||
def _set_theme(self, setting: str):
|
||||
"""Persist the theme choice and repaint the running window."""
|
||||
QSettings("BCC", "BetterClaudeConfig").setValue("ui/theme", setting)
|
||||
@@ -1681,10 +1788,9 @@ class MainWindow(QMainWindow):
|
||||
if release is None:
|
||||
return # offline/failed check: don't advance lastCheck, allow retry
|
||||
QSettings("BCC", "BetterClaudeConfig").setValue("update/lastCheck", time.time())
|
||||
if core.is_newer_version(core.__version__, release["version"]):
|
||||
self.status.setText(
|
||||
f"Update available: {release['version']} · Help ▸ About to view it."
|
||||
)
|
||||
notice = core.update_notice(core.__version__, release)
|
||||
if notice:
|
||||
self._show_update_notice(notice)
|
||||
|
||||
# --- layout persistence ---------------------------------------------- #
|
||||
def _restore_layout(self):
|
||||
@@ -2010,9 +2116,21 @@ class MainWindow(QMainWindow):
|
||||
return
|
||||
self.full_config = cfg
|
||||
repaired = True
|
||||
# extract_servers tolerates malformed entries rather than raising (#72),
|
||||
# but keep it inside the guard: a load failure must leave the previously
|
||||
# loaded profile intact instead of half-swapping the window's state.
|
||||
try:
|
||||
servers = core.extract_servers(self.full_config)
|
||||
except Exception as exc: # pragma: no cover - defence in depth
|
||||
QMessageBox.critical(
|
||||
self,
|
||||
"Could not read config",
|
||||
f"{profile.path}\n\nThe server list couldn't be read: {exc}",
|
||||
)
|
||||
return
|
||||
self._loaded_stat = core.config_fingerprint(profile.path)
|
||||
self.current_profile = profile
|
||||
self.servers = core.extract_servers(self.full_config)
|
||||
self.servers = servers
|
||||
self.dirty = False
|
||||
self.restart_btn.hide()
|
||||
self._undo_stack.clear()
|
||||
@@ -2335,7 +2453,7 @@ class MainWindow(QMainWindow):
|
||||
entry = self.servers[idx]
|
||||
old_name = entry.name
|
||||
entry.name = self.editor.current_name()
|
||||
entry.data = self.editor.dump_data()
|
||||
entry.set_data(self.editor.dump_data())
|
||||
# The server stays in its section (enable state unchanged), so update
|
||||
# its existing row in place rather than re-rendering.
|
||||
# An edit invalidates any cached "Test all" result -- the server that
|
||||
@@ -2429,7 +2547,7 @@ class MainWindow(QMainWindow):
|
||||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||||
)
|
||||
if ans == QMessageBox.StandardButton.Yes:
|
||||
self.servers[existing[name]].data = data
|
||||
self.servers[existing[name]].set_data(data)
|
||||
return False, True
|
||||
name = core.resolve_name_collision(name, {s.name for s in self.servers})
|
||||
self.servers.append(core.ServerEntry(name, data, True))
|
||||
@@ -2504,6 +2622,12 @@ class MainWindow(QMainWindow):
|
||||
self.save_btn.setEnabled(False)
|
||||
return False
|
||||
lint_warnings = core.lint_servers(self.servers)
|
||||
# ${VAR} references are only meaningful if the target client expands
|
||||
# them -- Claude Desktop doesn't, so the same config is fine in one
|
||||
# profile and broken in another (#76). Report against the loaded one.
|
||||
for entry in self.servers:
|
||||
for warning in core.env_ref_warnings(entry.data, self.current_profile):
|
||||
lint_warnings.append(f"'{entry.name}': {warning}")
|
||||
if lint_warnings:
|
||||
self.validation_lbl.setText(f"⚠ {lint_warnings[0]}")
|
||||
self.validation_lbl.setStyleSheet(f"color: {WARN};")
|
||||
@@ -2549,6 +2673,11 @@ class MainWindow(QMainWindow):
|
||||
except Exception as e:
|
||||
QMessageBox.critical(self, "Reload failed", str(e))
|
||||
return
|
||||
# The reload above is the on-disk truth for everything the user
|
||||
# didn't touch -- but it also wipes BCC-authored keys the user
|
||||
# changed in this session (named sets), which apply_servers
|
||||
# doesn't write. Carry them over before saving (#73).
|
||||
contested = core.carry_owned_keys(self.full_config, fresh)
|
||||
core.apply_servers(fresh, self.servers)
|
||||
try:
|
||||
backup = core.write_config(self.current_profile.path, fresh)
|
||||
@@ -2561,8 +2690,13 @@ class MainWindow(QMainWindow):
|
||||
self.dirty = False
|
||||
self.save_btn.setEnabled(False)
|
||||
bnote = f" · backup: {backup.name}" if backup else " · (new file)"
|
||||
cnote = (
|
||||
f" · kept your {', '.join(contested)} (the file on disk had a different copy)"
|
||||
if contested
|
||||
else ""
|
||||
)
|
||||
self.status.setText(
|
||||
f"Merged & saved {self.current_profile.path}{bnote}"
|
||||
f"Merged & saved {self.current_profile.path}{bnote}{cnote}"
|
||||
f" · Restart {self.current_profile.label} to apply."
|
||||
)
|
||||
self._offer_restart_button()
|
||||
|
||||
+327
-9
@@ -15,6 +15,7 @@ from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import contextlib
|
||||
import copy
|
||||
import difflib
|
||||
import functools
|
||||
import glob
|
||||
@@ -49,6 +50,12 @@ DISABLED_KEY = "_disabledMcpServers"
|
||||
# parks the rest under DISABLED_KEY.
|
||||
SETS_KEY = "_bccServerSets"
|
||||
|
||||
# Top-level keys BCC itself authors. They live in the client's config file, but
|
||||
# BCC is their owner, so on a stale-file merge the in-memory copy wins over the
|
||||
# on-disk one (see `carry_owned_keys`). Any future BCC-authored key belongs
|
||||
# here -- forgetting to add one is exactly how #73 happened.
|
||||
BCC_OWNED_KEYS = (SETS_KEY,)
|
||||
|
||||
BACKUP_DIRNAME = ".bcc_backups"
|
||||
MAX_BACKUPS = 15
|
||||
|
||||
@@ -163,6 +170,39 @@ def fetch_latest_release(timeout: float = 4.0) -> dict | None:
|
||||
return {"version": tag, "url": payload.get("html_url") or RELEASES_URL}
|
||||
|
||||
|
||||
def update_notice(
|
||||
current: str, release: dict | None, url_fallback: str = RELEASES_URL
|
||||
) -> dict | None:
|
||||
"""Decide whether to tell the user about a release, and what to say.
|
||||
|
||||
Returns {"version", "text", "url"} when `release` is newer than `current`,
|
||||
or None when it isn't, when the check failed, or when the payload is
|
||||
malformed. Kept here rather than in the GUI so the wording and the
|
||||
should-we-notify decision are testable -- bcc.py can't be imported by the
|
||||
test suite, which has no PySide6.
|
||||
|
||||
The text deliberately names no menu path. The old status-line notice read
|
||||
"Help > About to view it", which is wrong on macOS: Qt relocates the About
|
||||
action into the application menu (#79). A notice that carries its own
|
||||
action can't drift out of sync with the platform.
|
||||
"""
|
||||
if not isinstance(release, dict):
|
||||
return None
|
||||
version = release.get("version")
|
||||
if not version or not isinstance(version, str):
|
||||
return None
|
||||
if not is_newer_version(current, version):
|
||||
return None
|
||||
# Tags carry a "v" prefix and __version__ doesn't; render both the same way
|
||||
# so the notice doesn't read "Version v1.3.0 ... you're running 1.2.0".
|
||||
shown = version.lstrip("vV")
|
||||
return {
|
||||
"version": version,
|
||||
"text": f"Version {shown} is available. You're running {current.lstrip('vV')}.",
|
||||
"url": release.get("url") or url_fallback,
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Data model
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -178,16 +218,63 @@ class Profile:
|
||||
self.path = Path(self.path)
|
||||
|
||||
|
||||
class _NoRaw:
|
||||
"""Sentinel for ServerEntry.raw.
|
||||
|
||||
`None` can't do this job: `{"mcpServers": {"foo": null}}` is legal JSON and
|
||||
a real malformed-entry case, so None has to mean "the config said null",
|
||||
not "there was nothing here".
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def __repr__(self) -> str: # keeps ServerEntry reprs readable in test output
|
||||
return "<no raw>"
|
||||
|
||||
|
||||
NO_RAW = _NoRaw()
|
||||
|
||||
|
||||
@dataclass
|
||||
class ServerEntry:
|
||||
"""One server definition.
|
||||
|
||||
`data` is always a dict so every consumer can treat it as one. When the
|
||||
config held something that wasn't a JSON object for this server (a string,
|
||||
a number, a list -- all legal JSON, all wrong here), `data` is empty and
|
||||
the original value is preserved verbatim in `raw` so Save round-trips it
|
||||
instead of silently deleting the user's line. `lint_servers` surfaces it.
|
||||
`raw` defaults to the NO_RAW sentinel rather than None, because a config
|
||||
value of literal `null` is itself a malformed entry worth preserving.
|
||||
|
||||
Assigning `data` means the user replaced the definition through the editor,
|
||||
which retires `raw` -- use `set_data` so that can't be forgotten.
|
||||
"""
|
||||
|
||||
name: str
|
||||
data: dict
|
||||
enabled: bool = True
|
||||
raw: object = NO_RAW
|
||||
|
||||
@property
|
||||
def kind(self) -> str:
|
||||
return "remote" if "url" in self.data and "command" not in self.data else "stdio"
|
||||
|
||||
@property
|
||||
def malformed(self) -> bool:
|
||||
"""True when the config value for this server wasn't a JSON object."""
|
||||
return self.raw is not NO_RAW
|
||||
|
||||
def set_data(self, data: dict) -> None:
|
||||
"""Replace the definition from the editor, clearing any malformed original."""
|
||||
self.data = data
|
||||
self.raw = NO_RAW
|
||||
|
||||
def config_value(self):
|
||||
"""What to write back to the config: the edited dict, or the untouched
|
||||
malformed original when the user never edited it."""
|
||||
return self.data if self.raw is NO_RAW else self.raw
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Theming (issue #75)
|
||||
@@ -584,13 +671,30 @@ def repair_config_file(path: str | os.PathLike) -> tuple[dict, list[str], str]:
|
||||
return obj, notes, pretty
|
||||
|
||||
|
||||
def _server_entry(name: str, data, enabled: bool) -> ServerEntry:
|
||||
"""Build a ServerEntry, tolerating a value that isn't a JSON object.
|
||||
|
||||
A hand-edited config can legally hold `{"mcpServers": {"foo": "oops"}}` --
|
||||
valid JSON, wrong shape. Calling dict() on that raises, which used to take
|
||||
the whole load down before the linter ever got a look at it (#72). Keep the
|
||||
original instead and let the linter report it.
|
||||
"""
|
||||
if isinstance(data, dict):
|
||||
return ServerEntry(name=name, data=dict(data), enabled=enabled)
|
||||
return ServerEntry(name=name, data={}, enabled=enabled, raw=data)
|
||||
|
||||
|
||||
def extract_servers(cfg: dict) -> list[ServerEntry]:
|
||||
"""Pull enabled (`mcpServers`) and disabled (`_disabledMcpServers`) servers."""
|
||||
"""Pull enabled (`mcpServers`) and disabled (`_disabledMcpServers`) servers.
|
||||
|
||||
Never raises on a structurally-odd config -- malformed entries come back as
|
||||
empty-data entries carrying their original value (see `_server_entry`).
|
||||
"""
|
||||
out: list[ServerEntry] = []
|
||||
for name, data in (cfg.get("mcpServers") or {}).items():
|
||||
out.append(ServerEntry(name=name, data=dict(data), enabled=True))
|
||||
out.append(_server_entry(name, data, True))
|
||||
for name, data in (cfg.get(DISABLED_KEY) or {}).items():
|
||||
out.append(ServerEntry(name=name, data=dict(data), enabled=False))
|
||||
out.append(_server_entry(name, data, False))
|
||||
return out
|
||||
|
||||
|
||||
@@ -682,8 +786,8 @@ def apply_servers(cfg: dict, servers: list[ServerEntry]) -> dict:
|
||||
Write the server list back into `cfg` in place, preserving every other key
|
||||
and the position of `mcpServers`. Returns the same dict for convenience.
|
||||
"""
|
||||
enabled = {s.name: s.data for s in servers if s.enabled}
|
||||
disabled = {s.name: s.data for s in servers if not s.enabled}
|
||||
enabled = {s.name: s.config_value() for s in servers if s.enabled}
|
||||
disabled = {s.name: s.config_value() for s in servers if not s.enabled}
|
||||
|
||||
cfg["mcpServers"] = enabled # replaces value if key existed; appends otherwise
|
||||
if disabled:
|
||||
@@ -696,6 +800,34 @@ def apply_servers(cfg: dict, servers: list[ServerEntry]) -> dict:
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Write (atomic, with rotating backups)
|
||||
# --------------------------------------------------------------------------- #
|
||||
def carry_owned_keys(local_cfg: dict, fresh_cfg: dict) -> list[str]:
|
||||
"""Carry BCC-authored top-level keys from `local_cfg` onto `fresh_cfg`.
|
||||
|
||||
Used by the stale-file "Merge & save" path, which reloads the file from
|
||||
disk and re-applies the user's server edits. That reload used to drop
|
||||
anything BCC owns but `apply_servers` doesn't write -- named server sets
|
||||
vanished without a word (#73). BCC owns these keys, so the in-memory copy
|
||||
wins; mutates `fresh_cfg` in place.
|
||||
|
||||
Returns the keys where the on-disk copy differed and was overwritten, so
|
||||
the caller can tell the user something was actually contested rather than
|
||||
merely carried across.
|
||||
|
||||
Deliberately one-directional: a key absent locally is left alone on disk.
|
||||
We can't tell "user deleted their last set" from "user never had sets and
|
||||
another machine just added some", and silently deleting someone else's
|
||||
data is the worse of the two failures.
|
||||
"""
|
||||
conflicts: list[str] = []
|
||||
for key in BCC_OWNED_KEYS:
|
||||
if key not in local_cfg:
|
||||
continue
|
||||
if key in fresh_cfg and fresh_cfg[key] != local_cfg[key]:
|
||||
conflicts.append(key)
|
||||
fresh_cfg[key] = copy.deepcopy(local_cfg[key])
|
||||
return conflicts
|
||||
|
||||
|
||||
def _make_backup(path: Path) -> Path:
|
||||
bdir = path.parent / BACKUP_DIRNAME
|
||||
bdir.mkdir(exist_ok=True)
|
||||
@@ -761,13 +893,28 @@ def backup_label(backup_path: Path | str) -> str:
|
||||
return f"{ts[:4]}-{ts[4:6]}-{ts[6:8]} {ts[9:11]}:{ts[11:13]}:{ts[13:]}"
|
||||
|
||||
|
||||
def should_mask_value(key: str, value) -> bool:
|
||||
"""Whether an env/header value should be masked for display.
|
||||
|
||||
A ${VAR} reference is NOT a secret -- it's a pointer to one, and it's the
|
||||
thing we want users to adopt. Masking it to dots would make a reference
|
||||
indistinguishable from a stored credential, hiding exactly the distinction
|
||||
that makes the feature worth using (#76).
|
||||
"""
|
||||
if not is_secret_key(key):
|
||||
return False
|
||||
return not is_env_ref(value) if isinstance(value, str) else True
|
||||
|
||||
|
||||
def _redact_server_data(data: dict) -> dict:
|
||||
"""Return a copy of a server definition with secrets masked for display."""
|
||||
out = dict(data)
|
||||
if "args" in out:
|
||||
out["args"] = redact_args(list(out["args"] or []))
|
||||
if "env" in out:
|
||||
out["env"] = {k: (MASK if is_secret_key(k) else v) for k, v in (out["env"] or {}).items()}
|
||||
out["env"] = {
|
||||
k: (MASK if should_mask_value(k, v) else v) for k, v in (out["env"] or {}).items()
|
||||
}
|
||||
return out
|
||||
|
||||
|
||||
@@ -1318,6 +1465,155 @@ MASK = "••••••••"
|
||||
_EMBEDDED_CRED_RE = re.compile(r"://[^:@/\s]+:[^:@/\s]+@")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Environment-variable references (issue #76)
|
||||
#
|
||||
# Claude Code expands ${VAR} and ${VAR:-default} itself, in command, args, env,
|
||||
# url and headers. So BCC does NOT expand these on write -- resolving them into
|
||||
# the file would put the secret back on disk, which is the whole thing the user
|
||||
# is avoiding, and would defeat a feature the client already implements. BCC
|
||||
# authors, validates and warns.
|
||||
#
|
||||
# Claude Desktop has no documented support, so the same text there is passed to
|
||||
# the server literally. That makes this a per-client capability, not a global
|
||||
# one -- see client_expands_env_refs().
|
||||
# --------------------------------------------------------------------------- #
|
||||
# ${NAME} or ${NAME:-default}. Names follow the shell convention (letter or
|
||||
# underscore first) so a bare "${}" or "${1}" isn't mistaken for a reference.
|
||||
_ENV_REF_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)(?::-([^}]*))?\}")
|
||||
|
||||
# The five fields Claude Code documents as expansion sites.
|
||||
ENV_REF_FIELDS = ("command", "args", "env", "url", "headers")
|
||||
|
||||
|
||||
class EnvRef(NamedTuple):
|
||||
"""One ${VAR} / ${VAR:-default} occurrence found in a server definition."""
|
||||
|
||||
name: str
|
||||
default: str | None
|
||||
field: str # which of ENV_REF_FIELDS it was found in
|
||||
|
||||
@property
|
||||
def has_default(self) -> bool:
|
||||
return self.default is not None
|
||||
|
||||
|
||||
def find_env_refs(text: str, field: str = "") -> list[EnvRef]:
|
||||
"""Every ${VAR} / ${VAR:-default} reference in a single string."""
|
||||
if not isinstance(text, str):
|
||||
return []
|
||||
return [EnvRef(m.group(1), m.group(2), field) for m in _ENV_REF_RE.finditer(text)]
|
||||
|
||||
|
||||
def is_env_ref(value: str) -> bool:
|
||||
"""True when the value contains at least one ${VAR} reference.
|
||||
|
||||
Used to keep placeholders OUT of secret masking: `${API_KEY}` under a
|
||||
secret-looking key is a reference, not a secret, and masking it to dots
|
||||
would hide the one distinction the user needs to see.
|
||||
"""
|
||||
return bool(find_env_refs(value))
|
||||
|
||||
|
||||
def server_env_refs(data: dict) -> list[EnvRef]:
|
||||
"""Every env reference in a server definition, tagged with its field.
|
||||
|
||||
Only inspects the fields Claude Code actually expands; a ${VAR} written
|
||||
into some other key is not a reference and shouldn't be reported as one.
|
||||
"""
|
||||
out: list[EnvRef] = []
|
||||
if not isinstance(data, dict):
|
||||
return out
|
||||
for field in ENV_REF_FIELDS:
|
||||
value = data.get(field)
|
||||
if isinstance(value, str):
|
||||
out.extend(find_env_refs(value, field))
|
||||
elif isinstance(value, list):
|
||||
for item in value:
|
||||
out.extend(find_env_refs(item, field))
|
||||
elif isinstance(value, dict):
|
||||
for v in value.values():
|
||||
out.extend(find_env_refs(v, field))
|
||||
return out
|
||||
|
||||
|
||||
def expand_env_refs(text: str, environ: dict | None = None) -> str:
|
||||
"""Expand ${VAR} / ${VAR:-default} the way Claude Code documents it.
|
||||
|
||||
Provided for previewing what the client will do -- BCC never writes the
|
||||
expanded form back to the config. Unset with no default is left as the
|
||||
literal ${VAR} text, matching Claude Code: the config still loads and the
|
||||
unexpanded text is passed through.
|
||||
"""
|
||||
if not isinstance(text, str):
|
||||
return text
|
||||
env = os.environ if environ is None else environ
|
||||
|
||||
def repl(m: re.Match) -> str:
|
||||
name, default = m.group(1), m.group(2)
|
||||
if name in env:
|
||||
return env[name]
|
||||
return default if default is not None else m.group(0)
|
||||
|
||||
return _ENV_REF_RE.sub(repl, text)
|
||||
|
||||
|
||||
def unresolved_env_refs(data: dict, environ: dict | None = None) -> list[EnvRef]:
|
||||
"""References that would not resolve: variable unset AND no default.
|
||||
|
||||
Best-effort by nature -- BCC's environment isn't necessarily the client's,
|
||||
so this warns rather than blocks, and the warning text says so.
|
||||
"""
|
||||
env = os.environ if environ is None else environ
|
||||
return [r for r in server_env_refs(data) if not r.has_default and r.name not in env]
|
||||
|
||||
|
||||
def client_expands_env_refs(profile: Profile) -> bool:
|
||||
"""Whether the client behind `profile` expands ${VAR} itself.
|
||||
|
||||
Claude Code does, in command/args/env/url/headers, for both project
|
||||
`.mcp.json` and user-scope `~/.claude.json`. Claude Desktop has no
|
||||
documented support, so a reference there reaches the server as literal
|
||||
text -- which surfaces as a confusing auth failure rather than an obvious
|
||||
config error, hence the warning.
|
||||
"""
|
||||
return not profile_targets_claude_desktop(profile)
|
||||
|
||||
|
||||
def env_ref_warnings(
|
||||
data: dict, profile: Profile | None = None, environ: dict | None = None
|
||||
) -> list[str]:
|
||||
"""Advisory warnings about env references in one server definition.
|
||||
|
||||
Two distinct problems, deliberately worded differently:
|
||||
- the target client won't expand them at all (Claude Desktop)
|
||||
- the client will expand them, but a variable looks unset here
|
||||
"""
|
||||
refs = server_env_refs(data)
|
||||
if not refs:
|
||||
return []
|
||||
|
||||
if profile is not None and not client_expands_env_refs(profile):
|
||||
names = ", ".join(sorted({f"${{{r.name}}}" for r in refs}))
|
||||
return [
|
||||
f"{names} will NOT be expanded by Claude Desktop -- it has no "
|
||||
f"documented support for variable references, so the server "
|
||||
f"receives the literal text. Use a real value here, or move this "
|
||||
f"server to a Claude Code config."
|
||||
]
|
||||
|
||||
missing = unresolved_env_refs(data, environ)
|
||||
if not missing:
|
||||
return []
|
||||
names = ", ".join(sorted({r.name for r in missing}))
|
||||
return [
|
||||
f"{names} is not set in this environment and has no ':-default'. "
|
||||
f"Claude Code will pass the reference through unexpanded. "
|
||||
f"(Checked against BCC's environment, which may differ from the "
|
||||
f"client's.)"
|
||||
]
|
||||
|
||||
|
||||
def is_secret_key(name: str) -> bool:
|
||||
"""Does this env-var / header / flag name look like it holds a secret?"""
|
||||
return bool(_SECRET_KEY_RE.search(name or ""))
|
||||
@@ -1334,17 +1630,22 @@ def redact_args(args: list[str]) -> list[str]:
|
||||
--api-key=abc123 -> --api-key=•••••••• (inline flag=value)
|
||||
ghp_abc123 -> •••••••• (well-known token prefix)
|
||||
Everything else passes through untouched.
|
||||
|
||||
${VAR} references are left visible: they name a secret rather than being
|
||||
one, and hiding them would obscure the difference between "this config
|
||||
leaks a token" and "this config points at one" (#76).
|
||||
"""
|
||||
out: list[str] = []
|
||||
mask_next = False
|
||||
for a in args:
|
||||
s = str(a)
|
||||
if mask_next:
|
||||
out.append(MASK)
|
||||
mask_next = False
|
||||
out.append(s if is_env_ref(s) else MASK)
|
||||
continue
|
||||
if s.startswith("-") and "=" in s and is_secret_key(s.split("=", 1)[0]):
|
||||
out.append(s.split("=", 1)[0] + "=" + MASK)
|
||||
flag, value = s.split("=", 1)
|
||||
out.append(f"{flag}={value}" if is_env_ref(value) else f"{flag}={MASK}")
|
||||
continue
|
||||
if s.startswith("-") and is_secret_key(s):
|
||||
out.append(s)
|
||||
@@ -1371,6 +1672,11 @@ def args_secret_warning(data: dict) -> str | None:
|
||||
args = [str(a) for a in (data.get("args") or [])]
|
||||
mask_next = False
|
||||
for a in args:
|
||||
# A ${VAR} reference is the recommended fix for this very warning --
|
||||
# continuing to warn after the user adopts it punishes the fix (#76).
|
||||
if is_env_ref(a):
|
||||
mask_next = False
|
||||
continue
|
||||
if mask_next:
|
||||
mask_next = False
|
||||
if not a.startswith("-"):
|
||||
@@ -1512,9 +1818,21 @@ def lint_server(name: str, data: dict) -> list[str]:
|
||||
|
||||
|
||||
def lint_servers(servers: list[ServerEntry]) -> list[str]:
|
||||
"""Concatenate lint_server warnings across every entry, in order."""
|
||||
"""Concatenate lint_server warnings across every entry, in order.
|
||||
|
||||
Entries whose config value wasn't a JSON object at all are reported here
|
||||
rather than in lint_server, which takes an already-dict `data` (#72).
|
||||
"""
|
||||
out: list[str] = []
|
||||
for s in servers:
|
||||
if s.malformed:
|
||||
nm = s.name.strip() or "(unnamed)"
|
||||
out.append(
|
||||
f"'{nm}': server definition is not an object "
|
||||
f"(found {type(s.raw).__name__}) -- it is preserved as-is; "
|
||||
f"edit it to replace it with a proper definition"
|
||||
)
|
||||
continue
|
||||
out.extend(lint_server(s.name, s.data))
|
||||
return out
|
||||
|
||||
|
||||
@@ -2373,6 +2373,146 @@ def test_config_has_unfilled_placeholders_checks_env_too():
|
||||
assert c.config_has_unfilled_placeholders(cfg) is True
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# #72 -- a server value that isn't a JSON object must not take the load down
|
||||
# --------------------------------------------------------------------------- #
|
||||
@pytest.mark.parametrize("bad", ["not-a-dict", 123, ["a", "b"], None, True, 1.5])
|
||||
def test_extract_servers_survives_non_dict_server_value(bad):
|
||||
entries = c.extract_servers({"mcpServers": {"foo": bad}})
|
||||
assert len(entries) == 1
|
||||
assert entries[0].name == "foo"
|
||||
assert entries[0].data == {}
|
||||
assert entries[0].malformed is True
|
||||
assert entries[0].raw == bad
|
||||
|
||||
|
||||
def test_extract_servers_marks_only_the_bad_entry():
|
||||
cfg = {"mcpServers": {"good": {"command": "npx"}, "bad": "oops"}}
|
||||
by_name = {e.name: e for e in c.extract_servers(cfg)}
|
||||
assert by_name["good"].malformed is False
|
||||
assert by_name["good"].data == {"command": "npx"}
|
||||
assert by_name["bad"].malformed is True
|
||||
|
||||
|
||||
def test_extract_servers_handles_malformed_disabled_entry():
|
||||
entries = c.extract_servers({c.DISABLED_KEY: {"parked": ["nope"]}})
|
||||
assert entries[0].enabled is False
|
||||
assert entries[0].malformed is True
|
||||
|
||||
|
||||
def test_malformed_entry_round_trips_through_save_unchanged():
|
||||
"""The cardinal rule: never silently delete what the user had on disk."""
|
||||
cfg = {"mcpServers": {"good": {"command": "npx"}, "bad": "oops"}}
|
||||
servers = c.extract_servers(cfg)
|
||||
out = c.apply_servers(dict(cfg), servers)
|
||||
assert out["mcpServers"]["bad"] == "oops"
|
||||
assert out["mcpServers"]["good"] == {"command": "npx"}
|
||||
|
||||
|
||||
def test_editing_a_malformed_entry_retires_the_raw_value():
|
||||
entry = c.extract_servers({"mcpServers": {"bad": "oops"}})[0]
|
||||
entry.set_data({"command": "npx"})
|
||||
assert entry.malformed is False
|
||||
assert entry.config_value() == {"command": "npx"}
|
||||
assert c.apply_servers({}, [entry])["mcpServers"]["bad"] == {"command": "npx"}
|
||||
|
||||
|
||||
def test_lint_reports_the_malformed_entry_by_name():
|
||||
servers = c.extract_servers({"mcpServers": {"bad": "oops"}})
|
||||
warnings = c.lint_servers(servers)
|
||||
assert len(warnings) == 1
|
||||
assert "'bad'" in warnings[0]
|
||||
assert "not an object" in warnings[0]
|
||||
assert "str" in warnings[0]
|
||||
|
||||
|
||||
def test_lint_still_reports_normal_warnings_alongside_malformed():
|
||||
cfg = {"mcpServers": {"bad": "oops", "sloppy": {"command": "npx", "args": "one two"}}}
|
||||
warnings = c.lint_servers(c.extract_servers(cfg))
|
||||
assert any("not an object" in w for w in warnings)
|
||||
assert any("'args' should be a list" in w for w in warnings)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# #73 -- the stale-file merge must not discard BCC-authored keys
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_carry_owned_keys_moves_sets_onto_the_reloaded_config():
|
||||
local = {"mcpServers": {}, c.SETS_KEY: {"work": ["a", "b"]}}
|
||||
fresh = {"mcpServers": {"external": {"command": "npx"}}}
|
||||
contested = c.carry_owned_keys(local, fresh)
|
||||
assert contested == []
|
||||
assert fresh[c.SETS_KEY] == {"work": ["a", "b"]}
|
||||
assert fresh["mcpServers"] == {"external": {"command": "npx"}}
|
||||
|
||||
|
||||
def test_carry_owned_keys_reports_a_genuine_conflict():
|
||||
local = {c.SETS_KEY: {"work": ["a"]}}
|
||||
fresh = {c.SETS_KEY: {"work": ["a", "b"]}}
|
||||
assert c.carry_owned_keys(local, fresh) == [c.SETS_KEY]
|
||||
assert fresh[c.SETS_KEY] == {"work": ["a"]} # local wins: BCC owns the key
|
||||
|
||||
|
||||
def test_carry_owned_keys_is_quiet_when_both_sides_agree():
|
||||
local = {c.SETS_KEY: {"work": ["a"]}}
|
||||
fresh = {c.SETS_KEY: {"work": ["a"]}}
|
||||
assert c.carry_owned_keys(local, fresh) == []
|
||||
|
||||
|
||||
def test_carry_owned_keys_leaves_disk_alone_when_absent_locally():
|
||||
"""Can't distinguish 'deleted my last set' from 'never had sets'; keep theirs."""
|
||||
fresh = {c.SETS_KEY: {"remote": ["a"]}}
|
||||
assert c.carry_owned_keys({}, fresh) == []
|
||||
assert fresh[c.SETS_KEY] == {"remote": ["a"]}
|
||||
|
||||
|
||||
def test_carry_owned_keys_deep_copies_so_later_edits_do_not_leak():
|
||||
local = {c.SETS_KEY: {"work": ["a"]}}
|
||||
fresh = {}
|
||||
c.carry_owned_keys(local, fresh)
|
||||
local[c.SETS_KEY]["work"].append("b")
|
||||
assert fresh[c.SETS_KEY] == {"work": ["a"]}
|
||||
|
||||
|
||||
def test_merge_flow_preserves_sets_and_external_servers(tmp_path):
|
||||
"""End-to-end shape of the Merge & save path that lost sets in #73."""
|
||||
path = tmp_path / "claude.json"
|
||||
path.write_text(json.dumps({"mcpServers": {"old": {"command": "old"}}}))
|
||||
|
||||
# BCC loads, user saves a named set and edits servers in memory.
|
||||
local = c.load_config(path)
|
||||
servers = c.extract_servers(local)
|
||||
c.save_server_set(local, "work", servers)
|
||||
|
||||
# Something else rewrites the file underneath us.
|
||||
path.write_text(json.dumps({"mcpServers": {"external": {"command": "new"}}, "other": 1}))
|
||||
|
||||
# Merge & save: reload disk, carry BCC keys, re-apply the user's servers.
|
||||
fresh = c.load_config(path)
|
||||
c.carry_owned_keys(local, fresh)
|
||||
c.apply_servers(fresh, servers)
|
||||
c.write_config(path, fresh)
|
||||
|
||||
saved = c.load_config(path)
|
||||
assert saved[c.SETS_KEY] == {"work": ["old"]} # the set survived
|
||||
assert saved["other"] == 1 # unrelated external key preserved
|
||||
assert "old" in saved["mcpServers"] # user's servers re-applied
|
||||
|
||||
|
||||
def test_null_server_value_is_malformed_not_mistaken_for_absent():
|
||||
"""`{"mcpServers": {"foo": null}}` is legal JSON and a real malformed case,
|
||||
so None must not double as the 'nothing here' sentinel."""
|
||||
entry = c.extract_servers({"mcpServers": {"foo": None}})[0]
|
||||
assert entry.malformed is True
|
||||
assert entry.raw is None
|
||||
assert c.apply_servers({}, [entry])["mcpServers"]["foo"] is None
|
||||
|
||||
|
||||
def test_a_normal_entry_is_not_malformed():
|
||||
entry = c.extract_servers({"mcpServers": {"foo": {"command": "npx"}}})[0]
|
||||
assert entry.malformed is False
|
||||
assert entry.raw is c.NO_RAW
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# #75 -- theming
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -2486,3 +2626,192 @@ def test_every_palette_slot_is_consumed():
|
||||
if f.name == "name":
|
||||
continue
|
||||
assert f"p.{f.name}" in src, f"palette slot {f.name!r} is never consumed"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# #78/#79 -- update notice: when to show it, and what it says
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_update_notice_when_a_newer_release_exists():
|
||||
n = c.update_notice("1.2.0", {"version": "v1.3.0", "url": "https://example.test/rel"})
|
||||
assert n is not None
|
||||
assert n["version"] == "v1.3.0"
|
||||
assert n["url"] == "https://example.test/rel"
|
||||
assert "1.3.0" in n["text"] and "1.2.0" in n["text"]
|
||||
|
||||
|
||||
def test_update_notice_is_silent_when_current():
|
||||
assert c.update_notice("1.3.0", {"version": "v1.3.0"}) is None
|
||||
assert c.update_notice("1.4.0", {"version": "v1.3.0"}) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", [None, {}, {"version": ""}, {"version": None}, {"version": 3}, []])
|
||||
def test_update_notice_is_silent_on_a_failed_or_malformed_check(bad):
|
||||
"""fetch_latest_release returns None on any failure; a half-formed payload
|
||||
must not produce a notice pointing at nothing."""
|
||||
assert c.update_notice("1.0.0", bad) is None
|
||||
|
||||
|
||||
def test_update_notice_falls_back_to_the_releases_page_without_a_url():
|
||||
n = c.update_notice("1.0.0", {"version": "v2.0.0"})
|
||||
assert n["url"] == c.RELEASES_URL
|
||||
|
||||
|
||||
def test_update_notice_names_no_menu_path():
|
||||
"""The old status-line text said 'Help > About to view it', which is wrong
|
||||
on macOS -- Qt moves the About action into the application menu (#79). The
|
||||
notice carries its own action, so it must not describe a menu path."""
|
||||
n = c.update_notice("1.0.0", {"version": "v2.0.0"})
|
||||
lowered = n["text"].lower()
|
||||
for phrase in ("help", "about", "menu", "▸", ">"):
|
||||
assert phrase not in lowered, f"notice text should not reference {phrase!r}"
|
||||
|
||||
|
||||
def test_update_notice_handles_the_v_prefix_consistently():
|
||||
assert c.update_notice("1.2.0", {"version": "1.3.0"}) is not None
|
||||
assert c.update_notice("v1.2.0", {"version": "v1.3.0"}) is not None
|
||||
assert c.update_notice("1.3.0", {"version": "v1.3.0"}) is None
|
||||
|
||||
|
||||
def test_update_notice_renders_both_versions_the_same_way():
|
||||
"""Tags carry a 'v' prefix, __version__ doesn't -- don't show both forms
|
||||
in one sentence."""
|
||||
n = c.update_notice("1.2.0", {"version": "v1.3.0"})
|
||||
assert "v1.3.0" not in n["text"]
|
||||
assert "1.3.0" in n["text"] and "1.2.0" in n["text"]
|
||||
# the machine-readable field keeps the real tag
|
||||
assert n["version"] == "v1.3.0"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# #76 -- ${VAR} references. Semantics mirror Claude Code's documented
|
||||
# behaviour: ${VAR} and ${VAR:-default}, expanded in command/args/env/url/
|
||||
# headers, and an unset variable with no default left as literal text.
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_find_env_refs_plain_and_defaulted():
|
||||
refs = c.find_env_refs("${A} and ${B:-fallback}")
|
||||
assert [(r.name, r.default) for r in refs] == [("A", None), ("B", "fallback")]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("text", ["${}", "${1BAD}", "$NOTBRACED", "{NOPE}", "plain", "$${X"])
|
||||
def test_find_env_refs_ignores_non_references(text):
|
||||
assert c.find_env_refs(text) == []
|
||||
|
||||
|
||||
def test_find_env_refs_allows_an_empty_default():
|
||||
"""`${VAR:-}` is a documented way to say 'blank if unset'."""
|
||||
refs = c.find_env_refs("${A:-}")
|
||||
assert refs[0].default == ""
|
||||
assert refs[0].has_default is True
|
||||
|
||||
|
||||
def test_server_env_refs_covers_all_five_documented_fields():
|
||||
data = {
|
||||
"command": "${BIN}",
|
||||
"args": ["--x", "${ARG}"],
|
||||
"env": {"K": "${ENVV}"},
|
||||
"url": "${URL}/mcp",
|
||||
"headers": {"Authorization": "Bearer ${HDR}"},
|
||||
}
|
||||
found = {(r.name, r.field) for r in c.server_env_refs(data)}
|
||||
assert found == {
|
||||
("BIN", "command"),
|
||||
("ARG", "args"),
|
||||
("ENVV", "env"),
|
||||
("URL", "url"),
|
||||
("HDR", "headers"),
|
||||
}
|
||||
|
||||
|
||||
def test_server_env_refs_ignores_unexpanded_fields():
|
||||
"""Claude Code expands five fields; a ${VAR} elsewhere isn't a reference."""
|
||||
assert c.server_env_refs({"description": "${NOPE}", "timeout": "${ALSO_NO}"}) == []
|
||||
|
||||
|
||||
def test_expand_env_refs_matches_documented_semantics():
|
||||
env = {"SET": "value"}
|
||||
assert c.expand_env_refs("${SET}", env) == "value"
|
||||
assert c.expand_env_refs("${MISSING:-dflt}", env) == "dflt"
|
||||
assert c.expand_env_refs("${SET:-dflt}", env) == "value"
|
||||
# unset with no default: left as literal text, exactly as Claude Code does
|
||||
assert c.expand_env_refs("${MISSING}", env) == "${MISSING}"
|
||||
|
||||
|
||||
def test_expand_env_refs_handles_several_in_one_string():
|
||||
assert c.expand_env_refs("${A}/${B:-two}/${C}", {"A": "one"}) == "one/two/${C}"
|
||||
|
||||
|
||||
def test_unresolved_env_refs_only_flags_unset_without_default():
|
||||
data = {"env": {"A": "${SET}", "B": "${UNSET}", "C": "${OTHER:-has_default}"}}
|
||||
assert [r.name for r in c.unresolved_env_refs(data, {"SET": "x"})] == ["UNSET"]
|
||||
|
||||
|
||||
# --- the two interactions that were backwards for this feature ------------
|
||||
def test_placeholder_under_a_secret_key_is_not_masked():
|
||||
"""A ${VAR} names a secret rather than being one. Masking it would make a
|
||||
reference indistinguishable from a stored credential."""
|
||||
assert c.should_mask_value("API_KEY", "${API_KEY}") is False
|
||||
assert c.should_mask_value("API_KEY", "ghp_realsecret") is True
|
||||
assert c.should_mask_value("NOT_SECRET", "${API_KEY}") is False
|
||||
|
||||
|
||||
def test_redacted_display_keeps_placeholders_but_masks_real_secrets():
|
||||
out = c._redact_server_data({"env": {"API_KEY": "${API_KEY}", "TOKEN": "ghp_real"}})
|
||||
assert out["env"]["API_KEY"] == "${API_KEY}"
|
||||
assert out["env"]["TOKEN"] == c.MASK
|
||||
|
||||
|
||||
def test_redact_args_keeps_placeholders_visible():
|
||||
assert c.redact_args(["--token", "${GH_TOKEN}"]) == ["--token", "${GH_TOKEN}"]
|
||||
assert c.redact_args(["--api-key=${K}"]) == ["--api-key=${K}"]
|
||||
# real secrets still masked
|
||||
assert c.redact_args(["--token", "ghp_real"]) == ["--token", c.MASK]
|
||||
assert c.redact_args(["--api-key=sk-real"]) == [f"--api-key={c.MASK}"]
|
||||
|
||||
|
||||
def test_args_secret_warning_is_silenced_by_a_placeholder():
|
||||
"""Moving a token into ${VAR} is the recommended fix for this warning --
|
||||
still warning afterwards would punish the fix."""
|
||||
assert c.args_secret_warning({"args": ["--token", "ghp_real"]}) is not None
|
||||
assert c.args_secret_warning({"args": ["--token", "${GH_TOKEN}"]}) is None
|
||||
|
||||
|
||||
def test_args_secret_warning_still_fires_on_the_arg_after_a_placeholder():
|
||||
"""A placeholder must clear the pending-flag state, not blanket-suppress."""
|
||||
assert c.args_secret_warning({"args": ["${SAFE}", "--token", "ghp_real"]}) is not None
|
||||
|
||||
|
||||
# --- per-client gating ----------------------------------------------------
|
||||
def _profile(path):
|
||||
return c.Profile(label="p", path=Path(path), config_exists=True)
|
||||
|
||||
|
||||
def test_claude_code_profiles_expand_references():
|
||||
assert c.client_expands_env_refs(_profile(Path.home() / ".claude.json")) is True
|
||||
assert c.client_expands_env_refs(_profile("/repo/.mcp.json")) is True
|
||||
|
||||
|
||||
def test_claude_desktop_profile_does_not_expand_references():
|
||||
desktop = _profile(c.app_support_base() / "Claude" / c.CONFIG_FILENAME)
|
||||
assert c.client_expands_env_refs(desktop) is False
|
||||
|
||||
|
||||
def test_desktop_profile_warns_that_references_are_literal():
|
||||
data = {"env": {"API_KEY": "${API_KEY}"}}
|
||||
desktop = _profile(c.app_support_base() / "Claude" / c.CONFIG_FILENAME)
|
||||
warnings = c.env_ref_warnings(data, desktop, {"API_KEY": "set"})
|
||||
assert len(warnings) == 1
|
||||
assert "NOT be expanded" in warnings[0]
|
||||
assert "${API_KEY}" in warnings[0]
|
||||
|
||||
|
||||
def test_claude_code_profile_warns_only_about_unset_variables():
|
||||
code = _profile(Path.home() / ".claude.json")
|
||||
data = {"env": {"A": "${UNSET_ONE}"}}
|
||||
assert c.env_ref_warnings(data, code, {}) != []
|
||||
assert c.env_ref_warnings(data, code, {"UNSET_ONE": "x"}) == []
|
||||
# a default means it always resolves
|
||||
assert c.env_ref_warnings({"env": {"A": "${X:-d}"}}, code, {}) == []
|
||||
|
||||
|
||||
def test_no_references_means_no_warnings():
|
||||
assert c.env_ref_warnings({"command": "npx", "args": ["-y", "pkg"]}, None) == []
|
||||
|
||||
Reference in New Issue
Block a user