1039 lines
42 KiB
Python
1039 lines
42 KiB
Python
#!/usr/bin/env python3
|
|
import configparser
|
|
import hashlib
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import inspect
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import stat
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy"
|
|
BUILDER_PATH = SCRIPT_DIR / "build-gitea-fresh-install-artifact.mjs"
|
|
FIXTURE_ROOT = SCRIPT_DIR / "fixtures" / "gitea"
|
|
COMPOSE_PATH = FIXTURE_ROOT / "docker-compose.gitea.yml"
|
|
DESCRIPTOR_PATH = (
|
|
FIXTURE_ROOT / "deployment" / "gitea-fresh-install-v1.json"
|
|
)
|
|
|
|
|
|
def load_runner():
|
|
loader = importlib.machinery.SourceFileLoader(
|
|
"nodedc_gitea_deploy_under_test",
|
|
str(RUNNER_PATH),
|
|
)
|
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
|
module = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
RUNNER = load_runner()
|
|
|
|
|
|
def completed(returncode=0, stdout="", stderr=""):
|
|
return subprocess.CompletedProcess([], returncode, stdout, stderr)
|
|
|
|
|
|
class GiteaFreshInstallArtifactTest(unittest.TestCase):
|
|
def test_fixture_and_descriptor_are_digest_bound_and_hardened(self):
|
|
compose = COMPOSE_PATH.read_text(encoding="utf-8")
|
|
descriptor = json.loads(DESCRIPTOR_PATH.read_text(encoding="utf-8"))
|
|
digest = hashlib.sha256(COMPOSE_PATH.read_bytes()).hexdigest()
|
|
self.assertEqual(digest, RUNNER.GITEA_COMPOSE_SHA256)
|
|
self.assertEqual(descriptor, RUNNER.expected_gitea_fresh_install_descriptor())
|
|
self.assertIn(f"image: {RUNNER.GITEA_IMAGE}", compose)
|
|
for required in (
|
|
"platform: linux/amd64",
|
|
"pull_policy: never",
|
|
"network_mode: none",
|
|
"GITEA__server__PROTOCOL: http+unix",
|
|
"GITEA__server__HTTP_ADDR: /run/gitea/gitea.sock",
|
|
'GITEA__server__UNIX_SOCKET_PERMISSION: "0666"',
|
|
"GITEA__server__LOCAL_ROOT_URL: http://unix/",
|
|
"source: /volume1/docker/nodedc-gitea/socket",
|
|
"target: /run/gitea",
|
|
"create_host_path: false",
|
|
"stop_grace_period: 30s",
|
|
"driver: json-file",
|
|
'max-size: "10m"',
|
|
'max-file: "3"',
|
|
"GITEA__security__SECRET_KEY_URI: file:/run/secrets/gitea_secret_key",
|
|
"GITEA__security__INTERNAL_TOKEN_URI: file:/run/secrets/gitea_internal_token",
|
|
"GITEA__security__TWO_FACTOR_AUTH: enforced",
|
|
'GITEA__server__LFS_START_SERVER: "false"',
|
|
'GITEA__server__LFS_ALLOW_PURE_SSH: "false"',
|
|
"GITEA__security__ALLOWED_HOST_LIST: loopback",
|
|
'GITEA__service__ENABLE_BASIC_AUTHENTICATION: "false"',
|
|
'GITEA__repository__DISABLE_MIGRATIONS: "true"',
|
|
'GITEA__packages__ENABLED: "false"',
|
|
):
|
|
self.assertIn(required, compose)
|
|
for forbidden in (
|
|
"__FILE",
|
|
"4022",
|
|
"0.0.0.0:3000",
|
|
"ports:",
|
|
"networks:",
|
|
"/var/run/docker.sock",
|
|
"/volume1/docker/gitea",
|
|
"privileged: true",
|
|
"GITEA__server__LFS_JWT_SECRET_URI",
|
|
"gitea_lfs_jwt_secret",
|
|
"lfs-jwt-secret",
|
|
"GITEA__server__REVERSE_PROXY_LIMIT",
|
|
"GITEA__security__ENABLE_REVERSE_PROXY_AUTHENTICATION",
|
|
"GITEA__service__DISABLE_REGULAR_ORG_CREATION",
|
|
):
|
|
self.assertNotIn(forbidden, compose)
|
|
|
|
def test_compose_schema_is_accepted_without_pull_or_start(self):
|
|
result = subprocess.run(
|
|
[
|
|
"docker",
|
|
"compose",
|
|
"--project-name",
|
|
"nodedc-gitea-fixture-test",
|
|
"--file",
|
|
str(COMPOSE_PATH),
|
|
"config",
|
|
"--quiet",
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode == 127:
|
|
self.skipTest("docker compose is unavailable")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_builder_is_deterministic_and_archive_is_data_only(self):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-gitea-builder-") as directory:
|
|
environment = dict(os.environ)
|
|
environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = directory
|
|
command = ["node", str(BUILDER_PATH), "gitea-test-001"]
|
|
first = subprocess.run(
|
|
command,
|
|
env=environment,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
artifact = Path(json.loads(first.stdout)["artifact"])
|
|
first_digest = hashlib.sha256(artifact.read_bytes()).hexdigest()
|
|
second = subprocess.run(
|
|
command,
|
|
env=environment,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertEqual(
|
|
first_digest,
|
|
hashlib.sha256(artifact.read_bytes()).hexdigest(),
|
|
)
|
|
self.assertEqual(json.loads(second.stdout)["sha256"], first_digest)
|
|
with tarfile.open(artifact, "r:gz") as archive:
|
|
names = archive.getnames()
|
|
self.assertEqual(
|
|
names,
|
|
[
|
|
"manifest.env",
|
|
"files.txt",
|
|
"payload",
|
|
"payload/deployment",
|
|
"payload/deployment/gitea-fresh-install-v1.json",
|
|
"payload/docker-compose.gitea.yml",
|
|
],
|
|
)
|
|
manifest = archive.extractfile("manifest.env").read().decode()
|
|
files = archive.extractfile("files.txt").read().decode()
|
|
self.assertEqual(
|
|
manifest,
|
|
"id=gitea-test-001\ncomponent=gitea\ntype=app-overlay\n",
|
|
)
|
|
self.assertEqual(
|
|
files,
|
|
"docker-compose.gitea.yml\n"
|
|
"deployment/gitea-fresh-install-v1.json\n",
|
|
)
|
|
|
|
def test_payload_validator_accepts_exact_fixture_and_rejects_mutation(self):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-gitea-payload-") as directory:
|
|
payload = Path(directory)
|
|
(payload / "deployment").mkdir()
|
|
(payload / RUNNER.GITEA_COMPOSE_REL).write_bytes(COMPOSE_PATH.read_bytes())
|
|
(payload / RUNNER.GITEA_FRESH_INSTALL_DESCRIPTOR_REL).write_bytes(
|
|
DESCRIPTOR_PATH.read_bytes()
|
|
)
|
|
RUNNER.validate_gitea_fresh_install_payload(
|
|
payload,
|
|
RUNNER.GITEA_FRESH_INSTALL_ENTRIES,
|
|
)
|
|
(payload / RUNNER.GITEA_COMPOSE_REL).write_text(
|
|
COMPOSE_PATH.read_text(encoding="utf-8") + "\n# mutation\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(
|
|
RUNNER.DeployError,
|
|
"Compose digest mismatch",
|
|
):
|
|
RUNNER.validate_gitea_fresh_install_payload(
|
|
payload,
|
|
RUNNER.GITEA_FRESH_INSTALL_ENTRIES,
|
|
)
|
|
|
|
|
|
class GiteaFreshInstallRunnerTest(unittest.TestCase):
|
|
def test_registry_and_payload_boundary_are_exact(self):
|
|
component = RUNNER.COMPONENTS["gitea"]
|
|
self.assertEqual(component["payload_root"], RUNNER.GITEA_ROOT)
|
|
self.assertEqual(component["compose_project"], "nodedc-gitea")
|
|
self.assertEqual(component["services"], ("gitea",))
|
|
self.assertTrue(component["bootstrap_root"])
|
|
self.assertTrue(component["compose_no_deps"])
|
|
for allowed in RUNNER.GITEA_FRESH_INSTALL_ENTRIES:
|
|
self.assertTrue(RUNNER.allowed_payload_path("gitea", allowed))
|
|
for rejected in (
|
|
".env",
|
|
"data/gitea.db",
|
|
"config/app.ini",
|
|
"secrets/secret-key",
|
|
"repositories/org/repo.git",
|
|
"users/export.json",
|
|
"tokens/api-token",
|
|
"hooks/post-receive",
|
|
"docker-compose.yml",
|
|
):
|
|
with self.assertRaises(RUNNER.DeployError):
|
|
RUNNER.allowed_payload_path("gitea", rejected)
|
|
|
|
def test_runner_has_no_legacy_root_path_object_or_runtime_mount(self):
|
|
source = inspect.getsource(RUNNER.preflight_gitea_fresh_install)
|
|
prepare = inspect.getsource(RUNNER.prepare_gitea_fresh_runtime)
|
|
self.assertNotIn("GITEA_SALVAGE_LEGACY_ROOT", source)
|
|
self.assertNotIn("GITEA_SALVAGE_LEGACY_ROOT", prepare)
|
|
self.assertNotIn('source: /volume1/docker/gitea', COMPOSE_PATH.read_text())
|
|
self.assertEqual(RUNNER.GITEA_SALVAGE_LEGACY_ROOT, Path("/volume1/docker/gitea"))
|
|
|
|
def test_plan_and_apply_integrate_preflight_before_bootstrap(self):
|
|
plan_source = inspect.getsource(RUNNER.plan_artifact)
|
|
apply_source = inspect.getsource(RUNNER.apply_artifact)
|
|
self.assertIn("gitea_preflight = preflight_gitea_fresh_install()", plan_source)
|
|
preflight_index = apply_source.index("preflight_gitea_fresh_install()")
|
|
bootstrap_index = apply_source.index("if not root.is_dir()")
|
|
self.assertLess(preflight_index, bootstrap_index)
|
|
self.assertIn("defer_bootstrap_root = is_gitea_bootstrap_slice", apply_source)
|
|
self.assertLess(
|
|
apply_source.index("apply_started = True"),
|
|
apply_source.index("copy_payload_path(payload_dir, root, rel"),
|
|
)
|
|
|
|
def test_preflight_fails_closed_and_returns_exact_attestation(self):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-gitea-preflight-") as directory:
|
|
absent_root = Path(directory) / "absent"
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_ROOT", absent_root),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"gitea_compose_project_container_ids",
|
|
return_value=(),
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_gitea_legacy_candidate_network_absent",
|
|
return_value="absent",
|
|
),
|
|
mock.patch.object(RUNNER, "assert_loopback_tcp_port_closed") as port,
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_gitea_nginx_bridge_prerequisite",
|
|
return_value={"sha256": RUNNER.GITEA_NGINX_BRIDGE_SHA256},
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_gitea_reverse_proxy_prerequisite",
|
|
return_value={"upstream": "http://127.0.0.1:3000"},
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_gitea_firewall_prerequisite",
|
|
return_value={"legacy": "isolated"},
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_legacy_gitea_container_isolation",
|
|
return_value="stopped-restart-disabled",
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_gitea_no_docker_port_publications",
|
|
return_value="none",
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"inspect_gitea_docker_server_version",
|
|
return_value="24.0.2",
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"inspect_gitea_builtin_none_network",
|
|
return_value={"Id": "d" * 64},
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"inspect_gitea_docker_compose_version",
|
|
return_value="2.20.1-6047-g6817716",
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"inspect_gitea_local_image",
|
|
return_value=RUNNER.GITEA_IMAGE_ID,
|
|
),
|
|
):
|
|
result = RUNNER.preflight_gitea_fresh_install()
|
|
self.assertEqual(result["mode"], "fresh-root-absent")
|
|
self.assertEqual(result["image_id"], RUNNER.GITEA_IMAGE_ID)
|
|
self.assertEqual(result["legacy_container"], "stopped-restart-disabled")
|
|
self.assertEqual(
|
|
[call.args[0] for call in port.call_args_list],
|
|
[4022],
|
|
)
|
|
|
|
absent_root.mkdir()
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_ROOT", absent_root),
|
|
self.assertRaisesRegex(
|
|
RUNNER.DeployError,
|
|
"root already exists",
|
|
),
|
|
):
|
|
RUNNER.preflight_gitea_fresh_install()
|
|
|
|
def test_local_image_gate_requires_exact_linux_amd64_rootless_user(self):
|
|
valid = [{
|
|
"Id": RUNNER.GITEA_IMAGE_ID,
|
|
"RepoDigests": [RUNNER.GITEA_REPO_DIGEST],
|
|
"Architecture": "amd64",
|
|
"Os": "linux",
|
|
"Config": {"User": "1000:1000"},
|
|
}]
|
|
with mock.patch.object(RUNNER, "docker_json", return_value=valid):
|
|
self.assertEqual(
|
|
RUNNER.inspect_gitea_local_image(),
|
|
RUNNER.GITEA_IMAGE_ID,
|
|
)
|
|
invalid = json.loads(json.dumps(valid))
|
|
invalid[0]["Config"]["User"] = "root"
|
|
with (
|
|
mock.patch.object(RUNNER, "docker_json", return_value=invalid),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "identity/repo-digest"),
|
|
):
|
|
RUNNER.inspect_gitea_local_image()
|
|
wrong_identity = json.loads(json.dumps(valid))
|
|
wrong_identity[0]["Id"] = "sha256:" + "b" * 64
|
|
with (
|
|
mock.patch.object(RUNNER, "docker_json", return_value=wrong_identity),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "identity/repo-digest"),
|
|
):
|
|
RUNNER.inspect_gitea_local_image()
|
|
wrong_digest = json.loads(json.dumps(valid))
|
|
wrong_digest[0]["RepoDigests"] = ["docker.gitea.com/gitea@sha256:" + "c" * 64]
|
|
with (
|
|
mock.patch.object(RUNNER, "docker_json", return_value=wrong_digest),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "identity/repo-digest"),
|
|
):
|
|
RUNNER.inspect_gitea_local_image()
|
|
|
|
def test_docker_server_version_gate_is_exact(self):
|
|
with mock.patch.object(
|
|
RUNNER.subprocess,
|
|
"run",
|
|
return_value=completed(0, stdout="24.0.2\n"),
|
|
):
|
|
self.assertEqual(RUNNER.inspect_gitea_docker_server_version(), "24.0.2")
|
|
for version in ("24.0.1", "24.0.2-synology", "25.0.0", ""):
|
|
with (
|
|
mock.patch.object(
|
|
RUNNER.subprocess,
|
|
"run",
|
|
return_value=completed(0, stdout=version + "\n"),
|
|
),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "exact Docker Server"),
|
|
):
|
|
RUNNER.inspect_gitea_docker_server_version()
|
|
|
|
def test_firewall_gate_removes_only_broad_3000_drop_requirement(self):
|
|
fake_iptables = Path(__file__)
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_IPTABLES", fake_iptables),
|
|
mock.patch.object(
|
|
RUNNER.subprocess,
|
|
"run",
|
|
side_effect=[completed(1), completed(0), completed(0)],
|
|
) as run,
|
|
):
|
|
result = RUNNER.validate_gitea_firewall_prerequisite()
|
|
self.assertEqual(result["legacy_4022"], "input-drop-present")
|
|
commands = [call.args[0] for call in run.call_args_list]
|
|
self.assertIn(
|
|
[
|
|
str(fake_iptables), "-w", "5", "-C", "INPUT", "-p",
|
|
"tcp", "--dport", "3000", "-j", "DROP",
|
|
],
|
|
commands,
|
|
)
|
|
self.assertTrue(any("172.22.0.222/32" in command for command in commands))
|
|
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_IPTABLES", fake_iptables),
|
|
mock.patch.object(
|
|
RUNNER.subprocess,
|
|
"run",
|
|
return_value=completed(0),
|
|
),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "still blocked"),
|
|
):
|
|
RUNNER.validate_gitea_firewall_prerequisite()
|
|
|
|
def test_legacy_candidate_network_absence_fails_closed_on_unexpected_error(self):
|
|
expected = completed(
|
|
1,
|
|
stderr=f"Error: No such network: {RUNNER.GITEA_FORBIDDEN_LEGACY_NETWORK}\n",
|
|
)
|
|
with mock.patch.object(RUNNER.subprocess, "run", return_value=expected):
|
|
self.assertEqual(
|
|
RUNNER.validate_gitea_legacy_candidate_network_absent(),
|
|
"absent",
|
|
)
|
|
for result in (
|
|
completed(0, stdout="[]\n"),
|
|
completed(1, stderr="permission denied\n"),
|
|
completed(2, stderr="daemon unavailable\n"),
|
|
):
|
|
with (
|
|
mock.patch.object(RUNNER.subprocess, "run", return_value=result),
|
|
self.assertRaises(RUNNER.DeployError),
|
|
):
|
|
RUNNER.validate_gitea_legacy_candidate_network_absent()
|
|
|
|
def test_builtin_none_network_identity_is_exact(self):
|
|
exact = [{
|
|
"Name": "none",
|
|
"Id": "d" * 64,
|
|
"Scope": "local",
|
|
"Driver": "null",
|
|
"Internal": False,
|
|
"Attachable": False,
|
|
"Ingress": False,
|
|
"ConfigOnly": False,
|
|
"Containers": {},
|
|
}]
|
|
with mock.patch.object(RUNNER, "docker_json", return_value=exact):
|
|
self.assertEqual(
|
|
RUNNER.inspect_gitea_builtin_none_network()["Id"],
|
|
"d" * 64,
|
|
)
|
|
invalid = json.loads(json.dumps(exact))
|
|
invalid[0]["Driver"] = "bridge"
|
|
with (
|
|
mock.patch.object(RUNNER, "docker_json", return_value=invalid),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "identity mismatch"),
|
|
):
|
|
RUNNER.inspect_gitea_builtin_none_network()
|
|
|
|
def test_reverse_proxy_gate_requires_exact_persistent_and_generated_rule(self):
|
|
exact = {
|
|
RUNNER.GITEA_REVERSE_PROXY_UUID: {
|
|
"backend": {"fqdn": "127.0.0.1", "port": 3000, "protocol": 0},
|
|
"customize_headers": [],
|
|
"description": "Gittea",
|
|
"frontend": {
|
|
"acl": None,
|
|
"fqdn": "git.dcserve.ru",
|
|
"https": {"hsts": False},
|
|
"port": 443,
|
|
"protocol": 1,
|
|
},
|
|
"proxy_connect_timeout": 60,
|
|
"proxy_http_version": 1,
|
|
"proxy_intercept_errors": False,
|
|
"proxy_read_timeout": 60,
|
|
"proxy_send_timeout": 60,
|
|
}
|
|
}
|
|
generated = """
|
|
server {
|
|
server_name git.dcserve.ru ;
|
|
if ( $host !~ "(^git.dcserve.ru$)" ) { return 404; }
|
|
location / {
|
|
proxy_set_header Host $http_host;
|
|
proxy_set_header X-Real-IP $remote_addr;
|
|
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
|
proxy_set_header X-Forwarded-Proto $scheme;
|
|
proxy_pass http://127.0.0.1:3000;
|
|
}
|
|
}
|
|
server { server_name other.example ; proxy_pass http://172.22.0.222:3000; }
|
|
"""
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-gitea-proxy-") as directory:
|
|
root = Path(directory)
|
|
persistent = root / "ReverseProxy.json"
|
|
rendered = root / "gitea.w3conf"
|
|
persistent.write_text("{}", encoding="utf-8")
|
|
rendered.write_text(generated, encoding="utf-8")
|
|
stats = [
|
|
SimpleNamespace(
|
|
st_mode=stat.S_IFREG | 0o644,
|
|
st_uid=0,
|
|
st_gid=0,
|
|
st_size=128,
|
|
),
|
|
SimpleNamespace(
|
|
st_mode=stat.S_IFREG | 0o644,
|
|
st_uid=0,
|
|
st_gid=0,
|
|
st_size=len(generated),
|
|
),
|
|
]
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_REVERSE_PROXY_CONFIG", persistent),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"GITEA_REVERSE_PROXY_GENERATED_CONFIG",
|
|
rendered,
|
|
),
|
|
mock.patch.object(RUNNER, "read_strict_json", return_value=exact),
|
|
mock.patch.object(Path, "lstat", side_effect=stats),
|
|
):
|
|
result = RUNNER.validate_gitea_reverse_proxy_prerequisite()
|
|
self.assertEqual(result["upstream"], "http://127.0.0.1:3000")
|
|
|
|
rendered.write_text(
|
|
generated.replace("$remote_addr;", "$http_x_real_ip;", 1),
|
|
encoding="utf-8",
|
|
)
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_REVERSE_PROXY_CONFIG", persistent),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"GITEA_REVERSE_PROXY_GENERATED_CONFIG",
|
|
rendered,
|
|
),
|
|
mock.patch.object(RUNNER, "read_strict_json", return_value=exact),
|
|
mock.patch.object(Path, "lstat", side_effect=stats),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "prerequisite mismatch"),
|
|
):
|
|
RUNNER.validate_gitea_reverse_proxy_prerequisite()
|
|
rendered.write_text(generated, encoding="utf-8")
|
|
|
|
legacy = json.loads(json.dumps(exact))
|
|
legacy[RUNNER.GITEA_REVERSE_PROXY_UUID]["backend"]["fqdn"] = (
|
|
"172.22.0.222"
|
|
)
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_REVERSE_PROXY_CONFIG", persistent),
|
|
mock.patch.object(RUNNER, "read_strict_json", return_value=legacy),
|
|
mock.patch.object(Path, "lstat", return_value=stats[0]),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "prerequisite mismatch"),
|
|
):
|
|
RUNNER.validate_gitea_reverse_proxy_prerequisite()
|
|
|
|
def test_fresh_sqlite_gate_is_read_only_integrity_and_zero_counts(self):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-gitea-sqlite-") as directory:
|
|
data_root = Path(directory)
|
|
database = data_root / "data" / "gitea.db"
|
|
database.parent.mkdir()
|
|
connection = sqlite3.connect(database)
|
|
connection.executescript(
|
|
'CREATE TABLE "user" (id INTEGER PRIMARY KEY);'
|
|
'CREATE TABLE "repository" (id INTEGER PRIMARY KEY);'
|
|
)
|
|
connection.commit()
|
|
connection.close()
|
|
metadata = SimpleNamespace(
|
|
st_mode=stat.S_IFREG | 0o600,
|
|
st_uid=1000,
|
|
st_size=max(database.stat().st_size, 4096),
|
|
)
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_DATA_DIR", data_root),
|
|
mock.patch.object(Path, "lstat", return_value=metadata),
|
|
):
|
|
self.assertEqual(
|
|
RUNNER.validate_gitea_fresh_sqlite(),
|
|
{"user": 0, "repository": 0},
|
|
)
|
|
connection = sqlite3.connect(database)
|
|
connection.execute('INSERT INTO "repository" DEFAULT VALUES')
|
|
connection.commit()
|
|
connection.close()
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_DATA_DIR", data_root),
|
|
mock.patch.object(Path, "lstat", return_value=metadata),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "unexpectedly contains"),
|
|
):
|
|
RUNNER.validate_gitea_fresh_sqlite()
|
|
|
|
def test_nginx_uds_bridge_is_exact_and_listener_is_nginx_owned(self):
|
|
expected_fragments = (
|
|
"listen 127.0.0.1:3000;",
|
|
"proxy_pass http://unix:/volume1/docker/nodedc-gitea/socket/gitea.sock:;",
|
|
"proxy_set_header Host $http_host;",
|
|
"proxy_set_header X-Real-IP $http_x_real_ip;",
|
|
"proxy_set_header X-Forwarded-For $http_x_forwarded_for;",
|
|
"proxy_set_header X-Forwarded-Proto $http_x_forwarded_proto;",
|
|
"/user/login $http_x_real_ip;",
|
|
"rate=10r/m;",
|
|
"limit_conn nodedc_gitea_conn 40;",
|
|
)
|
|
self.assertTrue(
|
|
all(value in RUNNER.GITEA_NGINX_BRIDGE_CONTENT for value in expected_fragments)
|
|
)
|
|
self.assertEqual(
|
|
hashlib.sha256(RUNNER.GITEA_NGINX_BRIDGE_CONTENT.encode()).hexdigest(),
|
|
RUNNER.GITEA_NGINX_BRIDGE_SHA256,
|
|
)
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-gitea-nginx-") as directory:
|
|
bridge = Path(directory) / "http.nodedc-gitea-uds.conf"
|
|
bridge.write_text(RUNNER.GITEA_NGINX_BRIDGE_CONTENT, encoding="utf-8")
|
|
metadata = SimpleNamespace(
|
|
st_mode=stat.S_IFREG | 0o644,
|
|
st_uid=0,
|
|
st_gid=0,
|
|
)
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_NGINX_BRIDGE_CONFIG", bridge),
|
|
mock.patch.object(Path, "lstat", return_value=metadata),
|
|
mock.patch.object(
|
|
RUNNER.subprocess,
|
|
"run",
|
|
side_effect=[
|
|
completed(0, stderr=RUNNER.GITEA_NGINX_VERSION + "\n"),
|
|
completed(0, stderr="syntax is ok\n"),
|
|
],
|
|
) as run,
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_gitea_nginx_listener",
|
|
return_value={"address": "127.0.0.1:3000"},
|
|
),
|
|
):
|
|
result = RUNNER.validate_gitea_nginx_bridge_prerequisite()
|
|
self.assertEqual(result["sha256"], RUNNER.GITEA_NGINX_BRIDGE_SHA256)
|
|
self.assertEqual(
|
|
run.call_args_list[1].args[0],
|
|
[
|
|
str(RUNNER.GITEA_NGINX),
|
|
"-t",
|
|
"-c",
|
|
str(RUNNER.GITEA_NGINX_MAIN_CONFIG),
|
|
],
|
|
)
|
|
bridge.write_text(
|
|
RUNNER.GITEA_NGINX_BRIDGE_CONTENT.replace(
|
|
"$http_x_forwarded_proto",
|
|
"$scheme",
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_NGINX_BRIDGE_CONFIG", bridge),
|
|
mock.patch.object(Path, "lstat", return_value=metadata),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "identity/metadata"),
|
|
):
|
|
RUNNER.validate_gitea_nginx_bridge_prerequisite()
|
|
|
|
with (
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"gitea_loopback_listener_inodes",
|
|
return_value={"123"},
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"gitea_listener_process_owners",
|
|
return_value={"123": {("nginx", 0), ("nginx", 1023)}},
|
|
),
|
|
):
|
|
RUNNER.validate_gitea_nginx_listener()
|
|
with (
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"gitea_loopback_listener_inodes",
|
|
return_value={"123"},
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"gitea_listener_process_owners",
|
|
return_value={"123": {("docker-proxy", 0)}},
|
|
),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "not exclusively owned"),
|
|
):
|
|
RUNNER.validate_gitea_nginx_listener()
|
|
|
|
def test_installed_config_uses_uri_paths_and_never_secret_bytes(self):
|
|
parser = configparser.ConfigParser(interpolation=None)
|
|
parser.optionxform = str
|
|
for key, value in RUNNER.GITEA_EXPECTED_ENVIRONMENT.items():
|
|
if not key.startswith("GITEA__"):
|
|
continue
|
|
_, section, option = key.split("__", 2)
|
|
if not parser.has_section(section):
|
|
parser.add_section(section)
|
|
parser.set(section, option, value)
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-gitea-config-") as directory:
|
|
config_root = Path(directory)
|
|
app_ini = config_root / "app.ini"
|
|
with app_ini.open("w", encoding="utf-8") as output:
|
|
output.write("APP_NAME = NODE.DC Git\nRUN_USER = git\n\n")
|
|
parser.write(output)
|
|
metadata = SimpleNamespace(
|
|
st_mode=stat.S_IFREG | 0o600,
|
|
st_uid=1000,
|
|
st_size=app_ini.stat().st_size,
|
|
)
|
|
secrets = ("secret-a" * 10, "secret-b" * 10)
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_CONFIG_DIR", config_root),
|
|
mock.patch.object(Path, "lstat", return_value=metadata),
|
|
):
|
|
RUNNER.validate_gitea_installed_config(secrets)
|
|
parser.set("server", "LFS_JWT_SECRET", "unsafe-generated-secret")
|
|
with app_ini.open("w", encoding="utf-8") as output:
|
|
output.write("APP_NAME = NODE.DC Git\nRUN_USER = git\n\n")
|
|
parser.write(output)
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_CONFIG_DIR", config_root),
|
|
mock.patch.object(Path, "lstat", return_value=metadata),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "plaintext secret"),
|
|
):
|
|
RUNNER.validate_gitea_installed_config(secrets)
|
|
parser.remove_option("server", "LFS_JWT_SECRET")
|
|
with app_ini.open("w", encoding="utf-8") as output:
|
|
output.write("APP_NAME = NODE.DC Git\nRUN_USER = git\n\n")
|
|
parser.write(output)
|
|
with app_ini.open("a", encoding="utf-8") as output:
|
|
output.write(f"\n# leaked={secrets[0]}\n")
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_CONFIG_DIR", config_root),
|
|
mock.patch.object(Path, "lstat", return_value=metadata),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "secret bytes"),
|
|
):
|
|
RUNNER.validate_gitea_installed_config(secrets)
|
|
|
|
def test_installed_config_rejects_duplicate_and_root_shadow_options(self):
|
|
sections = {}
|
|
for key, value in RUNNER.GITEA_EXPECTED_ENVIRONMENT.items():
|
|
if key.startswith("GITEA__"):
|
|
_, section, option = key.split("__", 2)
|
|
sections.setdefault(section, []).append((option, value))
|
|
raw = "APP_NAME = NODE.DC Git\nRUN_USER = git\n\n" + "\n".join(
|
|
"[" + section + "]\n" + "\n".join(
|
|
f"{option} = {value}" for option, value in values
|
|
)
|
|
for section, values in sections.items()
|
|
) + "\n"
|
|
RUNNER.parse_gitea_app_ini_explicit(raw)
|
|
with self.assertRaisesRegex(RUNNER.DeployError, "duplicate option"):
|
|
RUNNER.parse_gitea_app_ini_explicit(
|
|
raw.replace(
|
|
"[server]\n",
|
|
"[server]\nPROTOCOL = http\n",
|
|
1,
|
|
)
|
|
)
|
|
with self.assertRaisesRegex(RUNNER.DeployError, "duplicate/unsafe section"):
|
|
RUNNER.parse_gitea_app_ini_explicit(raw + "\n[SERVER]\nROOT_URL=x\n")
|
|
with self.assertRaisesRegex(RUNNER.DeployError, "unsafe root-level"):
|
|
RUNNER.parse_gitea_app_ini_explicit(
|
|
"PROTOCOL = http\n" + raw
|
|
)
|
|
with self.assertRaisesRegex(RUNNER.DeployError, "duplicate/unsafe section"):
|
|
RUNNER.parse_gitea_app_ini_explicit(
|
|
"[DEFAULT]\nPROTOCOL = http\n" + raw
|
|
)
|
|
|
|
def test_runner_generates_distinct_non_artifact_secret_files(self):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-gitea-secrets-") as directory:
|
|
root = Path(directory)
|
|
paths = [root / name for name in ("secret-key", "internal-token")]
|
|
with mock.patch.object(RUNNER.os, "fchown") as chown:
|
|
for index, path in enumerate(paths):
|
|
RUNNER.create_gitea_runtime_secret(path, f"test-{index}")
|
|
values = [path.read_text(encoding="ascii").strip() for path in paths]
|
|
self.assertEqual(len(set(values)), 2)
|
|
self.assertTrue(all(RUNNER.GITEA_SECRET_RE.fullmatch(value) for value in values))
|
|
self.assertTrue(
|
|
all(stat.S_IMODE(path.stat().st_mode) == 0o400 for path in paths)
|
|
)
|
|
self.assertEqual(chown.call_count, 2)
|
|
|
|
def test_compose_runtime_is_no_pull_and_exact_service_only(self):
|
|
with (
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"compose_base_cmd",
|
|
return_value=["docker", "compose", "-f", "candidate.yml"],
|
|
),
|
|
mock.patch.object(RUNNER.subprocess, "run", return_value=completed()) as run,
|
|
):
|
|
RUNNER.run_gitea_fresh_install_compose(
|
|
("gitea",),
|
|
RUNNER.GITEA_FRESH_INSTALL_ENTRIES,
|
|
)
|
|
command = run.call_args_list[0].args[0]
|
|
self.assertEqual(
|
|
command[-7:],
|
|
[
|
|
"up", "-d", "--force-recreate", "--pull", "never",
|
|
"--no-deps", "gitea",
|
|
],
|
|
)
|
|
self.assertNotIn("pull", command[:command.index("up")])
|
|
|
|
def test_health_dispatch_is_gitea_specific(self):
|
|
with mock.patch.object(
|
|
RUNNER,
|
|
"accept_gitea_fresh_install_runtime",
|
|
) as accept:
|
|
RUNNER.run_healthchecks(
|
|
"gitea",
|
|
RUNNER.GITEA_FRESH_INSTALL_ENTRIES,
|
|
("gitea",),
|
|
)
|
|
accept.assert_called_once_with()
|
|
|
|
def test_socket_boundary_and_no_docker_publications_are_exact(self):
|
|
parent = SimpleNamespace(
|
|
st_mode=stat.S_IFDIR | 0o750,
|
|
st_uid=1000,
|
|
st_gid=1023,
|
|
)
|
|
uds = SimpleNamespace(
|
|
st_mode=stat.S_IFSOCK | 0o666,
|
|
st_uid=1000,
|
|
st_gid=1000,
|
|
)
|
|
with mock.patch.object(Path, "lstat", side_effect=[parent, uds]):
|
|
self.assertEqual(
|
|
RUNNER.validate_gitea_socket_boundary(),
|
|
str(RUNNER.GITEA_SOCKET_FILE),
|
|
)
|
|
bad = SimpleNamespace(**vars(uds))
|
|
bad.st_mode = stat.S_IFREG | 0o666
|
|
with (
|
|
mock.patch.object(Path, "lstat", side_effect=[parent, bad]),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "identity/metadata"),
|
|
):
|
|
RUNNER.validate_gitea_socket_boundary()
|
|
|
|
safe_container = {"HostConfig": {"PortBindings": {}}}
|
|
with (
|
|
mock.patch.object(
|
|
RUNNER.subprocess,
|
|
"run",
|
|
return_value=completed(0, stdout="a" * 12 + "\n"),
|
|
),
|
|
mock.patch.object(RUNNER, "docker_json", return_value=[safe_container]),
|
|
):
|
|
RUNNER.validate_gitea_no_docker_port_publications()
|
|
published = {
|
|
"HostConfig": {
|
|
"PortBindings": {
|
|
"3000/tcp": [{"HostIp": "127.0.0.1", "HostPort": "3000"}]
|
|
}
|
|
}
|
|
}
|
|
with (
|
|
mock.patch.object(
|
|
RUNNER.subprocess,
|
|
"run",
|
|
return_value=completed(0, stdout="a" * 12 + "\n"),
|
|
),
|
|
mock.patch.object(RUNNER, "docker_json", return_value=[published]),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "must not publish"),
|
|
):
|
|
RUNNER.validate_gitea_no_docker_port_publications()
|
|
|
|
def test_health_json_requires_pass_and_database_ping_pass(self):
|
|
response = mock.MagicMock()
|
|
response.status = 200
|
|
response.read.return_value = json.dumps({
|
|
"status": "pass",
|
|
"checks": {"database:ping": [{"status": "pass"}]},
|
|
}).encode("utf-8")
|
|
response.__enter__.return_value = response
|
|
with mock.patch.object(
|
|
RUNNER.NO_REDIRECT_OPENER,
|
|
"open",
|
|
return_value=response,
|
|
):
|
|
payload = RUNNER.healthcheck_gitea_json()
|
|
self.assertEqual(payload["status"], "pass")
|
|
|
|
response.read.return_value = json.dumps({
|
|
"status": "pass",
|
|
"checks": {"database:ping": [{"status": "fail"}]},
|
|
}).encode("utf-8")
|
|
with (
|
|
mock.patch.object(
|
|
RUNNER.NO_REDIRECT_OPENER,
|
|
"open",
|
|
return_value=response,
|
|
),
|
|
mock.patch.object(RUNNER.time, "sleep"),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "contract mismatch"),
|
|
):
|
|
RUNNER.healthcheck_gitea_json()
|
|
|
|
def test_uds_health_requires_database_ping_pass(self):
|
|
connection = mock.MagicMock()
|
|
response = mock.MagicMock()
|
|
response.status = 200
|
|
response.read.return_value = json.dumps({
|
|
"status": "pass",
|
|
"checks": {"database:ping": [{"status": "pass"}]},
|
|
}).encode("utf-8")
|
|
with (
|
|
mock.patch.object(RUNNER.socket, "socket", return_value=connection),
|
|
mock.patch.object(RUNNER.http.client, "HTTPResponse", return_value=response),
|
|
):
|
|
payload = RUNNER.healthcheck_gitea_uds_json()
|
|
self.assertEqual(payload["status"], "pass")
|
|
connection.connect.assert_called_once_with(str(RUNNER.GITEA_SOCKET_FILE))
|
|
self.assertIn(b"Host: git.dcserve.ru", connection.sendall.call_args.args[0])
|
|
|
|
response.read.return_value = json.dumps({
|
|
"status": "pass",
|
|
"checks": {"database:ping": [{"status": "fail"}]},
|
|
}).encode("utf-8")
|
|
with (
|
|
mock.patch.object(RUNNER.socket, "socket", return_value=connection),
|
|
mock.patch.object(RUNNER.http.client, "HTTPResponse", return_value=response),
|
|
mock.patch.object(RUNNER.time, "sleep"),
|
|
self.assertRaisesRegex(RUNNER.DeployError, "contract mismatch"),
|
|
):
|
|
RUNNER.healthcheck_gitea_uds_json()
|
|
|
|
def test_rollback_stops_only_candidate_and_quarantines_new_state(self):
|
|
entries = RUNNER.GITEA_FRESH_INSTALL_ENTRIES
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-gitea-rollback-") as directory:
|
|
root = Path(directory) / "nodedc-gitea"
|
|
backup = Path(directory) / "backup"
|
|
root.mkdir()
|
|
backup.mkdir()
|
|
data = root / "data"
|
|
config = root / "config"
|
|
secrets = root / "secrets"
|
|
socket_dir = root / "socket"
|
|
for path in (data, config, secrets, socket_dir):
|
|
path.mkdir()
|
|
(backup / "existing-files.txt").write_text("", encoding="utf-8")
|
|
(backup / "missing-files.txt").write_text(
|
|
"\n".join(entries) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_ROOT", root),
|
|
mock.patch.object(RUNNER, "GITEA_DATA_DIR", data),
|
|
mock.patch.object(RUNNER, "GITEA_CONFIG_DIR", config),
|
|
mock.patch.object(RUNNER, "GITEA_SECRET_DIR", secrets),
|
|
mock.patch.object(RUNNER, "GITEA_SOCKET_DIR", socket_dir),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"stop_and_remove_compose_services",
|
|
) as stop,
|
|
mock.patch.object(RUNNER, "restore_overlay_source") as restore,
|
|
mock.patch.object(RUNNER, "assert_loopback_tcp_port_closed"),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_gitea_legacy_candidate_network_absent",
|
|
return_value="absent",
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_gitea_no_docker_port_publications",
|
|
return_value="none",
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_legacy_gitea_container_isolation",
|
|
return_value="stopped-restart-disabled",
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"gitea_compose_project_container_ids",
|
|
return_value=(),
|
|
),
|
|
):
|
|
result = RUNNER.rollback_gitea_fresh_install(
|
|
root,
|
|
backup,
|
|
entries,
|
|
"test-stamp",
|
|
runtime_started=True,
|
|
)
|
|
stop.assert_called_once_with("gitea", ("gitea",))
|
|
restore.assert_called_once_with(root, backup, entries, "test-stamp")
|
|
self.assertFalse(root.exists())
|
|
retained = root.with_name("nodedc-gitea.failed-test-stamp")
|
|
self.assertTrue((retained / "data").is_dir())
|
|
self.assertTrue((retained / "config").is_dir())
|
|
self.assertTrue((retained / "secrets").is_dir())
|
|
self.assertTrue((retained / "socket").is_dir())
|
|
self.assertIn("runtime-state-preserved", result)
|
|
|
|
def test_rollback_preserves_root_when_candidate_absence_is_unproven(self):
|
|
entries = RUNNER.GITEA_FRESH_INSTALL_ENTRIES
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-gitea-reconcile-") as directory:
|
|
root = Path(directory) / "nodedc-gitea"
|
|
backup = Path(directory) / "backup"
|
|
root.mkdir()
|
|
backup.mkdir()
|
|
(backup / "existing-files.txt").write_text("", encoding="utf-8")
|
|
(backup / "missing-files.txt").write_text(
|
|
"\n".join(entries) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with (
|
|
mock.patch.object(RUNNER, "GITEA_ROOT", root),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"stop_and_remove_compose_services",
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"gitea_compose_project_container_ids",
|
|
return_value=("a" * 12,),
|
|
),
|
|
mock.patch.object(RUNNER, "restore_overlay_source") as restore,
|
|
self.assertRaisesRegex(
|
|
RUNNER.ReconciliationRequired,
|
|
"candidate_still_present",
|
|
),
|
|
):
|
|
RUNNER.rollback_gitea_fresh_install(
|
|
root,
|
|
backup,
|
|
entries,
|
|
"test-stamp",
|
|
runtime_started=True,
|
|
)
|
|
self.assertTrue(root.is_dir())
|
|
restore.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|