Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 32 additions & 14 deletions src/skillspector/providers/_agent_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -460,7 +460,8 @@ def _parse_gemini_output(raw: str) -> str:

_OPENCODE_AGENT_PREFIX = "skillspector-deny-all"
_OPENCODE_DENY_ALL = json.dumps({"*": "deny"}, separators=(",", ":"))
_OPENCODE_SUPPORTED_VERSION = "1.18.33"
_OPENCODE_MIN_VERSION = (1, 18, 31)
_OPENCODE_MAX_VERSION = (1, 18, 34)


def _opencode_agent_name(argv: list[str]) -> str:
Expand Down Expand Up @@ -502,15 +503,30 @@ def _opencode_config(agent_name: str) -> str:
)


def _parse_opencode_version(raw: bytes) -> str | None:
"""Parse an exact stable semantic version from ``opencode --version``."""
def _parse_opencode_version(raw: bytes) -> tuple[int, int, int] | None:
"""Parse a stable semantic version triple from ``opencode --version``.

Prereleases and unparsable output return ``None`` (fail closed).
"""
text = raw.decode("utf-8", errors="replace").strip()
match = re.fullmatch(
r"(?:opencode(?:\s+version)?\s+)?v?(\d+\.\d+\.\d+)",
r"(?:opencode(?:\s+version)?\s+)?v?(\d+)\.(\d+)\.(\d+)",
text,
flags=re.IGNORECASE,
)
return match.group(1) if match is not None else None
if match is None:
return None
return (int(match.group(1)), int(match.group(2)), int(match.group(3)))


def _format_opencode_version(version: tuple[int, int, int]) -> str:
"""Render a version triple for fail-closed error messages."""
return f"{version[0]}.{version[1]}.{version[2]}"


def _is_supported_opencode_version(version: tuple[int, int, int] | None) -> bool:
"""Return whether a parsed version is inside the verified range."""
return version is not None and _OPENCODE_MIN_VERSION <= version <= _OPENCODE_MAX_VERSION


def _prepare_opencode_env(
Expand Down Expand Up @@ -632,11 +648,13 @@ def run_probe(command: list[str], label: str) -> bytes:

version_raw = run_probe([binary, "--version"], "version")
version = _parse_opencode_version(version_raw)
if version != _OPENCODE_SUPPORTED_VERSION:
if not _is_supported_opencode_version(version):
version_text = version_raw.decode("utf-8", errors="replace").strip()
raise AgentCLIError(
"OpenCode security policy is verified only for version "
f"{_OPENCODE_SUPPORTED_VERSION}; found {version_text[:80]!r}"
"OpenCode security policy is verified only for versions "
f"{_format_opencode_version(_OPENCODE_MIN_VERSION)} through "
f"{_format_opencode_version(_OPENCODE_MAX_VERSION)}; found "
f"{version_text[:80]!r}"
)

agent_name = _opencode_agent_name(argv)
Expand Down Expand Up @@ -839,15 +857,15 @@ def _opencode_auth_check(binary: str) -> tuple[bool, str | None]:
timeout=15,
env=child_env,
)
if (
version_result.returncode != 0
or _parse_opencode_version(version_result.stdout or b"")
!= _OPENCODE_SUPPORTED_VERSION
if version_result.returncode != 0 or not _is_supported_opencode_version(
_parse_opencode_version(version_result.stdout or b"")
):
return (
False,
"opencode_cli requires exactly OpenCode "
f"{_OPENCODE_SUPPORTED_VERSION} for its verified deny-all policy",
"opencode_cli requires OpenCode "
f"{_format_opencode_version(_OPENCODE_MIN_VERSION)} through "
f"{_format_opencode_version(_OPENCODE_MAX_VERSION)} "
"for its verified deny-all policy",
)
result = subprocess.run(
[binary, "auth", "list"],
Expand Down
42 changes: 37 additions & 5 deletions tests/provider/test_opencode_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,11 +53,13 @@
from skillspector.providers._agent_cli import (
_OPENCODE_AGENT_PREFIX,
_OPENCODE_DENY_ALL,
_OPENCODE_SUPPORTED_VERSION,
_OPENCODE_MAX_VERSION,
_OPENCODE_MIN_VERSION,
AgentCLIError,
_build_opencode_argv,
_opencode_auth_check,
_parse_opencode_output,
_parse_opencode_version,
_prepare_opencode_env,
_run_bounded,
run_agent_cli,
Expand All @@ -73,7 +75,7 @@
_AUTH_LIST_EMPTY = b"0 credentials\n0 environment variables\n"
_AUTH_LIST_SINGULAR = b"1 credential\n1 environment variable\n"
_AUTH_LIST_UNPARSEABLE = b"authentication status unknown\n"
_VERSION_OK = f"{_OPENCODE_SUPPORTED_VERSION}\n".encode()
_VERSION_OK = b"1.18.33\n"


def _ok_result(stdout: bytes = _AUTH_LIST_OK) -> SimpleNamespace:
Expand Down Expand Up @@ -261,7 +263,7 @@ def test_probe_rejects_unverified_version(self, mock_run: MagicMock) -> None:
mock_run.return_value = _ok_result(b"1.18.99\n")
ok, reason = _opencode_auth_check(OPENCODE_BINARY)
assert ok is False
assert _OPENCODE_SUPPORTED_VERSION in (reason or "")
assert "1.18.31" in (reason or "") and "1.18.34" in (reason or "")
mock_run.assert_called_once()

@patch("skillspector.providers._agent_cli.subprocess.run")
Expand Down Expand Up @@ -291,6 +293,36 @@ def test_probe_unparseable_output_is_fail_closed(self, mock_run: MagicMock) -> N
assert "auth login" in (reason or "")


class TestOpencodeVersionRange:
@pytest.mark.parametrize("version", ["1.18.31", "1.18.32", "1.18.33", "1.18.34"])
def test_parse_supported_versions_compare_in_range(self, version: str) -> None:
parsed = _parse_opencode_version(f"{version}\n".encode())
assert parsed is not None
assert _OPENCODE_MIN_VERSION <= parsed <= _OPENCODE_MAX_VERSION

@pytest.mark.parametrize(
"raw", [b"1.18.30\n", b"1.18.35\n", b"1.18.34-1\n", b"garbage\n", b"\n"]
)
def test_parse_unsupported_versions_rejected(self, raw: bytes) -> None:
parsed = _parse_opencode_version(raw)
assert parsed is None or not (_OPENCODE_MIN_VERSION <= parsed <= _OPENCODE_MAX_VERSION)

@pytest.mark.parametrize("version", ["1.18.31", "1.18.32", "1.18.34"])
@patch("skillspector.providers._agent_cli.subprocess.run")
def test_probe_accepts_range_versions(self, mock_run: MagicMock, version: str) -> None:
mock_run.side_effect = [_ok_result(f"{version}\n".encode()), _ok_result()]
assert _opencode_auth_check(OPENCODE_BINARY) == (True, None)

@pytest.mark.parametrize("version", ["1.18.30", "1.18.35"])
@patch("skillspector.providers._agent_cli.subprocess.run")
def test_probe_rejects_out_of_range_versions(self, mock_run: MagicMock, version: str) -> None:
mock_run.return_value = _ok_result(f"{version}\n".encode())
ok, reason = _opencode_auth_check(OPENCODE_BINARY)
assert ok is False
assert "1.18.31" in (reason or "") and "1.18.34" in (reason or "")
mock_run.assert_called_once()


class TestOpencodeDenyAllPolicy:
def test_policy_overrides_hostile_ambient_configuration(self, tmp_path: Path) -> None:
base = {
Expand Down Expand Up @@ -335,7 +367,7 @@ def _write_fake_opencode(binary: Path) -> None:
from pathlib import Path

if sys.argv[1:] == ["--version"]:
print(os.environ.get("FAKE_OPENCODE_VERSION", {_OPENCODE_SUPPORTED_VERSION!r}))
print(os.environ.get("FAKE_OPENCODE_VERSION", '1.18.33'))
raise SystemExit(0)

config = json.loads(os.environ["OPENCODE_CONFIG_CONTENT"])
Expand Down Expand Up @@ -433,7 +465,7 @@ def test_unverified_runtime_version_fails_before_inference(
monkeypatch.setenv("FAKE_OPENCODE_VERSION", "1.18.99")
monkeypatch.setattr(_agent_cli, "find_binary", lambda _name: str(binary))

with pytest.raises(AgentCLIError, match=f"only for version {_OPENCODE_SUPPORTED_VERSION}"):
with pytest.raises(AgentCLIError, match="only for versions 1.18.31 through 1.18.34"):
run_agent_cli("opencode", "try a newer runtime", model="")
assert list(markers.iterdir()) == []

Expand Down
Loading