diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 0645a9d0..3f2506a2 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -7,12 +7,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- Credential files written through the default file backend now use private
+  temporary files and atomic replacement, requiring write access to the parent
+  directory. Unchanged regular files have their permissions repaired. Default
+  protected saves replace a destination symlink even when its bytes are unchanged; hard-linked
+  copies no longer share subsequent updates.
+- Authentication errors and verbose HTTP diagnostics omit provider-supplied
+  error descriptions, response reasons and redirect headers that could expose
+  credentials. OAuth response bodies and redirect destinations are also omitted
+  from verbose logs, along with all OAuth response header values.
+  Resumable Cloud account lookup uses the same protection. Encoded credential
+  query names are redacted, and cookie diagnostics omit names as well as values.
+- Generated manifests and bundles exclude `.rsconnect-python` directories and
+  the configured CLI credential directory, including explicit extra files and
+  symlink aliases. Publishing from inside that directory is rejected, including
+  static notebook rendering before execution. Prepared
+  bundle archives remain unchanged.
 - `login` and `add --connect-cloud` can start device-code authentication and return
   approval details as JSON. Finish a pending login by nickname with `--finish`.
   Use `--timeout` to bound the entire finish invocation, including account lookup.
   Resumable login requires POSIX (Linux or macOS). Pending state is validated
-  before saving and uses owner-only permissions. Resumable authentication omits
-  sensitive provider diagnostics. Connect Cloud finish retains the original OAuth client.
+  before saving and uses owner-only permissions. Connect Cloud finish retains the original OAuth client.
 - Added `rsconnect preflight` to check Python and Node.js runtime availability on
   self-hosted Posit Connect and Snowpark Container Services (SPCS), from POSIX systems. For first
   publishes, it considers server installations marked publishable. Python
diff --git a/docs/commands/deploy.md b/docs/commands/deploy.md
index 9d5b05e5..eef23fb0 100644
--- a/docs/commands/deploy.md
+++ b/docs/commands/deploy.md
@@ -1,3 +1,11 @@
+Generated bundles exclude `.rsconnect-python` directories and the configured
+CLI credential directory, including symlink aliases and files supplied as
+explicit extras. Keep application files outside that directory; publishing
+from inside it is rejected.
+
+Deploying an already prepared bundle uploads the archive as supplied. Check its
+contents before deployment.
+
 ::: mkdocs-click
     :module: rsconnect.main
     :command: deploy
diff --git a/docs/commands/write-manifest.md b/docs/commands/write-manifest.md
index 7b87cca6..da6868d7 100644
--- a/docs/commands/write-manifest.md
+++ b/docs/commands/write-manifest.md
@@ -1,3 +1,8 @@
+Generated manifests exclude `.rsconnect-python` directories and the configured
+CLI credential directory, including symlink aliases and files supplied as
+explicit extras. Keep application files outside that directory; generating
+a manifest from inside it is rejected.
+
 ::: mkdocs-click
     :module: rsconnect.main
     :command: write_manifest
diff --git a/rsconnect/api.py b/rsconnect/api.py
index 19c491d1..96b1ef33 100644
--- a/rsconnect/api.py
+++ b/rsconnect/api.py
@@ -167,7 +167,8 @@ def handle_bad_response(self, response: HTTPResponse | T, is_httpresponse: bool
             safe_uri = _redacted_uri_for_log(response.full_uri)
             if response.exception:
                 raise RSConnectException(
-                    "Could not connect to %s - %s" % (_redacted_uri_for_log(self.url), response.exception),
+                    "Could not connect to %s (%s)"
+                    % (_redacted_uri_for_log(self.url), type(response.exception).__name__),
                     cause=response.exception,
                 )
             # Sometimes an ISP will respond to an unknown server name by returning a friendly
@@ -188,12 +189,11 @@ def handle_bad_response(self, response: HTTPResponse | T, is_httpresponse: bool
                     raise RSConnectException(error, status=response.status)
                 if response.status is None or response.status < 200 or response.status > 299:
                     raise RSConnectException(
-                        "Received an unexpected response from %s (calling %s): %s %s"
+                        "Received an unexpected response from %s (calling %s): %s"
                         % (
                             self.remote_name,
                             safe_uri,
                             response.status,
-                            response.reason,
                         ),
                         status=response.status,
                     )
@@ -203,12 +203,11 @@ def handle_bad_response(self, response: HTTPResponse | T, is_httpresponse: bool
                     # prior function call was not converted from a HTTPResponse to JSON. This
                     # should never happen, so raise an exception.
                     raise RSConnectException(
-                        "Received an unexpected response from %s (calling %s): %s %s"
+                        "Received an unexpected response from %s (calling %s): %s"
                         % (
                             self.remote_name,
                             safe_uri,
                             response.status,
-                            response.reason,
                         )
                     )
         return response
@@ -400,6 +399,7 @@ def fmt_payload(self):
     def exchange_token(self) -> str:
         try:
             server = HTTPServer(url=self.token_endpoint())
+            server._suppress_oauth_response_logging = True
             payload = self.fmt_payload()
 
             response = server.request(
@@ -412,17 +412,17 @@ def exchange_token(self) -> str:
             # since we don't want to pick up its json decoding assumptions
             if response.exception is not None:
                 raise RSConnectException(
-                    "Could not connect to %s - %s" % (self.token_endpoint(), response.exception),
+                    "Could not connect to %s (%s)"
+                    % (_redacted_uri_for_log(self.token_endpoint()), type(response.exception).__name__),
                     cause=response.exception,
                 )
             if response.status is None or response.status < 200 or response.status > 299:
                 raise RSConnectException(
-                    "Received an unexpected response from %s (calling %s): %s %s"
+                    "Received an unexpected response from %s (calling %s): %s"
                     % (
                         self.url,
-                        response.full_uri,
+                        _redacted_uri_for_log(response.full_uri),
                         response.status,
-                        response.reason,
                     )
                 )
 
@@ -598,10 +598,12 @@ def _attempt_token_refresh(self) -> bool:
                     store._set(entry_name, entry)  # type: ignore[possibly-undefined]
                 logger.warning("OAuth client was re-registered; please run `rsconnect login` again.")
             except Exception as exc:
-                logger.warning(f"OAuth client re-registration failed: {exc}. Please run `rsconnect login` again.")
+                logger.warning(
+                    "OAuth client re-registration failed (%s). Please run `rsconnect login` again." % type(exc).__name__
+                )
             return False
         except Exception as exc:
-            logger.warning(f"OAuth token refresh failed: {exc}")
+            logger.warning("OAuth token refresh failed (%s)." % type(exc).__name__)
             return False
 
         new_access = token_response["access_token"]
@@ -3411,16 +3413,8 @@ def _refresh_user_token(self) -> dict[str, Any]:
             request_options["request_deadline"] = self.request_deadline
         if self._server.oauth_client_id is not None:
             request_options["client_id_override"] = self._server.oauth_client_id
-        if self._suppress_oauth_response_logging:
-            request_options["suppress_response_logging"] = True
         return connect_cloud.refresh(cast(str, self._server.refresh_token), self._server.environment, **request_options)
 
-    def _warn_refresh_failure(self, exception: Exception) -> None:
-        if self._suppress_oauth_response_logging:
-            logger.warning("Posit Connect Cloud token refresh failed (%s)." % type(exception).__name__)
-        else:
-            logger.warning("Posit Connect Cloud token refresh failed: %s" % exception)
-
     def _attempt_token_refresh(self) -> bool:
         """Mint a new access token and apply it to this client.
 
@@ -3448,7 +3442,7 @@ def _attempt_token_refresh(self) -> bool:
         except InvalidClientError as exc:
             if not service_account:
                 # This CLI's own OAuth client, not the user's credential.
-                self._warn_refresh_failure(exc)
+                logger.warning("Posit Connect Cloud token refresh failed (%s)." % type(exc).__name__)
                 return False
             raise RSConnectException(
                 "The Posit Connect Cloud service account credential was rejected — it has been revoked or "
@@ -3457,7 +3451,7 @@ def _attempt_token_refresh(self) -> bool:
             ) from exc
         except InvalidGrantError as exc:
             if service_account:
-                self._warn_refresh_failure(exc)
+                logger.warning("Posit Connect Cloud token refresh failed (%s)." % type(exc).__name__)
                 return False
             self._persist_tokens(None, None)
             raise RSConnectException(
@@ -3465,7 +3459,7 @@ def _attempt_token_refresh(self) -> bool:
                 "Authenticate again with `%s`." % self._add_command()
             ) from exc
         except RSConnectException as exc:
-            self._warn_refresh_failure(exc)
+            logger.warning("Posit Connect Cloud token refresh failed (%s)." % type(exc).__name__)
             return False
 
         access_token = tokens.get("access_token")
diff --git a/rsconnect/bundle.py b/rsconnect/bundle.py
index ea5edde1..7ca82db8 100644
--- a/rsconnect/bundle.py
+++ b/rsconnect/bundle.py
@@ -42,6 +42,8 @@
     cast,
 )
 
+from . import metadata
+
 # Even though TypedDict is available in Python 3.8, because it's used with NotRequired,
 # they should both come from the same typing module.
 # https://peps.python.org/pep-0655/#usage-in-python-3-11
@@ -79,10 +81,53 @@
     "node_modules/",
 ]
 directories_to_ignore = {Path(d) for d in directories_ignore_list}
+_config_dir_names = {".rsconnect-python"}
 
 mimetypes.add_type("text/ipynb", ".ipynb")
 
 
+def _resolved_config_dir() -> Path:
+    return Path(metadata.config_dirname()).expanduser().resolve()
+
+
+def _path_is_within(path: Path, directory: Path) -> bool:
+    return path == directory or directory in path.parents
+
+
+def _is_config_path(
+    path: str | Path,
+    base_dir: str | Path | None = None,
+    config_dir: Path | None = None,
+) -> bool:
+    candidate = Path(path)
+    if _config_dir_names.intersection(candidate.parts):
+        return True
+    if config_dir is None:
+        if base_dir is None and not candidate.is_absolute():
+            return False
+        config_dir = _resolved_config_dir()
+    if base_dir is not None and not candidate.is_absolute():
+        candidate = Path(base_dir) / candidate
+    return _path_is_within(candidate.resolve(), config_dir)
+
+
+def _ensure_content_root_is_safe(path: str | Path, config_dir: Path) -> None:
+    content_root = Path(path)
+    if content_root.is_file():
+        content_root = content_root.parent
+    content_root = content_root.resolve()
+    if _path_is_within(content_root, config_dir):
+        raise RSConnectException(
+            f"Cannot bundle content from '{content_root}' because it is inside the "
+            f"rsconnect-python configuration directory '{config_dir}'. "
+            "Choose a content directory outside the configuration directory."
+        )
+
+
+def _filter_config_paths(paths: Sequence[str], base_dir: str | Path, config_dir: Path) -> list[str]:
+    return [path for path in paths if not _is_config_path(path, base_dir, config_dir)]
+
+
 class ManifestDataFile(TypedDict):
     checksum: str
 
@@ -623,6 +668,11 @@ def make_notebook_source_bundle(
         extra_files = []
     base_dir = dirname(file)
     nb_name = basename(file)
+    config_dir = _resolved_config_dir()
+    _ensure_content_root_is_safe(base_dir, config_dir)
+    if _is_config_path(file, config_dir=config_dir):
+        raise RSConnectException("A credential configuration file cannot be bundled as notebook content.")
+    extra_files = _filter_config_paths(extra_files, base_dir, config_dir)
 
     manifest = make_source_manifest(
         AppModes.JUPYTER_NOTEBOOK,
@@ -740,6 +790,11 @@ def make_notebook_html_bundle(
     hide_tagged_input: bool,
     check_output: Callable[..., bytes] = subprocess.check_output,
 ) -> typing.IO[bytes]:
+    config_dir = _resolved_config_dir()
+    _ensure_content_root_is_safe(dirname(filename), config_dir)
+    if _is_config_path(filename, config_dir=config_dir):
+        raise RSConnectException("A credential configuration file cannot be bundled as notebook content.")
+
     # noinspection SpellCheckingInspection
     cmd = [
         python,
@@ -780,7 +835,12 @@ def make_notebook_html_bundle(
     return bundle_file
 
 
-def keep_manifest_specified_file(relative_path: str, ignore_path_set: set[Path] = directories_to_ignore) -> bool:
+def keep_manifest_specified_file(
+    relative_path: str,
+    ignore_path_set: set[Path] = directories_to_ignore,
+    base_dir: str | Path | None = None,
+    config_dir: Path | None = None,
+) -> bool:
     """
     A helper to see if the relative path given, which is assumed to have come
     from a manifest.json file, should be kept or ignored.
@@ -788,6 +848,8 @@ def keep_manifest_specified_file(relative_path: str, ignore_path_set: set[Path]
     :param relative_path: the relative path name to check.
     :return: True, if the path should kept or False, if it should be ignored.
     """
+    if _is_config_path(relative_path, base_dir, config_dir):
+        return False
     p = Path(relative_path)
     for parent in p.parents:
         if parent in ignore_path_set:
@@ -950,7 +1012,16 @@ def make_manifest_bundle(manifest_path: str | Path) -> typing.IO[bytes]:
     manifest, raw_manifest = read_manifest_file(manifest_path)
 
     base_dir = dirname(manifest_path)
-    files = list(filter(keep_manifest_specified_file, manifest.get("files", {}).keys()))
+    config_dir = _resolved_config_dir()
+    _ensure_content_root_is_safe(base_dir, config_dir)
+    manifest_files = manifest.get("files", {})
+    config_files = {path for path in manifest_files if _is_config_path(path, base_dir, config_dir)}
+    files = [
+        path for path in manifest_files if keep_manifest_specified_file(path, base_dir=base_dir, config_dir=config_dir)
+    ]
+    if config_files:
+        manifest["files"] = {path: value for path, value in manifest_files.items() if path not in config_files}
+        raw_manifest = json.dumps(manifest, indent=2)
 
     if "manifest.json" in files:
         # this will be created
@@ -1230,6 +1301,48 @@ def make_tensorflow_bundle(
     return bundle.to_file(directory)
 
 
+def _should_skip_walk_directory(
+    cur_dir: str,
+    sub_dirs: list[str],
+    exclude_paths: set[Path],
+    config_dir: Path,
+) -> bool:
+    sub_dirs[:] = [
+        sub_dir for sub_dir in sub_dirs if not _is_config_path(join(cur_dir, sub_dir), config_dir=config_dir)
+    ]
+    current_dir = Path(cur_dir)
+    return current_dir in exclude_paths or any(parent in exclude_paths for parent in current_dir.parents)
+
+
+def _iter_directory_files(
+    path: str,
+    extra_files: Sequence[str],
+    excludes: Sequence[str],
+    use_abspath: bool,
+    config_dir: Path,
+) -> Iterator[str]:
+    glob_set = create_glob_set(path, excludes)
+    exclude_paths = {Path(p) for p in excludes}
+    for cur_dir, sub_dirs, files in os.walk(path):
+        if _should_skip_walk_directory(cur_dir, sub_dirs, exclude_paths, config_dir):
+            continue
+        for file in files:
+            cur_path = os.path.join(cur_dir, file)
+            rel_path = relpath(cur_path, path)
+            if Path(cur_path) in exclude_paths:
+                continue
+            if not keep_manifest_specified_file(
+                rel_path,
+                exclude_paths | directories_to_ignore,
+                base_dir=path,
+                config_dir=config_dir,
+            ):
+                continue
+            if rel_path not in extra_files and glob_set.matches(cur_path):
+                continue
+            yield abspath(cur_path) if use_abspath else rel_path
+
+
 def create_file_list(
     path: str,
     extra_files: Sequence[str],
@@ -1247,32 +1360,19 @@ def create_file_list(
     :return: the list of relevant files, relative to the given directory.
     """
     extra_files = extra_files or []
-    excludes = excludes if excludes else []
-    glob_set = create_glob_set(path, excludes)
-    exclude_paths = {Path(p) for p in excludes}
-    file_set: set[str] = set(extra_files)
+    excludes = excludes or []
+    config_dir = _resolved_config_dir()
+    content_root = Path(path).parent if isfile(path) else Path(path)
+    _ensure_content_root_is_safe(content_root, config_dir)
+    file_set: set[str] = set(_filter_config_paths(extra_files, content_root, config_dir))
 
     if isfile(path):
-        path_to_add = abspath(path) if use_abspath else path
-        file_set.add(path_to_add)
+        if not _is_config_path(path, config_dir=config_dir):
+            path_to_add = abspath(path) if use_abspath else path
+            file_set.add(path_to_add)
         return sorted(file_set)
 
-    for cur_dir, _, files in os.walk(path):
-        if Path(cur_dir) in exclude_paths:
-            continue
-        if any(parent in exclude_paths for parent in Path(cur_dir).parents):
-            continue
-        for file in files:
-            cur_path = os.path.join(cur_dir, file)
-            rel_path = relpath(cur_path, path)
-
-            if Path(cur_path) in exclude_paths:
-                continue
-            if keep_manifest_specified_file(rel_path, exclude_paths | directories_to_ignore) and (
-                rel_path in extra_files or not glob_set.matches(cur_path)
-            ):
-                path_to_add = abspath(cur_path) if use_abspath else rel_path
-                file_set.add(path_to_add)
+    file_set.update(_iter_directory_files(path, extra_files, excludes, use_abspath, config_dir))
 
     return sorted(file_set)
 
@@ -1712,7 +1812,11 @@ def make_quarto_manifest(
         # Standalone Quarto document
         base_dir = dirname(file_or_directory)
         file_name = basename(file_or_directory)
-        relevant_files = [file_name] + list(extra_files or [])
+        config_dir = _resolved_config_dir()
+        _ensure_content_root_is_safe(base_dir, config_dir)
+        if _is_config_path(file_or_directory, config_dir=config_dir):
+            raise RSConnectException("A credential configuration file cannot be bundled as Quarto content.")
+        relevant_files = [file_name] + _filter_config_paths(extra_files or [], base_dir, config_dir)
 
     manifest = make_source_manifest(
         app_mode,
diff --git a/rsconnect/connect_cloud.py b/rsconnect/connect_cloud.py
index dd58dafd..1872b90e 100644
--- a/rsconnect/connect_cloud.py
+++ b/rsconnect/connect_cloud.py
@@ -242,8 +242,6 @@ def refresh(
     request_timeout: Optional[float] = None,
     request_deadline: Optional[float] = None,
     client_id_override: Optional[str] = None,
-    *,
-    suppress_response_logging: bool = False,
 ) -> dict[str, Any]:
     """Mint a new access token from a refresh token."""
     env = environment or environment_name()
@@ -252,8 +250,6 @@ def refresh(
         request_options["request_timeout"] = request_timeout
     if request_deadline is not None:
         request_options["request_deadline"] = request_deadline
-    if suppress_response_logging:
-        request_options["suppress_response_logging"] = True
     return refresh_access_token(
         metadata=urls(env).oauth_metadata(),
         client_id=client_id_override if client_id_override is not None else client_id(env),
diff --git a/rsconnect/device_login.py b/rsconnect/device_login.py
index 47ef70f7..643ba7d5 100644
--- a/rsconnect/device_login.py
+++ b/rsconnect/device_login.py
@@ -634,7 +634,6 @@ def _post_form(
         ca_data,
         request_timeout,
         request_deadline,
-        suppress_response_logging=True,
     )
 
 
@@ -662,13 +661,12 @@ def _register_device_client(
             ca_data,
             request_timeout=_remaining(deadline),
             request_deadline=deadline,
-            suppress_response_logging=True,
         )
+        return _identifier(client_id, "OAuth client ID")
     except InvalidClientError:
         raise
     except RSConnectException:
         raise RSConnectException("OAuth client registration failed.") from None
-    return _identifier(client_id, "OAuth client ID")
 
 
 def _start_device_request(
@@ -819,12 +817,21 @@ def _start_connect_login(
             ca_data,
             request_timeout=_remaining(deadline),
             request_deadline=deadline,
-            suppress_response_logging=True,
         ),
         server,
     )
     if not client_id:
-        client_id = _register_device_client(metadata, server, insecure, ca_data, deadline)
+        client_id = _identifier(
+            register_client(
+                metadata,
+                server,
+                insecure,
+                ca_data,
+                request_timeout=_remaining(deadline),
+                request_deadline=deadline,
+            ),
+            "OAuth client ID",
+        )
     client_id, response = _start_device_request(metadata, server, client_id, None, insecure, ca_data, deadline)
     state = _make_state(
         "connect", name, server, metadata, response, client_id, None, set_default, insecure, ca_data, None
@@ -1078,10 +1085,9 @@ def _lookup_cloud_account(
             raise InvalidGrantError() from exc
         if deadline <= time.monotonic():
             raise _FinishDeadline() from exc
-        if isinstance(exc, RSConnectException):
-            status_detail = " (HTTP %s)" % exc.status if exc.status is not None else ""
+        if isinstance(exc, RSConnectException) and exc.status is not None:
             safe_error = RSConnectException(
-                "Posit Connect Cloud account lookup failed%s." % status_detail,
+                "Posit Connect Cloud account lookup failed (HTTP %s)." % exc.status,
                 cause=exc.cause,
                 status=exc.status,
             )
diff --git a/rsconnect/http_support.py b/rsconnect/http_support.py
index 86b66a6e..fc5964c7 100644
--- a/rsconnect/http_support.py
+++ b/rsconnect/http_support.py
@@ -15,7 +15,7 @@
 from http import client as http
 from http.cookies import SimpleCookie
 from typing import IO, Any, Dict, List, Mapping, Optional, Tuple, Union, cast
-from urllib.parse import urlencode, urljoin, urlparse
+from urllib.parse import unquote_plus, urlencode, urljoin, urlparse
 from warnings import warn
 
 from . import VERSION
@@ -51,6 +51,7 @@
     "x-auth-token",
     "x-auth-signature",
     "x-rsc-authorization",
+    "location",
 }
 _SENSITIVE_FIELDS = (
     "client_secret",
@@ -72,6 +73,8 @@
     "subject_token",
     # PKCE (RFC 7636).
     "code_verifier",
+    "error_description",
+    "user_code",
     # The bootstrap response body carries a freshly minted admin API key.
     "api_key",
     "id_token",
@@ -82,6 +85,7 @@
 # redaction: in JSON bodies a bare "code" key is an error code (Connect,
 # shinyapps.io), which the debug log must keep readable.
 _SENSITIVE_FORM_ONLY_FIELDS = _SENSITIVE_FIELDS + ("code",)
+_SENSITIVE_QUERY_FIELDS = frozenset(_SENSITIVE_FORM_ONLY_FIELDS)
 _SENSITIVE_FORM_FIELD = re.compile(r"\b(%s)=[^&\s'\"]*" % "|".join(_SENSITIVE_FORM_ONLY_FIELDS), re.IGNORECASE)
 _SENSITIVE_JSON_FIELD = re.compile(r'"(%s)"\s*:\s*"[^"]*"' % "|".join(_SENSITIVE_FIELDS), re.IGNORECASE)
 
@@ -96,8 +100,26 @@ def _redacted_header_for_log(key: str, value: str) -> str:
 
 
 def _redacted_uri_for_log(uri: str) -> str:
-    """Redact credential-bearing query parameters, e.g. a presigned upload URL's."""
-    return _SENSITIVE_FORM_FIELD.sub(r"\1=<redacted>", uri)
+    """Redact credential fields in query strings and OAuth-style fragments."""
+    uri = _SENSITIVE_FORM_FIELD.sub(r"\1=<redacted>", uri)
+    before_fragment, fragment_marker, fragment = uri.partition("#")
+    suffix = fragment_marker + _redacted_query_for_log(fragment) if fragment_marker else ""
+    path, query_marker, query = before_fragment.partition("?")
+    if not query_marker:
+        return before_fragment + suffix
+    return path + query_marker + _redacted_query_for_log(query) + suffix
+
+
+def _redacted_query_for_log(query: str) -> str:
+    """Redact sensitive query values while preserving parameter spelling and order."""
+    parameters: list[str] = []
+    for parameter in query.split("&"):
+        name, separator, _ = parameter.partition("=")
+        if separator and unquote_plus(name).casefold() in _SENSITIVE_QUERY_FIELDS:
+            parameters.append(name + "=<redacted>")
+        else:
+            parameters.append(parameter)
+    return "&".join(parameters)
 
 
 def _redact_json_value(value: JsonData) -> JsonData:
@@ -121,8 +143,8 @@ def _redacted_body_for_log(body: object) -> object:
     Only affects what is logged; the body itself is sent untouched. Streams and
     other non-text bodies are logged as their repr, which carries no content.
     JSON bodies are parsed and redacted structurally, since a secret containing
-    an escaped quote would leak past a regex; everything else falls back to the
-    form-encoded pattern.
+    an escaped quote would leak past a regex. Malformed JSON is logged as a
+    placeholder instead of falling back to patterns that may miss escaped data.
     """
     if isinstance(body, bytes):
         text = body.decode("utf-8", errors="replace")
@@ -136,7 +158,7 @@ def _redacted_body_for_log(body: object) -> object:
         try:
             return json.dumps(_redact_json_value(json.loads(text)))
         except (json.JSONDecodeError, ValueError):
-            pass
+            return "<invalid JSON>"
 
     text = _SENSITIVE_FORM_FIELD.sub(r"\1=<redacted>", text)
     return _SENSITIVE_JSON_FIELD.sub(r'"\1": "<redacted>"', text)
@@ -639,8 +661,7 @@ def _log_request(
     ) -> None:
         if not logger.is_debugging():
             return
-        private_uri = suppress_uri_logging or self._suppress_oauth_response_logging
-        logged_uri = "<OAuth endpoint>" if private_uri else _redacted_uri_for_log(full_uri)
+        logged_uri = "<OAuth redirect>" if suppress_uri_logging else _redacted_uri_for_log(full_uri)
         logger.debug(f"Request: {method} {logged_uri}")
         logger.debug("Headers:")
         for key, value in headers.items():
@@ -651,10 +672,7 @@ def _log_request(
     def _log_response(self, response: http.HTTPResponse, response_body: str | bytes) -> None:
         if not logger.is_debugging():
             return
-        if self._suppress_oauth_response_logging:
-            logger.debug(f"Response: {response.status}")
-        else:
-            logger.debug(f"Response: {response.status} {response.reason}")
+        logger.debug(f"Response: {response.status}")
         logger.debug("Headers:")
         if self._suppress_oauth_response_logging:
             logger.debug("--> <OAuth response headers omitted>")
@@ -737,10 +755,7 @@ def _do_request(
             socket.gaierror,
             socket.timeout,
         ) as exception:
-            if self._suppress_oauth_response_logging:
-                logger.debug("An exception occurred processing the HTTP request (%s)." % type(exception).__name__)
-            else:
-                logger.debug("An exception occurred processing the HTTP request.", exc_info=True)
+            logger.debug("An exception occurred processing the HTTP request (%s)." % type(exception).__name__)
             return HTTPResponse(full_uri, exception=exception)
 
     def _follow_redirect(
@@ -789,19 +804,12 @@ def _tweak_response(self, response: HTTPResponse) -> JsonData | HTTPResponse:
         return response
 
     def _handle_set_cookie(self, response: http.HTTPResponse):
-        if self._suppress_oauth_response_logging:
-            self._cookies.store_cookies(response, suppress_logs=True)
-        else:
-            self._cookies.store_cookies(response)
+        self._cookies.store_cookies(response)
         self._inject_cookies()
 
     def _inject_cookies(self):
         if len(self._cookies) > 0:
-            self._headers["Cookie"] = (
-                self._cookies.get_cookie_header_value(suppress_logs=True)
-                if self._suppress_oauth_response_logging
-                else self._cookies.get_cookie_header_value()
-            )
+            self._headers["Cookie"] = self._cookies.get_cookie_header_value()
         elif "Cookie" in self._headers:
             del self._headers["Cookie"]
 
@@ -879,7 +887,7 @@ def __init__(self) -> None:
         self._content: dict[str, str] = {}
         self._reference = SimpleCookie()
 
-    def store_cookies(self, response: http.HTTPResponse, *, suppress_logs: bool = False):
+    def store_cookies(self, response: http.HTTPResponse):
         headers = filter(lambda h: h[0].lower() == "set-cookie", response.getheaders())
 
         for header in headers:
@@ -888,17 +896,12 @@ def store_cookies(self, response: http.HTTPResponse, *, suppress_logs: bool = Fa
                 if morsel.key not in self._keys:
                     self._keys.append(morsel.key)
                 self._content[morsel.key] = morsel.value
-                # Cookies are session credentials; names only, like the header log.
-                if not suppress_logs:
-                    logger.debug(f"--> Set cookie {morsel.key}: <redacted>")
 
-        if not suppress_logs:
-            logger.debug(f"CookieJar contents: {self._keys}")
+        logger.debug("CookieJar contains %d cookie(s).", len(self._keys))
 
-    def get_cookie_header_value(self, *, suppress_logs: bool = False):
+    def get_cookie_header_value(self):
         result = "; ".join([f"{key}={self._reference.value_encode(self._content[key])[1]}" for key in self._keys])
-        if not suppress_logs:
-            logger.debug(f"Cookie: {'; '.join(f'{key}=<redacted>' for key in self._keys)}")
+        logger.debug("Sending Cookie header with %d cookie(s).", len(self._keys))
         return result
 
     def as_dict(self):
diff --git a/rsconnect/metadata.py b/rsconnect/metadata.py
index 0905229f..0b6506e6 100644
--- a/rsconnect/metadata.py
+++ b/rsconnect/metadata.py
@@ -12,6 +12,8 @@
 import shutil
 import stat
 import sys
+import tempfile
+from contextlib import suppress
 from datetime import datetime, timezone
 from io import BufferedWriter
 from os.path import abspath, basename, dirname, exists, join
@@ -55,6 +57,8 @@
 # App deployment history is small; strict reads cap it at 1 MiB.
 _MAX_METADATA_BYTES = 1024 * 1024
 
+_DEFAULT_OPEN = open
+
 
 def resolve_server_alias(url: str) -> str:
     """Translate a --server value into the URL a saved server is stored under.
@@ -104,6 +108,13 @@ def makedirs(filepath: str):
         pass
 
 
+def _chmod_private_file(descriptor: int, path: str) -> None:
+    if hasattr(os, "fchmod"):
+        os.fchmod(descriptor, 0o600)
+    else:
+        os.chmod(path, 0o600)
+
+
 def _normalize_server_url(server_url: str):
     url = urlparse(server_url)
     return url.netloc.replace(".", "_").replace(":", "_")
@@ -291,13 +302,34 @@ def get_path(self):
     # noinspection PyShadowingBuiltins
     def save_to(self, path: str, data: bytes, open: Callable[..., BufferedWriter] = open):
         """
-        Save our data to the specified file.
+        Save our data to the specified file. Protected stores use an atomic write
+        with the default opener; supplied openers keep the direct destination-path
+        write behavior.
         """
-        with open(path, "wb") as f:
-            f.write(data)
+        if self._chmod and open is _DEFAULT_OPEN:
+            descriptor, temporary_path = tempfile.mkstemp(prefix=".%s." % basename(path), dir=dirname(path) or ".")
+            try:
+                with os.fdopen(descriptor, "wb") as stream:
+                    descriptor = -1
+                    _chmod_private_file(stream.fileno(), temporary_path)
+                    stream.write(data)
+                    stream.flush()
+                    os.fsync(stream.fileno())
+                os.replace(temporary_path, path)
+                temporary_path = ""
+            finally:
+                with suppress(OSError):
+                    os.close(descriptor)
+                with suppress(OSError):
+                    os.unlink(temporary_path)
+        else:
+            with open(path, "wb") as f:
+                f.write(data)
+            if self._chmod:
+                os.chmod(path, 0o600)
         self._real_path = path
 
-    def _already_holds(self, path: str, data: bytes) -> bool:
+    def _already_holds(self, path: str, data: bytes, opener: Callable[..., BufferedWriter] = open) -> bool:
         """
         Whether this store was loaded from `path` and that file already holds `data`.
 
@@ -306,6 +338,9 @@ def _already_holds(self, path: str, data: bytes) -> bool:
         """
         if self._real_path != path or not exists(path):
             return False
+        # Default protected writes must replace symlinks even when their target matches.
+        if self._chmod and opener is _DEFAULT_OPEN and os.path.islink(path):
+            return False
         try:
             with open(path, "rb") as f:
                 return f.read() == data
@@ -323,18 +358,22 @@ def save(self, open: Callable[..., BufferedWriter] = open):
         A save that would not change the file is skipped.
         """
         data = json.dumps(self._data, indent=4).encode("utf-8")
-        if not self._already_holds(self._primary_path, data):
-            try:
-                makedirs(self._primary_path)
-                self.save_to(self._primary_path, data, open)
-            except OSError:
-                if not self._secondary_path:
-                    raise
-                makedirs(self._secondary_path)
-                self.save_to(self._secondary_path, data, open)
-
-            if self._chmod and self._real_path is not None:
-                os.chmod(self._real_path, 0o600)
+        if self._already_holds(self._primary_path, data, open):
+            if self._chmod and open is _DEFAULT_OPEN:
+                metadata = os.lstat(self._primary_path)
+                # Avoid chmod following a symlink to an unrelated target.
+                if stat.S_ISREG(metadata.st_mode) and stat.S_IMODE(metadata.st_mode) != 0o600:
+                    os.chmod(self._primary_path, 0o600)
+            return
+
+        try:
+            makedirs(self._primary_path)
+            self.save_to(self._primary_path, data, open)
+        except OSError:
+            if not self._secondary_path:
+                raise
+            makedirs(self._secondary_path)
+            self.save_to(self._secondary_path, data, open)
 
 
 class ServerDataDict(TypedDict):
diff --git a/rsconnect/oauth.py b/rsconnect/oauth.py
index af097cbd..9373038a 100644
--- a/rsconnect/oauth.py
+++ b/rsconnect/oauth.py
@@ -20,7 +20,7 @@
 import click
 
 from .exception import RSConnectException
-from .http_support import HTTPResponse, HTTPServer
+from .http_support import HTTPResponse, HTTPServer, _redacted_uri_for_log
 from .log import logger
 
 # pyright: reportMissingTypeStubs=false
@@ -28,6 +28,29 @@
 _KEYRING_SERVICE = "rsconnect-python"
 _CLIENT_NAME = "rsconnect-python"
 _CALLBACK_TIMEOUT_SECONDS = 600
+_SAFE_OAUTH_ERROR_CODES = frozenset(
+    {
+        "access_denied",
+        "authorization_pending",
+        "expired_token",
+        "invalid_client",
+        "invalid_grant",
+        "invalid_request",
+        "invalid_scope",
+        "invalid_target",
+        "no_code",
+        "server_error",
+        "slow_down",
+        "temporarily_unavailable",
+        "unauthorized_client",
+        "unsupported_grant_type",
+        "unsupported_response_type",
+    }
+)
+
+
+def _safe_oauth_error_code(error: str) -> str:
+    return error if error in _SAFE_OAUTH_ERROR_CODES else "unknown_error"
 
 
 class InvalidClientError(RSConnectException):
@@ -47,8 +70,7 @@ class InvalidGrantError(RSConnectException):
 
     def __init__(self, description: Optional[str] = None) -> None:
         self.description = description
-        detail = f": {description}" if description else "."
-        super().__init__(f"The OAuth grant is invalid, expired, or has been revoked{detail}")
+        super().__init__("The OAuth grant is invalid, expired, or has been revoked.")
 
 
 def _check_oauth_error_response(response: HTTPResponse) -> None:
@@ -61,7 +83,7 @@ def _check_oauth_error_response(response: HTTPResponse) -> None:
         if error == "invalid_grant":
             raise InvalidGrantError(description or None)
         if description or error:
-            raise RSConnectException(f"OAuth error: {description or error}")
+            raise RSConnectException(f"OAuth error: {_safe_oauth_error_code(error)}")
 
 
 def _unwrap_json_response(response: Any) -> dict[str, Any]:
@@ -106,12 +128,10 @@ def _post_oauth_form_request(
     ca_data: Optional[str | bytes] = None,
     request_timeout: Optional[float] = None,
     request_deadline: Optional[float] = None,
-    *,
-    suppress_response_logging: bool = False,
 ) -> Any:
     """POST a form and return the raw response so device polling can inspect pending errors."""
     server = HTTPServer(base_url, disable_tls_check=insecure, ca_data=ca_data)
-    server._suppress_oauth_response_logging = suppress_response_logging
+    server._suppress_oauth_response_logging = True
     if request_timeout is not None:
         server.request_timeout = request_timeout
     if request_deadline is not None:
@@ -131,8 +151,6 @@ def discover_oauth_metadata(
     ca_data: Optional[str | bytes] = None,
     request_timeout: Optional[float] = None,
     request_deadline: Optional[float] = None,
-    *,
-    suppress_response_logging: bool = False,
 ) -> dict[str, Any]:
     """Fetch OAuth 2.0 Authorization Server Metadata (RFC 8414).
 
@@ -140,7 +158,7 @@ def discover_oauth_metadata(
     the server does not support OAuth.
     """
     server = HTTPServer(url, disable_tls_check=insecure, ca_data=ca_data)
-    server._suppress_oauth_response_logging = suppress_response_logging
+    server._suppress_oauth_response_logging = True
     server.request_timeout = request_timeout
     server.request_deadline = request_deadline
     with server:
@@ -170,8 +188,6 @@ def register_client(
     ca_data: Optional[str | bytes] = None,
     request_timeout: Optional[float] = None,
     request_deadline: Optional[float] = None,
-    *,
-    suppress_response_logging: bool = False,
 ) -> str:
     """Register an OAuth client via Dynamic Client Registration (RFC 7591).
 
@@ -190,7 +206,7 @@ def register_client(
         grant_types.append("urn:ietf:params:oauth:grant-type:device_code")
 
     server = HTTPServer(base, disable_tls_check=insecure, ca_data=ca_data)
-    server._suppress_oauth_response_logging = suppress_response_logging
+    server._suppress_oauth_response_logging = True
     server.request_timeout = request_timeout
     server.request_deadline = request_deadline
     with server:
@@ -285,7 +301,7 @@ def do_GET(self) -> None:  # noqa: N802
             self.result_queue.put(("error", "no_code", "No authorization code in callback"))
 
     def log_message(self, format: str, *args: object) -> None:
-        logger.debug(f"OAuth callback server: {format % args}")
+        logger.debug("OAuth callback server handled a request.")
 
 
 def login_with_browser(
@@ -346,7 +362,7 @@ def login_with_browser(
 
     result = result_queue.get_nowait()
     if result[0] == "error":
-        raise RSConnectException(f"OAuth authentication failed: {result[1]} — {result[2]}")
+        raise RSConnectException(f"OAuth authentication failed: {_safe_oauth_error_code(result[1] or '')}")
 
     _, code, returned_state = result
     if returned_state != state:
@@ -466,8 +482,7 @@ def _poll_for_device_token(
         elif error == "access_denied":
             raise RSConnectException("Authorization was denied by the user.")
         elif error:
-            description = str(json_data.get("error_description", error))
-            raise RSConnectException(f"Device code flow failed: {description}")
+            raise RSConnectException(f"Device code flow failed: {_safe_oauth_error_code(error)}")
         else:
             raise RSConnectException("Device code token request returned an unexpected response.")
 
@@ -483,8 +498,6 @@ def refresh_access_token(
     scope: Optional[str] = None,
     request_timeout: Optional[float] = None,
     request_deadline: Optional[float] = None,
-    *,
-    suppress_response_logging: bool = False,
 ) -> dict[str, Any]:
     """Refresh an OAuth access token using a refresh token.
 
@@ -505,8 +518,6 @@ def refresh_access_token(
         request_options["request_timeout"] = request_timeout
     if request_deadline is not None:
         request_options["request_deadline"] = request_deadline
-    if suppress_response_logging:
-        request_options["suppress_response_logging"] = True
     data = _post_token_request(str(metadata["token_endpoint"]), params, insecure, ca_data, **request_options)
     if "access_token" not in data:
         raise RSConnectException("Token refresh returned an unexpected response.")
@@ -550,14 +561,11 @@ def _post_token_request(
     ca_data: Optional[str | bytes] = None,
     request_timeout: Optional[float] = None,
     request_deadline: Optional[float] = None,
-    *,
-    suppress_response_logging: bool = False,
 ) -> dict[str, Any]:
     """POST a form-encoded request to an OAuth token endpoint and return the JSON body."""
     parsed = urlparse(token_endpoint)
     base = f"{parsed.scheme}://{parsed.netloc}"
 
-    logging_options = {"suppress_response_logging": True} if suppress_response_logging else {}
     response = _post_oauth_form_request(
         base,
         parsed.path,
@@ -566,7 +574,6 @@ def _post_token_request(
         ca_data,
         request_timeout,
         request_deadline,
-        **logging_options,
     )
 
     return _unwrap_json_response(response)
@@ -623,6 +630,7 @@ def exchange_token_for_api_key(
     ).encode("utf-8")
 
     server = HTTPServer(base, disable_tls_check=insecure, ca_data=ca_data)
+    server._suppress_oauth_response_logging = True
     with server:
         response = server.request(
             "POST",
@@ -635,7 +643,10 @@ def exchange_token_for_api_key(
         raise RSConnectException("Unexpected response from the OIDC token exchange.")
 
     if response.exception:
-        raise RSConnectException("Could not connect to %s - %s" % (url, response.exception), cause=response.exception)
+        raise RSConnectException(
+            "Could not connect to %s (%s)" % (_redacted_uri_for_log(url), type(response.exception).__name__),
+            cause=response.exception,
+        )
 
     status = response.status
     data = response.json_data if isinstance(response.json_data, dict) else {}
@@ -658,24 +669,21 @@ def _token_exchange_error(status: Optional[int], data: dict[str, Any]) -> RSConn
         lowered = description.lower()
         if "ambiguous" in lowered:
             return RSConnectException(
-                f"The identity token matched more than one service principal on Connect ({description}). "
+                "The identity token matched more than one service principal on Connect. "
                 "Resolve the duplicate access grants on the server, or authenticate with an API key."
             )
         if "verif" in lowered:
             return RSConnectException(
-                f"Connect could not verify the identity token ({description}). "
+                "Connect could not verify the identity token. "
                 "Check the server clock and the OIDC issuer configuration, or authenticate with an API key."
             )
         return RSConnectException(
-            f"Connect did not grant access for this identity token ({description or 'no match'}). "
+            "Connect did not grant access for this identity token. "
             "Confirm access has been configured for the target content and that the token's "
             "audience matches it, or authenticate with an API key."
         )
 
-    detail = error
-    if description:
-        detail = f"{error}: {description}" if error else description
-    suffix = f" ({detail})" if detail else ""
+    suffix = f" ({_safe_oauth_error_code(error)})" if error else ""
     return RSConnectException(f"OIDC token exchange failed (HTTP {status}){suffix}.")
 
 
diff --git a/tests/test_agent_cli_e2e.py b/tests/test_agent_cli_e2e.py
index cc779ee5..c922c5ce 100644
--- a/tests/test_agent_cli_e2e.py
+++ b/tests/test_agent_cli_e2e.py
@@ -8,6 +8,7 @@
 import stat
 import subprocess
 import sys
+import tarfile
 import threading
 import time
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -132,18 +133,24 @@ def do_GET(self) -> None:
             self._json(200, {"api_enabled": True, "installations": self.server.python_installations})  # type: ignore[attr-defined]
         elif path == "/__api__/v1/server_settings/nodejs":
             self._json(200, self.server.nodejs_settings)  # type: ignore[attr-defined]
-        elif path.startswith("/__api__/v1/content/"):
-            content_id = path.rsplit("/", 1)[-1]
-            status = self.server.content_statuses.get(content_id)  # type: ignore[attr-defined]
-            if status is not None and status != 200:
-                self._json(status, {"error": "content lookup failed"})
-            else:
-                self._json(200, {"py_version": "3.12.8", "node_version": "22.22.2"})
+        elif path.startswith("/__api__/v1/content"):
+            self._connect_content_response(path)
         elif path == "/v1/accounts":
             self._cloud_accounts(record)
         else:
             self._json(404, {"error": "not found"})
 
+    def _connect_content_response(self, path: str) -> None:
+        if path == "/__api__/v1/content":
+            self._json(200, [])
+            return
+        content_id = path.rsplit("/", 1)[-1]
+        status = self.server.content_statuses.get(content_id)  # type: ignore[attr-defined]
+        if status is not None and status != 200:
+            self._json(status, {"error": "content lookup failed"})
+        else:
+            self._json(200, {"py_version": "3.12.8", "node_version": "22.22.2"})
+
     def _cloud_accounts(self, record: dict[str, object]) -> None:
         with self.server.lock:  # type: ignore[attr-defined]
             status = self.server.account_statuses.pop(0) if self.server.account_statuses else 200  # type: ignore[attr-defined]
@@ -1540,6 +1547,121 @@ def test_cloud_finish_refresh_uses_the_client_id_saved_at_start(
     assert not _device_states(home, "cloud")
 
 
+@pytest.mark.parametrize("explicit_extra", [False, True])
+def test_pending_login_inside_project_is_excluded_from_manifest_and_bundle(
+    tmp_path: Path, local_http_server: _LocalHTTPServer, explicit_extra: bool
+) -> None:
+    project = tmp_path / "project-home"
+    environment = _cli_environment(project)
+    environment.pop("XDG_CONFIG_HOME", None)
+    (project / "app.py").write_text("app = object()\n", encoding="utf-8")
+    (project / "requirements.txt").write_text("", encoding="utf-8")
+    (project / "assets").mkdir()
+    (project / "assets" / "keep.txt").write_text("keep this content\n", encoding="utf-8")
+    started = _run_cli(["login", local_http_server.base_url, "--name", "inside-project", "--no-wait"], environment)
+    assert started.returncode == 0, _output(started)
+    state = _device_states(project, "connect")[0]
+    original_state = state.read_bytes()
+    arguments = ["write-manifest", "api", "--entrypoint", "app:app", "--exclude-renv", str(project)]
+    if explicit_extra:
+        arguments.append(str(state))
+
+    generated = _run_cli(arguments, environment)
+
+    assert generated.returncode == 0, _output(generated)
+    manifest_path = project / "manifest.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    assert {"app.py", "requirements.txt", "assets/keep.txt"} <= set(manifest["files"])
+    assert state.relative_to(project).as_posix() not in manifest["files"]
+    assert state.read_bytes() == original_state
+    output = tmp_path / "content.tar.gz"
+    packaged = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            "import shutil,sys; from rsconnect.bundle import make_manifest_bundle; "
+            "bundle=make_manifest_bundle(sys.argv[1]); "
+            "out=open(sys.argv[2],'wb'); shutil.copyfileobj(bundle,out); out.close(); bundle.close()",
+            str(manifest_path),
+            str(output),
+        ],
+        cwd=str(_REPO_ROOT),
+        env=environment,
+        capture_output=True,
+        text=True,
+        timeout=30,
+    )
+    assert packaged.returncode == 0, _output(packaged)
+    with tarfile.open(output, "r:gz") as bundle:
+        assert {"app.py", "requirements.txt", "assets/keep.txt", "manifest.json"} <= set(bundle.getnames())
+        assert state.relative_to(project).as_posix() not in bundle.getnames()
+        for member in bundle.getmembers():
+            if member.isfile():
+                file = bundle.extractfile(member)
+                assert file is not None
+                assert b"device-code-secret" not in file.read()
+
+
+def test_static_notebook_in_config_is_rejected_before_execution_or_upload(
+    tmp_path: Path, local_http_server: _LocalHTTPServer
+) -> None:
+    home = tmp_path / "home"
+    environment = _cli_environment(home)
+    started = _run_cli(
+        ["login", "--name", "static-guard", "--server", local_http_server.base_url, "--no-wait"],
+        environment,
+    )
+    assert started.returncode == 0, _output(started)
+    state = _device_states(home, "connect")[0]
+    notebook = state.parent / "restricted.ipynb"
+    marker = tmp_path / "notebook-executed"
+    notebook.write_text(
+        json.dumps(
+            {
+                "nbformat": 4,
+                "nbformat_minor": 5,
+                "metadata": {},
+                "cells": [
+                    {
+                        "cell_type": "code",
+                        "metadata": {},
+                        "execution_count": None,
+                        "outputs": [],
+                        "source": f"from pathlib import Path; Path({str(marker)!r}).write_text('executed')",
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    (notebook.parent / "requirements.txt").write_text("", encoding="utf-8")
+
+    deployed = _run_cli(
+        [
+            "deploy",
+            "notebook",
+            "--server",
+            local_http_server.base_url,
+            "--api-key",
+            "synthetic-api-key",
+            "--static",
+            "--new",
+            "--exclude-renv",
+            str(notebook),
+        ],
+        environment,
+    )
+
+    assert deployed.returncode == 1, _output(deployed)
+    assert "configuration directory" in _output(deployed)
+    assert not marker.exists()
+    assert state.exists()
+    assert not any(
+        record["method"] == "POST" and str(record["path"]).startswith("/__api__/")
+        for record in local_http_server.requests
+    )
+
+
 @pytest.mark.skipif(os.name != "posix", reason="POSIX FIFOs and symlinks are required")
 @pytest.mark.parametrize("runtime", ["python", "nodejs"])
 @pytest.mark.parametrize("record_kind", ["module-symlink", "manifest-symlink", "module-fifo"])
diff --git a/tests/test_api.py b/tests/test_api.py
index d435dddd..5e3c343f 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -54,6 +54,54 @@ def test_nodejs_settings_reports_permission_failure(self):
         with self.assertRaises(RSConnectException):
             client.nodejs_settings()
 
+    def test_http_failure_diagnostics_omit_untrusted_reason_and_exception_text(self):
+        from http.client import BadStatusLine
+
+        from rsconnect.http_support import HTTPResponse
+
+        server = RSConnectServer("https://connect.example.com", "api-key")
+        reason_secret = "reason-refresh-token"
+        response = HTTPResponse("https://connect.example.com/v1/user", body="")
+        response.status = 503
+        response.reason = reason_secret
+
+        with self.assertRaises(RSConnectException) as raised:
+            server.handle_bad_response(response, is_httpresponse=True)
+
+        self.assertIn("503", raised.exception.message)
+        self.assertNotIn(reason_secret, raised.exception.message)
+
+        exception_secret = "bad-status-refresh-token"
+        failure = BadStatusLine(exception_secret)
+        failed_response = HTTPResponse("https://connect.example.com/v1/user", exception=failure)
+        with self.assertRaises(RSConnectException) as raised:
+            server.handle_bad_response(failed_response, is_httpresponse=True)
+
+        self.assertNotIn(exception_secret, raised.exception.message)
+        self.assertIs(raised.exception.cause, failure)
+
+    def test_oauth_refresh_warning_omits_exception_text(self):
+        secret = "refresh-token-secret"
+        client = RSConnectClient(
+            RSConnectServer(
+                "https://connect.example.com",
+                None,
+                oauth_access_token="stale-access-token",
+                oauth_client_id="oauth-client",
+            )
+        )
+
+        with patch("rsconnect.oauth.keyring_get_tokens", return_value=(None, "refresh-token")):
+            with patch("rsconnect.oauth.discover_oauth_metadata", return_value={}):
+                with patch("rsconnect.oauth.refresh_access_token", side_effect=RSConnectException(secret)):
+                    with self.assertLogs("rsconnect", level="WARNING") as captured:
+                        self.assertFalse(client._attempt_token_refresh())
+
+        log_text = "\n".join(captured.output)
+        self.assertIn("OAuth token refresh failed", log_text)
+        self.assertIn("RSConnectException", log_text)
+        self.assertNotIn(secret, log_text)
+
     def test_executor_init(self):
         connect_server = require_connect()
         api_key = require_api_key()
@@ -198,8 +246,9 @@ def test_verify_api_key_connection_error(self):
         with patch.object(RSConnectClient, "get", return_value=failed_response):
             with self.assertRaises(RSConnectException) as cm:
                 ce.verify_api_key()
-        self.assertEqual(str(cm.exception), "Could not connect to http://test-server/ - connection refused")
+        self.assertEqual(str(cm.exception), "Could not connect to http://test-server/ (OSError)")
         self.assertIs(cm.exception.cause, failed_response.exception)
+        self.assertNotIn("connection refused", str(cm.exception))
 
     # The deprecated module-level verify_api_key() is reached via actions.test_api_key()
     # during `rsconnect add`, so it must accept the same credentials as the executor path.
@@ -553,8 +602,8 @@ def test_exchange_token_error_status(self, mock_fmt_payload, mock_token_endpoint
         mock_response = Mock()
         mock_response.status = 401
         mock_response.exception = None
-        mock_response.full_uri = "https://example.snowflakecomputing.com/oauth/token"
-        mock_response.reason = "Unauthorized"
+        mock_response.full_uri = "https://example.snowflakecomputing.com/oauth/token?refresh_token=uri-secret"
+        mock_response.reason = "reason-secret"
         mock_server_instance.request.return_value = mock_response
 
         # Mock the token endpoint and payload
@@ -566,13 +615,11 @@ def test_exchange_token_error_status(self, mock_fmt_payload, mock_token_endpoint
         }
 
         # Call the method and verify it raises the expected exception
-        with pytest.raises(RSConnectException) as raised:
+        with pytest.raises(RSConnectException, match="Failed to exchange Snowflake token") as raised:
             server.exchange_token()
-        self.assertEqual(
-            raised.value.message,
-            "Failed to exchange Snowflake token: Received an unexpected response from "
-            "https://spcs.example.com (calling https://example.snowflakecomputing.com/oauth/token): 401 Unauthorized",
-        )
+        self.assertIs(mock_server_instance._suppress_oauth_response_logging, True)
+        self.assertNotIn("reason-secret", raised.value.message)
+        self.assertNotIn("uri-secret", raised.value.message)
 
     @patch("rsconnect.api.HTTPServer")
     @patch("rsconnect.api.SPCSConnectServer.token_endpoint")
diff --git a/tests/test_bundle.py b/tests/test_bundle.py
index 15d91548..50ce6f81 100644
--- a/tests/test_bundle.py
+++ b/tests/test_bundle.py
@@ -782,9 +782,13 @@ def test_keep_manifest_specified_file(self):
         self.assertFalse(keep_manifest_specified_file("rsconnect/bogus.file"))
         self.assertFalse(keep_manifest_specified_file("rsconnect-python"))
         self.assertFalse(keep_manifest_specified_file("rsconnect-python/bogus.file"))
+        self.assertFalse(keep_manifest_specified_file(".rsconnect-python/servers.json"))
+        self.assertTrue(keep_manifest_specified_file("nested/rsconnect-python/app.json"))
+        self.assertFalse(keep_manifest_specified_file("nested/.rsconnect-python/pending.json"))
         self.assertFalse(keep_manifest_specified_file(".svn/bogus.file"))
         # noinspection SpellCheckingInspection
         self.assertFalse(keep_manifest_specified_file(".Rproj.user/bogus.file"))
+        self.assertTrue(keep_manifest_specified_file("nested/node_modules/local-data.js"))
 
     def test_manifest_bundle(self):
         # noinspection SpellCheckingInspection
@@ -3245,6 +3249,329 @@ def _make_node_env(**overrides):
     return NodeEnvironment(**defaults)
 
 
+def _make_test_environment():
+    return Environment.from_dict(
+        {
+            "contents": "flask==3.0.0\n",
+            "filename": "requirements.txt",
+            "locale": "en_US.UTF-8",
+            "package_manager": "pip",
+            "pip": "23.0.1",
+            "python": "3.12.0",
+            "source": "file",
+        }
+    )
+
+
+def _bundle_snapshot(bundle_file):
+    with tarfile.open(mode="r:gz", fileobj=bundle_file) as tar:
+        members = {name: tar.extractfile(name).read() for name in tar.getnames()}
+    return members, json.loads(members["manifest.json"].decode("utf-8"))
+
+
+def test_node_bundle_excludes_config_paths_and_explicit_extras(tmp_path, monkeypatch):
+    project = tmp_path / "project"
+    project.mkdir()
+    config_dir = project / "nested" / "custom-state"
+    config_dir.mkdir(parents=True)
+    monkeypatch.setattr(rsconnect.bundle.metadata, "config_dirname", lambda: str(config_dir))
+
+    (project / "app.js").write_text("// app")
+    (project / "package.json").write_text("{}")
+    (project / "data.csv").write_text("safe")
+    (project / "nested" / "keep.txt").write_text("safe")
+    nested_node_module = project / "feature" / "node_modules" / "kept.js"
+    nested_node_module.parent.mkdir(parents=True)
+    nested_node_module.write_text("safe")
+
+    secret_files = [
+        project / ".rsconnect-python" / "servers.json",
+        project / "nested" / ".rsconnect-python" / "pending.json",
+        config_dir / "servers.json",
+        config_dir / "pending.json",
+    ]
+    for secret_file in secret_files:
+        secret_file.parent.mkdir(parents=True, exist_ok=True)
+        secret_file.write_text("SECRET")
+    project_metadata = project / "nested" / "rsconnect-python" / "application.json"
+    project_metadata.parent.mkdir(parents=True)
+    project_metadata.write_text("project metadata")
+
+    extras = [
+        "data.csv",
+        ".rsconnect-python/servers.json",
+        "nested/.rsconnect-python/pending.json",
+        "nested/rsconnect-python/application.json",
+        "nested/custom-state/servers.json",
+    ]
+    try:
+        (project / "config-alias").symlink_to(config_dir, target_is_directory=True)
+    except (NotImplementedError, OSError):
+        pass
+    else:
+        extras.append("config-alias/pending.json")
+
+    bundle_file = make_nodejs_bundle(str(project), "app.js", _make_node_env(), extras, [])
+    members, manifest = _bundle_snapshot(bundle_file)
+    names = set(members)
+
+    expected = {
+        "manifest.json",
+        "app.js",
+        "package.json",
+        "data.csv",
+        "nested/keep.txt",
+        "nested/rsconnect-python/application.json",
+        "feature/node_modules/kept.js",
+    }
+    assert names == expected
+    assert set(manifest["files"]) == expected - {"manifest.json"}
+    assert b"SECRET" not in b"".join(members.values())
+
+
+def test_manifest_bundle_removes_config_entries_only(tmp_path, monkeypatch):
+    project = tmp_path / "project"
+    project.mkdir()
+    config_dir = project / "state" / "custom-store"
+    config_dir.mkdir(parents=True)
+    monkeypatch.setattr(rsconnect.bundle.metadata, "config_dirname", lambda: str(config_dir))
+
+    (project / "app.py").write_text("# app")
+    nested_node_module = project / "feature" / "node_modules" / "kept.js"
+    nested_node_module.parent.mkdir(parents=True)
+    nested_node_module.write_text("safe")
+    ignored_file = project / "packrat" / "packrat.lock"
+    ignored_file.parent.mkdir()
+    ignored_file.write_text("ignored by existing rules")
+
+    secret_paths = [
+        ".rsconnect-python/servers.json",
+        "nested/.rsconnect-python/pending.json",
+        "state/custom-store/servers.json",
+    ]
+    for relative_path in secret_paths:
+        secret_file = project / relative_path
+        secret_file.parent.mkdir(parents=True, exist_ok=True)
+        secret_file.write_text("SECRET")
+    project_metadata = project / "nested" / "rsconnect-python" / "application.json"
+    project_metadata.parent.mkdir(parents=True)
+    project_metadata.write_text("project metadata")
+
+    files = {
+        "app.py": {"checksum": "app"},
+        "nested/rsconnect-python/application.json": {"checksum": "project-metadata"},
+        "feature/node_modules/kept.js": {"checksum": "nested-node-module"},
+        "packrat/packrat.lock": {"checksum": "existing-ignore"},
+        **{path: {"checksum": "secret"} for path in secret_paths},
+    }
+    try:
+        (project / "config-alias").symlink_to(config_dir, target_is_directory=True)
+    except (NotImplementedError, OSError):
+        pass
+    else:
+        files["config-alias/servers.json"] = {"checksum": "secret-alias"}
+
+    manifest_path = project / "manifest.json"
+    manifest_path.write_text(json.dumps({"version": 1, "metadata": {"appmode": "python-api"}, "files": files}))
+    with make_manifest_bundle(manifest_path) as bundle_file:
+        members, manifest = _bundle_snapshot(bundle_file)
+    names = set(members)
+
+    assert names == {
+        "manifest.json",
+        "app.py",
+        "nested/rsconnect-python/application.json",
+        "feature/node_modules/kept.js",
+    }
+    assert set(manifest["files"]) == {
+        "app.py",
+        "nested/rsconnect-python/application.json",
+        "feature/node_modules/kept.js",
+        "packrat/packrat.lock",
+    }
+    assert not any(".rsconnect-python" in name for name in manifest["files"])
+    assert "state/custom-store/servers.json" not in manifest["files"]
+    assert "config-alias/servers.json" not in manifest["files"]
+    assert b"SECRET" not in b"".join(members.values())
+
+
+def test_notebook_source_bundle_filters_config_extras(tmp_path, monkeypatch):
+    project = tmp_path / "project"
+    project.mkdir()
+    config_dir = project / "nested" / "custom-state"
+    config_dir.mkdir(parents=True)
+    monkeypatch.setattr(rsconnect.bundle.metadata, "config_dirname", lambda: str(config_dir))
+
+    notebook = project / "notebook.ipynb"
+    notebook.write_text("{}")
+    (project / "data.csv").write_text("safe")
+    (config_dir / "servers.json").write_text("SECRET")
+    (project / ".rsconnect-python").mkdir()
+    (project / ".rsconnect-python" / "pending.json").write_text("SECRET")
+    extras = ["data.csv", "nested/custom-state/servers.json", ".rsconnect-python/pending.json"]
+    try:
+        (project / "config-alias").symlink_to(config_dir, target_is_directory=True)
+    except (NotImplementedError, OSError):
+        pass
+    else:
+        extras.append("config-alias/servers.json")
+
+    environment = _make_test_environment()
+    with make_notebook_source_bundle(
+        str(notebook),
+        environment,
+        extras,
+        hide_all_input=False,
+        hide_tagged_input=False,
+    ) as bundle_file:
+        members, manifest = _bundle_snapshot(bundle_file)
+    names = set(members)
+
+    assert names == {"manifest.json", "notebook.ipynb", "requirements.txt", "data.csv"}
+    assert set(manifest["files"]) == names - {"manifest.json"}
+    assert b"SECRET" not in b"".join(members.values())
+
+
+@pytest.mark.parametrize("path_kind", ["active_config", "reserved_path", "symlink_alias"])
+def test_notebook_html_bundle_rejects_config_paths_before_conversion(tmp_path, monkeypatch, path_kind):
+    project = tmp_path / "project"
+    project.mkdir()
+    if path_kind == "active_config":
+        config_dir = project / "custom-state"
+        notebook = config_dir / "private.ipynb"
+        error_match = "inside the rsconnect-python configuration directory"
+    elif path_kind == "reserved_path":
+        config_dir = tmp_path / "user-config"
+        notebook = project / ".rsconnect-python" / "private.ipynb"
+        error_match = "credential configuration file"
+    else:
+        config_dir = tmp_path / "user-config"
+        notebook = project / "private.ipynb"
+        error_match = "credential configuration file"
+
+    config_dir.mkdir(parents=True)
+    monkeypatch.setattr(rsconnect.bundle.metadata, "config_dirname", lambda: str(config_dir))
+    if path_kind == "symlink_alias":
+        target = config_dir / "private.ipynb"
+        target.write_text("{}")
+        try:
+            notebook.symlink_to(target)
+        except (NotImplementedError, OSError):
+            pytest.skip("symlinks are unavailable")
+    else:
+        notebook.parent.mkdir(parents=True, exist_ok=True)
+        notebook.write_text("{}")
+
+    callback_calls = []
+
+    def check_output(command):
+        callback_calls.append(command)
+        return b"Python 3.12"
+
+    with pytest.raises(RSConnectException, match=error_match):
+        make_notebook_html_bundle(
+            str(notebook),
+            sys.executable,
+            hide_all_input=False,
+            hide_tagged_input=True,
+            check_output=check_output,
+        )
+    assert callback_calls == []
+
+
+def test_standalone_quarto_bundle_filters_config_extras(tmp_path, monkeypatch):
+    project = tmp_path / "project"
+    project.mkdir()
+    config_dir = project / "nested" / "custom-state"
+    config_dir.mkdir(parents=True)
+    monkeypatch.setattr(rsconnect.bundle.metadata, "config_dirname", lambda: str(config_dir))
+
+    document = project / "report.qmd"
+    document.write_text("# report")
+    (project / "data.csv").write_text("safe")
+    (config_dir / "servers.json").write_text("SECRET")
+    extras = ["data.csv", "nested/custom-state/servers.json"]
+    try:
+        (project / "config-alias").symlink_to(config_dir, target_is_directory=True)
+    except (NotImplementedError, OSError):
+        pass
+    else:
+        extras.append("config-alias/servers.json")
+
+    inspect = {"quarto": {"version": "1.4.0"}, "engines": [], "files": {"input": [str(document)]}}
+    with make_quarto_source_bundle(
+        str(document),
+        inspect,
+        AppModes.STATIC_QUARTO,
+        None,
+        extras,
+        [],
+    ) as bundle_file:
+        members, manifest = _bundle_snapshot(bundle_file)
+    names = set(members)
+
+    assert names == {"manifest.json", "report.qmd", "data.csv"}
+    assert set(manifest["files"]) == names - {"manifest.json"}
+    assert b"SECRET" not in b"".join(members.values())
+
+
+def test_bundle_rejects_config_directory_as_content_root(tmp_path, monkeypatch):
+    config_dir = tmp_path / "custom-config"
+    config_dir.mkdir()
+    monkeypatch.setattr(rsconnect.bundle.metadata, "config_dirname", lambda: str(config_dir))
+    content_root = tmp_path / "config-alias"
+    try:
+        content_root.symlink_to(config_dir, target_is_directory=True)
+    except (NotImplementedError, OSError):
+        content_root = config_dir
+
+    with pytest.raises(RSConnectException, match="inside the rsconnect-python configuration directory"):
+        make_nodejs_bundle(str(content_root), "app.js", _make_node_env(), [], [])
+
+
+def test_write_api_manifest_excludes_current_home_config_extra(tmp_path, monkeypatch):
+    project = tmp_path / "project"
+    project.mkdir()
+    monkeypatch.setenv("HOME", str(project))
+    monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
+    monkeypatch.delenv("APPDATA", raising=False)
+
+    config_dir = Path(rsconnect.bundle.metadata.config_dirname()).resolve()
+    config_dir.mkdir(parents=True)
+    pending_state = config_dir / "pending.json"
+    pending_state.write_text("SECRET")
+    (project / "app.py").write_text("app = object()")
+    (project / "data.csv").write_text("safe")
+    (project / "requirements.txt").write_text("flask==3.0.0\n")
+    extra_files = [str(pending_state), str(project / "data.csv")]
+    environment = _make_test_environment()
+
+    assert write_api_manifest_json(
+        str(project),
+        "app:app",
+        environment,
+        AppModes.PYTHON_FASTAPI,
+        extra_files,
+        [],
+    )
+    manifest = json.loads((project / "manifest.json").read_text())
+    with make_manifest_bundle(project / "manifest.json") as bundle_file:
+        members, bundled_manifest = _bundle_snapshot(bundle_file)
+    names = set(members)
+
+    assert "app.py" in manifest["files"]
+    assert "data.csv" in manifest["files"]
+    assert all("pending.json" not in path for path in manifest["files"])
+    assert names == {"manifest.json", "app.py", "data.csv", "requirements.txt"}
+    assert set(bundled_manifest["files"]) == names - {"manifest.json"}
+    assert "app.py" in bundled_manifest["files"]
+    assert "data.csv" in bundled_manifest["files"]
+    assert all("pending.json" not in path for path in bundled_manifest["files"])
+    for path in ("app.py", "data.csv", "requirements.txt"):
+        assert bundled_manifest["files"][path]["checksum"] == file_checksum(project / path)
+    assert b"SECRET" not in b"".join(members.values())
+
+
 class TestNodeJSManifest:
     def test_manifest_structure(self):
         env = _make_node_env()
diff --git a/tests/test_connect_cloud.py b/tests/test_connect_cloud.py
index 2585f7de..a6de0cda 100644
--- a/tests/test_connect_cloud.py
+++ b/tests/test_connect_cloud.py
@@ -219,12 +219,6 @@ def test_refresh_can_limit_its_request_timeout(self):
             connect_cloud.refresh("rt", "production", request_timeout=0.25)
         self.assertEqual(refresh.call_args.kwargs["request_timeout"], 0.25)
 
-    def test_refresh_can_suppress_response_logging(self):
-        with mock.patch("rsconnect.connect_cloud.refresh_access_token") as refresh:
-            refresh.return_value = {"access_token": "new"}
-            connect_cloud.refresh("rt", "production", suppress_response_logging=True)
-        self.assertIs(refresh.call_args.kwargs["suppress_response_logging"], True)
-
     def test_refresh_honors_saved_client_override_and_deadline(self):
         with mock.patch.dict(
             os.environ,
@@ -833,21 +827,6 @@ def test_refresh_uses_the_remaining_finish_budget(self):
         )
         self.assertEqual(server.access_token, "fresh")
 
-    def test_private_refresh_suppresses_logging_and_limits_warning_details(self):
-        server = ConnectCloudServer("acme", access_token="stale", refresh_token="rt")
-        client = ConnectCloudClient(server)
-        client._suppress_oauth_response_logging = True
-        failure = RSConnectException("refresh response contained private details")
-
-        with mock.patch("rsconnect.connect_cloud.refresh", side_effect=failure) as refresh:
-            with self.assertLogs("rsconnect", level="WARNING") as captured:
-                self.assertFalse(client._attempt_token_refresh())
-
-        refresh.assert_called_once_with("rt", "production", suppress_response_logging=True)
-        log_text = "\n".join(captured.output)
-        self.assertIn("RSConnectException", log_text)
-        self.assertNotIn("refresh response contained private details", log_text)
-
     def test_expired_finish_budget_does_not_refresh(self):
         server = ConnectCloudServer("acme", access_token="stale", refresh_token="rt")
         client = ConnectCloudClient(server)
@@ -1062,7 +1041,8 @@ def test_a_transient_refresh_failure_leaves_the_original_401_and_warns(self):
         self.assertIn("401", exception.message)
         log_text = "\n".join(captured.output)
         self.assertIn("token refresh failed", log_text)
-        self.assertIn(secret, log_text)
+        self.assertIn("RSConnectException", log_text)
+        self.assertNotIn(secret, log_text)
         self.assertEqual(self._stored_entry()["connect_cloud_refresh_token"], "rt")
         self.assertEqual(len(httpretty.latest_requests()), 1)
 
diff --git a/tests/test_device_login.py b/tests/test_device_login.py
index b67759d5..a901e2cb 100644
--- a/tests/test_device_login.py
+++ b/tests/test_device_login.py
@@ -1007,7 +1007,7 @@ def test_bytes_ca_is_persisted_and_reused_during_finish(login_env: Any, monkeypa
             [RSConnectException("temporary account lookup failure"), {"id": "team-id"}],
             [],
             RSConnectException,
-            "account lookup failed",
+            "temporary account lookup failure",
         ),
         (
             [
@@ -1189,34 +1189,9 @@ def test_cloud_finish_refresh_uses_client_id_saved_at_start(login_env: Any, monk
         request_timeout=10.0,
         request_deadline=20.0,
         client_id_override="started-client",
-        suppress_response_logging=True,
     )
 
 
-def test_registration_error_is_private_and_can_be_retried(login_env: Any):
-    login_env.http.responses.extend(
-        [
-            _response(200, METADATA),
-            _response(503, {"error": "server_error", "error_description": "device-code-secret"}),
-        ]
-    )
-
-    with pytest.raises(RSConnectException, match="client registration failed") as failed:
-        device_login.start_connect_login(SERVER, "work")
-
-    assert "device-code-secret" not in str(failed.value)
-    assert not _state_path("connect", "work").exists()
-    assert login_env.http.instances[-1]._suppress_oauth_response_logging is True
-    login_env.http.responses.extend(
-        [
-            _response(200, METADATA),
-            _response(201, {"client_id": "registered-client"}),
-            _response(200, DEVICE_RESPONSE),
-        ]
-    )
-    assert device_login.start_connect_login(SERVER, "work")["status"] == "pending"
-
-
 def test_missing_cloud_account_removes_checkpoint_for_corrected_start(login_env: Any, monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setenv(connect_cloud.ENVIRONMENT_ENV_VAR, "production")
     _queue_device(login_env)
diff --git a/tests/test_git_metadata.py b/tests/test_git_metadata.py
index a4722a1e..44ff4d30 100644
--- a/tests/test_git_metadata.py
+++ b/tests/test_git_metadata.py
@@ -297,4 +297,6 @@ def test_upload_bundle_with_metadata_handles_bad_response(self):
         with patch.object(RSConnectClient, "post", return_value=bad_response):
             with pytest.raises(RSConnectException) as cm:
                 client.upload_bundle("guid", io.BytesIO(b"tarball"), metadata={"source": "git"})
-        assert "connection refused" in str(cm.value)
+        assert str(cm.value) == "Could not connect to http://test-server (OSError)"
+        assert cm.value.cause is bad_response.exception
+        assert "connection refused" not in str(cm.value)
diff --git a/tests/test_http_support.py b/tests/test_http_support.py
index 85c090c0..252a478c 100644
--- a/tests/test_http_support.py
+++ b/tests/test_http_support.py
@@ -312,29 +312,20 @@ def test_as_dict(self):
         )
 
     def test_cookie_values_do_not_reach_the_debug_log(self):
-        # Cookies are session credentials; the jar logs names only.
+        # Cookie names can carry credentials too; diagnostics expose only the count.
         jar = CookieJar()
+        cookie_name = "echoed-refresh-token-name"
+        cookie_value = "echoed-refresh-token-value"
         with self.assertLogs("rsconnect", level="DEBUG") as captured:
-            jar.store_cookies(FakeSetCookieResponse(["session=s3ssionv4lue"]))
+            jar.store_cookies(FakeSetCookieResponse([f"{cookie_name}={cookie_value}"]))
             header = jar.get_cookie_header_value()
 
-        self.assertEqual(header, "session=s3ssionv4lue")
-        log_text = "\n".join(captured.output)
-        self.assertNotIn("s3ssionv4lue", log_text)
-        self.assertIn("session", log_text)
-
-    def test_cookie_names_and_values_can_be_omitted_when_opted_in(self):
-        from unittest.mock import patch
-
-        jar = CookieJar()
-        cookie_name = "oauth-refresh-cookie-name"
-        cookie_value = "oauth-refresh-cookie-value"
-        with patch("rsconnect.http_support.logger.debug") as debug:
-            jar.store_cookies(FakeSetCookieResponse([f"{cookie_name}={cookie_value}"]), suppress_logs=True)
-            header = jar.get_cookie_header_value(suppress_logs=True)
-
-        debug.assert_not_called()
         self.assertEqual(header, f"{cookie_name}={cookie_value}")
+        self.assertEqual(jar.as_dict(), {"keys": [cookie_name], "content": {cookie_name: cookie_value}})
+        log_text = "\n".join(captured.output)
+        self.assertNotIn(cookie_name, log_text)
+        self.assertNotIn(cookie_value, log_text)
+        self.assertIn("1 cookie(s)", log_text)
 
 
 class TestDebugLogRedaction(TestCase):
@@ -416,6 +407,7 @@ def test_json_error_codes_stay_readable(self):
 
         body = '{"error": "An object with that name already exists.", "code": 26}'
         redacted = str(_redacted_body_for_log(body))
+        self.assertIn('"error": "An object with that name already exists."', redacted)
         self.assertIn('"code": 26', redacted)
 
     def test_streams_are_left_alone(self):
@@ -449,7 +441,72 @@ def test_cookie_values_with_spaces_leave_no_first_token(self):
 
         self.assertEqual(_redacted_header_for_log("Cookie", "session=abc; other=def"), "<redacted>")
 
-    def test_oauth_redirect_destination_is_suppressed_when_opted_in(self):
+    def test_oauth_error_fields_and_redirect_location_are_redacted_in_http_logs(self):
+        from unittest.mock import Mock, patch
+
+        location = (
+            "/callback?refresh%5Ftoken=redirect-secret"
+            "#access%5Ftoken=fragment-secret&error_description=location-secret&tab=summary"
+        )
+        body = '{"error":"invalid_grant","error_description":"description-secret","user_code":"code-secret"}'
+
+        def make_response(status, response_body, headers, reason):
+            response = Mock()
+            response.status = status
+            response.reason = reason
+            response.read.return_value = response_body
+            response.getheaders.return_value = headers
+            header_values = {key.lower(): value for key, value in headers}
+            response.getheader.side_effect = lambda key, default=None: header_values.get(key.lower(), default)
+            return response
+
+        redirect = make_response(302, b"", [("Location", location)], "reason-secret")
+        final = make_response(
+            200,
+            body.encode(),
+            [("Content-Type", "application/json"), ("X-Debug-Context", "ordinary-debug-context")],
+            "reason-secret",
+        )
+
+        with HTTPServer("http://example.com") as server:
+            transport = cast(Any, server._conn)
+            with self.assertLogs("rsconnect", level="DEBUG") as captured:
+                with patch.object(transport, "request") as send:
+                    with patch.object(transport, "getresponse", side_effect=[redirect, final]):
+                        response = cast(HTTPResponse, server.get("/start"))
+
+        log_text = "\n".join(captured.output)
+        for secret in (
+            "redirect-secret",
+            "location-secret",
+            "description-secret",
+            "code-secret",
+            "reason-secret",
+            "fragment-secret",
+        ):
+            self.assertNotIn(secret, log_text)
+        self.assertIn("Content-Type: application/json", log_text)
+        self.assertIn("X-Debug-Context: ordinary-debug-context", log_text)
+        self.assertIn(
+            "Redirected to: http://example.com/callback?refresh%5Ftoken=<redacted>"
+            "#access%5Ftoken=<redacted>&error_description=<redacted>&tab=summary",
+            log_text,
+        )
+        self.assertIn('"error_description": "<redacted>"', log_text)
+        self.assertIn('"user_code": "<redacted>"', log_text)
+        self.assertEqual(send.call_args_list[1].args[1], location)
+        self.assertEqual(response.reason, "reason-secret")
+        self.assertEqual(response.response_body, body)
+        self.assertEqual(
+            response.json_data,
+            {
+                "error": "invalid_grant",
+                "error_description": "description-secret",
+                "user_code": "code-secret",
+            },
+        )
+
+    def test_oauth_redirect_destination_is_suppressed_without_changing_routing(self):
         from unittest.mock import Mock, patch
 
         location = "http://example.com/opaque/path-refresh-token?resume=opaque-query-token"
@@ -486,24 +543,17 @@ def make_response(status, body, headers, reason):
         self.assertEqual(response.reason, "final-reason-refresh-token")
         self.assertEqual(response.response_body, "{}")
 
-    def test_oauth_response_body_is_omitted_when_opted_in(self):
+    def test_oauth_response_body_suppression_keeps_response_data_unchanged(self):
         from unittest.mock import Mock, patch
 
         secret = "echoed-refresh-token"
-        cookie_name = "oauth-cookie-name"
-        cookie_value = "oauth-cookie-value"
-        endpoint = "/oauth/token?refresh_token=endpoint-secret"
         body = f'{{"error":"{secret}","error_description":"{secret}"}}'
         content_type = f"application/json; debug={secret}"
         reply = Mock()
         reply.status = 503
         reply.reason = "reason-secret"
         reply.read.return_value = body.encode()
-        reply.getheaders.return_value = [
-            ("Content-Type", content_type),
-            ("X-Debug-Context", secret),
-            ("Set-Cookie", f"{cookie_name}={cookie_value}"),
-        ]
+        reply.getheaders.return_value = [("Content-Type", content_type), ("X-Debug-Context", secret)]
         header_values = {key.lower(): value for key, value in reply.getheaders.return_value}
         reply.getheader.side_effect = lambda key, default=None: header_values.get(key.lower(), default)
 
@@ -513,33 +563,38 @@ def test_oauth_response_body_is_omitted_when_opted_in(self):
             with self.assertLogs("rsconnect", level="DEBUG") as captured:
                 with patch.object(transport, "request"):
                     with patch.object(transport, "getresponse", return_value=reply):
-                        response = cast(HTTPResponse, server.get(endpoint))
+                        response = cast(HTTPResponse, server.get("/oauth/token"))
 
         log_text = "\n".join(captured.output)
         self.assertIn("Response: 503", log_text)
         self.assertIn("<OAuth response headers omitted>", log_text)
         self.assertIn("<OAuth response body omitted>", log_text)
-        for private_value in (secret, cookie_name, cookie_value, "endpoint-secret", "X-Debug-Context", "reason-secret"):
-            self.assertNotIn(private_value, log_text)
+        self.assertNotIn(secret, log_text)
+        self.assertNotIn("X-Debug-Context", log_text)
+        self.assertNotIn("reason-secret", log_text)
         self.assertEqual(response.reason, "reason-secret")
         self.assertEqual(response.content_type, content_type)
         self.assertEqual(response._response.getheader("Content-Type"), content_type)
         self.assertEqual(response._response.getheader("X-Debug-Context"), secret)
-        self.assertEqual(response._response.getheader("Set-Cookie"), f"{cookie_name}={cookie_value}")
         self.assertEqual(response.response_body, body)
         self.assertEqual(response.json_data, {"error": secret, "error_description": secret})
 
-    def test_default_oauth_response_logging_keeps_baseline_diagnostics(self):
+    def test_malformed_json_is_logged_as_a_placeholder(self):
         from unittest.mock import Mock, patch
 
-        body = '{"message":"ordinary response"}'
+        from rsconnect.http_support import _redacted_body_for_log
+
+        body = '{"error_description":"malformed-secret", "broken": "'
+        self.assertEqual(_redacted_body_for_log(body), "<invalid JSON>")
+
         reply = Mock()
-        reply.status = 200
-        reply.reason = "ordinary reason"
+        reply.status = 400
+        reply.reason = "Bad Request"
         reply.read.return_value = body.encode()
-        reply.getheaders.return_value = [("Content-Type", "application/json"), ("X-Debug-Context", "ordinary context")]
-        header_values = {key.lower(): value for key, value in reply.getheaders.return_value}
-        reply.getheader.side_effect = lambda key, default=None: header_values.get(key.lower(), default)
+        reply.getheaders.return_value = [("Content-Type", "application/json")]
+        reply.getheader.side_effect = lambda key, default=None: (
+            "application/json" if key.lower() == "content-type" else default
+        )
 
         with HTTPServer("http://example.com") as server:
             transport = cast(Any, server._conn)
@@ -549,82 +604,27 @@ def test_default_oauth_response_logging_keeps_baseline_diagnostics(self):
                         response = cast(HTTPResponse, server.get("/token"))
 
         log_text = "\n".join(captured.output)
-        self.assertIn("Response: 200 ordinary reason", log_text)
-        self.assertIn("X-Debug-Context: ordinary context", log_text)
-        self.assertIn("ordinary response", log_text)
-        self.assertNotIn("<OAuth response", log_text)
+        self.assertIn("<invalid JSON>", log_text)
+        self.assertNotIn("malformed-secret", log_text)
         self.assertEqual(response.response_body, body)
+        self.assertIsNone(response.json_data)
 
-    def test_default_redirect_logging_keeps_destination_visible(self):
-        from unittest.mock import Mock, patch
-
-        location = "http://example.com/reports/current?tab=summary"
-        request_target = "/reports/current?tab=summary"
-
-        def make_response(status, body, headers, reason):
-            response = Mock()
-            response.status = status
-            response.reason = reason
-            response.read.return_value = body
-            response.getheaders.return_value = headers
-            header_values = {key.lower(): value for key, value in headers}
-            response.getheader.side_effect = lambda key, default=None: header_values.get(key.lower(), default)
-            return response
-
-        redirect = make_response(302, b"", [("Location", location)], "Found")
-        final = make_response(
-            200,
-            b'{"message":"done"}',
-            [("Content-Type", "application/json")],
-            "OK",
-        )
-
-        with HTTPServer("http://example.com") as server:
-            transport = cast(Any, server._conn)
-            with self.assertLogs("rsconnect", level="DEBUG") as captured:
-                with patch.object(transport, "request") as send:
-                    with patch.object(transport, "getresponse", side_effect=[redirect, final]):
-                        response = cast(HTTPResponse, server.get("/start"))
-
-        log_text = "\n".join(captured.output)
-        self.assertIn("Location: " + location, log_text)
-        self.assertIn("Redirected to: " + location, log_text)
-        self.assertEqual(send.call_args_list[1].args[1], request_target)
-        self.assertEqual(response.response_body, '{"message":"done"}')
-
-    def test_default_http_failure_logging_keeps_exception_details(self):
+    def test_bad_status_line_text_is_not_logged_with_a_traceback(self):
+        from http.client import BadStatusLine
         from unittest.mock import patch
 
-        failure = OSError("ordinary transport detail")
+        failure = BadStatusLine("refresh-token-in-status-line")
         with HTTPServer("http://example.com") as server:
             transport = cast(Any, server._conn)
             with self.assertLogs("rsconnect", level="DEBUG") as captured:
                 with patch.object(transport, "request"):
                     with patch.object(transport, "getresponse", side_effect=failure):
-                        response = cast(HTTPResponse, server.get("/ordinary"))
-
-        log_text = "\n".join(captured.output)
-        self.assertIn("Traceback", log_text)
-        self.assertIn("ordinary transport detail", log_text)
-        self.assertIs(response.exception, failure)
-
-    def test_http_failure_logging_omits_exception_details_when_opted_in(self):
-        from unittest.mock import patch
-
-        failure = OSError("oauth transport detail")
-        with HTTPServer("http://example.com") as server:
-            server._suppress_oauth_response_logging = True
-            transport = cast(Any, server._conn)
-            with self.assertLogs("rsconnect", level="DEBUG") as captured:
-                with patch.object(transport, "request"):
-                    with patch.object(transport, "getresponse", side_effect=failure):
-                        response = cast(HTTPResponse, server.get("/oauth/token?access_token=endpoint-secret"))
+                        response = cast(HTTPResponse, server.get("/token"))
 
         log_text = "\n".join(captured.output)
-        self.assertIn("OSError", log_text)
-        self.assertNotIn("oauth transport detail", log_text)
-        self.assertNotIn("endpoint-secret", log_text)
+        self.assertNotIn("refresh-token-in-status-line", log_text)
         self.assertNotIn("Traceback", log_text)
+        self.assertIn("BadStatusLine", log_text)
         self.assertIs(response.exception, failure)
 
     def test_a_connection_failure_response_has_a_none_status(self):
@@ -658,6 +658,35 @@ def test_presigned_url_query_is_redacted(self):
         self.assertNotIn("AKIA", redacted)
         self.assertIn("X-Amz-Expires=300", redacted)
 
+    def test_encoded_query_names_are_redacted_without_rewriting_other_parameters(self):
+        from rsconnect.http_support import _redacted_uri_for_log
+
+        uri = "/path?keep=%2f+value&refresh%5Ftoken=echoed-secret&X-Amz%2dSignature=signature-secret&tail=a+b#section"
+        self.assertEqual(
+            _redacted_uri_for_log(uri),
+            "/path?keep=%2f+value&refresh%5Ftoken=<redacted>&X-Amz%2dSignature=<redacted>&tail=a+b#section",
+        )
+
+    def test_encoded_fragment_fields_are_redacted_with_and_without_a_query(self):
+        from rsconnect.http_support import _redacted_uri_for_log
+
+        for prefix in ("/callback", "/callback?keep=%2f+value"):
+            with self.subTest(prefix=prefix):
+                self.assertEqual(
+                    _redacted_uri_for_log(prefix + "#refresh%5Ftoken=fragment-secret&tab=a+b"),
+                    prefix + "#refresh%5Ftoken=<redacted>&tab=a+b",
+                )
+                self.assertEqual(_redacted_uri_for_log(prefix + "#ordinary-section"), prefix + "#ordinary-section")
+
+    def test_queryless_uri_text_and_fragment_credentials_remain_redacted(self):
+        from rsconnect.http_support import _redacted_uri_for_log
+
+        self.assertEqual(_redacted_uri_for_log("account token=plain-secret"), "account token=<redacted>")
+        self.assertEqual(
+            _redacted_uri_for_log("/callback#refresh_token=fragment-secret"),
+            "/callback#refresh_token=<redacted>",
+        )
+
     def test_azure_sas_sig_param_is_redacted(self):
         # Azure-style presigned URLs carry the signature in a bare "sig" param.
         from rsconnect.http_support import _redacted_uri_for_log
diff --git a/tests/test_metadata.py b/tests/test_metadata.py
index 4cc9f952..a08a2143 100644
--- a/tests/test_metadata.py
+++ b/tests/test_metadata.py
@@ -1,6 +1,7 @@
 import json
 import os
 import shutil
+import stat
 import subprocess
 import sys
 import tempfile
@@ -13,6 +14,7 @@
 from rsconnect.metadata import (
     AppStore,
     ContentBuildStore,
+    DataStore,
     ServerStore,
     _normalize_server_url,
     resolve_server_alias,
@@ -259,11 +261,240 @@ def test_remove_default_clears_it(self):
         self.assertIsNone(self.server_store.get_default())
 
     def test_save_skips_rewrite_when_file_unchanged(self):
+        with patch(
+            "rsconnect.metadata.tempfile.mkstemp",
+            side_effect=AssertionError("created a temporary file for unchanged data"),
+        ):
+            self.server_store.save()
+
         def fail_open(path_to_open, mode, *args, **kw):
             self.fail("rewrote %s when nothing had changed" % path_to_open)
 
         self.server_store.save(fail_open)
 
+    def test_private_store_creates_mode_600_file_under_umask_022(self):
+        if os.name != "posix":
+            self.skipTest("POSIX file permissions are unavailable")
+
+        base_dir = join(self.tempDir, "umask")
+        path = join(base_dir, "servers.json")
+        previous_umask = os.umask(0o022)
+        try:
+            store = ServerStore(base_dir=base_dir)
+            store.set("private", "https://connect.example.test", api_key="new-secret")
+        finally:
+            os.umask(previous_umask)
+
+        self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600)
+        self.assertEqual(store.get_path(), path)
+
+    def test_private_store_keeps_new_secret_off_existing_file_until_replace(self):
+        if os.name != "posix":
+            self.skipTest("POSIX file permissions are unavailable")
+
+        path = self.server_store_path
+        os.chmod(path, 0o644)
+        with open(path, "rb") as metadata_file:
+            previous_data = metadata_file.read()
+        new_secret = b"rotated-secret"
+        self.server_store._data["foo"]["api_key"] = new_secret.decode("ascii")
+        fdopen = os.fdopen
+        fchmod = os.fchmod
+        replace = os.replace
+
+        def check_private_fd(descriptor, mode):
+            self.assertEqual(stat.S_IMODE(os.fstat(descriptor).st_mode), 0o600)
+            return fdopen(descriptor, mode)
+
+        def check_fchmod(descriptor, mode):
+            self.assertEqual(mode, 0o600)
+            self.assertEqual(stat.S_IMODE(os.fstat(descriptor).st_mode), 0o600)
+            fchmod(descriptor, mode)
+
+        def check_atomic_replace(source, destination):
+            self.assertEqual(destination, path)
+            self.assertEqual(stat.S_IMODE(os.stat(source).st_mode), 0o600)
+            with open(source, "rb") as temporary_file:
+                self.assertIn(new_secret, temporary_file.read())
+            with open(destination, "rb") as existing_file:
+                self.assertEqual(existing_file.read(), previous_data)
+            replace(source, destination)
+
+        with patch("rsconnect.metadata.os.fdopen", side_effect=check_private_fd) as fdopen_mock, patch(
+            "rsconnect.metadata.os.fchmod", side_effect=check_fchmod
+        ) as fchmod_mock, patch("rsconnect.metadata.os.replace", side_effect=check_atomic_replace) as replace_mock:
+            self.server_store.save()
+
+        fdopen_mock.assert_called_once()
+        fchmod_mock.assert_called_once()
+        replace_mock.assert_called_once()
+        self.assertEqual(fchmod_mock.call_args[0][0], fdopen_mock.call_args[0][0])
+        self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600)
+        with open(path, "rb") as metadata_file:
+            self.assertIn(new_secret, metadata_file.read())
+
+    def test_fdopen_failure_closes_descriptor_removes_temp_and_preserves_file(self):
+        path = join(self.tempDir, "existing.json")
+        with open(path, "wb") as metadata_file:
+            metadata_file.write(b"old credentials")
+        previous_mode = stat.S_IMODE(os.stat(path).st_mode)
+        store = DataStore(path, chmod=True, autoload=False)
+        store._data = {"server": {"api_key": "new-secret"}}
+        files_before = set(os.listdir(self.tempDir))
+        mkstemp = tempfile.mkstemp
+        temporary = {}
+
+        def record_temp(*args, **kwargs):
+            descriptor, temporary_path = mkstemp(*args, **kwargs)
+            temporary.update(descriptor=descriptor, path=temporary_path)
+            return descriptor, temporary_path
+
+        with patch("rsconnect.metadata.tempfile.mkstemp", side_effect=record_temp), patch(
+            "rsconnect.metadata.os.fdopen", side_effect=OSError("fdopen failed")
+        ):
+            with self.assertRaisesRegex(OSError, "fdopen failed"):
+                store.save()
+
+        with self.assertRaises(OSError):
+            os.fstat(temporary["descriptor"])
+        self.assertFalse(os.path.exists(temporary["path"]))
+        self.assertEqual(set(os.listdir(self.tempDir)), files_before)
+        with open(path, "rb") as metadata_file:
+            self.assertEqual(metadata_file.read(), b"old credentials")
+        self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), previous_mode)
+        self.assertIsNone(store._real_path)
+
+    def test_unchanged_private_store_repairs_mode_without_rewriting(self):
+        if os.name != "posix":
+            self.skipTest("POSIX file permissions are unavailable")
+
+        path = self.server_store_path
+        os.chmod(path, 0o644)
+        before = os.stat(path)
+        with open(path, "rb") as metadata_file:
+            previous_data = metadata_file.read()
+
+        with patch(
+            "rsconnect.metadata.tempfile.mkstemp",
+            side_effect=AssertionError("rewrote unchanged credentials"),
+        ):
+            self.server_store.save()
+
+        after = os.stat(path)
+        self.assertEqual(stat.S_IMODE(after.st_mode), 0o600)
+        self.assertEqual(after.st_ino, before.st_ino)
+        with open(path, "rb") as metadata_file:
+            self.assertEqual(metadata_file.read(), previous_data)
+
+    def test_private_store_failure_preserves_file_and_real_path(self):
+        path = join(self.tempDir, "existing.json")
+        with open(path, "wb") as metadata_file:
+            metadata_file.write(b"old credentials")
+        previous_mode = stat.S_IMODE(os.stat(path).st_mode)
+        store = DataStore(path, chmod=True, autoload=False)
+        store._data = {"server": {"api_key": "new-secret"}}
+        files_before = set(os.listdir(self.tempDir))
+
+        with patch("rsconnect.metadata.os.fsync", side_effect=OSError("sync failed")):
+            with self.assertRaisesRegex(OSError, "sync failed"):
+                store.save()
+
+        with open(path, "rb") as metadata_file:
+            self.assertEqual(metadata_file.read(), b"old credentials")
+        self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), previous_mode)
+        self.assertIsNone(store._real_path)
+        self.assertEqual(set(os.listdir(self.tempDir)), files_before)
+
+    def test_private_store_falls_back_after_atomic_replace_failure(self):
+        primary = join(self.tempDir, "local", "servers.json")
+        secondary = join(self.tempDir, "config", "servers.json")
+        store = DataStore(primary, secondary, chmod=True, autoload=False)
+        store._data = {"server": {"api_key": "fallback-secret"}}
+        replace = os.replace
+
+        def fail_primary(source, destination):
+            if destination == primary:
+                raise OSError("primary unavailable")
+            replace(source, destination)
+
+        with patch("rsconnect.metadata.os.replace", side_effect=fail_primary):
+            store.save()
+
+        self.assertEqual(store.get_path(), secondary)
+        self.assertFalse(exists(primary))
+        self.assertEqual(set(os.listdir(os.path.dirname(primary))), set())
+        with open(secondary, "r", encoding="utf-8") as metadata_file:
+            self.assertEqual(json.load(metadata_file), store._data)
+
+    def test_private_store_replaces_symlink_without_writing_through(self):
+        if os.name != "posix":
+            self.skipTest("POSIX symlinks are unavailable")
+
+        path = self.server_store_path
+        target = join(self.tempDir, "servers-target.json")
+        os.replace(path, target)
+        os.chmod(target, 0o644)
+        with open(target, "rb") as metadata_file:
+            previous_data = metadata_file.read()
+        try:
+            os.symlink(target, path)
+        except OSError:
+            self.skipTest("symlinks are unavailable")
+
+        def fail_open(path_to_open, mode, *args, **kw):
+            self.fail("custom opener was called for unchanged symlink data")
+
+        self.server_store.save(fail_open)
+        self.assertTrue(os.path.islink(path))
+        with open(target, "rb") as metadata_file:
+            self.assertEqual(metadata_file.read(), previous_data)
+        self.assertEqual(stat.S_IMODE(os.stat(target).st_mode), 0o644)
+
+        self.server_store.save()
+
+        self.assertFalse(os.path.islink(path))
+        self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600)
+        with open(path, "rb") as metadata_file:
+            self.assertEqual(metadata_file.read(), previous_data)
+        with open(target, "rb") as metadata_file:
+            self.assertEqual(metadata_file.read(), previous_data)
+        self.assertEqual(stat.S_IMODE(os.stat(target).st_mode), 0o644)
+
+        self.server_store._data["foo"]["api_key"] = "symlink-secret"
+        self.server_store.save()
+
+        self.assertFalse(os.path.islink(path))
+        with open(target, "rb") as metadata_file:
+            self.assertEqual(metadata_file.read(), previous_data)
+        self.assertEqual(stat.S_IMODE(os.stat(target).st_mode), 0o644)
+        self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600)
+
+    def test_private_store_does_not_update_other_hardlinks(self):
+        if os.name != "posix":
+            self.skipTest("POSIX hard links are unavailable")
+
+        path = self.server_store_path
+        alias = join(self.tempDir, "servers-alias.json")
+        try:
+            os.link(path, alias)
+        except OSError:
+            self.skipTest("hard links are unavailable")
+        os.chmod(path, 0o644)
+        with open(path, "rb") as metadata_file:
+            previous_data = metadata_file.read()
+        self.server_store.save()
+        self.assertEqual(stat.S_IMODE(os.stat(alias).st_mode), 0o600)
+
+        self.server_store._data["foo"]["api_key"] = "hardlink-secret"
+        self.server_store.save()
+
+        with open(alias, "rb") as metadata_file:
+            alias_data = metadata_file.read()
+        self.assertEqual(alias_data, previous_data)
+        self.assertNotIn(b"hardlink-secret", alias_data)
+        self.assertEqual(stat.S_IMODE(os.stat(alias).st_mode), 0o600)
+        self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600)
+
     def test_save_rewrites_when_data_changed(self):
         writes = []
 
diff --git a/tests/test_oauth.py b/tests/test_oauth.py
index 66daaf54..a2d89745 100644
--- a/tests/test_oauth.py
+++ b/tests/test_oauth.py
@@ -13,6 +13,7 @@
 
 from .utils import failing_keyring
 from rsconnect.oauth import (
+    _CallbackHandler,
     InvalidClientError,
     InvalidGrantError,
     _exchange_code_for_token,
@@ -73,6 +74,7 @@ def test_success(self, mock_http_server: MagicMock):
         mock_http_server.get.return_value = _make_response(200, FAKE_METADATA)
         result = discover_oauth_metadata(FAKE_URL)
         assert result == FAKE_METADATA
+        assert mock_http_server._suppress_oauth_response_logging is True
 
     def test_client_specific_request_timeout_and_deadline(self, mock_http_server: MagicMock):
         mock_http_server.get.return_value = _make_response(200, FAKE_METADATA)
@@ -80,11 +82,6 @@ def test_client_specific_request_timeout_and_deadline(self, mock_http_server: Ma
         assert mock_http_server.request_timeout == 0.25
         assert mock_http_server.request_deadline == 12.5
 
-    def test_response_logging_can_be_suppressed(self, mock_http_server: MagicMock):
-        mock_http_server.get.return_value = _make_response(200, FAKE_METADATA)
-        assert discover_oauth_metadata(FAKE_URL, suppress_response_logging=True) == FAKE_METADATA
-        assert mock_http_server._suppress_oauth_response_logging is True
-
     def test_server_not_supporting_oauth(self, mock_http_server: MagicMock):
         mock_http_server.get.return_value = _make_response(404, None)
         with pytest.raises(RSConnectException, match="does not support OAuth"):
@@ -101,6 +98,7 @@ def test_success(self, mock_http_server: MagicMock):
         mock_http_server.post.return_value = _make_response(200, {"client_id": "test-client-123"})
         result = register_client(FAKE_METADATA, FAKE_URL)
         assert result == "test-client-123"
+        assert mock_http_server._suppress_oauth_response_logging is True
 
     def test_client_specific_request_timeout_and_deadline(self, mock_http_server: MagicMock):
         mock_http_server.post.return_value = _make_response(200, {"client_id": "bounded-client"})
@@ -108,18 +106,24 @@ def test_client_specific_request_timeout_and_deadline(self, mock_http_server: Ma
         assert mock_http_server.request_timeout == 0.25
         assert mock_http_server.request_deadline == 12.5
 
-    def test_response_logging_can_be_suppressed(self, mock_http_server: MagicMock):
-        mock_http_server.post.return_value = _make_response(200, {"client_id": "test-client-123"})
-        assert register_client(FAKE_METADATA, FAKE_URL, suppress_response_logging=True) == "test-client-123"
-        assert mock_http_server._suppress_oauth_response_logging is True
-
     def test_failure(self, mock_http_server: MagicMock):
         mock_http_server.post.return_value = _make_response(
-            400, {"error": "invalid_request", "error_description": "bad request"}
+            400, {"error": "invalid_request", "error_description": "refresh-token-secret"}
         )
         with pytest.raises(RSConnectException, match="OAuth error") as raised:
             register_client(FAKE_METADATA, FAKE_URL)
-        assert "bad request" in str(raised.value)
+        assert "invalid_request" in str(raised.value)
+        assert "refresh-token-secret" not in str(raised.value)
+
+    def test_unknown_oauth_error_code_is_not_echoed(self, mock_http_server: MagicMock):
+        secret = "refresh-token-secret"
+        mock_http_server.post.return_value = _make_response(400, {"error": secret, "error_description": secret})
+
+        with pytest.raises(RSConnectException) as raised:
+            register_client(FAKE_METADATA, FAKE_URL)
+
+        assert "unknown_error" in str(raised.value)
+        assert secret not in str(raised.value)
 
     def test_client_id_response_keeps_legacy_stringification(self, mock_http_server: MagicMock):
         mock_http_server.post.return_value = _make_response(200, {"client_id": "bad\nclient"})
@@ -161,6 +165,7 @@ def test_success(self, mock_http_server: MagicMock):
         mock_http_server.request.return_value = _make_response(200, {"access_token": "minted-key"})
         result = exchange_token_for_api_key(FAKE_URL, "oidc-token")
         assert result == "minted-key"
+        assert mock_http_server._suppress_oauth_response_logging is True
         # RFC 8693 token-exchange request shape.
         body = mock_http_server.request.call_args.kwargs["body"]
         assert b"grant_type=urn%3Aietf%3Aparams%3Aoauth%3Agrant-type%3Atoken-exchange" in body
@@ -229,18 +234,47 @@ def test_ambiguous_match(self, mock_http_server: MagicMock):
             exchange_token_for_api_key(FAKE_URL, "oidc-token")
 
     def test_verification_failure(self, mock_http_server: MagicMock):
+        secret = "subject-token-secret"
         self._set_metadata(mock_http_server)
         mock_http_server.request.return_value = _make_response(
-            400, {"error": "invalid_grant", "error_description": "could not verify token signature"}
+            400, {"error": "invalid_grant", "error_description": f"could not verify token signature {secret}"}
         )
-        with pytest.raises(RSConnectException, match="could not verify the identity token"):
+        with pytest.raises(RSConnectException, match="could not verify the identity token") as raised:
             exchange_token_for_api_key(FAKE_URL, "oidc-token")
+        assert secret not in str(raised.value)
 
     def test_generic_failure(self, mock_http_server: MagicMock):
         self._set_metadata(mock_http_server)
         mock_http_server.request.return_value = _make_response(500, {"error": "boom", "error_description": "kaboom"})
-        with pytest.raises(RSConnectException, match="HTTP 500"):
+        with pytest.raises(RSConnectException, match="HTTP 500") as raised:
             exchange_token_for_api_key(FAKE_URL, "oidc-token")
+        assert "boom" not in str(raised.value)
+        assert "kaboom" not in str(raised.value)
+
+
+def test_browser_callback_error_does_not_expose_description():
+    secret = "callback-token-secret"
+    with patch("rsconnect.oauth._HTTPServer") as server_factory:
+        server = server_factory.return_value
+        server.server_address = ("127.0.0.1", 0)
+        server.RequestHandlerClass = _CallbackHandler
+        server.handle_request.side_effect = lambda: _CallbackHandler.result_queue.put(
+            ("error", "invalid_request", secret)
+        )
+
+        with patch("rsconnect.oauth.webbrowser.open", return_value=True):
+            with pytest.raises(RSConnectException) as raised:
+                login_with_browser(FAKE_URL, "client-1", FAKE_METADATA)
+
+    assert "invalid_request" in str(raised.value)
+    assert secret not in str(raised.value)
+
+
+def test_callback_server_log_does_not_include_request_line():
+    request = "GET /callback?code=authorization-secret&state=state-secret HTTP/1.1"
+    with patch("rsconnect.oauth.logger.debug") as debug:
+        _CallbackHandler.log_message(MagicMock(), "%s", request)
+    debug.assert_called_once_with("OAuth callback server handled a request.")
 
 
 class TestDeviceCodeFlow:
@@ -330,6 +364,19 @@ def test_poll_malformed_response(self, _, mock_http_server: MagicMock):
         with pytest.raises(RSConnectException, match="unexpected response"):
             _poll_for_device_token(FAKE_METADATA, "client-1", "device-code-1", 5, 600)
 
+    @patch("rsconnect.oauth.time.sleep")
+    def test_poll_error_uses_code_without_description(self, _, mock_http_server: MagicMock):
+        secret = "refresh-token-secret"
+        mock_http_server.request.return_value = _make_response(
+            400, {"error": "server_error", "error_description": secret}
+        )
+
+        with pytest.raises(RSConnectException) as raised:
+            _poll_for_device_token(FAKE_METADATA, "client-1", "device-code-1", 5, 600)
+
+        assert "server_error" in str(raised.value)
+        assert secret not in str(raised.value)
+
     @pytest.mark.parametrize(
         ("response", "message"),
         [
@@ -361,19 +408,6 @@ def test_client_specific_request_timeout_and_deadline(self, mock_http_server: Ma
         )
         assert mock_http_server.request_timeout == 0.25
         assert mock_http_server.request_deadline == 12.5
-
-    def test_response_logging_can_be_suppressed(self, mock_http_server: MagicMock):
-        mock_http_server.request.return_value = _make_response(200, {"access_token": "new-at"})
-        refresh_access_token(
-            FAKE_METADATA,
-            "client-1",
-            "old-rt",
-            request_timeout=0.25,
-            request_deadline=12.5,
-            suppress_response_logging=True,
-        )
-        assert mock_http_server.request_timeout == 0.25
-        assert mock_http_server.request_deadline == 12.5
         assert mock_http_server._suppress_oauth_response_logging is True
 
     def test_refresh_keeps_legacy_path_only_for_token_query(self, mock_http_server: MagicMock):
@@ -397,12 +431,14 @@ def test_invalid_client(self, mock_http_server: MagicMock):
             refresh_access_token(FAKE_METADATA, "bad-client", "old-rt")
 
     def test_invalid_grant(self, mock_http_server: MagicMock):
+        secret = "refresh-token-secret"
         mock_http_server.request.return_value = _make_response(
-            400, {"error": "invalid_grant", "error_description": "refresh token expired"}
+            400, {"error": "invalid_grant", "error_description": secret}
         )
-        with pytest.raises(InvalidGrantError, match="refresh token expired") as raised:
+        with pytest.raises(InvalidGrantError) as raised:
             refresh_access_token(FAKE_METADATA, "client-1", "old-rt")
-        assert raised.value.description == "refresh token expired"
+        assert secret not in str(raised.value)
+        assert raised.value.description == secret
 
     def test_invalid_grant_without_a_description(self, mock_http_server: MagicMock):
         mock_http_server.request.return_value = _make_response(400, {"error": "invalid_grant"})