dashcaddy/ops/shipdeck-bridge/test_fleet_ops.py

import json
import os
import tempfile
import unittest
from unittest import mock

import gh_install
import image_install
import bridge


class FleetOpsTests(unittest.TestCase):
    def test_empty_bridge_token_fails_closed(self):
        with mock.patch("builtins.open", mock.mock_open(read_data="  \n")):
            with self.assertRaisesRegex(RuntimeError, "refusing to start unauthenticated"):
                bridge.read_token()

    def test_detects_go_node_python_at_requested_port(self):
        gh_install._STATE.update(PORT=8977)
        with mock.patch.object(gh_install, "_used_ports", return_value={8977}):
            cases = {
                "go": {"go.mod": "module x\n", "main.go": "package main\nfunc main(){}\n"},
                "node": {"package.json": json.dumps({"main": "server.js"}), "server.js": ""},
                "python": {"requirements.txt": "", "app.py": ""},
            }
            for i, (kind, files) in enumerate(cases.items()):
                with tempfile.TemporaryDirectory() as d:
                    for name, body in files.items():
                        with open(os.path.join(d, name), "w", encoding="utf-8") as fh: fh.write(body)
                    got = gh_install._detect_and_emit(d, "demo-" + kind, "demo-" + kind, "100.121.150.22", 8100 + i)
                    self.assertEqual(got["port"], 8100 + i)
                    self.assertTrue(got["mode"].startswith(kind))
                    self.assertTrue(os.path.isfile(got["shipdeckfile"]))

    def test_rejects_repository_controlled_go_package_shell_metachars(self):
        gh_install._STATE.update(PORT=8977)
        with tempfile.TemporaryDirectory() as d, mock.patch.object(
                gh_install, "_used_ports", return_value={8977}):
            os.mkdir(os.path.join(d, "cmd;touch-pwned"))
            with open(os.path.join(d, "go.mod"), "w", encoding="utf-8") as fh:
                fh.write("module x\n")
            with open(os.path.join(d, "cmd;touch-pwned", "main.go"), "w", encoding="utf-8") as fh:
                fh.write("package main\nfunc main(){}\n")
            with self.assertRaisesRegex(RuntimeError, "unsafe characters"):
                gh_install._detect_and_emit(d, "demo", "demo", "100.121.150.22", 8100)

    def test_gitea_repo_listing_rejects_arbitrary_hosts_before_http(self):
        with mock.patch("gh_install.urllib.request.urlopen") as urlopen:
            code, body = gh_install.list_repos({"gitea_url": "https://127.0.0.1"})
        self.assertEqual(code, 400)
        self.assertIn("configured fleet Gitea", body["error"])
        urlopen.assert_not_called()

    def test_unknown_git_host_skips_metadata_http_but_remains_cloneable(self):
        payload = {"repo_url": "https://code.example/owner/repo", "service": "demo"}
        lock = mock.MagicMock()
        lock.__enter__ = mock.Mock()
        lock.__exit__ = mock.Mock(return_value=False)
        gh_install._STATE.update(RE_SERVICE=gh_install.re.compile(r"^[a-z0-9][a-z0-9-]{0,62}$"),
                                 REPOS_ROOT="/tmp", PORT=8977, LOCK=lock, RUN=mock.Mock())
        with mock.patch("gh_install._http_json") as http_json, mock.patch(
                "gh_install.os.path.exists", return_value=True):
            code, _ = gh_install.gh_install(payload)
        self.assertEqual(code, 409)
        http_json.assert_not_called()

    def test_git_token_is_env_only_never_clone_url_or_error(self):
        secret = "ghp_private_secret"
        lock = mock.MagicMock()
        lock.__enter__ = mock.Mock()
        lock.__exit__ = mock.Mock(return_value=False)
        with tempfile.TemporaryDirectory() as root:
            gh_install._STATE.update(RE_SERVICE=gh_install.re.compile(r"^[a-z0-9][a-z0-9-]{0,62}$"),
                                     REPOS_ROOT=root, PORT=8977, LOCK=lock, RUN=mock.Mock())
            failed = mock.Mock(returncode=1, stdout="", stderr="clone rejected " + secret)
            with mock.patch.object(gh_install, "_http_json", return_value={"name": "repo", "owner": {}}), mock.patch(
                    "gh_install.subprocess.run", return_value=failed) as run:
                code, body = gh_install.gh_install({"repo_url": "https://github.com/acme/repo", "service": "demo", "token": secret})
        self.assertEqual(code, 400)
        argv = run.call_args.args[0]
        env = run.call_args.kwargs["env"]
        self.assertEqual(argv[-2], "https://github.com/acme/repo")
        self.assertNotIn(secret, " ".join(argv))
        self.assertIn(secret, env["GIT_CONFIG_VALUE_0"])
        self.assertNotIn(secret, json.dumps(body))

    def test_image_install_pins_digest_and_never_returns_env_secret(self):
        digest = "a" * 64
        pull = {"digest": "sha256:" + digest, "config": {"config": {"Entrypoint": [], "Cmd": ["/bin/app"]}}}
        calls = []
        def run(args, timeout):
            calls.append(args)
            return (0, json.dumps(pull)) if args[0] == "pull" else (0, "DEPLOYED")
        image_install.init_shared(mock.MagicMock(), run)
        image_install._STATE["LOCK"].__enter__ = mock.Mock()
        image_install._STATE["LOCK"].__exit__ = mock.Mock(return_value=False)
        payload = {"image": "alpine:3.20", "name": "demo", "subdomain": "demo", "port": 8080, "env": {"API_TOKEN": "private-value"}}
        with tempfile.TemporaryDirectory() as drafts, mock.patch.object(image_install, "DRAFT_ROOT", drafts), mock.patch("image_install.subprocess.run") as sp:
            sp.return_value.stdout = "100.121.150.22\n"
            code, body = image_install.install_image(payload)
        self.assertEqual(code, 200)
        self.assertNotIn("private-value", json.dumps(body))
        self.assertEqual(calls[0][:2], ["pull", "--json"])
        self.assertEqual(calls[1][0], "deploy")
        self.assertIn("@sha256:" + digest, body["service"]["image"])

    def test_image_install_rejects_traversal_before_pull(self):
        run = mock.Mock()
        image_install.init_shared(mock.MagicMock(), run)
        code, _ = image_install.install_image({"image": "alpine:3.20", "name": "demo", "subdomain": "demo", "port": 8080, "mounts": [{"source": "/tmp/../etc", "target": "/data"}]})
        self.assertEqual(code, 400)
        run.assert_not_called()


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