406 lines
12 KiB
Python
406 lines
12 KiB
Python
"""Adversarial contracts for the Atlas-only draft pull-request client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).parents[2]
|
|
CLIENT_PATH = ROOT / "services/hermes/scripts/gitea_api.py"
|
|
HEAD_SHA = "465cf9146b05c174a2a8d310aff6c64be58277b6"
|
|
|
|
|
|
def _load():
|
|
spec = importlib.util.spec_from_file_location("safe_gitea_api", CLIENT_PATH)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _draft_payload(**updates):
|
|
payload = {
|
|
"base": "main",
|
|
"body": "Review evidence",
|
|
"head": "hermes/review-fix",
|
|
"title": "WIP: Repair review findings",
|
|
}
|
|
payload.update(updates)
|
|
return payload
|
|
|
|
|
|
def _draft_response(**updates):
|
|
response = {
|
|
"number": 3,
|
|
"state": "open",
|
|
"draft": True,
|
|
"merged": False,
|
|
"html_url": "https://scm.bstein.dev/atlas/cassandra/pulls/3",
|
|
"url": "https://scm.bstein.dev/atlas/cassandra/pulls/3",
|
|
"title": "WIP: Focused fix",
|
|
"body": "Review evidence",
|
|
"base": {"ref": "main", "repo": {"full_name": "atlas/cassandra"}},
|
|
"head": {
|
|
"ref": "hermes/fix",
|
|
"sha": HEAD_SHA,
|
|
"repo": {"full_name": "atlas/cassandra"},
|
|
},
|
|
}
|
|
response.update(updates)
|
|
return response
|
|
|
|
|
|
class Response:
|
|
def __init__(self, body: object):
|
|
self.body = body if isinstance(body, bytes) else json.dumps(body).encode()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self, limit=-1):
|
|
return self.body if limit < 0 else self.body[:limit]
|
|
|
|
|
|
def test_redirect_handler_rejects_cross_origin_with_sentinel_authorization():
|
|
client = _load()
|
|
source = urllib.request.Request(
|
|
"https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
|
|
headers={"Authorization": "token redirect-sentinel"},
|
|
)
|
|
|
|
with pytest.raises(client.PolicyError, match="redirects are not allowed") as exc:
|
|
client.RejectRedirectHandler().redirect_request(
|
|
source,
|
|
None,
|
|
302,
|
|
"Found",
|
|
{},
|
|
"https://evil.example/collect",
|
|
)
|
|
|
|
assert "redirect-sentinel" not in str(exc.value)
|
|
assert any(
|
|
isinstance(handler, client.RejectRedirectHandler)
|
|
for handler in client._SAFE_OPENER.handlers
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("base_url", "path"),
|
|
[
|
|
("https://evil.example", "/api/v1/repos/atlas/cassandra"),
|
|
("http://scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
|
|
("https://scm.bstein.dev:443", "/api/v1/repos/atlas/cassandra"),
|
|
("https://scm.bstein.dev", "https://evil.example/api/v1/repos/atlas/cassandra"),
|
|
("https://scm.bstein.dev", "/api/v1/repos/evil/cassandra"),
|
|
("https://scm.bstein.dev", "/api/v1/repos/%61tlas/cassandra"),
|
|
("https://scm.bstein.dev", "/api/v1/repos/atlas/../admin"),
|
|
],
|
|
)
|
|
def test_host_owner_and_path_escape_attempts_are_rejected(base_url: str, path: str):
|
|
client = _load()
|
|
|
|
with pytest.raises(client.PolicyError):
|
|
client.build_request("GET", path, base_url=base_url, token="secret")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"/api/v1/repos/atlas/cassandra",
|
|
"/api/v1/repos/atlas/cassandra/pulls?state=open&limit=20&page=1",
|
|
"/api/v1/repos/atlas/cassandra/pulls/7",
|
|
"/api/v1/repos/atlas/cassandra/pulls/7/commits?limit=20",
|
|
"/api/v1/repos/atlas/cassandra/pulls/7/files?page=1",
|
|
"/api/v1/repos/atlas/cassandra/branches",
|
|
"/api/v1/repos/atlas/cassandra/branches/main",
|
|
"/api/v1/repos/atlas/cassandra/commits?limit=10",
|
|
f"/api/v1/repos/atlas/cassandra/git/commits/{HEAD_SHA}",
|
|
f"/api/v1/repos/atlas/cassandra/commits/{HEAD_SHA}/status",
|
|
f"/api/v1/repos/atlas/cassandra/commits/{HEAD_SHA}/statuses?limit=10",
|
|
f"/api/v1/repos/atlas/cassandra/statuses/{HEAD_SHA}?page=1",
|
|
],
|
|
)
|
|
def test_explicit_read_allowlist_accepts_only_engineering_metadata(path: str):
|
|
client = _load()
|
|
|
|
request = client.build_request(
|
|
"GET", path, base_url=client.CANONICAL_BASE_URL, token="runtime"
|
|
)
|
|
assert request.method == "GET"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"/api/v1/repos/atlas/cassandra/hooks",
|
|
"/api/v1/repos/atlas/cassandra/actions/secrets",
|
|
"/api/v1/repos/atlas/cassandra/actions/variables",
|
|
"/api/v1/repos/atlas/cassandra/collaborators",
|
|
"/api/v1/repos/atlas/cassandra/branch_protections",
|
|
"/api/v1/repos/atlas/cassandra/keys",
|
|
"/api/v1/repos/atlas/cassandra/pulls/7/reviews",
|
|
"/api/v1/repos/atlas/cassandra/pulls/7/merge",
|
|
"/api/v1/repos/atlas/cassandra/pulls/7.diff",
|
|
"/api/v1/repos/atlas/cassandra/releases",
|
|
],
|
|
)
|
|
def test_privileged_or_content_routes_are_denied_even_for_get(path: str):
|
|
client = _load()
|
|
|
|
with pytest.raises(client.PolicyError, match="outside the metadata read allowlist"):
|
|
client.authorize_request("GET", path, None)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"/api/v1/repos/atlas/cassandra/pulls?limit=51",
|
|
"/api/v1/repos/atlas/cassandra/pulls?state=merged",
|
|
"/api/v1/repos/atlas/cassandra/pulls?private=true",
|
|
"/api/v1/repos/atlas/cassandra/pulls?limit=1&limit=2",
|
|
"/api/v1/repos/atlas/cassandra?p=1",
|
|
],
|
|
)
|
|
def test_read_query_is_bounded(path: str):
|
|
client = _load()
|
|
|
|
with pytest.raises(client.PolicyError):
|
|
client.authorize_request("GET", path, None)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "path", "data"),
|
|
[
|
|
("DELETE", "/api/v1/repos/atlas/cassandra/pulls/4", None),
|
|
("POST", "/api/v1/repos/atlas/cassandra/pulls/4/merge", {}),
|
|
(
|
|
"POST",
|
|
"/api/v1/repos/atlas/cassandra/pulls/4/reviews",
|
|
{"event": "APPROVED"},
|
|
),
|
|
("PATCH", "/api/v1/repos/atlas/cassandra/pulls/4", {"title": "WIP: x"}),
|
|
("PUT", "/api/v1/repos/atlas/cassandra/branches/main", {}),
|
|
],
|
|
)
|
|
def test_merge_approve_close_delete_update_and_other_mutations_are_rejected(
|
|
method: str, path: str, data: object
|
|
):
|
|
client = _load()
|
|
|
|
with pytest.raises(client.PolicyError):
|
|
client.build_request(
|
|
method, path, base_url=client.CANONICAL_BASE_URL, token="secret", data=data
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"ref",
|
|
[
|
|
"foo/.bar",
|
|
"foo/bar.lock/baz",
|
|
"foo..bar",
|
|
"foo@{bar",
|
|
"foo//bar",
|
|
"-danger",
|
|
"danger.",
|
|
"danger~one",
|
|
"danger^one",
|
|
"danger:one",
|
|
"danger one",
|
|
],
|
|
)
|
|
def test_complete_git_ref_validation_rejects_invalid_names(ref: str):
|
|
client = _load()
|
|
|
|
with pytest.raises(client.PolicyError):
|
|
client._validate_ref(ref, "head")
|
|
|
|
|
|
def test_git_ref_validation_uses_fixed_trusted_binary():
|
|
client = _load()
|
|
|
|
assert client.GIT_BIN == "/usr/bin/git"
|
|
assert client._validate_ref("hermes/valid-fix", "head") == "hermes/valid-fix"
|
|
|
|
|
|
def test_create_forces_draft_title_and_same_repository_branch_names():
|
|
client = _load()
|
|
assert (
|
|
client.authorize_request(
|
|
"POST", "/api/v1/repos/atlas/cassandra/pulls", _draft_payload()
|
|
)
|
|
== "create-draft"
|
|
)
|
|
with pytest.raises(client.PolicyError, match="draft-title prefix"):
|
|
client.authorize_request(
|
|
"POST",
|
|
"/api/v1/repos/atlas/cassandra/pulls",
|
|
_draft_payload(title="Not a draft"),
|
|
)
|
|
with pytest.raises(client.PolicyError):
|
|
client.authorize_request(
|
|
"POST",
|
|
"/api/v1/repos/atlas/cassandra/pulls",
|
|
_draft_payload(head="someone:branch"),
|
|
)
|
|
|
|
|
|
def test_runtime_token_is_only_an_authorization_header():
|
|
client = _load()
|
|
request = client.build_request(
|
|
"POST",
|
|
"/api/v1/repos/atlas/cassandra/pulls",
|
|
base_url=client.CANONICAL_BASE_URL,
|
|
token="do-not-leak",
|
|
data=_draft_payload(),
|
|
)
|
|
|
|
assert "do-not-leak" not in request.full_url
|
|
assert b"do-not-leak" not in request.data
|
|
assert request.get_header("Authorization") == "token do-not-leak"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body",
|
|
[
|
|
"pass" + "word=not-a-real-credential",
|
|
"Authorization: " + "Bearer not-a-real-credential-value",
|
|
"token: " + "ghp_" + "notarealcredentialvalue123456",
|
|
"-----BEGIN OPENSSH " + "PRIVATE KEY-----",
|
|
"eyJnotarealheader." + "notarealpayloadvalue." + "notarealsignature",
|
|
],
|
|
)
|
|
def test_body_rejects_common_credential_shapes(body: str):
|
|
client = _load()
|
|
|
|
with pytest.raises(client.PolicyError, match="credential material"):
|
|
client._validate_body(body)
|
|
|
|
|
|
def test_create_rejects_exact_runtime_token_before_network():
|
|
client = _load()
|
|
called = False
|
|
|
|
def opener(*_args, **_kwargs):
|
|
nonlocal called
|
|
called = True
|
|
return Response(_draft_response())
|
|
|
|
with pytest.raises(client.PolicyError, match="runtime credential"):
|
|
client.create_draft(
|
|
"cassandra",
|
|
base="main",
|
|
head="hermes/fix",
|
|
head_sha=HEAD_SHA,
|
|
title="Focused fix",
|
|
body="accidental runtime-sentinel value",
|
|
token="runtime-sentinel",
|
|
opener=opener,
|
|
)
|
|
assert called is False
|
|
|
|
|
|
def test_create_verifies_every_server_postcondition():
|
|
client = _load()
|
|
calls = []
|
|
|
|
def opener(request, timeout):
|
|
calls.append((request, timeout))
|
|
return Response(_draft_response())
|
|
|
|
result = client.create_draft(
|
|
"cassandra",
|
|
base="main",
|
|
head="hermes/fix",
|
|
head_sha=HEAD_SHA,
|
|
title="Focused fix",
|
|
body="Review evidence",
|
|
token="runtime",
|
|
opener=opener,
|
|
)
|
|
|
|
assert json.loads(result) == _draft_response()
|
|
payload = json.loads(calls[0][0].data)
|
|
assert payload == {
|
|
"base": "main",
|
|
"body": "Review evidence",
|
|
"head": "hermes/fix",
|
|
"title": "WIP: Focused fix",
|
|
}
|
|
assert calls[0][1] == 30
|
|
|
|
|
|
def test_create_postcondition_rejects_every_material_mismatch():
|
|
client = _load()
|
|
mutations = [
|
|
("number", 0),
|
|
("state", "closed"),
|
|
("draft", False),
|
|
("merged", True),
|
|
("html_url", "https://evil.example/pulls/3"),
|
|
("url", "https://evil.example/api/pulls/3"),
|
|
("title", "Focused fix"),
|
|
("body", "different"),
|
|
]
|
|
documents = []
|
|
for key, value in mutations:
|
|
document = _draft_response()
|
|
document[key] = value
|
|
documents.append(document)
|
|
for path, value in [
|
|
(("base", "ref"), "master"),
|
|
(("base", "repo", "full_name"), "evil/cassandra"),
|
|
(("head", "ref"), "other"),
|
|
(("head", "sha"), "0" * 40),
|
|
(("head", "repo", "full_name"), "evil/cassandra"),
|
|
]:
|
|
document = copy.deepcopy(_draft_response())
|
|
target = document
|
|
for key in path[:-1]:
|
|
target = target[key]
|
|
target[path[-1]] = value
|
|
documents.append(document)
|
|
|
|
for document in documents:
|
|
with pytest.raises(client.PolicyError):
|
|
client._require_create_response(
|
|
json.dumps(document).encode(),
|
|
repo="cassandra",
|
|
base="main",
|
|
head="hermes/fix",
|
|
head_sha=HEAD_SHA,
|
|
title="WIP: Focused fix",
|
|
body="Review evidence",
|
|
)
|
|
|
|
|
|
def test_read_response_is_bounded():
|
|
client = _load()
|
|
|
|
with pytest.raises(client.PolicyError, match="safe size limit"):
|
|
client.read(
|
|
"/api/v1/repos/atlas/cassandra",
|
|
token="runtime",
|
|
opener=lambda *_a, **_k: Response(b"x" * (client.MAX_RESPONSE_BYTES + 1)),
|
|
)
|
|
|
|
|
|
def test_output_redaction_covers_exact_token_and_authorization_header():
|
|
client = _load()
|
|
raw = b'{"message":"do-not-leak","debug":"Authorization: token do-not-leak"}'
|
|
redacted = client.redact_bytes(raw, "do-not-leak")
|
|
|
|
assert b"do-not-leak" not in redacted
|
|
assert redacted.count(b"[REDACTED]") >= 1
|