akyldash/infra/deploy/test_gitea_api_binary.py

import hashlib
import os
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
FIXTURE = ROOT / "infra/deploy/gitea-api-base.fixture"
PATCH = ROOT / "infra/deploy/gitea-api-binary-download.patch"
FIXTURE_SHA256 = "8227332d28e27bcce9b4e3801f6b8d4323636174a82c93a0177970f560629df0"


FAKE_CURL = r'''#!/usr/bin/env python3
import os
import sys
from pathlib import Path

args = sys.argv[1:]
config = sys.stdin.read() if "--config" in args else ""
def value(flag, default=None):
    return args[args.index(flag) + 1] if flag in args else default

url = args[-1]
mode = os.environ["FAKE_CURL_MODE"]
log = Path(os.environ["FAKE_CURL_LOG"])
log.write_text(log.read_text() + url + "\n" if log.exists() else url + "\n")
headers = value("-D")
output = value("-o")
has_token = "Authorization: token fixture-secret" in config or any(arg == "Authorization: token fixture-secret" for arg in args)
if any("fixture-secret" in arg for arg in args):
    Path(os.environ["FAKE_TOKEN_ARGS"]).write_text("leaked")
if "gitea.test" in url and has_token:
    Path(os.environ["FAKE_GITEA_AUTH"]).write_text("True")
if "blob.test" in url or not value("-w"):
    if "blob.test" in url:
        Path(os.environ["FAKE_BLOB_AUTH"]).write_text(str(has_token))
    body = b"zip-fixture-bytes"
    if mode == "oversize":
        body = b"x" * (10 * 1024 * 1024 + 10)
    try:
        if output and output != "-":
            Path(output).write_bytes(body)
        else:
            sys.stdout.buffer.write(body)
            sys.stdout.buffer.flush()
    except BrokenPipeError:
        raise SystemExit(23)
    if value("-w"):
        print("200", end="")
    raise SystemExit(0)

if value("-w"):
    code = "302" if mode in {"redirect", "unlisted", "oversize"} else "200"
    if headers:
        location = "https://blob.test/archive.zip" if mode != "unlisted" else "https://evil.test/archive.zip"
        Path(headers).write_text(f"HTTP/1.1 {code} Fixture\r\nLocation: {location}\r\n\r\n")
    print(code, end="")
    raise SystemExit(0)

raise SystemExit(99)
'''


class GiteaApiBinaryDownloadTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.root = Path(self.temp.name)
        self.bin = self.root / "bin"
        self.bin.mkdir()
        self.helper = self.root / "gitea-api"
        shutil.copyfile(FIXTURE, self.helper)
        self.helper.chmod(0o700)
        applied = subprocess.run(
            ["patch", "-p1", "-i", str(PATCH)], cwd=self.root,
            text=True, capture_output=True,
        )
        if applied.returncode:
            self.fail(applied.stderr or applied.stdout)
        self.curl = self.bin / "curl"
        self.curl.write_text(FAKE_CURL, encoding="utf-8")
        self.curl.chmod(0o700)
        self.log = self.root / "calls"
        self.auth = self.root / "blob-auth"
        self.gitea_auth = self.root / "gitea-auth"
        self.token_args = self.root / "token-args"

    def tearDown(self):
        self.temp.cleanup()

    def run_download(self, mode, output, origins="https://blob.test"):
        env = {
            **os.environ,
            "PATH": f"{self.bin}:{os.environ['PATH']}",
            "GITEA_URL": "https://gitea.test",
            "GITEA_TOKEN": "fixture-secret",
            "GITEA_DOWNLOAD_ORIGINS": origins,
            "FAKE_CURL_MODE": mode,
            "FAKE_CURL_LOG": str(self.log),
            "FAKE_BLOB_AUTH": str(self.auth),
            "FAKE_GITEA_AUTH": str(self.gitea_auth),
            "FAKE_TOKEN_ARGS": str(self.token_args),
        }
        return subprocess.run(
            [str(self.helper), "DOWNLOAD", "/repos/admin/akyldash/actions/artifacts/7/zip", str(output)],
            env=env, text=True, capture_output=True,
        )

    def test_fixture_is_exact_audited_helper_and_patch_applies(self):
        self.assertEqual(hashlib.sha256(FIXTURE.read_bytes()).hexdigest(), FIXTURE_SHA256)

    def test_downloads_allowlisted_https_redirect_without_forwarding_token(self):
        output = self.root / "artifact.zip"
        result = self.run_download("redirect", output)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(output.read_bytes(), b"zip-fixture-bytes")
        self.assertEqual(output.stat().st_mode & 0o777, 0o600)
        self.assertEqual(self.auth.read_text(), "False")
        self.assertTrue(self.gitea_auth.exists())
        self.assertFalse(self.token_args.exists())

    def test_downloads_direct_https_response_from_gitea(self):
        output = self.root / "artifact.zip"
        result = self.run_download("direct", output)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(output.read_bytes(), b"zip-fixture-bytes")
        self.assertTrue(self.gitea_auth.exists())
        self.assertFalse(self.token_args.exists())

    def test_refuses_unlisted_redirect_before_fetching_target(self):
        output = self.root / "artifact.zip"
        result = self.run_download("unlisted", output)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("not allowlisted", result.stderr)
        self.assertFalse(output.exists())
        self.assertNotIn("evil.test", self.log.read_text())

    def test_caps_chunked_response_and_does_not_publish_partial_file(self):
        output = self.root / "artifact.zip"
        result = self.run_download("oversize", output)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("10 MiB", result.stderr)
        self.assertFalse(output.exists())

    def test_refuses_to_overwrite_existing_output(self):
        output = self.root / "artifact.zip"
        output.write_bytes(b"keep")
        result = self.run_download("redirect", output)
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual(output.read_bytes(), b"keep")


if __name__ == "__main__":
    unittest.main()