172 lines
8.2 KiB
Python
172 lines
8.2 KiB
Python
import contextlib
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
import threading
|
|
import importlib.util
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / "skills" / "updating-gitea-repositories" / "scripts" / "credential_migrate.py"
|
|
spec = importlib.util.spec_from_file_location("credential_migrate", SCRIPT)
|
|
credential_migrate = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(credential_migrate)
|
|
|
|
SITE = "https://gitea.invalid"
|
|
ACCOUNT = "fictional-alice"
|
|
SECRET = b"fictional-test-secret-do-not-use"
|
|
|
|
|
|
class FakeHelper:
|
|
def __init__(self, response=None, after_store=None):
|
|
self.response = response
|
|
self.after_store = after_store
|
|
self.operations = []
|
|
self.stored = None
|
|
|
|
def get(self, request):
|
|
self.operations.append("get")
|
|
if self.stored is not None:
|
|
return self.after_store if self.after_store is not None else self.stored
|
|
return self.response
|
|
|
|
def store(self, credential):
|
|
self.operations.append("store")
|
|
self.stored = dict(credential)
|
|
|
|
|
|
class MigrationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.good = {"username": ACCOUNT.encode(), "password": SECRET}
|
|
self.source = FakeHelper(self.good)
|
|
self.target = FakeHelper()
|
|
|
|
def migrate(self, site=SITE, account=ACCOUNT):
|
|
return credential_migrate.migrate(site, account, self.source, self.target)
|
|
|
|
def test_copies_without_touching_source_and_verifies_target(self):
|
|
self.migrate()
|
|
self.assertEqual(self.source.operations, ["get"])
|
|
self.assertEqual(self.target.operations, ["get", "store", "get"])
|
|
self.assertTrue(self.target.stored["password"] == SECRET, "stored credential differs")
|
|
|
|
def test_existing_target_is_not_overwritten(self):
|
|
self.target.response = {"username": b"other", "password": b"other-fake-secret"}
|
|
with self.assertRaises(credential_migrate.MigrationError):
|
|
self.migrate()
|
|
self.assertEqual(self.target.operations, ["get"])
|
|
|
|
def test_refuses_missing_or_wrong_source_account_and_password(self):
|
|
for response in (
|
|
{"password": SECRET},
|
|
{"username": b"different-account", "password": SECRET},
|
|
{"username": ACCOUNT.encode()},
|
|
{"username": ACCOUNT.encode(), "password": b""},
|
|
{"username": ACCOUNT.encode(), "password": SECRET, "host": b"other.invalid"},
|
|
{"username": ACCOUNT.encode(), "password": SECRET, "protocol": b"http"},
|
|
):
|
|
with self.subTest(fields=tuple(response.keys())):
|
|
self.source = FakeHelper(response)
|
|
self.target = FakeHelper()
|
|
with self.assertRaises(credential_migrate.MigrationError):
|
|
self.migrate()
|
|
self.assertNotIn("store", self.target.operations)
|
|
|
|
def test_refuses_unsafe_url_and_account_before_accessing_helpers(self):
|
|
for site, account in (
|
|
("http://gitea.invalid", ACCOUNT),
|
|
("https://name:secret@gitea.invalid", ACCOUNT),
|
|
("https://gitea.invalid/repo", ACCOUNT),
|
|
("https://gitea.invalid?secret=1", ACCOUNT),
|
|
(SITE, "alice\npassword=leak"),
|
|
):
|
|
with self.subTest(site=site.split("@")[0], account=account.splitlines()[0]):
|
|
self.source = FakeHelper(self.good)
|
|
self.target = FakeHelper()
|
|
with self.assertRaises(credential_migrate.MigrationError):
|
|
self.migrate(site, account)
|
|
self.assertEqual(self.source.operations, [])
|
|
self.assertEqual(self.target.operations, [])
|
|
|
|
def test_failed_readback_is_not_reported_as_success(self):
|
|
self.target.after_store = {"username": ACCOUNT.encode(), "password": b"wrong-fake-secret"}
|
|
with self.assertRaises(credential_migrate.MigrationError):
|
|
self.migrate()
|
|
self.assertEqual(self.source.operations, ["get"])
|
|
self.assertNotIn("erase", self.target.operations)
|
|
|
|
def test_helper_failure_message_does_not_leak_secret(self):
|
|
class BadHelper(FakeHelper):
|
|
def get(self, request):
|
|
raise RuntimeError(SECRET.decode())
|
|
self.source = BadHelper()
|
|
with self.assertRaises(credential_migrate.MigrationError) as caught:
|
|
self.migrate()
|
|
self.assertNotIn(SECRET.decode(), str(caught.exception))
|
|
|
|
def test_explicit_source_rejects_path_scoped_and_ambiguous_records(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source_file = Path(directory) / "credentials"
|
|
for contents in (
|
|
"https://fictional-alice:fictional-secret@gitea.invalid/team/first.git\n",
|
|
"https://fictional-alice:fictional-secret@gitea.invalid\nhttps://fictional-alice:another-secret@gitea.invalid\n",
|
|
):
|
|
source_file.write_text(contents, encoding="utf-8")
|
|
with self.assertRaises(credential_migrate.MigrationError) as caught:
|
|
credential_migrate.inspect_source_file(SITE, ACCOUNT, source_file)
|
|
self.assertNotIn("fictional-secret", str(caught.exception))
|
|
source_file.write_text("https://fictional-alice:fictional-secret@gitea.invalid\n", encoding="utf-8")
|
|
credential_migrate.inspect_source_file(SITE, ACCOUNT, source_file)
|
|
def test_git_http_does_not_emit_curl_trace(self):
|
|
git = shutil.which("git")
|
|
if not git:
|
|
self.skipTest("Git is not available")
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
def log_message(self, *args):
|
|
pass
|
|
server = HTTPServer(("127.0.0.1", 0), Handler)
|
|
worker = threading.Thread(target=server.serve_forever, daemon=True)
|
|
worker.start()
|
|
try:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
env = credential_migrate.safe_environment()
|
|
env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull})
|
|
proc = subprocess.run([git, "-c", "credential.helper=", "ls-remote", "http://127.0.0.1:" + str(server.server_port) + "/repo.git"], cwd=directory, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False)
|
|
self.assertNotIn(b"=> Send header", proc.stderr, "HTTP tracing was enabled")
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
worker.join()
|
|
def test_git_store_source_keeps_old_record_and_reuses_across_paths(self):
|
|
git = shutil.which("git")
|
|
if not git:
|
|
self.skipTest("Git is not available")
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source_file = Path(directory) / "source"
|
|
target_file = Path(directory) / "target"
|
|
base_env = credential_migrate.safe_environment()
|
|
source = credential_migrate.GitHelper([git, "credential-store", "--file=" + str(source_file)], base_env, directory)
|
|
target = credential_migrate.GitHelper([git, "credential-store", "--file=" + str(target_file)], base_env, directory)
|
|
original = {"protocol": b"https", "host": b"gitea.invalid", "username": ACCOUNT.encode(), "password": SECRET}
|
|
source.store(original)
|
|
before = source_file.read_bytes()
|
|
credential_migrate.migrate(SITE, ACCOUNT, source, target)
|
|
self.assertEqual(source_file.read_bytes(), before)
|
|
for path in ("team/first.git", "team/second.git"):
|
|
config = "!" + "git credential-store --file='" + str(target_file).replace("\\", "/") + "'"
|
|
request = ("url=" + SITE + "/" + path + "\n\n").encode()
|
|
completed = subprocess.run([git, "-c", "credential.helper=", "-c", "credential.helper=" + config, "credential", "fill"], input=request, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=base_env, cwd=directory, check=False)
|
|
self.assertEqual(completed.returncode, 0, "isolated git lookup failed")
|
|
self.assertTrue(b"password=" + SECRET + b"\n" in completed.stdout, "credential not reused")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|