381 lines
13 KiB
Python
381 lines
13 KiB
Python
"""Behavioral coverage for internal safe-Gitea client and policy paths."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import types
|
|
import urllib.error
|
|
import urllib.parse
|
|
from email.message import Message
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from testing.tests.test_hermes_gitea_support import CLIENT_PATH, Response, _load
|
|
from testing.tests.test_hermes_scm_broker_support import _load_path
|
|
|
|
|
|
def _policy(name: str = "gitea_policy_coverage"):
|
|
return _load_path(name, CLIENT_PATH.parent / "gitea_api_policy.py")
|
|
|
|
|
|
def test_runtime_token_reader_and_fixed_origin(tmp_path: Path, monkeypatch):
|
|
module = _load()
|
|
token = tmp_path / "token"
|
|
token.write_text(" synthetic-runtime-value \n", encoding="utf-8")
|
|
assert module.read_token(token) == "synthetic-runtime-value"
|
|
token.write_text(" \n", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="empty"):
|
|
module.read_token(token)
|
|
|
|
monkeypatch.setenv("GITEA_BASE_URL", module.CANONICAL_BASE_URL)
|
|
assert module.configured_base_url() == module.CANONICAL_BASE_URL
|
|
monkeypatch.setenv("GITEA_BASE_URL", "https://example.invalid")
|
|
with pytest.raises(module.PolicyError, match="fixed"):
|
|
module.configured_base_url()
|
|
|
|
|
|
def test_safe_urlopen_delegates_only_to_deadline_bounded_exchange(monkeypatch):
|
|
module = _load()
|
|
sentinel = object()
|
|
monkeypatch.setattr(
|
|
module.deadline_http,
|
|
"open_bounded",
|
|
lambda request, *, maximum, timeout: (request, maximum, timeout, sentinel),
|
|
)
|
|
request = object()
|
|
assert module._safe_urlopen(request, 7) == (
|
|
request,
|
|
module.MAX_RESPONSE_BYTES,
|
|
7,
|
|
sentinel,
|
|
)
|
|
|
|
|
|
def test_api_target_and_request_body_edge_paths(monkeypatch):
|
|
module = _load()
|
|
with pytest.raises(module.PolicyError, match="path exceeds"):
|
|
module._split_api_path("/api/v1/" + "x" * 510)
|
|
with pytest.raises(module.PolicyError, match="request body"):
|
|
module.authorize_request(
|
|
"GET", "/api/v1/repos/titan/cassandra", {"unexpected": True}
|
|
)
|
|
with pytest.raises(module.PolicyError, match="query"):
|
|
module.authorize_request(
|
|
"POST",
|
|
"/api/v1/repos/titan/cassandra/pulls?page=1",
|
|
{},
|
|
)
|
|
with pytest.raises(module.PolicyError, match="only read"):
|
|
module.authorize_request("TRACE", "/api/v1/repos/titan/cassandra/pulls", None)
|
|
with pytest.raises(module.PolicyError, match="accepts only"):
|
|
module.authorize_request("POST", "/api/v1/repos/titan/cassandra/pulls", [])
|
|
|
|
|
|
def test_response_status_fallback_and_nested_fail_closed():
|
|
module = _load()
|
|
|
|
class GetCodeResponse(Response):
|
|
def __init__(self):
|
|
super().__init__({"name": "cassandra"}, status=200)
|
|
del self.status
|
|
|
|
def getcode(self):
|
|
return 200
|
|
|
|
assert (
|
|
module.read(
|
|
"/api/v1/repos/titan/cassandra",
|
|
token="synthetic",
|
|
opener=lambda *_a, **_k: GetCodeResponse(),
|
|
)
|
|
== b'{"name": "cassandra"}'
|
|
)
|
|
with pytest.raises(module.PolicyError, match="omitted required"):
|
|
module._nested({"base": None}, "base", "ref")
|
|
|
|
|
|
@pytest.mark.parametrize("body", [b"not-json", b"[]"])
|
|
def test_create_response_requires_json_object(body: bytes):
|
|
module = _load()
|
|
with pytest.raises(module.PolicyError, match="invalid pull-request metadata"):
|
|
module._require_create_response(
|
|
body,
|
|
repo="cassandra",
|
|
base="main",
|
|
head="feature/test",
|
|
head_sha="a" * 40,
|
|
title="WIP: Test",
|
|
body="Evidence",
|
|
)
|
|
|
|
|
|
def test_write_body_handles_empty_newline_and_missing_newline(monkeypatch):
|
|
module = _load()
|
|
|
|
class Buffer:
|
|
value = bytearray()
|
|
|
|
@classmethod
|
|
def write(cls, value):
|
|
cls.value.extend(value)
|
|
|
|
monkeypatch.setattr(module.sys, "stdout", types.SimpleNamespace(buffer=Buffer))
|
|
module._write_body(b"")
|
|
module._write_body(b"one\n")
|
|
module._write_body(b"two")
|
|
assert bytes(Buffer.value) == b"one\ntwo\n"
|
|
|
|
|
|
def test_main_executes_broker_read_and_create_paths(monkeypatch):
|
|
module = _load()
|
|
outputs: list[bytes] = []
|
|
client = types.ModuleType("scm_broker_client")
|
|
client.read = lambda path: json.dumps({"path": path}).encode()
|
|
client.create_draft = lambda repo, **data: json.dumps(
|
|
{"repo": repo, **data}, sort_keys=True
|
|
).encode()
|
|
monkeypatch.setitem(sys.modules, "scm_broker_client", client)
|
|
monkeypatch.setattr(module, "_write_body", outputs.append)
|
|
|
|
assert module.main(["read", "/api/v1/repos/titan/cassandra"]) == 0
|
|
assert json.loads(outputs.pop()) == {"path": "/api/v1/repos/titan/cassandra"}
|
|
assert (
|
|
module.main(
|
|
[
|
|
"create-draft",
|
|
"cassandra",
|
|
"--base",
|
|
"main",
|
|
"--head",
|
|
"feature/coverage",
|
|
"--head-sha",
|
|
"a" * 40,
|
|
"--title",
|
|
"Coverage repair",
|
|
"--body",
|
|
"Review the focused tests",
|
|
]
|
|
)
|
|
== 0
|
|
)
|
|
assert json.loads(outputs.pop())["repo"] == "cassandra"
|
|
|
|
|
|
def test_main_handles_http_and_policy_failures(monkeypatch, capsys):
|
|
module = _load()
|
|
client = types.ModuleType("scm_broker_client")
|
|
|
|
def http_failure(_path):
|
|
headers = Message()
|
|
raise urllib.error.HTTPError("url", 503, "unavailable", headers, None)
|
|
|
|
client.read = http_failure
|
|
client.create_draft = lambda *_a, **_k: b"{}"
|
|
monkeypatch.setitem(sys.modules, "scm_broker_client", client)
|
|
assert module.main(["read", "/api/v1/repos/titan/cassandra"]) == 1
|
|
assert "HTTP 503" in capsys.readouterr().err
|
|
client.read = lambda _path: (_ for _ in ()).throw(module.PolicyError("rejected"))
|
|
assert module.main(["read", "/not-allowed"]) == 1
|
|
assert "no credential" in capsys.readouterr().err
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("function", "value"),
|
|
[
|
|
("_validate_repo", "."),
|
|
("_validate_repo", ".."),
|
|
("_validate_sha", "short"),
|
|
("_validate_pr_number", True),
|
|
("_validate_pr_number", 0),
|
|
("_validate_pr_number", 2_147_483_648),
|
|
],
|
|
)
|
|
def test_policy_scalar_validators_reject_ambiguous_values(function, value):
|
|
module = _policy("gitea_policy_scalar")
|
|
with pytest.raises(module.PolicyError):
|
|
getattr(module, function)(value)
|
|
|
|
|
|
def test_ref_bounds_cover_utf8_controls_git_failure_and_success(monkeypatch):
|
|
module = _policy("gitea_policy_refs")
|
|
with pytest.raises(module.PolicyError, match="valid UTF-8"):
|
|
module._validate_ref_bounds("\ud800", "head")
|
|
with pytest.raises(module.PolicyError, match="UTF-8"):
|
|
module._validate_ref_bounds("🧪" * 64, "head")
|
|
with pytest.raises(module.PolicyError, match="safe same"):
|
|
module._validate_ref("bad\nref", "head")
|
|
monkeypatch.setattr(
|
|
module.subprocess,
|
|
"run",
|
|
lambda *_a, **_k: subprocess.CompletedProcess([], 1),
|
|
)
|
|
with pytest.raises(module.PolicyError, match="same-repository"):
|
|
module._validate_ref("invalid-ref", "head")
|
|
monkeypatch.setattr(
|
|
module.subprocess,
|
|
"run",
|
|
lambda *_a, **_k: subprocess.CompletedProcess([], 0),
|
|
)
|
|
assert module._validate_ref("feature/valid", "head") == "feature/valid"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("value", "required", "match"),
|
|
[
|
|
(None, False, "must be text"),
|
|
(" ", True, "must not be empty"),
|
|
("\ud800", False, "valid UTF-8"),
|
|
("x\x00y", False, "safe request limit"),
|
|
],
|
|
)
|
|
def test_text_validation_rejects_nontext_empty_invalid_and_controls(
|
|
value, required, match
|
|
):
|
|
module = _policy("gitea_policy_text")
|
|
with pytest.raises(module.PolicyError, match=match):
|
|
module._validate_text(value, "field", 20, 20, required=required)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("key", "expected"),
|
|
[
|
|
("password", True),
|
|
("accountKey", True),
|
|
("registry-key", True),
|
|
("clientEmail", True),
|
|
("clientId", True),
|
|
("accessId", True),
|
|
("connectionString", True),
|
|
("dockerConfigJson", True),
|
|
("release_note", False),
|
|
],
|
|
)
|
|
def test_sensitive_key_semantics_cover_generic_forms(key, expected):
|
|
module = _policy("gitea_policy_keys")
|
|
assert module._is_sensitive_key(key) is expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("value", "expected"),
|
|
[
|
|
("", False),
|
|
('"synthetic"', True),
|
|
("{encoded}", True),
|
|
("Bearer value-12345678", True),
|
|
("ghp_" + "a" * 24, True),
|
|
("https://example.invalid/value", True),
|
|
("name@example.invalid", True),
|
|
("${RUNTIME_VALUE}", True),
|
|
("AbCdEf0123456789", True),
|
|
("reject empty values", False),
|
|
("ordinary", True),
|
|
],
|
|
)
|
|
def test_assignment_value_shape_distinguishes_prose_from_credentials(value, expected):
|
|
module = _policy("gitea_policy_assignment_values")
|
|
assert module._looks_sensitive_assignment_value(value) is expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("value", "expected"),
|
|
[
|
|
("single", False),
|
|
("add regression coverage", True),
|
|
("add regression_coverage", False),
|
|
("ordinary engineering prose", False),
|
|
],
|
|
)
|
|
def test_prose_classifier_covers_length_punctuation_and_lead_words(value, expected):
|
|
module = _policy("gitea_policy_prose")
|
|
assert module._looks_like_prose(value) is expected
|
|
|
|
|
|
def test_structured_json_walk_covers_lists_nonstring_keys_and_limits():
|
|
module = _policy("gitea_policy_json_walk")
|
|
assert module._json_value_has_sensitive_assignment({1: "ignored"}) is False
|
|
assert module._json_value_has_sensitive_assignment([{"release": "safe"}]) is False
|
|
assert (
|
|
module._json_value_has_sensitive_assignment({"client_secret": "value"}) is True
|
|
)
|
|
assert (
|
|
module._json_value_has_sensitive_assignment(
|
|
{"outer": {"client_secret": "value"}}
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
module._json_value_has_sensitive_assignment({"type": "service-account"}) is True
|
|
)
|
|
with pytest.raises(module.PolicyError, match="scan limit"):
|
|
module._json_value_has_sensitive_assignment([], depth=33)
|
|
with pytest.raises(module.PolicyError, match="scan limit"):
|
|
module._json_value_has_sensitive_assignment([], nodes=[2048])
|
|
|
|
|
|
def test_json_key_decoder_and_embedded_document_scanner_cover_failures():
|
|
module = _policy("gitea_policy_json_decoder")
|
|
assert module._decode_json_key("client\\u005fsecret") == ("client_secret", True)
|
|
decoded, valid = module._decode_json_key("client\\qsecret")
|
|
assert valid is False and "client" in decoded
|
|
assert list(module._decoded_json_documents("prose only")) == []
|
|
assert list(module._decoded_json_documents("x {bad y [1, 2]")) == [[1, 2]]
|
|
with pytest.raises(module.PolicyError, match="scan limit"):
|
|
list(module._decoded_json_documents("{" * 33))
|
|
assert (
|
|
module._has_structured_sensitive_assignment('"client\\qsecret": value') is True
|
|
)
|
|
assert (
|
|
module._has_structured_sensitive_assignment("type:\n service-account") is True
|
|
)
|
|
|
|
|
|
def test_high_entropy_detector_covers_long_mixed_and_low_entropy_values():
|
|
module = _policy("gitea_policy_entropy")
|
|
assert module._has_high_entropy_token("A1_" * 100) is True
|
|
assert (
|
|
module._has_high_entropy_token(
|
|
"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/_="
|
|
)
|
|
is True
|
|
)
|
|
assert module._has_high_entropy_token("a" * 64) is False
|
|
assert module._has_high_entropy_token("short prose") is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"query",
|
|
[
|
|
"bad",
|
|
"a=1&b=2&c=3&d=4",
|
|
"page=1&page=2",
|
|
"=value",
|
|
"unexpected=1",
|
|
"page=abc",
|
|
"page=01",
|
|
"page=10001",
|
|
"limit=51",
|
|
"state=merged",
|
|
],
|
|
)
|
|
def test_query_validator_covers_each_rejection_class(query):
|
|
module = _policy("gitea_policy_query")
|
|
target = urllib.parse.urlsplit("/api/v1/repos/titan/cassandra/pulls?" + query)
|
|
with pytest.raises(module.PolicyError):
|
|
module._validate_query(target, {"page", "limit", "state"})
|
|
|
|
|
|
def test_draft_title_rejects_prefix_without_content():
|
|
module = _policy("gitea_policy_empty_draft")
|
|
with pytest.raises(module.PolicyError, match="after the draft prefix"):
|
|
module._draft_title("WIP:")
|
|
|
|
|
|
def test_query_validator_rejects_noncanonical_raw_form():
|
|
module = _policy("gitea_policy_raw_query")
|
|
target = urllib.parse.SplitResult("", "", "/api/v1/repos/titan/cassandra", "%", "")
|
|
with pytest.raises(module.PolicyError, match="canonical ASCII"):
|
|
module._validate_query(target, set())
|