fix(storage): harden portable path diagnostics
This commit is contained in:
@@ -90,7 +90,67 @@ def test_doctor_json_does_not_echo_invalid_config_values(monkeypatch, tmp_path):
|
|||||||
payload = json.loads(result.stdout)
|
payload = json.loads(result.stdout)
|
||||||
assert payload["components"]["config"] == {
|
assert payload["components"]["config"] == {
|
||||||
"status": "error",
|
"status": "error",
|
||||||
"message": "configuration is invalid",
|
"message": "configuration is invalid or unreadable",
|
||||||
}
|
}
|
||||||
assert "super-secret" not in result.stdout
|
assert "super-secret" not in result.stdout
|
||||||
assert "pii_user" not in result.stdout
|
assert "pii_user" not in result.stdout
|
||||||
|
|
||||||
|
|
||||||
|
def test_doctor_json_normalizes_malformed_yaml(monkeypatch, tmp_path):
|
||||||
|
cfg = tmp_path / "demo.yaml"
|
||||||
|
cfg.write_text("password: super-secret\nroots: [unterminated")
|
||||||
|
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
|
||||||
|
|
||||||
|
result = runner.invoke(app, ["doctor", "--json", "--config", str(cfg)])
|
||||||
|
|
||||||
|
assert result.exit_code == 1
|
||||||
|
assert json.loads(result.stdout)["components"]["config"] == {
|
||||||
|
"status": "error",
|
||||||
|
"message": "configuration is invalid or unreadable",
|
||||||
|
}
|
||||||
|
assert result.stderr == ""
|
||||||
|
assert "super-secret" not in result.stdout
|
||||||
|
assert "Traceback" not in result.stdout
|
||||||
|
|
||||||
|
|
||||||
|
def test_doctor_json_normalizes_unreadable_config(monkeypatch, tmp_path):
|
||||||
|
cfg = tmp_path / "demo.yaml"
|
||||||
|
cfg.mkdir()
|
||||||
|
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
|
||||||
|
|
||||||
|
result = runner.invoke(app, ["doctor", "--json", "--config", str(cfg)])
|
||||||
|
|
||||||
|
assert result.exit_code == 1
|
||||||
|
assert json.loads(result.stdout)["components"]["config"] == {
|
||||||
|
"status": "error",
|
||||||
|
"message": "configuration is invalid or unreadable",
|
||||||
|
}
|
||||||
|
assert result.stderr == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_doctor_human_output_is_actionable_and_redacted(monkeypatch, tmp_path):
|
||||||
|
cfg = _config(tmp_path / "demo.yaml", absolute_sessions=tmp_path / "patient-private")
|
||||||
|
monkeypatch.delenv("THT_DATA_ROOT", raising=False)
|
||||||
|
|
||||||
|
result = runner.invoke(app, ["doctor", "--config", str(cfg)])
|
||||||
|
|
||||||
|
assert result.exit_code == 0
|
||||||
|
assert "data_root: warning - set THT_DATA_ROOT to enable portable storage" in result.stdout
|
||||||
|
assert "workspace_paths: warning - absolute legacy roots: sessions" in result.stdout
|
||||||
|
assert str(tmp_path) not in result.stdout
|
||||||
|
assert "patient_db" not in result.stdout
|
||||||
|
assert "super-secret" not in result.stdout
|
||||||
|
|
||||||
|
|
||||||
|
def test_doctor_human_config_error_is_actionable_and_redacted(monkeypatch, tmp_path):
|
||||||
|
cfg = tmp_path / "patient-private.yaml"
|
||||||
|
cfg.write_text("password: super-secret\nroots: [unterminated")
|
||||||
|
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
|
||||||
|
|
||||||
|
result = runner.invoke(app, ["doctor", "--config", str(cfg)])
|
||||||
|
|
||||||
|
assert result.exit_code == 1
|
||||||
|
assert "config: error - configuration is invalid or unreadable" in result.stdout
|
||||||
|
assert str(tmp_path) not in result.stdout
|
||||||
|
assert "super-secret" not in result.stdout
|
||||||
|
assert result.stderr == ""
|
||||||
|
|||||||
@@ -57,6 +57,28 @@ def test_path_escape_is_rejected(tmp_path):
|
|||||||
resolve_workspace_paths(cfg_path, cfg, tmp_path / "data")
|
resolve_workspace_paths(cfg_path, cfg, tmp_path / "data")
|
||||||
|
|
||||||
|
|
||||||
|
def test_workspace_symlink_escape_is_rejected(tmp_path):
|
||||||
|
cfg_path = _write_config(tmp_path / "demo.yaml")
|
||||||
|
cfg = load_config(cfg_path)
|
||||||
|
data_root = tmp_path / "data"
|
||||||
|
(data_root / "workspaces").mkdir(parents=True)
|
||||||
|
(data_root / "workspaces" / "demo").symlink_to(tmp_path / "private", target_is_directory=True)
|
||||||
|
|
||||||
|
with pytest.raises(ConfigError, match="outside workspaces root"):
|
||||||
|
resolve_workspace_paths(cfg_path, cfg, data_root)
|
||||||
|
|
||||||
|
|
||||||
|
def test_nested_root_symlink_escape_is_rejected(tmp_path):
|
||||||
|
cfg_path = _write_config(tmp_path / "demo.yaml")
|
||||||
|
cfg = load_config(cfg_path)
|
||||||
|
workspace = tmp_path / "data/workspaces/demo"
|
||||||
|
workspace.mkdir(parents=True)
|
||||||
|
(workspace / "sessions").symlink_to(tmp_path / "private", target_is_directory=True)
|
||||||
|
|
||||||
|
with pytest.raises(ConfigError, match="outside workspace root"):
|
||||||
|
resolve_workspace_paths(cfg_path, cfg, tmp_path / "data")
|
||||||
|
|
||||||
|
|
||||||
def test_data_root_environment_activates_portable_paths(monkeypatch, tmp_path):
|
def test_data_root_environment_activates_portable_paths(monkeypatch, tmp_path):
|
||||||
cfg_path = _write_config(tmp_path / "demo.yaml")
|
cfg_path = _write_config(tmp_path / "demo.yaml")
|
||||||
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
|
monkeypatch.setenv("THT_DATA_ROOT", str(tmp_path / "data"))
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ from pathlib import Path
|
|||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
import typer
|
import typer
|
||||||
|
import yaml
|
||||||
|
|
||||||
from tht.cli.config_cmd import CONFIG_OPT
|
from tht.cli.config_cmd import CONFIG_OPT
|
||||||
from tht.config import ConfigError, load_config
|
from tht.config import ConfigError, load_config
|
||||||
@@ -17,7 +18,14 @@ def _emit(payload: dict[str, Any], as_json: bool) -> None:
|
|||||||
typer.echo(json.dumps(payload, sort_keys=True))
|
typer.echo(json.dumps(payload, sort_keys=True))
|
||||||
return
|
return
|
||||||
for component, result in payload["components"].items():
|
for component, result in payload["components"].items():
|
||||||
typer.echo(f"{component}: {result['status']}")
|
detail = ""
|
||||||
|
if result["status"] == "error":
|
||||||
|
detail = f" - {result['message']}"
|
||||||
|
elif component == "data_root" and result["status"] == "warning":
|
||||||
|
detail = " - set THT_DATA_ROOT to enable portable storage"
|
||||||
|
elif component == "workspace_paths" and result["status"] == "warning":
|
||||||
|
detail = " - absolute legacy roots: " + ", ".join(result["legacy_absolute"])
|
||||||
|
typer.echo(f"{component}: {result['status']}{detail}")
|
||||||
|
|
||||||
|
|
||||||
def doctor(
|
def doctor(
|
||||||
@@ -32,11 +40,11 @@ def doctor(
|
|||||||
}
|
}
|
||||||
try:
|
try:
|
||||||
cfg = load_config(config)
|
cfg = load_config(config)
|
||||||
except ConfigError as exc:
|
except (ConfigError, yaml.YAMLError, OSError) as exc:
|
||||||
path_error = "outside workspace root" in str(exc)
|
path_error = isinstance(exc, ConfigError) and "outside workspace" in str(exc)
|
||||||
target = "workspace_paths" if path_error else "config"
|
target = "workspace_paths" if path_error else "config"
|
||||||
# Validation errors can contain Pydantic input excerpts, including credentials.
|
# Validation errors can contain Pydantic input excerpts, including credentials.
|
||||||
message = str(exc) if path_error else "configuration is invalid"
|
message = str(exc) if path_error else "configuration is invalid or unreadable"
|
||||||
components[target] = {"status": "error", "message": message}
|
components[target] = {"status": "error", "message": message}
|
||||||
payload = {"ok": False, "components": components}
|
payload = {"ok": False, "components": components}
|
||||||
_emit(payload, as_json)
|
_emit(payload, as_json)
|
||||||
|
|||||||
@@ -39,7 +39,11 @@ def resolve_workspace_paths(
|
|||||||
Relative roots are sandboxed to the logical workspace. Absolute roots are a
|
Relative roots are sandboxed to the logical workspace. Absolute roots are a
|
||||||
compatibility bridge for existing installations and are never rewritten.
|
compatibility bridge for existing installations and are never rewritten.
|
||||||
"""
|
"""
|
||||||
workspace = (data_root / "workspaces" / config_path.stem).resolve()
|
canonical_data_root = data_root.resolve()
|
||||||
|
workspaces_root = canonical_data_root / "workspaces"
|
||||||
|
workspace = (workspaces_root / config_path.stem).resolve()
|
||||||
|
if not workspace.is_relative_to(workspaces_root):
|
||||||
|
raise ConfigError("workspace resolves outside workspaces root")
|
||||||
roots = cfg.roots
|
roots = cfg.roots
|
||||||
return ResolvedPaths(
|
return ResolvedPaths(
|
||||||
workspace=workspace,
|
workspace=workspace,
|
||||||
|
|||||||
Reference in New Issue
Block a user