fix(python): add encoding=utf-8 to file reads and writes in sandbox.py (#1912)

* fix(python): add encoding=utf-8 to all file reads and writes in sandbox.py

read_text(), fdopen() without encoding use the system locale, which can
differ on non-UTF-8 systems. All config files (metadata.json,
oidc_token.json, active_gateway) are UTF-8 text.

* fix(python): catch UnicodeDecodeError in _read_oidc_token_bundle, add encoding tests

Corrupt or non-UTF-8 oidc_token.json now returns None consistently.
Add regression tests for non-ASCII UTF-8 paths in metadata, oidc token,
and active_gateway files.

* fix(python): use ensure_ascii=False in encoding tests, add from_active_cluster UTF-8 bytes test

* fix(python): assert non-ASCII cluster name and endpoint in encoding test

* fix(python): wrap long write_bytes line for ruff format

* fix(python): apply ruff format to sandbox_test.py

Author: moezdil

python/openshell/sandbox.py

@@ -318,7 +318,7 @@ def from_active_cluster(
         gateway_dir = _xdg_config_home() / "openshell" / "gateways" / cluster_name
         metadata_path = gateway_dir / "metadata.json"
         try:
-            metadata = json.loads(metadata_path.read_text())
+            metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
         except FileNotFoundError:
             raise SandboxError(f"gateway '{cluster_name}' not found") from None
         if "gateway_endpoint" not in metadata:
@@ -846,10 +846,10 @@ def _read_oidc_token_bundle(gateway_dir: pathlib.Path) -> dict | None:
     """
     token_path = gateway_dir / "oidc_token.json"
     try:
-        return json.loads(token_path.read_text())
+        return json.loads(token_path.read_text(encoding="utf-8"))
     except FileNotFoundError:
         return None
-    except (OSError, json.JSONDecodeError):
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError):
         return None
 
 
@@ -1299,7 +1299,7 @@ def _write_to_disk(self, bundle: dict) -> None:
         )
         tmp_path = pathlib.Path(tmp_name)
         try:
-            with os.fdopen(fd, "w") as f:
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
                 f.write(payload)
             with contextlib.suppress(OSError):
                 tmp_path.chmod(0o600)
@@ -1374,7 +1374,7 @@ def _resolve_active_cluster() -> str:
         return env_gateway
     active_file = _xdg_config_home() / "openshell" / "active_gateway"
     try:
-        value = active_file.read_text().strip()
+        value = active_file.read_text(encoding="utf-8").strip()
     except FileNotFoundError:
         raise SandboxError("no active gateway configured") from None
     if value == "":

python/openshell/sandbox_test.py

@@ -24,6 +24,7 @@
     _make_cluster_bearer_provider,
     _normalize_bearer,
     _OidcRefresher,
+    _read_oidc_token_bundle,
 )
 
 
@@ -1345,3 +1346,70 @@ def test_inference_set_cluster_forwards_no_verify_flag() -> None:
 
     assert stub.request is not None
     assert stub.request.no_verify is True
+
+
+# ---------------------------------------------------------------------------
+# Encoding regression tests (utf-8 explicit on all config file reads/writes)
+# ---------------------------------------------------------------------------
+
+
+def test_read_oidc_token_bundle_parses_non_ascii_utf8(tmp_path: Path) -> None:
+    gateway_dir = tmp_path / "gw"
+    gateway_dir.mkdir()
+    payload = {"refresh_token": "tok", "issuer": "https://example.com/é"}
+    (gateway_dir / "oidc_token.json").write_bytes(
+        json.dumps(payload, ensure_ascii=False).encode("utf-8")
+    )
+    result = _read_oidc_token_bundle(gateway_dir)
+    assert result == payload
+
+
+def test_read_oidc_token_bundle_returns_none_on_corrupt_bytes(tmp_path: Path) -> None:
+    gateway_dir = tmp_path / "gw"
+    gateway_dir.mkdir()
+    (gateway_dir / "oidc_token.json").write_bytes(b"\xff\xfe not utf-8")
+    assert _read_oidc_token_bundle(gateway_dir) is None
+
+
+def test_load_cluster_bearer_token_handles_non_ascii_utf8_oidc(tmp_path: Path) -> None:
+    gateway_dir = tmp_path / "gw"
+    gateway_dir.mkdir()
+    bundle = {
+        "access_token": "accéss",
+        "refresh_token": "ref",
+        "expiry": "2099-01-01T00:00:00Z",
+        "issuer": "https://example.com",
+        "client_id": "c",
+        "client_secret": "s",
+    }
+    (gateway_dir / "oidc_token.json").write_bytes(
+        json.dumps(bundle, ensure_ascii=False).encode("utf-8")
+    )
+    token = _load_cluster_bearer_token(gateway_dir)
+    assert token == "accéss"
+
+
+def test_from_active_cluster_reads_utf8_bytes_from_active_gateway_and_metadata(
+    tmp_path: Path,
+    monkeypatch: Any,
+) -> None:
+    gateway_name = "gw-é"
+    gateway_dir = tmp_path / "openshell" / "gateways" / gateway_name
+    gateway_dir.mkdir(parents=True)
+    (tmp_path / "openshell" / "active_gateway").write_bytes(
+        gateway_name.encode("utf-8")
+    )
+    meta = {"gateway_endpoint": "http://tést.example:8080"}
+    (gateway_dir / "metadata.json").write_bytes(
+        json.dumps(meta, ensure_ascii=False).encode("utf-8")
+    )
+
+    monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
+    monkeypatch.delenv("OPENSHELL_GATEWAY", raising=False)
+
+    client = SandboxClient.from_active_cluster()
+    try:
+        assert client._cluster_name == gateway_name
+        assert client._endpoint == "tést.example:8080"
+    finally:
+        client.close()