457 lines
16 KiB
Python
457 lines
16 KiB
Python
"""Boundary coverage for the three security-critical image-lane helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import sys
|
|
import urllib.error
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
RUNNER = REPO_ROOT / "dockerfiles/hermes-kaniko-heredoc-runner.py"
|
|
DOCKERFILE = REPO_ROOT / "dockerfiles/Dockerfile.hermes-agent"
|
|
HARBOR = (
|
|
REPO_ROOT
|
|
/ "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py"
|
|
)
|
|
TRIGGER = REPO_ROOT / "services/hermes/scripts/jenkins_image_build_trigger.py"
|
|
|
|
|
|
def _load(path: Path, name: str):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class Response(io.BytesIO):
|
|
"""Context-managed urllib response used without network access."""
|
|
|
|
def __init__(
|
|
self,
|
|
status: int,
|
|
body: bytes = b"",
|
|
headers: dict[str, str] | None = None,
|
|
) -> None:
|
|
super().__init__(body)
|
|
self.status = status
|
|
self.headers = headers or {}
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
self.close()
|
|
|
|
|
|
class FakeClient:
|
|
"""Ordered Harbor response fake with the pinned API origin."""
|
|
|
|
origin = "https://registry.bstein.dev/api/v2.0"
|
|
|
|
def __init__(self, responses) -> None:
|
|
self.responses = list(responses)
|
|
self.calls = []
|
|
|
|
def request(self, method, path, payload=None):
|
|
self.calls.append((method, path, payload))
|
|
return self.responses.pop(0)
|
|
|
|
|
|
def _count(value: int) -> dict[str, str]:
|
|
return {"X-Total-Count": str(value)}
|
|
|
|
|
|
def _robot(module, *, immutable: bool = False, duration=-1) -> dict:
|
|
access = [
|
|
{"resource": "repository", "action": "pull"},
|
|
{"resource": "repository", "action": "push"},
|
|
]
|
|
if immutable:
|
|
access.append({"resource": "immutable-tag", "action": "list"})
|
|
return {
|
|
"id": 41,
|
|
"name": module.PUBLISH_ROBOT,
|
|
"description": "publisher",
|
|
"level": "system",
|
|
"duration": duration,
|
|
"editable": True,
|
|
"disable": False,
|
|
"permissions": [
|
|
{
|
|
"kind": "project",
|
|
"namespace": module.PROJECT,
|
|
"access": access,
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _robot_list(module, *, robot_id=41, name=None):
|
|
return json.dumps(
|
|
[{"id": robot_id, "name": name or module.PUBLISH_ROBOT}]
|
|
).encode()
|
|
|
|
|
|
def test_runner_rejects_directive_and_heredoc_boundaries(tmp_path: Path) -> None:
|
|
"""Malformed parser state, bodies, sizes, and indices all fail closed."""
|
|
module = _load(RUNNER, "runner_boundary_coverage")
|
|
assert module._escape_character([""]) == "\\"
|
|
with pytest.raises(ValueError, match="multiple"):
|
|
module._escape_character(["# escape=\\", "# escape=\\"])
|
|
with pytest.raises(ValueError, match="unsupported"):
|
|
module._escape_character(["# escape=^"])
|
|
with pytest.raises(ValueError, match="unterminated.*continuation"):
|
|
module._logical_instruction(["R\\"], 0, "\\")
|
|
with pytest.raises(ValueError, match="unterminated node"):
|
|
module.extract_blocks("RUN node <<'NODE'\nbody")
|
|
|
|
source = DOCKERFILE.read_text(encoding="utf-8")
|
|
marker = "RUN node <<'NODE'\n"
|
|
body_start = source.index(marker) + len(marker)
|
|
body_end = source.index("\nNODE", body_start)
|
|
empty = source[:body_start] + source[body_end + 1 :]
|
|
with pytest.raises(ValueError, match="empty node"):
|
|
module.extract_blocks(empty)
|
|
|
|
dockerfile = tmp_path / "Dockerfile"
|
|
dockerfile.write_text("", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="size"):
|
|
module.replay(dockerfile, 1)
|
|
with dockerfile.open("wb") as stream:
|
|
stream.truncate(module.MAX_DOCKERFILE_BYTES + 1)
|
|
with pytest.raises(ValueError, match="size"):
|
|
module.replay(dockerfile, 1)
|
|
dockerfile.write_text(source, encoding="utf-8")
|
|
for index in (0, 10):
|
|
with pytest.raises(ValueError, match="index"):
|
|
module.replay(dockerfile, index)
|
|
|
|
|
|
def test_runner_main_dispatches_the_exact_path_and_index(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""The command-line wrapper cannot omit either bounded input."""
|
|
module = _load(RUNNER, "runner_main_coverage")
|
|
dockerfile = tmp_path / "Dockerfile"
|
|
seen = []
|
|
monkeypatch.setattr(module, "replay", lambda path, index: seen.append((path, index)))
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
["runner", "--dockerfile", str(dockerfile), "--block-index", "4"],
|
|
)
|
|
assert module.main() == 0
|
|
assert seen == [(dockerfile, 4)]
|
|
|
|
|
|
def test_trigger_redirect_wrapper_accepts_only_plugin_303(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""The low-level opener returns a real response or the one expected redirect."""
|
|
module = _load(TRIGGER, "trigger_opener_coverage")
|
|
assert module._NoRedirect().redirect_request(None, None, 0, None, None, None) is None
|
|
request = module.urllib.request.Request(module.JENKINS_BUILD_URL)
|
|
|
|
class Opener:
|
|
def __init__(self, result) -> None:
|
|
self.result = result
|
|
|
|
def open(self, _request, timeout):
|
|
assert timeout == 9
|
|
if isinstance(self.result, BaseException):
|
|
raise self.result
|
|
return self.result
|
|
|
|
success = Response(201)
|
|
monkeypatch.setattr(
|
|
module.urllib.request, "build_opener", lambda *_args: Opener(success)
|
|
)
|
|
assert module._open_without_redirect(request, 9) is success
|
|
redirect = module.urllib.error.HTTPError(
|
|
request.full_url, 303, "queued", {"Location": "/queue/item/1/"}, None
|
|
)
|
|
monkeypatch.setattr(
|
|
module.urllib.request, "build_opener", lambda *_args: Opener(redirect)
|
|
)
|
|
assert module._open_without_redirect(request, 9) is redirect
|
|
denied = module.urllib.error.HTTPError(request.full_url, 403, "denied", {}, None)
|
|
monkeypatch.setattr(
|
|
module.urllib.request, "build_opener", lambda *_args: Opener(denied)
|
|
)
|
|
with pytest.raises(urllib.error.HTTPError):
|
|
module._open_without_redirect(request, 9)
|
|
|
|
|
|
def test_trigger_main_reports_safe_success_and_errors(
|
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""CLI output contains safe metadata and converts expected failures to JSON."""
|
|
module = _load(TRIGGER, "trigger_main_coverage")
|
|
revision = "a" * 40
|
|
monkeypatch.setattr(sys, "argv", ["trigger", revision])
|
|
monkeypatch.setattr(
|
|
module,
|
|
"trigger_build",
|
|
lambda value: {"job": module.JOB_NAME, "source_revision": value, "status": 201},
|
|
)
|
|
assert module.main() == 0
|
|
assert json.loads(capsys.readouterr().out)["source_revision"] == revision
|
|
monkeypatch.setattr(
|
|
module, "trigger_build", lambda _value: (_ for _ in ()).throw(OSError("closed"))
|
|
)
|
|
assert module.main() == 1
|
|
assert json.loads(capsys.readouterr().out) == {"error": "closed"}
|
|
|
|
|
|
def test_harbor_http_client_bounds_requests_and_failures(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Credentials stay same-origin and all response/error paths remain bounded."""
|
|
module = _load(HARBOR, "harbor_client_coverage")
|
|
assert module.NoRedirect().redirect_request(None, None, 0, None, None, None) is None
|
|
|
|
class Opener:
|
|
result = Response(200, b"ok", {"X-Test": "yes"})
|
|
|
|
def open(self, request, timeout):
|
|
self.request = request
|
|
assert timeout == 20
|
|
if isinstance(self.result, BaseException):
|
|
raise self.result
|
|
return self.result
|
|
|
|
opener = Opener()
|
|
monkeypatch.setattr(module.urllib.request, "build_opener", lambda *_args: opener)
|
|
client = module.HarborClient(module.EXPECTED_ORIGIN + "/", "admin", "private")
|
|
status, body, headers = client.request("POST", "/rules", {"value": 1})
|
|
assert (status, body, headers["X-Test"]) == (200, b"ok", "yes")
|
|
assert opener.request.get_header("Content-type") == "application/json"
|
|
|
|
opener.result = module.urllib.error.HTTPError(
|
|
module.EXPECTED_ORIGIN, 404, "missing", {}, io.BytesIO(b"missing")
|
|
)
|
|
assert client.request("GET", "/missing")[:2] == (404, b"missing")
|
|
opener.result = module.urllib.error.URLError("offline")
|
|
with pytest.raises(module.HarborUnavailable):
|
|
client.request("GET", "/rules")
|
|
opener.result = Response(200, b"x" * (module.MAX_RESPONSE + 1))
|
|
with pytest.raises(RuntimeError, match="size"):
|
|
client.request("GET", "/rules")
|
|
|
|
|
|
@pytest.mark.parametrize("body", [b"\xff", b"not-json", b"{}", b"[1]"])
|
|
def test_harbor_json_lists_reject_invalid_shapes(body: bytes) -> None:
|
|
"""Malformed encodings, JSON, objects, and scalar members are never lists."""
|
|
module = _load(HARBOR, f"harbor_json_{body!r}")
|
|
with pytest.raises(RuntimeError, match="invalid"):
|
|
module._json_list(body, "fixture")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"robot,match",
|
|
[
|
|
({"permissions": "bad"}, "permissions"),
|
|
({"permissions": [1]}, "permissions"),
|
|
({"permissions": []}, "one existing"),
|
|
(
|
|
{
|
|
"permissions": [
|
|
{"kind": "project", "namespace": "bstein", "access": []},
|
|
{"kind": "project", "namespace": "bstein", "access": []},
|
|
]
|
|
},
|
|
"one existing",
|
|
),
|
|
(
|
|
{
|
|
"permissions": [
|
|
{"kind": "project", "namespace": "bstein", "access": "bad"}
|
|
]
|
|
},
|
|
"access",
|
|
),
|
|
(
|
|
{
|
|
"permissions": [
|
|
{"kind": "project", "namespace": "bstein", "access": []}
|
|
]
|
|
},
|
|
"pull/push",
|
|
),
|
|
],
|
|
)
|
|
def test_publisher_scope_rejects_malformed_or_incomplete_permissions(
|
|
robot: dict, match: str
|
|
) -> None:
|
|
"""Robot mutations require one structurally exact existing push scope."""
|
|
module = _load(HARBOR, f"harbor_scope_{match}")
|
|
with pytest.raises(RuntimeError, match=match):
|
|
module._publisher_scope(robot)
|
|
assert module._verified_publisher(robot) is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"responses,match,unavailable",
|
|
[
|
|
([(503, b"", {})], "503", True),
|
|
([(403, b"", {})], "403", False),
|
|
([(200, b"[]", _count(0))], "exactly one", False),
|
|
([(200, b"[]", _count(0))], "exactly one", False),
|
|
([(200, b'[{\"name\":\"robot$jenkins-pipelines\",\"id\":0}]', _count(1))], "valid ID", False),
|
|
],
|
|
)
|
|
def test_publisher_list_rejects_status_count_and_id(
|
|
responses, match: str, unavailable: bool
|
|
) -> None:
|
|
"""Publisher discovery distinguishes readiness from permanent policy errors."""
|
|
module = _load(HARBOR, f"harbor_publisher_list_{match}_{unavailable}")
|
|
error = module.HarborUnavailable if unavailable else RuntimeError
|
|
with pytest.raises(error, match=match):
|
|
module.ensure_publisher_can_read_rule(FakeClient(responses))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"second,match,unavailable",
|
|
[
|
|
((503, b"", {}), "503", True),
|
|
((403, b"", {}), "403", False),
|
|
((200, b"not-json", {}), "invalid robot JSON", False),
|
|
((200, b"[]", {}), "invalid shape", False),
|
|
((200, b'{}', {}), "not active and exact", False),
|
|
],
|
|
)
|
|
def test_publisher_read_rejects_status_and_identity(
|
|
second, match: str, unavailable: bool
|
|
) -> None:
|
|
"""The detailed robot reread must be available, valid, and exact."""
|
|
module = _load(HARBOR, f"harbor_publisher_read_{match}_{unavailable}")
|
|
responses = [(200, _robot_list(module), _count(1)), second]
|
|
error = module.HarborUnavailable if unavailable else RuntimeError
|
|
with pytest.raises(error, match=match):
|
|
module.ensure_publisher_can_read_rule(FakeClient(responses))
|
|
|
|
|
|
def test_publisher_update_rejects_duration_reference_and_status(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Only the preserved scope and a successful policy PUT can be accepted."""
|
|
module = _load(HARBOR, "harbor_publisher_update_coverage")
|
|
for duration in ("forever",):
|
|
responses = [
|
|
(200, _robot_list(module), _count(1)),
|
|
(200, json.dumps(_robot(module, duration=duration)).encode(), {}),
|
|
]
|
|
with pytest.raises(RuntimeError, match="duration"):
|
|
module.ensure_publisher_can_read_rule(FakeClient(responses))
|
|
|
|
robot = _robot(module)
|
|
monkeypatch.setattr(module, "_publisher_scope", lambda _robot: ({}, []))
|
|
responses = [
|
|
(200, _robot_list(module), _count(1)),
|
|
(200, json.dumps(robot).encode(), {}),
|
|
]
|
|
with pytest.raises(RuntimeError, match="normalization"):
|
|
module.ensure_publisher_can_read_rule(FakeClient(responses))
|
|
|
|
monkeypatch.undo()
|
|
for status, error in ((503, module.HarborUnavailable), (409, RuntimeError)):
|
|
responses = [
|
|
(200, _robot_list(module), _count(1)),
|
|
(200, json.dumps(_robot(module)).encode(), {}),
|
|
(status, b"", {}),
|
|
]
|
|
with pytest.raises(error):
|
|
module.ensure_publisher_can_read_rule(FakeClient(responses))
|
|
|
|
|
|
def test_publisher_and_rule_verification_retries_are_bounded(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Stale reads may retry, but never silently become successful evidence."""
|
|
module = _load(HARBOR, "harbor_retry_coverage")
|
|
monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
|
|
responses = [
|
|
(200, _robot_list(module), _count(1)),
|
|
(200, json.dumps(_robot(module)).encode(), {}),
|
|
(200, b"", {}),
|
|
*((500, b"", {}) for _ in range(5)),
|
|
]
|
|
with pytest.raises(RuntimeError, match="did not verify"):
|
|
module.ensure_publisher_can_read_rule(FakeClient(responses))
|
|
|
|
rule_path = "/api/v2.0/projects/bstein/immutabletagrules/23"
|
|
responses = [
|
|
(200, b"[]", _count(0)),
|
|
(201, b"", {"Location": rule_path}),
|
|
*((200, b"[]", _count(0)) for _ in range(5)),
|
|
]
|
|
with pytest.raises(RuntimeError, match="did not verify"):
|
|
module.ensure_rule(FakeClient(responses))
|
|
|
|
|
|
def test_rule_rejects_ids_create_status_and_location_suffix() -> None:
|
|
"""Neither existing nor newly created rules can omit their exact positive ID."""
|
|
module = _load(HARBOR, "harbor_rule_id_coverage")
|
|
bad_rule = {"id": 0, **module.EXPECTED_RULE}
|
|
with pytest.raises(RuntimeError, match="valid ID"):
|
|
module.ensure_rule(
|
|
FakeClient([(200, json.dumps([bad_rule]).encode(), _count(1))])
|
|
)
|
|
with pytest.raises(module.HarborUnavailable):
|
|
module.ensure_rule(
|
|
FakeClient([(200, b"[]", _count(0)), (503, b"", {})])
|
|
)
|
|
with pytest.raises(RuntimeError, match="invalid ID"):
|
|
module.ensure_rule(
|
|
FakeClient(
|
|
[
|
|
(200, b"[]", _count(0)),
|
|
(
|
|
201,
|
|
b"",
|
|
{
|
|
"location": "/api/v2.0/projects/bstein/immutabletagrules/nope"
|
|
},
|
|
),
|
|
]
|
|
)
|
|
)
|
|
|
|
|
|
def test_harbor_main_retries_then_reports_exact_ids(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""Runtime credentials are required and transient startup failures are bounded."""
|
|
module = _load(HARBOR, "harbor_main_coverage")
|
|
password_file = tmp_path / "password"
|
|
password_file.write_text("private\n", encoding="utf-8")
|
|
monkeypatch.setenv("HARBOR_API_ORIGIN", module.EXPECTED_ORIGIN)
|
|
monkeypatch.setenv("HARBOR_ADMIN_PASSWORD_FILE", str(password_file))
|
|
monkeypatch.setattr(module, "HarborClient", lambda *_args: object())
|
|
attempts = iter([module.HarborUnavailable("warming"), 17])
|
|
|
|
def ensure_rule(_client):
|
|
result = next(attempts)
|
|
if isinstance(result, BaseException):
|
|
raise result
|
|
return result
|
|
|
|
monkeypatch.setattr(module, "ensure_rule", ensure_rule)
|
|
monkeypatch.setattr(module, "ensure_publisher_can_read_rule", lambda _client: 41)
|
|
monkeypatch.setattr(module.time, "sleep", lambda seconds: None)
|
|
assert module.main() == 0
|
|
assert "id=17" in capsys.readouterr().out
|
|
|
|
password_file.write_text("\n", encoding="utf-8")
|
|
with pytest.raises(RuntimeError, match="empty"):
|
|
module.main()
|