147 lines
5.1 KiB
Python
147 lines
5.1 KiB
Python
"""Atlas route and Git protocol contracts for the Hermes SCM broker."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from testing.tests.test_hermes_scm_broker_support import (
|
|
Response,
|
|
_load,
|
|
_receive_command,
|
|
)
|
|
|
|
@pytest.mark.parametrize(
|
|
"target",
|
|
[
|
|
"/api/v1/repos/atlas/cassandra/pulls/1/merge",
|
|
"/api/v1/repos/atlas/cassandra/pulls/1/reviews",
|
|
"/api/v1/admin/users",
|
|
"/git/evil/cassandra.git/info/refs?service=git-upload-pack",
|
|
"/git/atlas/cassandra.git/hooks",
|
|
"/git/atlas/cassandra.git/git-receive-pack?force=true",
|
|
"/git/atlas/cassandra.git/info/refs?service=git-upload-pack\nignored",
|
|
"/git/atlas/cassandra.git/info/refs?service=git-upload-pack\tignored",
|
|
"/git/atlas/cassandra.git/info/refs?service=git-upload-pack#fragment",
|
|
],
|
|
)
|
|
def test_broker_rejects_privileged_or_non_atlas_routes(target: str):
|
|
broker = _load("scm_broker")
|
|
|
|
with pytest.raises(broker.PolicyError):
|
|
broker._git_target(target)
|
|
|
|
|
|
def test_broker_metadata_boundary_reuses_explicit_read_allowlist():
|
|
api = _load("gitea_api")
|
|
|
|
assert (
|
|
api.authorize_request("GET", "/api/v1/repos/atlas/cassandra/pulls/1", None)
|
|
== "pull"
|
|
)
|
|
for path in (
|
|
"/api/v1/repos/atlas/cassandra/pulls/1/merge",
|
|
"/api/v1/repos/atlas/cassandra/pulls/1/reviews",
|
|
"/api/v1/repos/atlas/cassandra/hooks",
|
|
"/api/v1/admin/users",
|
|
):
|
|
with pytest.raises(api.PolicyError):
|
|
api.authorize_request("GET", path, None)
|
|
|
|
|
|
def test_receive_pack_allows_only_new_namespaced_feature_branch():
|
|
broker = _load("scm_broker")
|
|
zero = b"0" * 40
|
|
commit = b"1" * 40
|
|
|
|
broker._validate_receive_pack(
|
|
_receive_command(zero, commit, b"refs/heads/hermes/focused-fix"),
|
|
"runtime-sentinel",
|
|
)
|
|
for old, new, ref in (
|
|
(commit, b"2" * 40, b"refs/heads/hermes/focused-fix"),
|
|
(commit, zero, b"refs/heads/hermes/focused-fix"),
|
|
(zero, commit, b"refs/heads/main"),
|
|
(zero, commit, b"refs/heads/master"),
|
|
(zero, commit, b"refs/tags/release"),
|
|
):
|
|
with pytest.raises(broker.PolicyError):
|
|
broker._validate_receive_pack(
|
|
_receive_command(old, new, ref), "runtime-sentinel"
|
|
)
|
|
|
|
|
|
def test_git_proxy_uses_fixed_origin_and_never_reflects_credential():
|
|
broker = _load("scm_broker")
|
|
seen = []
|
|
|
|
def opener(request, timeout):
|
|
seen.append((request, timeout))
|
|
return Response(
|
|
b"git-result",
|
|
content_type="application/x-git-upload-pack-result",
|
|
)
|
|
|
|
result = broker._upstream_git_request(
|
|
"/atlas/cassandra.git/git-upload-pack",
|
|
method="POST",
|
|
body=b"request",
|
|
content_type="application/x-git-upload-pack-request",
|
|
expected_type="application/x-git-upload-pack-result",
|
|
token="runtime-sentinel",
|
|
opener=opener,
|
|
)
|
|
|
|
assert result == b"git-result"
|
|
request = seen[0][0]
|
|
assert request.full_url == (
|
|
"https://scm.bstein.dev/atlas/cassandra.git/git-upload-pack"
|
|
)
|
|
assert b"runtime-sentinel" not in request.data
|
|
assert "runtime-sentinel" not in request.full_url
|
|
|
|
with pytest.raises(broker.PolicyError, match="credential material"):
|
|
broker._upstream_git_request(
|
|
"/atlas/cassandra.git/git-upload-pack",
|
|
method="POST",
|
|
body=b"request",
|
|
content_type="application/x-git-upload-pack-request",
|
|
expected_type="application/x-git-upload-pack-result",
|
|
token="runtime-sentinel",
|
|
opener=lambda *_a, **_k: Response(
|
|
b"runtime-sentinel",
|
|
content_type="application/x-git-upload-pack-result",
|
|
),
|
|
)
|
|
encoded = broker._credential_forms("runtime-sentinel")[1]
|
|
with pytest.raises(broker.PolicyError, match="credential material"):
|
|
broker._upstream_git_request(
|
|
"/atlas/cassandra.git/git-upload-pack",
|
|
method="POST",
|
|
body=b"request",
|
|
content_type="application/x-git-upload-pack-request",
|
|
expected_type="application/x-git-upload-pack-result",
|
|
token="runtime-sentinel",
|
|
opener=lambda *_a, **_k: Response(
|
|
b"Authorization: Basic " + encoded,
|
|
content_type="application/x-git-upload-pack-result",
|
|
),
|
|
)
|
|
@pytest.mark.parametrize("status", [201, 202, 204, 206, 301, 302, 307, 308])
|
|
def test_git_proxy_requires_exact_http_200(status: int):
|
|
broker = _load("scm_broker")
|
|
|
|
with pytest.raises(broker.PolicyError, match="unexpected HTTP status"):
|
|
broker._upstream_git_request(
|
|
"/atlas/cassandra.git/git-upload-pack",
|
|
method="POST",
|
|
body=b"request",
|
|
content_type="application/x-git-upload-pack-request",
|
|
expected_type="application/x-git-upload-pack-result",
|
|
token="runtime-sentinel",
|
|
opener=lambda *_a, **_k: Response(
|
|
b"result",
|
|
status=status,
|
|
content_type="application/x-git-upload-pack-result",
|
|
),
|
|
)
|