wehub-resource-sync
334 lines
12 KiB
Python
334 lines
12 KiB
Python
"""Redaction round-trip + D19 secret-inheritance regression tests.
|
|
|
|
Pins the behavior of the consolidated secret-inheritance service
|
|
(``opensquilla.gateway.config_secrets``) and proves that every config-mutation
|
|
surface routes through it identically:
|
|
|
|
* the redaction round-trip restores a redaction *marker* back to the stored
|
|
secret without ever exposing the value;
|
|
* the D19 inherit-then-clear-explicit rule keeps an inherited secret marked
|
|
and unmarks a path that received an explicit new value;
|
|
* no secret value ever reaches a wire response, a persisted config file, or a
|
|
structlog event.
|
|
|
|
A single distinctive sentinel secret is used throughout so any leak is
|
|
unambiguous. These are offline, credential-free tests.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tomllib
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import structlog
|
|
|
|
import opensquilla.gateway.rpc_config # noqa: F401 ensures config.* handlers register
|
|
from opensquilla.gateway import config_secrets
|
|
from opensquilla.gateway.auth import Principal
|
|
from opensquilla.gateway.config import GatewayConfig
|
|
from opensquilla.gateway.rpc import RpcContext, get_dispatcher
|
|
from opensquilla.gateway.rpc_config import _persist_config
|
|
from opensquilla.onboarding.mutations import upsert_llm_provider
|
|
from opensquilla.onboarding.redaction import REDACTED_PLACEHOLDER
|
|
|
|
# A secret that must never appear in any wire response, persisted file, or log.
|
|
SENTINEL_SECRET = "sk-SEN71NEL-do-not-leak-Zz9"
|
|
MARKER = config_secrets.REDACTED_PUBLIC_VALUE
|
|
|
|
|
|
def _admin_ctx(config: GatewayConfig) -> RpcContext:
|
|
return RpcContext(
|
|
conn_id="t",
|
|
config=config,
|
|
principal=Principal(
|
|
role="operator",
|
|
scopes=frozenset({"operator.admin"}),
|
|
is_owner=True,
|
|
authenticated=True,
|
|
),
|
|
)
|
|
|
|
|
|
# --- Part 1: the consolidated service in isolation (the D19 rule) -----------
|
|
|
|
|
|
def test_marker_constant_matches_read_side_redaction() -> None:
|
|
# The write-side marker must equal the read-side redaction value so a
|
|
# public config round-trips.
|
|
from opensquilla.gateway.config import _REDACTED
|
|
|
|
assert config_secrets.REDACTED_PUBLIC_VALUE == _REDACTED
|
|
|
|
|
|
def test_restore_redacted_values_inherits_marker_at_secret_path() -> None:
|
|
payload = {"remote": {"api_key": MARKER, "model": "new-model"}}
|
|
source = {"remote": {"api_key": SENTINEL_SECRET, "model": "old-model"}}
|
|
|
|
restored, redacted_paths = config_secrets.restore_redacted_values(payload, source)
|
|
|
|
# Marker inherited the stored secret; explicit non-secret value replaced.
|
|
assert restored["remote"]["api_key"] == SENTINEL_SECRET
|
|
assert restored["remote"]["model"] == "new-model"
|
|
assert redacted_paths == {"remote.api_key"}
|
|
|
|
|
|
def test_restore_redacted_values_explicit_new_secret_passes_through() -> None:
|
|
payload = {"remote": {"api_key": "sk-brand-new"}}
|
|
source = {"remote": {"api_key": SENTINEL_SECRET}}
|
|
|
|
restored, redacted_paths = config_secrets.restore_redacted_values(payload, source)
|
|
|
|
assert restored["remote"]["api_key"] == "sk-brand-new"
|
|
assert redacted_paths == set()
|
|
|
|
|
|
def test_restore_redacted_values_literal_marker_at_nonsecret_path_is_kept() -> None:
|
|
# "[redacted]" at a non-secret path is a legitimate value, not a marker.
|
|
payload = {"remote": {"model": MARKER}}
|
|
source = {"remote": {"model": "old"}}
|
|
|
|
restored, redacted_paths = config_secrets.restore_redacted_values(payload, source)
|
|
|
|
assert restored["remote"]["model"] == MARKER
|
|
assert redacted_paths == set()
|
|
|
|
|
|
def test_restore_redacted_values_marker_without_stored_secret_raises() -> None:
|
|
with pytest.raises(ValueError, match="no existing secret"):
|
|
config_secrets.restore_redacted_values(
|
|
{"remote": {"api_key": MARKER}}, {"remote": {}}
|
|
)
|
|
|
|
|
|
def test_inherit_then_clear_explicit_applies_d19_rule() -> None:
|
|
old = GatewayConfig()
|
|
old.mark_runtime_secret("llm.api_key")
|
|
old.mark_runtime_secret("search_api_key")
|
|
new = GatewayConfig()
|
|
|
|
# The path that received an explicit new value is cleared; the inherited
|
|
# (marker-preserved) path keeps its runtime-secret marker.
|
|
config_secrets.inherit_then_clear_explicit(old, new, {"search_api_key"})
|
|
|
|
assert "llm.api_key" in new._runtime_secret_paths
|
|
assert "search_api_key" not in new._runtime_secret_paths
|
|
|
|
|
|
def test_inherit_runtime_secrets_none_source_is_noop() -> None:
|
|
target = GatewayConfig()
|
|
target.mark_runtime_secret("llm.api_key")
|
|
config_secrets.inherit_runtime_secrets(None, target)
|
|
# A None source must not wipe the target's markers.
|
|
assert "llm.api_key" in target._runtime_secret_paths
|
|
|
|
|
|
# --- Part 2: RPC surface (config.get / config.apply) round-trip -------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_secret_reads_back_as_marker_never_value(tmp_path) -> None:
|
|
cfg = GatewayConfig(
|
|
config_path=str(tmp_path / "c.toml"),
|
|
memory={"embedding": {"remote": {"api_key": SENTINEL_SECRET}}},
|
|
)
|
|
|
|
res = await get_dispatcher().dispatch("r1", "config.get", {}, _admin_ctx(cfg))
|
|
|
|
assert res.error is None, res.error
|
|
assert res.payload["memory"]["embedding"]["remote"]["api_key"] == MARKER
|
|
assert SENTINEL_SECRET not in json.dumps(res.payload)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_marker_apply_preserves_secret_and_never_leaks(tmp_path) -> None:
|
|
config_path = tmp_path / "c.toml"
|
|
cfg = GatewayConfig(
|
|
config_path=str(config_path),
|
|
memory={"embedding": {"remote": {"api_key": SENTINEL_SECRET}}},
|
|
)
|
|
dispatcher = get_dispatcher()
|
|
|
|
get_res = await dispatcher.dispatch("r1", "config.get", {}, _admin_ctx(cfg))
|
|
assert get_res.payload["memory"]["embedding"]["remote"]["api_key"] == MARKER
|
|
|
|
apply_res = await dispatcher.dispatch(
|
|
"r2", "config.apply", {"config": get_res.payload}, _admin_ctx(cfg)
|
|
)
|
|
|
|
assert apply_res.error is None, apply_res.error
|
|
# Marker resubmit inherited the stored secret: value unchanged.
|
|
assert cfg.memory.embedding.remote.api_key == SENTINEL_SECRET
|
|
# Never echoed on the wire; never written to disk as the marker string.
|
|
assert SENTINEL_SECRET not in json.dumps(apply_res.payload)
|
|
assert MARKER not in config_path.read_text(encoding="utf-8")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_runtime_marked_secret_survives_and_is_not_persisted(tmp_path) -> None:
|
|
config_path = tmp_path / "c.toml"
|
|
cfg = GatewayConfig(
|
|
config_path=str(config_path),
|
|
llm={"provider": "openai", "model": "gpt-x", "api_key": SENTINEL_SECRET},
|
|
)
|
|
# Emulate an env-sourced runtime-secret marker (as config_store stamps).
|
|
cfg.mark_runtime_secret("llm.api_key")
|
|
dispatcher = get_dispatcher()
|
|
|
|
get_res = await dispatcher.dispatch("r1", "config.get", {}, _admin_ctx(cfg))
|
|
assert get_res.payload["llm"]["api_key"] == MARKER
|
|
|
|
apply_res = await dispatcher.dispatch(
|
|
"r2", "config.apply", {"config": get_res.payload}, _admin_ctx(cfg)
|
|
)
|
|
|
|
assert apply_res.error is None, apply_res.error
|
|
# Secret preserved by value AND marker; marker keeps it off disk.
|
|
assert cfg.llm.api_key == SENTINEL_SECRET
|
|
assert "llm.api_key" in cfg._runtime_secret_paths
|
|
assert SENTINEL_SECRET not in config_path.read_text(encoding="utf-8")
|
|
assert SENTINEL_SECRET not in json.dumps(apply_res.payload)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_explicit_new_secret_replaces_and_clears_marker(tmp_path) -> None:
|
|
config_path = tmp_path / "c.toml"
|
|
cfg = GatewayConfig(
|
|
config_path=str(config_path),
|
|
memory={"embedding": {"remote": {"api_key": "sk-original"}}},
|
|
)
|
|
cfg.mark_runtime_secret("memory.embedding.remote.api_key")
|
|
|
|
res = await get_dispatcher().dispatch(
|
|
"r1",
|
|
"config.set",
|
|
{"path": "memory.embedding.remote.api_key", "value": SENTINEL_SECRET},
|
|
_admin_ctx(cfg),
|
|
)
|
|
|
|
assert res.error is None, res.error
|
|
# Explicit value replaced the stored secret and cleared its marker, so it
|
|
# is now persisted (the value belongs on disk once explicitly provided).
|
|
assert cfg.memory.embedding.remote.api_key == SENTINEL_SECRET
|
|
assert "memory.embedding.remote.api_key" not in cfg._runtime_secret_paths
|
|
persisted = tomllib.loads(config_path.read_text(encoding="utf-8"))
|
|
assert persisted["memory"]["embedding"]["remote"]["api_key"] == SENTINEL_SECRET
|
|
# The response body never carries the value.
|
|
assert SENTINEL_SECRET not in json.dumps(res.payload)
|
|
|
|
|
|
# --- Part 3: onboarding mutation surface ------------------------------------
|
|
|
|
|
|
def test_onboarding_blank_key_inherits_stored_secret() -> None:
|
|
res1 = upsert_llm_provider(
|
|
GatewayConfig(), provider_id="openrouter", model="m1", api_key=SENTINEL_SECRET
|
|
)
|
|
assert res1.config.llm.api_key == SENTINEL_SECRET
|
|
|
|
# Re-saving the same provider with a blank key keeps the stored secret.
|
|
res2 = upsert_llm_provider(
|
|
res1.config, provider_id="openrouter", model="m2", api_key=""
|
|
)
|
|
|
|
assert res2.config.llm.api_key == SENTINEL_SECRET
|
|
assert res2.public_payload["api_key"] == REDACTED_PLACEHOLDER
|
|
assert SENTINEL_SECRET not in json.dumps(res2.public_payload)
|
|
|
|
|
|
def test_onboarding_explicit_key_replaces_and_clears_marker() -> None:
|
|
res1 = upsert_llm_provider(
|
|
GatewayConfig(), provider_id="openrouter", model="m1", api_key="sk-old"
|
|
)
|
|
# Emulate a runtime-secret marker carried on the stored config.
|
|
res1.config.mark_runtime_secret("llm.api_key")
|
|
|
|
res2 = upsert_llm_provider(
|
|
res1.config, provider_id="openrouter", model="m1", api_key=SENTINEL_SECRET
|
|
)
|
|
|
|
assert res2.config.llm.api_key == SENTINEL_SECRET
|
|
assert "llm.api_key" not in res2.config._runtime_secret_paths
|
|
assert SENTINEL_SECRET not in json.dumps(res2.public_payload)
|
|
|
|
|
|
# --- Part 4: no secret value ever reaches a structlog event -----------------
|
|
|
|
|
|
# --- Part 5: env-absorbed auth token must not be baked to disk -------------
|
|
|
|
|
|
async def test_config_set_does_not_bake_env_auth_token_into_toml(
|
|
tmp_path, monkeypatch
|
|
) -> None:
|
|
monkeypatch.setenv("OPENSQUILLA_AUTH_TOKEN", "tok-env-original")
|
|
cfg_path = tmp_path / "config.toml"
|
|
cfg_path.write_text('[auth]\nmode = "token"\n')
|
|
|
|
cfg = GatewayConfig.load(cfg_path)
|
|
assert cfg.auth.token == "tok-env-original"
|
|
|
|
res = await get_dispatcher().dispatch(
|
|
"r1", "config.set", {"path": "port", "value": 18795}, _admin_ctx(cfg)
|
|
)
|
|
assert res.error is None, res.error
|
|
|
|
text = cfg_path.read_text()
|
|
assert tomllib.loads(text)["port"] == 18795
|
|
assert "tok-env-original" not in text
|
|
|
|
|
|
async def test_env_auth_token_rotation_survives_config_save(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("OPENSQUILLA_AUTH_TOKEN", "tok-env-original")
|
|
cfg_path = tmp_path / "config.toml"
|
|
cfg_path.write_text('[auth]\nmode = "token"\n')
|
|
|
|
cfg = GatewayConfig.load(cfg_path)
|
|
await get_dispatcher().dispatch(
|
|
"r1", "config.set", {"path": "port", "value": 18795}, _admin_ctx(cfg)
|
|
)
|
|
|
|
monkeypatch.setenv("OPENSQUILLA_AUTH_TOKEN", "tok-env-rotated")
|
|
rebooted = GatewayConfig.load(cfg_path)
|
|
assert rebooted.auth.token == "tok-env-rotated"
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="Windows st_mode does not represent DACLs")
|
|
def test_persist_config_fresh_file_has_owner_only_posix_mode(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("OPENSQUILLA_AUTH_TOKEN", "tok-env-original")
|
|
old_umask = os.umask(0o022)
|
|
try:
|
|
cfg = GatewayConfig()
|
|
cfg_path = tmp_path / "fresh" / "config.toml"
|
|
cfg.config_path = str(cfg_path)
|
|
_persist_config(cfg)
|
|
finally:
|
|
os.umask(old_umask)
|
|
|
|
mode = Path(cfg_path).stat().st_mode & 0o777
|
|
assert mode & 0o077 == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_secret_never_appears_in_any_log_event(tmp_path) -> None:
|
|
cfg = GatewayConfig(
|
|
config_path=str(tmp_path / "c.toml"),
|
|
memory={"embedding": {"remote": {"api_key": SENTINEL_SECRET}}},
|
|
)
|
|
dispatcher = get_dispatcher()
|
|
|
|
with structlog.testing.capture_logs() as captured:
|
|
get_res = await dispatcher.dispatch("r1", "config.get", {}, _admin_ctx(cfg))
|
|
await dispatcher.dispatch(
|
|
"r2", "config.apply", {"config": get_res.payload}, _admin_ctx(cfg)
|
|
)
|
|
# Onboarding surface in the same capture window.
|
|
upsert_llm_provider(
|
|
cfg, provider_id="openrouter", model="m", api_key=SENTINEL_SECRET
|
|
)
|
|
|
|
blob = json.dumps(captured, default=repr)
|
|
assert SENTINEL_SECRET not in blob
|