275 lines
10 KiB
Python
275 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import importlib.util
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SKILL_ROOT = ROOT / "skills" / "gitea-fix-ci"
|
|
SCRIPT = SKILL_ROOT / "scripts" / "fetch_ci_logs.py"
|
|
|
|
|
|
def load_module():
|
|
spec = importlib.util.spec_from_file_location("fetch_ci_logs", SCRIPT)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"cannot load {SCRIPT}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class FakeResponse:
|
|
status = 200
|
|
headers = {"Content-Type": "application/json"}
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self) -> bytes:
|
|
return b'{"ok": true}'
|
|
|
|
|
|
class RecordingOpener:
|
|
def __init__(self) -> None:
|
|
self.requests = []
|
|
|
|
def open(self, request, *, timeout):
|
|
self.requests.append((request, timeout))
|
|
return FakeResponse()
|
|
|
|
|
|
class GiteaFixCiSkillTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.module = load_module()
|
|
|
|
def setUp(self) -> None:
|
|
self.target = self.module.RepoTarget(
|
|
base_url="https://git.example.test",
|
|
owner="team",
|
|
repo="project",
|
|
)
|
|
|
|
def test_token_auth_takes_precedence_without_reading_git_credentials(self):
|
|
args = SimpleNamespace(use_git_credential=True)
|
|
with mock.patch.dict(os.environ, {"GITEA_TOKEN": "token-secret"}, clear=True):
|
|
with mock.patch.object(
|
|
self.module,
|
|
"_git_credential_auth",
|
|
side_effect=AssertionError("git credential must not run"),
|
|
):
|
|
auth = self.module.resolve_auth(args, self.target)
|
|
|
|
self.assertEqual(auth.source, "GITEA_TOKEN")
|
|
self.assertEqual(auth.authorization, "token token-secret")
|
|
self.assertNotIn("token-secret", repr(auth))
|
|
|
|
def test_git_credential_auth_is_explicit_scoped_and_non_interactive(self):
|
|
credential = subprocess.CompletedProcess(
|
|
args=["git", "credential", "fill"],
|
|
returncode=0,
|
|
stdout=(
|
|
"protocol=https\n"
|
|
"host=git.example.test\n"
|
|
"username=ci-user\n"
|
|
"password=basic-secret\n"
|
|
),
|
|
stderr="",
|
|
)
|
|
args = SimpleNamespace(use_git_credential=True)
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
with mock.patch.object(
|
|
self.module.subprocess, "run", return_value=credential
|
|
) as run:
|
|
auth = self.module.resolve_auth(args, self.target)
|
|
|
|
self.assertEqual(auth.source, "git credential")
|
|
expected = base64.b64encode(b"ci-user:basic-secret").decode("ascii")
|
|
self.assertEqual(auth.authorization, f"Basic {expected}")
|
|
self.assertNotIn("basic-secret", repr(auth))
|
|
|
|
call = run.call_args
|
|
self.assertEqual(call.args[0], ["git", "credential", "fill"])
|
|
self.assertIn("protocol=https\n", call.kwargs["input"])
|
|
self.assertIn("host=git.example.test\n", call.kwargs["input"])
|
|
self.assertIn("path=team/project.git\n", call.kwargs["input"])
|
|
self.assertEqual(call.kwargs["env"]["GIT_TERMINAL_PROMPT"], "0")
|
|
self.assertEqual(call.kwargs["env"]["GCM_INTERACTIVE"], "Never")
|
|
|
|
def test_git_credential_scope_includes_gitea_base_path(self):
|
|
target = self.module.RepoTarget(
|
|
base_url="https://git.example.test/gitea",
|
|
owner="team",
|
|
repo="project",
|
|
)
|
|
self.assertEqual(target.credential_path, "gitea/team/project.git")
|
|
|
|
def test_git_credentials_are_not_read_without_explicit_flag(self):
|
|
args = SimpleNamespace(use_git_credential=False)
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
with mock.patch.object(
|
|
self.module.subprocess,
|
|
"run",
|
|
side_effect=AssertionError("git credential must not run"),
|
|
):
|
|
auth = self.module.resolve_auth(args, self.target)
|
|
self.assertIsNone(auth)
|
|
|
|
def test_all_credentials_require_https(self):
|
|
target = self.module.RepoTarget(
|
|
base_url="http://git.example.test",
|
|
owner="team",
|
|
repo="project",
|
|
)
|
|
with mock.patch.dict(os.environ, {"GITEA_TOKEN": "secret"}, clear=True):
|
|
with self.assertRaisesRegex(self.module.ConfigError, "HTTPS"):
|
|
self.module.resolve_auth(
|
|
SimpleNamespace(use_git_credential=False), target
|
|
)
|
|
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaisesRegex(self.module.ConfigError, "HTTPS"):
|
|
self.module.resolve_auth(
|
|
SimpleNamespace(use_git_credential=True), target
|
|
)
|
|
|
|
auth = self.module.ApiAuth(
|
|
source="GITEA_TOKEN",
|
|
authorization="token sensitive-value",
|
|
)
|
|
with self.assertRaisesRegex(self.module.ConfigError, "HTTPS"):
|
|
self.module.ApiClient(target, auth=auth, opener=RecordingOpener())
|
|
|
|
def test_invalid_token_header_is_rejected_without_echoing_secret(self):
|
|
secret = "token-secret\ninjected-header"
|
|
with mock.patch.dict(os.environ, {"GITEA_TOKEN": secret}, clear=True):
|
|
with self.assertRaises(self.module.ConfigError) as raised:
|
|
self.module.resolve_auth(
|
|
SimpleNamespace(use_git_credential=False), self.target
|
|
)
|
|
self.assertNotIn("token-secret", str(raised.exception))
|
|
self.assertNotIn("injected-header", str(raised.exception))
|
|
|
|
def test_missing_git_credential_fails_without_echoing_helper_output(self):
|
|
credential = subprocess.CompletedProcess(
|
|
args=["git", "credential", "fill"],
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="helper diagnostic containing basic-secret",
|
|
)
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
with mock.patch.object(
|
|
self.module.subprocess, "run", return_value=credential
|
|
):
|
|
with self.assertRaises(self.module.ConfigError) as raised:
|
|
self.module.resolve_auth(
|
|
SimpleNamespace(use_git_credential=True), self.target
|
|
)
|
|
self.assertNotIn("basic-secret", str(raised.exception))
|
|
|
|
def test_git_credential_timeout_does_not_echo_helper_output(self):
|
|
timeout = subprocess.TimeoutExpired(
|
|
cmd=["git", "credential", "fill"],
|
|
timeout=10,
|
|
output="username=ci-user\npassword=basic-secret\n",
|
|
stderr="helper diagnostic containing basic-secret",
|
|
)
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
with mock.patch.object(self.module.subprocess, "run", side_effect=timeout):
|
|
with self.assertRaises(self.module.ConfigError) as raised:
|
|
self.module.resolve_auth(
|
|
SimpleNamespace(use_git_credential=True), self.target
|
|
)
|
|
self.assertNotIn("basic-secret", str(raised.exception))
|
|
|
|
def test_api_client_sends_auth_only_to_the_configured_origin(self):
|
|
opener = RecordingOpener()
|
|
auth = self.module.ApiAuth(
|
|
source="git credential",
|
|
authorization="Basic sensitive-value",
|
|
)
|
|
client = self.module.ApiClient(self.target, auth=auth, opener=opener)
|
|
|
|
status, body, _ = client.request(
|
|
self.target.repo_path("/actions/runs"), accept_json=True
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(body, b'{"ok": true}')
|
|
request, timeout = opener.requests[0]
|
|
self.assertEqual(request.get_header("Authorization"), "Basic sensitive-value")
|
|
self.assertEqual(timeout, self.module.REQUEST_TIMEOUT)
|
|
|
|
with self.assertRaisesRegex(self.module.ConfigError, "origin"):
|
|
client.request("https://evil.example/actions/runs", accept_json=True)
|
|
self.assertEqual(len(opener.requests), 1)
|
|
|
|
def test_api_client_rejects_cross_origin_redirects(self):
|
|
auth = self.module.ApiAuth(
|
|
source="git credential",
|
|
authorization="Basic sensitive-value",
|
|
)
|
|
handler = self.module.SameOriginRedirectHandler(self.target)
|
|
request = urllib.request.Request(
|
|
self.target.repo_path("/actions/runs"),
|
|
headers={"Authorization": auth.authorization},
|
|
)
|
|
|
|
with self.assertRaises(self.module.ConfigError) as raised:
|
|
handler.redirect_request(
|
|
request,
|
|
None,
|
|
302,
|
|
"Found",
|
|
{},
|
|
"https://evil.example/actions/runs",
|
|
)
|
|
self.assertNotIn("sensitive-value", str(raised.exception))
|
|
|
|
def test_401_diagnostics_distinguish_missing_and_rejected_auth(self):
|
|
anonymous = self.module._explain_status(401, self.target, None)
|
|
self.assertIn("GITEA_TOKEN", anonymous)
|
|
self.assertIn("--use-git-credential", anonymous)
|
|
|
|
auth = self.module.ApiAuth(
|
|
source="git credential",
|
|
authorization="Basic sensitive-value",
|
|
)
|
|
rejected = self.module._explain_status(401, self.target, auth)
|
|
self.assertIn("git credential", rejected)
|
|
self.assertNotIn("sensitive-value", rejected)
|
|
|
|
def test_403_and_404_diagnostics_have_distinct_actions(self):
|
|
auth = self.module.ApiAuth(
|
|
source="GITEA_TOKEN",
|
|
authorization="token sensitive-value",
|
|
)
|
|
forbidden = self.module._explain_status(403, self.target, auth)
|
|
self.assertIn("permission", forbidden)
|
|
self.assertIn("GITEA_TOKEN", forbidden)
|
|
|
|
missing = self.module._explain_status(404, self.target, auth)
|
|
self.assertIn("endpoint", missing)
|
|
self.assertIn("version", missing)
|
|
self.assertNotIn("sensitive-value", forbidden + missing)
|
|
|
|
def test_skill_documents_explicit_git_credential_auth(self):
|
|
skill = (SKILL_ROOT / "SKILL.md").read_text(encoding="utf-8")
|
|
self.assertIn("--use-git-credential", skill)
|
|
self.assertIn("GITEA_TOKEN", skill)
|
|
self.assertIn("HTTPS", skill)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|