feat: deliver owner-bound credentials into bwrap commands
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0726e-5232-73f2-aaca-2c05ceb62efb
This commit is contained in:
parent
0196f083c4
commit
17d4160b6e
10 changed files with 505 additions and 6 deletions
|
|
@ -251,7 +251,7 @@ def test_execute_sends_bounded_request_to_in_namespace_owner(tmp_path) -> None:
|
|||
result = ext.execute(
|
||||
handle,
|
||||
["python3", "-V"],
|
||||
credential_route_refs=["rein-openweights-openrouter-approle"],
|
||||
credential_route_refs=[],
|
||||
execution_context={"actor": "agt", "run_id": "run-1"},
|
||||
timeout_seconds=30,
|
||||
max_output_bytes=1024,
|
||||
|
|
@ -262,7 +262,7 @@ def test_execute_sends_bounded_request_to_in_namespace_owner(tmp_path) -> None:
|
|||
|
||||
assert received["command"] == ["python3", "-V"]
|
||||
assert received["execution_context"] == {"actor": "agt", "run_id": "run-1"}
|
||||
assert received["credential_route_refs"] == ["rein-openweights-openrouter-approle"]
|
||||
assert received["credential_route_refs"] == []
|
||||
assert received["timeout_seconds"] == 30
|
||||
assert received["max_output_bytes"] == 1024
|
||||
assert received["stdin_text"] == "task payload"
|
||||
|
|
|
|||
126
tests/test_credential_delivery.py
Normal file
126
tests/test_credential_delivery.py
Normal file
|
|
@ -0,0 +1,126 @@
|
|||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from sandboxer.extensions.bwrap_runner import _run
|
||||
from sandboxer.extensions.credential_delivery import execute, provider_argv
|
||||
|
||||
CONTEXT = {
|
||||
"profile_id": "profile.proof",
|
||||
"project": "glas-harness",
|
||||
"actor": "agt",
|
||||
"run_id": "run-1",
|
||||
}
|
||||
|
||||
|
||||
def config(command=None):
|
||||
return {
|
||||
"credential_routes": {
|
||||
"proof": {
|
||||
"profiles": ["profile.proof"],
|
||||
"projects": ["glas-harness"],
|
||||
"actors": ["agt"],
|
||||
"exec_argv": command or ["/usr/bin/provider", "--"],
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"field,value", [("profile_id", "wrong"), ("project", "wrong"), ("actor", "usr"), ("run_id", "")]
|
||||
)
|
||||
def test_wrong_consumer_refuses_before_provider(field, value):
|
||||
context = {**CONTEXT, field: value}
|
||||
with pytest.raises(ValueError, match="binding denied"):
|
||||
provider_argv(config(), ["proof"], context)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("refs", [[], ["unknown"], ["proof", "proof"]])
|
||||
def test_unknown_or_multiple_routes_refuse(refs):
|
||||
with pytest.raises(ValueError):
|
||||
provider_argv(config(), refs, CONTEXT)
|
||||
|
||||
|
||||
def test_provider_requires_absolute_command():
|
||||
with pytest.raises(ValueError):
|
||||
provider_argv(config(["relative-provider", "--"]), ["proof"], CONTEXT)
|
||||
|
||||
|
||||
def test_redaction_precedes_truncation_and_environment_is_scoped(tmp_path):
|
||||
value = "synthetic-value-for-test"
|
||||
process = MagicMock(returncode=0)
|
||||
process.communicate.return_value = (value.encode() + b" suffix", value.encode())
|
||||
payload = {
|
||||
"command": ["true"],
|
||||
"timeout_seconds": 1,
|
||||
"max_output_bytes": 5,
|
||||
"credential_env": {"ANTHROPIC_API_KEY": value},
|
||||
}
|
||||
with patch("sandboxer.extensions.bwrap_runner.subprocess.Popen", return_value=process) as popen:
|
||||
response = _run(payload, tmp_path)
|
||||
assert response["stdout"] == "[REDA" and response["stderr"] == "[REDA"
|
||||
env = popen.call_args.kwargs["env"]
|
||||
assert env["ANTHROPIC_API_KEY"] == value
|
||||
assert "BAO_TOKEN" not in env and "VAULT_TOKEN" not in env
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"values", [{"PATH": "bad"}, {"ANTHROPIC_API_KEY": ""}, {"ANTHROPIC_API_KEY": "has newline\n"}]
|
||||
)
|
||||
def test_bad_credential_envelope_refuses_before_execution(tmp_path, values):
|
||||
with patch("sandboxer.extensions.bwrap_runner.subprocess.Popen") as popen:
|
||||
with pytest.raises(ValueError):
|
||||
_run(
|
||||
{
|
||||
"command": ["true"],
|
||||
"timeout_seconds": 1,
|
||||
"max_output_bytes": 8,
|
||||
"credential_env": values,
|
||||
},
|
||||
tmp_path,
|
||||
)
|
||||
popen.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body",
|
||||
[
|
||||
"print('sensitive-provider-failure'); raise SystemExit(1)",
|
||||
"print('sensitive-provider-failure')",
|
||||
"print('x'*100000)",
|
||||
],
|
||||
)
|
||||
def test_provider_failure_never_escapes(tmp_path, body):
|
||||
provider = tmp_path / "provider.py"
|
||||
provider.write_text(body)
|
||||
request = {
|
||||
"credential_route_refs": ["proof"],
|
||||
"execution_context": CONTEXT,
|
||||
"timeout_seconds": 1,
|
||||
"max_output_bytes": 8,
|
||||
"command": ["true"],
|
||||
}
|
||||
with pytest.raises(RuntimeError, match="^owner credential delivery failed$"):
|
||||
execute(config([sys.executable, str(provider), "--"]), request, tmp_path / "unused.sock")
|
||||
|
||||
|
||||
def test_parent_api_key_is_not_provider_fallback(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("ANTHROPIC_API_KEY", "ambient-secret")
|
||||
provider = tmp_path / "provider.py"
|
||||
provider.write_text(
|
||||
"import os,json; assert 'ANTHROPIC_API_KEY' not in os.environ; "
|
||||
"print(json.dumps({'exit_code':0}))"
|
||||
)
|
||||
result = execute(
|
||||
config([sys.executable, str(provider), "--"]),
|
||||
{
|
||||
"credential_route_refs": ["proof"],
|
||||
"execution_context": CONTEXT,
|
||||
"timeout_seconds": 1,
|
||||
"max_output_bytes": 8,
|
||||
},
|
||||
Path("/unused"),
|
||||
)
|
||||
assert result == {"exit_code": 0}
|
||||
Loading…
Add table
Add a link
Reference in a new issue